diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 84c78b91453c2..021907b952895 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2382,20 +2382,28 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); - it.effect.each(["failed", "interrupted", "Stop"] as const)( - "settles unfinished compaction on %s without changing completed compaction", - (outcome) => + it.effect.each([ + { itemType: "contextCompaction", outcome: "failed" }, + { itemType: "contextCompaction", outcome: "interrupted" }, + { itemType: "contextCompaction", outcome: "Stop" }, + { itemType: "webSearch", outcome: "completed" }, + { itemType: "webSearch", outcome: "failed" }, + { itemType: "webSearch", outcome: "interrupted" }, + { itemType: "webSearch", outcome: "Stop" }, + ] as const)( + "settles unfinished $itemType on $outcome without changing completed items", + ({ itemType, outcome }) => Effect.scoped( Effect.gen(function* () { - const nativeThreadId = "compact-thread"; - const nativeTurnId = "compact-turn"; + const nativeThreadId = "unfinished-item-thread"; + const nativeTurnId = "unfinished-item-turn"; const error = { - message: "Remote compaction stream disconnected", + message: "Remote item stream disconnected", codexErrorInfo: "other", additionalDetails: null, }; const transcript = makeCodexReplayTranscript({ - scenario: `automatic-compaction-${outcome}`, + scenario: `unfinished-${itemType}-${outcome}`, entries: [ ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "Continue." }), ...( @@ -2412,7 +2420,11 @@ describe("CodexAdapterV2 post-settle continuation", () => { params: { threadId: nativeThreadId, turnId: nativeTurnId, - item: { type: "contextCompaction", id }, + item: { + type: itemType, + id, + ...(itemType === "webSearch" ? { query: "T3 Code", action: null } : {}), + }, ...timestamps, }, }, @@ -2452,13 +2464,13 @@ describe("CodexAdapterV2 post-settle continuation", () => { ]), ], }); - const compactionStarted = yield* Deferred.make(); + const itemStarted = yield* Deferred.make(); const harness = yield* makeCodexReplayHarness(transcript, (event) => event.type === "turn_item.updated" && event.turnItem.nativeItemRef?.nativeId === "unfinished-item" && event.turnItem.status === "running" && event.turnItem.providerTurnId !== null - ? Deferred.succeed(compactionStarted, event.turnItem.providerTurnId) + ? Deferred.succeed(itemStarted, event.turnItem.providerTurnId) : Effect.void, ); yield* TestClock.setTime(1782622445000); @@ -2467,12 +2479,12 @@ describe("CodexAdapterV2 post-settle continuation", () => { threadId: harness.threadId, providerThread: harness.providerThread, now: yield* DateTime.now, - attemptId: RunAttemptId.make(`compact-${outcome}-attempt`), + attemptId: RunAttemptId.make(`unfinished-${itemType}-${outcome}-attempt`), text: "Continue.", }), ); if (outcome === "Stop") { - const providerTurnId = yield* Deferred.await(compactionStarted); + const providerTurnId = yield* Deferred.await(itemStarted); const interrupt = yield* harness.runtime .interruptTurn({ providerThread: harness.providerThread, providerTurnId }) .pipe(Effect.exit); @@ -2480,7 +2492,8 @@ describe("CodexAdapterV2 post-settle continuation", () => { } yield* harness.firstTerminal; const items = harness.events.flatMap((event) => - event.type === "turn_item.updated" && event.turnItem.type === "compaction" + event.type === "turn_item.updated" && + event.turnItem.type === (itemType === "webSearch" ? "web_search" : "compaction") ? [event.turnItem] : [], ); @@ -2490,13 +2503,26 @@ describe("CodexAdapterV2 post-settle continuation", () => { const unfinished = items.filter( (item) => item.nativeItemRef?.nativeId === "unfinished-item", ); - const status = outcome === "failed" ? "failed" : "interrupted"; + const status = + outcome === "completed" ? "cancelled" : outcome === "failed" ? "failed" : "interrupted"; assert.deepEqual( unfinished.map((item) => item.status), ["running", status], ); assert.equal(unfinished[0]?.id, unfinished[1]?.id); assert.deepEqual(unfinished[1]?.startedAt, DateTime.makeUnsafe(1782622443000)); + if (unfinished[1]?.type === "web_search") { + assert.deepEqual(unfinished[1].patterns, ["T3 Code"]); + assert.deepEqual( + harness.events.flatMap((event) => + event.type === "node.updated" && + event.node.nativeItemRef?.nativeId === "unfinished-item" + ? [event.node.status] + : [], + ), + ["running", status], + ); + } assert.deepEqual( unfinished[1]?.completedAt, DateTime.makeUnsafe(outcome === "Stop" ? 1782622445000 : 1782622450000), @@ -2506,8 +2532,14 @@ describe("CodexAdapterV2 post-settle continuation", () => { ["running", "completed"], ); assert.deepEqual(completed[1]?.startedAt, DateTime.makeUnsafe(1782622441000)); - assert.deepEqual(completed[1]?.completedAt, DateTime.makeUnsafe(1782622442000)); - assert.equal(harness.terminalEvents()[0]?.status, status); + assert.deepEqual( + completed[1]?.completedAt, + DateTime.makeUnsafe(itemType === "webSearch" ? 1782622445000 : 1782622442000), + ); + assert.equal( + harness.terminalEvents()[0]?.status, + outcome === "Stop" ? "interrupted" : outcome, + ); if (outcome === "failed") { assert.equal(harness.terminalEvents()[0]?.failure?.message, error.message); } diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 07c4e4e72eca7..6e811c4f4a31d 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1202,6 +1202,7 @@ interface ActiveCodexTurnContext { readonly subagent: CodexSubagentThreadContext | null; readonly startedAt: DateTime.Utc; readonly runningCompactionItemIds: Set; + readonly runningWebSearchItems: Map; // Modern approval requests omit paths; keep the kinds without retaining patch bodies. readonly fileChangeKindsByItem: Map; readonly itemPositions: Map< @@ -2129,6 +2130,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( itemPositions: new Map(), fileChangeKindsByItem: new Map(), runningCompactionItemIds: new Set(), + runningWebSearchItems: new Map(), }; const nativeThreadId = input.turnInput.providerThread.nativeThreadRef?.nativeId; const continuation = @@ -2735,6 +2737,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( itemPositions: new Map(), fileChangeKindsByItem: new Map(), runningCompactionItemIds: new Set(), + runningWebSearchItems: new Map(), }; beginTurnTokenUsage(activeContext); endSessionApprovalsOnModeChange(activeContext); @@ -3772,10 +3775,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( function* (input: { readonly context: ActiveCodexTurnContext; readonly item: CodexWebSearchItem; - readonly completed: boolean; + readonly status: "running" | "completed" | "cancelled" | "failed" | "interrupted"; + readonly updatedAt?: DateTime.Utc; }) { - const updatedAt = yield* DateTime.now; - const completedAt = input.completed ? updatedAt : null; + const updatedAt = input.updatedAt ?? (yield* DateTime.now); + const completedAt = input.status === "running" ? null : updatedAt; const nodeId = idAllocator.derive.nodeFromProviderItem({ driver: CODEX_PROVIDER, nativeItemId: input.item.id, @@ -3786,7 +3790,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }); const { ordinal, startedAt } = yield* resolveItemPosition(input.context, input.item.id); const patterns = webSearchPatterns(input.item); - const status = input.completed ? "completed" : "running"; + const { status } = input; const node: OrchestrationV2ExecutionNode = { id: nodeId, threadId: input.context.projectionThreadId, @@ -4971,10 +4975,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( return; } + context.runningWebSearchItems.set(payload.item.id, payload.item); const artifacts = yield* buildWebSearchArtifacts({ context, item: payload.item, - completed: false, + status: "running", }); yield* emitProviderEvent({ type: "node.updated", @@ -5151,10 +5156,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } if (payload.item.type === "webSearch") { + context.runningWebSearchItems.delete(payload.item.id); const artifacts = yield* buildWebSearchArtifacts({ context, item: payload.item, - completed: true, + status: "completed", }); yield* emitProviderEvent({ type: "node.updated", @@ -6338,14 +6344,34 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); yield* terminalizeMcpAppCaptures(input.context, input.nativeTurnId); } - const dynamicToolStatus: "cancelled" | "interrupted" | "failed" = + const unfinishedToolStatus: "cancelled" | "interrupted" | "failed" = input.status === "interrupted" || input.status === "failed" ? input.status : "cancelled"; + // A completed turn does not establish whether an unfinished search succeeded. + for (const item of input.context.runningWebSearchItems.values()) { + const artifacts = yield* buildWebSearchArtifacts({ + context: input.context, + item, + status: unfinishedToolStatus, + updatedAt: input.completedAt, + }); + yield* emitProviderEvent({ + type: "node.updated", + driver: CODEX_PROVIDER, + node: artifacts.node, + }); + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CODEX_PROVIDER, + turnItem: artifacts.turnItem, + }); + } + input.context.runningWebSearchItems.clear(); yield* terminalizeRunningDynamicTools( input.context, input.nativeTurnId, - dynamicToolStatus, + unfinishedToolStatus, input.completedAt, input.status === "interrupted" || input.status === "failed", ); @@ -6621,6 +6647,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( itemPositions: new Map(), fileChangeKindsByItem: new Map(), runningCompactionItemIds: new Set(), + runningWebSearchItems: new Map(), }; const providerTurn = { id: context.providerTurnId,