From 22c7c3142312bbe7d2ccf8432b57ab709f8344fb Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 07:04:14 +0000 Subject: [PATCH 01/12] test(server): cover Codex context resend after app-server restart Regression test for #25191: after a restarted app-server resumes a thread, an unchanged additionalContext must not be resent, a changed one must, and compaction must still restore it. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 204 ++++++++++++++++++ 1 file changed, 204 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index bd7b6f420876..5b2f9dfe520a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1817,6 +1817,210 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + // regression: #25191 a restarted app-server re-appends unchanged context after thread/resume + it.effect("does not resend unchanged T3 context after an app-server restart", () => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "restart-context-thread"; + const mcpSessions = yield* McpProviderSessions.McpProviderSessions; + const withT3Mcp = (threadId: ThreadId) => + mcpSessions.set({ + environmentId: EnvironmentId.make("test"), + threadId, + providerSessionId: "restart-context-session", + providerInstanceId: ProviderInstanceId.make("codex"), + endpoint: "http://127.0.0.1:43123/mcp", + authorizationHeader: "Bearer test", + browserToolsAvailable: true, + }); + const contextParams = ( + text: string, + appContext?: ProviderAdapter.ProviderAdapterV2TurnInput["appContext"], + ) => + CodexAdapterV2.buildCodexTurnStartParams({ + nativeThreadId, + codexInput: [{ type: "text", text }], + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + modelSelection: CODEX_TEST_MODEL_SELECTION, + hasT3Mcp: true, + ...(appContext === undefined ? {} : { appContext }), + }); + const completed = (id: number, nativeTurnId: string, method: string, params: unknown) => + [ + { type: "expect_outbound", label: method, frame: { id, method, params } }, + { + type: "emit_inbound", + label: method, + frame: { + id, + result: { turn: makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }) }, + }, + }, + { + type: "emit_inbound", + label: `${nativeTurnId} completed`, + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + }, + }, + }, + ] satisfies Array; + const preamble = codexReplayPreamble({ + nativeThreadId, + nativeTurnId: "unused-turn", + prompt: "unused-prompt", + }).slice(0, 5); + const firstParams = yield* contextParams("before restart"); + + const beforeRestart = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "restart-context-before", + entries: [...preamble, ...completed(3, "turn-before", "turn/start", firstParams)], + }), + ); + yield* withT3Mcp(beforeRestart.threadId); + yield* beforeRestart.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: beforeRestart.threadId, + providerThread: beforeRestart.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("restart-context-before"), + text: "before restart", + }), + ); + yield* beforeRestart.firstTerminal; + const persistedProviderThread = + beforeRestart.events.findLast((event) => event.type === "provider_thread.updated") + ?.providerThread ?? beforeRestart.providerThread; + + const unchangedParams = yield* contextParams("after restart"); + const { additionalContext: _unchanged, ...unchangedWithoutContext } = unchangedParams; + const appContext = [{ key: "mcp_app_todos_item-1", text: "Filtered to overdue" }]; + const changedParams = yield* contextParams("changed context", appContext); + const afterRestart = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "restart-context-after", + entries: [ + ...preamble, + { + type: "expect_outbound", + label: "thread/resume", + frame: { + id: 3, + method: "thread/resume", + params: { + threadId: nativeThreadId, + excludeTurns: true, + config: CodexAdapterV2.CODEX_THREAD_CONFIG, + }, + }, + }, + { + type: "emit_inbound", + label: "thread/resume", + frame: { id: 3, result: { thread: { id: nativeThreadId, updatedAt: 1782622450 } } }, + }, + { + type: "expect_outbound", + label: "unchanged context turn", + frame: { id: 4, method: "turn/start", params: unchangedWithoutContext }, + }, + { + type: "emit_inbound", + label: "unchanged context turn", + frame: { + id: 4, + result: { turn: makeCodexReplayTurn({ id: "turn-after", status: "inProgress" }) }, + }, + }, + { + type: "emit_inbound", + label: "compacted", + frame: { + method: "item/completed", + params: { + threadId: nativeThreadId, + turnId: "turn-after", + item: { type: "contextCompaction", id: "compact-after-restart" }, + }, + }, + }, + { + type: "expect_outbound", + label: "restore context", + frame: { + id: 5, + method: "thread/inject_items", + params: { + threadId: nativeThreadId, + items: Object.entries(unchangedParams.additionalContext ?? {}).map( + ([key, entry]) => ({ + type: "message", + role: "developer", + content: [{ type: "input_text", text: `<${key}>${entry.value}` }], + }), + ), + }, + }, + }, + { type: "emit_inbound", label: "restored", frame: { id: 5, result: {} } }, + { + type: "emit_inbound", + label: "turn-after completed", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "turn-after", status: "completed" }), + }, + }, + }, + ...completed(6, "turn-changed", "turn/start", changedParams), + ], + }), + ); + yield* withT3Mcp(afterRestart.threadId); + const resumed = yield* afterRestart.runtime.resumeThread({ + providerThread: persistedProviderThread, + modelSelection: CODEX_TEST_MODEL_SELECTION, + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + }); + yield* afterRestart.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: afterRestart.threadId, + providerThread: resumed, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("restart-context-unchanged"), + text: "after restart", + }), + ); + yield* afterRestart.awaitTerminal(1); + yield* afterRestart.runtime.startTurn({ + ...makeCodexTestTurnInput({ + threadId: afterRestart.threadId, + providerThread: resumed, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("restart-context-changed"), + text: "changed context", + }), + appContext, + }); + yield* afterRestart.awaitTerminal(2); + assert.deepEqual( + afterRestart.terminalEvents().map((event) => event.status), + ["completed", "completed"], + ); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + it.effect("compacts Codex with the native RPC and completes the compaction turn", () => Effect.scoped( Effect.gen(function* () { From 1c2f42627da75eec186cd26cc3369a0923f4c5bf Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 07:13:23 +0000 Subject: [PATCH 02/12] fix(server): Codex no longer resends unchanged context after an app-server restart Codex dedupes turn additionalContext in an in-memory cache, so after an app-server restart and thread/resume every turn re-appended the same T3 context. Record a SHA-256 digest of the last successfully delivered context in the provider thread's nativeMetadata and omit additionalContext when the next turn's context matches it. Compaction still restores the full current context, changed context is sent in full, and rollback drops the digest. Fixes pingdotgg/t3code#25191 Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.ts | 57 +++++++++++++++++-- packages/contracts/src/orchestrationV2.ts | 2 + 2 files changed, 55 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 10b673459e56..356dfc8c17be 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -79,6 +79,7 @@ import * as CodexSchema from "effect-codex-app-server/schema"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; +import * as Hex from "effect/encoding/Hex"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -665,12 +666,17 @@ const decodeTurnReasoningEffort = Schema.decodeUnknownEffect( Schema.Union([CodexSchema.V2TurnStartParams__ReasoningEffort, Schema.Null]), ); +const CodexAdditionalContext = Schema.Record( + Schema.String, + CodexSchema.V2TurnStartParams__AdditionalContextEntry, +); +const encodeCodexAdditionalContextJson = Schema.encodeEffect( + Schema.fromJsonString(CodexAdditionalContext), +); const CodexTurnStartParamsWithCollaborationMode = CodexSchema.V2TurnStartParams.pipe( Schema.fieldsAssign({ collaborationMode: Schema.optionalKey(CodexSchema.ClientRequest__CollaborationMode), - additionalContext: Schema.optionalKey( - Schema.Record(Schema.String, CodexSchema.V2TurnStartParams__AdditionalContextEntry), - ), + additionalContext: Schema.optionalKey(CodexAdditionalContext), }), ); type CodexTurnStartParamsWithCollaborationMode = @@ -1834,6 +1840,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const turnTerminalizationPermit = yield* Semaphore.make(1); const goalsByNativeThread = new Map(); const rootProviderThreads = new Map(); + // Absent: fall back to persisted metadata. Undefined: Codex holds no known context. + const deliveredContextDigests = new Map(); const goalHolds = new Map(); const goalRuns = new Map>(); const goalActivations = new Map(); @@ -6274,7 +6282,44 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( else next.delete(threadId); return next; }); - const started = yield* client.request("turn/start", turnStartParams); + const { additionalContext, ...paramsWithoutContext } = turnStartParams; + const contextDigest = + additionalContext === undefined + ? undefined + : Hex.encode( + yield* crypto.digest( + "SHA-256", + new TextEncoder().encode( + yield* encodeCodexAdditionalContextJson(additionalContext), + ), + ), + ); + const deliveredDigest = deliveredContextDigests.has(threadId) + ? deliveredContextDigests.get(threadId) + : turnInput.providerThread.nativeMetadata?.codexAdditionalContextDigest; + const started = yield* client.request( + "turn/start", + contextDigest !== undefined && contextDigest === deliveredDigest + ? paramsWithoutContext + : turnStartParams, + ); + if (contextDigest !== undefined && contextDigest !== deliveredDigest) { + deliveredContextDigests.set(threadId, contextDigest); + const providerThread = rootProviderThreads.get(threadId) ?? turnInput.providerThread; + const updated = { + ...providerThread, + nativeMetadata: { + ...providerThread.nativeMetadata, + codexAdditionalContextDigest: contextDigest, + }, + }; + rootProviderThreads.set(threadId, updated); + yield* emitProviderEvent({ + type: "provider_thread.updated", + driver: CODEX_PROVIDER, + providerThread: updated, + }); + } yield* registerRootTurn({ turnInput, nativeTurnId: started.turn.id, @@ -7312,9 +7357,13 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( Effect.andThen(revertCodexThread(client, threadId, numTurns)), ); turnTokenUsageByThread.delete(threadId); + deliveredContextDigests.set(response.thread.id, undefined); + const { codexAdditionalContextDigest: _reverted, ...nativeMetadata } = + threadInput.providerThread.nativeMetadata ?? {}; return { providerThread: { ...threadInput.providerThread, + ...(_reverted === undefined ? {} : { nativeMetadata }), nativeThreadRef: { driver: CODEX_PROVIDER, nativeId: response.thread.id, diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index b3a85fa3700d..7c6f620e0656 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -921,6 +921,8 @@ export const OrchestrationV2ProviderThreadNativeMetadata = Schema.Struct({ claudeAttachmentBudget: Schema.optional( Schema.Struct({ bytesUsed: NonNegativeInt, pagesUsed: NonNegativeInt }), ), + // Codex dedupes additionalContext in memory only; this survives app-server restarts. + codexAdditionalContextDigest: Schema.optional(TrimmedNonEmptyString), }); export type OrchestrationV2ProviderThreadNativeMetadata = typeof OrchestrationV2ProviderThreadNativeMetadata.Type; From 54c3734a05d66481aaa20894e7d8fbab96c1c383 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 07:23:46 +0000 Subject: [PATCH 03/12] fix(server): Codex resends context after an empty compaction restore Compacting a resumed Codex thread before its first turn had no context to restore, yet the persisted digest still suppressed the next turn's additionalContext, so it was never resent. Clear the delivered digest when restoration has nothing to inject or fails, forcing a full resend. The digest provider_thread.updated was built from the adapter's cached root provider thread, which could predate this turn and drop lastRunOrdinal and handoffIds. Build it from the turn's provider thread and register the turn with the updated thread so later goal updates keep the digest. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 118 +++++++++++++++++- .../Adapters/CodexAdapterV2.ts | 37 +++--- 2 files changed, 140 insertions(+), 15 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 5b2f9dfe520a..f606abb0060b 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -8,6 +8,7 @@ import { CommandId, CheckpointId, CodexSettings, + ContextHandoffId, EnvironmentId, MessageId, type ModelSelection, @@ -1998,10 +1999,15 @@ describe("CodexAdapterV2 post-settle continuation", () => { }), ); yield* afterRestart.awaitTerminal(1); + const advancedProviderThread = { + ...resumed, + lastRunOrdinal: 2, + handoffIds: [ContextHandoffId.make("restart-context-handoff")], + }; yield* afterRestart.runtime.startTurn({ ...makeCodexTestTurnInput({ threadId: afterRestart.threadId, - providerThread: resumed, + providerThread: advancedProviderThread, now: yield* DateTime.now, attemptId: RunAttemptId.make("restart-context-changed"), text: "changed context", @@ -2013,6 +2019,116 @@ describe("CodexAdapterV2 post-settle continuation", () => { afterRestart.terminalEvents().map((event) => event.status), ["completed", "completed"], ); + const changedProviderThread = afterRestart.events.findLast( + (event) => event.type === "provider_thread.updated", + )?.providerThread; + assert.equal(changedProviderThread?.lastRunOrdinal, 2); + assert.deepEqual(changedProviderThread?.handoffIds, advancedProviderThread.handoffIds); + assert.notEqual( + changedProviderThread?.nativeMetadata?.codexAdditionalContextDigest, + persistedProviderThread.nativeMetadata?.codexAdditionalContextDigest, + ); + + // Compaction before the first post-restart turn has nothing to restore, so the next turn resends. + const compactTurnId = "turn-compact"; + const compactionItem = { type: "contextCompaction", id: "compact-before-turn" }; + const compactRestart = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "restart-context-compact", + entries: [ + ...preamble, + { + type: "expect_outbound", + label: "thread/resume", + frame: { + id: 3, + method: "thread/resume", + params: { + threadId: nativeThreadId, + excludeTurns: true, + config: CodexAdapterV2.CODEX_THREAD_CONFIG, + }, + }, + }, + { + type: "emit_inbound", + label: "thread/resume", + frame: { id: 3, result: { thread: { id: nativeThreadId, updatedAt: 1782622450 } } }, + }, + { + type: "expect_outbound", + label: "compact", + frame: { + id: 4, + method: "thread/compact/start", + params: { threadId: nativeThreadId }, + }, + }, + { type: "emit_inbound", label: "compact", frame: { id: 4, result: {} } }, + { + type: "emit_inbound", + label: "compact started", + frame: { + method: "turn/started", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: compactTurnId, status: "inProgress" }), + }, + }, + }, + ...(["item/started", "item/completed"] as const).map((method) => ({ + type: "emit_inbound" as const, + label: method, + frame: { + method, + params: { threadId: nativeThreadId, turnId: compactTurnId, item: compactionItem }, + }, + })), + { + type: "emit_inbound", + label: "compact completed", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: compactTurnId, status: "completed" }), + }, + }, + }, + ...completed(5, "turn-after-compact", "turn/start", unchangedParams), + ], + }), + ); + yield* withT3Mcp(compactRestart.threadId); + const compactResumed = yield* compactRestart.runtime.resumeThread({ + providerThread: persistedProviderThread, + modelSelection: CODEX_TEST_MODEL_SELECTION, + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + }); + yield* compactRestart.runtime.compactThread!( + makeCodexTestTurnInput({ + threadId: compactRestart.threadId, + providerThread: compactResumed, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("restart-context-compact"), + text: "/compact", + }), + ); + yield* compactRestart.awaitTerminal(1); + yield* compactRestart.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: compactRestart.threadId, + providerThread: compactResumed, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("restart-context-after-compact"), + text: "after restart", + }), + ); + yield* compactRestart.awaitTerminal(2); + assert.deepEqual( + compactRestart.terminalEvents().map((event) => event.status), + ["completed", "completed"], + ); }).pipe( Effect.provide( Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 356dfc8c17be..252cc8ef0805 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1672,12 +1672,18 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( NonNullable >(), ); + // Absent: fall back to persisted metadata. Undefined: Codex holds no known context. + const deliveredContextDigests = new Map(); // Codex drops client developer messages during compaction but only // resends additionalContext when it changes. const restoreAdditionalContext = Effect.fn("CodexAdapterV2.restoreAdditionalContext")( function* (threadId: string) { const context = (yield* Ref.get(additionalContextByThread)).get(threadId); - if (!context) return; + // Nothing to restore before this session's first turn, so the next turn must resend. + if (!context) { + deliveredContextDigests.set(threadId, undefined); + return; + } yield* client.request("thread/inject_items", { threadId, items: Object.entries(context).map(([key, entry]) => @@ -1700,12 +1706,17 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ), }); }, - Effect.timeout("10 seconds"), - Effect.catch((cause) => - Effect.logWarning("Failed to restore Codex additional context after compaction.", { - cause, - }), - ), + (effect, threadId) => + effect.pipe( + Effect.timeout("10 seconds"), + Effect.catch((cause) => { + deliveredContextDigests.set(threadId, undefined); + return Effect.logWarning( + "Failed to restore Codex additional context after compaction.", + { cause }, + ); + }), + ), ); const initialized = yield* Ref.make(false); // Codex rejects a second initialize on the shared app-server. @@ -1840,8 +1851,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const turnTerminalizationPermit = yield* Semaphore.make(1); const goalsByNativeThread = new Map(); const rootProviderThreads = new Map(); - // Absent: fall back to persisted metadata. Undefined: Codex holds no known context. - const deliveredContextDigests = new Map(); const goalHolds = new Map(); const goalRuns = new Map>(); const goalActivations = new Map(); @@ -6303,25 +6312,25 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ? paramsWithoutContext : turnStartParams, ); + let providerThread = turnInput.providerThread; if (contextDigest !== undefined && contextDigest !== deliveredDigest) { deliveredContextDigests.set(threadId, contextDigest); - const providerThread = rootProviderThreads.get(threadId) ?? turnInput.providerThread; - const updated = { + providerThread = { ...providerThread, nativeMetadata: { ...providerThread.nativeMetadata, codexAdditionalContextDigest: contextDigest, }, }; - rootProviderThreads.set(threadId, updated); + rootProviderThreads.set(threadId, providerThread); yield* emitProviderEvent({ type: "provider_thread.updated", driver: CODEX_PROVIDER, - providerThread: updated, + providerThread, }); } yield* registerRootTurn({ - turnInput, + turnInput: { ...turnInput, providerThread }, nativeTurnId: started.turn.id, startedAt: codexTimestamp(started.turn.startedAt), waitForNativeStart: started.turn.startedAt === null, From 0692553d3401eabcbf18ddb45f2e373bd5038c4e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 07:25:45 +0000 Subject: [PATCH 04/12] fix(server): persist Codex context invalidation before compaction --- .../Adapters/CodexAdapterV2.test.ts | 5 +++ .../Adapters/CodexAdapterV2.ts | 37 +++++++++++++++---- 2 files changed, 34 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index f606abb0060b..22ec498df886 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2115,6 +2115,11 @@ describe("CodexAdapterV2 post-settle continuation", () => { }), ); yield* compactRestart.awaitTerminal(1); + const invalidatedThread = compactRestart.events.findLast( + (event) => event.type === "provider_thread.updated", + )?.providerThread; + assert.isDefined(invalidatedThread); + assert.isUndefined(invalidatedThread?.nativeMetadata?.codexAdditionalContextDigest); yield* compactRestart.runtime.startTurn( makeCodexTestTurnInput({ threadId: compactRestart.threadId, diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 252cc8ef0805..13d8db7e529d 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1681,7 +1681,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const context = (yield* Ref.get(additionalContextByThread)).get(threadId); // Nothing to restore before this session's first turn, so the next turn must resend. if (!context) { - deliveredContextDigests.set(threadId, undefined); + yield* invalidateAdditionalContext(threadId); return; } yield* client.request("thread/inject_items", { @@ -1709,13 +1709,16 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( (effect, threadId) => effect.pipe( Effect.timeout("10 seconds"), - Effect.catch((cause) => { - deliveredContextDigests.set(threadId, undefined); - return Effect.logWarning( - "Failed to restore Codex additional context after compaction.", - { cause }, - ); - }), + Effect.catch((cause) => + invalidateAdditionalContext(threadId).pipe( + Effect.andThen( + Effect.logWarning( + "Failed to restore Codex additional context after compaction.", + { cause }, + ), + ), + ), + ), ), ); const initialized = yield* Ref.make(false); @@ -1862,6 +1865,22 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const emitProviderEvent = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); + const invalidateAdditionalContext = Effect.fn("CodexAdapterV2.invalidateAdditionalContext")( + function* (threadId: string) { + deliveredContextDigests.set(threadId, undefined); + const providerThread = rootProviderThreads.get(threadId); + if (!providerThread?.nativeMetadata?.codexAdditionalContextDigest) return; + const { codexAdditionalContextDigest: _digest, ...nativeMetadata } = + providerThread.nativeMetadata; + const updated = { ...providerThread, nativeMetadata }; + rootProviderThreads.set(threadId, updated); + yield* emitProviderEvent({ + type: "provider_thread.updated", + driver: CODEX_PROVIDER, + providerThread: updated, + }); + }, + ); const emitGoalUpdate = Effect.fnUntraced(function* (nativeThreadId: string) { const providerThread = rootProviderThreads.get(nativeThreadId); if (providerThread === undefined || !goalsByNativeThread.has(nativeThreadId)) return; @@ -6687,6 +6706,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* Ref.update(pendingRootTurns, (current) => new Map(current).set(threadId, turnInput), ); + rootProviderThreads.set(threadId, turnInput.providerThread); + yield* invalidateAdditionalContext(threadId); // Clear on interruption; a late turn/started must not be adopted after an interrupted start. yield* client.request("thread/compact/start", { threadId }).pipe( Effect.onError(() => From 70b2e1a615e9d2d4a1cbac3633d057d44c6d8c31 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 07:35:09 +0000 Subject: [PATCH 05/12] fix(server): retain Codex context delivery state across compaction --- .../Adapters/CodexAdapterV2.test.ts | 250 ++++++++++++------ .../Adapters/CodexAdapterV2.ts | 98 ++++--- 2 files changed, 221 insertions(+), 127 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 22ec498df886..b0e0e22c879a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1720,102 +1720,178 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); - it.effect("preserves T3 context on the wire and restores it after compaction", () => - Effect.scoped( - Effect.gen(function* () { - const nativeThreadId = "context-thread"; - const nativeTurnId = "context-turn"; - const params = yield* CodexAdapterV2.buildCodexTurnStartParams({ - nativeThreadId, - codexInput: [{ type: "text", text: "work" }], - runtimePolicy: CODEX_TEST_RUNTIME_POLICY, - modelSelection: CODEX_TEST_MODEL_SELECTION, - hasT3Mcp: true, - }); - assert.include( - params.additionalContext?.t3_code_orchestration_01?.value ?? "", - "delegate_task", - ); - const entries = codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "work" }); - const transcript = makeCodexReplayTranscript({ - scenario: "restore-context", - entries: [ - ...entries.slice(0, 5), - { - type: "expect_outbound", - label: "context turn", - frame: { id: 3, method: "turn/start", params }, - }, - ...entries.slice(6), - { - type: "emit_inbound", - label: "compacted", - frame: { - method: "item/completed", - params: { - threadId: nativeThreadId, - turnId: nativeTurnId, - item: { type: "contextCompaction", id: "compact-context" }, + it.effect.each([false, true])( + "restores T3 context and its digest after compaction (explicit: %s)", + (explicit) => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "context-thread"; + const nativeTurnId = "context-turn"; + const params = yield* CodexAdapterV2.buildCodexTurnStartParams({ + nativeThreadId, + codexInput: [{ type: "text", text: "work" }], + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + modelSelection: CODEX_TEST_MODEL_SELECTION, + hasT3Mcp: true, + }); + assert.include( + params.additionalContext?.t3_code_orchestration_01?.value ?? "", + "delegate_task", + ); + const entries = codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "work" }); + const transcript = makeCodexReplayTranscript({ + scenario: "restore-context", + entries: [ + ...entries.slice(0, 5), + { + type: "expect_outbound", + label: "context turn", + frame: { id: 3, method: "turn/start", params }, + }, + ...entries.slice(6), + ...(explicit + ? [ + { + type: "emit_inbound" as const, + label: "first turn completed", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + }, + }, + }, + { + type: "expect_outbound" as const, + label: "explicit compact", + frame: { + id: 4, + method: "thread/compact/start", + params: { threadId: nativeThreadId }, + }, + }, + { + type: "emit_inbound" as const, + label: "explicit compact", + frame: { id: 4, result: {} }, + }, + { + type: "emit_inbound" as const, + label: "compact started", + frame: { + method: "turn/started", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ + id: "explicit-compact-turn", + status: "inProgress", + }), + }, + }, + }, + ] + : []), + { + type: "emit_inbound", + label: "compacted", + frame: { + method: "item/completed", + params: { + threadId: nativeThreadId, + turnId: explicit ? "explicit-compact-turn" : nativeTurnId, + item: { type: "contextCompaction", id: "compact-context" }, + }, }, }, - }, - { - type: "expect_outbound", - label: "restore context", - frame: { - id: 4, - method: "thread/inject_items", - params: { - threadId: nativeThreadId, - items: Object.entries(params.additionalContext ?? {}).map(([key, entry]) => ({ - type: "message", - role: "developer", - content: [{ type: "input_text", text: `<${key}>${entry.value}` }], - })), + { + type: "expect_outbound", + label: "restore context", + frame: { + id: explicit ? 5 : 4, + method: "thread/inject_items", + params: { + threadId: nativeThreadId, + items: Object.entries(params.additionalContext ?? {}).map(([key, entry]) => ({ + type: "message", + role: "developer", + content: [{ type: "input_text", text: `<${key}>${entry.value}` }], + })), + }, }, }, - }, - { type: "emit_inbound", label: "restored", frame: { id: 4, result: {} } }, - { - type: "emit_inbound", - label: "done", - frame: { - method: "turn/completed", - params: { - threadId: nativeThreadId, - turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + { + type: "emit_inbound", + label: "restored", + frame: { id: explicit ? 5 : 4, result: {} }, + }, + { + type: "emit_inbound", + label: "done", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ + id: explicit ? "explicit-compact-turn" : nativeTurnId, + status: "completed", + }), + }, }, }, - }, - ], - }); - const harness = yield* makeCodexReplayHarness(transcript); - yield* (yield* McpProviderSessions.McpProviderSessions).set({ - environmentId: EnvironmentId.make("test"), - threadId: harness.threadId, - providerSessionId: "context-session", - providerInstanceId: ProviderInstanceId.make("codex"), - endpoint: "http://127.0.0.1:43123/mcp", - authorizationHeader: "Bearer test", - browserToolsAvailable: true, - }); - yield* harness.runtime.startTurn( - makeCodexTestTurnInput({ + ], + }); + const harness = yield* makeCodexReplayHarness(transcript); + yield* (yield* McpProviderSessions.McpProviderSessions).set({ + environmentId: EnvironmentId.make("test"), threadId: harness.threadId, - providerThread: harness.providerThread, - now: yield* DateTime.now, - attemptId: RunAttemptId.make("context-attempt"), - text: "work", - }), - ); - yield* harness.firstTerminal; - assert.equal(harness.terminalEvents()[0]?.status, "completed"); - }).pipe( - Effect.provide( - Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + providerSessionId: "context-session", + providerInstanceId: ProviderInstanceId.make("codex"), + endpoint: "http://127.0.0.1:43123/mcp", + authorizationHeader: "Bearer test", + browserToolsAvailable: true, + }); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("context-attempt"), + text: "work", + }), + ); + yield* harness.firstTerminal; + assert.equal(harness.terminalEvents()[0]?.status, "completed"); + if (explicit) { + const delivered = harness.events.findLast( + (event) => event.type === "provider_thread.updated", + )?.providerThread; + assert.isDefined(delivered?.nativeMetadata?.codexAdditionalContextDigest); + if (!delivered) return yield* Effect.die("Missing delivered provider thread"); + yield* harness.runtime.compactThread!( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: delivered, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("explicit-context-compact"), + text: "/compact", + }), + ); + yield* harness.awaitTerminal(2); + const restored = harness.events.findLast( + (event) => event.type === "provider_thread.updated", + )?.providerThread; + assert.equal( + restored?.nativeMetadata?.codexAdditionalContextDigest, + delivered?.nativeMetadata?.codexAdditionalContextDigest, + ); + } + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), ), ), - ), ); // regression: #25191 a restarted app-server re-appends unchanged context after thread/resume diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 13d8db7e529d..b587273f6c94 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1672,6 +1672,16 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( NonNullable >(), ); + const digestAdditionalContext = Effect.fn("CodexAdapterV2.digestAdditionalContext")( + function* (context: typeof CodexAdditionalContext.Type) { + return Hex.encode( + yield* crypto.digest( + "SHA-256", + new TextEncoder().encode(yield* encodeCodexAdditionalContextJson(context)), + ), + ); + }, + ); // Absent: fall back to persisted metadata. Undefined: Codex holds no known context. const deliveredContextDigests = new Map(); // Codex drops client developer messages during compaction but only @@ -1705,6 +1715,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }, ), }); + yield* recordAdditionalContextDigest(threadId, yield* digestAdditionalContext(context)); }, (effect, threadId) => effect.pipe( @@ -1865,22 +1876,42 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const emitProviderEvent = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); - const invalidateAdditionalContext = Effect.fn("CodexAdapterV2.invalidateAdditionalContext")( - function* (threadId: string) { - deliveredContextDigests.set(threadId, undefined); - const providerThread = rootProviderThreads.get(threadId); - if (!providerThread?.nativeMetadata?.codexAdditionalContextDigest) return; - const { codexAdditionalContextDigest: _digest, ...nativeMetadata } = - providerThread.nativeMetadata; - const updated = { ...providerThread, nativeMetadata }; - rootProviderThreads.set(threadId, updated); - yield* emitProviderEvent({ - type: "provider_thread.updated", - driver: CODEX_PROVIDER, - providerThread: updated, - }); - }, - ); + const withCurrentContextDigest = ( + threadId: string, + providerThread: OrchestrationV2ProviderThread, + ) => { + if (!deliveredContextDigests.has(threadId)) return providerThread; + const { codexAdditionalContextDigest: _digest, ...nativeMetadata } = + providerThread.nativeMetadata ?? {}; + const digest = deliveredContextDigests.get(threadId); + return { + ...providerThread, + nativeMetadata: { + ...nativeMetadata, + ...(digest === undefined ? {} : { codexAdditionalContextDigest: digest }), + }, + }; + }; + const recordAdditionalContextDigest = Effect.fn( + "CodexAdapterV2.recordAdditionalContextDigest", + )(function* (threadId: string, digest: string | undefined) { + deliveredContextDigests.set(threadId, digest); + const providerThread = rootProviderThreads.get(threadId); + if ( + !providerThread || + providerThread.nativeMetadata?.codexAdditionalContextDigest === digest + ) + return; + const updated = withCurrentContextDigest(threadId, providerThread); + rootProviderThreads.set(threadId, updated); + yield* emitProviderEvent({ + type: "provider_thread.updated", + driver: CODEX_PROVIDER, + providerThread: updated, + }); + }); + const invalidateAdditionalContext = (threadId: string) => + recordAdditionalContextDigest(threadId, undefined); const emitGoalUpdate = Effect.fnUntraced(function* (nativeThreadId: string) { const providerThread = rootProviderThreads.get(nativeThreadId); if (providerThread === undefined || !goalsByNativeThread.has(nativeThreadId)) return; @@ -1900,7 +1931,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ) { const nativeThreadId = providerThread.nativeThreadRef?.nativeId; if (nativeThreadId == null) return; - rootProviderThreads.set(nativeThreadId, providerThread); + rootProviderThreads.set( + nativeThreadId, + withCurrentContextDigest(nativeThreadId, providerThread), + ); yield* emitGoalUpdate(nativeThreadId); }); @@ -6314,14 +6348,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const contextDigest = additionalContext === undefined ? undefined - : Hex.encode( - yield* crypto.digest( - "SHA-256", - new TextEncoder().encode( - yield* encodeCodexAdditionalContextJson(additionalContext), - ), - ), - ); + : yield* digestAdditionalContext(additionalContext); const deliveredDigest = deliveredContextDigests.has(threadId) ? deliveredContextDigests.get(threadId) : turnInput.providerThread.nativeMetadata?.codexAdditionalContextDigest; @@ -6331,23 +6358,14 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ? paramsWithoutContext : turnStartParams, ); - let providerThread = turnInput.providerThread; + rootProviderThreads.set( + threadId, + withCurrentContextDigest(threadId, turnInput.providerThread), + ); if (contextDigest !== undefined && contextDigest !== deliveredDigest) { - deliveredContextDigests.set(threadId, contextDigest); - providerThread = { - ...providerThread, - nativeMetadata: { - ...providerThread.nativeMetadata, - codexAdditionalContextDigest: contextDigest, - }, - }; - rootProviderThreads.set(threadId, providerThread); - yield* emitProviderEvent({ - type: "provider_thread.updated", - driver: CODEX_PROVIDER, - providerThread, - }); + yield* recordAdditionalContextDigest(threadId, contextDigest); } + const providerThread = rootProviderThreads.get(threadId) ?? turnInput.providerThread; yield* registerRootTurn({ turnInput: { ...turnInput, providerThread }, nativeTurnId: started.turn.id, From da0d8e2e2e702161ad1252e90e65198c37dbc251 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 07:56:34 +0000 Subject: [PATCH 06/12] fix(server): invalidate context before automatic compaction restoration --- .../Adapters/CodexAdapterV2.test.ts | 101 +++++++++++++++--- .../Adapters/CodexAdapterV2.ts | 7 +- 2 files changed, 86 insertions(+), 22 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index b0e0e22c879a..4afeb6f68c19 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1734,6 +1734,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { modelSelection: CODEX_TEST_MODEL_SELECTION, hasT3Mcp: true, }); + const { additionalContext: _context, ...paramsWithoutContext } = params; assert.include( params.additionalContext?.t3_code_orchestration_01?.value ?? "", "delegate_task", @@ -1749,19 +1750,19 @@ describe("CodexAdapterV2 post-settle continuation", () => { frame: { id: 3, method: "turn/start", params }, }, ...entries.slice(6), + { + type: "emit_inbound" as const, + label: "first turn completed", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + }, + }, + }, ...(explicit ? [ - { - type: "emit_inbound" as const, - label: "first turn completed", - frame: { - method: "turn/completed", - params: { - threadId: nativeThreadId, - turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), - }, - }, - }, { type: "expect_outbound" as const, label: "explicit compact", @@ -1791,7 +1792,44 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, }, ] - : []), + : [ + { + type: "expect_outbound" as const, + label: "next turn", + frame: { + id: 4, + method: "turn/start", + params: paramsWithoutContext, + }, + }, + { + type: "emit_inbound" as const, + label: "next turn", + frame: { + id: 4, + result: { + turn: makeCodexReplayTurn({ + id: "automatic-compact-turn", + status: "inProgress", + }), + }, + }, + }, + { + type: "emit_inbound" as const, + label: "next turn started", + frame: { + method: "turn/started", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ + id: "automatic-compact-turn", + status: "inProgress", + }), + }, + }, + }, + ]), { type: "emit_inbound", label: "compacted", @@ -1799,7 +1837,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { method: "item/completed", params: { threadId: nativeThreadId, - turnId: explicit ? "explicit-compact-turn" : nativeTurnId, + turnId: explicit ? "explicit-compact-turn" : "automatic-compact-turn", item: { type: "contextCompaction", id: "compact-context" }, }, }, @@ -1808,7 +1846,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { type: "expect_outbound", label: "restore context", frame: { - id: explicit ? 5 : 4, + id: 5, method: "thread/inject_items", params: { threadId: nativeThreadId, @@ -1823,7 +1861,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { { type: "emit_inbound", label: "restored", - frame: { id: explicit ? 5 : 4, result: {} }, + frame: { id: 5, result: {} }, }, { type: "emit_inbound", @@ -1833,7 +1871,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { params: { threadId: nativeThreadId, turn: makeCodexReplayTurn({ - id: explicit ? "explicit-compact-turn" : nativeTurnId, + id: explicit ? "explicit-compact-turn" : "automatic-compact-turn", status: "completed", }), }, @@ -1841,7 +1879,15 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, ], }); - const harness = yield* makeCodexReplayHarness(transcript); + const contextDigests: Array = []; + const harness = yield* makeCodexReplayHarness(transcript, (event) => + Effect.sync(() => { + if (event.type === "provider_thread.updated") + contextDigests.push( + event.providerThread.nativeMetadata?.codexAdditionalContextDigest, + ); + }), + ); yield* (yield* McpProviderSessions.McpProviderSessions).set({ environmentId: EnvironmentId.make("test"), threadId: harness.threadId, @@ -1862,6 +1908,27 @@ describe("CodexAdapterV2 post-settle continuation", () => { ); yield* harness.firstTerminal; assert.equal(harness.terminalEvents()[0]?.status, "completed"); + if (!explicit) { + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("automatic-compact"), + text: "work", + }), + ); + yield* harness.awaitTerminal(2); + } + if (!explicit) { + assert.lengthOf(contextDigests, 3); + assert.isDefined(contextDigests[0]); + assert.isUndefined( + contextDigests[1], + "automatic compaction must persist invalidation before restoration", + ); + assert.equal(contextDigests[2], contextDigests[0]); + } if (explicit) { const delivered = harness.events.findLast( (event) => event.type === "provider_thread.updated", diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 77bbc8ccfd45..b749b023e7bb 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1688,12 +1688,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( // resends additionalContext when it changes. const restoreAdditionalContext = Effect.fn("CodexAdapterV2.restoreAdditionalContext")( function* (threadId: string) { + yield* invalidateAdditionalContext(threadId); const context = (yield* Ref.get(additionalContextByThread)).get(threadId); - // Nothing to restore before this session's first turn, so the next turn must resend. - if (!context) { - yield* invalidateAdditionalContext(threadId); - return; - } + if (!context) return; yield* client.request("thread/inject_items", { threadId, items: Object.entries(context).map(([key, entry]) => From f9dc8ac7cce07b7eaa5e44deb3ca2493e72fd6cc Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 30 Oct 2026 06:47:09 +0000 Subject: [PATCH 07/12] test(server): cover Codex context digest when turn/started precedes its response Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 83 +++++++++++++++++++ 1 file changed, 83 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index bd2f05f3c214..78a3a269b499 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2458,6 +2458,89 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("persists the T3 context digest when turn/started precedes its response", () => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "early-started-context-thread"; + const nativeTurnId = "early-started-context-turn"; + const params = yield* CodexAdapterV2.buildCodexTurnStartParams({ + nativeThreadId, + codexInput: [{ type: "text", text: "hello" }], + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + modelSelection: CODEX_TEST_MODEL_SELECTION, + hasT3Mcp: true, + }); + const turnEvent = (method: "turn/started" | "turn/completed", status: string) => ({ + type: "emit_inbound" as const, + label: method, + frame: { + method, + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status }), + }, + }, + }); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "early-started-context", + entries: [ + ...codexReplayPreamble({ + nativeThreadId, + nativeTurnId: "unused-turn", + prompt: "unused-prompt", + }).slice(0, 5), + { + type: "expect_outbound", + label: "turn/start", + frame: { id: 3, method: "turn/start", params }, + }, + turnEvent("turn/started", "inProgress"), + { + type: "emit_inbound", + label: "turn/start", + frame: { + id: 3, + result: { turn: makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }) }, + }, + }, + { ...turnEvent("turn/completed", "completed"), afterMs: 1000 }, + ], + }), + ); + const mcpSessions = yield* McpProviderSessions.McpProviderSessions; + yield* mcpSessions.set({ + environmentId: EnvironmentId.make("test"), + threadId: harness.threadId, + providerSessionId: "early-started-context-session", + providerInstanceId: ProviderInstanceId.make("codex"), + endpoint: "http://127.0.0.1:43123/mcp", + authorizationHeader: "Bearer test", + browserToolsAvailable: true, + }); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("early-started-context"), + text: "hello", + }), + ); + yield* TestClock.adjust("1 second"); + yield* harness.firstTerminal; + const persisted = harness.events.findLast( + (event) => event.type === "provider_thread.updated", + )?.providerThread; + assert.isString(persisted?.nativeMetadata?.codexAdditionalContextDigest); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + it.effect("compacts Codex with the native RPC and completes the compaction turn", () => Effect.scoped( Effect.gen(function* () { From 32cb2ebc43e22aac1bb7ea727f229ff7ac697bbe Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 30 Oct 2026 06:52:41 +0000 Subject: [PATCH 08/12] fix(server): persist the Codex context digest before a fast turn completes Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 17 +++++--- .../Adapters/CodexAdapterV2.ts | 40 +++++++++++-------- 2 files changed, 36 insertions(+), 21 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 78a3a269b499..cdd3cc724f93 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2458,7 +2458,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); - it.effect("persists the T3 context digest when turn/started precedes its response", () => + it.effect("persists the T3 context digest when the turn completes before its response", () => Effect.scoped( Effect.gen(function* () { const nativeThreadId = "early-started-context-thread"; @@ -2470,7 +2470,10 @@ describe("CodexAdapterV2 post-settle continuation", () => { modelSelection: CODEX_TEST_MODEL_SELECTION, hasT3Mcp: true, }); - const turnEvent = (method: "turn/started" | "turn/completed", status: string) => ({ + const turnEvent = ( + method: "turn/started" | "turn/completed", + status: "inProgress" | "completed", + ) => ({ type: "emit_inbound" as const, label: method, frame: { @@ -2496,6 +2499,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { frame: { id: 3, method: "turn/start", params }, }, turnEvent("turn/started", "inProgress"), + turnEvent("turn/completed", "completed"), { type: "emit_inbound", label: "turn/start", @@ -2504,7 +2508,6 @@ describe("CodexAdapterV2 post-settle continuation", () => { result: { turn: makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }) }, }, }, - { ...turnEvent("turn/completed", "completed"), afterMs: 1000 }, ], }), ); @@ -2527,9 +2530,13 @@ describe("CodexAdapterV2 post-settle continuation", () => { text: "hello", }), ); - yield* TestClock.adjust("1 second"); yield* harness.firstTerminal; - const persisted = harness.events.findLast( + // Run ingestion stops at the terminal event, so the digest must be published before it. + const beforeTerminal = harness.events.slice( + 0, + harness.events.findIndex((event) => event.type === "turn.terminal"), + ); + const persisted = beforeTerminal.findLast( (event) => event.type === "provider_thread.updated", )?.providerThread; assert.isString(persisted?.nativeMetadata?.codexAdditionalContextDigest); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 7e7c83732082..295d61790220 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1678,6 +1678,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); // Absent: fall back to persisted metadata. Undefined: Codex holds no known context. const deliveredContextDigests = new Map(); + // Recorded once Codex accepts the turn, before its turn/started can lead to a terminal. + const pendingContextDigests = new Map(); // Codex drops client developer messages during compaction but only // resends additionalContext when it changes. const restoreAdditionalContext = Effect.fn("CodexAdapterV2.restoreAdditionalContext")( @@ -1932,6 +1934,13 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }); const invalidateAdditionalContext = (threadId: string) => recordAdditionalContextDigest(threadId, undefined); + const recordPendingContextDigest = (threadId: string) => + Effect.suspend(() => { + const digest = pendingContextDigests.get(threadId); + if (digest === undefined) return Effect.void; + pendingContextDigests.delete(threadId); + return recordAdditionalContextDigest(threadId, digest); + }); // Setting a goal passes through cleared and paused states; publish only the result. const goalUpdatesHeld = new Set(); const emitGoalUpdate = Effect.fnUntraced(function* (nativeThreadId: string) { @@ -4480,6 +4489,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( nativeTurnId: payload.turn.id, startedAt: codexTimestamp(payload.turn.startedAt), }); + yield* recordPendingContextDigest(payload.threadId); yield* Ref.update(pendingRootTurns, (current) => { const updated = new Map(current); updated.delete(payload.threadId); @@ -6588,6 +6598,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const deliveredDigest = deliveredContextDigests.has(threadId) ? deliveredContextDigests.get(threadId) : turnInput.providerThread.nativeMetadata?.codexAdditionalContextDigest; + if (contextDigest !== undefined && contextDigest !== deliveredDigest) { + pendingContextDigests.set(threadId, contextDigest); + } const started = yield* client.request( "turn/start", contextDigest !== undefined && contextDigest === deliveredDigest @@ -6595,18 +6608,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( : turnStartParams, ); // A turn/started notification may have registered and completed this turn already. - const registeredByNotification = - (yield* Ref.get(pendingRootTurns)).get(threadId) !== turnInput; - if (!registeredByNotification) { - rootProviderThreads.set( - threadId, - withCurrentContextDigest(threadId, turnInput.providerThread), - ); - } - if (contextDigest !== undefined && contextDigest !== deliveredDigest) { - yield* recordAdditionalContextDigest(threadId, contextDigest); - } - if (registeredByNotification) return; + if ((yield* Ref.get(pendingRootTurns)).get(threadId) !== turnInput) return; + rootProviderThreads.set( + threadId, + withCurrentContextDigest(threadId, turnInput.providerThread), + ); + yield* recordPendingContextDigest(threadId); const providerThread = rootProviderThreads.get(threadId) ?? turnInput.providerThread; yield* registerRootTurn({ turnInput: { ...turnInput, providerThread }, @@ -6618,14 +6625,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( (effect, turnInput) => effect.pipe( Effect.ensuring( - Effect.flatMap(getNativeThreadId(turnInput.providerThread), (threadId) => - Ref.update(pendingRootTurns, (current) => { + Effect.flatMap(getNativeThreadId(turnInput.providerThread), (threadId) => { + pendingContextDigests.delete(threadId); + return Ref.update(pendingRootTurns, (current) => { if (current.get(threadId) !== turnInput) return current; const updated = new Map(current); updated.delete(threadId); return updated; - }), - ).pipe(Effect.ignore), + }); + }).pipe(Effect.ignore), ), ), ); From eccc492a58a32a7be8569380a7b7798501e7536e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 30 Oct 2026 07:05:37 +0000 Subject: [PATCH 09/12] fix(server): republish a Codex context digest the stored thread lost Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 100 +++++++++++++++++- .../Adapters/CodexAdapterV2.ts | 27 +++-- 2 files changed, 112 insertions(+), 15 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index cdd3cc724f93..f76121754966 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2085,7 +2085,9 @@ describe("CodexAdapterV2 post-settle continuation", () => { yield* harness.runtime.startTurn( makeCodexTestTurnInput({ threadId: harness.threadId, - providerThread: harness.providerThread, + providerThread: + harness.events.findLast((event) => event.type === "provider_thread.updated") + ?.providerThread ?? harness.providerThread, now: yield* DateTime.now, attemptId: RunAttemptId.make("automatic-compact"), text: "work", @@ -2548,6 +2550,102 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("republishes a T3 context digest the stored provider thread lost", () => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "lost-digest-context-thread"; + const params = (text: string) => + CodexAdapterV2.buildCodexTurnStartParams({ + nativeThreadId, + codexInput: [{ type: "text", text }], + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + modelSelection: CODEX_TEST_MODEL_SELECTION, + hasT3Mcp: true, + }); + const completed = (id: number, nativeTurnId: string, turnParams: unknown) => + [ + { + type: "expect_outbound", + label: nativeTurnId, + frame: { id, method: "turn/start", params: turnParams }, + }, + { + type: "emit_inbound", + label: nativeTurnId, + frame: { + id, + result: { turn: makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }) }, + }, + }, + { + type: "emit_inbound", + label: `${nativeTurnId} completed`, + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + }, + }, + }, + ] satisfies Array; + const { additionalContext: _unchanged, ...secondWithoutContext } = yield* params("second"); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "lost-digest-context", + entries: [ + ...codexReplayPreamble({ + nativeThreadId, + nativeTurnId: "unused-turn", + prompt: "unused-prompt", + }).slice(0, 5), + ...completed(3, "turn-first", yield* params("first")), + ...completed(4, "turn-second", secondWithoutContext), + ], + }), + ); + const mcpSessions = yield* McpProviderSessions.McpProviderSessions; + yield* mcpSessions.set({ + environmentId: EnvironmentId.make("test"), + threadId: harness.threadId, + providerSessionId: "lost-digest-context-session", + providerInstanceId: ProviderInstanceId.make("codex"), + endpoint: "http://127.0.0.1:43123/mcp", + authorizationHeader: "Bearer test", + browserToolsAvailable: true, + }); + const turnInput = (text: string) => + Effect.map(DateTime.now, (now) => + makeCodexTestTurnInput({ + threadId: harness.threadId, + // An update published after a terminal is never stored, so the stored thread lacks it. + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make(`lost-digest-${text}`), + text, + }), + ); + yield* harness.runtime.startTurn(yield* turnInput("first")); + yield* harness.awaitTerminal(1); + const firstTerminal = harness.events.length; + yield* harness.runtime.startTurn(yield* turnInput("second")); + yield* harness.awaitTerminal(2); + const secondTurnEvents = harness.events.slice(firstTerminal); + const republished = secondTurnEvents + .slice( + 0, + secondTurnEvents.findIndex((event) => event.type === "turn.terminal"), + ) + .findLast((event) => event.type === "provider_thread.updated")?.providerThread; + assert.isString(republished?.nativeMetadata?.codexAdditionalContextDigest); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + it.effect("compacts Codex with the native RPC and completes the compaction turn", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 295d61790220..88c4acc873fb 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1679,7 +1679,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( // Absent: fall back to persisted metadata. Undefined: Codex holds no known context. const deliveredContextDigests = new Map(); // Recorded once Codex accepts the turn, before its turn/started can lead to a terminal. - const pendingContextDigests = new Map(); + const pendingContextDigests = new Map(); // Codex drops client developer messages during compaction but only // resends additionalContext when it changes. const restoreAdditionalContext = Effect.fn("CodexAdapterV2.restoreAdditionalContext")( @@ -1934,11 +1934,16 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }); const invalidateAdditionalContext = (threadId: string) => recordAdditionalContextDigest(threadId, undefined); - const recordPendingContextDigest = (threadId: string) => + // Compare against the stored thread: an update emitted after a terminal is never stored. + const recordPendingContextDigest = ( + threadId: string, + storedThread: OrchestrationV2ProviderThread, + ) => Effect.suspend(() => { + if (!pendingContextDigests.has(threadId)) return Effect.void; const digest = pendingContextDigests.get(threadId); - if (digest === undefined) return Effect.void; pendingContextDigests.delete(threadId); + rootProviderThreads.set(threadId, storedThread); return recordAdditionalContextDigest(threadId, digest); }); // Setting a goal passes through cleared and paused states; publish only the result. @@ -4484,12 +4489,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } const pendingRootTurn = (yield* Ref.get(pendingRootTurns)).get(payload.threadId); if (pendingRootTurn !== undefined) { + yield* recordPendingContextDigest(payload.threadId, pendingRootTurn.providerThread); yield* registerRootTurn({ turnInput: pendingRootTurn, nativeTurnId: payload.turn.id, startedAt: codexTimestamp(payload.turn.startedAt), }); - yield* recordPendingContextDigest(payload.threadId); yield* Ref.update(pendingRootTurns, (current) => { const updated = new Map(current); updated.delete(payload.threadId); @@ -6598,9 +6603,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const deliveredDigest = deliveredContextDigests.has(threadId) ? deliveredContextDigests.get(threadId) : turnInput.providerThread.nativeMetadata?.codexAdditionalContextDigest; - if (contextDigest !== undefined && contextDigest !== deliveredDigest) { - pendingContextDigests.set(threadId, contextDigest); - } + pendingContextDigests.set(threadId, contextDigest ?? deliveredDigest); const started = yield* client.request( "turn/start", contextDigest !== undefined && contextDigest === deliveredDigest @@ -6609,11 +6612,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); // A turn/started notification may have registered and completed this turn already. if ((yield* Ref.get(pendingRootTurns)).get(threadId) !== turnInput) return; - rootProviderThreads.set( - threadId, - withCurrentContextDigest(threadId, turnInput.providerThread), - ); - yield* recordPendingContextDigest(threadId); + yield* recordPendingContextDigest(threadId, turnInput.providerThread); const providerThread = rootProviderThreads.get(threadId) ?? turnInput.providerThread; yield* registerRootTurn({ turnInput: { ...turnInput, providerThread }, @@ -7715,12 +7714,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); turnTokenUsageByThread.delete(threadId); deliveredContextDigests.set(response.thread.id, undefined); - const { codexAdditionalContextDigest: _reverted, ...nativeMetadata } = + const { codexAdditionalContextDigest: revertedDigest, ...nativeMetadata } = threadInput.providerThread.nativeMetadata ?? {}; return { providerThread: { ...threadInput.providerThread, - ...(_reverted === undefined ? {} : { nativeMetadata }), + ...(revertedDigest === undefined ? {} : { nativeMetadata }), nativeThreadRef: { driver: CODEX_PROVIDER, nativeId: response.thread.id, From ccfc1d6c98ab4bcaf99d4e3843297f51e3567edb Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 30 Oct 2026 07:19:04 +0000 Subject: [PATCH 10/12] fix(server): keep sending T3 context to a live Codex app-server Omitting additionalContext empties Codex's in-memory context map, so a later change would repeat every entry. Omit only when history holds the context and this app-server's map does not. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 13 ++++--------- .../src/orchestration-v2/Adapters/CodexAdapterV2.ts | 13 ++++++++++--- 2 files changed, 14 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index f76121754966..2754504ee1e2 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1907,7 +1907,6 @@ describe("CodexAdapterV2 post-settle continuation", () => { modelSelection: CODEX_TEST_MODEL_SELECTION, hasT3Mcp: true, }); - const { additionalContext: _context, ...paramsWithoutContext } = params; assert.include( params.additionalContext?.t3_code_orchestration_01?.value ?? "", "delegate_task", @@ -1969,11 +1968,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { { type: "expect_outbound" as const, label: "next turn", - frame: { - id: 4, - method: "turn/start", - params: paramsWithoutContext, - }, + frame: { id: 4, method: "turn/start", params }, }, { type: "emit_inbound" as const, @@ -2550,7 +2545,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); - it.effect("republishes a T3 context digest the stored provider thread lost", () => + it.effect("keeps sending live Codex its T3 context and republishes a lost digest", () => Effect.scoped( Effect.gen(function* () { const nativeThreadId = "lost-digest-context-thread"; @@ -2589,7 +2584,6 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, }, ] satisfies Array; - const { additionalContext: _unchanged, ...secondWithoutContext } = yield* params("second"); const harness = yield* makeCodexReplayHarness( makeCodexReplayTranscript({ scenario: "lost-digest-context", @@ -2600,7 +2594,8 @@ describe("CodexAdapterV2 post-settle continuation", () => { prompt: "unused-prompt", }).slice(0, 5), ...completed(3, "turn-first", yield* params("first")), - ...completed(4, "turn-second", secondWithoutContext), + // Omitting context would empty Codex's map, so a later change would repeat every entry. + ...completed(4, "turn-second", yield* params("second")), ], }), ); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 88c4acc873fb..08d3b50c916e 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1678,6 +1678,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); // Absent: fall back to persisted metadata. Undefined: Codex holds no known context. const deliveredContextDigests = new Map(); + // Codex dedupes against its in-memory map, which a turn without additionalContext empties. + const contextStoreDigests = new Map(); // Recorded once Codex accepts the turn, before its turn/started can lead to a terminal. const pendingContextDigests = new Map(); // Codex drops client developer messages during compaction but only @@ -6603,13 +6605,18 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const deliveredDigest = deliveredContextDigests.has(threadId) ? deliveredContextDigests.get(threadId) : turnInput.providerThread.nativeMetadata?.codexAdditionalContextDigest; + // Omit only when history holds the context but this app-server's map does not. + const omitContext = + contextDigest !== undefined && + contextDigest === deliveredDigest && + contextStoreDigests.get(threadId) !== contextDigest; pendingContextDigests.set(threadId, contextDigest ?? deliveredDigest); const started = yield* client.request( "turn/start", - contextDigest !== undefined && contextDigest === deliveredDigest - ? paramsWithoutContext - : turnStartParams, + omitContext ? paramsWithoutContext : turnStartParams, ); + if (omitContext || contextDigest === undefined) contextStoreDigests.delete(threadId); + else contextStoreDigests.set(threadId, contextDigest); // A turn/started notification may have registered and completed this turn already. if ((yield* Ref.get(pendingRootTurns)).get(threadId) !== turnInput) return; yield* recordPendingContextDigest(threadId, turnInput.providerThread); From 5d0755517da5e2d5f1f84152a7bfb3417a5dbd57 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 30 Oct 2026 07:38:31 +0000 Subject: [PATCH 11/12] fix(server): omit unchanged Codex context after a thread resume A resumed thread may have been unloaded, which empties Codex's context map. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 34 ++++++++++++++++++- .../Adapters/CodexAdapterV2.ts | 2 ++ 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 2754504ee1e2..15f92b533da5 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2545,7 +2545,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); - it.effect("keeps sending live Codex its T3 context and republishes a lost digest", () => + it.effect("sends live Codex its T3 context until a resume and republishes a lost digest", () => Effect.scoped( Effect.gen(function* () { const nativeThreadId = "lost-digest-context-thread"; @@ -2584,6 +2584,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, }, ] satisfies Array; + const { additionalContext: _unchanged, ...thirdWithoutContext } = yield* params("third"); const harness = yield* makeCodexReplayHarness( makeCodexReplayTranscript({ scenario: "lost-digest-context", @@ -2596,6 +2597,25 @@ describe("CodexAdapterV2 post-settle continuation", () => { ...completed(3, "turn-first", yield* params("first")), // Omitting context would empty Codex's map, so a later change would repeat every entry. ...completed(4, "turn-second", yield* params("second")), + { + type: "expect_outbound", + label: "thread/resume", + frame: { + id: 5, + method: "thread/resume", + params: { + threadId: nativeThreadId, + excludeTurns: true, + config: CodexAdapterV2.CODEX_THREAD_CONFIG, + }, + }, + }, + { + type: "emit_inbound", + label: "thread/resume", + frame: { id: 5, result: { thread: { id: nativeThreadId, updatedAt: 1782622450 } } }, + }, + ...completed(6, "turn-third", thirdWithoutContext), ], }), ); @@ -2633,6 +2653,18 @@ describe("CodexAdapterV2 post-settle continuation", () => { ) .findLast((event) => event.type === "provider_thread.updated")?.providerThread; assert.isString(republished?.nativeMetadata?.codexAdditionalContextDigest); + if (!republished) return yield* Effect.die("Missing republished provider thread"); + // A resumed thread may have been unloaded, which empties Codex's context map. + const resumed = yield* harness.runtime.resumeThread({ + providerThread: republished, + modelSelection: CODEX_TEST_MODEL_SELECTION, + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + }); + yield* harness.runtime.startTurn({ + ...(yield* turnInput("third")), + providerThread: resumed, + }); + yield* harness.awaitTerminal(3); }).pipe( Effect.provide( Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 08d3b50c916e..6d5bc82187d4 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -6954,6 +6954,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ), Effect.flatMap(decodeCodexResumeMetadata), ); + // An unloaded thread reloads with an empty context map. + contextStoreDigests.delete(response.thread.id); return { ...threadInput.providerThread, providerSessionId: input.providerSessionId, From a1744f2c398993d45d4107ee26320d69b2741998 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 01:10:33 +0000 Subject: [PATCH 12/12] fix(server): track Codex's context map as soon as turn/start is accepted Main's earlier-run guard can return before the bookkeeping ran, leaving the context-store digest stale for a request Codex already accepted. AI-assisted; verified with the tests and tools listed above. --- apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 079afeb2c94e..a868a8bd97a7 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -7102,6 +7102,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( "turn/start", omitContext ? paramsWithoutContext : turnStartParams, ); + if (omitContext || contextDigest === undefined) contextStoreDigests.delete(threadId); + else contextStoreDigests.set(threadId, contextDigest); const previousTurn = previousTurns.get(started.turn.id) ?? (yield* Ref.get(activeTurns)).get(started.turn.id); @@ -7115,8 +7117,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( "Codex returned a turn belonging to an earlier run. Stop the existing turn before retrying this message.", }); } - if (omitContext || contextDigest === undefined) contextStoreDigests.delete(threadId); - else contextStoreDigests.set(threadId, contextDigest); // A turn/started notification may have registered and completed this turn already. if ((yield* Ref.get(pendingRootTurns)).get(threadId) !== turnInput) return; yield* recordPendingContextDigest(threadId, turnInput.providerThread);