diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 14bb6a03d8111163e2e963261842c61bdd50df72..d4f9f70993e277f8c6dd6f6982e7cdabc3f1f1e9 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -3679,6 +3679,94 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("streams a proposed plan a few times a second instead of once per delta", () => + Effect.scoped( + Effect.gen(function* () { + const scenario = "codex-plan-delta-coalescing"; + const nativeThreadId = `native-${scenario}-thread`; + const nativeTurnId = `native-${scenario}-turn`; + const prompt = "Plan the change."; + // 100 deltas, 20 ms apart: a two second stream. + const deltas = Array.from({ length: 100 }, (_, index) => `step ${index}\n`); + const transcript = makeCodexReplayTranscript({ + scenario, + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }), + ...deltas.map((delta, index): CodexReplay.CodexAppServerReplayEntry => ({ + type: "emit_inbound", + label: `item/plan/delta/${index}`, + afterMs: 20, + frame: { + method: "item/plan/delta", + params: { threadId: nativeThreadId, turnId: nativeTurnId, itemId: "plan", delta }, + }, + })), + { + type: "emit_inbound", + label: "item/completed/plan", + frame: { + method: "item/completed", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + item: { type: "plan", id: "plan", text: "" }, + }, + }, + }, + { + type: "emit_inbound", + label: "turn/completed", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + }, + }, + }, + ], + }); + const harness = yield* makeCodexReplayHarness(transcript); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make(`attempt-${scenario}`), + text: prompt, + }), + ); + for (let step = 0; step < 200 && harness.terminalEvents().length === 0; step += 1) { + yield* Effect.yieldNow; + yield* TestClock.adjust("20 millis"); + } + yield* harness.firstTerminal; + + const plans = harness.events.flatMap((event) => + event.type === "plan.updated" && event.plan.kind === "proposed_plan" ? [event.plan] : [], + ); + const planItems = harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "proposed_plan" + ? [event.turnItem] + : [], + ); + // One update per 400 ms of the stream, plus the completion. + assert.isAtLeast(plans.length, 2); + assert.isAtMost(plans.length, 7); + assert.equal(planItems.length, plans.length); + assert.equal(plans.at(-1)?.markdown, deltas.join("")); + assert.deepEqual( + planItems.map((item) => item.streaming), + [...Array.from({ length: planItems.length - 1 }, () => true), false], + ); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + const finalAnswerTranscript = ( scenario: string, answers: ReadonlyArray<{ diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 5a51136396ad8c4dc5fa333551a5a730733ac25a..0e18e2857d14878e7d9ddfb1436984c47d445d88 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -203,6 +203,9 @@ export function codexProviderTurnTokenUsage( } const DEFAULT_CODEX_SETTINGS = Schema.decodeSync(CodexSettings)({}); const CODEX_ASSISTANT_DELTA_FLUSH_INTERVAL_MS = 50; +// Each plan flush stores the whole plan three times (node, plan, turn item), so +// flush at the cadence clients already see for assistant text. +const CODEX_PLAN_DELTA_FLUSH_INTERVAL_MS = 400; const CodexBackgroundTerminalTerminateResponse = Schema.Struct({ terminated: Schema.Boolean, }); @@ -1864,8 +1867,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const providerRetries = yield* Ref.make( new Map(), ); - // Streamed plan text per plan item, dropped when the item completes. - const planDeltas = new Map(); const planIds = yield* Ref.make(new Map()); const pendingRuntimeRequests = yield* Ref.make( new Map(), @@ -3948,6 +3949,43 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( return { node, plan, turnItem }; }); + // Streamed plan text, flushed at most once per interval. A turn that ends + // early leaves its partial plan streaming, as an unfinished plan item. + const planDeltas = yield* makeProviderTextDeltaCoalescer({ + flushIntervalMs: CODEX_PLAN_DELTA_FLUSH_INTERVAL_MS, + emit: (update) => + Effect.gen(function* () { + const resolved = yield* resolveItemEventContext(update.turnId); + if (resolved === undefined) return; + const interrupted = (yield* Ref.get(terminalizedNonCompletedNativeTurns)).has( + update.turnId, + ); + const artifacts = yield* buildProposedPlanArtifacts({ + context: resolved.context, + nativeItemId: update.itemId, + // A finished proposal stays active until Implement consumes it. + status: "active", + markdown: update.text, + completed: update.completed && !interrupted, + }); + yield* emitProviderEvent({ + type: "node.updated", + driver: CODEX_PROVIDER, + node: artifacts.node, + }); + yield* emitProviderEvent({ + type: "plan.updated", + driver: CODEX_PROVIDER, + plan: artifacts.plan, + }); + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CODEX_PROVIDER, + turnItem: artifacts.turnItem, + }); + }).pipe(Effect.orDie), + }); + const buildTodoListArtifacts = (input: { readonly context: ActiveCodexTurnContext; readonly nativeItemId: string; @@ -4245,28 +4283,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( return; } yield* completeProviderRetry(context, yield* DateTime.now); - const markdown = `${planDeltas.get(payload.itemId) ?? ""}${payload.delta}`; - planDeltas.set(payload.itemId, markdown); - const artifacts = yield* buildProposedPlanArtifacts({ - context, - nativeItemId: payload.itemId, - status: "active", - markdown, - }); - yield* emitProviderEvent({ - type: "node.updated", - driver: CODEX_PROVIDER, - node: artifacts.node, - }); - yield* emitProviderEvent({ - type: "plan.updated", - driver: CODEX_PROVIDER, - plan: artifacts.plan, - }); - yield* emitProviderEvent({ - type: "turn_item.updated", - driver: CODEX_PROVIDER, - turnItem: artifacts.turnItem, + yield* planDeltas.append({ + turnId: payload.turnId, + itemId: payload.itemId, + delta: payload.delta, }); }).pipe(Effect.orDie), ); @@ -4916,33 +4936,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } if (payload.item.type === "plan") { - const markdown = - payload.item.text.length > 0 - ? payload.item.text - : (planDeltas.get(payload.item.id) ?? ""); - planDeltas.delete(payload.item.id); - // A finished proposal stays active until Implement consumes it. - const artifacts = yield* buildProposedPlanArtifacts({ - context, - nativeItemId: payload.item.id, - status: "active", - markdown, - completed: true, - }); - yield* emitProviderEvent({ - type: "node.updated", - driver: CODEX_PROVIDER, - node: artifacts.node, - }); - yield* emitProviderEvent({ - type: "plan.updated", - driver: CODEX_PROVIDER, - plan: artifacts.plan, - }); - yield* emitProviderEvent({ - type: "turn_item.updated", - driver: CODEX_PROVIDER, - turnItem: artifacts.turnItem, + yield* planDeltas.complete({ + turnId: payload.turnId, + itemId: payload.item.id, + ...(payload.item.text.length > 0 ? { finalText: payload.item.text } : {}), }); return; } @@ -5780,6 +5777,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } yield* agentMessageDeltas.flushTurn(input.nativeTurnId); yield* reasoningDeltas.flushTurn(input.nativeTurnId); + yield* planDeltas.flushTurn(input.nativeTurnId); for (const [key, part] of reasoningParts) { if (part.turnId === input.nativeTurnId) reasoningParts.delete(key); }