diff --git a/packages/agent/README.md b/packages/agent/README.md index 6a226fa616..337d6e38fb 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -171,6 +171,16 @@ Required environment variables (validated by zod in `src/server/bin.ts`): - `POSTHOG_PERSONAL_API_KEY` — API key for PostHog requests - `POSTHOG_PROJECT_ID` — numeric project ID +Optional run telemetry (the logs pair must both be set, otherwise telemetry stays off): + +- `POSTHOG_AGENT_OTEL_LOGS_URL` — full OTLP logs URL for run metadata, e.g. `https://us.i.posthog.com/i/v1/logs` +- `POSTHOG_AGENT_OTEL_LOGS_TOKEN` — project API key of the telemetry project +- `POSTHOG_AGENT_OTEL_TRACES_URL` — full OTLP traces URL, e.g. `https://us.i.posthog.com/i/v1/traces`; additionally enables one APM trace per run + +When set, `AgentServer` ships an allowlisted metadata subset of the session log (run/turn/tool lifecycle, usage, error provenance — never message content, tool arguments, or raw error text; see `src/otel-telemetry.ts`) to PostHog Logs, tagged with `service.name=posthog-code-agent` and `run_id`/`task_id`/`team_id`/`user_id`/`distinct_id` resource attributes so cloud runs are filterable per user. With the traces URL set, each run also produces an APM trace (`task_run` root span, a `turn` span per prompt, a `tool_call:` span per tool call; see `src/otel-trace-builder.ts`), and log records carry the matching trace/span ids so Logs and APM cross-link. + +The `task_run` root span is ended and exported at the run's in-process terminal point — a background run's prompt settling, a terminal failure, or session cleanup (`close`). It cannot wait for teardown: agent-server is an exec'd process inside the sandbox, so `docker stop` / Modal terminate kill it without SIGTERM ever arriving, and a span still open at that moment is lost. Interactive sessions that end via hard teardown (e.g. inactivity timeout) therefore lose the root span; turn/tool spans and logs still assemble under the same trace id. + ## Agent SDK The `Agent` class (`src/agent.ts`) is the entrypoint for local/programmatic usage. It handles LLM gateway configuration, log writer setup, and model filtering — then delegates to `createAcpConnection()`. @@ -209,12 +219,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: - -- **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`. +`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. -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, error provenance — never message content, tool arguments, or raw error text) 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 diff --git a/packages/agent/package.json b/packages/agent/package.json index cd94555a5c..fdf66665c4 100644 --- a/packages/agent/package.json +++ b/packages/agent/package.json @@ -163,10 +163,13 @@ "@earendil-works/pi-coding-agent": "catalog:", "@hono/node-server": "^1.19.9", "@openai/codex": "0.144.0", + "@opentelemetry/api": "^1.9.1", "@opentelemetry/api-logs": "^0.208.0", "@opentelemetry/exporter-logs-otlp-http": "^0.208.0", + "@opentelemetry/exporter-trace-otlp-http": "^0.208.0", "@opentelemetry/resources": "^2.0.0", "@opentelemetry/sdk-logs": "^0.208.0", + "@opentelemetry/sdk-trace-base": "^2.8.0", "@opentelemetry/semantic-conventions": "^1.28.0", "@types/jsonwebtoken": "^9.0.10", "commander": "^14.0.2", diff --git a/packages/agent/src/otel-attributes.ts b/packages/agent/src/otel-attributes.ts new file mode 100644 index 0000000000..0aff31796a --- /dev/null +++ b/packages/agent/src/otel-attributes.ts @@ -0,0 +1,84 @@ +const MAX_BODY_CHARS = 2000; +// PostHog Logs only facets attribute key/value pairs shorter than 256 chars, +// so free-text attribute values are capped well below that. +const MAX_ATTR_CHARS = 200; +// The SDK default export timeout is 30s; a hanging endpoint must not hold up +// session cleanup (the sandbox can be torn down right after), so keep it short. +const EXPORT_TIMEOUT_MS = 5000; + +// Batch flush cadence shared by the log and span processors. +const DEFAULT_FLUSH_INTERVAL_MS = 2000; + +export type AttributeValue = string | number | boolean; +export type Attributes = Record; + +export { + DEFAULT_FLUSH_INTERVAL_MS, + EXPORT_TIMEOUT_MS, + MAX_ATTR_CHARS, + MAX_BODY_CHARS, +}; + +/** + * extNotification() can double-prefix custom methods (see matchesExt in + * acp-extensions.ts); normalize so both spellings map identically. + */ +export function normalizeMethod(method: string): string { + return method.startsWith("__posthog/") ? method.slice(1) : method; +} + +export function asString(value: unknown): string | undefined { + return typeof value === "string" ? value : undefined; +} + +export function truncate(value: string, max: number): string { + return value.length <= max ? value : `${value.slice(0, max)}…`; +} + +export function asRecord(value: unknown): Record | undefined { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? (value as Record) + : undefined; +} + +export function strAttr( + attrs: Attributes, + key: string, + value: unknown, + max = MAX_ATTR_CHARS, +): string | undefined { + if (typeof value !== "string" || value.length === 0) return undefined; + const truncated = truncate(value, max); + attrs[key] = truncated; + return truncated; +} + +export function numAttr(attrs: Attributes, key: string, value: unknown): void { + if (typeof value === "number" && Number.isFinite(value)) { + attrs[key] = value; + } +} + +export function usageAttributes(params: Record): Attributes { + const attrs: Attributes = {}; + const used = asRecord(params.used); + if (used) { + numAttr(attrs, "tokens_input", used.inputTokens); + numAttr(attrs, "tokens_output", used.outputTokens); + numAttr(attrs, "tokens_cached_read", used.cachedReadTokens); + numAttr(attrs, "tokens_cached_write", used.cachedWriteTokens); + } + // Claude sends a plain number; other shapes carry { amount }. + const cost = + typeof params.cost === "number" + ? params.cost + : asRecord(params.cost)?.amount; + numAttr(attrs, "cost_usd", cost); + return attrs; +} + +/** Timestamp of a stored entry as a Date, falling back to now when invalid. */ +export function entryTime(timestamp: string): Date { + const parsed = new Date(timestamp); + return Number.isNaN(parsed.getTime()) ? new Date() : parsed; +} diff --git a/packages/agent/src/otel-log-writer.test.ts b/packages/agent/src/otel-log-writer.test.ts deleted file mode 100644 index 29fc78f035..0000000000 --- a/packages/agent/src/otel-log-writer.test.ts +++ /dev/null @@ -1,105 +0,0 @@ -import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { OtelLogWriter } from "./otel-log-writer"; -import type { StoredNotification } from "./types"; - -// Mock the OTEL exporter -const mockExport = vi.fn((_logs, callback) => { - callback({ code: 0 }); // Success -}); - -vi.mock("@opentelemetry/exporter-logs-otlp-http", () => ({ - OTLPLogExporter: class { - export = mockExport; - shutdown = vi.fn().mockResolvedValue(undefined); - }, -})); - -describe("OtelLogWriter", () => { - let writer: OtelLogWriter; - - beforeEach(() => { - mockExport.mockClear(); - // Session context (taskId, runId) is now passed in constructor as resource attributes - writer = new OtelLogWriter( - { - posthogHost: "https://us.i.posthog.com", - apiKey: "phc_test_key", - flushIntervalMs: 100, - }, - { - taskId: "task-123", - runId: "run-456", - }, - ); - }); - - afterEach(async () => { - await writer.shutdown(); - }); - - it("should emit a log entry with event_type as regular attribute", async () => { - const notification: StoredNotification = { - type: "notification", - timestamp: new Date().toISOString(), - notification: { - jsonrpc: "2.0", - method: "_posthog/test_event", - params: { foo: "bar" }, - }, - }; - - // taskId and runId are now resource attributes set in constructor, - // only notification is passed per-emit - writer.emit({ notification }); - - // Force flush to trigger export - await writer.flush(); - - // Verify export was called - expect(mockExport).toHaveBeenCalled(); - - // Get the logs that were exported - const exportedLogs = mockExport.mock.calls[0][0]; - expect(exportedLogs.length).toBe(1); - - const log = exportedLogs[0]; - // task_id and run_id are now resource attributes, not regular attributes - expect(log.attributes.task_id).toBeUndefined(); - expect(log.attributes.run_id).toBeUndefined(); - // event_type is still a regular attribute (varies per log entry) - expect(log.attributes.event_type).toBe("_posthog/test_event"); - expect(log.body).toBe(JSON.stringify(notification)); - - // Verify resource attributes contain task_id and run_id - expect(log.resource.attributes.task_id).toBe("task-123"); - expect(log.resource.attributes.run_id).toBe("run-456"); - expect(log.resource.attributes["service.name"]).toBe("posthog-code-agent"); - }); - - it("should batch multiple log entries", async () => { - const makeNotification = (method: string): StoredNotification => ({ - type: "notification", - timestamp: new Date().toISOString(), - notification: { - jsonrpc: "2.0", - method, - }, - }); - - writer.emit({ notification: makeNotification("event_1") }); - writer.emit({ notification: makeNotification("event_2") }); - writer.emit({ notification: makeNotification("event_3") }); - - await writer.flush(); - - expect(mockExport).toHaveBeenCalled(); - const exportedLogs = mockExport.mock.calls[0][0]; - expect(exportedLogs.length).toBe(3); - - // All logs should share the same resource attributes - for (const log of exportedLogs) { - expect(log.resource.attributes.task_id).toBe("task-123"); - expect(log.resource.attributes.run_id).toBe("run-456"); - } - }); -}); diff --git a/packages/agent/src/otel-log-writer.ts b/packages/agent/src/otel-log-writer.ts deleted file mode 100644 index 83099749ae..0000000000 --- a/packages/agent/src/otel-log-writer.ts +++ /dev/null @@ -1,94 +0,0 @@ -import { SeverityNumber } from "@opentelemetry/api-logs"; -import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-http"; -import { resourceFromAttributes } from "@opentelemetry/resources"; -import { - BatchLogRecordProcessor, - LoggerProvider, -} from "@opentelemetry/sdk-logs"; -import { ATTR_SERVICE_NAME } from "@opentelemetry/semantic-conventions"; -import type { StoredNotification } from "./types"; -import type { Logger } from "./utils/logger"; - -export interface OtelLogConfig { - /** PostHog ingest host, e.g., "https://us.i.posthog.com" */ - posthogHost: string; - /** Project API key, e.g., "phc_xxx" */ - apiKey: string; - /** Batch flush interval in ms (default: 500) */ - flushIntervalMs?: number; - /** Override the logs endpoint path (default: /i/v1/agent-logs) */ - logsPath?: string; -} - -/** - * Session context for resource attributes. - * These are set once per OTEL logger instance and indexed via resource_fingerprint - */ -export interface SessionContext { - /** Parent task grouping - all runs for a task share this */ - taskId: string; - /** Primary conversation identifier - all events in a run share this */ - runId: string; - /** Deployment environment - "local" for desktop, "cloud" for cloud sandbox */ - deviceType?: "local" | "cloud"; -} - -export class OtelLogWriter { - private loggerProvider: LoggerProvider; - private logger: ReturnType; - - constructor( - config: OtelLogConfig, - sessionContext: SessionContext, - _debugLogger?: Logger, - ) { - const logsPath = config.logsPath ?? "/i/v1/agent-logs"; - const exporter = new OTLPLogExporter({ - url: `${config.posthogHost}${logsPath}`, - headers: { Authorization: `Bearer ${config.apiKey}` }, - }); - - const processor = new BatchLogRecordProcessor(exporter, { - scheduledDelayMillis: config.flushIntervalMs ?? 500, - }); - - // Resource attributes are set ONCE per session and indexed via resource_fingerprint - // So we have fast queries by run_id/task_id in PostHog Logs UI - this.loggerProvider = new LoggerProvider({ - resource: resourceFromAttributes({ - [ATTR_SERVICE_NAME]: "posthog-code-agent", - run_id: sessionContext.runId, - task_id: sessionContext.taskId, - device_type: sessionContext.deviceType ?? "local", - }), - processors: [processor], - }); - - this.logger = this.loggerProvider.getLogger("agent-session"); - } - - /** - * Emit an agent event to PostHog Logs via OTEL. - */ - emit(entry: { notification: StoredNotification }): void { - const { notification } = entry; - const eventType = notification.notification.method; - - this.logger.emit({ - severityNumber: SeverityNumber.INFO, - severityText: "INFO", - body: JSON.stringify(notification), - attributes: { - event_type: eventType, - }, - }); - } - - async flush(): Promise { - await this.loggerProvider.forceFlush(); - } - - async shutdown(): Promise { - await this.loggerProvider.shutdown(); - } -} diff --git a/packages/agent/src/otel-telemetry.test.ts b/packages/agent/src/otel-telemetry.test.ts new file mode 100644 index 0000000000..9b761521b7 --- /dev/null +++ b/packages/agent/src/otel-telemetry.test.ts @@ -0,0 +1,701 @@ +import { SpanKind, SpanStatusCode } from "@opentelemetry/api"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { mapNotificationToLogRecord, OtelRunTelemetry } from "./otel-telemetry"; +import type { StoredNotification } from "./types"; + +const mockLogExport = vi.fn((_logs, callback) => { + callback({ code: 0 }); // Success +}); + +const mockSpanExport = vi.fn((_spans, callback) => { + callback({ code: 0 }); // Success +}); + +const mockLogShutdown = vi.fn(() => Promise.resolve()); +const mockSpanShutdown = vi.fn(() => Promise.resolve()); + +vi.mock("@opentelemetry/exporter-logs-otlp-http", () => ({ + OTLPLogExporter: class { + export = mockLogExport; + shutdown = mockLogShutdown; + }, +})); + +vi.mock("@opentelemetry/exporter-trace-otlp-http", () => ({ + OTLPTraceExporter: class { + export = mockSpanExport; + shutdown = mockSpanShutdown; + }, +})); + +const RUN_ID = "run-456"; + +const RESOURCE = { + taskId: "task-123", + runId: RUN_ID, + deviceType: "cloud" as const, + teamId: 42, + userId: 7, + distinctId: "distinct-1", + adapter: "claude", + mode: "background", + agentVersion: "1.2.3", +}; + +function makeEntry( + method: string, + params?: Record, +): StoredNotification { + return { + type: "notification", + timestamp: "2026-07-06T12:00:00.000Z", + notification: { jsonrpc: "2.0", method, params }, + }; +} + +function sessionUpdate(update: Record): StoredNotification { + return makeEntry("session/update", { update }); +} + +interface ExportedLog { + body: string; + attributes: Record; + resource: { attributes: Record }; + spanContext?: { traceId: string; spanId: string }; +} + +interface ExportedSpan { + name: string; + kind: number; + status: { code: number; message?: string }; + attributes: Record; + parentSpanContext?: { spanId: string }; + spanContext: () => { traceId: string; spanId: string }; +} + +function exportedLogs(): ExportedLog[] { + return mockLogExport.mock.calls.flatMap((call) => call[0]); +} + +function exportedSpans(): ExportedSpan[] { + return mockSpanExport.mock.calls.flatMap((call) => call[0]); +} + +function spanByName(name: string): ExportedSpan { + const span = exportedSpans().find((s) => s.name === name); + expect(span, `span ${name} should be exported`).toBeDefined(); + return span as ExportedSpan; +} + +describe("OtelRunTelemetry", () => { + beforeEach(() => { + mockLogExport.mockClear(); + mockSpanExport.mockClear(); + mockLogShutdown.mockClear(); + mockSpanShutdown.mockClear(); + }); + + describe("mapNotificationToLogRecord", () => { + it.each([ + { + name: "run_started", + entry: makeEntry("_posthog/run_started", { + agentVersion: "1.0.0", + sessionId: "acp-1", + }), + body: "run started", + attrs: { + event_type: "_posthog/run_started", + agent_version: "1.0.0", + session_id: "acp-1", + }, + }, + { + name: "double-prefixed extension method", + entry: makeEntry("__posthog/run_started", {}), + body: "run started", + attrs: { event_type: "_posthog/run_started" }, + }, + { + name: "sdk_session", + entry: makeEntry("_posthog/sdk_session", { + adapter: "claude", + sessionId: "acp-1", + }), + body: "sdk session created (claude)", + attrs: { adapter: "claude", session_id: "acp-1" }, + }, + { + name: "usage_update with numeric cost", + entry: makeEntry("_posthog/usage_update", { + used: { + inputTokens: 100, + outputTokens: 20, + cachedReadTokens: 5, + cachedWriteTokens: 2, + }, + cost: 0.42, + }), + body: "usage update", + attrs: { + tokens_input: 100, + tokens_output: 20, + tokens_cached_read: 5, + tokens_cached_write: 2, + cost_usd: 0.42, + }, + }, + { + name: "usage_update with cost object", + entry: makeEntry("_posthog/usage_update", { + cost: { amount: 1.5, currency: "USD" }, + }), + body: "usage update", + attrs: { cost_usd: 1.5 }, + }, + { + name: "turn_complete", + entry: makeEntry("_posthog/turn_complete", { stopReason: "end_turn" }), + body: "turn complete (end_turn)", + attrs: { stop_reason: "end_turn" }, + }, + { + name: "task_complete", + entry: makeEntry("_posthog/task_complete", {}), + body: "task complete", + attrs: { event_type: "_posthog/task_complete" }, + }, + { + name: "error", + entry: makeEntry("_posthog/error", { + source: "agent_server", + stopReason: "error", + error: "boom", + }), + severityText: "ERROR", + body: "run error", + attrs: { error_source: "agent_server", stop_reason: "error" }, + }, + { + name: "progress", + entry: makeEntry("_posthog/progress", { + group: "setup:run-1", + step: "agent", + status: "completed", + label: "Started agent", + }), + body: "progress: agent completed (Started agent)", + attrs: { + progress_group: "setup:run-1", + progress_step: "agent", + progress_status: "completed", + }, + }, + { + name: "git_checkpoint", + entry: makeEntry("_posthog/git_checkpoint", { + branch: "posthog-code/fix", + }), + body: "git checkpoint", + attrs: { branch: "posthog-code/fix" }, + }, + { + name: "branch_created", + entry: makeEntry("_posthog/branch_created", { branch: "b1" }), + body: "branch created", + attrs: { branch: "b1" }, + }, + { + name: "permission_request without tool content", + entry: makeEntry("_posthog/permission_request", { + requestId: "r1", + toolCallId: "t1", + toolCall: { title: "rm -rf /" }, + }), + body: "permission request", + attrs: { request_id: "r1", tool_call_id: "t1" }, + }, + { + name: "tool_call start", + entry: sessionUpdate({ + sessionUpdate: "tool_call", + toolCallId: "t1", + kind: "execute", + status: "pending", + title: "Run tests", + }), + body: "tool call started (execute)", + attrs: { + session_update_type: "tool_call", + tool_call_id: "t1", + tool_kind: "execute", + tool_status: "pending", + }, + }, + { + name: "terminal tool_call_update completed", + entry: sessionUpdate({ + sessionUpdate: "tool_call_update", + toolCallId: "t1", + status: "completed", + }), + body: "tool call completed", + attrs: { tool_call_id: "t1", tool_status: "completed" }, + }, + { + name: "terminal tool_call_update failed", + entry: sessionUpdate({ + sessionUpdate: "tool_call_update", + toolCallId: "t1", + status: "failed", + }), + severityText: "WARN", + body: "tool call failed", + }, + ])("maps $name", ({ entry, severityText = "INFO", body, attrs = {} }) => { + const mapped = mapNotificationToLogRecord(entry); + + expect(mapped).not.toBeNull(); + expect(mapped?.severityText).toBe(severityText); + expect(mapped?.body).toBe(body); + expect(mapped?.attributes).toMatchObject(attrs); + }); + + // Content-bearing notifications must stay in the session log: exporting + // them would ship customer prompts and repo content to the telemetry + // project, and in-progress tool snapshots would multiply billed bytes. + it.each([ + [ + "agent_message", + sessionUpdate({ + sessionUpdate: "agent_message", + content: { type: "text", text: "SECRET" }, + }), + ], + [ + "agent_message_chunk", + sessionUpdate({ + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "SECRET" }, + }), + ], + [ + "agent_thought_chunk", + sessionUpdate({ + sessionUpdate: "agent_thought_chunk", + content: { type: "text", text: "SECRET" }, + }), + ], + [ + "in-progress tool_call_update", + sessionUpdate({ + sessionUpdate: "tool_call_update", + toolCallId: "t1", + status: "in_progress", + rawInput: { command: "SECRET" }, + }), + ], + [ + "available_commands_update", + sessionUpdate({ + sessionUpdate: "available_commands_update", + availableCommands: [], + }), + ], + [ + "user prompt request", + makeEntry("session/prompt", { + prompt: [{ type: "text", text: "SECRET" }], + }), + ], + [ + "user_message", + makeEntry("_posthog/user_message", { message: "SECRET" }), + ], + // Console lines are free-text agent-server diagnostics; some interpolate + // content (e.g. the prompt preview logged on user_message handling). + [ + "console with interpolated prompt preview", + makeEntry("_posthog/console", { + level: "debug", + message: "Processing user message (detectedPrUrl=none): SECRET...", + }), + ], + [ + "console error", + makeEntry("_posthog/console", { level: "error", message: "SECRET" }), + ], + ["unknown extension method", makeEntry("_posthog/some_new_event", {})], + ])("drops %s", (_name, entry) => { + expect(mapNotificationToLogRecord(entry)).toBeNull(); + }); + + // Run errors export provenance only: the raw message is free text that + // can embed prompt or repo content (exception paths, provider errors). + it("never exports the raw error message", () => { + const mapped = mapNotificationToLogRecord( + makeEntry("_posthog/error", { + source: "agent_server_crash", + error: "Agent server crashed: ENOENT open '/repos/acme/SECRET/.env'", + }), + ); + + expect(mapped).not.toBeNull(); + expect(JSON.stringify([mapped?.body, mapped?.attributes])).not.toContain( + "SECRET", + ); + }); + + it("caps body length", () => { + const mapped = mapNotificationToLogRecord( + makeEntry("_posthog/progress", { + group: "setup:run-1", + step: "agent", + status: "completed", + label: "x".repeat(5000), + }), + ); + + expect(mapped?.body.length).toBeLessThanOrEqual(2001); + }); + }); + + describe("logs export", () => { + let telemetry: OtelRunTelemetry; + + beforeEach(() => { + telemetry = new OtelRunTelemetry( + { + url: "https://us.i.posthog.com/i/v1/logs", + token: "phc_test_key", + flushIntervalMs: 100, + }, + RESOURCE, + ); + }); + + afterEach(async () => { + await telemetry.shutdown(); + }); + + it("pins run identity as resource attributes so runs are filterable per user", async () => { + telemetry.append(RUN_ID, makeEntry("_posthog/run_started", {})); + + await telemetry.flush(); + + const record = exportedLogs()[0]; + expect(record.resource.attributes).toMatchObject({ + "service.name": "posthog-code-agent", + "service.version": "1.2.3", + run_id: RUN_ID, + task_id: "task-123", + team_id: "42", + user_id: "7", + distinct_id: "distinct-1", + device_type: "cloud", + adapter: "claude", + run_mode: "background", + }); + // Without a traces URL, no spans are built and logs carry no trace ids. + expect(mockSpanExport).not.toHaveBeenCalled(); + expect(record.spanContext).toBeUndefined(); + }); + + it("never exports tool arguments, titles, or output content", async () => { + telemetry.append( + RUN_ID, + sessionUpdate({ + sessionUpdate: "tool_call", + toolCallId: "t1", + kind: "execute", + status: "pending", + title: "bash: cat .env", + rawInput: { command: "cat .env" }, + }), + ); + telemetry.append( + RUN_ID, + sessionUpdate({ + sessionUpdate: "tool_call_update", + toolCallId: "t1", + status: "completed", + rawOutput: { stdout: "AWS_KEY=leaked" }, + }), + ); + + await telemetry.flush(); + + const records = exportedLogs(); + expect(records).toHaveLength(2); + const surface = JSON.stringify( + records.map((record) => [record.body, record.attributes]), + ); + expect(surface).not.toContain(".env"); + expect(surface).not.toContain("leaked"); + }); + + it("ignores entries for other sessions", async () => { + telemetry.append("other-run", makeEntry("_posthog/run_started", {})); + + await telemetry.flush(); + + expect(mockLogExport).not.toHaveBeenCalled(); + }); + + it("swallows malformed entries instead of throwing", () => { + const junk = { + type: "notification", + timestamp: "not-a-date", + notification: { jsonrpc: "2.0", method: 123 }, + } as unknown as StoredNotification; + + expect(() => telemetry.append(RUN_ID, junk)).not.toThrow(); + }); + }); + + describe("trace export", () => { + let telemetry: OtelRunTelemetry; + + beforeEach(() => { + telemetry = new OtelRunTelemetry( + { + url: "https://us.i.posthog.com/i/v1/logs", + token: "phc_test_key", + tracesUrl: "https://us.i.posthog.com/i/v1/traces", + flushIntervalMs: 100, + }, + RESOURCE, + ); + }); + + function driveSuccessfulRun(): void { + telemetry.append(RUN_ID, makeEntry("_posthog/run_started", {})); + telemetry.append( + RUN_ID, + makeEntry("session/prompt", { + prompt: [{ type: "text", text: "SECRET" }], + }), + ); + telemetry.append( + RUN_ID, + sessionUpdate({ + sessionUpdate: "tool_call", + toolCallId: "t1", + kind: "execute", + status: "pending", + title: "bash: cat .env", + }), + ); + telemetry.append( + RUN_ID, + sessionUpdate({ + sessionUpdate: "tool_call_update", + toolCallId: "t1", + status: "completed", + }), + ); + telemetry.append( + RUN_ID, + makeEntry("_posthog/usage_update", { + used: { inputTokens: 10, outputTokens: 5 }, + cost: 0.1, + }), + ); + telemetry.append( + RUN_ID, + makeEntry("_posthog/turn_complete", { stopReason: "end_turn" }), + ); + // 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 () => { + driveSuccessfulRun(); + + await telemetry.shutdown(); + + const root = spanByName("task_run"); + const turn = spanByName("turn"); + const tool = spanByName("tool_call:execute"); + + expect(root.kind).toBe(SpanKind.SERVER); + expect(root.parentSpanContext).toBeUndefined(); + expect(turn.parentSpanContext?.spanId).toBe(root.spanContext().spanId); + expect(tool.parentSpanContext?.spanId).toBe(turn.spanContext().spanId); + + const traceId = root.spanContext().traceId; + expect(turn.spanContext().traceId).toBe(traceId); + expect(tool.spanContext().traceId).toBe(traceId); + + expect(root.status.code).toBe(SpanStatusCode.OK); + expect(turn.status.code).toBe(SpanStatusCode.OK); + expect(tool.status.code).toBe(SpanStatusCode.OK); + + expect(turn.attributes).toMatchObject({ + turn_index: 1, + stop_reason: "end_turn", + tokens_input: 10, + tokens_output: 5, + cost_usd: 0.1, + }); + expect(tool.attributes).toMatchObject({ + tool_kind: "execute", + tool_call_id: "t1", + tool_status: "completed", + }); + + // Same allowlist stance as logs: no prompt or tool content on spans. + const surface = JSON.stringify( + exportedSpans().map((span) => [span.name, span.attributes]), + ); + expect(surface).not.toContain("SECRET"); + expect(surface).not.toContain(".env"); + }); + + it("does not leave root OK when a later turn ends non-clean", async () => { + driveSuccessfulRun(); + // Second turn gets cancelled: the latest outcome wins, so the earlier + // clean turn must not leave the run marked OK. + telemetry.append( + RUN_ID, + makeEntry("session/prompt", { + prompt: [{ type: "text", text: "again" }], + }), + ); + telemetry.append( + RUN_ID, + makeEntry("_posthog/turn_complete", { stopReason: "cancelled" }), + ); + + await telemetry.shutdown(); + + expect(spanByName("task_run").status.code).toBe(SpanStatusCode.UNSET); + expect(exportedSpans().filter((s) => s.name === "turn")).toHaveLength(2); + }); + + it("stamps log records with the span they belong to", async () => { + driveSuccessfulRun(); + + await telemetry.shutdown(); + + const root = spanByName("task_run"); + const tool = spanByName("tool_call:execute"); + const logs = exportedLogs(); + + const runStartedLog = logs.find((log) => log.body === "run started"); + const toolStartedLog = logs.find((log) => + log.body.startsWith("tool call started"), + ); + expect(runStartedLog?.spanContext?.spanId).toBe( + root.spanContext().spanId, + ); + expect(toolStartedLog?.spanContext?.spanId).toBe( + tool.spanContext().spanId, + ); + for (const log of logs) { + expect(log.spanContext?.traceId).toBe(root.spanContext().traceId); + } + }); + + it("marks failed tools, interrupted tools, errored turns, and the errored run", async () => { + telemetry.append(RUN_ID, makeEntry("session/prompt", {})); + telemetry.append( + RUN_ID, + sessionUpdate({ + sessionUpdate: "tool_call", + toolCallId: "t1", + kind: "execute", + }), + ); + telemetry.append( + RUN_ID, + sessionUpdate({ + sessionUpdate: "tool_call_update", + toolCallId: "t1", + status: "failed", + }), + ); + // Still open when the run error below lands. + telemetry.append( + RUN_ID, + sessionUpdate({ + sessionUpdate: "tool_call", + toolCallId: "t2", + kind: "fetch", + }), + ); + telemetry.append( + RUN_ID, + makeEntry("_posthog/error", { + source: "agent_server", + stopReason: "error", + error: "gateway exploded reading SECRET", + }), + ); + // 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(); + + expect(spanByName("tool_call:execute").status.code).toBe( + SpanStatusCode.ERROR, + ); + const interrupted = spanByName("tool_call:fetch"); + expect(interrupted.status.code).toBe(SpanStatusCode.ERROR); + expect(interrupted.attributes).toMatchObject({ + tool_status: "interrupted", + }); + const turn = spanByName("turn"); + expect(turn.status.code).toBe(SpanStatusCode.ERROR); + expect(turn.attributes).toMatchObject({ stop_reason: "error" }); + const root = spanByName("task_run"); + expect(root.status.code).toBe(SpanStatusCode.ERROR); + expect(root.attributes).toMatchObject({ error_source: "agent_server" }); + // The raw error message is free text; it must not reach span status + // messages or attributes. + const surface = JSON.stringify( + exportedSpans().map((span) => [ + span.name, + span.attributes, + span.status, + ]), + ); + expect(surface).not.toContain("SECRET"); + }); + + it("shuts down logs even when the traces endpoint fails", async () => { + mockSpanShutdown.mockRejectedValueOnce(new Error("traces endpoint down")); + telemetry.append(RUN_ID, makeEntry("_posthog/run_started", {})); + + await expect(telemetry.shutdown()).resolves.toBeUndefined(); + + expect(mockLogShutdown).toHaveBeenCalled(); + expect(exportedLogs().map((log) => log.body)).toContain("run started"); + }); + + it("exports open spans on shutdown even without terminal events", async () => { + telemetry.append(RUN_ID, makeEntry("session/prompt", {})); + telemetry.append( + RUN_ID, + sessionUpdate({ + sessionUpdate: "tool_call", + toolCallId: "t1", + kind: "read", + }), + ); + + await telemetry.shutdown(); + + expect( + exportedSpans() + .map((span) => span.name) + .sort(), + ).toEqual(["task_run", "tool_call:read", "turn"]); + }); + }); +}); diff --git a/packages/agent/src/otel-telemetry.ts b/packages/agent/src/otel-telemetry.ts new file mode 100644 index 0000000000..8d7456ad04 --- /dev/null +++ b/packages/agent/src/otel-telemetry.ts @@ -0,0 +1,367 @@ +import { SeverityNumber } from "@opentelemetry/api-logs"; +import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-http"; +import { resourceFromAttributes } from "@opentelemetry/resources"; +import { + BatchLogRecordProcessor, + LoggerProvider, +} from "@opentelemetry/sdk-logs"; +import { + ATTR_SERVICE_NAME, + ATTR_SERVICE_VERSION, +} from "@opentelemetry/semantic-conventions"; +import { POSTHOG_NOTIFICATIONS } from "./acp-extensions"; +import { + type Attributes, + asRecord, + asString, + DEFAULT_FLUSH_INTERVAL_MS, + EXPORT_TIMEOUT_MS, + entryTime, + MAX_BODY_CHARS, + normalizeMethod, + strAttr, + truncate, + usageAttributes, +} from "./otel-attributes"; +import { RunTraceBuilder } from "./otel-trace-builder"; +import type { SessionLogSink } from "./session-log-writer"; +import type { StoredNotification } from "./types"; +import type { Logger } from "./utils/logger"; + +const SERVICE_NAME = "posthog-code-agent"; + +export interface OtelTelemetryConfig { + /** Full OTLP logs endpoint URL, e.g. "https://us.i.posthog.com/i/v1/logs" */ + url: string; + /** Project API key sent as a Bearer token */ + token: string; + /** Full OTLP traces endpoint URL; spans are off when unset */ + tracesUrl?: string; + /** Batch flush interval in ms (default: 2000) */ + flushIntervalMs?: number; +} + +/** + * Session identity pinned as OTel resource attributes. Resource attributes are + * indexed via resource_fingerprint in PostHog Logs, so runs are directly + * filterable per user/task/run in the Logs UI. + */ +export interface OtelSessionResource { + /** Parent task grouping - all runs for a task share this */ + taskId: string; + /** Primary conversation identifier - all events in a run share this */ + runId: string; + /** Deployment environment - "local" for desktop, "cloud" for cloud sandbox */ + deviceType: "local" | "cloud"; + teamId?: number; + userId?: number; + distinctId?: string; + /** Runtime adapter: "claude" or "codex" */ + adapter?: string; + /** Run mode: "interactive" or "background" */ + mode?: string; + agentVersion?: string; +} + +export interface MappedLogRecord { + severityNumber: SeverityNumber; + severityText: string; + body: string; + attributes: Attributes; +} + +function record( + severity: [SeverityNumber, string], + body: string, + eventType: string, + attributes: Attributes = {}, +): MappedLogRecord { + return { + severityNumber: severity[0], + severityText: severity[1], + body: truncate(body, MAX_BODY_CHARS), + attributes: { event_type: eventType, ...attributes }, + }; +} + +const INFO: [SeverityNumber, string] = [SeverityNumber.INFO, "INFO"]; +const WARN: [SeverityNumber, string] = [SeverityNumber.WARN, "WARN"]; +const ERROR: [SeverityNumber, string] = [SeverityNumber.ERROR, "ERROR"]; + +function mapSessionUpdate( + method: string, + params: Record, +): MappedLogRecord | null { + const update = asRecord(params.update); + const updateType = update?.sessionUpdate; + if (!update || typeof updateType !== "string") return null; + + switch (updateType) { + case "tool_call": { + const attrs: Attributes = { session_update_type: updateType }; + strAttr(attrs, "tool_call_id", update.toolCallId); + const kind = strAttr(attrs, "tool_kind", update.kind); + strAttr(attrs, "tool_status", update.status ?? "pending"); + return record( + INFO, + `tool call started${kind ? ` (${kind})` : ""}`, + method, + attrs, + ); + } + case "tool_call_update": { + const status = update.status; + // In-progress snapshots re-send the growing tool input/output; only the + // terminal transition is run metadata. + if (status !== "completed" && status !== "failed") return null; + const attrs: Attributes = { session_update_type: updateType }; + strAttr(attrs, "tool_call_id", update.toolCallId); + strAttr(attrs, "tool_status", status); + return record( + status === "failed" ? WARN : INFO, + `tool call ${status}`, + method, + attrs, + ); + } + case "usage_update": + return record(INFO, "usage update", method, { + session_update_type: updateType, + ...usageAttributes(update), + }); + default: + return null; + } +} + +/** + * Maps a stored session notification to an exportable log record, or null for + * notification types that must not leave the sandbox. + * + * Allowlist by design: agent message/thought text, tool arguments, and tool + * output stay in the session log (the product source of truth) - only + * run-lifecycle metadata is exported, so prompts and repo content never reach + * the telemetry project. + */ +export function mapNotificationToLogRecord( + entry: StoredNotification, +): MappedLogRecord | null { + const rawMethod = entry.notification.method; + if (typeof rawMethod !== "string") return null; + const method = normalizeMethod(rawMethod); + const params = asRecord(entry.notification.params) ?? {}; + + if (method === "session/update") { + return mapSessionUpdate(method, params); + } + + switch (method) { + case POSTHOG_NOTIFICATIONS.RUN_STARTED: { + const attrs: Attributes = {}; + strAttr(attrs, "agent_version", params.agentVersion); + strAttr(attrs, "session_id", params.sessionId); + return record(INFO, "run started", method, attrs); + } + case POSTHOG_NOTIFICATIONS.SDK_SESSION: { + const attrs: Attributes = {}; + const adapter = strAttr(attrs, "adapter", params.adapter); + strAttr(attrs, "session_id", params.sessionId); + return record( + INFO, + `sdk session created${adapter ? ` (${adapter})` : ""}`, + method, + attrs, + ); + } + case POSTHOG_NOTIFICATIONS.USAGE_UPDATE: + return record(INFO, "usage update", method, usageAttributes(params)); + case POSTHOG_NOTIFICATIONS.TURN_COMPLETE: { + const attrs: Attributes = {}; + const stopReason = strAttr(attrs, "stop_reason", params.stopReason); + return record( + INFO, + `turn complete${stopReason ? ` (${stopReason})` : ""}`, + method, + attrs, + ); + } + case POSTHOG_NOTIFICATIONS.TASK_COMPLETE: { + const attrs: Attributes = {}; + strAttr(attrs, "stop_reason", params.stopReason); + return record(INFO, "task complete", method, attrs); + } + case POSTHOG_NOTIFICATIONS.ERROR: { + // params.error is free text that can embed prompt or repo content + // (exception messages, provider errors), so only its provenance is + // exported. The raw message stays in the session log and on the task + // run's error_message. + const attrs: Attributes = {}; + strAttr(attrs, "error_source", params.source); + strAttr(attrs, "stop_reason", params.stopReason); + return record(ERROR, "run error", method, attrs); + } + // POSTHOG_NOTIFICATIONS.CONSOLE is deliberately NOT exported: those are + // free-text agent-server diagnostics that interpolate arbitrary data + // (prompt previews, stringified extension params), so shipping them would + // leak content the allowlist exists to keep in the sandbox. They remain + // in the S3 session log and the event-ingest stream. + case POSTHOG_NOTIFICATIONS.PROGRESS: { + const attrs: Attributes = {}; + strAttr(attrs, "progress_group", params.group); + const step = strAttr(attrs, "progress_step", params.step); + const status = strAttr(attrs, "progress_status", params.status); + const label = asString(params.label); + return record( + INFO, + `progress: ${step ?? "step"} ${status ?? ""}${label ? ` (${label})` : ""}`.trim(), + method, + attrs, + ); + } + case POSTHOG_NOTIFICATIONS.GIT_CHECKPOINT: { + const attrs: Attributes = {}; + strAttr(attrs, "branch", params.branch); + return record(INFO, "git checkpoint", method, attrs); + } + case POSTHOG_NOTIFICATIONS.BRANCH_CREATED: { + const attrs: Attributes = {}; + strAttr(attrs, "branch", params.branch ?? params.branchName); + return record(INFO, "branch created", method, attrs); + } + case POSTHOG_NOTIFICATIONS.MODE_CHANGE: { + const attrs: Attributes = {}; + strAttr(attrs, "run_mode", params.mode); + return record(INFO, "mode change", method, attrs); + } + case POSTHOG_NOTIFICATIONS.COMPACT_BOUNDARY: + return record(INFO, "compact boundary", method, {}); + case POSTHOG_NOTIFICATIONS.PERMISSION_REQUEST: + case POSTHOG_NOTIFICATIONS.PERMISSION_RESPONSE: + case POSTHOG_NOTIFICATIONS.PERMISSION_RESOLVED: { + // params.toolCall/options carry tool content; export identifiers only. + const attrs: Attributes = {}; + strAttr(attrs, "request_id", params.requestId); + strAttr(attrs, "tool_call_id", params.toolCallId); + const action = method.slice(method.indexOf("/") + 1).replace("_", " "); + return record(INFO, action, method, attrs); + } + default: + return null; + } +} + +/** + * Ships run telemetry to PostHog over OTLP: an allowlisted metadata subset of + * the session log to PostHog Logs (see mapNotificationToLogRecord) and, when a + * traces URL is configured, an APM trace per run (see RunTraceBuilder) with + * trace/span ids stamped on the log records so the two cross-link in the UI. + * Registered as a SessionLogWriter sink; the S3 session log remains the source + * of truth for the full transcript. + */ +export class OtelRunTelemetry implements SessionLogSink { + private loggerProvider: LoggerProvider; + private otelLogger: ReturnType; + private traceBuilder?: RunTraceBuilder; + private runId: string; + private debugLogger?: Logger; + private shutdownStarted = false; + + constructor( + config: OtelTelemetryConfig, + resource: OtelSessionResource, + debugLogger?: Logger, + ) { + this.runId = resource.runId; + this.debugLogger = debugLogger; + + const exporter = new OTLPLogExporter({ + url: config.url, + headers: { Authorization: `Bearer ${config.token}` }, + }); + + const processor = new BatchLogRecordProcessor(exporter, { + scheduledDelayMillis: config.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS, + exportTimeoutMillis: EXPORT_TIMEOUT_MS, + }); + + const resourceAttributes: Attributes = { + [ATTR_SERVICE_NAME]: SERVICE_NAME, + }; + strAttr(resourceAttributes, ATTR_SERVICE_VERSION, resource.agentVersion); + strAttr(resourceAttributes, "run_id", resource.runId); + strAttr(resourceAttributes, "task_id", resource.taskId); + strAttr(resourceAttributes, "device_type", resource.deviceType); + strAttr(resourceAttributes, "team_id", resource.teamId?.toString()); + strAttr(resourceAttributes, "user_id", resource.userId?.toString()); + strAttr(resourceAttributes, "distinct_id", resource.distinctId); + strAttr(resourceAttributes, "adapter", resource.adapter); + strAttr(resourceAttributes, "run_mode", resource.mode); + + const otelResource = resourceFromAttributes(resourceAttributes); + + this.loggerProvider = new LoggerProvider({ + resource: otelResource, + processors: [processor], + }); + + this.otelLogger = this.loggerProvider.getLogger("agent-session"); + + if (config.tracesUrl) { + this.traceBuilder = new RunTraceBuilder( + { + url: config.tracesUrl, + token: config.token, + flushIntervalMs: config.flushIntervalMs, + }, + otelResource, + ); + } + } + + append(sessionId: string, entry: StoredNotification): void { + // Resource attributes pin this writer to one run; ignore entries for any + // other session so records are never mislabeled. + if (sessionId !== this.runId || this.shutdownStarted) return; + try { + // The span state machine must see every entry: e.g. session/prompt is a + // turn boundary even though it never becomes a log record. + const context = this.traceBuilder?.handle(entry); + const mapped = mapNotificationToLogRecord(entry); + if (!mapped) return; + this.otelLogger.emit({ + ...mapped, + timestamp: entryTime(entry.timestamp), + context, + }); + } catch (error) { + // Telemetry must never interfere with the run. + this.debugLogger?.debug("Failed to emit OTel telemetry", { error }); + } + } + + /** + * Best-effort: logs and traces flush independently, so a rejecting or + * hanging traces endpoint can never block or fail the log flush (and vice + * versa). Never rejects. + */ + async flush(): Promise { + await Promise.allSettled([ + this.loggerProvider.forceFlush(), + this.traceBuilder?.flush(), + ]); + } + + /** + * Ends open spans, flushes batched records, then stops the providers. + * Idempotent and best-effort: the two providers shut down independently + * and a failure in one never skips the other. Never rejects. + */ + async shutdown(): Promise { + if (this.shutdownStarted) return; + this.shutdownStarted = true; + await Promise.allSettled([ + this.traceBuilder?.shutdown(), + this.loggerProvider.shutdown(), + ]); + } +} diff --git a/packages/agent/src/otel-trace-builder.ts b/packages/agent/src/otel-trace-builder.ts new file mode 100644 index 0000000000..1d1091d066 --- /dev/null +++ b/packages/agent/src/otel-trace-builder.ts @@ -0,0 +1,310 @@ +import { + type Context, + ROOT_CONTEXT, + type Span, + SpanKind, + SpanStatusCode, + type Tracer, + trace, +} from "@opentelemetry/api"; +import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http"; +import type { Resource } from "@opentelemetry/resources"; +import { + BasicTracerProvider, + BatchSpanProcessor, +} from "@opentelemetry/sdk-trace-base"; +import { POSTHOG_NOTIFICATIONS } from "./acp-extensions"; +import { + type Attributes, + asRecord, + asString, + DEFAULT_FLUSH_INTERVAL_MS, + EXPORT_TIMEOUT_MS, + entryTime, + normalizeMethod, + strAttr, + usageAttributes, +} from "./otel-attributes"; +import type { StoredNotification } from "./types"; + +export interface RunTraceBuilderConfig { + /** Full OTLP traces endpoint URL, e.g. "https://us.i.posthog.com/i/v1/traces" */ + url: string; + /** Project API key sent as a Bearer token */ + token: string; + /** Batch flush interval in ms (default: 2000) */ + flushIntervalMs?: number; +} + +/** + * Builds one APM trace per run from the session notification stream: a root + * `task_run` span, a child `turn` span per prompt/turn, and a child + * `tool_call:` span per tool call. `handle()` returns the OTel context + * 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 is resolved at shutdown from the LATEST turn outcome (the + * root span only exports when it ends, so earlier turns must not leave a + * sticky OK): OK when the last turn ended cleanly (`end_turn`, or an explicit + * task_complete), ERROR on a run error (which always wins) or a last turn + * that stopped with `error`, unset otherwise (cancelled / refused / timed out + * / no completed turns). + * + * Spans carry the same allowlist stance as the log export: lifecycle, status, + * usage, and identifiers only — never prompts, tool arguments, or output. + */ +export class RunTraceBuilder { + private provider: BasicTracerProvider; + private tracer: Tracer; + private rootSpan: Span; + private rootContext: Context; + private rootErrored = false; + /** stopReason of the most recent completed turn; drives root status at shutdown */ + private lastStopReason?: string; + private turnSpan?: Span; + private turnContext?: Context; + private turnIndex = 0; + private toolSpans = new Map(); + private ended = false; + + constructor(config: RunTraceBuilderConfig, resource: Resource) { + const exporter = new OTLPTraceExporter({ + url: config.url, + headers: { Authorization: `Bearer ${config.token}` }, + }); + + this.provider = new BasicTracerProvider({ + resource, + spanProcessors: [ + new BatchSpanProcessor(exporter, { + scheduledDelayMillis: + config.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS, + exportTimeoutMillis: EXPORT_TIMEOUT_MS, + }), + ], + }); + + this.tracer = this.provider.getTracer("agent-session"); + this.rootSpan = this.tracer.startSpan("task_run", { + kind: SpanKind.SERVER, + }); + this.rootContext = trace.setSpan(ROOT_CONTEXT, this.rootSpan); + } + + /** + * Advances the span state machine with one stored entry and returns the + * context the entry's log record (if any) should attach to. + */ + handle(entry: StoredNotification): Context { + if (this.ended) return this.rootContext; + const rawMethod = entry.notification.method; + if (typeof rawMethod !== "string") return this.currentContext(); + const method = normalizeMethod(rawMethod); + const params = asRecord(entry.notification.params) ?? {}; + const time = entryTime(entry.timestamp); + + switch (method) { + // The ACP prompt request is what starts a turn; its content never leaves + // the sandbox — it is only a turn boundary marker here. + case "session/prompt": + return this.startTurn(time); + case POSTHOG_NOTIFICATIONS.TURN_COMPLETE: + return this.endTurn(params, time); + case POSTHOG_NOTIFICATIONS.USAGE_UPDATE: { + (this.turnSpan ?? this.rootSpan).setAttributes(usageAttributes(params)); + return this.currentContext(); + } + case POSTHOG_NOTIFICATIONS.TASK_COMPLETE: + // Explicit success signal (forward-compat; production decides the + // terminal status outside the sandbox) — treated as a clean outcome. + this.lastStopReason = "end_turn"; + return this.rootContext; + case POSTHOG_NOTIFICATIONS.ERROR: + return this.handleError(params, time); + case "session/update": + return this.handleSessionUpdate(params, time); + default: + return this.currentContext(); + } + } + + async flush(): Promise { + await this.provider.forceFlush(); + } + + /** + * Ends any open spans (status unset), resolves the root status from the + * latest turn outcome, then flushes and stops the provider. + */ + async shutdown(): Promise { + if (this.ended) return; + this.ended = true; + const now = new Date(); + this.closeOpenTools(now); + this.closeTurn(undefined, now); + if (!this.rootErrored) { + if (this.lastStopReason === "end_turn") { + this.rootSpan.setStatus({ code: SpanStatusCode.OK }); + } else if (this.lastStopReason === "error") { + this.rootSpan.setStatus({ code: SpanStatusCode.ERROR }); + } + // Any other latest outcome (cancelled, refusal, max_tokens, none) + // leaves the status unset: neither success nor failure. + } + this.rootSpan.end(now); + await this.provider.shutdown(); + } + + private currentContext(): Context { + return this.turnContext ?? this.rootContext; + } + + private startTurn(time: Date): Context { + // A new prompt while a turn is still open means we missed its completion; + // close it without a status rather than nesting turns. + this.closeOpenTools(time); + this.closeTurn(undefined, time); + this.turnIndex += 1; + this.turnSpan = this.tracer.startSpan( + "turn", + { + kind: SpanKind.INTERNAL, + startTime: time, + attributes: { turn_index: this.turnIndex }, + }, + this.rootContext, + ); + this.turnContext = trace.setSpan(this.rootContext, this.turnSpan); + return this.turnContext; + } + + private endTurn(params: Record, time: Date): Context { + const context = this.turnContext ?? this.rootContext; + const stopReason = asString(params.stopReason); + 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 the latest + // turn outcome is the run's success signal — recorded here, resolved into + // the root span status at shutdown so an early clean turn can't leave a + // stale OK on a run whose last turn was cancelled. + this.lastStopReason = stopReason; + return context; + } + + private closeTurn( + end: { stopReason?: string; errored?: boolean } | undefined, + time: Date, + ): void { + if (!this.turnSpan) return; + if (end?.stopReason) { + this.turnSpan.setAttribute("stop_reason", end.stopReason); + } + if (end) { + this.turnSpan.setStatus({ + code: end.errored ? SpanStatusCode.ERROR : SpanStatusCode.OK, + }); + } + this.turnSpan.end(time); + this.turnSpan = undefined; + this.turnContext = undefined; + } + + private handleSessionUpdate( + params: Record, + time: Date, + ): Context { + const update = asRecord(params.update); + const updateType = update?.sessionUpdate; + if (!update || typeof updateType !== "string") { + return this.currentContext(); + } + if (updateType === "tool_call") return this.startTool(update, time); + if (updateType === "tool_call_update") { + return this.handleToolUpdate(update, time); + } + if (updateType === "usage_update") { + (this.turnSpan ?? this.rootSpan).setAttributes(usageAttributes(update)); + } + return this.currentContext(); + } + + private startTool(update: Record, time: Date): Context { + const toolCallId = asString(update.toolCallId); + const existing = toolCallId ? this.toolSpans.get(toolCallId) : undefined; + if (existing) return existing.context; + + const kind = asString(update.kind) ?? "unknown"; + const parentContext = this.turnContext ?? this.rootContext; + const attributes: Attributes = { tool_kind: kind }; + if (toolCallId) attributes.tool_call_id = toolCallId; + + const span = this.tracer.startSpan( + // Kind (read/edit/execute/...) is a small enum, so per-kind span names + // stay low-cardinality and make APM latency breakdowns useful. + `tool_call:${kind}`, + { kind: SpanKind.INTERNAL, startTime: time, attributes }, + parentContext, + ); + const context = trace.setSpan(parentContext, span); + if (toolCallId) { + this.toolSpans.set(toolCallId, { span, context }); + } else { + // Without an id there is no terminal update to match; record a marker. + span.end(time); + } + return context; + } + + private handleToolUpdate( + update: Record, + time: Date, + ): Context { + const toolCallId = asString(update.toolCallId); + const open = toolCallId ? this.toolSpans.get(toolCallId) : undefined; + if (!open || !toolCallId) return this.currentContext(); + + const status = update.status; + if (status === "completed" || status === "failed") { + open.span.setAttribute("tool_status", status); + open.span.setStatus({ + code: status === "failed" ? SpanStatusCode.ERROR : SpanStatusCode.OK, + }); + open.span.end(time); + this.toolSpans.delete(toolCallId); + } + return open.context; + } + + private handleError(params: Record, time: Date): Context { + this.rootErrored = true; + this.closeOpenTools(time, { interrupted: true }); + const stopReason = asString(params.stopReason); + this.closeTurn({ stopReason, errored: true }, time); + // params.error is free text that can embed prompt or repo content, so + // only the error's provenance is exported; the raw message stays in the + // session log and on the task run's error_message. + const attrs: Attributes = {}; + strAttr(attrs, "error_source", params.source); + this.rootSpan.setAttributes(attrs); + this.rootSpan.setStatus({ code: SpanStatusCode.ERROR }); + return this.rootContext; + } + + /** + * Ends every open tool span. Interrupted (a run error aborted the tool + * mid-flight) marks them errored so APM doesn't show a healthy-looking + * active tool under a failed run; otherwise the outcome is unknown and the + * status stays unset. + */ + private closeOpenTools(time: Date, opts?: { interrupted?: boolean }): void { + for (const { span } of this.toolSpans.values()) { + if (opts?.interrupted) { + span.setAttribute("tool_status", "interrupted"); + span.setStatus({ code: SpanStatusCode.ERROR }); + } + span.end(time); + } + this.toolSpans.clear(); + } +} diff --git a/packages/agent/src/server/agent-server.test.ts b/packages/agent/src/server/agent-server.test.ts index 99e0c67a9d..49cd5ca9d2 100644 --- a/packages/agent/src/server/agent-server.test.ts +++ b/packages/agent/src/server/agent-server.test.ts @@ -952,6 +952,124 @@ describe("AgentServer HTTP Mode", () => { ); }); + // Sandbox teardown kills the exec'd agent-server without SIGTERM, so the + // trace's root span only exports if telemetry is shut down at the run's + // in-process terminal points. + it("shuts down telemetry after mirroring the terminal failure record", async () => { + const order: string[] = []; + const testServer = new AgentServer({ + port, + jwtPublicKey: TEST_PUBLIC_KEY, + repositoryPath: repo.path, + apiUrl: "http://localhost:8000", + apiKey: "test-api-key", + projectId: 1, + mode: "interactive", + taskId: "test-task-id", + runId: "test-run-id", + }) as unknown as { + session: { + payload: { run_id: string }; + logWriter: { flush: ReturnType }; + telemetry: { + append: ReturnType; + shutdown: ReturnType; + }; + }; + eventStreamSender: { + enqueue: (event: Record) => void; + stop: () => Promise; + }; + posthogAPI: { + updateTaskRun: ( + taskId: string, + runId: string, + payload: Record, + ) => Promise; + }; + signalTaskComplete( + payload: JwtPayload, + stopReason: string, + errorMessage?: string, + ): Promise; + }; + testServer.eventStreamSender = { + enqueue: vi.fn(), + stop: vi.fn(async () => {}), + }; + testServer.posthogAPI = { + updateTaskRun: vi.fn(async () => ({})), + }; + testServer.session = { + payload: { run_id: "run-1" }, + logWriter: { flush: vi.fn(async () => {}) }, + telemetry: { + append: vi.fn(() => { + order.push("append"); + }), + shutdown: vi.fn(async () => { + order.push("shutdown"); + }), + }, + }; + + await testServer.signalTaskComplete( + { + run_id: "run-1", + task_id: "task-1", + team_id: 1, + user_id: 1, + distinct_id: "distinct-id", + mode: "background", + }, + "error", + "boom", + ); + + // The error mirror must land before shutdown so the root span exports + // with ERROR status. + expect(order).toEqual(["append", "shutdown"]); + }); + + it.each([ + { mode: "background" as const, shutdownCalls: 1 }, + { mode: "interactive" as const, shutdownCalls: 0 }, + ])( + "finalizeRunTelemetry shuts down telemetry only for $mode runs", + async ({ mode, shutdownCalls }) => { + const testServer = new AgentServer({ + port, + jwtPublicKey: TEST_PUBLIC_KEY, + repositoryPath: repo.path, + apiUrl: "http://localhost:8000", + apiKey: "test-api-key", + projectId: 1, + mode: "interactive", + taskId: "test-task-id", + runId: "test-run-id", + }) as unknown as { + session: { telemetry: { shutdown: ReturnType } }; + finalizeRunTelemetry(payload: JwtPayload): Promise; + }; + testServer.session = { + telemetry: { shutdown: vi.fn(async () => {}) }, + }; + + await testServer.finalizeRunTelemetry({ + run_id: "run-1", + task_id: "task-1", + team_id: 1, + user_id: 1, + distinct_id: "distinct-id", + mode, + }); + + expect(testServer.session.telemetry.shutdown).toHaveBeenCalledTimes( + shutdownCalls, + ); + }, + ); + function createFailureTestServer() { const appendRawLine = vi.fn(); const testServer = new AgentServer({ diff --git a/packages/agent/src/server/agent-server.ts b/packages/agent/src/server/agent-server.ts index 3a1853f7b9..224f42fa16 100644 --- a/packages/agent/src/server/agent-server.ts +++ b/packages/agent/src/server/agent-server.ts @@ -55,6 +55,7 @@ import { import type { PermissionMode } from "../execution-mode"; import { DEFAULT_CODEX_MODEL, fetchGatewayModels } from "../gateway-models"; import { HandoffCheckpointTracker } from "../handoff-checkpoint"; +import { OtelRunTelemetry } from "../otel-telemetry"; import { configurePersistentAgentState } from "../persistent-agent-state"; import { PostHogAPIClient } from "../posthog-api"; import { @@ -277,6 +278,8 @@ interface ActiveSession { sseController: SseController | null; deviceInfo: DeviceInfo; logWriter: SessionLogWriter; + /** Ships run telemetry (logs + spans) to PostHog; unset when the sandbox has no OTLP config */ + telemetry?: OtelRunTelemetry; /** Current permission mode, tracked for relay decisions */ permissionMode: PermissionMode; /** Whether a desktop client has ever connected via SSE during this session */ @@ -517,6 +520,47 @@ export class AgentServer { return payload.mode ?? this.config.mode; } + /** + * Ships run telemetry to PostHog when the sandbox provides an OTLP endpoint + * + token (POSTHOG_AGENT_OTEL_LOGS_URL/_TOKEN): metadata log records, plus an + * APM trace per run when POSTHOG_AGENT_OTEL_TRACES_URL is also set. Resource + * attributes carry the run/user identifiers so cloud runs are filterable per + * user, task, and run in the Logs UI. Returns undefined when unconfigured or + * on failure — telemetry must never block session startup. + */ + private createRunTelemetry( + payload: JwtPayload, + deviceInfo: DeviceInfo, + adapter: "claude" | "codex", + ): OtelRunTelemetry | undefined { + const { otelLogsUrl, otelLogsToken } = this.config; + if (!otelLogsUrl || !otelLogsToken) return undefined; + try { + return new OtelRunTelemetry( + { + url: otelLogsUrl, + token: otelLogsToken, + tracesUrl: this.config.otelTracesUrl, + }, + { + taskId: payload.task_id, + runId: payload.run_id, + deviceType: deviceInfo.type, + teamId: payload.team_id, + userId: payload.user_id, + distinctId: payload.distinct_id, + adapter, + mode: this.getEffectiveMode(payload), + agentVersion: this.config.version ?? packageJson.version, + }, + new Logger({ debug: false, prefix: "[OtelRunTelemetry]" }), + ); + } catch (error) { + this.logger.warn("Failed to initialize OTel run telemetry", error); + return undefined; + } + } + private getSessionPermissionMode(): PermissionMode { if (this.session?.permissionMode) { return this.session.permissionMode; @@ -939,6 +983,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( @@ -1523,9 +1594,16 @@ export class AgentServer { userAgent: `posthog/cloud.hog.dev; version: ${this.config.version ?? packageJson.version}`, }); + const telemetry = this.createRunTelemetry( + payload, + deviceInfo, + runtimeAdapter, + ); + const logWriter = new SessionLogWriter({ posthogAPI, logger: new Logger({ debug: true, prefix: "[SessionLogWriter]" }), + sinks: telemetry ? [telemetry] : undefined, }); const acpConnection = createAcpConnection({ @@ -1733,6 +1811,7 @@ export class AgentServer { sseController, deviceInfo, logWriter, + telemetry, permissionMode: initialPermissionMode, hasDesktopConnected: sseController !== null, pendingHandoffGitState: undefined, @@ -2090,6 +2169,8 @@ export class AgentServer { if (result.stopReason === "end_turn") { await this.relayAgentResponse(payload); } + + await this.finalizeRunTelemetry(payload); } catch (error) { this.logger.error("Failed to send initial task message", error); if (this.session) { @@ -2312,6 +2393,8 @@ export class AgentServer { if (result.stopReason === "end_turn") { await this.relayAgentResponse(payload); } + + await this.finalizeRunTelemetry(payload); } catch (error) { this.logger.error(`Failed to send ${logLabel.toLowerCase()}`, error); if (this.session) { @@ -3655,6 +3738,25 @@ ${signedCommitInstructions}${prLinkInstructions}${shellEfficiencyInstructions} } } + /** + * Ends the run's telemetry (root span + final flush) at the in-sandbox + * terminal point of a background run. Sandbox teardown cannot be relied on + * for this: agent-server is an exec'd process inside the sandbox, so + * `docker stop` signals only the container's PID 1 and Modal terminate is + * immediate — the SIGTERM handler (and thus cleanupSession) never runs, and + * an unended root span would never export. Once the background prompt + * settles the run is over in-sandbox; the workflow marks the terminal + * status and destroys the sandbox right after. + */ + private async finalizeRunTelemetry(payload: JwtPayload): Promise { + if (this.getEffectiveMode(payload) !== "background") return; + try { + await this.session?.telemetry?.shutdown(); + } catch (error) { + this.logger.debug("Failed to finalize run telemetry", error); + } + } + private async signalTaskComplete( payload: JwtPayload, stopReason: string, @@ -3700,6 +3802,11 @@ ${signedCommitInstructions}${prLinkInstructions}${shellEfficiencyInstructions} } finally { await this.emitRtkSavings(); await this.eventStreamSender?.stop(); + // The run is terminal and the sandbox is torn down right after — and + // teardown kills this exec'd process without SIGTERM, so this is the + // last chance to end the root span and drain the OTel queues. The + // error mirror was appended above, so the root span exports as ERROR. + await this.session?.telemetry?.shutdown().catch(() => {}); } } @@ -3709,15 +3816,20 @@ ${signedCommitInstructions}${prLinkInstructions}${shellEfficiencyInstructions} | typeof POSTHOG_NOTIFICATIONS.ERROR, params: Record, ): void { - this.eventStreamSender?.enqueue({ - type: "notification", + const entry = { + type: "notification" as const, timestamp: new Date().toISOString(), notification: { - jsonrpc: "2.0", + jsonrpc: "2.0" as const, method, params, }, - }); + }; + this.eventStreamSender?.enqueue(entry); + // Terminal events bypass the SessionLogWriter (and its sinks), so mirror + // them onto the OTel writer directly — a failed run is exactly what the + // telemetry must record. + this.session?.telemetry?.append(this.session.payload.run_id, entry); } private configureEnvironment({ @@ -4365,6 +4477,15 @@ ${signedCommitInstructions}${prLinkInstructions}${shellEfficiencyInstructions} this.mcpRelayServer = null; } + // Shutdown ends open spans and flushes batched records; without it, + // sandbox teardown races the OTel batch delay and drops the tail of the + // run's telemetry. + try { + await this.session.telemetry?.shutdown(); + } catch (error) { + this.logger.error("Failed to shut down OTel run telemetry", error); + } + // Drain pending permissions before ACP cleanup to avoid deadlocks — // cleanup may await operations that are blocked on a permission response. for (const [, pending] of this.pendingPermissions) { diff --git a/packages/agent/src/server/bin.ts b/packages/agent/src/server/bin.ts index 1bdfef445e..6b5ac2e188 100644 --- a/packages/agent/src/server/bin.ts +++ b/packages/agent/src/server/bin.ts @@ -54,6 +54,11 @@ const envSchema = z.object({ .enum(["true", "false"]) .transform((value) => value === "true") .optional(), + // OTLP pair for shipping run metadata to PostHog Logs; telemetry stays off + // unless both are set. The traces URL additionally enables APM spans. + POSTHOG_AGENT_OTEL_LOGS_URL: z.url().optional(), + POSTHOG_AGENT_OTEL_LOGS_TOKEN: z.string().min(1).optional(), + POSTHOG_AGENT_OTEL_TRACES_URL: z.url().optional(), }); const program = new Command(); @@ -165,6 +170,13 @@ program const env = envResult.data; + // The telemetry token is only ever consumed here (into the server config); + // drop it from the process environment so tool subprocesses spawned by the + // agent don't inherit it and `env` dumps in persisted transcripts or PR + // bodies can't leak it. Defense in depth, not a boundary: same-UID + // processes can still read the container's initial env via /proc. + delete process.env.POSTHOG_AGENT_OTEL_LOGS_TOKEN; + const mode = options.mode === "background" ? "background" : "interactive"; const createPr = parseBooleanOption(options.createPr, "--createPr"); const autoPublish = parseBooleanOption( @@ -225,6 +237,9 @@ program env.POSTHOG_TASK_RUN_EVENT_INGEST_STREAM_WINDOW_MS, eventIngestKeepStreamOpen: env.POSTHOG_TASK_RUN_EVENT_INGEST_KEEP_STREAM_OPEN, + otelLogsUrl: env.POSTHOG_AGENT_OTEL_LOGS_URL, + otelLogsToken: env.POSTHOG_AGENT_OTEL_LOGS_TOKEN, + otelTracesUrl: env.POSTHOG_AGENT_OTEL_TRACES_URL, repositoryPath: options.repositoryPath, repoReadyFile: options.repoReadyFile, apiUrl: env.POSTHOG_API_URL, diff --git a/packages/agent/src/server/types.ts b/packages/agent/src/server/types.ts index 34b94a45e1..dc2471dedb 100644 --- a/packages/agent/src/server/types.ts +++ b/packages/agent/src/server/types.ts @@ -24,6 +24,12 @@ export interface AgentServerConfig { eventIngestBaseUrl?: string; eventIngestStreamWindowMs?: number; eventIngestKeepStreamOpen?: boolean; + /** Full OTLP logs URL for run telemetry, e.g. https://us.i.posthog.com/i/v1/logs */ + otelLogsUrl?: string; + /** Project API key for the OTLP logs/traces endpoints */ + otelLogsToken?: string; + /** Full OTLP traces URL for run spans, e.g. https://us.i.posthog.com/i/v1/traces */ + otelTracesUrl?: string; mode: AgentMode; taskId: string; runId: string; diff --git a/packages/agent/src/session-log-writer.test.ts b/packages/agent/src/session-log-writer.test.ts index efb2b91092..2a492b58ac 100644 --- a/packages/agent/src/session-log-writer.test.ts +++ b/packages/agent/src/session-log-writer.test.ts @@ -104,6 +104,52 @@ describe("SessionLogWriter", () => { }); }); + describe("sinks", () => { + it("delivers non-chunk entries to sinks and keeps persistence when a sink throws", async () => { + const goodSink = { append: vi.fn() }; + const badSink = { + append: vi.fn(() => { + throw new Error("sink down"); + }), + }; + const writer = new SessionLogWriter({ + posthogAPI: mockPosthogAPI, + sinks: [badSink, goodSink], + }); + writer.register("run-1", { taskId: "task-1", runId: "run-1" }); + + writer.appendRawLine( + "run-1", + makeSessionUpdate("agent_message_chunk", { + content: { type: "text", text: "chunk" }, + }), + ); + writer.appendRawLine( + "run-1", + JSON.stringify({ + jsonrpc: "2.0", + method: "_posthog/turn_complete", + params: { stopReason: "end_turn" }, + }), + ); + + // Buffered message chunks never reach sinks; the non-chunk entry does, + // even after the sink listed first has thrown. + expect(goodSink.append).toHaveBeenCalledTimes(1); + expect(goodSink.append.mock.calls[0][0]).toBe("run-1"); + expect(goodSink.append.mock.calls[0][1].notification.method).toBe( + "_posthog/turn_complete", + ); + + await writer.flush("run-1"); + + const persistedMethods = mockAppendLog.mock.calls + .flatMap((call) => call[2]) + .map((entry: StoredNotification) => entry.notification.method); + expect(persistedMethods).toContain("_posthog/turn_complete"); + }); + }); + describe("agent_message_chunk coalescing", () => { it("coalesces consecutive chunks into a single agent_message", async () => { const sessionId = "s1"; diff --git a/packages/agent/src/session-log-writer.ts b/packages/agent/src/session-log-writer.ts index 7941b8487e..a11566902c 100644 --- a/packages/agent/src/session-log-writer.ts +++ b/packages/agent/src/session-log-writer.ts @@ -2,12 +2,35 @@ import fs from "node:fs"; import fsp from "node:fs/promises"; import path from "node:path"; import { serializeError } from "@posthog/shared"; -import type { SessionContext } from "./otel-log-writer"; import type { PostHogAPIClient } from "./posthog-api"; import type { StoredNotification } from "./types"; import { isEmptyContentBlock } from "./utils/acp-content"; import { Logger } from "./utils/logger"; +/** + * Session context for a registered session. + * These are set once per session and shared by every persisted entry. + */ +export interface SessionContext { + /** Parent task grouping - all runs for a task share this */ + taskId: string; + /** Primary conversation identifier - all events in a run share this */ + runId: string; + /** Deployment environment - "local" for desktop, "cloud" for cloud sandbox */ + deviceType?: "local" | "cloud"; +} + +/** + * Receives every parsed non-chunk notification the writer persists, in + * arrival order. Streamed agent message/thought chunks are not delivered + * (neither raw nor coalesced) - sinks carry run metadata, not transcript + * content. Sink failures are swallowed so they can never break session log + * persistence. + */ +export interface SessionLogSink { + append(sessionId: string, entry: StoredNotification): void; +} + export interface SessionLogWriterOptions { /** PostHog API client for log persistence */ posthogAPI?: PostHogAPIClient; @@ -15,6 +38,8 @@ export interface SessionLogWriterOptions { logger?: Logger; /** Local cache path for instant log loading (e.g., ~/.posthog-code) */ localCachePath?: string; + /** Additional consumers of persisted entries (e.g. OTel telemetry) */ + sinks?: SessionLogSink[]; } interface ChunkBuffer { @@ -72,6 +97,8 @@ export class SessionLogWriter { private retryCounts: Map = new Map(); private sessions: Map = new Map(); private flushQueues: Map> = new Map(); + private sinks: SessionLogSink[]; + private warnedSinks: Set = new Set(); private logger: Logger; private localCachePath?: string; @@ -79,6 +106,7 @@ export class SessionLogWriter { constructor(options: SessionLogWriterOptions = {}) { this.posthogAPI = options.posthogAPI; this.localCachePath = options.localCachePath; + this.sinks = options.sinks ?? []; this.logger = options.logger ?? new Logger({ debug: false, prefix: "[SessionLogWriter]" }); @@ -187,6 +215,8 @@ export class SessionLogWriter { notification: message, }; + this.emitToSinks(sessionId, entry); + // Coalesce the local cache: buffer in-progress tool_call_update // snapshots (they re-send the full growing output) and write one merged // update per toolCallId. Written on a terminal update, any non-tool @@ -338,6 +368,24 @@ export class SessionLogWriter { } } + private emitToSinks(sessionId: string, entry: StoredNotification): void { + for (const sink of this.sinks) { + try { + sink.append(sessionId, entry); + } catch (error) { + // Warn once per sink: a broken sink at streaming rate would otherwise + // flood the console without ever affecting persistence. + if (!this.warnedSinks.has(sink)) { + this.warnedSinks.add(sink); + this.logger.warn( + "Session log sink failed; suppressing further errors from this sink", + { error: serializeError(error) }, + ); + } + } + } + } + private getUpdate( message: Record, ): Record | undefined { diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index c8d6a0ae0b..8bbffe5e70 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -77,6 +77,12 @@ export type OnLogCallback = ( data?: unknown, ) => void; +/** + * @deprecated Ignored. The in-process OTel log transport was removed; run + * telemetry is exported by the agent server, configured via the + * POSTHOG_AGENT_OTEL_LOGS_URL/_TOKEN environment variables. Kept only so + * existing consumers keep compiling; will be removed in a future major. + */ export interface OtelTransportConfig { /** PostHog ingest host, e.g., "https://us.i.posthog.com" */ host: string; @@ -88,7 +94,7 @@ export interface OtelTransportConfig { export interface AgentConfig { posthog?: PostHogAPIConfig; - /** OTEL transport config for shipping logs to PostHog Logs */ + /** @deprecated Ignored — see {@link OtelTransportConfig}. */ otelTransport?: OtelTransportConfig; /** Skip session log persistence (e.g. for preview sessions with no real task) */ skipLogPersistence?: boolean; diff --git a/packages/agent/tsup.config.ts b/packages/agent/tsup.config.ts index 57ebcd9f6a..9315c383da 100644 --- a/packages/agent/tsup.config.ts +++ b/packages/agent/tsup.config.ts @@ -88,6 +88,7 @@ const sharedOptions = { "@posthog/git", "@posthog/enricher", "@posthog/harness", + /^@opentelemetry\//, "fflate", ], external: [ diff --git a/packages/core/src/sessions/sessionService.ts b/packages/core/src/sessions/sessionService.ts index a22f9590aa..d7bc02d754 100644 --- a/packages/core/src/sessions/sessionService.ts +++ b/packages/core/src/sessions/sessionService.ts @@ -1591,6 +1591,7 @@ export class SessionService { { startedAtTs: number; agentTextChunks: number; agentOutputEvents: number } >(); private pendingPermissionHydratedRuns = new Set(); + /** In-flight hydrations keyed by `${taskRunId}:${hydrationMode}` */ private cloudHydrationPromises = new Map< string, Promise @@ -5529,6 +5530,25 @@ export class SessionService { if (onStatusChange) { existingWatcher.onStatusChange = onStatusChange; } + // The run finished while a live watcher was still attached: apply the + // final transcript, mark the terminal status, and tear the watcher down + // instead of leaving a live stream on a dead run. + if (isTerminalStatus(runStatus)) { + const terminalSession = this.d.store.getSessionByTaskId(taskId); + if (terminalSession?.taskRunId === taskRunId) { + void this.hydrateCloudTaskSessionFromLogs( + taskId, + taskRunId, + logUrl, + taskDescription, + runStatus, + runState, + ); + } + this.finalizeTerminalCloudTask(taskRunId, runStatus); + this.stopCloudTaskWatch(taskId); + return () => {}; + } // Ensure configOptions is populated on revisit const existing = this.d.store.getSessionByTaskId(taskId); if (existing) { @@ -5708,6 +5728,24 @@ export class SessionService { initialReasoningEffort, ); + // A run that is already terminal has no live stream to subscribe to: + // hydrate the final transcript, settle the terminal status, and return + // without registering a watcher. Plain hydration already fetches the + // full resume chain for terminal runs, and the resume wrapper only + // exists to buffer a live stream that does not exist here. + if (isTerminalStatus(runStatus)) { + void this.hydrateCloudTaskSessionFromLogs( + taskId, + taskRunId, + logUrl, + taskDescription, + runStatus, + runState, + ); + this.finalizeTerminalCloudTask(taskRunId, runStatus); + return () => {}; + } + const processCloudUpdate = (update: CloudTaskUpdatePayload): void => { if (update.kind === "logs" || update.kind === "snapshot") { this.d.store.updateSession(taskRunId, { @@ -5728,11 +5766,18 @@ export class SessionService { ), } : update; + // Evaluate staleness before handleCloudTaskUpdate mutates cloudStatus: + // a late non-terminal update must not re-notify after the run settled. + const isStaleNonTerminalStatus = this.isStaleNonTerminalCloudUpdate( + taskRunId, + normalizedUpdate, + ); this.handleCloudTaskUpdate(taskRunId, normalizedUpdate); if ( (update.kind === "status" || update.kind === "snapshot" || update.kind === "error") && + !isStaleNonTerminalStatus && watcher?.onStatusChange ) { watcher.onStatusChange(); @@ -5853,7 +5898,16 @@ export class SessionService { runStatus?: TaskRunStatus, runState?: Record, ): Promise { - const existing = this.cloudHydrationPromises.get(taskRunId); + // Key by hydration mode, not just run: a run going terminal must start + // its final-transcript hydration even while a resume-chain or single-run + // hydration for the same run is still in flight. + const hydrationMode = isTerminalStatus(runStatus) + ? "terminal-chain" + : runState?.resume_from_run_id + ? "resume-chain" + : "single"; + const hydrationKey = `${taskRunId}:${hydrationMode}`; + const existing = this.cloudHydrationPromises.get(hydrationKey); if (existing) { return existing; } @@ -5872,10 +5926,10 @@ export class SessionService { }); return undefined; }); - this.cloudHydrationPromises.set(taskRunId, hydration); + this.cloudHydrationPromises.set(hydrationKey, hydration); void hydration.finally(() => { - if (this.cloudHydrationPromises.get(taskRunId) === hydration) { - this.cloudHydrationPromises.delete(taskRunId); + if (this.cloudHydrationPromises.get(hydrationKey) === hydration) { + this.cloudHydrationPromises.delete(hydrationKey); } }); return hydration; @@ -5949,7 +6003,8 @@ export class SessionService { ? runState.resume_from_run_id : undefined; const isResumeRun = Boolean(resumeFromRunId); - if (isTerminalStatus(runStatus) || isResumeRun) { + const isTerminalRun = isTerminalStatus(runStatus); + if (isTerminalRun || isResumeRun) { // Resume chains need the full history even while the leaf run is still // active; otherwise a renderer restart hydrates only the final run. // Non-resume in-progress runs keep using the single-run log so hydrate @@ -6049,6 +6104,16 @@ export class SessionService { } rawEntries = result.entries; liveStreamLineCount = rawEntries.length; + // A terminal run whose persisted chain comes back empty can still + // have a complete S3 session log (persistence raced teardown); fall + // back to it rather than hydrating an empty final transcript. + if (rawEntries.length === 0 && logUrl) { + const parsed = await this.fetchSessionLogs(logUrl, taskRunId); + if (parsed.rawEntries.length > 0) { + rawEntries = parsed.rawEntries; + liveStreamLineCount = parsed.totalLineCount; + } + } } } else { const parsed = await this.fetchSessionLogs(logUrl, taskRunId); @@ -6096,15 +6161,17 @@ export class SessionService { // its chip renders right away) over the bare task description. const seedContent = this.initialCloudOptimisticPrompt.get(taskId) ?? taskDescription; - if (!hasUserPrompt && seedContent?.trim()) { + if (!isTerminalRun && !hasUserPrompt && seedContent?.trim()) { this.d.store.appendOptimisticItem(taskRunId, { type: "user_message", content: seedContent, timestamp: Date.now(), }); } - if (hasUserPrompt) { - // The real prompt has landed; the stash is no longer needed. + if (hasUserPrompt || isTerminalRun) { + // The stash is no longer needed once the real prompt lands - and a + // finished run gets no further echoes, so leftover optimistic items + // would otherwise linger as phantom tail items on the final transcript. this.initialCloudOptimisticPrompt.delete(taskId); this.d.store.clearTailOptimisticItems(taskRunId); } @@ -6119,25 +6186,42 @@ export class SessionService { // If live updates already populated a processed count, don't overwrite // that newer state with the persisted baseline fetched during startup. - if ( - session.processedLineCount !== undefined && - session.processedLineCount > 0 && - !isResumeRun - ) { + // Terminal hydration is different: it is the final transcript, so apply + // it whenever the persisted chain has more lines than the local stream. + const effectiveLineCount = Math.max(liveStreamLineCount, rawEntries.length); + const alreadyApplied = isTerminalRun + ? (session.processedLineCount ?? 0) >= effectiveLineCount + : session.processedLineCount !== undefined && + session.processedLineCount > 0 && + !isResumeRun; + if (alreadyApplied) { this.surfacePersistedPendingPermissions(taskRunId, rawEntries); this.pendingPermissionHydratedRuns.add(taskRunId); return { historyEntryCount: rawEntries.length, - liveStreamLineCount: session.processedLineCount, + liveStreamLineCount: session.processedLineCount ?? liveStreamLineCount, }; } + // A concurrent terminal-chain hydration (they memoize under different + // mode keys) may have already recorded the full chain as processed; a + // leaf-stream cursor from this older hydration must never lower it once + // the run has settled. + const settled = this.d.store.getSessions()[taskRunId]; + const settledCursor = isTerminalStatus(settled?.cloudStatus) + ? (settled?.processedLineCount ?? 0) + : 0; + this.d.store.updateSession(taskRunId, { events, isCloud: true, logUrl: logUrl ?? session.logUrl, cloudTranscriptEntryCount: rawEntries.length, - processedLineCount: liveStreamLineCount, + // Terminal hydration records the whole chain as processed so nothing + // re-applies it; live resume runs keep the leaf-stream cursor. + processedLineCount: isTerminalRun + ? effectiveLineCount + : Math.max(liveStreamLineCount, settledCursor), }); this.surfacePersistedPendingPermissions(taskRunId, rawEntries); this.pendingPermissionHydratedRuns.add(taskRunId); @@ -6145,12 +6229,63 @@ export class SessionService { // baseline already contains an in-flight session/prompt — the live delta // path otherwise sees delta <= 0 and never re-evaluates the tail. this.updatePromptStateFromEvents(taskRunId, events); + if (isTerminalRun) { + this.clearTerminalCloudPromptState(taskRunId); + } return { historyEntryCount: rawEntries.length, liveStreamLineCount, }; } + private finalizeTerminalCloudTask( + taskRunId: string, + status: TaskRunStatus | undefined, + ): void { + this.d.store.updateCloudStatus(taskRunId, { status }); + this.clearTerminalCloudPromptState(taskRunId); + } + + /** + * A terminal run can never flush its queue or answer a pending prompt; + * leaving those set keeps the composer in a busy state forever. + */ + private clearTerminalCloudPromptState(taskRunId: string): void { + const session = this.d.store.getSessions()[taskRunId]; + if ( + !session || + (!session.isPromptPending && session.messageQueue.length === 0) + ) { + return; + } + + this.d.store.clearMessageQueue(session.taskId); + this.d.store.updateSession(taskRunId, { + isPromptPending: false, + }); + } + + /** + * SSE replays and out-of-order bursts can deliver a non-terminal status + * after the run already settled; applying it would revive a finished run. + */ + private isStaleNonTerminalCloudUpdate( + taskRunId: string, + update: CloudTaskUpdatePayload, + ): boolean { + if (update.kind !== "status" && update.kind !== "snapshot") { + return false; + } + if (update.status === undefined) { + return false; + } + const currentCloudStatus = + this.d.store.getSessions()[taskRunId]?.cloudStatus; + return ( + isTerminalStatus(currentCloudStatus) && !isTerminalStatus(update.status) + ); + } + private isCurrentCloudTaskWatcher( taskId: string, runId: string, @@ -7074,7 +7209,17 @@ export class SessionService { } } - if (update.kind === "snapshot" && !isTerminalStatus(update.status)) { + // Evaluated once, before updateCloudStatus below can mutate cloudStatus. + const isStaleNonTerminalStatus = this.isStaleNonTerminalCloudUpdate( + taskRunId, + update, + ); + + if ( + update.kind === "snapshot" && + !isTerminalStatus(update.status) && + !isStaleNonTerminalStatus + ) { this.surfacePersistedPendingPermissions(taskRunId, update.newEntries); } @@ -7088,32 +7233,25 @@ export class SessionService { // Update cloud status fields if present if (update.kind === "status" || update.kind === "snapshot") { - this.d.store.updateCloudStatus(taskRunId, { - status: update.status, - stage: update.stage, - output: update.output, - errorMessage: update.errorMessage, - branch: update.branch, - }); - - if (update.status === "in_progress") { - this.tryRecoverIdleCloudQueue(taskRunId, { - serverSandboxAlive: update.sandboxAlive, + if (!isStaleNonTerminalStatus) { + this.d.store.updateCloudStatus(taskRunId, { + status: update.status, + stage: update.stage, + output: update.output, + errorMessage: update.errorMessage, + branch: update.branch, }); - } - if (isTerminalStatus(update.status)) { - // Clean up any pending resume messages that couldn't be sent - const session = this.d.store.getSessions()[taskRunId]; - if ( - session && - (session.messageQueue.length > 0 || session.isPromptPending) - ) { - this.d.store.clearMessageQueue(session.taskId); - this.d.store.updateSession(taskRunId, { - isPromptPending: false, + if (update.status === "in_progress") { + this.tryRecoverIdleCloudQueue(taskRunId, { + serverSandboxAlive: update.sandboxAlive, }); } + } + + if (isTerminalStatus(update.status)) { + // Pending resume messages can never be sent to a settled run. + this.clearTerminalCloudPromptState(taskRunId); this.stopCloudTaskWatch(update.taskId); } } diff --git a/packages/core/src/sessions/sessionStore.ts b/packages/core/src/sessions/sessionStore.ts index f8300eb8af..ddb1aff9f7 100644 --- a/packages/core/src/sessions/sessionStore.ts +++ b/packages/core/src/sessions/sessionStore.ts @@ -8,6 +8,7 @@ import { sendableQueuePrefixLength, type TaskRunStatus, } from "@posthog/shared"; +import { isTerminalStatus } from "@posthog/shared/domain-types"; import { setAutoFreeze } from "immer"; import { immer } from "zustand/middleware/immer"; import { createStore } from "zustand/vanilla"; @@ -180,7 +181,16 @@ export const sessionStoreSetters = { sessionStore.setState((state) => { const session = state.sessions[taskRunId]; if (!session) return; - if (fields.status !== undefined) session.cloudStatus = fields.status; + if (fields.status !== undefined) { + const currentStatus = session.cloudStatus; + if ( + isTerminalStatus(currentStatus) && + !isTerminalStatus(fields.status) + ) { + return; + } + session.cloudStatus = fields.status; + } if (fields.stage !== undefined) session.cloudStage = fields.stage; if (fields.output !== undefined) session.cloudOutput = fields.output; if (fields.errorMessage !== undefined) diff --git a/packages/core/src/sessions/sessionViewState.test.ts b/packages/core/src/sessions/sessionViewState.test.ts new file mode 100644 index 0000000000..1e218fe246 --- /dev/null +++ b/packages/core/src/sessions/sessionViewState.test.ts @@ -0,0 +1,87 @@ +import type { AgentSession } from "@posthog/shared"; +import type { Task, TaskRunStatus } from "@posthog/shared/domain-types"; +import { describe, expect, it } from "vitest"; +import { deriveSessionViewState } from "./sessionViewState"; + +function makeTask(runStatus: TaskRunStatus, runId = "run-1"): Task { + return { + id: "task-1", + task_number: 1, + slug: "task-1", + title: "Task", + description: "", + created_at: "2026-01-01T00:00:00.000Z", + updated_at: "2026-01-01T00:00:00.000Z", + origin_product: "user_created", + latest_run: { + id: runId, + status: runStatus, + environment: "cloud", + } as Task["latest_run"], + }; +} + +function makeSession( + cloudStatus: TaskRunStatus, + taskRunId = "run-1", +): AgentSession { + return { + taskId: "task-1", + taskRunId, + taskTitle: "Task", + channel: `agent-event:${taskRunId}`, + status: "connected", + events: [], + startedAt: 0, + isCloud: true, + cloudStatus, + isPromptPending: false, + isCompacting: false, + promptStartedAt: null, + pendingPermissions: new Map(), + pausedDurationMs: 0, + messageQueue: [], + optimisticItems: [], + }; +} + +describe("deriveSessionViewState", () => { + it("uses terminal task status over stale same-run session status", () => { + const state = deriveSessionViewState( + makeSession("in_progress"), + makeTask("completed"), + null, + true, + ); + + expect(state.cloudStatus).toBe("completed"); + expect(state.isCloudRunTerminal).toBe(true); + expect(state.isInitializing).toBe(false); + }); + + it("uses the task status when the session belongs to an older run", () => { + const state = deriveSessionViewState( + makeSession("completed", "old-run"), + makeTask("in_progress", "new-run"), + null, + true, + ); + + expect(state.cloudStatus).toBe("in_progress"); + expect(state.isCloudRunNotTerminal).toBe(true); + }); + + it("treats not_started as a non-terminal cloud state", () => { + const state = deriveSessionViewState( + undefined, + makeTask("not_started"), + null, + true, + ); + + expect(state.cloudStatus).toBe("not_started"); + expect(state.isCloudRunNotTerminal).toBe(true); + expect(state.isCloudRunTerminal).toBe(false); + expect(state.isInitializing).toBe(true); + }); +}); diff --git a/packages/core/src/sessions/sessionViewState.ts b/packages/core/src/sessions/sessionViewState.ts index de25c07556..a41c356550 100644 --- a/packages/core/src/sessions/sessionViewState.ts +++ b/packages/core/src/sessions/sessionViewState.ts @@ -1,5 +1,10 @@ import type { AcpMessage, AgentSession, Workspace } from "@posthog/shared"; -import type { Task, TaskRunStatus } from "@posthog/shared/domain-types"; +import { + isTerminalStatus, + type Task, + type TaskRunStatus, +} from "@posthog/shared/domain-types"; +import { resolveEffectiveCloudStatus } from "../task-detail/cloudRunState"; export interface SessionViewState { isCloudRunNotTerminal: boolean; @@ -23,11 +28,9 @@ export function deriveSessionViewState( workspace: Workspace | null, isCloud: boolean, ): SessionViewState { - const cloudStatus = session?.cloudStatus ?? null; - const isCloudRunNotTerminal = - isCloud && - (!cloudStatus || cloudStatus === "queued" || cloudStatus === "in_progress"); - const isCloudRunTerminal = isCloud && !isCloudRunNotTerminal; + const cloudStatus = resolveEffectiveCloudStatus(task, session); + const isCloudRunTerminal = isCloud && isTerminalStatus(cloudStatus); + const isCloudRunNotTerminal = isCloud && !isCloudRunTerminal; const hasError = session?.status === "error" && !session?.idleKilled; const handoffInProgress = session?.handoffInProgress ?? false; diff --git a/packages/core/src/task-detail/cloudRunState.test.ts b/packages/core/src/task-detail/cloudRunState.test.ts new file mode 100644 index 0000000000..5f48138e40 --- /dev/null +++ b/packages/core/src/task-detail/cloudRunState.test.ts @@ -0,0 +1,51 @@ +import type { Task, TaskRunStatus } from "@posthog/shared/domain-types"; +import { describe, expect, it } from "vitest"; +import { deriveCloudRunState } from "./cloudRunState"; + +function makeTask(runStatus: TaskRunStatus, runId = "run-1"): Task { + return { + id: "task-1", + task_number: 1, + slug: "task-1", + title: "Task", + description: "", + created_at: "2026-01-01T00:00:00.000Z", + updated_at: "2026-01-01T00:00:00.000Z", + origin_product: "user_created", + latest_run: { + id: runId, + status: runStatus, + environment: "cloud", + } as Task["latest_run"], + }; +} + +describe("deriveCloudRunState", () => { + it("uses terminal task status over stale same-run session status", () => { + const state = deriveCloudRunState( + makeTask("completed"), + { + taskRunId: "run-1", + cloudStatus: "in_progress", + }, + null, + ); + + expect(state.cloudStatus).toBe("completed"); + expect(state.isRunActive).toBe(false); + }); + + it("uses task status when the session belongs to an older run", () => { + const state = deriveCloudRunState( + makeTask("in_progress", "new-run"), + { + taskRunId: "old-run", + cloudStatus: "completed", + }, + null, + ); + + expect(state.cloudStatus).toBe("in_progress"); + expect(state.isRunActive).toBe(true); + }); +}); diff --git a/packages/core/src/task-detail/cloudRunState.ts b/packages/core/src/task-detail/cloudRunState.ts index f076d3ca24..298f94b445 100644 --- a/packages/core/src/task-detail/cloudRunState.ts +++ b/packages/core/src/task-detail/cloudRunState.ts @@ -1,8 +1,37 @@ -import type { ChangedFile, Task } from "@posthog/shared/domain-types"; +import { + type ChangedFile, + isTerminalStatus, + type Task, + type TaskRunStatus, +} from "@posthog/shared/domain-types"; export interface CloudRunSessionLike { + taskRunId?: string | null; cloudBranch?: string | null; - cloudStatus?: string | null; + cloudStatus?: TaskRunStatus | null; +} + +/** + * Effective run status for the UI: a terminal task status always wins; + * otherwise the session's live status wins while the session belongs to the + * task's latest run. + */ +export function resolveEffectiveCloudStatus( + task: Task, + session: + | { taskRunId?: string | null; cloudStatus?: TaskRunStatus | null } + | null + | undefined, +): TaskRunStatus | null { + const taskRunStatus = task.latest_run?.status ?? null; + const taskRunId = task.latest_run?.id; + const sessionMatchesLatestRun = + !!taskRunId && session?.taskRunId === taskRunId; + return sessionMatchesLatestRun + ? isTerminalStatus(taskRunStatus) + ? taskRunStatus + : (session?.cloudStatus ?? taskRunStatus) + : (taskRunStatus ?? session?.cloudStatus ?? null); } export interface CloudRunStateResult { @@ -23,7 +52,7 @@ export function deriveCloudRunState( const effectiveBranch = branch ?? cloudBranch; const repo = task.repository ?? null; - const cloudStatus = session?.cloudStatus ?? task.latest_run?.status ?? null; + const cloudStatus = resolveEffectiveCloudStatus(task, session); const isRunActive = cloudStatus === "queued" || cloudStatus === "in_progress" || diff --git a/packages/ui/src/features/sessions/sessionServiceHost.test.ts b/packages/ui/src/features/sessions/sessionServiceHost.test.ts index 90a90ca460..e4f47c7cc8 100644 --- a/packages/ui/src/features/sessions/sessionServiceHost.test.ts +++ b/packages/ui/src/features/sessions/sessionServiceHost.test.ts @@ -1529,6 +1529,45 @@ describe("SessionService", () => { expect(unsubscribe).not.toHaveBeenCalled(); }); + it("marks a reused same-run watcher terminal when task data reports completion", () => { + const service = getSessionService(); + const unsubscribe = vi.fn(); + const onStatusChange = vi.fn(); + mockTrpcCloudTask.onUpdate.subscribe.mockReturnValueOnce({ + unsubscribe, + }); + + service.watchCloudTask( + "task-123", + "run-123", + "https://api.anthropic.com", + 123, + onStatusChange, + ); + service.watchCloudTask( + "task-123", + "run-123", + "https://api.anthropic.com", + 123, + undefined, + "https://example.com/logs/run-123", + undefined, + "claude", + undefined, + undefined, + undefined, + "completed", + ); + + expect(mockSessionStoreSetters.updateCloudStatus).toHaveBeenCalledWith( + "run-123", + { status: "completed" }, + ); + expect(unsubscribe).toHaveBeenCalledTimes(1); + expect(mockTrpcCloudTask.watch.mutate).toHaveBeenCalledTimes(1); + expect(onStatusChange).not.toHaveBeenCalled(); + }); + it.each<[string, Partial, boolean]>([ [ "skips a hydrated terminal (completed) run", @@ -1584,6 +1623,399 @@ describe("SessionService", () => { ); }); + it("hydrates a caller-reported terminal run without watching it", async () => { + const service = getSessionService(); + const onStatusChange = vi.fn(); + const session = createMockSession({ + taskId: "task-123", + taskRunId: "run-123", + cloudStatus: "in_progress", + events: [ + { + type: "acp_message", + ts: 1700000000, + message: { + jsonrpc: "2.0", + method: "session/update", + params: {}, + }, + } as AcpMessage, + ], + isPromptPending: true, + messageQueue: [ + { + id: "queued-1", + content: "follow up", + queuedAt: 1700000001, + }, + ], + processedLineCount: 1, + }); + mockSessionStoreSetters.getSessionByTaskId.mockReturnValue(session); + mockSessionStoreSetters.getSessions.mockReturnValue({ + "run-123": session, + }); + const finalEntries = [ + { timestamp: "2024-01-01T00:00:00Z", notification: {} }, + { timestamp: "2024-01-01T00:01:00Z", notification: {} }, + ]; + const finalEvents = [ + { + type: "acp_message", + ts: 1700000000, + message: { + jsonrpc: "2.0", + method: "session/update", + params: {}, + }, + } as AcpMessage, + { + type: "acp_message", + ts: 1700000001, + message: { + jsonrpc: "2.0", + method: "session/update", + params: {}, + }, + } as AcpMessage, + ]; + mockAuthenticatedClient.getTaskRunSessionLogsResult.mockResolvedValue({ + entries: finalEntries, + complete: true, + }); + mockConvertStoredEntriesToEvents.mockReturnValueOnce(finalEvents); + + service.watchCloudTask( + "task-123", + "run-123", + "https://api.anthropic.com", + 123, + onStatusChange, + "https://example.com/logs/run-123", + undefined, + "claude", + undefined, + undefined, + undefined, + "completed", + ); + + expect(mockSessionStoreSetters.updateCloudStatus).toHaveBeenCalledWith( + "run-123", + { status: "completed" }, + ); + expect(mockSessionStoreSetters.clearMessageQueue).toHaveBeenCalledWith( + "task-123", + ); + expect(mockSessionStoreSetters.updateSession).toHaveBeenCalledWith( + "run-123", + expect.objectContaining({ isPromptPending: false }), + ); + + expect(mockTrpcCloudTask.onUpdate.subscribe).not.toHaveBeenCalled(); + expect(mockTrpcCloudTask.watch.mutate).not.toHaveBeenCalled(); + expect(onStatusChange).not.toHaveBeenCalled(); + await vi.waitFor(() => { + expect( + mockAuthenticatedClient.getTaskRunSessionLogsResult, + ).toHaveBeenCalledWith("task-123", "run-123", { limit: 100000 }); + }); + expect(mockSessionStoreSetters.updateSession).toHaveBeenCalledWith( + "run-123", + expect.objectContaining({ + events: finalEvents, + processedLineCount: finalEntries.length, + }), + ); + }); + + it("falls back to the run log URL when terminal chain hydration is empty", async () => { + const service = getSessionService(); + const session = createMockSession({ + taskId: "task-123", + taskRunId: "run-123", + cloudStatus: "in_progress", + isCloud: true, + events: [], + }); + mockSessionStoreSetters.getSessionByTaskId.mockReturnValue(session); + mockSessionStoreSetters.getSessions.mockReturnValue({ + "run-123": session, + }); + mockAuthenticatedClient.getTaskRunSessionLogs.mockResolvedValue([]); + mockTrpcLogs.readLocalLogs.query.mockResolvedValue(""); + mockTrpcLogs.fetchS3Logs.query.mockResolvedValue( + JSON.stringify({ + type: "notification", + timestamp: "2024-01-01T00:00:00Z", + notification: { + method: "session/update", + params: {}, + }, + }), + ); + mockTrpcLogs.writeLocalLogs.mutate.mockResolvedValue(undefined); + const finalEvents = [ + { + type: "acp_message", + ts: 1700000000, + message: { + jsonrpc: "2.0", + method: "session/update", + params: {}, + }, + } as AcpMessage, + ]; + mockConvertStoredEntriesToEvents.mockReturnValueOnce(finalEvents); + + service.watchCloudTask( + "task-123", + "run-123", + "https://api.anthropic.com", + 123, + undefined, + "https://example.com/logs/run-123", + undefined, + "claude", + undefined, + "build me a thing", + undefined, + "completed", + ); + + await vi.waitFor(() => { + expect(mockTrpcLogs.fetchS3Logs.query).toHaveBeenCalledWith({ + logUrl: "https://example.com/logs/run-123", + }); + }); + await vi.waitFor(() => { + expect(mockSessionStoreSetters.updateSession).toHaveBeenCalledWith( + "run-123", + expect.objectContaining({ + events: finalEvents, + processedLineCount: 1, + }), + ); + }); + expect( + mockSessionStoreSetters.clearTailOptimisticItems, + ).toHaveBeenCalledWith("run-123"); + expect( + mockSessionStoreSetters.appendOptimisticItem, + ).not.toHaveBeenCalled(); + }); + + it("does not cache or seed an empty terminal hydration", async () => { + const service = getSessionService(); + const session = createMockSession({ + taskId: "task-123", + taskRunId: "run-123", + cloudStatus: "in_progress", + isCloud: true, + events: [], + }); + mockSessionStoreSetters.getSessionByTaskId.mockReturnValue(session); + mockSessionStoreSetters.getSessions.mockReturnValue({ + "run-123": session, + }); + mockAuthenticatedClient.getTaskRunSessionLogs.mockResolvedValue([]); + mockTrpcLogs.readLocalLogs.query.mockResolvedValue(""); + mockTrpcLogs.fetchS3Logs.query.mockResolvedValue(""); + + service.watchCloudTask( + "task-123", + "run-123", + "https://api.anthropic.com", + 123, + undefined, + "https://example.com/logs/run-123", + undefined, + "claude", + undefined, + "build me a thing", + undefined, + "completed", + ); + + await vi.waitFor(() => { + expect( + mockSessionStoreSetters.clearTailOptimisticItems, + ).toHaveBeenCalledWith("run-123"); + }); + expect( + mockSessionStoreSetters.appendOptimisticItem, + ).not.toHaveBeenCalled(); + expect(mockSessionStoreSetters.updateSession).not.toHaveBeenCalledWith( + "run-123", + expect.objectContaining({ processedLineCount: 0 }), + ); + }); + + it("starts terminal hydration even when resume-chain hydration is already in flight", async () => { + const service = getSessionService(); + const session = createMockSession({ + taskId: "task-123", + taskRunId: "run-123", + isCloud: true, + events: [], + }); + mockSessionStoreSetters.getSessionByTaskId.mockReturnValue(session); + mockSessionStoreSetters.getSessions.mockReturnValue({ + "run-123": session, + }); + let resolveFirstHydration!: (result: { + entries: Array<{ timestamp: string; notification: object }>; + complete: boolean; + }) => void; + const firstHydration = new Promise<{ + entries: Array<{ timestamp: string; notification: object }>; + complete: boolean; + }>((resolve) => { + resolveFirstHydration = resolve; + }); + mockAuthenticatedClient.getTaskRunSessionLogsResult + .mockReturnValueOnce(firstHydration) + .mockResolvedValueOnce({ entries: [], complete: true }); + + service.watchCloudTask( + "task-123", + "run-123", + "https://api.anthropic.com", + 123, + undefined, + "https://example.com/logs/run-123", + undefined, + "claude", + undefined, + undefined, + undefined, + "in_progress", + undefined, + { resume_from_run_id: "previous-run" }, + ); + + // Resume hydration fetches the ancestor and current run in parallel; + // the pending ancestor fetch keeps this first hydration in flight. + await vi.waitFor(() => { + expect( + mockAuthenticatedClient.getTaskRunSessionLogsResult, + ).toHaveBeenCalledTimes(2); + }); + + service.watchCloudTask( + "task-123", + "run-123", + "https://api.anthropic.com", + 123, + undefined, + "https://example.com/logs/run-123", + undefined, + "claude", + undefined, + undefined, + undefined, + "completed", + undefined, + { resume_from_run_id: "previous-run" }, + ); + + // The terminal hydration issues its own single terminal-chain fetch + // instead of being deduped onto the in-flight resume-chain hydration. + await vi.waitFor(() => { + expect( + mockAuthenticatedClient.getTaskRunSessionLogsResult, + ).toHaveBeenCalledTimes(3); + }); + resolveFirstHydration({ entries: [], complete: true }); + }); + + it("keeps the settled terminal cursor when a resume-chain hydration resolves late", async () => { + const service = getSessionService(); + const session = createMockSession({ + taskId: "task-123", + taskRunId: "run-123", + isCloud: true, + events: [], + }); + mockSessionStoreSetters.getSessionByTaskId.mockReturnValue(session); + mockSessionStoreSetters.getSessions.mockReturnValue({ + "run-123": session, + }); + mockTrpcLogs.readLocalLogs.query.mockResolvedValue(""); + mockTrpcLogs.fetchS3Logs.query.mockResolvedValue(""); + const entry = (text: string) => ({ + timestamp: "2026-07-15T10:00:00+00:00", + notification: { + method: "session/update", + params: { + update: { + sessionUpdate: "agent_message", + content: { type: "text", text }, + }, + }, + }, + }); + let resolveAncestor!: (r: { + entries: object[]; + complete: boolean; + }) => void; + let resolveCurrent!: (r: { + entries: object[]; + complete: boolean; + }) => void; + mockAuthenticatedClient.getTaskRunSessionLogsResult + .mockReturnValueOnce( + new Promise((r) => { + resolveAncestor = r; + }), + ) + .mockReturnValueOnce( + new Promise((r) => { + resolveCurrent = r; + }), + ); + + service.watchCloudTask( + "task-123", + "run-123", + "https://api.anthropic.com", + 123, + undefined, + "https://example.com/logs/run-123", + undefined, + "claude", + undefined, + undefined, + undefined, + "in_progress", + undefined, + { resume_from_run_id: "previous-run" }, + ); + + await vi.waitFor(() => { + expect( + mockAuthenticatedClient.getTaskRunSessionLogsResult, + ).toHaveBeenCalledTimes(2); + }); + + // The run settles while the resume-chain fetches are still in flight, + // exactly as a concurrent terminal-chain hydration records it. + session.cloudStatus = "completed"; + session.processedLineCount = 5; + + resolveAncestor({ entries: [entry("a"), entry("b")], complete: true }); + resolveCurrent({ entries: [entry("c")], complete: true }); + + // The late resume-chain write must not lower the settled cursor to its + // leaf-only count. + await vi.waitFor(() => { + expect(mockSessionStoreSetters.updateSession).toHaveBeenCalledWith( + "run-123", + expect.objectContaining({ processedLineCount: 5 }), + ); + }); + }); + it("does not re-subscribe across repeated calls for a hydrated terminal run", () => { const service = getSessionService(); mockSessionStoreSetters.getSessionByTaskId.mockReturnValue( @@ -1649,6 +2081,47 @@ describe("SessionService", () => { expect(onStatusChange).toHaveBeenCalledTimes(1); }); + it("ignores stale non-terminal stream status after a terminal status", () => { + const service = getSessionService(); + const onStatusChange = vi.fn(); + const completedSession = createMockSession({ + taskId: "task-123", + taskRunId: "run-123", + isCloud: true, + cloudStatus: "completed", + }); + mockSessionStoreSetters.getSessions.mockReturnValue({ + "run-123": completedSession, + }); + + service.watchCloudTask( + "task-123", + "run-123", + "https://api.anthropic.com", + 123, + onStatusChange, + ); + + const subscribeOptions = mockTrpcCloudTask.onUpdate.subscribe.mock + .calls[0][1] as { + onData: (update: { + kind: "status"; + taskId: string; + runId: string; + status: "in_progress"; + }) => void; + }; + subscribeOptions.onData({ + kind: "status", + taskId: "task-123", + runId: "run-123", + status: "in_progress", + }); + + expect(mockSessionStoreSetters.updateCloudStatus).not.toHaveBeenCalled(); + expect(onStatusChange).not.toHaveBeenCalled(); + }); + it("hydrates a fresh cloud session from persisted logs before replay arrives", async () => { const service = getSessionService(); const hydratedSession = createMockSession({ diff --git a/packages/ui/src/features/sessions/sessionStore.test.ts b/packages/ui/src/features/sessions/sessionStore.test.ts index 25d58fb5ba..d088a7ed31 100644 --- a/packages/ui/src/features/sessions/sessionStore.test.ts +++ b/packages/ui/src/features/sessions/sessionStore.test.ts @@ -116,6 +116,47 @@ describe("dequeueMessages", () => { }); }); +describe("updateCloudStatus", () => { + beforeEach(() => { + useSessionStore.setState((state) => { + state.sessions = {}; + state.taskIdIndex = {}; + }); + }); + + it("does not downgrade a terminal run when a stale non-terminal status arrives", () => { + sessionStoreSetters.setSession({ + taskRunId: "run-123", + taskId: "task-123", + taskTitle: "Test", + channel: "agent-event:run-123", + events: [], + startedAt: 0, + status: "connected", + isPromptPending: false, + isCompacting: false, + promptStartedAt: null, + pendingPermissions: new Map(), + pausedDurationMs: 0, + messageQueue: [], + optimisticItems: [], + cloudStatus: "completed", + }); + + sessionStoreSetters.updateCloudStatus("run-123", { + status: "in_progress", + branch: "stale-branch", + }); + + expect(useSessionStore.getState().sessions["run-123"].cloudStatus).toBe( + "completed", + ); + expect(useSessionStore.getState().sessions["run-123"].cloudBranch).toBe( + undefined, + ); + }); +}); + describe("dequeueMessagesAsText", () => { beforeEach(() => { useSessionStore.setState((state) => { diff --git a/packages/ui/src/features/task-detail/hooks/useCloudRunState.ts b/packages/ui/src/features/task-detail/hooks/useCloudRunState.ts index 88d2d82518..cff2f5cec0 100644 --- a/packages/ui/src/features/task-detail/hooks/useCloudRunState.ts +++ b/packages/ui/src/features/task-detail/hooks/useCloudRunState.ts @@ -4,14 +4,19 @@ import type { Task } from "@posthog/shared/domain-types"; import { useMemo } from "react"; import { resolveCloudPrUrl } from "../../git-interaction/cloudPrUrl"; import { useSessionForTask } from "../../sessions/useSession"; +import { pickFreshestTask } from "../../tasks/taskFreshness"; import { useTasks } from "../../tasks/useTasks"; import { useCloudEventSummary } from "./useCloudEventSummary"; export function useCloudRunState(taskId: string, task: Task) { const { data: tasks = [] } = useTasks(); const freshTask = useMemo( - () => tasks.find((t) => t.id === taskId) ?? task, - [tasks, taskId, task], + () => + pickFreshestTask( + task, + tasks.find((t) => t.id === taskId), + ), + [task, taskId, tasks], ); const session = useSessionForTask(taskId); diff --git a/packages/ui/src/features/tasks/queries.test.ts b/packages/ui/src/features/tasks/queries.test.ts new file mode 100644 index 0000000000..e2c0005e0e --- /dev/null +++ b/packages/ui/src/features/tasks/queries.test.ts @@ -0,0 +1,23 @@ +import { ApiRequestError } from "@posthog/api-client/fetcher"; +import { describe, expect, it } from "vitest"; +import { isTaskDetailNotFoundError } from "./queries"; + +describe("task queries", () => { + it("detects task detail 404 errors from the shared API fetcher", () => { + expect( + isTaskDetailNotFoundError( + new ApiRequestError(404, '{"detail":"Not found."}'), + ), + ).toBe(true); + expect( + isTaskDetailNotFoundError( + new ApiRequestError(500, '{"detail":"Server error."}'), + ), + ).toBe(false); + // A plain error merely mentioning a 404 (e.g. quoting an upstream body) + // must not read as one. + expect(isTaskDetailNotFoundError(new Error("Failed request: [404]"))).toBe( + false, + ); + }); +}); diff --git a/packages/ui/src/features/tasks/queries.ts b/packages/ui/src/features/tasks/queries.ts index 5333809331..58013affe0 100644 --- a/packages/ui/src/features/tasks/queries.ts +++ b/packages/ui/src/features/tasks/queries.ts @@ -1,3 +1,4 @@ +import { requestErrorStatus } from "@posthog/api-client/fetcher"; import { resolveService } from "@posthog/di/container"; import { NotAuthenticatedError } from "@posthog/shared"; import type { Task } from "@posthog/shared/domain-types"; @@ -23,9 +24,17 @@ export function taskDetailQuery(taskId: string) { return (await client.getTask(taskId)) as unknown as Task; }, meta: AUTH_SCOPED_QUERY_META, + // A 404 is a definitive answer (optimistic/cloud-pending tasks aren't + // returnable by the API yet) - retrying it only multiplies the miss. + retry: (failureCount, error) => + !isTaskDetailNotFoundError(error) && failureCount < 3, }); } +export function isTaskDetailNotFoundError(error: unknown): boolean { + return requestErrorStatus(error) === 404; +} + // Read a task from the already-loaded sidebar list cache without fetching. // Lets the task-detail route loader resolve synchronously from cache. export function getCachedTask(taskId: string): Task | undefined { diff --git a/packages/ui/src/features/tasks/taskFreshness.test.ts b/packages/ui/src/features/tasks/taskFreshness.test.ts new file mode 100644 index 0000000000..71d6a8815f --- /dev/null +++ b/packages/ui/src/features/tasks/taskFreshness.test.ts @@ -0,0 +1,65 @@ +import type { Task } from "@posthog/shared/domain-types"; +import { describe, expect, it } from "vitest"; +import { pickFreshestTask } from "./taskFreshness"; + +function makeTask( + title: string, + updatedAt: string, + runUpdatedAt?: string, +): Task { + return { + id: title, + task_number: 1, + slug: title, + title, + description: "", + created_at: "2026-01-01T00:00:00.000Z", + updated_at: updatedAt, + origin_product: "user_created", + latest_run: runUpdatedAt + ? ({ + id: `${title}-run`, + updated_at: runUpdatedAt, + } as Task["latest_run"]) + : undefined, + }; +} + +describe("pickFreshestTask", () => { + it("prefers the first task when timestamps tie", () => { + const detail = makeTask("detail", "2026-01-01T00:00:00.000Z"); + const list = makeTask("list", "2026-01-01T00:00:00.000Z"); + + expect(pickFreshestTask(detail, list)).toBe(detail); + }); + + it("uses latest run activity when it is newer than the task", () => { + const completedDetail = makeTask( + "detail", + "2026-01-01T00:00:00.000Z", + "2026-01-01T00:05:00.000Z", + ); + const staleList = makeTask( + "list", + "2026-01-01T00:03:00.000Z", + "2026-01-01T00:03:00.000Z", + ); + + expect(pickFreshestTask(staleList, completedDetail)).toBe(completedDetail); + }); + + it("keeps a newer list update when it arrives after detail data", () => { + const detail = makeTask( + "detail", + "2026-01-01T00:00:00.000Z", + "2026-01-01T00:05:00.000Z", + ); + const newerList = makeTask( + "list", + "2026-01-01T00:06:00.000Z", + "2026-01-01T00:06:00.000Z", + ); + + expect(pickFreshestTask(detail, newerList)).toBe(newerList); + }); +}); diff --git a/packages/ui/src/features/tasks/taskFreshness.ts b/packages/ui/src/features/tasks/taskFreshness.ts new file mode 100644 index 0000000000..02b9bd83fc --- /dev/null +++ b/packages/ui/src/features/tasks/taskFreshness.ts @@ -0,0 +1,27 @@ +import type { Task } from "@posthog/shared/domain-types"; + +function parseTime(value: string | null | undefined): number { + const timestamp = value ? Date.parse(value) : Number.NEGATIVE_INFINITY; + return Number.isFinite(timestamp) ? timestamp : Number.NEGATIVE_INFINITY; +} + +export function getTaskFreshness(task: Task): number { + return Math.max( + parseTime(task.updated_at), + parseTime(task.latest_run?.updated_at), + ); +} + +export function pickFreshestTask(a: Task, b: Task | null | undefined): Task; +export function pickFreshestTask( + a: Task | null | undefined, + b: Task | null | undefined, +): Task | undefined; +export function pickFreshestTask( + a: Task | null | undefined, + b: Task | null | undefined, +): Task | undefined { + if (!a) return b ?? undefined; + if (!b) return a; + return getTaskFreshness(b) > getTaskFreshness(a) ? b : a; +} diff --git a/packages/ui/src/router/routes/code/tasks/$taskId.tsx b/packages/ui/src/router/routes/code/tasks/$taskId.tsx index 9fd3ff3af7..dcea2e4bd5 100644 --- a/packages/ui/src/router/routes/code/tasks/$taskId.tsx +++ b/packages/ui/src/router/routes/code/tasks/$taskId.tsx @@ -5,6 +5,7 @@ import { getCachedTaskDetail, taskDetailQuery, } from "@posthog/ui/features/tasks/queries"; +import { pickFreshestTask } from "@posthog/ui/features/tasks/taskFreshness"; import { useTasks } from "@posthog/ui/features/tasks/useTasks"; import { TaskDetailSkeleton } from "@posthog/ui/router/routeSkeletons"; import { yieldToPaint } from "@posthog/ui/router/yieldToPaint"; @@ -37,25 +38,25 @@ function TaskDetailRoute() { const loaderTask = Route.useLoaderData(); const { data: tasks } = useTasks(); const fromList = tasks?.find((t) => t.id === taskId); + const initialTask = pickFreshestTask(fromList, loaderTask); - // Cold deep-link / URL restore: nothing cached. Fetch the single task here so - // a hang or 404 only affects this view's spinner, never the router. - const needsFetch = !fromList && !loaderTask; + // Always fetch so a stale cached copy converges on the server's latest run + // state; render whichever copy is freshest. const { data: fetched, isError, isSuccess, - } = useQuery({ - ...taskDetailQuery(taskId), - enabled: needsFetch, - }); + } = useQuery(taskDetailQuery(taskId)); - // Prefer the live list task (kept fresh by polling + subscriptions). - const task = fromList ?? loaderTask ?? fetched; + const task = pickFreshestTask(fetched, initialTask); - // Task doesn't exist (deleted, 404, or stale deep link): the cold fetch - // settled with an error or empty result. Redirect to the new-task screen - // rather than spin forever — matches the old navigationStore.hydrateTask. + // Cold deep-link / URL restore with nothing cached: if the fetch settled + // with an error or empty result, redirect to the new-task screen rather + // than spin forever — matches the old navigationStore.hydrateTask. While a + // cached/list copy exists, a 404 is NOT authoritative (optimistic and + // cloud-pending tasks aren't returnable by the API yet — see the loader + // comment), so never redirect away from a usable task. + const needsFetch = !initialTask; if (needsFetch && (isError || (isSuccess && !fetched))) { return ; } diff --git a/packages/ui/src/router/routes/website/$channelId/tasks/$taskId.tsx b/packages/ui/src/router/routes/website/$channelId/tasks/$taskId.tsx index 62ce110a5d..bd9c0ceb2b 100644 --- a/packages/ui/src/router/routes/website/$channelId/tasks/$taskId.tsx +++ b/packages/ui/src/router/routes/website/$channelId/tasks/$taskId.tsx @@ -6,13 +6,16 @@ import { TaskDetail } from "@posthog/ui/features/task-detail/components/TaskDeta import { getCachedTask, getCachedTaskDetail, + isTaskDetailNotFoundError, taskDetailQuery, } from "@posthog/ui/features/tasks/queries"; +import { pickFreshestTask } from "@posthog/ui/features/tasks/taskFreshness"; import { useTasks } from "@posthog/ui/features/tasks/useTasks"; import { TaskDetailSkeleton } from "@posthog/ui/router/routeSkeletons"; import { yieldToPaint } from "@posthog/ui/router/yieldToPaint"; +import { Button, Flex, Text } from "@radix-ui/themes"; import { useQuery } from "@tanstack/react-query"; -import { createFileRoute } from "@tanstack/react-router"; +import { createFileRoute, Navigate } from "@tanstack/react-router"; import { useEffect } from "react"; export const Route = createFileRoute("/website/$channelId/tasks/$taskId")({ @@ -36,6 +39,7 @@ function ChannelTaskDetailRoute() { const loaderTask = Route.useLoaderData(); const { data: tasks } = useTasks(); const fromList = tasks?.find((t) => t.id === taskId); + const initialTask = pickFreshestTask(fromList, loaderTask); const { channels } = useChannels(); const channelName = channels.find((c) => c.id === channelId)?.name; @@ -48,12 +52,52 @@ function ChannelTaskDetailRoute() { markAsViewed(taskId); }, [taskId, markAsViewed]); - const { data: fetched } = useQuery({ + const { + data: fetched, + error, + isError, + isFetching, + isSuccess, + refetch, + } = useQuery({ ...taskDetailQuery(taskId), - enabled: !fromList && !loaderTask, }); - const task = fromList ?? loaderTask ?? fetched; + const task = pickFreshestTask(fetched, initialTask); + + // While a cached/list copy exists, a 404 is NOT authoritative (optimistic + // and cloud-pending tasks aren't returnable by the API yet — see the loader + // comment), so only treat the task as gone when nothing cached is usable. + if (!initialTask && isTaskDetailNotFoundError(error)) { + return ; + } + + if (!task && isSuccess && !fetched) { + return ; + } + + if (!task && isError) { + const message = + error instanceof Error ? error.message : "Failed to load task"; + return ( + + + Failed to load task + + {message} + + + + + ); + } if (!task) { return ; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 4a8c6ced07..6680b87dd7 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -309,7 +309,7 @@ importers: version: 4.6.2(patch_hash=a8cd00b963d4ae6787cc267a2345cc1b32bcba5fa1131328b8017ee0d316ab0e)(react-dom@19.2.6(react@19.2.6))(react@19.2.6) react-scan: specifier: ^0.5.6 - version: 0.5.7(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(esbuild@0.27.2)(eslint@10.5.0(jiti@2.7.0))(react-dom@19.2.6(react@19.2.6))(react@19.2.6)(rollup@4.57.1) + version: 0.5.7(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(@opentelemetry/exporter-trace-otlp-http@0.208.0(@opentelemetry/api@1.9.1))(esbuild@0.27.2)(eslint@10.5.0(jiti@2.7.0))(react-dom@19.2.6(react@19.2.6))(react@19.2.6)(rollup@4.57.1) reflect-metadata: specifier: ^0.2.2 version: 0.2.2 @@ -778,18 +778,27 @@ importers: '@openai/codex': specifier: 0.144.0 version: 0.144.0 + '@opentelemetry/api': + specifier: ^1.9.1 + version: 1.9.1 '@opentelemetry/api-logs': specifier: ^0.208.0 version: 0.208.0 '@opentelemetry/exporter-logs-otlp-http': specifier: ^0.208.0 version: 0.208.0(@opentelemetry/api@1.9.1) + '@opentelemetry/exporter-trace-otlp-http': + specifier: ^0.208.0 + version: 0.208.0(@opentelemetry/api@1.9.1) '@opentelemetry/resources': specifier: ^2.0.0 version: 2.5.0(@opentelemetry/api@1.9.1) '@opentelemetry/sdk-logs': specifier: ^0.208.0 version: 0.208.0(@opentelemetry/api@1.9.1) + '@opentelemetry/sdk-trace-base': + specifier: ^2.8.0 + version: 2.8.0(@opentelemetry/api@1.9.1) '@opentelemetry/semantic-conventions': specifier: ^1.28.0 version: 1.39.0 @@ -2843,11 +2852,11 @@ packages: '@esbuild-kit/core-utils@3.3.2': resolution: {integrity: sha512-sPRAnw9CdSsRmEtnsl2WXWdyquogVpB3yZ3dgwJfe8zrOzTsV7cJvmwrKVa+0ma5BoiGJ+BoqkMvawbayKUsqQ==} - deprecated: 'Merged into tsx: https://tsx.is' + deprecated: 'Merged into tsx: https://tsx.hirok.io' '@esbuild-kit/esm-loader@2.6.5': resolution: {integrity: sha512-FxEMIkJKnodyA1OaCUoEvbYRkoZlLZ4d/eXFu9Fh8CbBBgP5EmZxrfTRyN0qpXZ4vOvqnE5YdRdcrmUUXuU+dA==} - deprecated: 'Merged into tsx: https://tsx.is' + deprecated: 'Merged into tsx: https://tsx.hirok.io' '@esbuild/aix-ppc64@0.25.12': resolution: {integrity: sha512-Hhmwd6CInZ3dwpuGTF8fJG6yoWmsToE+vYgD4nytZVxcu1ulHpUQRAB1UJ8+N1Am3Mz4+xOByoQoSZf4D+CpkA==} @@ -4477,6 +4486,12 @@ packages: peerDependencies: '@opentelemetry/api': ^1.3.0 + '@opentelemetry/exporter-trace-otlp-http@0.208.0': + resolution: {integrity: sha512-jbzDw1q+BkwKFq9yxhjAJ9rjKldbt5AgIy1gmEIJjEV/WRxQ3B6HcLVkwbjJ3RcMif86BDNKR846KJ0tY0aOJA==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': ^1.3.0 + '@opentelemetry/instrumentation@0.214.0': resolution: {integrity: sha512-MHqEX5Dk59cqVah5LiARMACku7jXSVk9iVDWOea4x3cr7VfdByeDCURK6o1lntT1JS/Tsovw01UJrBhN3/uC5w==} engines: {node: ^18.19.0 || >=20.6.0} @@ -9554,8 +9569,8 @@ packages: resolution: {integrity: sha512-0je+qPKHEMohvfRTCEo3CrPG6cAzAYgmzKyxRiYSSDkS6eGJdyVJm7WaYA5ECaAD9wLB2T4EEeymA5aFVcYXCA==} engines: {node: '>=6'} - deslop-js@0.7.7: - resolution: {integrity: sha512-mUr2B3FZM+vJZK4w6ZRwuHm6cXVWM5kS4xGRjRLC0yyen/zD+/ZI7QtH/UqUN0LaEIQZL2+P8PMcsDOIB5xgPQ==} + deslop-js@0.7.8: + resolution: {integrity: sha512-QMmb3Z/ARvYZmZneudb8cnY/4mVvZTdhUyA9TC2skwOcm7KvY9zyOdn0TApQc4rL0VM2TffFkmo3ky/lJZX7qw==} destroy@1.2.0: resolution: {integrity: sha512-2sJGJTaXIIaR1w4iJSNoN0hnMY7Gpc/n8D4qSCJw8QqFWXf7cuAgnEHxBpweaVcPevC2l3KpjYCx3NypQQgaJg==} @@ -12924,8 +12939,8 @@ packages: engines: {node: ^20.19.0 || >=22.12.0} hasBin: true - oxlint-plugin-react-doctor@0.7.7: - resolution: {integrity: sha512-OMl/8k3rcxTSyMJWUL/pdq/ZZKFtv1F4kyM+AWYsj/YYaL/nftL/idrJOxyd2CuFa5Nrbz2DlK3QFRej1uwQ/Q==} + oxlint-plugin-react-doctor@0.7.8: + resolution: {integrity: sha512-3f9/jFLIC/KRLPYqxiXSk20cq47luGy9Oz5Ru7nK7w0EI9B9zuMvAU92bK9jFiwFE7Lc9PA8ER6Y+naYaGFQGw==} engines: {node: ^20.19.0 || >=22.13.0} oxlint@1.66.0: @@ -13571,8 +13586,8 @@ packages: resolution: {integrity: sha512-+NRMYs2DyTP4/tqWz371Oo50JqmWltR1h2gcdgUMAWZJIAvrd0/SqlCfx7tpzpl/s36rzw6qH2MjoNrxtRNYhA==} engines: {node: ^20.9.0 || >=22} - react-doctor@0.7.7: - resolution: {integrity: sha512-kHlzPDZNRKlQbXsXgnFh1I4mL0ybt0xJca3/uRMZafaXR4AG5zf7iFMqtdRUiFWHeefE4vGf4YQmkYkXzE1rlg==} + react-doctor@0.7.8: + resolution: {integrity: sha512-G3spmtZJE/gWWPRJ3rpgUWTPRDJpEmdRja7iNZ7RAXlfpEO+NWVzPTca/cPI9hLwPo2Aq5/BZggo5JDBrwGrlA==} engines: {node: ^20.19.0 || >=22.13.0} hasBin: true @@ -19190,6 +19205,13 @@ snapshots: '@tybys/wasm-util': 0.10.2 optional: true + '@napi-rs/wasm-runtime@1.1.6(@emnapi/core@1.10.0)(@emnapi/runtime@1.10.0)': + dependencies: + '@emnapi/core': 1.10.0 + '@emnapi/runtime': 1.10.0 + '@tybys/wasm-util': 0.10.3 + optional: true + '@napi-rs/wasm-runtime@1.1.6(@emnapi/core@1.11.1)(@emnapi/runtime@1.11.1)': dependencies: '@emnapi/core': 1.11.1 @@ -19300,6 +19322,15 @@ snapshots: '@opentelemetry/otlp-transformer': 0.208.0(@opentelemetry/api@1.9.1) '@opentelemetry/sdk-logs': 0.208.0(@opentelemetry/api@1.9.1) + '@opentelemetry/exporter-trace-otlp-http@0.208.0(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/core': 2.2.0(@opentelemetry/api@1.9.1) + '@opentelemetry/otlp-exporter-base': 0.208.0(@opentelemetry/api@1.9.1) + '@opentelemetry/otlp-transformer': 0.208.0(@opentelemetry/api@1.9.1) + '@opentelemetry/resources': 2.2.0(@opentelemetry/api@1.9.1) + '@opentelemetry/sdk-trace-base': 2.2.0(@opentelemetry/api@1.9.1) + '@opentelemetry/instrumentation@0.214.0(@opentelemetry/api@1.9.1)': dependencies: '@opentelemetry/api': 1.9.1 @@ -19362,7 +19393,7 @@ snapshots: '@opentelemetry/api': 1.9.1 '@opentelemetry/core': 2.2.0(@opentelemetry/api@1.9.1) '@opentelemetry/resources': 2.2.0(@opentelemetry/api@1.9.1) - '@opentelemetry/semantic-conventions': 1.39.0 + '@opentelemetry/semantic-conventions': 1.41.1 '@opentelemetry/sdk-trace-base@2.8.0(@opentelemetry/api@1.9.1)': dependencies: @@ -21257,7 +21288,7 @@ snapshots: '@sentry/core@10.61.0': {} - '@sentry/node-core@10.61.0(@opentelemetry/api@1.9.1)(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(@opentelemetry/instrumentation@0.214.0(@opentelemetry/api@1.9.1))(@opentelemetry/sdk-trace-base@2.8.0(@opentelemetry/api@1.9.1))': + '@sentry/node-core@10.61.0(@opentelemetry/api@1.9.1)(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(@opentelemetry/exporter-trace-otlp-http@0.208.0(@opentelemetry/api@1.9.1))(@opentelemetry/instrumentation@0.214.0(@opentelemetry/api@1.9.1))(@opentelemetry/sdk-trace-base@2.8.0(@opentelemetry/api@1.9.1))': dependencies: '@sentry/conventions': 0.12.0 '@sentry/core': 10.61.0 @@ -21266,17 +21297,18 @@ snapshots: optionalDependencies: '@opentelemetry/api': 1.9.1 '@opentelemetry/core': 2.8.0(@opentelemetry/api@1.9.1) + '@opentelemetry/exporter-trace-otlp-http': 0.208.0(@opentelemetry/api@1.9.1) '@opentelemetry/instrumentation': 0.214.0(@opentelemetry/api@1.9.1) '@opentelemetry/sdk-trace-base': 2.8.0(@opentelemetry/api@1.9.1) - '@sentry/node@10.61.0(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))': + '@sentry/node@10.61.0(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(@opentelemetry/exporter-trace-otlp-http@0.208.0(@opentelemetry/api@1.9.1))': dependencies: '@opentelemetry/api': 1.9.1 '@opentelemetry/instrumentation': 0.214.0(@opentelemetry/api@1.9.1) '@opentelemetry/sdk-trace-base': 2.8.0(@opentelemetry/api@1.9.1) '@opentelemetry/semantic-conventions': 1.41.1 '@sentry/core': 10.61.0 - '@sentry/node-core': 10.61.0(@opentelemetry/api@1.9.1)(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(@opentelemetry/instrumentation@0.214.0(@opentelemetry/api@1.9.1))(@opentelemetry/sdk-trace-base@2.8.0(@opentelemetry/api@1.9.1)) + '@sentry/node-core': 10.61.0(@opentelemetry/api@1.9.1)(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(@opentelemetry/exporter-trace-otlp-http@0.208.0(@opentelemetry/api@1.9.1))(@opentelemetry/instrumentation@0.214.0(@opentelemetry/api@1.9.1))(@opentelemetry/sdk-trace-base@2.8.0(@opentelemetry/api@1.9.1)) '@sentry/opentelemetry': 10.61.0(@opentelemetry/api@1.9.1)(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(@opentelemetry/sdk-trace-base@2.8.0(@opentelemetry/api@1.9.1)) '@sentry/server-utils': 10.61.0 import-in-the-middle: 3.2.0 @@ -22617,7 +22649,7 @@ snapshots: dependencies: '@emnapi/core': 1.10.0 '@emnapi/runtime': 1.10.0 - '@napi-rs/wasm-runtime': 1.1.5(@emnapi/core@1.10.0)(@emnapi/runtime@1.10.0) + '@napi-rs/wasm-runtime': 1.1.6(@emnapi/core@1.10.0)(@emnapi/runtime@1.10.0) optional: true '@unrs/resolver-binding-win32-arm64-msvc@1.12.2': @@ -24193,7 +24225,7 @@ snapshots: dequal@2.0.3: {} - deslop-js@0.7.7: + deslop-js@0.7.8: dependencies: '@oxc-project/types': 0.138.0 fast-glob: 3.3.3 @@ -28570,7 +28602,7 @@ snapshots: '@oxfmt/binding-win32-ia32-msvc': 0.45.0 '@oxfmt/binding-win32-x64-msvc': 0.45.0 - oxlint-plugin-react-doctor@0.7.7: + oxlint-plugin-react-doctor@0.7.8: dependencies: '@typescript-eslint/types': 8.62.0 eslint-scope: 9.1.2 @@ -29323,19 +29355,19 @@ snapshots: transitivePeerDependencies: - supports-color - react-doctor@0.7.7(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(eslint@10.5.0(jiti@2.7.0)): + react-doctor@0.7.8(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(@opentelemetry/exporter-trace-otlp-http@0.208.0(@opentelemetry/api@1.9.1))(eslint@10.5.0(jiti@2.7.0)): dependencies: '@babel/code-frame': 7.29.0 - '@sentry/node': 10.61.0(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1)) + '@sentry/node': 10.61.0(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(@opentelemetry/exporter-trace-otlp-http@0.208.0(@opentelemetry/api@1.9.1)) agent-install: 0.0.5 conf: 15.1.0 confbox: 0.2.4 - deslop-js: 0.7.7 + deslop-js: 0.7.8 eslint-plugin-react-hooks: 7.1.1(eslint@10.5.0(jiti@2.7.0)) jiti: 2.7.0 magicast: 0.5.3 oxlint: 1.66.0 - oxlint-plugin-react-doctor: 0.7.7 + oxlint-plugin-react-doctor: 0.7.8 prompts: 2.4.2 typescript: 5.9.3 vscode-languageserver: 9.0.1 @@ -29593,7 +29625,7 @@ snapshots: react: 19.2.6 react-dom: 19.2.6(react@19.2.6) - react-scan@0.5.7(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(esbuild@0.27.2)(eslint@10.5.0(jiti@2.7.0))(react-dom@19.2.6(react@19.2.6))(react@19.2.6)(rollup@4.57.1): + react-scan@0.5.7(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(@opentelemetry/exporter-trace-otlp-http@0.208.0(@opentelemetry/api@1.9.1))(esbuild@0.27.2)(eslint@10.5.0(jiti@2.7.0))(react-dom@19.2.6(react@19.2.6))(react@19.2.6)(rollup@4.57.1): dependencies: '@babel/core': 7.29.0 '@babel/types': 7.29.7 @@ -29605,7 +29637,7 @@ snapshots: preact: 10.29.2 prompts: 2.4.2 react: 19.2.6 - react-doctor: 0.7.7(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(eslint@10.5.0(jiti@2.7.0)) + react-doctor: 0.7.8(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.1))(@opentelemetry/exporter-trace-otlp-http@0.208.0(@opentelemetry/api@1.9.1))(eslint@10.5.0(jiti@2.7.0)) react-dom: 19.2.6(react@19.2.6) react-grab: 0.1.48(react@19.2.6) optionalDependencies: