diff --git a/packages/core/src/sessions/sessionService.ts b/packages/core/src/sessions/sessionService.ts index 1ea7c02b8a..75435aacb2 100644 --- a/packages/core/src/sessions/sessionService.ts +++ b/packages/core/src/sessions/sessionService.ts @@ -22,6 +22,7 @@ import { isJsonRpcRequest, isJsonRpcResponse, isRateLimitError, + isTransientUpstreamError, mergeConfigOptions, type OptimisticItem, type PermissionRequest, @@ -903,7 +904,11 @@ export class SessionService { const previous = this.d.store.getSessions()[taskRunId]; const session = createBaseSession(taskRunId, taskId, taskTitle); - session.events = events; + // Repainting from the log must not blank a transcript we already hold: + // fetchSessionLogs swallows read errors and returns empty, so keep the + // previous in-memory events when the log read produced nothing. + session.events = + events.length === 0 && previous?.events.length ? previous.events : events; if (logUrl) { session.logUrl = logUrl; } @@ -2460,6 +2465,21 @@ export class SessionService { }); } + // A provider request that timed out or dropped leaves the session + // healthy — no recovery ran above — so tell the user to just re-send + // instead of surfacing the raw "Internal error: API Error: …" text. + if (isTransientUpstreamError(errorMessage, errorDetails)) { + this.d.log.warn("Transient upstream provider failure during prompt", { + taskRunId: session.taskRunId, + errorMessage, + errorDetails, + }); + throw new Error( + "The AI provider timed out or dropped the connection. Your session is unaffected — please send the message again.", + { cause: error }, + ); + } + throw error; } } @@ -3392,14 +3412,17 @@ export class SessionService { * effect loop. * * If the session failed before any conversation started (has an - * initialPrompt saved from the original creation attempt), creates - * a fresh session and re-sends the prompt instead of reconnecting - * to an empty session. + * initialPrompt saved from the original creation attempt, and no + * conversation in memory or in the run log), creates a fresh session + * and re-sends the prompt instead of reconnecting to an empty session. */ async clearSessionError(taskId: string, repoPath: string): Promise { this.localRepoPaths.set(taskId, repoPath); const session = this.d.store.getSessionByTaskId(taskId); - if (session?.initialPrompt?.length) { + if ( + session?.initialPrompt?.length && + !(await this.runHasConversationHistory(session)) + ) { const { taskTitle, initialPrompt, @@ -3434,6 +3457,34 @@ export class SessionService { await this.reconnectInPlace(taskId, repoPath); } + /** + * Whether the run already holds conversation beyond the user's prompt + * echoes. A set `initialPrompt` alone doesn't prove the conversation never + * started: it is only cleared when a live agent event arrives, so it + * survives when the event subscription drops before the first agent event + * while the agent keeps working and logging, and error sessions carry it + * forward. Recreating the run in that state orphans the populated run log + * behind a fresh latest_run — the task's entire history disappears — so + * check the in-memory transcript first and fall back to the persisted log. + */ + private async runHasConversationHistory( + session: AgentSession, + ): Promise { + const isPromptEcho = (event: AcpMessage): boolean => + isJsonRpcRequest(event.message) && + event.message.method === "session/prompt"; + if (session.events.some((event) => !isPromptEcho(event))) { + return true; + } + const { rawEntries } = await this.fetchSessionLogs( + session.logUrl, + session.taskRunId, + ); + return convertStoredEntriesToEvents(rawEntries).some( + (event) => !isPromptEcho(event), + ); + } + /** * Start a fresh session for a task, abandoning the old conversation. * Clears the backend sessionId so the next reconnect creates a new diff --git a/packages/core/src/sessions/sessionServiceRetryConfig.test.ts b/packages/core/src/sessions/sessionServiceRetryConfig.test.ts index 6c99ac3dc5..2e482d41d3 100644 --- a/packages/core/src/sessions/sessionServiceRetryConfig.test.ts +++ b/packages/core/src/sessions/sessionServiceRetryConfig.test.ts @@ -1,4 +1,4 @@ -import type { AgentSession } from "@posthog/shared"; +import type { AcpMessage, AgentSession } from "@posthog/shared"; import type { Task } from "@posthog/shared/domain-types"; import { describe, expect, it, vi } from "vitest"; import { @@ -7,6 +7,32 @@ import { type SessionServiceDeps, } from "./sessionService"; +const PROMPT_ECHO_EVENT: AcpMessage = { + type: "acp_message", + ts: 1, + message: { + jsonrpc: "2.0", + id: 1, + method: "session/prompt", + params: { prompt: [{ type: "text", text: "Ship the fix" }] }, + } as AcpMessage["message"], +}; + +const AGENT_MESSAGE_EVENT: AcpMessage = { + type: "acp_message", + ts: 2, + message: { + jsonrpc: "2.0", + method: "session/update", + params: { + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Working on it" }, + }, + }, + } as AcpMessage["message"], +}; + function makeSession(overrides: Partial = {}): AgentSession { return { taskRunId: "run-1", @@ -66,8 +92,28 @@ function createHarness(session: AgentSession) { "createNewLocalSession", ) .mockResolvedValue(undefined); + const reconnectInPlace = vi + .spyOn( + service as unknown as { + reconnectInPlace: (...args: unknown[]) => Promise; + }, + "reconnectInPlace", + ) + .mockResolvedValue(true); + const fetchSessionLogs = vi + .spyOn( + service as unknown as { + fetchSessionLogs: (...args: unknown[]) => Promise; + }, + "fetchSessionLogs", + ) + .mockResolvedValue({ + rawEntries: [], + totalLineCount: 0, + parseFailureCount: 0, + }); - return { service, createNewLocalSession }; + return { service, createNewLocalSession, reconnectInPlace, fetchSessionLogs }; } describe("SessionService.clearSessionError retry config", () => { @@ -94,6 +140,56 @@ describe("SessionService.clearSessionError retry config", () => { "high", // reasoningLevel ); }); + + it("reconnects in place instead of recreating when the transcript has agent events", async () => { + const session = makeSession({ + events: [PROMPT_ECHO_EVENT, AGENT_MESSAGE_EVENT], + }); + const { service, createNewLocalSession, reconnectInPlace } = + createHarness(session); + + await service.clearSessionError("task-1", "/repo"); + + expect(createNewLocalSession).not.toHaveBeenCalled(); + expect(reconnectInPlace).toHaveBeenCalledWith("task-1", "/repo"); + }); + + it("reconnects in place when the run log has history even if in-memory events are empty", async () => { + const session = makeSession({ events: [] }); + const { + service, + createNewLocalSession, + reconnectInPlace, + fetchSessionLogs, + } = createHarness(session); + fetchSessionLogs.mockResolvedValue({ + rawEntries: [ + { + type: "notification", + timestamp: "2026-07-06T00:00:00.000Z", + notification: AGENT_MESSAGE_EVENT.message, + }, + ], + totalLineCount: 1, + parseFailureCount: 0, + }); + + await service.clearSessionError("task-1", "/repo"); + + expect(createNewLocalSession).not.toHaveBeenCalled(); + expect(reconnectInPlace).toHaveBeenCalledWith("task-1", "/repo"); + }); + + it("recreates when the transcript holds only the user's prompt echo", async () => { + const session = makeSession({ events: [PROMPT_ECHO_EVENT] }); + const { service, createNewLocalSession, reconnectInPlace } = + createHarness(session); + + await service.clearSessionError("task-1", "/repo"); + + expect(createNewLocalSession).toHaveBeenCalled(); + expect(reconnectInPlace).not.toHaveBeenCalled(); + }); }); const CONNECT_PARAMS: ConnectParams = { diff --git a/packages/shared/src/errors.test.ts b/packages/shared/src/errors.test.ts index 067923c6be..15058eb0da 100644 --- a/packages/shared/src/errors.test.ts +++ b/packages/shared/src/errors.test.ts @@ -5,6 +5,7 @@ import { isFatalSessionError, isNotAuthenticatedError, isRateLimitError, + isTransientUpstreamError, NotAuthenticatedError, serializeError, } from "./errors"; @@ -103,6 +104,56 @@ describe("isFatalSessionError", () => { it("returns false for ordinary recoverable errors", () => { expect(isFatalSessionError("temporary network blip")).toBe(false); }); + + it.each([ + "Internal error: API Error: the operation timed out", + "Internal error: API Error: Request timeout", + "Internal error: API Error: terminated", + "Internal error: API Error: Connection error", + "Internal error: API Error: 529 overloaded_error", + ])("does not treat the transient upstream failure %j as fatal", (message) => { + expect(isFatalSessionError(message)).toBe(false); + }); + + it("does not treat a transient upstream failure in the details as fatal", () => { + expect( + isFatalSessionError( + "internal error", + "API Error: the operation timed out", + ), + ).toBe(false); + }); +}); + +describe("isTransientUpstreamError", () => { + it.each([ + "API Error: the operation timed out", + "API Error: terminated", + "API Error: Connection error", + "API Error: 500 internal server error", + "API Error: 529 overloaded_error", + "Internal error: API Error: request timed out", + ])("recognises %j", (message) => { + expect(isTransientUpstreamError(message)).toBe(true); + }); + + it("matches against the details when the message is generic", () => { + expect( + isTransientUpstreamError( + "Internal error", + "API Error: the operation timed out", + ), + ).toBe(true); + }); + + it.each([ + "process exited", + "session not found", + "the operation timed out", // no "API Error:" marker — not an upstream turn failure + "API Error: 400 invalid_request_error", + ])("does not match %j", (message) => { + expect(isTransientUpstreamError(message)).toBe(false); + }); }); describe("serializeError", () => { diff --git a/packages/shared/src/errors.ts b/packages/shared/src/errors.ts index c8f4d27118..0287bf6670 100644 --- a/packages/shared/src/errors.ts +++ b/packages/shared/src/errors.ts @@ -82,6 +82,22 @@ const FATAL_SESSION_ERROR_PATTERNS = [ "session not found", ] as const; +/** + * Transient upstream provider failures, as surfaced by agent adapters in + * "API Error: …" result strings (kept in sync with classifyAgentError in + * @posthog/agent). The agent process and session are healthy — a single + * provider request timed out, dropped, or returned a retryable status — so + * these must not count as fatal session errors: the fix is re-sending the + * prompt, never tearing the session down. Checked before the fatal patterns + * because the ACP layer wraps them as "Internal error: API Error: …". + */ +const UPSTREAM_TRANSIENT_ERROR_REGEXES = [ + /API Error:\s*terminated\b/i, + /API Error:\s*Connection error\b/i, + /API Error:.*\b(?:timed out|timeout)\b/i, + /API Error:\s*(?:429|5\d\d)\b/i, +] as const; + function includesAny( value: string | undefined, patterns: readonly string[], @@ -101,11 +117,22 @@ export function isRateLimitError( ); } +export function isTransientUpstreamError( + errorMessage: string, + errorDetails?: string, +): boolean { + return UPSTREAM_TRANSIENT_ERROR_REGEXES.some( + (regex) => + regex.test(errorMessage) || (!!errorDetails && regex.test(errorDetails)), + ); +} + export function isFatalSessionError( errorMessage: string, errorDetails?: string, ): boolean { if (isRateLimitError(errorMessage, errorDetails)) return false; + if (isTransientUpstreamError(errorMessage, errorDetails)) return false; return ( includesAny(errorMessage, FATAL_SESSION_ERROR_PATTERNS) || includesAny(errorDetails, FATAL_SESSION_ERROR_PATTERNS) diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index 9c71d7df07..4f555f09b0 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -75,6 +75,7 @@ export { isFatalSessionError, isNotAuthenticatedError, isRateLimitError, + isTransientUpstreamError, NotAuthenticatedError, type SerializedError, serializeError, diff --git a/packages/ui/src/features/sessions/sessionServiceHost.test.ts b/packages/ui/src/features/sessions/sessionServiceHost.test.ts index 49e3c443d9..689cb9d8bb 100644 --- a/packages/ui/src/features/sessions/sessionServiceHost.test.ts +++ b/packages/ui/src/features/sessions/sessionServiceHost.test.ts @@ -4599,6 +4599,34 @@ describe("SessionService", () => { }), ); }); + + it("does not run session recovery for a transient upstream API timeout", async () => { + const service = getSessionService(); + const mockSession = createMockSession(); + mockSessionStoreSetters.getSessionByTaskId.mockReturnValue(mockSession); + mockSessionStoreSetters.getSessions.mockReturnValue({ + "run-123": mockSession, + }); + mockTrpcAgent.prompt.mutate.mockRejectedValue( + new Error("Internal error: API Error: the operation timed out"), + ); + + await expect(service.sendPrompt("task-123", "Hello")).rejects.toThrow( + /provider timed out/, + ); + + // The session stays as-is: no recovery reconnect, no error overlay — + // only the pending-prompt state is cleared so the user can re-send. + expect(mockTrpcAgent.reconnect.mutate).not.toHaveBeenCalled(); + expect(mockSessionStoreSetters.updateSession).not.toHaveBeenCalledWith( + "run-123", + expect.objectContaining({ status: "disconnected" }), + ); + expect(mockSessionStoreSetters.updateSession).toHaveBeenCalledWith( + "run-123", + expect.objectContaining({ isPromptPending: false }), + ); + }); }); describe("local turn_complete + JSON-RPC response ordering", () => { @@ -5292,6 +5320,56 @@ describe("SessionService", () => { expect(mockTrpcAgent.reconnect.mutate).toHaveBeenCalled(); }); + it("keeps the in-memory transcript when the log read returns nothing", async () => { + const service = getSessionService(); + const previousEvents = [ + { + type: "acp_message" as const, + ts: 1, + message: { + jsonrpc: "2.0", + method: "session/update", + params: { + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Working on it" }, + }, + }, + }, + }, + ]; + const mockSession = createMockSession({ + status: "error", + logUrl: "https://logs.example.com/run-123", + events: previousEvents as AgentSession["events"], + }); + mockSessionStoreSetters.getSessionByTaskId.mockReturnValue(mockSession); + mockSessionStoreSetters.getSessions.mockReturnValue({ + "run-123": mockSession, + }); + mockTrpcAgent.reconnect.mutate.mockResolvedValue({ + sessionId: "run-123", + channel: "agent-event:run-123", + configOptions: [], + }); + mockTrpcAgent.onSessionEvent.subscribe.mockReturnValue({ + unsubscribe: vi.fn(), + }); + mockTrpcAgent.onPermissionRequest.subscribe.mockReturnValue({ + unsubscribe: vi.fn(), + }); + mockTrpcWorkspace.verify.query.mockResolvedValue({ exists: true }); + // Both the local cache and S3 reads come back empty (e.g. unreadable + // log after an agent crash) — the repaint must not blank the transcript. + mockTrpcLogs.readLocalLogs.query.mockResolvedValue(""); + mockTrpcLogs.fetchS3Logs.query.mockResolvedValue(""); + + await service.clearSessionError("task-123", "/repo"); + + const stored = mockSessionStoreSetters.setSession.mock.calls.at(-1)?.[0]; + expect(stored.events).toBe(previousEvents); + }); + it("creates fresh session when initialPrompt is set (prompt never delivered)", async () => { const service = getSessionService(); const mockSession = createMockSession({