fix: add back s3 writer (#811)

Author: jonathanlab

packages/agent/src/agent.ts

@@ -27,13 +27,16 @@ export class Agent {
       this.posthogAPI = new PostHogAPIClient(config.posthog);
     }
 
-    if (config.otelTransport) {
+    if (config.posthog || config.otelTransport) {
       this.sessionLogWriter = new SessionLogWriter({
-        otelConfig: {
-          posthogHost: config.otelTransport.host,
-          apiKey: config.otelTransport.apiKey,
-          logsPath: config.otelTransport.logsPath,
-        },
+        posthogAPI: this.posthogAPI,
+        otelConfig: config.otelTransport
+          ? {
+              posthogHost: config.otelTransport.host,
+              apiKey: config.otelTransport.apiKey,
+              logsPath: config.otelTransport.logsPath,
+            }
+          : undefined,
         logger: this.logger.child("SessionLogWriter"),
       });
     }

packages/agent/src/session-log-writer.ts

@@ -3,12 +3,15 @@ import {
   OtelLogWriter,
   type SessionContext,
 } from "./otel-log-writer.js";
+import type { PostHogAPIClient } from "./posthog-api.js";
 import type { StoredNotification } from "./types.js";
 import { Logger } from "./utils/logger.js";
 
 export interface SessionLogWriterOptions {
   /** OTEL config for creating writers per session */
   otelConfig?: OtelLogConfig;
+  /** PostHog API client for S3 log persistence */
+  posthogAPI?: PostHogAPIClient;
   /** Logger instance */
   logger?: Logger;
 }
@@ -19,11 +22,15 @@ interface SessionState {
 }
 
 export class SessionLogWriter {
+  private posthogAPI?: PostHogAPIClient;
   private otelConfig?: OtelLogConfig;
+  private pendingEntries: Map<string, StoredNotification[]> = new Map();
+  private flushTimeouts: Map<string, NodeJS.Timeout> = new Map();
   private sessions: Map<string, SessionState> = new Map();
   private logger: Logger;
 
   constructor(options: SessionLogWriterOptions = {}) {
+    this.posthogAPI = options.posthogAPI;
     this.otelConfig = options.otelConfig;
     this.logger =
       options.logger ??
@@ -84,11 +91,6 @@ export class SessionLogWriter {
       return;
     }
 
-    if (!session.otelWriter) {
-      this.logger.debug("No OTEL writer configured for session, skipping log");
-      return;
-    }
-
     try {
       const message = JSON.parse(line);
       const entry: StoredNotification = {
@@ -97,7 +99,16 @@ export class SessionLogWriter {
         notification: message,
       };
 
-      session.otelWriter.emit({ notification: entry });
+      if (session.otelWriter) {
+        session.otelWriter.emit({ notification: entry });
+      }
+
+      if (this.posthogAPI) {
+        const pending = this.pendingEntries.get(sessionId) ?? [];
+        pending.push(entry);
+        this.pendingEntries.set(sessionId, pending);
+        this.scheduleFlush(sessionId);
+      }
     } catch {
       this.logger.warn("Failed to parse raw line for persistence", {
         sessionId,
@@ -108,8 +119,37 @@ export class SessionLogWriter {
 
   async flush(sessionId: string): Promise<void> {
     const session = this.sessions.get(sessionId);
-    if (session?.otelWriter) {
+    if (!session) return;
+
+    if (session.otelWriter) {
       await session.otelWriter.flush();
     }
+
+    const pending = this.pendingEntries.get(sessionId);
+    if (!this.posthogAPI || !pending?.length) return;
+
+    this.pendingEntries.delete(sessionId);
+    const timeout = this.flushTimeouts.get(sessionId);
+    if (timeout) {
+      clearTimeout(timeout);
+      this.flushTimeouts.delete(sessionId);
+    }
+
+    try {
+      await this.posthogAPI.appendTaskRunLog(
+        session.context.taskId,
+        session.context.runId,
+        pending,
+      );
+    } catch (error) {
+      this.logger.error("Failed to persist session logs:", error);
+    }
+  }
+
+  private scheduleFlush(sessionId: string): void {
+    const existing = this.flushTimeouts.get(sessionId);
+    if (existing) clearTimeout(existing);
+    const timeout = setTimeout(() => this.flush(sessionId), 500);
+    this.flushTimeouts.set(sessionId, timeout);
   }
 }