authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-14 01:36:34-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-14 01:36:34-07:00
logfd9cdaed17f615e364ee113829ec39b2bd2ed6fa
treef67ae01bdd3d202305f6840be3ccdc5f3d9b3d93
parent614b432a79b0acae38c27c00b1e4eece223ecfe4
signature Signed by PGP key B5690EEEBB952194

perf(server): Codex plan streaming no longer commits every token (#19317)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

2 files changed, 137 insertions(+), 51 deletions(-)

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts+88
......@@ -3679,6 +3679,94 @@ describe("CodexAdapterV2 post-settle continuation", () => {
36793679 ),
36803680 );
36813681
3682 it.effect("streams a proposed plan a few times a second instead of once per delta", () =>
3683 Effect.scoped(
3684 Effect.gen(function* () {
3685 const scenario = "codex-plan-delta-coalescing";
3686 const nativeThreadId = `native-${scenario}-thread`;
3687 const nativeTurnId = `native-${scenario}-turn`;
3688 const prompt = "Plan the change.";
3689 // 100 deltas, 20 ms apart: a two second stream.
3690 const deltas = Array.from({ length: 100 }, (_, index) => `step ${index}\n`);
3691 const transcript = makeCodexReplayTranscript({
3692 scenario,
3693 entries: [
3694 ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }),
3695 ...deltas.map((delta, index): CodexReplay.CodexAppServerReplayEntry => ({
3696 type: "emit_inbound",
3697 label: `item/plan/delta/${index}`,
3698 afterMs: 20,
3699 frame: {
3700 method: "item/plan/delta",
3701 params: { threadId: nativeThreadId, turnId: nativeTurnId, itemId: "plan", delta },
3702 },
3703 })),
3704 {
3705 type: "emit_inbound",
3706 label: "item/completed/plan",
3707 frame: {
3708 method: "item/completed",
3709 params: {
3710 threadId: nativeThreadId,
3711 turnId: nativeTurnId,
3712 item: { type: "plan", id: "plan", text: "" },
3713 },
3714 },
3715 },
3716 {
3717 type: "emit_inbound",
3718 label: "turn/completed",
3719 frame: {
3720 method: "turn/completed",
3721 params: {
3722 threadId: nativeThreadId,
3723 turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }),
3724 },
3725 },
3726 },
3727 ],
3728 });
3729 const harness = yield* makeCodexReplayHarness(transcript);
3730 yield* harness.runtime.startTurn(
3731 makeCodexTestTurnInput({
3732 threadId: harness.threadId,
3733 providerThread: harness.providerThread,
3734 now: yield* DateTime.now,
3735 attemptId: RunAttemptId.make(`attempt-${scenario}`),
3736 text: prompt,
3737 }),
3738 );
3739 for (let step = 0; step < 200 && harness.terminalEvents().length === 0; step += 1) {
3740 yield* Effect.yieldNow;
3741 yield* TestClock.adjust("20 millis");
3742 }
3743 yield* harness.firstTerminal;
3744
3745 const plans = harness.events.flatMap((event) =>
3746 event.type === "plan.updated" && event.plan.kind === "proposed_plan" ? [event.plan] : [],
3747 );
3748 const planItems = harness.events.flatMap((event) =>
3749 event.type === "turn_item.updated" && event.turnItem.type === "proposed_plan"
3750 ? [event.turnItem]
3751 : [],
3752 );
3753 // One update per 400 ms of the stream, plus the completion.
3754 assert.isAtLeast(plans.length, 2);
3755 assert.isAtMost(plans.length, 7);
3756 assert.equal(planItems.length, plans.length);
3757 assert.equal(plans.at(-1)?.markdown, deltas.join(""));
3758 assert.deepEqual(
3759 planItems.map((item) => item.streaming),
3760 [...Array.from({ length: planItems.length - 1 }, () => true), false],
3761 );
3762 }).pipe(
3763 Effect.provide(
3764 Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
3765 ),
3766 ),
3767 ),
3768 );
3769
36823770 const finalAnswerTranscript = (
36833771 scenario: string,
36843772 answers: ReadonlyArray<{
apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts+49-51
......@@ -203,6 +203,9 @@ export function codexProviderTurnTokenUsage(
203203}
204204const DEFAULT_CODEX_SETTINGS = Schema.decodeSync(CodexSettings)({});
205205const CODEX_ASSISTANT_DELTA_FLUSH_INTERVAL_MS = 50;
206// Each plan flush stores the whole plan three times (node, plan, turn item), so
207// flush at the cadence clients already see for assistant text.
208const CODEX_PLAN_DELTA_FLUSH_INTERVAL_MS = 400;
206209const CodexBackgroundTerminalTerminateResponse = Schema.Struct({
207210 terminated: Schema.Boolean,
208211});
......@@ -1864,8 +1867,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
18641867 const providerRetries = yield* Ref.make(
18651868 new Map<ProviderTurnId, ActiveCodexProviderRetry>(),
18661869 );
1867 // Streamed plan text per plan item, dropped when the item completes.
1868 const planDeltas = new Map<string, string>();
18691870 const planIds = yield* Ref.make(new Map<string, OrchestrationV2PlanArtifact["id"]>());
18701871 const pendingRuntimeRequests = yield* Ref.make(
18711872 new Map<string, PendingCodexRuntimeRequest>(),
......@@ -3948,6 +3949,43 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
39483949 return { node, plan, turnItem };
39493950 });
39503951
3952 // Streamed plan text, flushed at most once per interval. A turn that ends
3953 // early leaves its partial plan streaming, as an unfinished plan item.
3954 const planDeltas = yield* makeProviderTextDeltaCoalescer({
3955 flushIntervalMs: CODEX_PLAN_DELTA_FLUSH_INTERVAL_MS,
3956 emit: (update) =>
3957 Effect.gen(function* () {
3958 const resolved = yield* resolveItemEventContext(update.turnId);
3959 if (resolved === undefined) return;
3960 const interrupted = (yield* Ref.get(terminalizedNonCompletedNativeTurns)).has(
3961 update.turnId,
3962 );
3963 const artifacts = yield* buildProposedPlanArtifacts({
3964 context: resolved.context,
3965 nativeItemId: update.itemId,
3966 // A finished proposal stays active until Implement consumes it.
3967 status: "active",
3968 markdown: update.text,
3969 completed: update.completed && !interrupted,
3970 });
3971 yield* emitProviderEvent({
3972 type: "node.updated",
3973 driver: CODEX_PROVIDER,
3974 node: artifacts.node,
3975 });
3976 yield* emitProviderEvent({
3977 type: "plan.updated",
3978 driver: CODEX_PROVIDER,
3979 plan: artifacts.plan,
3980 });
3981 yield* emitProviderEvent({
3982 type: "turn_item.updated",
3983 driver: CODEX_PROVIDER,
3984 turnItem: artifacts.turnItem,
3985 });
3986 }).pipe(Effect.orDie),
3987 });
3988
39513989 const buildTodoListArtifacts = (input: {
39523990 readonly context: ActiveCodexTurnContext;
39533991 readonly nativeItemId: string;
......@@ -4245,28 +4283,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
42454283 return;
42464284 }
42474285 yield* completeProviderRetry(context, yield* DateTime.now);
4248 const markdown = `${planDeltas.get(payload.itemId) ?? ""}${payload.delta}`;
4249 planDeltas.set(payload.itemId, markdown);
4250 const artifacts = yield* buildProposedPlanArtifacts({
4251 context,
4252 nativeItemId: payload.itemId,
4253 status: "active",
4254 markdown,
4255 });
4256 yield* emitProviderEvent({
4257 type: "node.updated",
4258 driver: CODEX_PROVIDER,
4259 node: artifacts.node,
4260 });
4261 yield* emitProviderEvent({
4262 type: "plan.updated",
4263 driver: CODEX_PROVIDER,
4264 plan: artifacts.plan,
4265 });
4266 yield* emitProviderEvent({
4267 type: "turn_item.updated",
4268 driver: CODEX_PROVIDER,
4269 turnItem: artifacts.turnItem,
4286 yield* planDeltas.append({
4287 turnId: payload.turnId,
4288 itemId: payload.itemId,
4289 delta: payload.delta,
42704290 });
42714291 }).pipe(Effect.orDie),
42724292 );
......@@ -4916,33 +4936,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
49164936 }
49174937
49184938 if (payload.item.type === "plan") {
4919 const markdown =
4920 payload.item.text.length > 0
4921 ? payload.item.text
4922 : (planDeltas.get(payload.item.id) ?? "");
4923 planDeltas.delete(payload.item.id);
4924 // A finished proposal stays active until Implement consumes it.
4925 const artifacts = yield* buildProposedPlanArtifacts({
4926 context,
4927 nativeItemId: payload.item.id,
4928 status: "active",
4929 markdown,
4930 completed: true,
4931 });
4932 yield* emitProviderEvent({
4933 type: "node.updated",
4934 driver: CODEX_PROVIDER,
4935 node: artifacts.node,
4936 });
4937 yield* emitProviderEvent({
4938 type: "plan.updated",
4939 driver: CODEX_PROVIDER,
4940 plan: artifacts.plan,
4941 });
4942 yield* emitProviderEvent({
4943 type: "turn_item.updated",
4944 driver: CODEX_PROVIDER,
4945 turnItem: artifacts.turnItem,
4939 yield* planDeltas.complete({
4940 turnId: payload.turnId,
4941 itemId: payload.item.id,
4942 ...(payload.item.text.length > 0 ? { finalText: payload.item.text } : {}),
49464943 });
49474944 return;
49484945 }
......@@ -5780,6 +5777,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
57805777 }
57815778 yield* agentMessageDeltas.flushTurn(input.nativeTurnId);
57825779 yield* reasoningDeltas.flushTurn(input.nativeTurnId);
5780 yield* planDeltas.flushTurn(input.nativeTurnId);
57835781 for (const [key, part] of reasoningParts) {
57845782 if (part.turnId === input.nativeTurnId) reasoningParts.delete(key);
57855783 }