diff --git a/packages/core/src/sessions/sessionEvents.ts b/packages/core/src/sessions/sessionEvents.ts index 7370397428..f9e916b19f 100644 --- a/packages/core/src/sessions/sessionEvents.ts +++ b/packages/core/src/sessions/sessionEvents.ts @@ -358,7 +358,11 @@ export function normalizePromptToBlocks( ); } -export { isFatalSessionError, isRateLimitError } from "@posthog/shared"; +export { + isFatalSessionError, + isRateLimitError, + isTransportError, +} from "@posthog/shared"; /** * Whether a list of events already contains a `session/prompt` request. diff --git a/packages/core/src/sessions/sessionService.ts b/packages/core/src/sessions/sessionService.ts index e7f18d46a2..0f60e1a74f 100644 --- a/packages/core/src/sessions/sessionService.ts +++ b/packages/core/src/sessions/sessionService.ts @@ -22,6 +22,7 @@ import { isJsonRpcRequest, isJsonRpcResponse, isRateLimitError, + isTransportError, mergeConfigOptions, type OptimisticItem, type PermissionRequest, @@ -89,7 +90,7 @@ const LOCAL_SESSION_RECONNECT_BACKOFF = { const LOCAL_SESSION_RECOVERY_MESSAGE = "Lost connection to the agent. Reconnecting…"; const LOCAL_SESSION_RECOVERY_FAILED_MESSAGE = - "Connecting to to the agent has been lost. Retry, or start a new session."; + "Connection to the agent has been lost. Retry, or start a new session."; const GITHUB_AUTHORIZATION_REQUIRED_CODE = "github_authorization_required"; const AUTO_RETRY_MAX_ATTEMPTS = 2; const AUTO_RETRY_DELAY_MS = 10_000; @@ -152,6 +153,9 @@ export interface SessionTrpc { onSessionEvent: TrpcSubscription; onPermissionRequest: TrpcSubscription; onSessionIdleKilled: TrpcSubscription; + /** Optional: only hosts with a power manager detect turns stalled by + * system sleep. Core subscribes when the host exposes it. */ + onSessionStalled?: TrpcSubscription; }; workspace: { verify: TrpcQuery }; cloudTask: { @@ -559,6 +563,7 @@ export class SessionService { { startedAtTs: number; agentTextChunks: number; agentOutputEvents: number } >(); private idleKilledSubscription: { unsubscribe: () => void } | null = null; + private stalledTurnSubscription: { unsubscribe: () => void } | null = null; /** * Cached preview-config-options responses keyed by `${apiHost}::${adapter}`. * Shared across cloud sessions so switching model/adapter reuses the list. @@ -612,6 +617,38 @@ export class SessionService { }, }, ); + this.stalledTurnSubscription = + d.trpc.agent.onSessionStalled?.subscribe(undefined, { + onData: (event: { taskRunId: string; taskId: string }) => { + this.handleStalledTurn(event.taskId, event.taskRunId); + }, + onError: (err: unknown) => { + d.log.debug("Stalled-turn subscription error", { error: err }); + }, + }) ?? null; + } + + /** + * The main process detected a turn that produced no agent traffic after a + * system resume: the agent is wedged on a connection that died during + * sleep. Reconnect in place — the resumed session keeps its history. + */ + private handleStalledTurn(taskId: string, taskRunId: string): void { + const session = this.d.store.getSessionByTaskId(taskId); + if (!session || session.taskRunId !== taskRunId || session.isCloud) { + return; + } + this.d.log.warn("Recovering turn stalled by system sleep", { + taskId, + taskRunId, + }); + this.startAutoRecoverLocalSession( + taskId, + taskRunId, + session.taskTitle, + "Turn stalled after system sleep", + "The agent lost its connection while the computer was asleep. Please retry or start a new session.", + ); } /** @@ -976,6 +1013,16 @@ export class SessionService { }); if (result) { + if (sessionId && result.resumedExistingSession === false) { + this.d.log.warn( + "Reconnect could not restore prior agent context; continuing with a fresh session", + { taskId, taskRunId }, + ); + this.d.toast.info( + "Couldn't restore the agent's previous context, so it may not remember earlier turns. The transcript is unaffected.", + ); + } + // Cast and merge live configOptions with persisted values. // Fall back to persisted options if the agent doesn't return any // (e.g. after session compaction). @@ -1237,8 +1284,16 @@ export class SessionService { reason: string, fallbackMessage: string, ): void { - void this.tryAutoRecoverLocalSession(taskId, taskRunId, reason).then( - (recovered) => { + void this.tryAutoRecoverLocalSession(taskId, taskRunId, reason) + .catch((error) => { + this.d.log.error("Local session recovery threw", { + taskId, + taskRunId, + error, + }); + return false; + }) + .then((recovered) => { if (recovered) { return; } @@ -1257,8 +1312,7 @@ export class SessionService { "Connection lost", ); } - }, - ); + }); } private async createNewLocalSession( @@ -1686,6 +1740,8 @@ export class SessionService { this.cloudRunIdleTracker.clear(); this.idleKilledSubscription?.unsubscribe(); this.idleKilledSubscription = null; + this.stalledTurnSubscription?.unsubscribe(); + this.stalledTurnSubscription = null; } /** @@ -2436,8 +2492,11 @@ export class SessionService { return { stopReason: "rate_limited" }; } - if (isFatalSessionError(errorMessage, errorDetails)) { - this.d.log.error("Fatal prompt error, attempting recovery", { + if ( + isFatalSessionError(errorMessage, errorDetails) || + isTransportError(errorMessage, errorDetails) + ) { + this.d.log.error("Prompt error, attempting session recovery", { taskRunId: session.taskRunId, errorMessage, errorDetails, diff --git a/packages/core/src/sessions/sessionServiceRecovery.test.ts b/packages/core/src/sessions/sessionServiceRecovery.test.ts index cb48522111..89d5a84134 100644 --- a/packages/core/src/sessions/sessionServiceRecovery.test.ts +++ b/packages/core/src/sessions/sessionServiceRecovery.test.ts @@ -48,6 +48,9 @@ function createHarness({ spyConnect = true } = {}) { updateSession: vi.fn(), }; const log = { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }; + let emitStalledTurn: + | ((event: { taskRunId: string; taskId: string }) => void) + | undefined; const deps = { store, log, @@ -65,6 +68,17 @@ function createHarness({ spyConnect = true } = {}) { onSessionIdleKilled: { subscribe: () => ({ unsubscribe: vi.fn() }), }, + onSessionStalled: { + subscribe: ( + _input: unknown, + handlers: { + onData: (event: { taskRunId: string; taskId: string }) => void; + }, + ) => { + emitStalledTurn = handlers.onData; + return { unsubscribe: vi.fn() }; + }, + }, }, }, } as unknown as SessionServiceDeps; @@ -73,7 +87,16 @@ function createHarness({ spyConnect = true } = {}) { const connectToTask = spyConnect ? vi.spyOn(service, "connectToTask").mockResolvedValue(undefined) : undefined; - return { service, sessions, connectToTask, log }; + return { + service, + sessions, + connectToTask, + log, + emitStalledTurn: (event: { taskRunId: string; taskId: string }) => { + if (!emitStalledTurn) throw new Error("Stalled-turn handler not wired"); + emitStalledTurn(event); + }, + }; } function reconcile( @@ -188,3 +211,59 @@ describe("SessionService run-less local task recovery", () => { expect(connectToTask).toHaveBeenCalledWith({ task, repoPath: "/repo" }); }); }); + +describe("SessionService stalled-turn recovery", () => { + it("auto-recovers a local session when the main process reports a stalled turn", async () => { + const { service, sessions, emitStalledTurn } = createHarness(); + sessions["run-task-1"] = makeSession("task-1"); + ( + service as unknown as { localRepoPaths: Map } + ).localRepoPaths.set("task-1", "/repo"); + const reconnectInPlace = vi + .spyOn( + service as unknown as { + reconnectInPlace: (taskId: string, repoPath: string) => unknown; + }, + "reconnectInPlace", + ) + .mockResolvedValue(true); + + emitStalledTurn({ taskId: "task-1", taskRunId: "run-task-1" }); + + await vi.waitFor(() => + expect(reconnectInPlace).toHaveBeenCalledWith("task-1", "/repo"), + ); + }); + + it("ignores stalled-turn reports for cloud sessions", async () => { + const { service, sessions, emitStalledTurn } = createHarness(); + sessions["run-task-1"] = { ...makeSession("task-1"), isCloud: true }; + const reconnectInPlace = vi.spyOn( + service as unknown as { + reconnectInPlace: (taskId: string, repoPath: string) => unknown; + }, + "reconnectInPlace", + ); + + emitStalledTurn({ taskId: "task-1", taskRunId: "run-task-1" }); + await Promise.resolve(); + + expect(reconnectInPlace).not.toHaveBeenCalled(); + }); + + it("ignores stalled-turn reports for superseded runs", async () => { + const { service, sessions, emitStalledTurn } = createHarness(); + sessions["run-task-1"] = makeSession("task-1"); + const reconnectInPlace = vi.spyOn( + service as unknown as { + reconnectInPlace: (taskId: string, repoPath: string) => unknown; + }, + "reconnectInPlace", + ); + + emitStalledTurn({ taskId: "task-1", taskRunId: "run-old" }); + await Promise.resolve(); + + expect(reconnectInPlace).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/host-router/src/routers/agent.router.ts b/packages/host-router/src/routers/agent.router.ts index e62ea18cf9..ac2ddc3a60 100644 --- a/packages/host-router/src/routers/agent.router.ts +++ b/packages/host-router/src/routers/agent.router.ts @@ -200,6 +200,16 @@ export const agentRouter = router({ } }), + onSessionStalled: publicProcedure.subscription(async function* (opts) { + const service = opts.ctx.container.get(AGENT_SERVICE); + for await (const event of service.toIterable( + AgentServiceEvent.SessionStalled, + { signal: opts.signal }, + )) { + yield event; + } + }), + onAgentFileActivity: publicProcedure.subscription(async function* (opts) { const service = opts.ctx.container.get(AGENT_SERVICE); for await (const event of service.toIterable( diff --git a/packages/shared/src/errors.test.ts b/packages/shared/src/errors.test.ts index 067923c6be..562cb23d38 100644 --- a/packages/shared/src/errors.test.ts +++ b/packages/shared/src/errors.test.ts @@ -5,6 +5,7 @@ import { isFatalSessionError, isNotAuthenticatedError, isRateLimitError, + isTransportError, NotAuthenticatedError, serializeError, } from "./errors"; @@ -105,6 +106,38 @@ describe("isFatalSessionError", () => { }); }); +describe("isTransportError", () => { + it.each([ + "read ECONNRESET", + "connect ECONNREFUSED 127.0.0.1:443", + "write EPIPE", + "request timed out: ETIMEDOUT", + "getaddrinfo ENOTFOUND api.anthropic.com", + "socket hang up", + "TypeError: fetch failed", + "Connection reset by peer", + "stream closed unexpectedly", + "UND_ERR_SOCKET: other side closed", + ])("treats %j as a transport error", (message) => { + expect(isTransportError(message)).toBe(true); + }); + + it("matches in details when the message is generic", () => { + expect(isTransportError("Unknown error", "read ECONNRESET")).toBe(true); + }); + + it("does not treat a rate-limit error as a transport error", () => { + expect(isTransportError("network error", "rate limit exceeded")).toBe( + false, + ); + }); + + it("returns false for non-transport errors", () => { + expect(isTransportError("invalid prompt")).toBe(false); + expect(isTransportError("")).toBe(false); + }); +}); + describe("serializeError", () => { it("captures name, message and code from an Error", () => { const err = Object.assign(new TypeError("boom"), { code: "ERR_X" }); diff --git a/packages/shared/src/errors.ts b/packages/shared/src/errors.ts index c8f4d27118..9ca4fb4372 100644 --- a/packages/shared/src/errors.ts +++ b/packages/shared/src/errors.ts @@ -82,6 +82,29 @@ const FATAL_SESSION_ERROR_PATTERNS = [ "session not found", ] as const; +// Network/socket failures cut a turn short without killing the agent process — +// typical after system sleep resets the agent's upstream connection. They are +// recoverable by reconnecting the session, unlike FATAL_SESSION_ERROR_PATTERNS +// which indicate the agent itself is gone. +const TRANSPORT_ERROR_PATTERNS = [ + "econnreset", + "econnrefused", + "epipe", + "etimedout", + "enetdown", + "enetunreach", + "ehostunreach", + "enotfound", + "socket hang up", + "network error", + "fetch failed", + "connection reset", + "connection closed", + "connection error", + "stream closed", + "other side closed", +] as const; + function includesAny( value: string | undefined, patterns: readonly string[], @@ -111,3 +134,14 @@ export function isFatalSessionError( includesAny(errorDetails, FATAL_SESSION_ERROR_PATTERNS) ); } + +export function isTransportError( + errorMessage: string, + errorDetails?: string, +): boolean { + if (isRateLimitError(errorMessage, errorDetails)) return false; + return ( + includesAny(errorMessage, TRANSPORT_ERROR_PATTERNS) || + includesAny(errorDetails, TRANSPORT_ERROR_PATTERNS) + ); +} diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index 9c71d7df07..7d821ee3ea 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -75,6 +75,7 @@ export { isFatalSessionError, isNotAuthenticatedError, isRateLimitError, + isTransportError, NotAuthenticatedError, type SerializedError, serializeError, diff --git a/packages/workspace-server/src/services/agent/agent.test.ts b/packages/workspace-server/src/services/agent/agent.test.ts index adeffc2afb..8a1a930e70 100644 --- a/packages/workspace-server/src/services/agent/agent.test.ts +++ b/packages/workspace-server/src/services/agent/agent.test.ts @@ -520,6 +520,97 @@ describe("AgentService", () => { expect.anything(), ); }); + + describe("stalled turn detection after resume", () => { + const STALLED_TURN_GRACE_MS = ( + AgentService as unknown as { STALLED_TURN_GRACE_MS: number } + ).STALLED_TURN_GRACE_MS; + + function getResumeHandler() { + return ( + deps.powerManager.onResume.mock.calls[0] as unknown as [() => void] + )[0]; + } + + function getSession(taskRunId: string) { + return ( + service as unknown as { + sessions: Map; + } + ).sessions.get(taskRunId); + } + + it("reports a pending turn with no agent traffic since resume as stalled", () => { + injectSession(service, "run-1", { promptPending: true }); + service.recordActivity("run-1"); + + vi.advanceTimersByTime(10 * 60 * 1000); + getResumeHandler()(); + vi.advanceTimersByTime(STALLED_TURN_GRACE_MS); + + expect(service.emit).toHaveBeenCalledWith( + "session-stalled", + expect.objectContaining({ + taskRunId: "run-1", + taskId: "task-for-run-1", + }), + ); + }); + + it("does not report a turn that produced agent traffic after resume", () => { + injectSession(service, "run-1", { promptPending: true }); + service.recordActivity("run-1"); + + vi.advanceTimersByTime(10 * 60 * 1000); + getResumeHandler()(); + + vi.advanceTimersByTime(10 * 1000); + const session = getSession("run-1"); + if (!session) throw new Error("Expected session to exist"); + session.lastActivityAt = Date.now(); + + vi.advanceTimersByTime(STALLED_TURN_GRACE_MS); + + expect(service.emit).not.toHaveBeenCalledWith( + "session-stalled", + expect.anything(), + ); + }); + + it("does not report a turn that settled before the grace period ends", () => { + injectSession(service, "run-1", { promptPending: true }); + service.recordActivity("run-1"); + + vi.advanceTimersByTime(10 * 60 * 1000); + getResumeHandler()(); + + const session = getSession("run-1") as unknown as { + promptPending: boolean; + }; + session.promptPending = false; + + vi.advanceTimersByTime(STALLED_TURN_GRACE_MS); + + expect(service.emit).not.toHaveBeenCalledWith( + "session-stalled", + expect.anything(), + ); + }); + + it("does not report sessions that had no pending turn at resume", () => { + injectSession(service, "run-1", { promptPending: false }); + service.recordActivity("run-1"); + + vi.advanceTimersByTime(10 * 60 * 1000); + getResumeHandler()(); + vi.advanceTimersByTime(STALLED_TURN_GRACE_MS); + + expect(service.emit).not.toHaveBeenCalledWith( + "session-stalled", + expect.anything(), + ); + }); + }); }); describe("channel system prompt local folders", () => { diff --git a/packages/workspace-server/src/services/agent/agent.ts b/packages/workspace-server/src/services/agent/agent.ts index 4ee49e1fad..308bfd7b90 100644 --- a/packages/workspace-server/src/services/agent/agent.ts +++ b/packages/workspace-server/src/services/agent/agent.ts @@ -307,6 +307,8 @@ interface ManagedSession { // Reset per session. `evaluatedPrUrls` dedupes the GitHub lookup per URL. prAttributed: boolean; evaluatedPrUrls: Set; + /** Whether this session restored a prior conversation (vs starting fresh). */ + resumedExistingSession: boolean; } /** Get the agent session ID from a managed session, throwing if not set. */ @@ -341,6 +343,7 @@ interface PendingPermission { @injectable() export class AgentService extends TypedEventEmitter { private static readonly IDLE_TIMEOUT_MS = 15 * 60 * 1000; + private static readonly STALLED_TURN_GRACE_MS = 45 * 1000; private sessions = new Map(); private pendingPermissions = new Map(); @@ -349,6 +352,7 @@ export class AgentService extends TypedEventEmitter { string, { handle: ReturnType; deadline: number } >(); + private stalledTurnCheckHandle: ReturnType | null = null; private processTracking: ProcessTrackingService; private sleepService: AgentSleepCoordinator; private fsService: AgentRepoFiles; @@ -398,7 +402,42 @@ export class AgentService extends TypedEventEmitter { this.log = loggerFactory.scope("agent-service"); this.onAgentLog = makeOnAgentLog(loggerFactory); - powerManager.onResume(() => this.checkIdleDeadlines()); + powerManager.onResume(() => { + this.checkIdleDeadlines(); + this.scheduleStalledTurnCheck(); + }); + } + + /** + * A system sleep mid-turn can leave the agent subprocess waiting forever on + * an upstream connection that died, without the turn ever erroring. After + * resume, give in-flight turns a grace period; any that produce no agent + * traffic in that window are reported stalled so the client reconnects them. + */ + private scheduleStalledTurnCheck(): void { + const resumedAt = Date.now(); + const pendingAtResume = [...this.sessions.values()] + .filter((session) => session.promptPending) + .map((session) => session.taskRunId); + if (pendingAtResume.length === 0) return; + + if (this.stalledTurnCheckHandle) clearTimeout(this.stalledTurnCheckHandle); + this.stalledTurnCheckHandle = setTimeout(() => { + this.stalledTurnCheckHandle = null; + for (const taskRunId of pendingAtResume) { + const session = this.sessions.get(taskRunId); + if (!session?.promptPending) continue; + if (session.lastActivityAt >= resumedAt) continue; + this.log.warn("Turn stalled after system resume", { + taskRunId, + taskId: session.taskId, + }); + this.emit(AgentServiceEvent.SessionStalled, { + taskRunId, + taskId: session.taskId, + }); + } + }, AgentService.STALLED_TURN_GRACE_MS); } private getClaudeCliPath(): string { @@ -908,6 +947,7 @@ If a repository IS genuinely required, attach one in this priority order: let configOptions: SessionConfigOption[] | undefined; let agentSessionId: string | undefined; + let resumedExistingSession = false; // Imported Claude Code CLI session: the transcript JSONL was copied // into CLAUDE_CONFIG_DIR at import time, so load it directly and let @@ -940,6 +980,7 @@ If a repository IS genuinely required, attach one in this priority order: }); configOptions = loadResponse?.configOptions ?? undefined; agentSessionId = importedSessionId; + resumedExistingSession = true; } catch (err) { this.log.warn( "Failed to load imported session, creating new session instead", @@ -1012,6 +1053,7 @@ If a repository IS genuinely required, attach one in this priority order: }); configOptions = resumeResponse?.configOptions ?? undefined; agentSessionId = existingSessionId; + resumedExistingSession = true; } else if (agentSessionId === undefined) { if (isReconnect) { this.log.info("No sessionId for reconnect, creating new session", { @@ -1059,6 +1101,7 @@ If a repository IS genuinely required, attach one in this priority order: toolInstallations, prAttributed: false, evaluatedPrUrls: new Set(), + resumedExistingSession, }; this.sessions.set(taskRunId, session); @@ -1123,7 +1166,8 @@ If a repository IS genuinely required, attach one in this priority order: config.sessionId = undefined; return this.getOrCreateSession(config, false, false); } - if (isReconnect) return null; + // Rethrow (rather than returning null) so the client can show the real + // failure instead of a generic "could not be resumed" message. throw err; } } @@ -1536,6 +1580,10 @@ For git operations while detached: async cleanupAll(): Promise { for (const { handle } of this.idleTimeouts.values()) clearTimeout(handle); this.idleTimeouts.clear(); + if (this.stalledTurnCheckHandle) { + clearTimeout(this.stalledTurnCheckHandle); + this.stalledTurnCheckHandle = null; + } const sessionIds = Array.from(this.sessions.keys()); this.log.info("Cleaning up all agent sessions", { sessionCount: sessionIds.length, @@ -1646,6 +1694,11 @@ For git operations while detached: }; const onAcpMessage = (message: unknown) => { + const session = this.sessions.get(taskRunId); + if (session) { + session.lastActivityAt = Date.now(); + } + const acpMessage: AcpMessage = { type: "acp_message", ts: Date.now(), @@ -1967,6 +2020,7 @@ For git operations while detached: sessionId: session.taskRunId, channel: session.channel, configOptions: session.configOptions, + resumedExistingSession: session.resumedExistingSession, }; } diff --git a/packages/workspace-server/src/services/agent/schemas.ts b/packages/workspace-server/src/services/agent/schemas.ts index 493e79943e..1f77216fdf 100644 --- a/packages/workspace-server/src/services/agent/schemas.ts +++ b/packages/workspace-server/src/services/agent/schemas.ts @@ -136,6 +136,12 @@ export const sessionResponseSchema = z.object({ sessionId: z.string(), channel: z.string(), configOptions: z.array(sessionConfigOptionSchema).optional(), + /** + * Whether the agent restored the prior conversation. False when a reconnect + * fell back to a fresh session (no stored history, or the resume failed), so + * the client can tell the user the agent lost its context. + */ + resumedExistingSession: z.boolean().optional(), }); export type SessionResponse = z.infer; @@ -228,6 +234,7 @@ export const AgentServiceEvent = { PermissionRequest: "permission-request", SessionsIdle: "sessions-idle", SessionIdleKilled: "session-idle-killed", + SessionStalled: "session-stalled", AgentFileActivity: "agent-file-activity", LlmActivity: "llm-activity", } as const; @@ -250,6 +257,11 @@ export interface SessionIdleKilledPayload { taskId: string; } +export interface SessionStalledPayload { + taskRunId: string; + taskId: string; +} + export interface AgentFileActivityPayload { taskId: string; branchName: string | null; @@ -260,6 +272,7 @@ export interface AgentServiceEvents { [AgentServiceEvent.PermissionRequest]: PermissionRequestPayload; [AgentServiceEvent.SessionsIdle]: undefined; [AgentServiceEvent.SessionIdleKilled]: SessionIdleKilledPayload; + [AgentServiceEvent.SessionStalled]: SessionStalledPayload; [AgentServiceEvent.AgentFileActivity]: AgentFileActivityPayload; [AgentServiceEvent.LlmActivity]: undefined; }