From b22686af4032bf1e886cfba0dd0477428c8360fc Mon Sep 17 00:00:00 2001 From: fixfon <12512757+fixfon@users.noreply.github.com> Date: Tue, 13 Oct 2026 13:39:33 +0300 Subject: [PATCH] fix(server): held assistant drafts keep their position when flushed at turn end --- .../RunExecutionService.test.ts | 20 +++ .../orchestration-v2/RunExecutionService.ts | 19 +++ .../SteeringCompletion.integration.test.ts | 118 ++++++++++++++++++ .../src/orchestration-v2/runtimeLayer.ts | 1 + .../testkit/ProviderReplayHarness.ts | 1 + .../fixtures/opencode2_tool_call/output.ts | 5 +- 6 files changed, 162 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index d877922a644a..4dd82ae41b22 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -49,14 +49,20 @@ import * as ProviderEventIngestor from "./ProviderEventIngestor.ts"; import * as McpAppModelContext from "../mcpApps/McpAppModelContext.ts"; import * as RunExecutionService from "./RunExecutionService.ts"; import * as RunFinalizationService from "./RunFinalizationService.ts"; +import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; const driver = ProviderDriverKind.make("codex"); +const layerTurnItemPositionsTest = Layer.mock(TurnItemPositionStore.TurnItemPositionStoreV2)({ + allocate: () => Effect.succeed(1), +}); + const layerRunExecutionTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({}), IdAllocator.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ @@ -755,6 +761,7 @@ it.effect("passes the thread's MCP app context to the provider under a safe key" Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void, }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({}), IdAllocator.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ @@ -782,6 +789,7 @@ it.effect("fails the run when its ownership check cannot be read before calling Layer.mergeAll( McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({ writeIfRunCurrent: (input) => Ref.update(writes, (current) => [...current, ...input.events]).pipe( @@ -1065,6 +1073,7 @@ it.effect("starts the provider when checkpoint baseline capture fails", () => }), ), }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({ writeWithEffects: (input) => Ref.update(writes, (current) => [ @@ -1177,6 +1186,7 @@ it.effect.each(["failure", "interruption", "stale-attempt", "start-guard"] as co captureBaseline: () => scenario === "start-guard" ? Effect.void : Effect.die("not reached"), }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({ writeIfRunCurrent: (input) => Effect.gen(function* () { @@ -1346,6 +1356,7 @@ it.effect("ingests new member threads of an owned child after the root turn term Layer.mergeAll( McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({ write: () => Effect.succeed([]), writeWithEffects: (input) => @@ -1575,6 +1586,7 @@ it.effect("ends a nested native subagent and its thread when the root run is int Layer.mergeAll( McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({ write: () => Effect.succeed([]), writeWithEffects: (input) => @@ -1932,6 +1944,7 @@ it.effect( Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void, }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({ write: () => Effect.succeed([]), writeWithEffects: (input) => @@ -2148,6 +2161,7 @@ it.effect("drops late root provider-thread writes from a superseded attempt", () Layer.mergeAll( McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({ write: () => Effect.succeed([]), writeWithEffects: (input) => @@ -2355,6 +2369,7 @@ it.effect( Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void, }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({ write: () => Effect.succeed([]), writeWithEffects: (input) => @@ -2527,6 +2542,7 @@ it.effect( Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void, }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({ write: () => Effect.succeed([]), writeWithEffects: (input) => @@ -2697,6 +2713,7 @@ it.effect( Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void, }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({ write: () => Effect.succeed([]), writeWithEffects: (input) => @@ -3138,6 +3155,7 @@ it.effect( Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void, }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({ write: () => Effect.succeed([]), writeWithEffects: (input) => @@ -3862,6 +3880,7 @@ function captureRootRunTermination(input: { Layer.mergeAll( McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({ write: (payload) => Effect.gen(function* () { @@ -4343,6 +4362,7 @@ function runBackgroundItemScenario( Layer.mergeAll( McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), + layerTurnItemPositionsTest, Layer.mock(EventSink.EventSinkV2)({ write: () => Effect.succeed([]), writeWithEffects: (input) => diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index b2fe77f6cd85..f73c8224ab77 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -49,6 +49,7 @@ import { makeProviderFailureTurnItem, } from "@t3tools/provider-core/server/failure"; import * as RunFinalizationService from "./RunFinalizationService.ts"; +import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; export interface ProviderEventRoutingState { readonly ownedThreadIds: ReadonlySet; @@ -569,6 +570,7 @@ export const layer: Layer.Layer< | ProviderEventIngestor.ProviderEventIngestorV2 | ServerSettings.ServerSettingsService | McpAppModelContext.McpAppModelContext + | TurnItemPositionStore.TurnItemPositionStoreV2 > = Layer.effect( RunExecutionServiceV2, Effect.gen(function* () { @@ -578,6 +580,7 @@ export const layer: Layer.Layer< const providerEventIngestor = yield* ProviderEventIngestor.ProviderEventIngestorV2; const serverSettings = yield* ServerSettings.ServerSettingsService; const mcpAppModelContext = yield* McpAppModelContext.McpAppModelContext; + const turnItemPositions = yield* TurnItemPositionStore.TurnItemPositionStoreV2; const finalizationObserver = yield* RunFinalizationService.RunFinalizationObserver; const writeFinalRunEvents = (input: { @@ -1224,6 +1227,9 @@ export const layer: Layer.Layer< return true; }); const filterAssistantEvent = makeAssistantStreamingFilter(responseStreamingMode); + // Held text writes nothing, so its item would take a position only + // when finally written, behind every item that began after it. + const reservedHeldTurnItemIds = new Set(); const providerEventFiber = yield* eventSubscription.events.pipe( Stream.filterEffect((event) => Ref.modify(eventRouting, (state) => routeProviderEvent(event, routeIdentity, state)), @@ -1286,6 +1292,19 @@ export const layer: Layer.Layer< // stream on this run's background probe. yield* Ref.set(providerThreadOwnerLost, true); } + } else if ( + event.type === "turn_item.updated" && + (event.turnItem.type === "assistant_message" || + event.turnItem.type === "reasoning") && + event.turnItem.streaming && + !reservedHeldTurnItemIds.has(event.turnItem.id) + ) { + yield* turnItemPositions.allocate({ + threadId: event.turnItem.threadId, + turnItemId: event.turnItem.id, + runId: event.turnItem.runId, + }); + reservedHeldTurnItemIds.add(event.turnItem.id); } if (event.type === "provider_thread.updated") { if (event.providerThread.id === input.providerThread.id && storedEventCount > 0) { diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 730b72feea6b..5ee2e2f1e37f 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -10,6 +10,7 @@ import { ProviderThreadId, ProviderTurnId, ThreadId, + TurnItemId, type ModelSelection, type OrchestrationV2DomainEvent, } from "@t3tools/contracts"; @@ -702,3 +703,120 @@ it.effect("starts a steer that missed the turn on the saved next-turn selection" }), ), ); + +// A draft that never reaches a paragraph boundary is held back while it +// streams. Codex only flushes it at turn end, after the final answer is written, +// so the draft must already own its place from before the steer. +it.effect("keeps a held commentary draft before the steer when turn end flushes it", () => + Effect.scoped( + Effect.gen(function* () { + const { events, started, layer, startFirstTurn } = + yield* nextTurnSelectionHarness("steering-held-draft"); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const first = started[0]!; + const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; + const now = yield* DateTime.now; + const assistantItem = (id: string, text: string, streaming: boolean) => + ({ + type: "turn_item.updated", + driver, + turnItem: { + id: TurnItemId.make(id), + threadId, + runId: first.runId, + nodeId: first.rootNodeId, + providerThreadId: turn.providerThreadId, + providerTurnId: turn.id, + nativeItemRef: null, + parentItemId: null, + ordinal: 0, + status: streaming ? "running" : "completed", + title: null, + startedAt: now, + completedAt: streaming ? null : now, + updatedAt: now, + type: "assistant_message", + messageId: MessageId.make(`message:${id}`), + text, + streaming, + }, + }) satisfies ProviderAdapter.ProviderAdapterV2Event; + const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + + // The cut-off draft starts first; the commentary after it is written, + // so the draft has been seen by the time the steer arrives. + yield* Queue.offer(events, assistantItem("draft", "Let me look at the", true)); + const commentaryWritten = yield* watch( + (event) => event.type === "turn-item.updated" && event.payload.id === "commentary", + ); + yield* Queue.offer(events, assistantItem("commentary", "Checking the popover.", false)); + yield* Fiber.join(commentaryWritten); + + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("steer"), + threadId, + messageId: MessageId.make("message:steer"), + text: "fix the popover", + attachments: [], + dispatchMode: { type: "steer_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + + const settled = yield* watch( + (event) => + event.type === "run.updated" && + event.payload.id === first.runId && + event.payload.status === "waiting", + ); + yield* Queue.offer(events, assistantItem("answer", "All fixed.", false)); + yield* Queue.offer(events, assistantItem("draft", "Let me look at the", false)); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...turn, status: "completed", completedAt: yield* DateTime.now }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: turn.providerThreadId, + providerTurnId: turn.id, + runOrdinal: first.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(settled); + yield* worker.drain(); + + const items = (yield* orchestrator.getThreadProjection(threadId)).turnItems + .toSorted((left, right) => left.ordinal - right.ordinal) + .flatMap((item) => + item.type === "user_message" + ? [String(item.messageId)] + : item.type === "assistant_message" + ? [String(item.id)] + : [], + ); + assert.deepEqual(items, [ + "message:first", + "draft", + "commentary", + "message:steer", + "answer", + ]); + }).pipe(Effect.provide(layer)); + }), + ), +); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 18ad606d9767..c48eae1dd6c8 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -148,6 +148,7 @@ const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe( layerEventSinkProvided, IdAllocator.layer, layerProviderEventIngestorProvided, + TurnItemPositionStore.layer, ), ), ); diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 1ee5fb996484..f1e78fac80e6 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -380,6 +380,7 @@ export function layerWithRegistry( IdAllocator.layer, layerProviderEventIngestorProvided, layerServerSettings, + layerStores, ), ), ); diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/opencode2_tool_call/output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/opencode2_tool_call/output.ts index 7118276536f8..5df8be80c7d5 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/opencode2_tool_call/output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/opencode2_tool_call/output.ts @@ -23,13 +23,14 @@ export function assertOpenCode2ToolCallOutput( assertSemanticProjectionIntegrity(projection); assertVisibleTurnItemsMirrorLocalTurnItems(projection); // Buffered delivery keeps streaming text out of the read model until - // `text.ended`, which OpenCode sends after both tool calls started. + // `text.ended`, which OpenCode sends after both tool calls started. The text + // keeps the position it took when it began, ahead of the tool calls. assertTurnItemTypeSequence(projection, [ "user_message", + "assistant_message", "dynamic_tool", "command_execution", "assistant_message", - "assistant_message", "checkpoint", ]); assertUserMessagesInclude(projection, [OPENCODE2_TOOL_CALL_PROMPT]);