Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions .pylon/features.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -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.
9 changes: 9 additions & 0 deletions .pylon/upstream-review.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
1 change: 1 addition & 0 deletions packages/ai/.changes/fix-codex-chain-reset.md
Original file line number Diff line number Diff line change
@@ -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.
120 changes: 77 additions & 43 deletions packages/ai/src/providers/openai-codex-responses.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}

Expand Down Expand Up @@ -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<Record<string, unknown>>): AsyncGenerator<ResponseStreamEvent> {
for await (const event of events) {
const type = typeof event.type === "string" ? event.type : undefined;
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -1144,6 +1177,7 @@ async function processWebSocketStream(
lastRequestBody: fullBody,
lastResponseId: output.responseId,
lastResponseItems: responseItems,
connection: entry.socket,
};
}
} catch (error) {
Expand Down
Loading
Loading