feat(pam): real-time session log sync via incremental batch uploads

backend/src/@types/knex.d.ts

@@ -712,6 +712,11 @@ import {
   TPamResourceRotationRulesUpdate
 } from "@app/db/schemas/pam-resource-rotation-rules";
 import { TPamResources, TPamResourcesInsert, TPamResourcesUpdate } from "@app/db/schemas/pam-resources";
+import {
+  TPamSessionEventBatches,
+  TPamSessionEventBatchesInsert,
+  TPamSessionEventBatchesUpdate
+} from "@app/db/schemas/pam-session-event-batches";
 import { TPamSessions, TPamSessionsInsert, TPamSessionsUpdate } from "@app/db/schemas/pam-sessions";
 import {
   TProjectMicrosoftTeamsConfigs,
@@ -1592,6 +1597,11 @@ declare module "knex/types/tables" {
     >;
     [TableName.PamAccount]: KnexOriginal.CompositeTableType<TPamAccounts, TPamAccountsInsert, TPamAccountsUpdate>;
     [TableName.PamSession]: KnexOriginal.CompositeTableType<TPamSessions, TPamSessionsInsert, TPamSessionsUpdate>;
+    [TableName.PamSessionEventBatch]: KnexOriginal.CompositeTableType<
+      TPamSessionEventBatches,
+      TPamSessionEventBatchesInsert,
+      TPamSessionEventBatchesUpdate
+    >;
     [TableName.PamDiscoverySource]: KnexOriginal.CompositeTableType<
       TPamDiscoverySources,
       TPamDiscoverySourcesInsert,

backend/src/db/migrations/20260407000001_pam-session-event-batches.ts

@@ -0,0 +1,30 @@
+import { Knex } from "knex";
+
+import { TableName } from "../schemas";
+import { createOnUpdateTrigger, dropOnUpdateTrigger } from "../utils";
+
+export async function up(knex: Knex): Promise<void> {
+  if (!(await knex.schema.hasTable(TableName.PamSessionEventBatch))) {
+    await knex.schema.createTable(TableName.PamSessionEventBatch, (t) => {
+      t.uuid("id", { primaryKey: true }).defaultTo(knex.fn.uuid());
+
+      t.uuid("sessionId").notNullable();
+      t.foreign("sessionId").references("id").inTable(TableName.PamSession).onDelete("CASCADE");
+      t.index("sessionId");
+
+      t.bigInteger("startOffset").notNullable();
+      t.binary("encryptedEventsBlob").notNullable();
+
+      t.timestamps(true, true, true);
+
+      t.unique(["sessionId", "startOffset"]);
+    });
+
+    await createOnUpdateTrigger(knex, TableName.PamSessionEventBatch);
+  }
+}
+
+export async function down(knex: Knex): Promise<void> {
+  await dropOnUpdateTrigger(knex, TableName.PamSessionEventBatch);
+  await knex.schema.dropTableIfExists(TableName.PamSessionEventBatch);
+}

backend/src/db/schemas/index.ts

@@ -120,6 +120,7 @@ export * from "./pam-folders";
 export * from "./pam-resource-favorites";
 export * from "./pam-resource-rotation-rules";
 export * from "./pam-resources";
+export * from "./pam-session-event-batches";
 export * from "./pam-sessions";
 export * from "./pki-acme-accounts";
 export * from "./pki-acme-auths";

backend/src/db/schemas/models.ts

@@ -227,6 +227,7 @@ export enum TableName {
   PamResource = "pam_resources",
   PamAccount = "pam_accounts",
   PamSession = "pam_sessions",
+  PamSessionEventBatch = "pam_session_event_batches",
   PamDiscoverySource = "pam_discovery_sources",
   PamDiscoverySourceRun = "pam_discovery_source_runs",
   PamDiscoverySourceResource = "pam_discovery_source_resources",

backend/src/db/schemas/pam-session-event-batches.ts

@@ -0,0 +1,25 @@
+// Code generated by automation script, DO NOT EDIT.
+// Automated by pulling database and generating zod schema
+// To update. Just run npm run generate:schema
+// Written by akhilmhdh.
+
+import { z } from "zod";
+
+import { zodBuffer } from "@app/lib/zod";
+
+import { TImmutableDBKeys } from "./models";
+
+export const PamSessionEventBatchesSchema = z.object({
+  id: z.string().uuid(),
+  sessionId: z.string().uuid(),
+  startOffset: z.coerce.number(),
+  encryptedEventsBlob: zodBuffer,
+  createdAt: z.date(),
+  updatedAt: z.date()
+});
+
+export type TPamSessionEventBatches = z.infer<typeof PamSessionEventBatchesSchema>;
+export type TPamSessionEventBatchesInsert = Omit<z.input<typeof PamSessionEventBatchesSchema>, TImmutableDBKeys>;
+export type TPamSessionEventBatchesUpdate = Partial<
+  Omit<z.input<typeof PamSessionEventBatchesSchema>, TImmutableDBKeys>
+>;

backend/src/ee/routes/v1/pam-session-router.ts

@@ -28,6 +28,10 @@ const SessionCredentialsSchema = z.union([
 ]);
 
 export const registerPamSessionRouter = async (server: FastifyZodProvider) => {
+  server.addContentTypeParser("application/octet-stream", { parseAs: "buffer" }, (_req, body, done) => {
+    done(null, body);
+  });
+
   // Meant to be hit solely by gateway identities
   server.route({
     method: "GET",
@@ -320,4 +324,52 @@ export const registerPamSessionRouter = async (server: FastifyZodProvider) => {
       return response;
     }
   });
+
+  // Meant to be hit solely by gateway identities
+  server.route({
+    method: "POST",
+    url: "/:sessionId/event-batches",
+    config: {
+      rateLimit: writeLimit
+    },
+    schema: {
+      description: "Upload a PAM session event batch",
+      params: z.object({
+        sessionId: z.string().uuid()
+      }),
+      querystring: z.object({
+        startOffset: z.coerce.number().int().nonnegative()
+      }),
+      body: z.instanceof(Buffer),
+      response: {
+        200: z.object({ ok: z.literal(true) })
+      }
+    },
+    onRequest: verifyAuth([AuthMode.IDENTITY_ACCESS_TOKEN]),
+    handler: async (req) => {
+      const { projectId } = await server.services.pamSession.uploadEventBatch(
+        {
+          sessionId: req.params.sessionId,
+          startOffset: req.query.startOffset,
+          events: req.body
+        },
+        req.permission
+      );
+
+      await server.services.auditLog.createAuditLog({
+        ...req.auditLogInfo,
+        orgId: req.permission.orgId,
+        projectId,
+        event: {
+          type: EventType.PAM_SESSION_EVENT_BATCH_UPLOAD,
+          metadata: {
+            sessionId: req.params.sessionId,
+            startOffset: req.query.startOffset
+          }
+        }
+      });
+
+      return { ok: true as const };
+    }
+  });
 };

backend/src/ee/services/audit-log/audit-log-types.ts

@@ -596,6 +596,7 @@ export enum EventType {
   PAM_SESSION_TERMINATE = "pam-session-terminate",
   PAM_SESSION_GET = "pam-session-get",
   PAM_SESSION_LIST = "pam-session-list",
+  PAM_SESSION_EVENT_BATCH_UPLOAD = "pam-session-event-batch-upload",
   PAM_FOLDER_CREATE = "pam-folder-create",
   PAM_FOLDER_UPDATE = "pam-folder-update",
   PAM_FOLDER_DELETE = "pam-folder-delete",
@@ -4746,6 +4747,14 @@ interface PamSessionListEvent {
   };
 }
 
+interface PamSessionEventBatchUploadEvent {
+  type: EventType.PAM_SESSION_EVENT_BATCH_UPLOAD;
+  metadata: {
+    sessionId: string;
+    startOffset: number;
+  };
+}
+
 interface PamFolderCreateEvent {
   type: EventType.PAM_FOLDER_CREATE;
   metadata: {
@@ -6152,6 +6161,7 @@ export type Event =
   | PamSessionTerminateEvent
   | PamSessionGetEvent
   | PamSessionListEvent
+  | PamSessionEventBatchUploadEvent
   | PamFolderCreateEvent
   | PamFolderUpdateEvent
   | PamFolderDeleteEvent

backend/src/ee/services/pam-session/pam-session-event-batch-dal.ts

@@ -0,0 +1,27 @@
+import { Knex } from "knex";
+
+import { TDbClient } from "@app/db";
+import { TableName } from "@app/db/schemas";
+import { ormify } from "@app/lib/knex";
+
+export type TPamSessionEventBatchDALFactory = ReturnType<typeof pamSessionEventBatchDALFactory>;
+
+export const pamSessionEventBatchDALFactory = (db: TDbClient) => {
+  const orm = ormify(db, TableName.PamSessionEventBatch);
+
+  const findBySessionId = async (sessionId: string, tx?: Knex) => {
+    return (tx || db.replicaNode())(TableName.PamSessionEventBatch)
+      .where("sessionId", sessionId)
+      .orderBy("startOffset", "asc")
+      .select("*");
+  };
+
+  const upsertBatch = async (sessionId: string, startOffset: number, encryptedEventsBlob: Buffer, tx?: Knex) => {
+    return (tx || db)(TableName.PamSessionEventBatch)
+      .insert({ sessionId, startOffset, encryptedEventsBlob })
+      .onConflict(["sessionId", "startOffset"])
+      .merge(["encryptedEventsBlob"]); // on re-upload of the same offset, overwrite the blob instead of erroring or skipping
+  };
+
+  return { ...orm, findBySessionId, upsertBatch };
+};

backend/src/ee/services/pam-session/pam-session-fns.ts

@@ -1,4 +1,4 @@
-import { TPamSessions } from "@app/db/schemas";
+import { TPamSessionEventBatches, TPamSessions } from "@app/db/schemas";
 import { TKmsServiceFactory } from "@app/services/kms/kms-service";
 import { KmsDataKey } from "@app/services/kms/kms-types";
 
@@ -25,6 +25,25 @@ export const decryptSessionCommandLogs = async ({
   return JSON.parse(decryptedPlainTextBlob.toString()) as (TPamSessionCommandLog | TTerminalEvent)[];
 };
 
+export const decryptBatches = async (
+  batches: TPamSessionEventBatches[],
+  projectId: string,
+  kmsService: Pick<TKmsServiceFactory, "createCipherPairWithDataKey">
+) => {
+  const { decryptor } = await kmsService.createCipherPairWithDataKey({
+    type: KmsDataKey.SecretManager,
+    projectId
+  });
+
+  const events: (TPamSessionCommandLog | TTerminalEvent)[] = [];
+  for (const batch of batches) {
+    const plain = decryptor({ cipherTextBlob: batch.encryptedEventsBlob });
+    const batchEvents = JSON.parse(plain.toString()) as (TPamSessionCommandLog | TTerminalEvent)[];
+    events.push(...batchEvents);
+  }
+  return events;
+};
+
 export const decryptSession = async (
   session: TPamSessions,
   projectId: string,

backend/src/ee/services/pam-session/pam-session-service.ts

@@ -1,4 +1,4 @@
 import { ForbiddenError } from "@casl/ability";
 import net from "net";
 
 import { ActionProjectType, OrganizationActionScope } from "@app/db/schemas";
@@ -18,13 +18,15 @@
 import { PamResource } from "../pam-resource/pam-resource-enums";
 import { OrgPermissionGatewayActions, OrgPermissionSubjects } from "../permission/org-permission";
 import { ProjectPermissionPamSessionActions, ProjectPermissionSub } from "../permission/project-permission";
+import { TPamSessionEventBatchDALFactory } from "./pam-session-event-batch-dal";
 import { TPamSessionDALFactory } from "./pam-session-dal";
 import { PamSessionStatus } from "./pam-session-enums";
-import { decryptSession } from "./pam-session-fns";
-import { TUpdateSessionLogsDTO } from "./pam-session-types";
+import { decryptBatches, decryptSession } from "./pam-session-fns";
+import { TUpdateSessionLogsDTO, TUploadEventBatchDTO } from "./pam-session-types";
 
 type TPamSessionServiceFactoryDep = {
   pamSessionDAL: TPamSessionDALFactory;
+  pamSessionEventBatchDAL: TPamSessionEventBatchDALFactory;
   projectDAL: TProjectDALFactory;
   userDAL: Pick<TUserDALFactory, "findById">;
   permissionService: Pick<TPermissionServiceFactory, "getProjectPermission" | "getOrgPermission">;
@@ -36,6 +38,7 @@
 
 export const pamSessionServiceFactory = ({
   pamSessionDAL,
+  pamSessionEventBatchDAL,
   projectDAL,
   userDAL,
   permissionService,
@@ -91,6 +94,12 @@
       ProjectPermissionSub.PamSessions
     );
 
+    const batches = await pamSessionEventBatchDAL.findBySessionId(sessionId);
+    if (batches.length > 0) {
+      const logs = await decryptBatches(batches, session.projectId, kmsService);
+      return { session: { ...session, logs, encryptedLogsBlob: undefined } as unknown as Awaited<ReturnType<typeof decryptSession>> };
+    }
+
     return {
       session: await decryptSession(session, session.projectId, kmsService)
     };
@@ -288,5 +297,29 @@
     return { session: updatedSession, projectId: project.id, alreadyEnded: false };
   };
 
-  return { getById, list, updateLogsById, endSessionById, terminateSessionById };
+  const uploadEventBatch = async ({ sessionId, startOffset, events }: TUploadEventBatchDTO, actor: OrgServiceActor) => {
+    if (actor.type !== ActorType.IDENTITY) {
+      throw new ForbiddenRequestError({ message: "Only gateways can perform this action" });
+    }
+
+    const session = await pamSessionDAL.findById(sessionId);
+    if (!session) throw new NotFoundError({ message: `Session with ID '${sessionId}' not found` });
+
+    if (session.gatewayIdentityId && session.gatewayIdentityId !== actor.id) {
+      throw new ForbiddenRequestError({ message: "Identity does not have access to upload events for this session" });
+    }
+
+    const { encryptor } = await kmsService.createCipherPairWithDataKey({
+      type: KmsDataKey.SecretManager,
+      projectId: session.projectId
+    });
+
+    const { cipherTextBlob } = encryptor({ plainText: events });
+
+    await pamSessionEventBatchDAL.upsertBatch(sessionId, startOffset, cipherTextBlob);
+
+    return { projectId: session.projectId };
+  };
+
+  return { getById, list, updateLogsById, endSessionById, terminateSessionById, uploadEventBatch };
 };

backend/src/ee/services/pam-session/pam-session-types.ts

@@ -17,3 +17,9 @@ export type TUpdateSessionLogsDTO = {
   sessionId: string;
   logs: (TPamSessionCommandLog | TTerminalEvent | THttpEvent)[];
 };
+
+export type TUploadEventBatchDTO = {
+  sessionId: string;
+  startOffset: number;
+  events: Buffer;
+};

backend/src/server/routes/index.ts

@@ -100,6 +100,7 @@ import { pamResourceRotationRulesDALFactory } from "@app/ee/services/pam-resourc
 import { pamResourceRotationRulesServiceFactory } from "@app/ee/services/pam-resource/pam-resource-rotation-rules-service";
 import { pamResourceServiceFactory } from "@app/ee/services/pam-resource/pam-resource-service";
 import { pamSessionDALFactory } from "@app/ee/services/pam-session/pam-session-dal";
+import { pamSessionEventBatchDALFactory } from "@app/ee/services/pam-session/pam-session-event-batch-dal";
 import { pamSessionServiceFactory } from "@app/ee/services/pam-session/pam-session-service";
 import { pamWebAccessServiceFactory } from "@app/ee/services/pam-web-access/pam-web-access-service";
 import { permissionDALFactory } from "@app/ee/services/permission/permission-dal";
@@ -2787,6 +2788,7 @@ export const registerRoutes = async (
   const pamResourceFavoriteDAL = pamResourceFavoriteDALFactory(db);
   const pamAccountDAL = pamAccountDALFactory(db);
   const pamSessionDAL = pamSessionDALFactory(db);
+  const pamSessionEventBatchDAL = pamSessionEventBatchDALFactory(db);
   const pamDiscoveryRunDAL = pamDiscoveryRunDALFactory(db);
   const pamDiscoverySourceResourcesDAL = pamDiscoverySourceResourcesDALFactory(db);
   const pamDiscoverySourceAccountsDAL = pamDiscoverySourceAccountsDALFactory(db);
@@ -2863,6 +2865,7 @@ export const registerRoutes = async (
 
   const pamSessionService = pamSessionServiceFactory({
     pamSessionDAL,
+    pamSessionEventBatchDAL,
     projectDAL,
     userDAL,
     permissionService,

docs/documentation/platform/pam/architecture.mdx

@@ -57,7 +57,7 @@ A key feature of the Gateway is its ability to act as a "middleman" for all sess
 
 - **Interception**: Because the Gateway sits between the secure tunnel and the target resource, it intercepts all data flowing through the connection.
 - **Logging**: This traffic is logged as part of [Session Recording](/documentation/platform/pam/product-reference/session-recording). The Gateway temporarily stores encrypted session logs locally.
-- **Upload**: Once the session concludes, the logs are securely uploaded to the Infisical platform for storage and review.
+- **Upload**: Every 10 seconds, the Gateway incrementally uploads new session events to the Infisical platform, enabling near real-time monitoring. The upload offset is persisted locally so that uploads resume correctly if the Gateway restarts mid-session.
 
 ## Security Architecture
 

docs/documentation/platform/pam/product-reference/session-recording.mdx

@@ -10,7 +10,7 @@ Infisical PAM provides robust session recording capabilities to help you audit a
 
 When a user initiates a session by accessing an account, a recording of the session begins. The Gateway securely caches all recording data in temporary encrypted files on its local system.
 
-Once the session concludes, the gateway transmits the complete recording to the Infisical platform for long-term, centralized storage. This asynchronous process ensures that sessions remain operational even if the connection to the Infisical platform is temporarily lost. After the upload is complete, administrators can search and review the session logs on the Infisical platform.
+Every 10 seconds, the Gateway incrementally uploads new session events to the Infisical platform, allowing administrators to monitor activity in near real-time without waiting for the session to end. If the connection to the Infisical platform is temporarily lost, the Gateway continues recording locally and resumes uploading once connectivity is restored, ensuring no data is lost. After the session concludes, administrators can search and review the complete session logs on the Infisical platform.
 
 ## What's Captured
 
@@ -54,7 +54,7 @@ You can use the search bar to quickly find relevant information:
     <Accordion title="Are session recordings encrypted?">
         Yes. All session recordings are encrypted at rest by default, ensuring your data is always secure.
     </Accordion>
-    <Accordion title="Why aren't recordings streamed in real-time?">
-        Currently, Infisical uses an asynchronous approach where the gateway records the entire session locally before uploading it. This design makes your PAM sessions more resilient, as they don't depend on a constant, active connection to the Infisical platform. We may introduce live streaming capabilities in a future release.
+    <Accordion title="How quickly do session logs appear on the platform?">
+        Session events are uploaded every 10 seconds, so logs appear on the platform in near real-time during an active session. The session detail page refreshes automatically while a session is active.
     </Accordion>
 </AccordionGroup>