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

Replaces end-of-session bulk upload with incremental 10-second batch
uploads to enable live monitoring and session intervention.

- Add CallUploadPamSessionEventBatch to post raw event bytes to the new
  batch endpoint
- Add sessionUploadState with per-session file offset tracking and mutex
- Add RegisterSession/UnregisterSession for active session management
- Add readFromOffset to read and decrypt new records from a byte offset
- Persist upload progress to a .offset file alongside the .enc recording
  for crash recovery; resume from saved offset on restart
- Flush all active sessions every 10 seconds via a new ticker
- Re-register all on-disk session files at startup to resume after crash
- Fall back to legacy bulk upload if the batch endpoint returns 404
- Remove end-of-session bulk upload call from CleanupPAMSession

Author: bernie-g

packages/api/api.go

@@ -54,6 +54,7 @@ const (
 	operationCallGetPamSessionKey                  = "CallGetPamSessionKey"
 	operationCallUploadPamSessionLog               = "CallUploadPamSessionLog"
 	operationCallPAMSessionTermination             = "CallPAMSessionTermination"
+	operationCallUploadPamSessionEventBatch        = "CallUploadPamSessionEventBatch"
 	operationCallGetMFASessionStatus               = "CallGetMFASessionStatus"
 	operationCallOrgRelayHeartBeat                 = "CallOrgRelayHeartBeat"
 	operationCallInstanceRelayHeartBeat            = "CallInstanceRelayHeartBeat"
@@ -1008,6 +1009,23 @@ func CallUploadPamSessionLogs(httpClient *resty.Client, sessionId string, reques
 	return nil
 }
 
+func CallUploadPamSessionEventBatch(httpClient *resty.Client, sessionId string, startOffset int64, data []byte) error {
+	response, err := httpClient.
+		R().
+		SetHeader("User-Agent", USER_AGENT).
+		SetHeader("Content-Type", "application/octet-stream").
+		SetBody(data).
+		Post(fmt.Sprintf("%v/v1/pam/sessions/%s/event-batches?startOffset=%d", config.INFISICAL_URL, sessionId, startOffset))
+
+	if err != nil {
+		return NewGenericRequestError(operationCallUploadPamSessionEventBatch, err)
+	}
+	if response.IsError() {
+		return NewAPIErrorWithResponse(operationCallUploadPamSessionEventBatch, response, nil)
+	}
+	return nil
+}
+
 func CallPAMSessionTermination(httpClient *resty.Client, sessionId string) error {
 	response, err := httpClient.
 		R().

packages/pam/pam-proxy.go

@@ -151,6 +151,7 @@ func HandlePAMProxy(ctx context.Context, conn *tls.Conn, pamConfig *GatewayPAMCo
 	if err != nil {
 		return fmt.Errorf("failed to create session logger: %w", err)
 	}
+	pamConfig.SessionUploader.RegisterSession(pamConfig.SessionId)
 
 	serverName := credentials.Host
 	if pamConfig.ResourceType == session.ResourceTypeKubernetes {