fix(agent): mark successful runs ok and mirror fatal crashes into telemetry

Addresses three review findings:

- The sandbox never emits task_complete for successful runs (the terminal
  "completed" status is decided by the workflow outside), so the root span
  was ending with unset status on success. The trace builder now marks the
  root span OK when a turn ends cleanly with end_turn; a run error still
  always wins and a turn completion arriving after an error cannot flip the
  status back.
- reportFatalError (uncaught exception / unhandled rejection) marked the run
  failed via the API but never touched telemetry, leaving hard crashes
  invisible in the new pipeline. It now mirrors an error record
  (error_source=agent_server_crash) and shuts telemetry down, ending the root
  span as errored and flushing before the process dies.
- The README "Writing logs" section still described the removed
  otel-log-writer as the preferred full-transcript path to the nonexistent
  /i/v1/agent-logs endpoint; rewritten to match the sink-based metadata
  export.

Generated-By: PostHog Code
Task-Id: 0c511836-2180-455a-9b58-45df2a0661ec

Author: tatoalo

packages/agent/README.md

@@ -217,12 +217,9 @@ Logs serve two purposes: real-time observability and session resume. Every ACP m
 
 ### Writing logs
 
-`SessionLogWriter` (`src/session-log-writer.ts`) is a per-session multiplexer that buffers raw ndJson lines. On flush (auto-scheduled 500ms after writes, or explicit), it dispatches to whichever backend is configured:
+`SessionLogWriter` (`src/session-log-writer.ts`) is a per-session multiplexer that buffers raw ndJson lines. On flush (auto-scheduled 500ms after writes, or explicit), it POSTs batched `StoredNotification` entries to the Django API via `PostHogAPIClient.appendTaskRunLog()`; the API stores them in S3 as the task run's `log_url`. This S3 log is the source of truth for the full transcript and for session resume.
 
-- **OTEL** (`src/otel-log-writer.ts`) — preferred path. Creates an OpenTelemetry `LoggerProvider` per session with resource attributes (`task_id`, `run_id`, `device_type`) set once and indexed via `resource_fingerprint`. Each ndJson line is emitted as an OTEL log record with an `event_type` attribute (the ACP method name) and exported via OTLP HTTP to PostHog's `/i/v1/agent-logs` endpoint. Batch flush interval defaults to 500ms.
-- **Legacy S3** — falls back to `PostHogAPIClient.appendTaskRunLog()`, which POSTs batched `StoredNotification` entries to the Django API. The API stores them as the task run's `log_url`.
-
-Both backends can be active simultaneously — OTEL for fast indexed queries, S3 for full log download.
+Independently of that flush cycle, the writer tees every parsed non-chunk entry to optional `SessionLogSink`s at append time. In cloud, `OtelRunTelemetry` (`src/otel-telemetry.ts`) is wired as a sink and exports an allowlisted metadata subset (run/turn/tool lifecycle, usage, errors — never message content or tool arguments) to PostHog Logs, plus an APM trace per run when configured. See "Optional run telemetry" above for the env vars and behavior. Sink failures can never break S3 log persistence.
 
 ### Resuming from logs
 

packages/agent/src/otel-telemetry.test.ts

@@ -483,7 +483,9 @@ describe("OtelRunTelemetry", () => {
         RUN_ID,
         makeEntry("_posthog/turn_complete", { stopReason: "end_turn" }),
       );
-      telemetry.append(RUN_ID, makeEntry("_posthog/task_complete", {}));
+      // Deliberately no task_complete: production never emits it (the
+      // terminal "completed" status is decided outside the sandbox), so the
+      // root span's OK must come from the clean end_turn above.
     }
 
     it("builds a run trace: root span, turn span, tool span", async () => {
@@ -578,6 +580,12 @@ describe("OtelRunTelemetry", () => {
           error: "gateway exploded",
         }),
       );
+      // A turn completion arriving after the error must not flip the run
+      // back to OK.
+      telemetry.append(
+        RUN_ID,
+        makeEntry("_posthog/turn_complete", { stopReason: "end_turn" }),
+      );
 
       await telemetry.shutdown();
 

packages/agent/src/otel-trace-builder.ts

@@ -41,6 +41,10 @@ export interface RunTraceBuilderConfig {
  * the corresponding log record should be emitted under, so logs and spans
  * cross-link in the UI via trace_id/span_id.
  *
+ * Root span status: OK once a turn ends cleanly (`end_turn`) or on
+ * task_complete, ERROR on a run error (which always wins), unset when the run
+ * ends without either (cancelled / timed out).
+ *
  * Spans carry the same allowlist stance as the log export: lifecycle, status,
  * usage, and identifiers only — never prompts, tool arguments, or output.
  */
@@ -162,6 +166,13 @@ export class RunTraceBuilder {
       typeof params.stopReason === "string" ? params.stopReason : undefined;
     this.closeOpenTools(time);
     this.closeTurn({ stopReason, errored: stopReason === "error" }, time);
+    // The sandbox never emits task_complete for successful runs (the terminal
+    // "completed" status is decided by the workflow outside), so a cleanly
+    // finished turn is the success signal for the run. A later error still
+    // wins via rootErrored/handleError.
+    if (stopReason === "end_turn" && !this.rootErrored) {
+      this.rootSpan.setStatus({ code: SpanStatusCode.OK });
+    }
     return context;
   }
 

packages/agent/src/server/agent-server.ts

@@ -979,6 +979,33 @@ export class AgentServer {
         stopError,
       );
     }
+
+    // Mirror the crash into run telemetry and shut it down (ends the root
+    // span as errored and flushes) - the process is about to die, so nothing
+    // else will get this record out.
+    try {
+      const session = this.session;
+      if (session?.telemetry) {
+        session.telemetry.append(session.payload.run_id, {
+          type: "notification",
+          timestamp: new Date().toISOString(),
+          notification: {
+            jsonrpc: "2.0",
+            method: POSTHOG_NOTIFICATIONS.ERROR,
+            params: {
+              source: "agent_server_crash",
+              error: `Agent server crashed: ${errorMessage}`,
+            },
+          },
+        });
+        await session.telemetry.shutdown();
+      }
+    } catch (telemetryError) {
+      this.logger.error(
+        "Failed to flush telemetry after fatal error",
+        telemetryError,
+      );
+    }
   }
 
   private authenticateRequest(