Skip to content

fix(codex): settle unfinished web searches when turns end #32177

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Sidreyas wants to merge 1 commit into
base: main
Choose a base branch
from
+83 −24
Open

fix(codex): settle unfinished web searches when turns end #32177

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
64 changes: 48 additions & 16 deletions apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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." }),
...(
Expand All @@ -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,
},
},
Expand Down Expand Up @@ -2452,13 +2464,13 @@ describe("CodexAdapterV2 post-settle continuation", () => {
]),
],
});
const compactionStarted = yield* Deferred.make<ProviderTurnId>();
const itemStarted = yield* Deferred.make<ProviderTurnId>();
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);
Expand All @@ -2467,20 +2479,21 @@ 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);
assert.equal(interrupt._tag, "Failure");
}
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]
: [],
);
Expand All @@ -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),
Expand All @@ -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);
}
Expand Down
43 changes: 35 additions & 8 deletions apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1202,6 +1202,7 @@ interface ActiveCodexTurnContext {
readonly subagent: CodexSubagentThreadContext | null;
readonly startedAt: DateTime.Utc;
readonly runningCompactionItemIds: Set<string>;
readonly runningWebSearchItems: Map<string, CodexWebSearchItem>;
// Modern approval requests omit paths; keep the kinds without retaining patch bodies.
readonly fileChangeKindsByItem: Map<string, CodexFileChangeKinds>;
readonly itemPositions: Map<
Expand Down Expand Up @@ -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 =
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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",
);
Expand Down Expand Up @@ -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,
Expand Down
Loading