From 1a3e6ac6dd6e04402a834d5fd6f8a745911095e4 Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Mon, 14 Sep 2026 10:39:06 -0600 Subject: [PATCH] fix(runtime): adopt kernel, idle wait and Codex recovery fixes Fixes #69. Selectively adopts Prime upstream #2213 (90ca4a457e0fef0bd925215d4c973abd6264f067), #2151 (8eaa00c37385c6b25fed9d46584b50ffc2e87f9c), and #2314 (7d1913969d33fa5a31366321e3f5db0070dcf2bf). --- .pylon/features.yaml | 17 + .pylon/upstream-review.md | 9 + packages/ai/.changes/fix-codex-chain-reset.md | 1 + .../src/providers/openai-codex-responses.ts | 120 +++--- packages/ai/test/openai-codex-stream.test.ts | 363 ++++++++++++++++++ .../.changes/pylon-runtime-reliability.md | 2 + .../coding-agent/src/core/agent-session.ts | 11 +- .../src/core/kernel/repl-manager.ts | 13 + .../test/repl-kernel-pipe-errors.test.ts | 62 +++ .../test/repl-kernel-shutdown.test.ts | 8 +- .../test/suite/agent-session-queue.test.ts | 47 +++ 11 files changed, 607 insertions(+), 46 deletions(-) create mode 100644 packages/ai/.changes/fix-codex-chain-reset.md create mode 100644 packages/coding-agent/.changes/pylon-runtime-reliability.md create mode 100644 packages/coding-agent/test/repl-kernel-pipe-errors.test.ts diff --git a/.pylon/features.yaml b/.pylon/features.yaml index 49722c37768..78f47f6ace8 100644 --- a/.pylon/features.yaml +++ b/.pylon/features.yaml @@ -417,3 +417,20 @@ decisions: revisit_when: - Prime ships a repository-neutral immutable release and promotion primitive that preserves Pylon's exact source, workflow, attestation, stable-history, and withdrawal policy without upstream credentials or channel names. - Pylon adopts another verifiable distribution transport that fully supersedes GitHub immutable Releases. + + runtime-reliability-september-followup: + area: runtime-reliability + state: candidate + owner: upstream + decision: adopt + pylon_refs: + - https://github.com/pylon-code/prime-agent/issues/69 + - https://github.com/pylon-code/pylon/issues/114 + upstream_refs: + - https://github.com/PrimeIntellect-ai/prime-agent/commit/90ca4a457e0fef0bd925215d4c973abd6264f067 + - https://github.com/PrimeIntellect-ai/prime-agent/commit/8eaa00c37385c6b25fed9d46584b50ffc2e87f9c + - https://github.com/PrimeIntellect-ai/prime-agent/commit/7d1913969d33fa5a31366321e3f5db0070dcf2bf + fork_change: selective-kernel-idle-codex-reliability + upstream_support: Absorb kernel pipe errors, park blocked idle waiters, and retry stale Codex continuation IDs once before output; retain Pylon lifecycle, cancellation, scoped identity, recovery cursors and ownership without wire changes. + revisit_when: + - The next complete upstream integration includes these commits; reconcile the selective patches without duplicating behavior. diff --git a/.pylon/upstream-review.md b/.pylon/upstream-review.md index 0f02764187b..a50295bcc05 100644 --- a/.pylon/upstream-review.md +++ b/.pylon/upstream-review.md @@ -325,3 +325,12 @@ Follow-up: Task10 builds/packs the exact merged tree into a private prefix and r - This is an additive public SDK metadata correction within the existing recovery contract. No daemon wire, schema, negotiated capability, SDK feature token, artifact recipe or publication policy changes. Legacy cursorless events remain cursorless. Changed SDK bytes require a new immutable managed build. - Validation: the new faux-provider SDK regression fails without forwarding and passes with it; accepted events retain their exact cursors, duplicate/foreign/retired-generation/invalid-attribution frames stay suppressed, duplicate lifecycles stay suppressed, and sequence-only events gain no fabricated cursor. All 226 focused SDK, adoption/export and regression tests pass. `npm run check` passes formatting/lint, TypeScript, installer rendering and browser bundle smoke. - Revisit when upstream provides equivalent per-event SDK progress and recovery guarantees. The Pylon idle-restart regression and real web/mobile acceptance remain consumer gates. + +## 2026-09-14 — selective runtime reliability fixes + +- Product base: `65d9c0274866f1bacb3959a16718718d102ee593`. Tracking: #69 under #1 and Pylon #114. The maintainer authorized implementation, review and merge after green checks in the Pylon task conversation. This selective batch does not advance the complete upstream checkpoint `1eee2938b4eeb7a4d72e17035adda669a89b63de` or adopt pending distribution/automation changes. +- Adopt upstream #2213 (`90ca4a457e0fef0bd925215d4c973abd6264f067`): observe stdin/stdout stream errors without terminating the worker; retain existing pending-write failure, generation checks and cleanup ownership. +- Adopt upstream #2151 (`8eaa00c37385c6b25fed9d46584b50ffc2e87f9c`): park `waitForIdle` on the same busy predicate used by the input pump. Preserve fork correlated settlement and post-compaction cancellation waiters. +- Adopt upstream #2314 (`7d1913969d33fa5a31366321e3f5db0070dcf2bf`): bind Codex continuation state to the producing connection and retry `previous_response_not_found` once with full context only before output. Preserve scoped provider hooks and avoid duplicate stream events or prompt replay. +- Compatibility: no daemon commands, events, response shapes, protocol/schema revisions or public capability tokens change. No credentials, managed runtime, live session or release is modified. Existing installations require a separately verified immutable managed build to receive these fixes. +- Validation: 125 focused coding-agent tests (including the real isolated kernel pipe regression) and 25 Codex stream tests pass. `npm run check` passes Biome, TypeScript, installer and browser bundle checks. Required hosted validation and final review are recorded in the owning PR. diff --git a/packages/ai/.changes/fix-codex-chain-reset.md b/packages/ai/.changes/fix-codex-chain-reset.md new file mode 100644 index 00000000000..07315980d06 --- /dev/null +++ b/packages/ai/.changes/fix-codex-chain-reset.md @@ -0,0 +1 @@ +- Fixed codex websocket sessions repeatedly failing after a stale cached previous_response_id: the continuation chain is now connection-scoped (a reconnected connection always resends the full context), and a previous_response_not_found failure drops the cached chain and retries once with the full request body, re-anchoring on the successful response. diff --git a/packages/ai/src/providers/openai-codex-responses.ts b/packages/ai/src/providers/openai-codex-responses.ts index b1eca527d12..a3af0629156 100644 --- a/packages/ai/src/providers/openai-codex-responses.ts +++ b/packages/ai/src/providers/openai-codex-responses.ts @@ -142,50 +142,64 @@ export const streamOpenAICodexResponses: StreamFunction<"openai-codex-responses" if (transport !== "sse" && !websocketDisabledForSession) { let websocketStarted = false; - try { - await processWebSocketStream( - resolveCodexWebSocketUrl(model.baseUrl), - body, - websocketHeaders, - output, - stream, - model, - () => { - websocketStarted = true; - }, - options, - ); - - if (options?.signal?.aborted) { - throw new Error("Request was aborted"); - } - stream.push({ - type: "done", - reason: output.stopReason as "stop" | "length" | "toolUse", - message: output, - }); - stream.end(); - return; - } catch (error) { - const aborted = options?.signal?.aborted; - if (aborted || isCodexNonTransportError(error)) { - throw error; - } - appendAssistantMessageDiagnostic( - output, - createAssistantMessageDiagnostic("provider_transport_failure", error, { - configuredTransport: transport, - fallbackTransport: websocketStarted ? undefined : "sse", - eventsEmitted: websocketStarted, - phase: websocketStarted ? "after_message_stream_start" : "before_message_stream_start", - requestBytes: new TextEncoder().encode(bodyJson).byteLength, - }), - ); - recordWebSocketFailure(options?.sessionId, error); - if (websocketStarted) { - throw error; + // Retry a stale previous_response_id once on a fresh connection: the + // failed attempt's error cleanup already dropped the cached connection + // entry, so the retry resends the full request body. Any further + // failure takes the shared error handling below. + let chainResetRetried = false; + for (;;) { + try { + await processWebSocketStream( + resolveCodexWebSocketUrl(model.baseUrl), + body, + websocketHeaders, + output, + stream, + model, + () => { + websocketStarted = true; + }, + options, + ); + + if (options?.signal?.aborted) { + throw new Error("Request was aborted"); + } + stream.push({ + type: "done", + reason: output.stopReason as "stop" | "length" | "toolUse", + message: output, + }); + stream.end(); + return; + } catch (error) { + const aborted = options?.signal?.aborted; + // Only reset the chain while nothing was streamed yet: after the + // first event the retry would duplicate "start"/content events. + if (!aborted && !websocketStarted && !chainResetRetried && isStaleCodexContinuationError(error)) { + chainResetRetried = true; + continue; + } + if (aborted || isCodexNonTransportError(error)) { + throw error; + } + appendAssistantMessageDiagnostic( + output, + createAssistantMessageDiagnostic("provider_transport_failure", error, { + configuredTransport: transport, + fallbackTransport: websocketStarted ? undefined : "sse", + eventsEmitted: websocketStarted, + phase: websocketStarted ? "after_message_stream_start" : "before_message_stream_start", + requestBytes: new TextEncoder().encode(bodyJson).byteLength, + }), + ); + recordWebSocketFailure(options?.sessionId, error); + if (websocketStarted) { + throw error; + } + recordWebSocketSseFallback(options?.sessionId); + break; } - recordWebSocketSseFallback(options?.sessionId); } } @@ -425,6 +439,16 @@ function isCodexNonTransportError(error: unknown): boolean { return error instanceof CodexApiError || error instanceof CodexProtocolError; } +const STALE_CONTINUATION_ERROR_CODE = "previous_response_not_found"; + +/** + * Codex API error for a previous_response_id the server does not recognize; + * the request must be resent in full instead of chained. + */ +function isStaleCodexContinuationError(error: unknown): boolean { + return error instanceof CodexApiError && error.code?.toLowerCase() === STALE_CONTINUATION_ERROR_CODE; +} + async function* mapCodexEvents(events: AsyncIterable>): AsyncGenerator { for await (const event of events) { const type = typeof event.type === "string" ? event.type : undefined; @@ -545,6 +569,8 @@ interface CachedWebSocketContinuationState { lastRequestBody: RequestBody; lastResponseId: string; lastResponseItems: ResponseInput; + /** Connection whose server-side response state this chain is anchored to. */ + connection: WebSocketLike; } interface CachedWebSocketConnection { @@ -1051,6 +1077,13 @@ function buildCachedWebSocketRequestBody(entry: CachedWebSocketConnection, body: return body; } + // Continuations are anchored to the connection that produced the response; + // a different socket cannot resolve their previous_response_id. + if (continuation.connection !== entry.socket) { + entry.continuation = undefined; + return body; + } + const delta = getCachedWebSocketInputDelta(body, continuation); if (!delta || !continuation.lastResponseId) { entry.continuation = undefined; @@ -1144,6 +1177,7 @@ async function processWebSocketStream( lastRequestBody: fullBody, lastResponseId: output.responseId, lastResponseItems: responseItems, + connection: entry.socket, }; } } catch (error) { diff --git a/packages/ai/test/openai-codex-stream.test.ts b/packages/ai/test/openai-codex-stream.test.ts index cd7f6ab979a..2f1645357a0 100644 --- a/packages/ai/test/openai-codex-stream.test.ts +++ b/packages/ai/test/openai-codex-stream.test.ts @@ -1008,6 +1008,247 @@ describe("openai-codex streaming", () => { }); }); + it("sends full context without a stale previous_response_id after a reconnect and re-anchors", async () => { + const token = mockToken(); + let firstSocket: ScriptedWebSocketHandle | undefined; + const sentBodies = installScriptedCodexWebSocket([ + (socket) => { + firstSocket = socket; + socket.emit(codexResponseEvents({ responseId: "resp_1", messageId: "msg_1", text: "Hello" })); + }, + (socket) => socket.emit(codexResponseEvents({ responseId: "resp_2", messageId: "msg_2", text: "Done" })), + (socket) => socket.emit(codexResponseEvents({ responseId: "resp_3", messageId: "msg_3", text: "Again" })), + ]); + + const model = codexTestModel(); + const firstContext: Context = { + systemPrompt: "You are a helpful assistant.", + messages: [{ role: "user", content: "Say hello", timestamp: 1 }], + }; + const first = await streamOpenAICodexResponses(model, firstContext, { + apiKey: token, + sessionId: "session-reconnect", + transport: "websocket-cached", + }).result(); + + // The server dropped the connection during the idle gap between turns. + firstSocket?.drop(); + + const secondContext: Context = { + systemPrompt: "You are a helpful assistant.", + messages: [...firstContext.messages, first, { role: "user", content: "Now finish", timestamp: 2 }], + }; + const second = await streamOpenAICodexResponses(model, secondContext, { + apiKey: token, + sessionId: "session-reconnect", + transport: "websocket-cached", + }).result(); + + expect(second.stopReason).toBe("stop"); + + const thirdContext: Context = { + systemPrompt: "You are a helpful assistant.", + messages: [...secondContext.messages, second, { role: "user", content: "And again", timestamp: 3 }], + }; + await streamOpenAICodexResponses(model, thirdContext, { + apiKey: token, + sessionId: "session-reconnect", + transport: "websocket-cached", + }).result(); + + expect(sentBodies).toHaveLength(3); + const reconnectedBody = sentBodies[1] as { previous_response_id?: string; input: unknown[] }; + const resumedBody = sentBodies[2] as { previous_response_id?: string; input: unknown[] }; + // The reconnected connection starts from the full context, never the dead + // connection's previous_response_id. + expect(reconnectedBody.previous_response_id).toBeUndefined(); + expect(reconnectedBody.input).toEqual([ + { role: "user", content: [{ type: "input_text", text: "Say hello" }] }, + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "Hello", annotations: [] }], + status: "completed", + id: "msg_1", + }, + { role: "user", content: [{ type: "input_text", text: "Now finish" }] }, + ]); + // The chain re-anchors on the reconnected connection's response. + expect(resumedBody.previous_response_id).toBe("resp_2"); + expect(resumedBody.input).toEqual([{ role: "user", content: [{ type: "input_text", text: "And again" }] }]); + expect(global.fetch).not.toHaveBeenCalled(); + expect(getOpenAICodexWebSocketDebugStats("session-reconnect")).toMatchObject({ + requests: 3, + connectionsCreated: 2, + connectionsReused: 1, + fullContextRequests: 2, + deltaRequests: 1, + lastPreviousResponseId: "resp_2", + }); + }); + + it("recovers a stale previous_response_id with one full-context chain-reset retry", async () => { + const token = mockToken(); + const sentBodies = installScriptedCodexWebSocket([ + (socket) => socket.emit(codexResponseEvents({ responseId: "resp_1", messageId: "msg_1", text: "Hello" })), + (socket) => + socket.emit( + codexErrorEvents("previous_response_not_found", "Previous response with id 'resp_1' not found"), + ), + (socket) => socket.emit(codexResponseEvents({ responseId: "resp_2", messageId: "msg_2", text: "Done" })), + (socket) => socket.emit(codexResponseEvents({ responseId: "resp_3", messageId: "msg_3", text: "Again" })), + ]); + + const model = codexTestModel(); + const firstContext: Context = { + systemPrompt: "You are a helpful assistant.", + messages: [{ role: "user", content: "Say hello", timestamp: 1 }], + }; + + const first = await streamOpenAICodexResponses(model, firstContext, { + apiKey: token, + sessionId: "session-chain-reset", + transport: "websocket-cached", + }).result(); + + const secondContext: Context = { + systemPrompt: "You are a helpful assistant.", + messages: [...firstContext.messages, first, { role: "user", content: "Now finish", timestamp: 2 }], + }; + const second = await streamOpenAICodexResponses(model, secondContext, { + apiKey: token, + sessionId: "session-chain-reset", + transport: "websocket-cached", + }).result(); + + // The stale continuation was recovered within the same turn. + expect(second.stopReason).toBe("stop"); + expect(second.content[0]).toMatchObject({ type: "text", text: "Done" }); + + // A later turn proves the chain resumed from the retried response. + const thirdContext: Context = { + systemPrompt: "You are a helpful assistant.", + messages: [...secondContext.messages, second, { role: "user", content: "And again", timestamp: 3 }], + }; + await streamOpenAICodexResponses(model, thirdContext, { + apiKey: token, + sessionId: "session-chain-reset", + transport: "websocket-cached", + }).result(); + + expect(sentBodies).toHaveLength(4); + const deltaBody = sentBodies[1] as { previous_response_id?: string; input: unknown[] }; + const retryBody = sentBodies[2] as { previous_response_id?: string; input: unknown[] }; + const resumedBody = sentBodies[3] as { previous_response_id?: string; input: unknown[] }; + expect(deltaBody.previous_response_id).toBe("resp_1"); + expect(deltaBody.input).toEqual([{ role: "user", content: [{ type: "input_text", text: "Now finish" }] }]); + // The chain reset resends the full request body without previous_response_id. + expect(retryBody.previous_response_id).toBeUndefined(); + expect(retryBody.input).toEqual([ + { role: "user", content: [{ type: "input_text", text: "Say hello" }] }, + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "Hello", annotations: [] }], + status: "completed", + id: "msg_1", + }, + { role: "user", content: [{ type: "input_text", text: "Now finish" }] }, + ]); + expect(resumedBody.previous_response_id).toBe("resp_2"); + expect(resumedBody.input).toEqual([{ role: "user", content: [{ type: "input_text", text: "And again" }] }]); + expect(global.fetch).not.toHaveBeenCalled(); + expect(getOpenAICodexWebSocketDebugStats("session-chain-reset")).toMatchObject({ + requests: 4, + connectionsCreated: 2, + connectionsReused: 2, + cachedContextRequests: 4, + fullContextRequests: 2, + deltaRequests: 2, + websocketFailures: 0, + sseFallbacks: 0, + }); + }); + + it("surfaces the error without further retries when the chain-reset retry fails again", async () => { + const token = mockToken(); + const sentBodies = installScriptedCodexWebSocket([ + (socket) => socket.emit(codexResponseEvents({ responseId: "resp_1", messageId: "msg_1", text: "Hello" })), + (socket) => + socket.emit( + codexErrorEvents("previous_response_not_found", "Previous response with id 'resp_1' not found"), + ), + (socket) => + socket.emit( + codexErrorEvents("previous_response_not_found", "Previous response with id 'resp_9' not found"), + ), + ]); + + const model = codexTestModel(); + const firstContext: Context = { + systemPrompt: "You are a helpful assistant.", + messages: [{ role: "user", content: "Say hello", timestamp: 1 }], + }; + const first = await streamOpenAICodexResponses(model, firstContext, { + apiKey: token, + sessionId: "session-chain-reset-fail", + transport: "websocket-cached", + }).result(); + + const secondContext: Context = { + systemPrompt: "You are a helpful assistant.", + messages: [...firstContext.messages, first, { role: "user", content: "Now finish", timestamp: 2 }], + }; + const second = await streamOpenAICodexResponses(model, secondContext, { + apiKey: token, + sessionId: "session-chain-reset-fail", + transport: "websocket-cached", + }).result(); + + expect(second.stopReason).toBe("error"); + expect(second.errorMessage).toBe("Codex error: Previous response with id 'resp_9' not found"); + // Exactly one chain-reset retry, then the error surfaces unchanged. + expect(sentBodies).toHaveLength(3); + const retryBody = sentBodies[2] as { previous_response_id?: string }; + expect(retryBody.previous_response_id).toBeUndefined(); + expect(global.fetch).not.toHaveBeenCalled(); + expect(failureDetails(second)).toMatchObject({ providerErrorType: "previous_response_not_found" }); + }); + + it("does not chain-reset retry other codex api errors", async () => { + const token = mockToken(); + const sentBodies = installScriptedCodexWebSocket([ + (socket) => socket.emit(codexResponseEvents({ responseId: "resp_1", messageId: "msg_1", text: "Hello" })), + (socket) => socket.emit(codexErrorEvents("invalid_request", "Invalid request")), + ]); + + const model = codexTestModel(); + const firstContext: Context = { + systemPrompt: "You are a helpful assistant.", + messages: [{ role: "user", content: "Say hello", timestamp: 1 }], + }; + const first = await streamOpenAICodexResponses(model, firstContext, { + apiKey: token, + sessionId: "session-chain-reset-other", + transport: "websocket-cached", + }).result(); + + const secondContext: Context = { + systemPrompt: "You are a helpful assistant.", + messages: [...firstContext.messages, first, { role: "user", content: "Now finish", timestamp: 2 }], + }; + const second = await streamOpenAICodexResponses(model, secondContext, { + apiKey: token, + sessionId: "session-chain-reset-other", + transport: "websocket-cached", + }).result(); + + expect(second.stopReason).toBe("error"); + expect(second.errorMessage).toBe("Codex error: Invalid request"); + expect(sentBodies).toHaveLength(2); + expect(global.fetch).not.toHaveBeenCalled(); + }); + function codexTestModel(): Model<"openai-codex-responses"> { return { id: "gpt-5.1-codex", @@ -1059,6 +1300,128 @@ describe("openai-codex streaming", () => { : undefined; } + /** Handle the send scripts use to drive a mocked websocket connection. */ + interface ScriptedWebSocketHandle { + emit: (events: unknown[]) => void; + /** Simulate the server dropping the connection between turns. */ + drop: () => void; + } + + /** + * Install a scripted WebSocket mock: every websocket send consumes one + * script entry (across socket instances, so reconnects stay scripted) and + * fetch is stubbed so an unexpected SSE fallback fails fast. + */ + function installScriptedCodexWebSocket( + scripts: Array<(socket: ScriptedWebSocketHandle) => void>, + ): Record[] { + const sentBodies: Record[] = []; + + class MockWebSocket { + static OPEN = 1; + readyState = MockWebSocket.OPEN; + private listeners = new Map void>>(); + + constructor(_url: string, _protocols?: string | string[] | { headers?: Record }) { + queueMicrotask(() => this.dispatch("open", {})); + } + + addEventListener(type: string, listener: (event: unknown) => void): void { + let listeners = this.listeners.get(type); + if (!listeners) { + listeners = new Set(); + this.listeners.set(type, listeners); + } + listeners.add(listener); + } + + removeEventListener(type: string, listener: (event: unknown) => void): void { + this.listeners.get(type)?.delete(listener); + } + + send(data: string): void { + sentBodies.push(JSON.parse(data) as Record); + const script = scripts.shift(); + if (!script) throw new Error("unexpected websocket request"); + queueMicrotask(() => + script({ + emit: (events: unknown[]) => this.emit(events), + drop: () => { + this.readyState = 3; + }, + }), + ); + } + + close(): void { + this.readyState = 3; + } + + emit(events: unknown[]): void { + for (const event of events) { + this.dispatch("message", { data: JSON.stringify(event) }); + } + } + + private dispatch(type: string, event: unknown): void { + for (const listener of this.listeners.get(type) ?? []) { + listener(event); + } + } + } + + global.fetch = vi.fn(async () => new Response("unexpected fetch", { status: 500 })) as typeof fetch; + globalThis.WebSocket = MockWebSocket as unknown as typeof WebSocket; + return sentBodies; + } + + function codexResponseEvents({ + responseId, + messageId, + text, + }: { + responseId: string; + messageId: string; + text: string; + }): unknown[] { + return [ + { type: "response.created", response: { id: responseId } }, + { + type: "response.output_item.added", + item: { type: "message", id: messageId, role: "assistant", status: "in_progress", content: [] }, + }, + { type: "response.content_part.added", part: { type: "output_text", text: "" } }, + { type: "response.output_text.delta", delta: text }, + { + type: "response.output_item.done", + item: { + type: "message", + id: messageId, + role: "assistant", + status: "completed", + content: [{ type: "output_text", text }], + }, + }, + { + type: "response.completed", + response: { + id: responseId, + status: "completed", + usage: { + input_tokens: 5, + output_tokens: 3, + total_tokens: 8, + input_tokens_details: { cached_tokens: 0 }, + }, + }, + }, + ]; + } + + function codexErrorEvents(code: string, message: string): unknown[] { + return [{ type: "error", code, message }]; + } + it("throws a structured failure after a single attempt on HTTP 500", async () => { const counter = stubCodexFetch( () => new Response(JSON.stringify({ error: { type: "server_error", message: "boom" } }), { status: 500 }), diff --git a/packages/coding-agent/.changes/pylon-runtime-reliability.md b/packages/coding-agent/.changes/pylon-runtime-reliability.md new file mode 100644 index 00000000000..5c81e56f589 --- /dev/null +++ b/packages/coding-agent/.changes/pylon-runtime-reliability.md @@ -0,0 +1,2 @@ +- Fixed a kernel pipe write error (write EPIPE) crashing the whole session worker: pipe errors are now recorded as kernel diagnostics while the pending write rejects cleanly. +- Fixed a session worker wedging at 100% CPU: waiting for a session to go idle while queued input was blocked by a running bash command, compaction, or retry spun in microtasks without ever yielding to IO, freezing every session in the worker and starving daemon IPC. diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 5509dd4bf02..8d67ea6d033 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -6759,6 +6759,12 @@ export class AgentSession { } } + /** + * Busy inputs: compaction, retry, bash, plus (for "pump") disposal, + * suspension, queued-work pauses, and branch-summary mutation. Waiters + * parked on this predicate rely on every clear site notifying the + * session-input checkpoint waiters. + */ private _isBusyForSessionInput(point: "preflight" | "pump"): boolean { const externalBusy = this.isCompacting || this.isRetrying || this.isBashRunning; if (point === "pump") { @@ -7777,7 +7783,10 @@ export class AgentSession { private async _waitForIdleOrSettlement(settlement?: PostCompactionContinuationSettlement): Promise { while (settlement === undefined || this._postCompactionContinuationSettlement === settlement) { if (this._actionStore.queuedActions().length > 0) { - if (this._sessionInputPumpSuspended || this._queuedWorkPauses.size > 0) { + // Park while the pump would refuse scheduling or selection: rescheduling + // a blocked pump completes on already-resolved promises, so looping here + // would spin on the microtask queue and starve the IO that ends the busy state. + if (this._isBusyForSessionInput("pump")) { let wake = () => {}; const changed = new Promise((resolve) => { wake = resolve; diff --git a/packages/coding-agent/src/core/kernel/repl-manager.ts b/packages/coding-agent/src/core/kernel/repl-manager.ts index 5023e487299..31a1408f30b 100644 --- a/packages/coding-agent/src/core/kernel/repl-manager.ts +++ b/packages/coding-agent/src/core/kernel/repl-manager.ts @@ -432,6 +432,19 @@ export class ReplKernelManager { this.appendKernelDiagnostic(`kernel stderr log close failed: ${errorMessage(error)}`); } }); + // A pipe write into a dead kernel surfaces as an 'error' event on the + // stream (write EPIPE); without a listener Node rethrows it and takes + // down the whole worker. The pending writeLine rejection and the child + // 'exit' handler below own the fallout, so this only records the + // diagnosis. + child.stdin?.on("error", (error) => { + if (this.child !== child) return; + this.appendKernelDiagnostic(`kernel stdin error: ${errorMessage(error)}`); + }); + child.stdout?.on("error", (error) => { + if (this.child !== child) return; + this.appendKernelDiagnostic(`kernel stdout error: ${errorMessage(error)}`); + }); child.once("exit", () => { // One turn for the poll phase to deliver the bytes the kernel wrote // before dying (the pipe buffer bounds them), then destroy: EOF may diff --git a/packages/coding-agent/test/repl-kernel-pipe-errors.test.ts b/packages/coding-agent/test/repl-kernel-pipe-errors.test.ts new file mode 100644 index 00000000000..46266cd42cb --- /dev/null +++ b/packages/coding-agent/test/repl-kernel-pipe-errors.test.ts @@ -0,0 +1,62 @@ +import { type ChildProcess, spawnSync } from "node:child_process"; +import { existsSync, mkdtempSync, rmSync } from "node:fs"; +import { homedir, tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { ReplKernelManager } from "../src/core/kernel/index.js"; + +function resolveReplPython(): string | null { + const candidates = [ + process.env.PRIME_AGENT_KERNEL_PYTHON, + resolve(__dirname, "..", "..", "..", "prime-agent-runtime", ".venv", "bin", "python"), + join(homedir(), ".prime", "agent", "kernel-venv", "bin", "python"), + ].filter((p): p is string => Boolean(p)); + for (const python of candidates) { + if (!existsSync(python)) continue; + const check = spawnSync(python, ["-c", "import rlm.repl, dill"], { encoding: "utf8" }); + if (check.status === 0) return python; + } + return null; +} + +const python = resolveReplPython(); +const describeIf = python ? describe : describe.skip; + +describeIf("ReplKernelManager pipe errors (real runtime)", () => { + let dir = ""; + let manager: ReplKernelManager | undefined; + + beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "prime-agent-repl-pipe-")); + }); + + afterEach(async () => { + await manager?.shutdown({ snapshot: true, drainHostRequests: true }); + manager = undefined; + if (dir) { + rmSync(dir, { recursive: true, force: true }); + dir = ""; + } + }); + + it("absorbs kernel pipe write errors instead of crashing the host process", async () => { + manager = new ReplKernelManager({ python: python as string, cwd: dir }); + const first = await manager.execute("x = 1"); + expect(first.status).toBe("ok"); + + const child = (manager as unknown as { child?: ChildProcess }).child; + expect(child?.stdin).toBeDefined(); + expect(child?.stdout).toBeDefined(); + // A write racing the kernel's death lands as an 'error' event on the pipe; + // without a listener Node would crash the worker (observed in production as + // "uncaught exception: Error: write EPIPE"). + expect(() => child?.stdin?.emit("error", new Error("write EPIPE"))).not.toThrow(); + expect(() => child?.stdout?.emit("error", new Error("read ECONNRESET"))).not.toThrow(); + + // The kernel is still healthy: further cells keep executing and shutdown + // (afterEach) still completes. + const second = await manager.execute("x + 1"); + expect(second.status).toBe("ok"); + expect(second.result).toBe("2"); + }, 30_000); +}); diff --git a/packages/coding-agent/test/repl-kernel-shutdown.test.ts b/packages/coding-agent/test/repl-kernel-shutdown.test.ts index 081942902b1..86d404e9167 100644 --- a/packages/coding-agent/test/repl-kernel-shutdown.test.ts +++ b/packages/coding-agent/test/repl-kernel-shutdown.test.ts @@ -17,7 +17,11 @@ type ShutdownInternals = { signalCode: NodeJS.Signals | null; kill: (signal?: NodeJS.Signals | number) => boolean; pid?: number; - stdin: { destroyed: boolean; destroy: () => void }; + stdin: { + destroyed: boolean; + destroy: () => void; + on: (event: string, listener: (...args: unknown[]) => void) => void; + }; stdout?: { destroy: () => void; on: (event: string, listener: (...args: unknown[]) => void) => void }; stderr?: { destroy: () => void; @@ -46,7 +50,7 @@ function configuredManager( signalCode: null, kill: vi.fn(() => true), pid: undefined, - stdin: { destroyed: false, destroy: vi.fn() }, + stdin: { destroyed: false, destroy: vi.fn(), on: vi.fn() }, stdout: { destroy: vi.fn(), on: vi.fn() }, stderr: { destroy: vi.fn(), on: vi.fn(), once: vi.fn() }, }); diff --git a/packages/coding-agent/test/suite/agent-session-queue.test.ts b/packages/coding-agent/test/suite/agent-session-queue.test.ts index c993bd65821..afe121da903 100644 --- a/packages/coding-agent/test/suite/agent-session-queue.test.ts +++ b/packages/coding-agent/test/suite/agent-session-queue.test.ts @@ -29,6 +29,7 @@ import { saveHarnessState, } from "../../src/core/refinement/index.js"; import { parseSessionSlashCommand } from "../../src/core/slash-commands.js"; +import type { BashOperations } from "../../src/core/tools/bash.js"; import { conversationMessages, createHarness, @@ -3631,4 +3632,50 @@ describe("AgentSession scheduler scenarios", () => { } } }); + + it("waitForIdle parks instead of microtask-spinning while a running bash blocks queued input", async () => { + const harness = await createHarness(); + harnesses.push(harness); + const session = harness.session; + let releaseBash!: () => void; + const gate = new Promise<{ exitCode: number | null }>((resolve) => { + releaseBash = () => resolve({ exitCode: 0 }); + }); + const operations: BashOperations = { exec: async () => await gate }; + const bashPromise = session.executeBash("blocked", undefined, { operations }); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(session.isBashRunning).toBe(true); + + harness.setResponses([fauxAssistantMessage("first done"), fauxAssistantMessage("second done")]); + const firstPrompt = session.prompt("queued while bash runs"); + await new Promise((resolve) => setTimeout(resolve, 0)); + + // Count pump scheduling from the idle wait. Unfixed, the wait loop respun the + // blocked pump purely in microtasks, so the setImmediate below never fired and + // only the 200th reschedule (the escape hatch) released the gate. + const internals = session as unknown as { _scheduleSessionInputPump(): void }; + const originalSchedule = internals._scheduleSessionInputPump.bind(session); + let scheduleCount = 0; + let secondPrompt: Promise | undefined; + internals._scheduleSessionInputPump = () => { + scheduleCount++; + if (scheduleCount === 200) releaseBash(); + originalSchedule(); + }; + + const idle = session.waitForIdle(); + setImmediate(() => { + // An arrival during the park must not be lost once the busy state clears. + secondPrompt = session.prompt("queued during park"); + secondPrompt.catch(() => undefined); + releaseBash(); + }); + await idle; + await bashPromise; + await firstPrompt; + await secondPrompt; + + expect(scheduleCount).toBeLessThan(200); + expect(getAssistantTexts(harness)).toEqual(["first done", "second done"]); + }); });