From 5c5431d77774e60684efad9545184ff97f8bb8db Mon Sep 17 00:00:00 2001 From: Alessandro Pogliaghi Date: Thu, 30 Jul 2026 09:56:18 +0100 Subject: [PATCH 1/3] fix(agent): add session initialization timeout diagnostics Generated-By: PostHog Code Task-Id: cd137de1-90a0-4cb9-8509-56f5f210c0e9 --- .../claude/claude-agent.resume-model.test.ts | 49 ++++++++++++++++++- .../agent/src/adapters/claude/claude-agent.ts | 13 ++++- 2 files changed, 59 insertions(+), 3 deletions(-) diff --git a/packages/agent/src/adapters/claude/claude-agent.resume-model.test.ts b/packages/agent/src/adapters/claude/claude-agent.resume-model.test.ts index b16a800f82..5b5d535d0a 100644 --- a/packages/agent/src/adapters/claude/claude-agent.resume-model.test.ts +++ b/packages/agent/src/adapters/claude/claude-agent.resume-model.test.ts @@ -1,7 +1,10 @@ import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; import * as os from "node:os"; import * as path from "node:path"; -import type { AgentSideConnection } from "@agentclientprotocol/sdk"; +import { + type AgentSideConnection, + RequestError, +} from "@agentclientprotocol/sdk"; import type { HookInput, Options } from "@anthropic-ai/claude-agent-sdk"; import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; import { DEFAULT_GATEWAY_MODEL } from "../../gateway-models"; @@ -453,6 +456,50 @@ describe("ClaudeAcpAgent session creation", () => { expect(createdQueries[0]?.close).toHaveBeenCalledTimes(1); }); + it("logs diagnostics and closes the query when new-session init times out", async () => { + vi.useFakeTimers(); + try { + nextInitPromise = new Promise(() => {}); + const agent = makeAgent(); + const errorSpy = vi.spyOn(agent.logger, "error"); + + const promise = agent.newSession({ + cwd, + mcpServers: [], + _meta: { + taskRunId: "run-init-timeout-new", + model: "claude-opus-5", + }, + }); + promise.catch(() => {}); + + await vi.waitFor(() => { + expect(createdQueries[0]?.initializationResult).toHaveBeenCalledTimes( + 1, + ); + }); + await vi.advanceTimersByTimeAsync(30_001); + + await expect(promise).rejects.toBeInstanceOf(RequestError); + expect(createdQueries[0]?.close).toHaveBeenCalledTimes(1); + expect(errorSpy).toHaveBeenCalledWith( + "Session initialization failed", + expect.objectContaining({ + initializationPhase: "sdk_initialization", + timeoutMs: 30_000, + initMs: expect.any(Number), + requestedModel: "claude-opus-5", + gatewayConfigured: false, + errorDetail: expect.objectContaining({ + message: "Session initialization timed out after 30000ms", + }), + }), + ); + } finally { + vi.useRealTimers(); + } + }); + it("closes the query and rethrows when resume init fails", async () => { const failedInit = Promise.reject(new Error("resume boom")); failedInit.catch(() => {}); diff --git a/packages/agent/src/adapters/claude/claude-agent.ts b/packages/agent/src/adapters/claude/claude-agent.ts index 9189b1e04d..e07a2c404f 100644 --- a/packages/agent/src/adapters/claude/claude-agent.ts +++ b/packages/agent/src/adapters/claude/claude-agent.ts @@ -2197,10 +2197,12 @@ export class ClaudeAcpAgent extends BaseAcpAgent { const initPromise = !isResume ? withTimeout(q.initializationResult(), SESSION_VALIDATION_TIMEOUT_MS) : undefined; + const requestedModel = + meta?.model || settingsManager.getSettings().model || undefined; const [rawModelOptions] = await Promise.all([ this.getModelConfigOptions( - meta?.model || settingsManager.getSettings().model || undefined, + requestedModel, this.options?.gatewayEnv?.anthropicBaseUrl, this.options?.gatewayEnv?.anthropicAuthToken, ), @@ -2252,12 +2254,19 @@ export class ClaudeAcpAgent extends BaseAcpAgent { } catch (err) { settingsManager.dispose(); this.terminateQuery(q, abortController); + const initMs = Date.now() - initStartedAt; this.logger.error("Session initialization failed", { sessionId, taskId, taskRunId: meta?.taskRunId, + initializationPhase: "sdk_initialization", + timeoutMs: SESSION_VALIDATION_TIMEOUT_MS, modelConfigMs, - initMs: Date.now() - initStartedAt, + initMs, + requestedModel: requestedModel ?? null, + gatewayConfigured: Boolean( + this.options?.gatewayEnv?.anthropicBaseUrl, + ), errorDetail: serializeError(err), }); throw err; From a612a526a47d296f00393151b3d288e6822da806 Mon Sep 17 00:00:00 2001 From: Alessandro Pogliaghi Date: Thu, 30 Jul 2026 09:56:20 +0100 Subject: [PATCH 2/3] fix(agent): add Codex and Pi initialization diagnostics Generated-By: PostHog Code Task-Id: cd137de1-90a0-4cb9-8509-56f5f210c0e9 --- .../codex-app-server-agent.test.ts | 39 +++++++++++++++++++ .../codex-app-server-agent.ts | 33 ++++++++++++++-- .../agent/src/server/pi-agent-server.test.ts | 38 ++++++++++++++++++ packages/agent/src/server/pi-agent-server.ts | 16 ++++++++ 4 files changed, 122 insertions(+), 4 deletions(-) diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts index def493b9e8..1dbfc4b87d 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts @@ -101,6 +101,45 @@ function makeFakeClient( const init = { protocolVersion: 1 } as unknown as InitializeRequest; describe("CodexAppServerAgent", () => { + it("logs session initialization diagnostics when thread setup fails", async () => { + const stub = makeStubRpc({ + "thread/start": new Error("thread setup failed"), + }); + const { client } = makeFakeClient(); + const agent = new CodexAppServerAgent(client, { + processOptions: { + binaryPath: "/bundle/codex", + apiBaseUrl: "https://gateway.example.com", + }, + model: "gpt-5.5", + rpcFactory: stub.factory, + }); + const errorSpy = vi.spyOn(agent.logger, "error"); + + await expect( + agent.newSession({ + cwd: "/repo", + _meta: { taskId: "task-1", taskRunId: "run-1" }, + } as unknown as NewSessionRequest), + ).rejects.toThrow("thread setup failed"); + + expect(errorSpy).toHaveBeenCalledWith( + "Codex app-server session initialization failed", + expect.objectContaining({ + runtimeAdapter: "codex", + initializationPhase: "thread/start", + initMs: expect.any(Number), + requestedModel: "gpt-5.5", + gatewayConfigured: true, + taskId: "task-1", + taskRunId: "run-1", + errorDetail: expect.objectContaining({ + message: "thread setup failed", + }), + }), + ); + }); + it("runs initialize -> thread/start -> turn/start and streams agent text", async () => { const stub = makeStubRpc({ initialize: {}, diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts index 74a7c15390..25e54abe5b 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts @@ -25,6 +25,7 @@ import { classifyGatewayLimitError, mcpToolKey, posthogToolMeta, + serializeError, } from "@posthog/shared"; import { type NativeGoalState, @@ -251,6 +252,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { ) => Promise; /** Codex-specific guidance injected at spawn time; replayed per-thread. */ private readonly developerInstructions?: string; + private readonly gatewayConfigured: boolean; private threadId?: string; /** JSON schema constraining the final message; set per session via `_meta`. */ private jsonSchema?: Record; @@ -306,6 +308,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { ); this.onStructuredOutput = options.onStructuredOutput; this.developerInstructions = options.processOptions.developerInstructions; + this.gatewayConfigured = Boolean(options.processOptions.apiBaseUrl); const handlers: AppServerClientHandlers = { logger: this.logger, @@ -388,7 +391,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { } async newSession(params: NewSessionRequest): Promise { - const { threadId } = await this.setupThread( + const { threadId } = await this.setupThreadWithDiagnostics( APP_SERVER_METHODS.THREAD_START, { cwd: params.cwd, @@ -403,7 +406,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { async resumeSession( params: ResumeSessionRequest, ): Promise { - await this.setupThread(APP_SERVER_METHODS.THREAD_RESUME, { + await this.setupThreadWithDiagnostics(APP_SERVER_METHODS.THREAD_RESUME, { cwd: params.cwd, mcpServers: params.mcpServers, meta: params._meta as AppServerSessionMeta | undefined, @@ -415,7 +418,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { /** Re-attach to an existing thread without starting a turn: resume it, then replay the transcript. */ async loadSession(params: LoadSessionRequest): Promise { - const { thread } = await this.setupThread( + const { thread } = await this.setupThreadWithDiagnostics( APP_SERVER_METHODS.THREAD_RESUME, { cwd: params.cwd, @@ -432,7 +435,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { async unstable_forkSession( params: ForkSessionRequest, ): Promise { - const { threadId } = await this.setupThread( + const { threadId } = await this.setupThreadWithDiagnostics( APP_SERVER_METHODS.THREAD_FORK, { cwd: params.cwd, @@ -485,6 +488,28 @@ export class CodexAppServerAgent extends BaseAcpAgent { } } + private async setupThreadWithDiagnostics( + method: string, + params: Parameters[1], + ): ReturnType { + const initStartedAt = Date.now(); + try { + return await this.setupThread(method, params); + } catch (error) { + this.logger.error("Codex app-server session initialization failed", { + runtimeAdapter: "codex", + initializationPhase: method, + initMs: Date.now() - initStartedAt, + requestedModel: this.config.model, + gatewayConfigured: this.gatewayConfigured, + taskId: params.meta?.taskId, + taskRunId: params.meta?.taskRunId, + errorDetail: serializeError(error), + }); + throw error; + } + } + /** Shared thread setup for start/resume/fork. `threadId` present => resume/fork; absent => new thread. */ private async setupThread( method: string, diff --git a/packages/agent/src/server/pi-agent-server.test.ts b/packages/agent/src/server/pi-agent-server.test.ts index c185666512..23a5a79f09 100644 --- a/packages/agent/src/server/pi-agent-server.test.ts +++ b/packages/agent/src/server/pi-agent-server.test.ts @@ -22,6 +22,44 @@ function config(overrides: Partial = {}): AgentServerConfig { } describe("PiAgentServer", () => { + it("logs session initialization diagnostics when setup fails", async () => { + const payload = { task_id: "task-1", run_id: "run-1" }; + const server = new PiAgentServer( + config({ model: "anthropic/claude-opus-5" }), + ) as unknown as { + logger: { error: ReturnType }; + createSession(sessionPayload: typeof payload): Promise; + initializeSession( + sessionPayload: typeof payload, + sseController: null, + ): Promise; + }; + server.createSession = vi.fn(async () => { + throw new Error("Pi RPC startup failed"); + }); + const errorSpy = vi.spyOn(server.logger, "error"); + + await expect(server.initializeSession(payload, null)).rejects.toThrow( + "Pi RPC startup failed", + ); + + expect(errorSpy).toHaveBeenCalledWith( + "Pi session initialization failed", + expect.objectContaining({ + runtimeAdapter: "pi", + initializationPhase: "session_setup", + initMs: expect.any(Number), + requestedModel: "anthropic/claude-opus-5", + gatewayConfigured: true, + taskId: "task-1", + taskRunId: "run-1", + errorDetail: expect.objectContaining({ + message: "Pi RPC startup failed", + }), + }), + ); + }); + it.each([ ["task", { task_id: "task-2", run_id: "run-1", team_id: 1 }], ["run", { task_id: "task-1", run_id: "run-2", team_id: 1 }], diff --git a/packages/agent/src/server/pi-agent-server.ts b/packages/agent/src/server/pi-agent-server.ts index 6a754f4758..bd9e1bfb76 100644 --- a/packages/agent/src/server/pi-agent-server.ts +++ b/packages/agent/src/server/pi-agent-server.ts @@ -8,6 +8,7 @@ import type { StoredLogEntry, TaskRunArtifact, } from "@posthog/shared"; +import { serializeError } from "@posthog/shared"; import { Hono } from "hono"; import { z } from "zod/v4"; import { createPiRpcClient, type PiRpcClient } from "../pi/rpc-client"; @@ -383,8 +384,23 @@ export class PiAgentServer { const initializationPromise = this.createSession(payload); this.initializationPromise = initializationPromise; + const initStartedAt = Date.now(); try { await initializationPromise; + } catch (error) { + this.logger.error("Pi session initialization failed", { + runtimeAdapter: "pi", + initializationPhase: "session_setup", + initMs: Date.now() - initStartedAt, + requestedModel: this.config.model ?? null, + gatewayConfigured: Boolean( + process.env.LLM_GATEWAY_URL || this.config.apiUrl, + ), + taskId: payload.task_id, + taskRunId: payload.run_id, + errorDetail: serializeError(error), + }); + throw error; } finally { if (this.initializationPromise === initializationPromise) { this.initializationPromise = null; From 1f9762458aee46165212251c9b9cbedd2780b532 Mon Sep 17 00:00:00 2001 From: Alessandro Pogliaghi Date: Thu, 30 Jul 2026 09:56:21 +0100 Subject: [PATCH 3/3] fix(agent): export initialization failures to OTLP Generated-By: PostHog Code Task-Id: cd137de1-90a0-4cb9-8509-56f5f210c0e9 --- packages/agent/src/acp-extensions.ts | 3 + packages/agent/src/otel-telemetry.test.ts | 38 +++++++++++++ packages/agent/src/otel-telemetry.ts | 14 +++++ .../agent/src/server/agent-server.test.ts | 52 +++++++++++++++++ packages/agent/src/server/agent-server.ts | 30 ++++++++++ .../agent/src/server/pi-agent-server.test.ts | 18 ++++++ packages/agent/src/server/pi-agent-server.ts | 56 +++++++++++++++++++ 7 files changed, 211 insertions(+) diff --git a/packages/agent/src/acp-extensions.ts b/packages/agent/src/acp-extensions.ts index c700c32bb7..3a3bd9dab5 100644 --- a/packages/agent/src/acp-extensions.ts +++ b/packages/agent/src/acp-extensions.ts @@ -33,6 +33,9 @@ export const POSTHOG_NOTIFICATIONS = { /** Error occurred during task execution */ ERROR: "_posthog/error", + /** Agent runtime failed before its session became ready */ + INITIALIZATION_FAILED: "_posthog/initialization_failed", + /** Console/log output from the agent */ CONSOLE: "_posthog/console", diff --git a/packages/agent/src/otel-telemetry.test.ts b/packages/agent/src/otel-telemetry.test.ts index 9b761521b7..66ac11c759 100644 --- a/packages/agent/src/otel-telemetry.test.ts +++ b/packages/agent/src/otel-telemetry.test.ts @@ -176,6 +176,30 @@ describe("OtelRunTelemetry", () => { body: "run error", attrs: { error_source: "agent_server", stop_reason: "error" }, }, + { + name: "initialization_failed", + entry: makeEntry("_posthog/initialization_failed", { + runtimeAdapter: "codex", + initializationPhase: "session_setup", + initMs: 12_345, + requestedModel: "gpt-5.2-codex", + gatewayConfigured: true, + errorType: "timeout", + timeoutMs: 30_000, + errorDetail: "SECRET provider response", + }), + severityText: "ERROR", + body: "agent initialization failed", + attrs: { + runtime_adapter: "codex", + initialization_phase: "session_setup", + init_ms: 12_345, + requested_model: "gpt-5.2-codex", + gateway_configured: true, + error_type: "timeout", + timeout_ms: 30_000, + }, + }, { name: "progress", entry: makeEntry("_posthog/progress", { @@ -346,6 +370,20 @@ describe("OtelRunTelemetry", () => { ); }); + it("never exports initialization error detail", () => { + const mapped = mapNotificationToLogRecord( + makeEntry("_posthog/initialization_failed", { + runtimeAdapter: "pi", + errorDetail: "SECRET provider response", + }), + ); + + expect(mapped).not.toBeNull(); + expect(JSON.stringify([mapped?.body, mapped?.attributes])).not.toContain( + "SECRET", + ); + }); + it("caps body length", () => { const mapped = mapNotificationToLogRecord( makeEntry("_posthog/progress", { diff --git a/packages/agent/src/otel-telemetry.ts b/packages/agent/src/otel-telemetry.ts index 8d7456ad04..00f5fbc96a 100644 --- a/packages/agent/src/otel-telemetry.ts +++ b/packages/agent/src/otel-telemetry.ts @@ -19,6 +19,7 @@ import { entryTime, MAX_BODY_CHARS, normalizeMethod, + numAttr, strAttr, truncate, usageAttributes, @@ -200,6 +201,19 @@ export function mapNotificationToLogRecord( strAttr(attrs, "stop_reason", params.stopReason); return record(ERROR, "run error", method, attrs); } + case POSTHOG_NOTIFICATIONS.INITIALIZATION_FAILED: { + const attrs: Attributes = {}; + strAttr(attrs, "runtime_adapter", params.runtimeAdapter); + strAttr(attrs, "initialization_phase", params.initializationPhase); + numAttr(attrs, "init_ms", params.initMs); + strAttr(attrs, "requested_model", params.requestedModel); + if (typeof params.gatewayConfigured === "boolean") { + attrs.gateway_configured = params.gatewayConfigured; + } + strAttr(attrs, "error_type", params.errorType); + numAttr(attrs, "timeout_ms", params.timeoutMs); + return record(ERROR, "agent initialization failed", 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 diff --git a/packages/agent/src/server/agent-server.test.ts b/packages/agent/src/server/agent-server.test.ts index 016d96a80d..5be8bf1c84 100644 --- a/packages/agent/src/server/agent-server.test.ts +++ b/packages/agent/src/server/agent-server.test.ts @@ -448,6 +448,58 @@ describe("AgentServer HTTP Mode", () => { ); }; + it("exports safe telemetry when session initialization fails", async () => { + const append = vi.fn(); + const shutdown = vi.fn(async () => {}); + const testServer = createServer({ + runtimeAdapter: "codex", + model: "gpt-5.2-codex", + }) as unknown as { + initializingTelemetry: { + append: typeof append; + shutdown: typeof shutdown; + }; + _doInitializeSession( + payload: JwtPayload, + controller: null, + ): Promise; + initializeSession(payload: JwtPayload, controller: null): Promise; + }; + testServer._doInitializeSession = vi.fn(async () => { + testServer.initializingTelemetry = { append, shutdown }; + throw new Error("SECRET provider response"); + }); + const payload = { + task_id: "test-task-id", + run_id: "test-run-id", + team_id: 1, + user_id: 1, + distinct_id: "test-distinct-id", + mode: "interactive" as const, + }; + + await expect(testServer.initializeSession(payload, null)).rejects.toThrow( + "SECRET provider response", + ); + + expect(append).toHaveBeenCalledWith( + "test-run-id", + expect.objectContaining({ + notification: expect.objectContaining({ + method: POSTHOG_NOTIFICATIONS.INITIALIZATION_FAILED, + params: expect.objectContaining({ + runtimeAdapter: "codex", + initializationPhase: "session_setup", + requestedModel: "gpt-5.2-codex", + errorType: "error", + }), + }), + }), + ); + expect(JSON.stringify(append.mock.calls)).not.toContain("SECRET"); + expect(shutdown).toHaveBeenCalledOnce(); + }); + it("replays ACP notifications emitted before cloud session assignment", () => { const testServer = createServer() as unknown as { session: { sseController: null } | null; diff --git a/packages/agent/src/server/agent-server.ts b/packages/agent/src/server/agent-server.ts index c7fad47310..274fe25b7b 100644 --- a/packages/agent/src/server/agent-server.ts +++ b/packages/agent/src/server/agent-server.ts @@ -401,6 +401,7 @@ export class AgentServer { // often arrives while newSession() is still awaited (this.session is still null), // causing a second session to be created and duplicate Slack messages to be sent. private initializationPromise: Promise | null = null; + private initializingTelemetry: OtelRunTelemetry | undefined; private pendingEvents: Record[] = []; /** ACP notifications emitted by newSession/resumeSession before this.session is assigned. */ private preSessionEvents: Record[] = []; @@ -1496,9 +1497,36 @@ export class AgentServer { payload, sseController, ); + const initStartedAt = Date.now(); try { await this.initializationPromise; + } catch (error) { + const telemetry = this.initializingTelemetry; + telemetry?.append(payload.run_id, { + type: "notification", + timestamp: new Date().toISOString(), + notification: { + jsonrpc: "2.0", + method: POSTHOG_NOTIFICATIONS.INITIALIZATION_FAILED, + params: { + runtimeAdapter: this.getRuntimeAdapter(), + initializationPhase: "session_setup", + initMs: Date.now() - initStartedAt, + requestedModel: this.config.model, + gatewayConfigured: Boolean( + process.env.LLM_GATEWAY_URL || this.config.apiUrl, + ), + errorType: + error instanceof Error && error.name === "TimeoutError" + ? "timeout" + : "error", + }, + }, + }); + await telemetry?.shutdown(); + throw error; } finally { + this.initializingTelemetry = undefined; this.initializationPromise = null; } } @@ -1612,6 +1640,7 @@ export class AgentServer { deviceInfo, runtimeAdapter, ); + this.initializingTelemetry = telemetry; const logWriter = new SessionLogWriter({ posthogAPI, @@ -1847,6 +1876,7 @@ export class AgentServer { pendingHandoffGitState: undefined, sessionMeta: effectiveSessionMeta, }; + this.initializingTelemetry = undefined; this.flushPreSessionEvents(); this.logger = new Logger({ diff --git a/packages/agent/src/server/pi-agent-server.test.ts b/packages/agent/src/server/pi-agent-server.test.ts index 23a5a79f09..21c2463223 100644 --- a/packages/agent/src/server/pi-agent-server.test.ts +++ b/packages/agent/src/server/pi-agent-server.test.ts @@ -33,7 +33,11 @@ describe("PiAgentServer", () => { sessionPayload: typeof payload, sseController: null, ): Promise; + createRunTelemetry: ReturnType; }; + const append = vi.fn(); + const shutdown = vi.fn(async () => {}); + server.createRunTelemetry = vi.fn(() => ({ append, shutdown })); server.createSession = vi.fn(async () => { throw new Error("Pi RPC startup failed"); }); @@ -58,6 +62,20 @@ describe("PiAgentServer", () => { }), }), ); + expect(append).toHaveBeenCalledWith( + "run-1", + expect.objectContaining({ + notification: expect.objectContaining({ + method: "_posthog/initialization_failed", + params: expect.objectContaining({ + runtimeAdapter: "pi", + initializationPhase: "session_setup", + errorType: "error", + }), + }), + }), + ); + expect(shutdown).toHaveBeenCalledOnce(); }); it.each([ diff --git a/packages/agent/src/server/pi-agent-server.ts b/packages/agent/src/server/pi-agent-server.ts index bd9e1bfb76..0fb220e900 100644 --- a/packages/agent/src/server/pi-agent-server.ts +++ b/packages/agent/src/server/pi-agent-server.ts @@ -11,6 +11,8 @@ import type { import { serializeError } from "@posthog/shared"; import { Hono } from "hono"; import { z } from "zod/v4"; +import { POSTHOG_NOTIFICATIONS } from "../acp-extensions"; +import { OtelRunTelemetry } from "../otel-telemetry"; import { createPiRpcClient, type PiRpcClient } from "../pi/rpc-client"; import { piRpcCommandSchema, type RpcCommand } from "../pi/rpc-transport"; import { PiRuntime } from "../pi/runtime"; @@ -132,6 +134,37 @@ export class PiAgentServer { this.app = this.createApp(); } + private createRunTelemetry( + payload: JwtPayload, + ): 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: "cloud", + teamId: payload.team_id, + userId: payload.user_id, + distinctId: payload.distinct_id, + adapter: "pi", + mode: payload.mode ?? this.config.mode, + agentVersion: this.config.version, + }, + new Logger({ debug: false, prefix: "[OtelRunTelemetry]" }), + ); + } catch (error) { + this.logger.warn("Failed to initialize OTel run telemetry", error); + return undefined; + } + } + async start(): Promise { await new Promise((resolve) => { this.server = serve( @@ -388,6 +421,29 @@ export class PiAgentServer { try { await initializationPromise; } catch (error) { + const telemetry = this.createRunTelemetry(payload); + telemetry?.append(payload.run_id, { + type: "notification", + timestamp: new Date().toISOString(), + notification: { + jsonrpc: "2.0", + method: POSTHOG_NOTIFICATIONS.INITIALIZATION_FAILED, + params: { + runtimeAdapter: "pi", + initializationPhase: "session_setup", + initMs: Date.now() - initStartedAt, + requestedModel: this.config.model, + gatewayConfigured: Boolean( + process.env.LLM_GATEWAY_URL || this.config.apiUrl, + ), + errorType: + error instanceof Error && error.name === "TimeoutError" + ? "timeout" + : "error", + }, + }, + }); + await telemetry?.shutdown(); this.logger.error("Pi session initialization failed", { runtimeAdapter: "pi", initializationPhase: "session_setup",