From f65c2d53a1b5f66e4fcef7ac2e95f252de943689 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Fri, 6 Nov 2026 05:19:08 +0000 Subject: [PATCH 01/11] fix(server): show held assistant text while a turn stays open Paragraph streaming held a turn's latest text back until the next Markdown boundary or the final update, so a provider that paused mid-turn (a backgrounded command, a long tool call) left that text hidden while the turn stayed open. A held snapshot is now delivered whole after 800 ms of inactivity. The filter keeps the latest held message.updated and turn_item.updated snapshot per message. It never shrinks delivered text, but a provider rewrite or truncation replaces it. Turn mode is unchanged. The idle timer only wakes ingestion. The flush runs as its own step of the run's provider event consumer, ordered with routing and snapshots and outside any batched commit, so it never waits on the shared batch worker while that worker waits on it. A write has committed before its text counts as delivered. consumeProviderEvents accepts such non-snapshot control items, which it already orders outside the shared commit. The flush rechecks routing, so a child handed to another run has its held text dropped. Root finalization and the end of ingestion close the timer before it can reopen settled items. A failed idle write keeps the text held and retries it with backoff (1s doubling to 30s, warning at most every 10 minutes) while provider ingestion continues; a newer snapshot with an earlier deadline reschedules the wake. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/EventSink.ts | 8 +- .../orchestration-v2/ProviderCommitBatch.ts | 10 +- .../RunExecutionService.test.ts | 666 +++++++++++++++++- .../orchestration-v2/RunExecutionService.ts | 210 ++++-- .../assistantStreaming.test.ts | 58 +- .../orchestration-v2/assistantStreaming.ts | 172 ++++- 6 files changed, 1005 insertions(+), 119 deletions(-) diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index 81e64ddb70920..0c42ac73ff9cb 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -12,7 +12,6 @@ import { type ProjectId, ThreadId, } from "@t3tools/contracts"; -import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProviderCommitBatch from "./ProviderCommitBatch.ts"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; @@ -79,9 +78,10 @@ const CurrentProviderBatch = Context.Reference( - source: Stream.Stream, - process: (event: ProviderAdapter.ProviderAdapterV2Event) => Effect.Effect, + /** Items that are not provider snapshots, such as a consumer's own control items, never batch. */ + readonly consumeProviderEvents: ( + source: Stream.Stream, + process: (event: A) => Effect.Effect, after: Effect.Effect, canBatch: Effect.Effect, ) => Effect.Effect< diff --git a/apps/server/src/orchestration-v2/ProviderCommitBatch.ts b/apps/server/src/orchestration-v2/ProviderCommitBatch.ts index 60cee2b632f66..1a6ca8d28a609 100644 --- a/apps/server/src/orchestration-v2/ProviderCommitBatch.ts +++ b/apps/server/src/orchestration-v2/ProviderCommitBatch.ts @@ -1,4 +1,3 @@ -import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; import * as Config from "effect/Config"; @@ -19,8 +18,7 @@ export const enabled = Config.Boolean("T3CODE_PROVIDER_COMMIT_BATCHING").pipe( Effect.orDie, ); -type Event = ProviderAdapter.ProviderAdapterV2Event; -const isSnapshot = (event: Event) => +const isSnapshot = (event: { readonly type: string }) => event.type === "node.updated" || event.type === "message.updated" || event.type === "turn_item.updated" || @@ -232,9 +230,9 @@ export const make = Effect.fn("ProviderCommitBatch.make")(function* ( ), ); }); - const consume = Effect.fnUntraced(function* ( - source: Stream.Stream, - process: (event: Event) => Effect.Effect, + const consume = Effect.fnUntraced(function* ( + source: Stream.Stream, + process: (event: A) => Effect.Effect, after: Effect.Effect, canBatch: Effect.Effect = Effect.succeed(true), ) { diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 5bfb4fe34dc6a..459647be6c914 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -29,13 +29,16 @@ import { } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { CodexAppServerRequestError } from "effect-codex-app-server/errors"; import { SqlError, classifySqliteError } from "effect/sql/SqlError"; @@ -47,6 +50,7 @@ import * as EventSink from "./EventSink.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProjectionStore from "./ProjectionStore.ts"; +import * as ProviderCommitBatch from "./ProviderCommitBatch.ts"; import * as ProviderEventIngestor from "./ProviderEventIngestor.ts"; import * as ProviderUsageAnalytics from "../telemetry/ProviderUsageAnalytics.ts"; import * as McpAppModelContext from "../mcpApps/McpAppModelContext.ts"; @@ -3692,8 +3696,19 @@ function rootTerminalEvent( function runBackgroundItemScenario( key: string, - makeEvents: (ids: BackgroundScenarioIds) => ReadonlyArray, + makeEvents: ( + ids: BackgroundScenarioIds, + ) => + | ReadonlyArray + | Stream.Stream, options?: { + readonly onIngest?: ( + event: ProviderAdapter.ProviderAdapterV2Event, + write: { readonly inSharedCommit: boolean }, + ) => Effect.Effect; + // Consumes through the real commit coordinator, as the event sink does. + readonly batched?: boolean; + readonly startTurn?: Effect.Effect; readonly keepEventStreamOpen?: boolean; readonly hasPendingBackgroundWorkForThread?: ( providerThread: OrchestrationV2ProviderThread, @@ -3709,6 +3724,9 @@ function runBackgroundItemScenario( readonly onFinalized?: ( events: ReadonlyArray, ) => Effect.Effect; + readonly onInterruptTurn?: (input: { + readonly requestRuntimeRestart?: boolean; + }) => Effect.Effect; }, ) { return Effect.gen(function* () { @@ -3719,6 +3737,24 @@ function runBackgroundItemScenario( const previewActivity = yield* PreviewAgentActivity.PreviewAgentActivity.pipe( Effect.provide(PreviewAgentActivity.layer), ); + let inSharedCommit = false; + const coordinator = + options?.batched === true + ? yield* ProviderCommitBatch.make( + (work) => + Effect.suspend(() => { + inSharedCommit = true; + return work; + }).pipe( + Effect.ensuring( + Effect.sync(() => { + inSharedCommit = false; + }), + ), + ), + true, + ) + : undefined; const layerTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( @@ -3727,6 +3763,7 @@ function runBackgroundItemScenario( Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), layerTurnItemPositionsTest, mockEventSink({ + ...(coordinator === undefined ? {} : { consumeProviderEvents: coordinator.consume }), write: () => Effect.succeed([]), writeWithEffects: (input) => Effect.gen(function* () { @@ -3744,9 +3781,10 @@ function runBackgroundItemScenario( }), IdAllocator.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ - ingestNormalized: (input) => - Effect.gen(function* () { + ingestNormalized: (input) => { + const write = Effect.gen(function* () { const event = input.event; + yield* options?.onIngest?.(event, { inSharedCommit }) ?? Effect.void; if (event.type === "turn_item.updated") { yield* Ref.update(observed, (current) => [ ...current, @@ -3760,7 +3798,10 @@ function runBackgroundItemScenario( ]); } return []; - }), + }); + // A write outside a batch waits for queued snapshots, as the event sink's does. + return coordinator === undefined ? write : coordinator.barrier(write); + }, }), ServerSettings.layerTest(), ), @@ -3780,7 +3821,8 @@ function runBackgroundItemScenario( : { hasPendingBackgroundWorkForThread: options.hasPendingBackgroundWorkForThread }), subscribeEvents: Effect.gen(function* () { yield* options?.onSubscribe ?? Effect.void; - const events = Stream.fromIterable(makeEvents(ids)).pipe( + const source = makeEvents(ids); + const events = (Stream.isStream(source) ? source : Stream.fromIterable(source)).pipe( Stream.concat( Stream.fromEffect( Effect.andThen( @@ -3798,7 +3840,9 @@ function runBackgroundItemScenario( close: Deferred.succeed(ingestionDone, undefined), }; }), - startTurn: () => Effect.void, + startTurn: () => options?.startTurn ?? Effect.void, + interruptTurn: (interrupt: { readonly requestRuntimeRestart?: boolean }) => + options?.onInterruptTurn?.(interrupt) ?? Effect.void, } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime, run: { id: ids.runId, @@ -3847,8 +3891,9 @@ function runBackgroundItemScenario( }); }).pipe(Effect.provide(layerTest)); - const closed = yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds")); - assert.isTrue(Option.isSome(closed), "event ingestion fiber did not finish"); + // No timeout: tests advance the TestClock, which would fire it early. + yield* Deferred.await(ingestionDone); + return yield* Ref.get(observed); }); } @@ -3900,3 +3945,608 @@ it.effect("keeps a failed root's still-running native children live while the ru ); }), ); + +it.effect.each([ + { name: "", batched: false }, + { name: " through batched commits", batched: true }, +])( + "delivers a held paragraph after the provider goes idle with its prompt still open$name", + ({ batched }) => + Effect.gen(function* () { + const key = `idle-paragraph${batched ? "-batched" : ""}`; + const ids = backgroundScenarioIds(key); + const now = yield* DateTime.now; + const providerEvents = yield* Queue.unbounded(); + const delivered = yield* Ref.make>([]); + const paragraphDelivered = yield* Deferred.make(); + const idleDelivered = yield* Deferred.make(); + const assistantText = (text: string, streaming: boolean) => + [ + { + type: "message.updated", + driver, + message: { + createdBy: "agent", + creationSource: "provider", + id: MessageId.make(`message:${key}`), + threadId: ids.threadId, + runId: ids.runId, + nodeId: ids.rootNodeId, + role: "assistant", + text, + attachments: [], + streaming, + createdAt: now, + updatedAt: now, + }, + }, + { + type: "turn_item.updated", + driver, + turnItem: { + ...makeLinkedChildTurnItemFixture({ ids, driver, type: "assistant_message" }), + id: ids.itemId, + threadId: ids.threadId, + runId: ids.runId, + nodeId: ids.rootNodeId, + providerTurnId: ids.rootProviderTurnId, + text, + streaming, + }, + }, + ] as ReadonlyArray; + + const scenario = yield* runBackgroundItemScenario( + key, + () => Stream.fromQueue(providerEvents), + { + batched, + onIngest: (event, write) => + event.type === "message.updated" || event.type === "turn_item.updated" + ? Effect.gen(function* () { + const message = event.type === "message.updated" ? event.message : event.turnItem; + const text = "text" in message ? message.text : ""; + const streaming = "streaming" in message && message.streaming; + yield* Ref.update(delivered, (current) => [ + ...current, + `${event.type}:${streaming ? "streaming" : "final"}:${JSON.stringify(text)}${ + write.inSharedCommit ? " in commit" : "" + }`, + ]); + if (event.type === "turn_item.updated") { + yield* Deferred.succeed(paragraphDelivered, undefined); + if (streaming && text === "First.\n\nWaiting on the build") + yield* Deferred.succeed(idleDelivered, undefined); + } + }) + : Effect.void, + }, + ).pipe(Effect.forkChild); + + yield* Queue.offerAll(providerEvents, assistantText("First.\n\nWaiting on the build", true)); + yield* Deferred.await(paragraphDelivered); + // The provider keeps its prompt open while a backgrounded command runs. + yield* TestClock.adjust("1 second"); + yield* Deferred.await(idleDelivered); + yield* Queue.offerAll(providerEvents, [ + ...assistantText("First.\n\nWaiting on the build to", true), + ...assistantText("First.\n\nWaiting on the build to finish.", false), + rootTerminalEvent(ids, "completed"), + ]); + yield* Fiber.join(scenario); + + // Snapshots share the batch commit; the idle write commits alone, before it counts as shown. + const committed = batched ? " in commit" : ""; + assert.deepEqual(yield* Ref.get(delivered), [ + `message.updated:streaming:"First.\\n\\n"${committed}`, + `turn_item.updated:streaming:"First.\\n\\n"${committed}`, + 'message.updated:streaming:"First.\\n\\nWaiting on the build"', + 'turn_item.updated:streaming:"First.\\n\\nWaiting on the build"', + `message.updated:final:"First.\\n\\nWaiting on the build to finish."${committed}`, + `turn_item.updated:final:"First.\\n\\nWaiting on the build to finish."${committed}`, + ]); + }), +); + +const heldChildText = "partial response"; + +function childAssistantSnapshot( + ids: BackgroundScenarioIds, +): ProviderAdapter.ProviderAdapterV2Event { + return { + type: "turn_item.updated", + driver, + turnItem: { + ...makeLinkedChildTurnItemFixture({ ids, driver, type: "assistant_message" }), + text: heldChildText, + }, + } as ProviderAdapter.ProviderAdapterV2Event; +} + +function linkedSubagentEvent( + ids: BackgroundScenarioIds, + status: "running" | "completed", +): ProviderAdapter.ProviderAdapterV2Event { + return { + type: "subagent.updated", + driver, + subagent: { + ...makeRunOwnedSubagentFixture({ + ids, + providerInstanceId: ProviderInstanceId.make("codex"), + childThreadId: ids.childThreadId, + driver, + status: "running", + }), + status, + }, + }; +} + +const isHeldChildSnapshot = (event: ProviderAdapter.ProviderAdapterV2Event) => + event.type === "turn_item.updated" && + "text" in event.turnItem && + event.turnItem.text === heldChildText; + +// Feeds a held child snapshot, then waits until ingestion has read past it. +const holdChildSnapshot = Effect.fn(function* ( + ids: BackgroundScenarioIds, + events: Queue.Enqueue, + subagentIngested: Deferred.Deferred, +) { + yield* Queue.offerAll(events, [ + childThreadCreatedEvent(ids), + childAssistantSnapshot(ids), + linkedSubagentEvent(ids, "running"), + ]); + yield* Deferred.await(subagentIngested); +}); + +it.effect("settles a child message only an idle flush wrote when the root is interrupted", () => + Effect.gen(function* () { + const ids = backgroundScenarioIds("idle-child-interrupt"); + const events = yield* Queue.unbounded(); + const subagentIngested = yield* Deferred.make(); + const flushed = yield* Deferred.make(); + const finalized = yield* Ref.make>([]); + const scenario = yield* runBackgroundItemScenario( + "idle-child-interrupt", + () => Stream.fromQueue(events), + { + onIngest: (event) => + event.type === "subagent.updated" + ? Deferred.succeed(subagentIngested, undefined) + : isHeldChildSnapshot(event) + ? Deferred.succeed(flushed, undefined) + : Effect.void, + onFinalized: (written) => Ref.set(finalized, written), + }, + ).pipe(Effect.forkChild); + + yield* holdChildSnapshot(ids, events, subagentIngested); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(flushed); + yield* Queue.offer(events, rootTerminalEvent(ids, "interrupted")); + yield* Fiber.join(scenario); + + const settled = (yield* Ref.get(finalized)).flatMap((event) => + event.type === "turn-item.updated" && event.payload.id === ids.childItemId + ? [event.payload] + : [], + ); + assert.equal(settled.length, 1); + assert.equal(settled[0]?.status, "interrupted"); + assert.equal("text" in settled[0]! ? settled[0].text : null, heldChildText); + }), +); + +it.effect.each([ + { name: "completed", terminalStatus: "completed" as const }, + { name: "failed with its children continuing", terminalStatus: "failed" as const }, +])("idle-flushes a continuing child's text after the root $name", ({ terminalStatus }) => + Effect.gen(function* () { + const key = `idle-child-after-${terminalStatus}`; + const ids = backgroundScenarioIds(key); + const events = yield* Queue.unbounded(); + const subagentIngested = yield* Deferred.make(); + const rootFinalized = yield* Deferred.make(); + const flushed = yield* Deferred.make(); + const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { + onIngest: (event) => + event.type === "subagent.updated" + ? Deferred.succeed(subagentIngested, undefined) + : isHeldChildSnapshot(event) + ? Deferred.succeed(flushed, undefined) + : Effect.void, + onFinalized: () => Deferred.succeed(rootFinalized, undefined), + }).pipe(Effect.forkChild); + + yield* holdChildSnapshot(ids, events, subagentIngested); + const terminal = rootTerminalEvent(ids, terminalStatus); + yield* Queue.offer( + events, + terminalStatus === "failed" + ? ({ ...terminal, childTurnsContinue: true } as ProviderAdapter.ProviderAdapterV2Event) + : terminal, + ); + yield* Deferred.await(rootFinalized); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(flushed); + yield* Queue.offer(events, linkedSubagentEvent(ids, "completed")); + const observed = yield* Fiber.join(scenario); + + assert.deepEqual(observed.slice(-3), [ + "root-finalized", + "turn_item:running", + "subagent:completed", + ]); + }), +); + +it.effect.each([ + { name: "before", afterRoot: false, batched: false }, + { name: "after", afterRoot: true, batched: false }, + { name: "before", afterRoot: false, batched: true }, + { name: "after", afterRoot: true, batched: true }, +])( + "retries a failed idle flush $name the root completes and keeps the run live (batched: $batched)", + ({ afterRoot, batched }) => + Effect.gen(function* () { + const key = `idle-flush-retry-${afterRoot ? "after" : "before"}-root${batched ? "-batched" : ""}`; + const ids = backgroundScenarioIds(key); + const events = yield* Queue.unbounded(); + const subagentIngested = yield* Deferred.make(); + const rootFinalized = yield* Deferred.make(); + const firstWriteFailed = yield* Deferred.make(); + const failureHandled = yield* Deferred.make(); + const retried = yield* Deferred.make(); + const attempts = yield* Ref.make(0); + const childWrites = yield* Ref.make>([]); + const interrupts = yield* Ref.make(0); + const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { + batched, + onIngest: (event) => + Effect.gen(function* () { + if (event.type === "subagent.updated") { + yield* Deferred.succeed(subagentIngested, undefined); + if ((yield* Ref.get(attempts)) > 0) + yield* Deferred.succeed(failureHandled, undefined); + return; + } + if (event.type !== "turn_item.updated" || event.turnItem.type !== "assistant_message") + return; + if ( + isHeldChildSnapshot(event) && + (yield* Ref.getAndUpdate(attempts, (n) => n + 1)) === 0 + ) { + yield* Deferred.succeed(firstWriteFailed, undefined); + yield* Effect.sleep("2 seconds"); + return yield* Effect.die(new Error("disk I/O error")); + } + const { text, streaming } = event.turnItem; + yield* Ref.update(childWrites, (current) => [ + ...current, + `${streaming ? "streaming" : "final"}:${text}`, + ]); + if (isHeldChildSnapshot(event)) yield* Deferred.succeed(retried, undefined); + }), + onFinalized: () => Deferred.succeed(rootFinalized, undefined), + onInterruptTurn: () => Ref.update(interrupts, (n) => n + 1), + }).pipe(Effect.forkChild); + + yield* holdChildSnapshot(ids, events, subagentIngested); + if (afterRoot) { + yield* Queue.offer(events, rootTerminalEvent(ids, "completed")); + yield* Deferred.await(rootFinalized); + } + yield* TestClock.adjust("1 second"); + yield* Deferred.await(firstWriteFailed); + yield* TestClock.adjust("2 seconds"); + yield* Queue.offer(events, linkedSubagentEvent(ids, "running")); + yield* Deferred.await(failureHandled); + assert.equal(yield* Ref.get(attempts), 1); + assert.deepEqual(yield* Ref.get(childWrites), []); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(retried); + const childText = (text: string, streaming: boolean) => { + const snapshot = childAssistantSnapshot(ids); + return snapshot.type === "turn_item.updated" + ? ({ + ...snapshot, + turnItem: { ...snapshot.turnItem, text, streaming }, + } as ProviderAdapter.ProviderAdapterV2Event) + : snapshot; + }; + yield* Queue.offerAll(events, [ + childText(`${heldChildText} and more`, true), + childText(`${heldChildText} and more.`, false), + ...(afterRoot ? [] : [rootTerminalEvent(ids, "completed")]), + linkedSubagentEvent(ids, "completed"), + ]); + const observed = yield* Fiber.join(scenario); + + assert.deepEqual(yield* Ref.get(childWrites), [ + `streaming:${heldChildText}`, + `final:${heldChildText} and more.`, + ]); + assert.equal(yield* Ref.get(interrupts), 0); + assert.include(observed, "root-finalized"); + assert.equal(observed.at(-1), "subagent:completed"); + }), +); + +it.effect("reschedules idle text when a newer snapshot precedes a failed write's retry", () => + Effect.gen(function* () { + const key = "idle-flush-earlier-deadline"; + const ids = backgroundScenarioIds(key); + const events = yield* Queue.unbounded(); + const subagentIngested = yield* Deferred.make(); + const markers = yield* Queue.unbounded(); + const newerWrite = yield* Deferred.make(); + const attempts = yield* Ref.make(0); + const childWrites = yield* Ref.make>([]); + const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { + onIngest: (event) => + Effect.gen(function* () { + if (event.type === "subagent.updated") + yield* Deferred.succeed(subagentIngested, undefined); + if (event.type === "message.updated" && event.message.role === "user") + yield* Queue.offer(markers, undefined); + if (event.type !== "turn_item.updated" || event.turnItem.type !== "assistant_message") + return; + if ((yield* Ref.getAndUpdate(attempts, (n) => n + 1)) < 2) + return yield* Effect.die(new Error("disk I/O error")); + const { text } = event.turnItem; + yield* Ref.update(childWrites, (current) => [...current, text]); + if (text === "New partial response") { + yield* Deferred.succeed(newerWrite, undefined); + } + }), + }).pipe(Effect.forkChild); + const marker = streamedMessageEvent({ + id: `message:${key}:marker`, + threadId: ids.threadId, + runId: ids.runId, + role: "user", + text: "marker", + streaming: false, + }); + + yield* holdChildSnapshot(ids, events, subagentIngested); + yield* TestClock.adjust("1 second"); + yield* Queue.offer(events, marker); + yield* Queue.take(markers); + assert.equal(yield* Ref.get(attempts), 1); + yield* TestClock.adjust("1 second"); + yield* Queue.offer(events, marker); + yield* Queue.take(markers); + assert.equal(yield* Ref.get(attempts), 2); + + const newer = childAssistantSnapshot(ids); + assert(newer.type === "turn_item.updated" && newer.turnItem.type === "assistant_message"); + yield* Queue.offerAll(events, [ + { ...newer, turnItem: { ...newer.turnItem, text: "New partial response" } }, + marker, + ]); + yield* Queue.take(markers); + yield* TestClock.adjust("800 millis"); + yield* Deferred.await(newerWrite); + assert.deepEqual(yield* Ref.get(childWrites), ["New partial response"]); + + yield* Queue.offerAll(events, [ + rootTerminalEvent(ids, "completed"), + linkedSubagentEvent(ids, "completed"), + ]); + yield* Fiber.join(scenario); + }), +); + +function streamedMessageEvent(input: { + readonly id: string; + readonly threadId: ThreadId; + readonly runId: RunId | null; + readonly role: "assistant" | "user"; + readonly text: string; + readonly streaming: boolean; +}): ProviderAdapter.ProviderAdapterV2Event { + const now = DateTime.makeUnsafe("2026-07-21T12:00:00.000Z"); + return { + type: "message.updated", + driver, + message: { + createdBy: input.role === "user" ? "user" : "agent", + creationSource: "provider", + id: MessageId.make(input.id), + threadId: input.threadId, + runId: input.runId, + nodeId: null, + role: input.role, + text: input.text, + attachments: [], + streaming: input.streaming, + createdAt: now, + updatedAt: now, + }, + } as ProviderAdapter.ProviderAdapterV2Event; +} + +it.effect.each([ + { name: "", failFirst: false }, + { name: " after its first idle write failed", failFirst: true }, +])("drops a child's held text once the child is handed to another run$name", ({ failFirst }) => + Effect.gen(function* () { + const key = `idle-child-handoff${failFirst ? "-after-failure" : ""}`; + const ids = backgroundScenarioIds(key); + const events = yield* Queue.unbounded(); + const subagentIngested = yield* Deferred.make(); + const firstWriteFailed = yield* Deferred.make(); + const handoffRead = yield* Deferred.make(); + const rootFlushed = yield* Deferred.make(); + const staleWrites = yield* Ref.make>([]); + const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { + onIngest: (event) => { + if (event.type === "subagent.updated") return Deferred.succeed(subagentIngested, undefined); + if (event.type !== "message.updated" && event.type !== "turn_item.updated") + return Effect.void; + const message = event.type === "message.updated" ? event.message : event.turnItem; + if (message.threadId === ids.childThreadId) { + return Deferred.succeed(firstWriteFailed, undefined).pipe( + Effect.flatMap((first) => + failFirst && first + ? Effect.die(new Error("disk I/O error")) + : Ref.update(staleWrites, (current) => [...current, event.type]), + ), + ); + } + if (event.type === "message.updated" && event.message.role === "user") + return Deferred.succeed(handoffRead, undefined); + return Deferred.succeed(rootFlushed, undefined); + }, + }).pipe(Effect.forkChild); + + yield* holdChildSnapshot(ids, events, subagentIngested); + if (failFirst) { + // The held snapshot stays held for a retry that must not outlive the handoff. + yield* TestClock.adjust("1 second"); + yield* Deferred.await(firstWriteFailed); + } + const handoff = linkedSubagentEvent(ids, "running"); + yield* Queue.offerAll(events, [ + streamedMessageEvent({ + id: `message:${key}:child`, + threadId: ids.childThreadId, + runId: null, + role: "assistant", + text: heldChildText, + streaming: true, + }), + handoff.type === "subagent.updated" + ? { ...handoff, subagent: { ...handoff.subagent, runId: RunId.make(`run:${key}:next`) } } + : handoff, + // Due in the same flush as the child's text, after it. + streamedMessageEvent({ + id: `message:${key}:root`, + threadId: ids.threadId, + runId: ids.runId, + role: "assistant", + text: "Root partial", + streaming: true, + }), + streamedMessageEvent({ + id: `message:${key}:marker`, + threadId: ids.threadId, + runId: ids.runId, + role: "user", + text: "marker", + streaming: false, + }), + ]); + yield* Deferred.await(handoffRead); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(rootFlushed); + yield* Queue.end(events); + yield* Fiber.join(scenario); + + assert.deepEqual(yield* Ref.get(staleWrites), []); + }), +); + +it.effect("flushes idle text only after the snapshots queued ahead of its wake", () => + Effect.gen(function* () { + const key = "idle-flush-queued-snapshot"; + const ids = backgroundScenarioIds(key); + const events = yield* Queue.unbounded(); + const paragraphWritten = yield* Deferred.make(); + const blockerStarted = yield* Deferred.make(); + const releaseBlocker = yield* Deferred.make(); + const writes = yield* Ref.make>([]); + const message = (id: string, text: string, streaming: boolean) => + streamedMessageEvent({ + id: `message:${key}:${id}`, + threadId: ids.threadId, + runId: ids.runId, + role: "assistant", + text, + streaming, + }); + const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { + batched: true, + onIngest: (event) => { + if (event.type !== "message.updated") return Effect.void; + if (event.message.text === "Other") + return Deferred.succeed(blockerStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseBlocker)), + ); + const { streaming, text } = event.message; + return Ref.update(writes, (current) => [ + ...current, + `${streaming ? "streaming" : "final"}:${text}`, + ]).pipe(Effect.andThen(Deferred.succeed(paragraphWritten, undefined))); + }, + }).pipe(Effect.forkChild); + + yield* Queue.offer(events, message("text", "First.\n\nWaiting on the build", true)); + yield* Deferred.await(paragraphWritten); + yield* Queue.offerAll(events, [ + message("blocker", "Other", false), + message("text", "First.\n\nWaiting on the build to finish.", false), + ]); + yield* Deferred.await(blockerStarted); + // The held text comes due while its final snapshot waits behind the blocked write. + yield* TestClock.adjust("1 second"); + yield* Deferred.succeed(releaseBlocker, undefined); + yield* Queue.offer(events, rootTerminalEvent(ids, "completed")); + yield* Fiber.join(scenario); + + assert.deepEqual(yield* Ref.get(writes), [ + "streaming:First.\n\n", + "final:First.\n\nWaiting on the build to finish.", + ]); + }), +); + +it.effect("interrupts a blocked idle flush write when ingestion is cancelled", () => + Effect.gen(function* () { + const ids = backgroundScenarioIds("idle-flush-cancelled"); + const events = yield* Queue.unbounded(); + const subagentIngested = yield* Deferred.make(); + const writeStarted = yield* Deferred.make(); + const writeInterrupted = yield* Deferred.make(); + const failStart = yield* Deferred.make(); + const scenario = yield* runBackgroundItemScenario( + "idle-flush-cancelled", + () => Stream.fromQueue(events), + { + batched: true, + // A failed turn start cancels ingestion. + startTurn: Deferred.await(failStart).pipe( + Effect.andThen( + new ProviderAdapter.ProviderAdapterTurnStartError({ + driver, + threadId: ids.threadId, + providerThreadId: ids.providerThreadId, + runId: ids.runId, + cause: "start failed", + }), + ), + ), + onIngest: (event) => + event.type === "subagent.updated" + ? Deferred.succeed(subagentIngested, undefined) + : isHeldChildSnapshot(event) + ? Deferred.succeed(writeStarted, undefined).pipe( + Effect.andThen(Effect.never), + Effect.onInterrupt(() => Deferred.succeed(writeInterrupted, undefined)), + ) + : Effect.void, + }, + ).pipe(Effect.forkChild); + + yield* holdChildSnapshot(ids, events, subagentIngested); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(writeStarted); + yield* Deferred.succeed(failStart, undefined); + yield* Deferred.await(writeInterrupted); + yield* Fiber.join(scenario); + }), +); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 304eeaba43421..5c0ac4e248747 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -32,9 +32,11 @@ import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts"; import * as McpAppModelContext from "../mcpApps/McpAppModelContext.ts"; @@ -1072,6 +1074,10 @@ export const layer: Layer.Layer< ReadonlySet >(new Set(inheritedBackgroundTurnItemsById.keys())); const openRunOwnedSubagents = yield* Ref.make(emptyOpenRunOwnedSubagentProjection()); + const assistantStreaming = makeAssistantStreamingFilter(responseStreamingMode); + // Set when ingestion ends or finalization settles the run's open items, which a later + // streaming snapshot must not reopen. + let idleFlushClosed = false; const finalizeRootRun = Effect.fn("RunExecutionService.finalizeRootRun")(function* ( terminal: ProviderTerminalEvent, ) { @@ -1114,6 +1120,7 @@ export const layer: Layer.Layer< ); if (isRunOwnedSubagentTerminalStatus(terminal.status) && !childTurnsContinue(terminal)) { yield* Ref.set(openRunOwnedSubagents, emptyOpenRunOwnedSubagentProjection()); + idleFlushClosed = true; } yield* Ref.set(rootRunFinalized, true); }); @@ -1314,7 +1321,6 @@ 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(); @@ -1397,6 +1403,137 @@ export const layer: Layer.Layer< } return { event, accepted }; }); + const ingestDelivered = Effect.fnUntraced(function* ( + event: ProviderAdapter.ProviderAdapterV2Event, + deliveredEvent: ProviderAdapter.ProviderAdapterV2Event, + ) { + const rootTerminalAlreadySeen = yield* Ref.get(rootTerminalSeen); + const isRootProviderThreadUpdate = + event.type === "provider_thread.updated" && + event.providerThread.id === input.providerThread.id; + const nativeChildTurn = + event.type === "provider_turn.updated" && + event.providerTurn.providerThreadId !== input.providerThread.id; + if (event.type === "provider_turn.updated" && !nativeChildTurn) { + firstRootProviderTurnId ??= event.providerTurn.id; + } + const storedEvents = yield* providerEventIngestor.ingestNormalized({ + analyticsContext: nativeChildTurn + ? { + ...analyticsContext, + origin: NATIVE_SUBAGENT_TURN_ORIGIN, + segment: "native_child", + } + : { + ...analyticsContext, + segment: + event.type === "provider_turn.updated" && + event.providerTurn.id !== firstRootProviderTurnId + ? "continuation" + : input.attempt.reason, + }, + providerSessionId: input.providerSessionId, + providerInstanceId: input.run.providerInstanceId, + threadId: input.run.threadId, + runId: input.run.id, + nodeId: input.rootNode.id, + event: deliveredEvent, + ...(isRootProviderThreadUpdate + ? rootTerminalAlreadySeen + ? { + writeIfProviderThreadOwner: { + providerThreadId: input.providerThread.id, + runId: input.run.id, + activeAttemptId: input.attempt.id, + expectedLastRunOrdinal: input.run.ordinal, + }, + } + : { + writeIfRunCurrent: { + runId: input.run.id, + activeAttemptId: input.attempt.id, + expectedStatus: "running" as const, + }, + } + : {}), + }); + if (isRootProviderThreadUpdate && rootTerminalAlreadySeen && storedEvents.length === 0) { + yield* Ref.set(providerThreadOwnerLost, true); + } + return storedEvents.length; + }); + // The timer only wakes ingestion. The flush runs as an ingestion step outside any provider + // batch, so it stays ordered with routing and snapshots, and a write has committed when + // it returns. + const idleFlushTick = { type: "idle_flush" } as const; + const idleFlushTicks = yield* Queue.dropping(1); + let idleFlushWake: { readonly at: number; readonly fiber: Fiber.Fiber } | undefined; + const scheduleIdleFlush = Effect.suspend(() => { + const at = assistantStreaming.nextIdleFlushAt(); + if ( + idleFlushClosed || + at === null || + (idleFlushWake !== undefined && idleFlushWake.at <= at) + ) + return Effect.void; + const previous = idleFlushWake?.fiber; + return Effect.gen(function* () { + if (previous !== undefined) yield* Fiber.interrupt(previous); + const now = DateTime.toEpochMillis(yield* DateTime.now); + const fiber = yield* Effect.sleep(Duration.millis(Math.max(0, at - now))).pipe( + Effect.andThen(Queue.offer(idleFlushTicks, idleFlushTick)), + Effect.asVoid, + Effect.forkDetach, + ); + idleFlushWake = { at, fiber }; + }); + }); + const closeIdleFlush = Effect.suspend(() => { + idleFlushClosed = true; + const wake = idleFlushWake; + idleFlushWake = undefined; + return wake === undefined ? Effect.void : Fiber.interrupt(wake.fiber); + }); + // A failed idle write keeps its text held and retries with backoff while provider events + // keep ingesting. The provider's next update still writes, and fails ingestion, the normal way. + let idleFlushFailures = 0; + let nextIdleFlushWarningAt = Number.NEGATIVE_INFINITY; + const flushIdleAssistantText = Effect.gen(function* () { + const now = DateTime.toEpochMillis(yield* DateTime.now); + // A wake due by now has fired; a later deadline needs a new one. + if (idleFlushWake !== undefined && idleFlushWake.at <= now) idleFlushWake = undefined; + if (idleFlushClosed) return; + const due = assistantStreaming.dueIdleFlushes(now); + for (const [index, flush] of due.entries()) { + // The held text's child may have been handed to another run since it arrived. + if (!routeProviderEvent(flush.event, routeIdentity, yield* Ref.get(eventRouting))[0]) { + flush.discard(); + continue; + } + const written = yield* Effect.exit(ingestDelivered(flush.event, flush.event)); + if (Exit.isFailure(written)) { + if (Cause.hasInterruptsOnly(written.cause)) return yield* written; + idleFlushFailures += 1; + const failedAt = DateTime.toEpochMillis(yield* DateTime.now); + if (failedAt >= nextIdleFlushWarningAt) { + yield* Effect.logWarning("orchestration V2 idle assistant text flush failed", { + runId: input.run.id, + failures: idleFlushFailures, + cause: written.cause, + }); + nextIdleFlushWarningAt = failedAt + 10 * 60_000; + } + const retryAt = failedAt + Math.min(1_000 * 2 ** (idleFlushFailures - 1), 30_000); + for (const pending of due.slice(index)) pending.retryAt(retryAt); + break; + } + idleFlushFailures = 0; + nextIdleFlushWarningAt = Number.NEGATIVE_INFINITY; + flush.delivered(now); + yield* trackChildLifecycle(flush.event, true); + } + yield* scheduleIdleFlush; + }); const ingestEvent = Effect.fnUntraced(function* ({ event, accepted, @@ -1413,65 +1550,12 @@ export const layer: Layer.Layer< yield* Ref.set(latestRootProviderTurn, event.providerTurn); } let storedEventCount = 0; - const deliveredEvent = filterAssistantEvent( + const deliveredEvent = assistantStreaming.filter( event, DateTime.toEpochMillis(yield* DateTime.now), ); if (deliveredEvent) { - const rootTerminalAlreadySeen = yield* Ref.get(rootTerminalSeen); - const isRootProviderThreadUpdate = - event.type === "provider_thread.updated" && - event.providerThread.id === input.providerThread.id; - const nativeChildTurn = - event.type === "provider_turn.updated" && - event.providerTurn.providerThreadId !== input.providerThread.id; - if (event.type === "provider_turn.updated" && !nativeChildTurn) { - firstRootProviderTurnId ??= event.providerTurn.id; - } - const storedEvents = yield* providerEventIngestor.ingestNormalized({ - analyticsContext: nativeChildTurn - ? { - ...analyticsContext, - origin: NATIVE_SUBAGENT_TURN_ORIGIN, - segment: "native_child", - } - : { - ...analyticsContext, - segment: - event.type === "provider_turn.updated" && - event.providerTurn.id !== firstRootProviderTurnId - ? "continuation" - : input.attempt.reason, - }, - providerSessionId: input.providerSessionId, - providerInstanceId: input.run.providerInstanceId, - threadId: input.run.threadId, - runId: input.run.id, - nodeId: input.rootNode.id, - event: deliveredEvent, - ...(isRootProviderThreadUpdate - ? rootTerminalAlreadySeen - ? { - writeIfProviderThreadOwner: { - providerThreadId: input.providerThread.id, - runId: input.run.id, - activeAttemptId: input.attempt.id, - expectedLastRunOrdinal: input.run.ordinal, - }, - } - : { - writeIfRunCurrent: { - runId: input.run.id, - activeAttemptId: input.attempt.id, - expectedStatus: "running" as const, - }, - } - : {}), - }); - storedEventCount = storedEvents.length; - if (isRootProviderThreadUpdate && rootTerminalAlreadySeen && storedEventCount === 0) { - yield* Ref.set(providerThreadOwnerLost, true); - } + storedEventCount = yield* ingestDelivered(event, deliveredEvent); } else if ( event.type === "turn_item.updated" && (event.turnItem.type === "assistant_message" || event.turnItem.type === "reasoning") && @@ -1511,14 +1595,19 @@ export const layer: Layer.Layer< } } yield* trackChildLifecycle(event, deliveredEvent !== null); + yield* scheduleIdleFlush; }); const providerEventFiber = yield* eventSink .consumeProviderEvents( - eventSubscription.events, + eventSubscription.events.pipe( + Stream.merge(Stream.fromQueue(idleFlushTicks), { haltStrategy: "left" }), + ), (event) => - Effect.gen(function* () { - yield* ingestEvent(yield* routeEvent(event)); - }), + event.type === "idle_flush" + ? flushIdleAssistantText + : Effect.gen(function* () { + yield* ingestEvent(yield* routeEvent(event)); + }), Effect.gen(function* () { if (yield* agentTurnsSettled) yield* endPreviewActivity; return yield* shouldStopProviderEventIngestion; @@ -1526,6 +1615,7 @@ export const layer: Layer.Layer< Ref.get(rootTerminalSeen).pipe(Effect.map((seen) => !seen)), ) .pipe( + Effect.ensuring(closeIdleFlush), Effect.mapError((cause) => new RunExecutionIngestError({ runId: input.run.id, cause })), Effect.flatMap(() => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/assistantStreaming.test.ts b/apps/server/src/orchestration-v2/assistantStreaming.test.ts index d71042312bfb1..08da5bf79eb38 100644 --- a/apps/server/src/orchestration-v2/assistantStreaming.test.ts +++ b/apps/server/src/orchestration-v2/assistantStreaming.test.ts @@ -1,5 +1,6 @@ -import { describe, expect, it } from "vite-plus/test"; -import { splitBufferedAssistantText } from "./assistantStreaming.ts"; +import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import { assert, describe, expect, it } from "vite-plus/test"; +import { makeAssistantStreamingFilter, splitBufferedAssistantText } from "./assistantStreaming.ts"; describe("V2 assistant streaming", () => { it("keeps code fences intact", () => { @@ -24,3 +25,56 @@ it("holds a streamed section heading until its content has a boundary", () => { rest: "Next", }); }); + +it("delivers held text once idle, never after its final, and never in turn mode", () => { + const snapshot = (text: string, streaming = true) => + ({ + type: "turn_item.updated", + turnItem: { id: "item", type: "reasoning", text, streaming }, + }) as unknown as ProviderAdapter.ProviderAdapterV2Event; + const text = (event: ProviderAdapter.ProviderAdapterV2Event | undefined) => + event?.type === "turn_item.updated" && "text" in event.turnItem ? event.turnItem.text : null; + // Flushes everything due and reports what it delivered. + const flush = (filter: ReturnType, now: number) => + filter.dueIdleFlushes(now).map((due) => { + due.delivered(now); + return text(due.event); + }); + + const paragraph = makeAssistantStreamingFilter("paragraph", 800); + assert.equal(text(paragraph.filter(snapshot("A\n\n```ts\nx()"), 0) ?? undefined), "A\n\n"); + // Inside the 400ms throttle even though a boundary is ready. + assert.isNull(paragraph.filter(snapshot("A\n\n```ts\nx()\n```\n\nB"), 100)); + assert.equal(paragraph.nextIdleFlushAt(), 900); + assert.deepEqual(flush(paragraph, 899), []); + // A failed write keeps the snapshot held until its retry time. + paragraph.dueIdleFlushes(900)[0]?.retryAt(1_900); + assert.equal(paragraph.nextIdleFlushAt(), 1_900); + assert.deepEqual(flush(paragraph, 1_899), []); + assert.deepEqual(flush(paragraph, 1_900), ["A\n\n```ts\nx()\n```\n\nB"]); + assert.isNull(paragraph.nextIdleFlushAt()); + + paragraph.filter(snapshot("A\n\n```ts\nx()\n```\n\nB and C"), 2_500); + const superseded = paragraph.dueIdleFlushes(3_300); + assert.isNotNull(paragraph.filter(snapshot("A\n\n```ts\nx()\n```\n\nB and C.", false), 3_400)); + // An outcome for a snapshot the final replaced changes nothing. + superseded[0]?.retryAt(3_500); + assert.isNull(paragraph.nextIdleFlushAt()); + assert.deepEqual(flush(paragraph, 5_000), []); + + // Provider rewrites replace delivered text: shorter at once, same-length after the throttle. + const rewrite = makeAssistantStreamingFilter("paragraph", 800); + assert.equal( + text(rewrite.filter(snapshot("Original long paragraph.\n\n"), 0) ?? undefined), + "Original long paragraph.\n\n", + ); + assert.equal(text(rewrite.filter(snapshot("New.\n\n"), 1_000) ?? undefined), "New.\n\n"); + assert.isNull(rewrite.filter(snapshot("Old.\n\n"), 1_100)); + assert.deepEqual(flush(rewrite, 1_900), ["Old.\n\n"]); + assert.isNull(rewrite.filter(snapshot("Old"), 2_500)); + assert.deepEqual(flush(rewrite, 3_300), ["Old"]); + + const turn = makeAssistantStreamingFilter("turn", 800); + assert.isNull(turn.filter(snapshot("A\n\nB"), 0)); + assert.isNull(turn.nextIdleFlushAt()); +}); diff --git a/apps/server/src/orchestration-v2/assistantStreaming.ts b/apps/server/src/orchestration-v2/assistantStreaming.ts index 62a173d7318c7..59f6aaa47dabf 100644 --- a/apps/server/src/orchestration-v2/assistantStreaming.ts +++ b/apps/server/src/orchestration-v2/assistantStreaming.ts @@ -68,44 +68,138 @@ export function splitBufferedAssistantText(text: string): { ready: string; rest: return { ready: text.slice(0, boundary), rest: text.slice(boundary) }; } -export function makeAssistantStreamingFilter(mode: ResponseStreamingMode) { - const delivered = new Map(); - return ( - event: ProviderAdapter.ProviderAdapterV2Event, - now: number, - ): ProviderAdapter.ProviderAdapterV2Event | null => { - if ( - event.type === "node.updated" && - event.node.kind === "assistant_message" && - event.node.status === "running" - ) - return null; - const message = - event.type === "message.updated" && event.message.role === "assistant" - ? event.message - : event.type === "turn_item.updated" && - (event.turnItem.type === "assistant_message" || event.turnItem.type === "reasoning") - ? event.turnItem - : null; - if (!message) return event; - const key = `${event.type}:${message.id}`; - if (!message.streaming) { - delivered.delete(key); - return event; - } - if (mode === "turn") return null; - const previous = delivered.get(key); - if (previous && now - previous.at < 400) return null; - const { ready } = splitBufferedAssistantText(message.text); - if (!ready || ready === previous?.text) return null; - delivered.set(key, { text: ready, at: now }); - if (event.type === "message.updated") - return { ...event, message: { ...event.message, text: ready } }; - if ( - event.type === "turn_item.updated" && - (event.turnItem.type === "assistant_message" || event.turnItem.type === "reasoning") - ) - return { ...event, turnItem: { ...event.turnItem, text: ready } }; - return event; +// Long enough to ride out normal gaps between chunks; short enough that text written before a +// long tool call or background wait does not look lost. +const ASSISTANT_IDLE_FLUSH_MS = 800; + +type StreamedMessage = { + readonly id: string; + readonly text: string; + readonly streaming: boolean; +}; + +interface StreamTrack { + readonly text: string; + readonly at: number; + // Latest snapshot whose text has not all been delivered yet. + readonly held: ProviderAdapter.ProviderAdapterV2Event | null; + readonly dueAt: number; +} + +/** A held snapshot due for delivery. Each outcome applies only while nothing newer replaced it. */ +interface IdleFlush { + readonly event: ProviderAdapter.ProviderAdapterV2Event; + readonly delivered: (now: number) => void; + readonly retryAt: (at: number) => void; + readonly discard: () => void; +} + +function streamedMessage(event: ProviderAdapter.ProviderAdapterV2Event): StreamedMessage | null { + if (event.type === "message.updated" && event.message.role === "assistant") return event.message; + if ( + event.type === "turn_item.updated" && + (event.turnItem.type === "assistant_message" || event.turnItem.type === "reasoning") + ) + return event.turnItem; + return null; +} + +function withText( + event: ProviderAdapter.ProviderAdapterV2Event, + text: string, +): ProviderAdapter.ProviderAdapterV2Event { + if (event.type === "message.updated") return { ...event, message: { ...event.message, text } }; + if ( + event.type === "turn_item.updated" && + (event.turnItem.type === "assistant_message" || event.turnItem.type === "reasoning") + ) + return { ...event, turnItem: { ...event.turnItem, text } }; + return event; +} + +/** + * Paragraph mode delivers streamed text at Markdown boundaries, at most every 400ms. A snapshot + * held back that way is delivered whole once its message has been idle for `idleFlushMs`, so a + * provider that pauses mid-turn still shows everything it wrote. + */ +export function makeAssistantStreamingFilter( + mode: ResponseStreamingMode, + idleFlushMs = ASSISTANT_IDLE_FLUSH_MS, +) { + const tracks = new Map(); + return { + filter( + event: ProviderAdapter.ProviderAdapterV2Event, + now: number, + ): ProviderAdapter.ProviderAdapterV2Event | null { + if ( + event.type === "node.updated" && + event.node.kind === "assistant_message" && + event.node.status === "running" + ) + return null; + const message = streamedMessage(event); + if (!message) return event; + const key = `${event.type}:${message.id}`; + if (!message.streaming) { + tracks.delete(key); + return event; + } + if (mode === "turn") return null; + const previous = tracks.get(key); + const delivered = previous?.text ?? ""; + const hold = () => { + tracks.set(key, { + text: delivered, + at: previous?.at ?? Number.NEGATIVE_INFINITY, + held: message.text !== delivered ? event : null, + dueAt: now + idleFlushMs, + }); + return null; + }; + if (previous && now - previous.at < 400) return hold(); + const { ready } = splitBufferedAssistantText(message.text); + // An idle delivery may already show more than the next boundary; never shrink it. A + // provider rewrite or truncation is not a prefix and replaces it. + if (delivered.startsWith(ready)) return hold(); + tracks.set(key, { + text: ready, + at: now, + held: ready !== message.text ? event : null, + dueAt: now + idleFlushMs, + }); + return withText(event, ready); + }, + nextIdleFlushAt(): number | null { + let next: number | null = null; + for (const track of tracks.values()) { + if (track.held !== null && (next === null || track.dueAt < next)) next = track.dueAt; + } + return next; + }, + dueIdleFlushes(now: number): ReadonlyArray { + const due: Array = []; + for (const [key, track] of tracks) { + const event = track.held; + if (event === null || track.dueAt > now) continue; + const settle = (next: (current: StreamTrack) => StreamTrack) => { + const current = tracks.get(key); + if (current?.held === event) tracks.set(key, next(current)); + }; + due.push({ + event, + delivered: (at) => + settle((current) => ({ + text: streamedMessage(event)?.text ?? current.text, + at, + held: null, + dueAt: at, + })), + retryAt: (at) => settle((current) => ({ ...current, dueAt: at })), + discard: () => settle((current) => ({ ...current, held: null })), + }); + } + return due; + }, }; } From 2ecbf3eef96386104dfb4b2d8bdee1df60bb9abf Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Fri, 6 Nov 2026 05:39:08 +0000 Subject: [PATCH 02/11] fix(server): keep a failing idle write's backoff when another message's succeeds Co-Authored-By: Claude Opus 5.5 --- .../RunExecutionService.test.ts | 74 +++++++++++++++++++ .../orchestration-v2/RunExecutionService.ts | 23 +++--- 2 files changed, 88 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 459647be6c914..b0fd625d3e9ef 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -34,6 +34,7 @@ import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Logger from "effect/Logger"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; @@ -4341,6 +4342,79 @@ it.effect("reschedules idle text when a newer snapshot precedes a failed write's }), ); +it.effect( + "keeps a failing idle write's backoff and warning limit when another message's succeeds", + () => + Effect.gen(function* () { + const key = "idle-flush-unrelated-success"; + const ids = backgroundScenarioIds(key); + const events = yield* Queue.unbounded(); + const markers = yield* Queue.unbounded(); + const failingAttempts = yield* Queue.unbounded(); + const otherWritten = yield* Deferred.make(); + const recovered = yield* Deferred.make(); + const attempts = yield* Ref.make(0); + let warnings = 0; + const message = (id: string, role: "assistant" | "user", text: string) => + streamedMessageEvent({ + id: `message:${key}:${id}`, + threadId: ids.threadId, + runId: ids.runId, + role, + text, + streaming: role === "assistant", + }); + const marker = message("marker", "user", "marker"); + const readPast = (...items: ReadonlyArray) => + Queue.offerAll(events, [...items, marker]).pipe(Effect.andThen(Queue.take(markers))); + const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { + onIngest: (event) => + Effect.gen(function* () { + if (event.type !== "message.updated") return; + if (event.message.role === "user") return yield* Queue.offer(markers, undefined); + if (event.message.text === "Other partial") + return yield* Deferred.succeed(otherWritten, undefined); + const attempt = yield* Ref.updateAndGet(attempts, (n) => n + 1); + yield* Queue.offer(failingAttempts, attempt); + if (attempt <= 2) return yield* Effect.die(new Error("disk I/O error")); + yield* Deferred.succeed(recovered, undefined); + }), + }).pipe( + Effect.provide( + Logger.layer( + [ + Logger.make(({ logLevel }) => { + if (logLevel === "Warn") warnings += 1; + }), + ], + { mergeWithExisting: false }, + ), + ), + Effect.forkChild, + ); + + yield* readPast(message("failing", "assistant", "Failing partial")); + yield* TestClock.adjust("800 millis"); + assert.equal(yield* Queue.take(failingAttempts), 1); + // Due before the failing write's 1s retry. + yield* readPast(message("other", "assistant", "Other partial")); + yield* TestClock.adjust("800 millis"); + yield* Deferred.await(otherWritten); + yield* TestClock.adjust("200 millis"); + assert.equal(yield* Queue.take(failingAttempts), 2); + // A second failure backs off 2s; the other message's success must not restart at 1s. + yield* TestClock.adjust("1 second"); + yield* readPast(); + assert.equal(yield* Ref.get(attempts), 2); + assert.equal(warnings, 1); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(recovered); + + yield* Queue.offer(events, rootTerminalEvent(ids, "completed")); + yield* Fiber.join(scenario); + }), +); + function streamedMessageEvent(input: { readonly id: string; readonly threadId: ThreadId; diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 5c0ac4e248747..5b660c0c86d50 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1496,8 +1496,12 @@ export const layer: Layer.Layer< }); // A failed idle write keeps its text held and retries with backoff while provider events // keep ingesting. The provider's next update still writes, and fails ingestion, the normal way. - let idleFlushFailures = 0; - let nextIdleFlushWarningAt = Number.NEGATIVE_INFINITY; + // Keyed by held snapshot: another message's success keeps this backoff and warning limit, + // while a newer snapshot or a delivery leaves the old entry unreachable. + const idleFlushFailures = new WeakMap< + ProviderAdapter.ProviderAdapterV2Event, + { readonly count: number; readonly nextWarningAt: number } + >(); const flushIdleAssistantText = Effect.gen(function* () { const now = DateTime.toEpochMillis(yield* DateTime.now); // A wake due by now has fired; a later deadline needs a new one. @@ -1513,22 +1517,23 @@ export const layer: Layer.Layer< const written = yield* Effect.exit(ingestDelivered(flush.event, flush.event)); if (Exit.isFailure(written)) { if (Cause.hasInterruptsOnly(written.cause)) return yield* written; - idleFlushFailures += 1; + const previous = idleFlushFailures.get(flush.event); + const failures = (previous?.count ?? 0) + 1; const failedAt = DateTime.toEpochMillis(yield* DateTime.now); - if (failedAt >= nextIdleFlushWarningAt) { + let nextWarningAt = previous?.nextWarningAt ?? Number.NEGATIVE_INFINITY; + if (failedAt >= nextWarningAt) { yield* Effect.logWarning("orchestration V2 idle assistant text flush failed", { runId: input.run.id, - failures: idleFlushFailures, + failures, cause: written.cause, }); - nextIdleFlushWarningAt = failedAt + 10 * 60_000; + nextWarningAt = failedAt + 10 * 60_000; } - const retryAt = failedAt + Math.min(1_000 * 2 ** (idleFlushFailures - 1), 30_000); + idleFlushFailures.set(flush.event, { count: failures, nextWarningAt }); + const retryAt = failedAt + Math.min(1_000 * 2 ** (failures - 1), 30_000); for (const pending of due.slice(index)) pending.retryAt(retryAt); break; } - idleFlushFailures = 0; - nextIdleFlushWarningAt = Number.NEGATIVE_INFINITY; flush.delivered(now); yield* trackChildLifecycle(flush.event, true); } From 67b87c30ab9bd96f4911d68956a13bba9160e04c Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Fri, 6 Nov 2026 05:46:48 +0000 Subject: [PATCH 03/11] test(server): prove a failed idle COMMIT neither persists nor publishes held text Co-Authored-By: Claude Opus 5.5 --- .../RunExecutionService.test.ts | 161 +++++++++++++++--- 1 file changed, 136 insertions(+), 25 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index b0fd625d3e9ef..129c9248d5afc 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -4,6 +4,7 @@ import { ChatAttachmentId, ChatFileAttachment, CommandId, + EventId, MessageId, NodeId, type OrchestrationV2AppThread, @@ -39,15 +40,18 @@ import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/sql/SqlClient"; import * as TestClock from "effect/testing/TestClock"; import { CodexAppServerRequestError } from "effect-codex-app-server/errors"; -import { SqlError, classifySqliteError } from "effect/sql/SqlError"; +import { SqlError, classifySqliteError, isSqlError } from "effect/sql/SqlError"; import { toPersistenceSqlError } from "../persistence/Errors.ts"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as CheckpointService from "./CheckpointService.ts"; import type { PendingOrchestrationEffectV2 } from "./EffectOutbox.ts"; import * as EventSink from "./EventSink.ts"; +import * as EventStore from "./EventStore.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProjectionStore from "./ProjectionStore.ts"; @@ -3709,6 +3713,8 @@ function runBackgroundItemScenario( ) => Effect.Effect; // Consumes through the real commit coordinator, as the event sink does. readonly batched?: boolean; + // Consumes through this sink instead; `onIngest` makes its own writes. + readonly eventSink?: EventSink.EventSinkV2["Service"]; readonly startTurn?: Effect.Effect; readonly keepEventStreamOpen?: boolean; readonly hasPendingBackgroundWorkForThread?: ( @@ -3764,6 +3770,9 @@ function runBackgroundItemScenario( Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), layerTurnItemPositionsTest, mockEventSink({ + ...(options?.eventSink === undefined + ? {} + : { consumeProviderEvents: options.eventSink.consumeProviderEvents }), ...(coordinator === undefined ? {} : { consumeProviderEvents: coordinator.consume }), write: () => Effect.succeed([]), writeWithEffects: (input) => @@ -4184,16 +4193,89 @@ it.effect.each([ }), ); +const layerPersistedEventSink = EventSink.layer.pipe( + Layer.provideMerge(Layer.merge(EventStore.layer, ProjectionStore.layer)), + Layer.provideMerge(SqlitePersistence.layerMemory), +); + +/** Writes turn items through a real event sink and SQLite, failing each COMMIT until restored. */ +const persistTurnItemsWithFailingCommit = Effect.fn(function* ( + sentinel: OrchestrationV2TurnItem, + onPublished: (turnItem: OrchestrationV2TurnItem) => Effect.Effect, +) { + const eventSink = yield* EventSink.EventSinkV2; + const sql = yield* SqlClient.SqlClient; + let written = 0; + const write = (turnItem: OrchestrationV2TurnItem) => + eventSink + .write({ + events: [ + { + id: EventId.make(`event:commit-fault:${written++}`), + type: "turn-item.updated", + threadId: turnItem.threadId, + ...(turnItem.runId === null ? {} : { runId: turnItem.runId }), + occurredAt: DateTime.makeUnsafe("2026-07-21T12:00:00.000Z"), + payload: turnItem, + }, + ], + }) + .pipe( + // Only a failed COMMIT leaves its transaction open here. A full disk's ends it, so end + // this one the same way; the ROLLBACK fails if the write failed before COMMIT. + Effect.tapError(() => sql`ROLLBACK`.pipe(Effect.orDie)), + ); + const subscribed = yield* Deferred.make(); + yield* eventSink.stream({ eventType: "turn-item.updated" }).pipe( + Stream.runForEach(({ event }) => + event.type !== "turn-item.updated" + ? Effect.void + : event.payload.id === sentinel.id + ? Deferred.succeed(subscribed, undefined) + : onPublished(event.payload), + ), + Effect.forkChild, + ); + yield* write(sentinel); + yield* Deferred.await(subscribed); + // Every write statement succeeds; the deferred dangling reference fails only the COMMIT. + yield* sql`CREATE TABLE commit_fault_parent (id TEXT PRIMARY KEY)`; + yield* sql`CREATE TABLE commit_fault_child ( + parent_id TEXT REFERENCES commit_fault_parent (id) DEFERRABLE INITIALLY DEFERRED + )`; + yield* sql`CREATE TABLE commit_fault (enabled INTEGER NOT NULL)`; + yield* sql`INSERT INTO commit_fault VALUES (1)`; + yield* sql`CREATE TRIGGER commit_fault AFTER INSERT ON orchestration_events + WHEN (SELECT enabled FROM commit_fault) = 1 + BEGIN INSERT INTO commit_fault_child VALUES ('missing'); END`; + return { + eventSink, + write, + restore: sql`UPDATE commit_fault SET enabled = 0`.pipe(Effect.orDie), + persistedTexts: (itemId: TurnItemId) => + sql<{ readonly text: string }>` + SELECT json_extract(payload_json, '$.text') AS text FROM orchestration_events + WHERE event_type = 'turn-item.updated' AND json_extract(payload_json, '$.id') = ${itemId} + ORDER BY sequence + `.pipe( + Effect.map((rows) => rows.map((row) => row.text)), + Effect.orDie, + ), + }; +}); + it.effect.each([ - { name: "before", afterRoot: false, batched: false }, - { name: "after", afterRoot: true, batched: false }, - { name: "before", afterRoot: false, batched: true }, - { name: "after", afterRoot: true, batched: true }, + { name: "before", afterRoot: false, batched: false, persisted: false }, + { name: "after", afterRoot: true, batched: false, persisted: false }, + { name: "before", afterRoot: false, batched: true, persisted: false }, + { name: "after", afterRoot: true, batched: true, persisted: false }, + // The idle write's body runs and only its COMMIT fails. + { name: "before", afterRoot: false, batched: true, persisted: true }, ])( - "retries a failed idle flush $name the root completes and keeps the run live (batched: $batched)", - ({ afterRoot, batched }) => + "retries a failed idle flush $name the root completes and keeps the run live (batched: $batched, persisted: $persisted)", + ({ afterRoot, batched, persisted }) => Effect.gen(function* () { - const key = `idle-flush-retry-${afterRoot ? "after" : "before"}-root${batched ? "-batched" : ""}`; + const key = `idle-flush-retry-${afterRoot ? "after" : "before"}-root${batched ? "-batched" : ""}${persisted ? "-persisted" : ""}`; const ids = backgroundScenarioIds(key); const events = yield* Queue.unbounded(); const subagentIngested = yield* Deferred.make(); @@ -4202,10 +4284,30 @@ it.effect.each([ const failureHandled = yield* Deferred.make(); const retried = yield* Deferred.make(); const attempts = yield* Ref.make(0); + // What subscribers saw when persisted; otherwise what the write received. const childWrites = yield* Ref.make>([]); const interrupts = yield* Ref.make(0); + const recordChildWrite = Effect.fnUntraced(function* (turnItem: OrchestrationV2TurnItem) { + if (turnItem.type !== "assistant_message") return; + const { text, streaming } = turnItem; + yield* Ref.update(childWrites, (current) => [ + ...current, + `${streaming ? "streaming" : "final"}:${text}`, + ]); + if (text === heldChildText) yield* Deferred.succeed(retried, undefined); + }); + const snapshot = childAssistantSnapshot(ids); + assert(snapshot.type === "turn_item.updated"); + const persistence = persisted + ? yield* persistTurnItemsWithFailingCommit( + { ...snapshot.turnItem, id: TurnItemId.make(`${ids.childItemId}:sentinel`) }, + recordChildWrite, + ).pipe(Effect.provideContext(yield* Layer.build(layerPersistedEventSink))) + : undefined; + const commitFailure = yield* Ref.make(null); const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { - batched, + batched: batched && persistence === undefined, + ...(persistence === undefined ? {} : { eventSink: persistence.eventSink }), onIngest: (event) => Effect.gen(function* () { if (event.type === "subagent.updated") { @@ -4222,14 +4324,16 @@ it.effect.each([ ) { yield* Deferred.succeed(firstWriteFailed, undefined); yield* Effect.sleep("2 seconds"); - return yield* Effect.die(new Error("disk I/O error")); + if (persistence === undefined) return yield* Effect.die(new Error("disk I/O error")); + return yield* persistence.write(event.turnItem).pipe( + Effect.tapError((error) => + Ref.set(commitFailure, isSqlError(error.cause) ? error.cause.reason._tag : null), + ), + Effect.orDie, + ); } - const { text, streaming } = event.turnItem; - yield* Ref.update(childWrites, (current) => [ - ...current, - `${streaming ? "streaming" : "final"}:${text}`, - ]); - if (isHeldChildSnapshot(event)) yield* Deferred.succeed(retried, undefined); + if (persistence === undefined) return yield* recordChildWrite(event.turnItem); + yield* persistence.write(event.turnItem).pipe(Effect.orDie); }), onFinalized: () => Deferred.succeed(rootFinalized, undefined), onInterruptTurn: () => Ref.update(interrupts, (n) => n + 1), @@ -4247,17 +4351,18 @@ it.effect.each([ yield* Deferred.await(failureHandled); assert.equal(yield* Ref.get(attempts), 1); assert.deepEqual(yield* Ref.get(childWrites), []); + if (persistence !== undefined) { + assert.equal(yield* Ref.get(commitFailure), "ConstraintError"); + assert.deepEqual(yield* persistence.persistedTexts(ids.childItemId), []); + yield* persistence.restore; + } yield* TestClock.adjust("1 second"); yield* Deferred.await(retried); - const childText = (text: string, streaming: boolean) => { - const snapshot = childAssistantSnapshot(ids); - return snapshot.type === "turn_item.updated" - ? ({ - ...snapshot, - turnItem: { ...snapshot.turnItem, text, streaming }, - } as ProviderAdapter.ProviderAdapterV2Event) - : snapshot; - }; + const childText = (text: string, streaming: boolean) => + ({ + ...snapshot, + turnItem: { ...snapshot.turnItem, text, streaming }, + }) as ProviderAdapter.ProviderAdapterV2Event; yield* Queue.offerAll(events, [ childText(`${heldChildText} and more`, true), childText(`${heldChildText} and more.`, false), @@ -4270,6 +4375,12 @@ it.effect.each([ `streaming:${heldChildText}`, `final:${heldChildText} and more.`, ]); + if (persistence !== undefined) { + assert.deepEqual(yield* persistence.persistedTexts(ids.childItemId), [ + heldChildText, + `${heldChildText} and more.`, + ]); + } assert.equal(yield* Ref.get(interrupts), 0); assert.include(observed, "root-finalized"); assert.equal(observed.at(-1), "subagent:completed"); From c2eba7006f04559a2c998b307a1e84ae16f080ba Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Fri, 6 Nov 2026 05:50:18 +0000 Subject: [PATCH 04/11] fix(server): cap idle storage warnings across the run --- .../RunExecutionService.test.ts | 145 ++++++++++-------- .../orchestration-v2/RunExecutionService.ts | 19 +-- 2 files changed, 84 insertions(+), 80 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 129c9248d5afc..b769e283868be 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -4453,77 +4453,86 @@ it.effect("reschedules idle text when a newer snapshot precedes a failed write's }), ); -it.effect( - "keeps a failing idle write's backoff and warning limit when another message's succeeds", - () => - Effect.gen(function* () { - const key = "idle-flush-unrelated-success"; - const ids = backgroundScenarioIds(key); - const events = yield* Queue.unbounded(); - const markers = yield* Queue.unbounded(); - const failingAttempts = yield* Queue.unbounded(); - const otherWritten = yield* Deferred.make(); - const recovered = yield* Deferred.make(); - const attempts = yield* Ref.make(0); - let warnings = 0; - const message = (id: string, role: "assistant" | "user", text: string) => - streamedMessageEvent({ - id: `message:${key}:${id}`, - threadId: ids.threadId, - runId: ids.runId, - role, - text, - streaming: role === "assistant", - }); - const marker = message("marker", "user", "marker"); - const readPast = (...items: ReadonlyArray) => - Queue.offerAll(events, [...items, marker]).pipe(Effect.andThen(Queue.take(markers))); - const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { - onIngest: (event) => - Effect.gen(function* () { - if (event.type !== "message.updated") return; - if (event.message.role === "user") return yield* Queue.offer(markers, undefined); - if (event.message.text === "Other partial") - return yield* Deferred.succeed(otherWritten, undefined); - const attempt = yield* Ref.updateAndGet(attempts, (n) => n + 1); - yield* Queue.offer(failingAttempts, attempt); - if (attempt <= 2) return yield* Effect.die(new Error("disk I/O error")); - yield* Deferred.succeed(recovered, undefined); - }), - }).pipe( - Effect.provide( - Logger.layer( - [ - Logger.make(({ logLevel }) => { - if (logLevel === "Warn") warnings += 1; - }), - ], - { mergeWithExisting: false }, - ), +it.effect("keeps a failing idle write's backoff and caps warnings across other messages", () => + Effect.gen(function* () { + const key = "idle-flush-unrelated-success"; + const ids = backgroundScenarioIds(key); + const events = yield* Queue.unbounded(); + const markers = yield* Queue.unbounded(); + const failingAttempts = yield* Queue.unbounded(); + const otherWritten = yield* Deferred.make(); + const recovered = yield* Deferred.make(); + const anotherFailed = yield* Deferred.make(); + const attempts = yield* Ref.make(0); + let warnings = 0; + const message = (id: string, role: "assistant" | "user", text: string) => + streamedMessageEvent({ + id: `message:${key}:${id}`, + threadId: ids.threadId, + runId: ids.runId, + role, + text, + streaming: role === "assistant", + }); + const marker = message("marker", "user", "marker"); + const readPast = (...items: ReadonlyArray) => + Queue.offerAll(events, [...items, marker]).pipe(Effect.andThen(Queue.take(markers))); + const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { + onIngest: (event) => + Effect.gen(function* () { + if (event.type !== "message.updated") return; + if (event.message.role === "user") return yield* Queue.offer(markers, undefined); + if (event.message.text === "Other partial") + return yield* Deferred.succeed(otherWritten, undefined); + if (event.message.text === "Another failing partial") { + yield* Deferred.succeed(anotherFailed, undefined); + return yield* Effect.die(new Error("disk I/O error")); + } + const attempt = yield* Ref.updateAndGet(attempts, (n) => n + 1); + yield* Queue.offer(failingAttempts, attempt); + if (attempt <= 2) return yield* Effect.die(new Error("disk I/O error")); + yield* Deferred.succeed(recovered, undefined); + }), + }).pipe( + Effect.provide( + Logger.layer( + [ + Logger.make(({ logLevel }) => { + if (logLevel === "Warn") warnings += 1; + }), + ], + { mergeWithExisting: false }, ), - Effect.forkChild, - ); + ), + Effect.forkChild, + ); - yield* readPast(message("failing", "assistant", "Failing partial")); - yield* TestClock.adjust("800 millis"); - assert.equal(yield* Queue.take(failingAttempts), 1); - // Due before the failing write's 1s retry. - yield* readPast(message("other", "assistant", "Other partial")); - yield* TestClock.adjust("800 millis"); - yield* Deferred.await(otherWritten); - yield* TestClock.adjust("200 millis"); - assert.equal(yield* Queue.take(failingAttempts), 2); - // A second failure backs off 2s; the other message's success must not restart at 1s. - yield* TestClock.adjust("1 second"); - yield* readPast(); - assert.equal(yield* Ref.get(attempts), 2); - assert.equal(warnings, 1); - yield* TestClock.adjust("1 second"); - yield* Deferred.await(recovered); + yield* readPast(message("failing", "assistant", "Failing partial")); + yield* TestClock.adjust("800 millis"); + assert.equal(yield* Queue.take(failingAttempts), 1); + // Due before the failing write's 1s retry. + yield* readPast(message("other", "assistant", "Other partial")); + yield* TestClock.adjust("800 millis"); + yield* Deferred.await(otherWritten); + yield* TestClock.adjust("200 millis"); + assert.equal(yield* Queue.take(failingAttempts), 2); + // A second failure backs off 2s; the other message's success must not restart at 1s. + yield* TestClock.adjust("1 second"); + yield* readPast(); + assert.equal(yield* Ref.get(attempts), 2); + assert.equal(warnings, 1); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(recovered); - yield* Queue.offer(events, rootTerminalEvent(ids, "completed")); - yield* Fiber.join(scenario); - }), + yield* readPast(message("another", "assistant", "Another failing partial")); + yield* TestClock.adjust("800 millis"); + yield* Deferred.await(anotherFailed); + yield* readPast(); + assert.equal(warnings, 1); + + yield* Queue.offer(events, rootTerminalEvent(ids, "completed")); + yield* Fiber.join(scenario); + }), ); function streamedMessageEvent(input: { diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 5b660c0c86d50..aff32448d6324 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1496,12 +1496,9 @@ export const layer: Layer.Layer< }); // A failed idle write keeps its text held and retries with backoff while provider events // keep ingesting. The provider's next update still writes, and fails ingestion, the normal way. - // Keyed by held snapshot: another message's success keeps this backoff and warning limit, - // while a newer snapshot or a delivery leaves the old entry unreachable. - const idleFlushFailures = new WeakMap< - ProviderAdapter.ProviderAdapterV2Event, - { readonly count: number; readonly nextWarningAt: number } - >(); + // Backoff belongs to each held snapshot; the warning cap covers the whole run. + const idleFlushFailures = new WeakMap(); + let nextIdleFlushWarningAt = Number.NEGATIVE_INFINITY; const flushIdleAssistantText = Effect.gen(function* () { const now = DateTime.toEpochMillis(yield* DateTime.now); // A wake due by now has fired; a later deadline needs a new one. @@ -1517,19 +1514,17 @@ export const layer: Layer.Layer< const written = yield* Effect.exit(ingestDelivered(flush.event, flush.event)); if (Exit.isFailure(written)) { if (Cause.hasInterruptsOnly(written.cause)) return yield* written; - const previous = idleFlushFailures.get(flush.event); - const failures = (previous?.count ?? 0) + 1; + const failures = (idleFlushFailures.get(flush.event) ?? 0) + 1; const failedAt = DateTime.toEpochMillis(yield* DateTime.now); - let nextWarningAt = previous?.nextWarningAt ?? Number.NEGATIVE_INFINITY; - if (failedAt >= nextWarningAt) { + if (failedAt >= nextIdleFlushWarningAt) { yield* Effect.logWarning("orchestration V2 idle assistant text flush failed", { runId: input.run.id, failures, cause: written.cause, }); - nextWarningAt = failedAt + 10 * 60_000; + nextIdleFlushWarningAt = failedAt + 10 * 60_000; } - idleFlushFailures.set(flush.event, { count: failures, nextWarningAt }); + idleFlushFailures.set(flush.event, failures); const retryAt = failedAt + Math.min(1_000 * 2 ** (failures - 1), 30_000); for (const pending of due.slice(index)) pending.retryAt(retryAt); break; From 9a3ac59da77de24374422a5d834d3bafd891edd5 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Fri, 6 Nov 2026 06:25:08 +0000 Subject: [PATCH 05/11] fix(server): keep nested background ingestion open until the root snapshot settles The idle-flush merge pulls provider events ahead of the consumer, so OpenCode's live pending-work probe could report false while a consumed root snapshot still said work remained, ending ingestion before the middle child's continuation text arrived. The stop check now trusts the last stored root snapshot's pendingBackgroundWork before the live probe. Adapters that omit the flag keep the previous behavior. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/RunExecutionService.ts | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index aff32448d6324..4b3b0e8a942bd 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1074,6 +1074,8 @@ export const layer: Layer.Layer< ReadonlySet >(new Set(inheritedBackgroundTurnItemsById.keys())); const openRunOwnedSubagents = yield* Ref.make(emptyOpenRunOwnedSubagentProjection()); + // The last stored root snapshot's pendingBackgroundWork; adapters that omit it leave it false. + let rootSnapshotPendingWork = false; const assistantStreaming = makeAssistantStreamingFilter(responseStreamingMode); // Set when ingestion ends or finalization settles the run's open items, which a later // streaming snapshot must not reopen. @@ -1307,6 +1309,10 @@ export const layer: Layer.Layer< if (yield* Ref.get(providerThreadOwnerLost)) { return true; } + // The live probe reads adapter state, which can be ahead of the events consumed here. + if (rootSnapshotPendingWork) { + return false; + } // Claude background Bash has no turn-item projection. Use only the // thread-scoped probe: session-wide pending work (siblings, wake // buffers, session subagents) must not pin this root subscription. @@ -1572,6 +1578,7 @@ export const layer: Layer.Layer< if (event.type === "provider_thread.updated") { if (event.providerThread.id === input.providerThread.id && storedEventCount > 0) { yield* Ref.set(latestProviderThread, event.providerThread); + rootSnapshotPendingWork = event.pendingBackgroundWork === true; } } if ( From 079b9a2a3cf843535d60e704cef691ad8b571e7f Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Sat, 7 Nov 2026 18:45:36 +0000 Subject: [PATCH 06/11] fix(server): drop idle flush retries and trim its tests A failed idle write now fails ingestion like any provider write, and the earliest idle deadline is cached instead of scanned on every event. Co-Authored-By: Claude Opus 5.5 --- .../RunExecutionService.test.ts | 655 ++++-------------- .../orchestration-v2/RunExecutionService.ts | 35 +- .../assistantStreaming.test.ts | 38 +- .../orchestration-v2/assistantStreaming.ts | 65 +- 4 files changed, 176 insertions(+), 617 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index b769e283868be..be3562920b485 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -4,7 +4,6 @@ import { ChatAttachmentId, ChatFileAttachment, CommandId, - EventId, MessageId, NodeId, type OrchestrationV2AppThread, @@ -35,23 +34,19 @@ import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; -import * as Logger from "effect/Logger"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; -import * as SqlClient from "effect/sql/SqlClient"; import * as TestClock from "effect/testing/TestClock"; import { CodexAppServerRequestError } from "effect-codex-app-server/errors"; -import { SqlError, classifySqliteError, isSqlError } from "effect/sql/SqlError"; +import { SqlError, classifySqliteError } from "effect/sql/SqlError"; import { toPersistenceSqlError } from "../persistence/Errors.ts"; -import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as CheckpointService from "./CheckpointService.ts"; import type { PendingOrchestrationEffectV2 } from "./EffectOutbox.ts"; import * as EventSink from "./EventSink.ts"; -import * as EventStore from "./EventStore.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProjectionStore from "./ProjectionStore.ts"; @@ -1894,6 +1889,40 @@ it.effect("keeps ingesting until queued background work clears after the probe g }), ); +it.effect("keeps ingesting while the last stored root snapshot reports pending work", () => + Effect.gen(function* () { + // The live probe is already idle, as when it reads adapter state ahead of the queued events. + const consumed = yield* Deferred.make(); + const result = yield* Effect.raceFirst( + runBackgroundItemScenario( + "root-snapshot-pending-work", + (ids) => { + const snapshot = (pendingBackgroundWork: boolean) => + ({ + type: "provider_thread.updated", + driver, + providerThread: { id: ids.providerThreadId, appThreadId: ids.threadId }, + pendingBackgroundWork, + }) as ProviderAdapter.ProviderAdapterV2Event; + return [ + snapshot(true), + rootTerminalEvent(ids, "completed"), + backgroundTurnItemEvent(ids, "dynamic_tool", "completed", 1), + snapshot(false), + ]; + }, + { + keepEventStreamOpen: true, + hasPendingBackgroundWorkForThread: () => Effect.succeed(false), + onEventsConsumed: Deferred.succeed(consumed, undefined).pipe(Effect.asVoid), + }, + ).pipe(Effect.asSome), + Deferred.await(consumed).pipe(Effect.as(Option.none())), + ); + assert.deepEqual(result, Option.some(["root-finalized", "turn_item:completed"])); + }), +); + it.effect("keeps ingesting a child thread's late background item completion", () => Effect.gen(function* () { const observed = yield* runBackgroundItemScenario("bg-child-item", (ids) => [ @@ -3713,9 +3742,6 @@ function runBackgroundItemScenario( ) => Effect.Effect; // Consumes through the real commit coordinator, as the event sink does. readonly batched?: boolean; - // Consumes through this sink instead; `onIngest` makes its own writes. - readonly eventSink?: EventSink.EventSinkV2["Service"]; - readonly startTurn?: Effect.Effect; readonly keepEventStreamOpen?: boolean; readonly hasPendingBackgroundWorkForThread?: ( providerThread: OrchestrationV2ProviderThread, @@ -3731,9 +3757,6 @@ function runBackgroundItemScenario( readonly onFinalized?: ( events: ReadonlyArray, ) => Effect.Effect; - readonly onInterruptTurn?: (input: { - readonly requestRuntimeRestart?: boolean; - }) => Effect.Effect; }, ) { return Effect.gen(function* () { @@ -3770,9 +3793,6 @@ function runBackgroundItemScenario( Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), layerTurnItemPositionsTest, mockEventSink({ - ...(options?.eventSink === undefined - ? {} - : { consumeProviderEvents: options.eventSink.consumeProviderEvents }), ...(coordinator === undefined ? {} : { consumeProviderEvents: coordinator.consume }), write: () => Effect.succeed([]), writeWithEffects: (input) => @@ -3807,7 +3827,10 @@ function runBackgroundItemScenario( `subagent:${event.subagent.status}`, ]); } - return []; + // Only the count of stored events matters to the run. + return event.type === "provider_thread.updated" + ? [{} as OrchestrationV2DomainEvent] + : []; }); // A write outside a batch waits for queued snapshots, as the event sink's does. return coordinator === undefined ? write : coordinator.barrier(write); @@ -3850,9 +3873,7 @@ function runBackgroundItemScenario( close: Deferred.succeed(ingestionDone, undefined), }; }), - startTurn: () => options?.startTurn ?? Effect.void, - interruptTurn: (interrupt: { readonly requestRuntimeRestart?: boolean }) => - options?.onInterruptTurn?.(interrupt) ?? Effect.void, + startTurn: () => Effect.void, } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime, run: { id: ids.runId, @@ -3956,106 +3977,96 @@ it.effect("keeps a failed root's still-running native children live while the ru }), ); -it.effect.each([ - { name: "", batched: false }, - { name: " through batched commits", batched: true }, -])( - "delivers a held paragraph after the provider goes idle with its prompt still open$name", - ({ batched }) => - Effect.gen(function* () { - const key = `idle-paragraph${batched ? "-batched" : ""}`; - const ids = backgroundScenarioIds(key); - const now = yield* DateTime.now; - const providerEvents = yield* Queue.unbounded(); - const delivered = yield* Ref.make>([]); - const paragraphDelivered = yield* Deferred.make(); - const idleDelivered = yield* Deferred.make(); - const assistantText = (text: string, streaming: boolean) => - [ - { - type: "message.updated", - driver, - message: { - createdBy: "agent", - creationSource: "provider", - id: MessageId.make(`message:${key}`), - threadId: ids.threadId, - runId: ids.runId, - nodeId: ids.rootNodeId, - role: "assistant", - text, - attachments: [], - streaming, - createdAt: now, - updatedAt: now, - }, - }, - { - type: "turn_item.updated", - driver, - turnItem: { - ...makeLinkedChildTurnItemFixture({ ids, driver, type: "assistant_message" }), - id: ids.itemId, - threadId: ids.threadId, - runId: ids.runId, - nodeId: ids.rootNodeId, - providerTurnId: ids.rootProviderTurnId, - text, - streaming, - }, +it.effect("delivers a held paragraph after the provider goes idle with its prompt still open", () => + Effect.gen(function* () { + const key = "idle-paragraph"; + const ids = backgroundScenarioIds(key); + const now = yield* DateTime.now; + const providerEvents = yield* Queue.unbounded(); + const delivered = yield* Ref.make>([]); + const paragraphDelivered = yield* Deferred.make(); + const idleDelivered = yield* Deferred.make(); + const assistantText = (text: string, streaming: boolean) => + [ + { + type: "message.updated", + driver, + message: { + createdBy: "agent", + creationSource: "provider", + id: MessageId.make(`message:${key}`), + threadId: ids.threadId, + runId: ids.runId, + nodeId: ids.rootNodeId, + role: "assistant", + text, + attachments: [], + streaming, + createdAt: now, + updatedAt: now, }, - ] as ReadonlyArray; - - const scenario = yield* runBackgroundItemScenario( - key, - () => Stream.fromQueue(providerEvents), + }, { - batched, - onIngest: (event, write) => - event.type === "message.updated" || event.type === "turn_item.updated" - ? Effect.gen(function* () { - const message = event.type === "message.updated" ? event.message : event.turnItem; - const text = "text" in message ? message.text : ""; - const streaming = "streaming" in message && message.streaming; - yield* Ref.update(delivered, (current) => [ - ...current, - `${event.type}:${streaming ? "streaming" : "final"}:${JSON.stringify(text)}${ - write.inSharedCommit ? " in commit" : "" - }`, - ]); - if (event.type === "turn_item.updated") { - yield* Deferred.succeed(paragraphDelivered, undefined); - if (streaming && text === "First.\n\nWaiting on the build") - yield* Deferred.succeed(idleDelivered, undefined); - } - }) - : Effect.void, + type: "turn_item.updated", + driver, + turnItem: { + ...makeLinkedChildTurnItemFixture({ ids, driver, type: "assistant_message" }), + id: ids.itemId, + threadId: ids.threadId, + runId: ids.runId, + nodeId: ids.rootNodeId, + providerTurnId: ids.rootProviderTurnId, + text, + streaming, + }, }, - ).pipe(Effect.forkChild); - - yield* Queue.offerAll(providerEvents, assistantText("First.\n\nWaiting on the build", true)); - yield* Deferred.await(paragraphDelivered); - // The provider keeps its prompt open while a backgrounded command runs. - yield* TestClock.adjust("1 second"); - yield* Deferred.await(idleDelivered); - yield* Queue.offerAll(providerEvents, [ - ...assistantText("First.\n\nWaiting on the build to", true), - ...assistantText("First.\n\nWaiting on the build to finish.", false), - rootTerminalEvent(ids, "completed"), - ]); - yield* Fiber.join(scenario); - - // Snapshots share the batch commit; the idle write commits alone, before it counts as shown. - const committed = batched ? " in commit" : ""; - assert.deepEqual(yield* Ref.get(delivered), [ - `message.updated:streaming:"First.\\n\\n"${committed}`, - `turn_item.updated:streaming:"First.\\n\\n"${committed}`, - 'message.updated:streaming:"First.\\n\\nWaiting on the build"', - 'turn_item.updated:streaming:"First.\\n\\nWaiting on the build"', - `message.updated:final:"First.\\n\\nWaiting on the build to finish."${committed}`, - `turn_item.updated:final:"First.\\n\\nWaiting on the build to finish."${committed}`, - ]); - }), + ] as ReadonlyArray; + + const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(providerEvents), { + batched: true, + onIngest: (event, write) => + event.type === "message.updated" || event.type === "turn_item.updated" + ? Effect.gen(function* () { + const message = event.type === "message.updated" ? event.message : event.turnItem; + const text = "text" in message ? message.text : ""; + const streaming = "streaming" in message && message.streaming; + yield* Ref.update(delivered, (current) => [ + ...current, + `${event.type}:${streaming ? "streaming" : "final"}:${JSON.stringify(text)}${ + write.inSharedCommit ? " in commit" : "" + }`, + ]); + if (event.type === "turn_item.updated") { + yield* Deferred.succeed(paragraphDelivered, undefined); + if (streaming && text === "First.\n\nWaiting on the build") + yield* Deferred.succeed(idleDelivered, undefined); + } + }) + : Effect.void, + }).pipe(Effect.forkChild); + + yield* Queue.offerAll(providerEvents, assistantText("First.\n\nWaiting on the build", true)); + yield* Deferred.await(paragraphDelivered); + // The provider keeps its prompt open while a backgrounded command runs. + yield* TestClock.adjust("1 second"); + yield* Deferred.await(idleDelivered); + yield* Queue.offerAll(providerEvents, [ + ...assistantText("First.\n\nWaiting on the build to", true), + ...assistantText("First.\n\nWaiting on the build to finish.", false), + rootTerminalEvent(ids, "completed"), + ]); + yield* Fiber.join(scenario); + + // Snapshots share the batch commit; the idle write commits alone, before it counts as shown. + assert.deepEqual(yield* Ref.get(delivered), [ + `message.updated:streaming:"First.\\n\\n" in commit`, + `turn_item.updated:streaming:"First.\\n\\n" in commit`, + 'message.updated:streaming:"First.\\n\\nWaiting on the build"', + 'turn_item.updated:streaming:"First.\\n\\nWaiting on the build"', + `message.updated:final:"First.\\n\\nWaiting on the build to finish." in commit`, + `turn_item.updated:final:"First.\\n\\nWaiting on the build to finish." in commit`, + ]); + }), ); const heldChildText = "partial response"; @@ -4193,348 +4204,6 @@ it.effect.each([ }), ); -const layerPersistedEventSink = EventSink.layer.pipe( - Layer.provideMerge(Layer.merge(EventStore.layer, ProjectionStore.layer)), - Layer.provideMerge(SqlitePersistence.layerMemory), -); - -/** Writes turn items through a real event sink and SQLite, failing each COMMIT until restored. */ -const persistTurnItemsWithFailingCommit = Effect.fn(function* ( - sentinel: OrchestrationV2TurnItem, - onPublished: (turnItem: OrchestrationV2TurnItem) => Effect.Effect, -) { - const eventSink = yield* EventSink.EventSinkV2; - const sql = yield* SqlClient.SqlClient; - let written = 0; - const write = (turnItem: OrchestrationV2TurnItem) => - eventSink - .write({ - events: [ - { - id: EventId.make(`event:commit-fault:${written++}`), - type: "turn-item.updated", - threadId: turnItem.threadId, - ...(turnItem.runId === null ? {} : { runId: turnItem.runId }), - occurredAt: DateTime.makeUnsafe("2026-07-21T12:00:00.000Z"), - payload: turnItem, - }, - ], - }) - .pipe( - // Only a failed COMMIT leaves its transaction open here. A full disk's ends it, so end - // this one the same way; the ROLLBACK fails if the write failed before COMMIT. - Effect.tapError(() => sql`ROLLBACK`.pipe(Effect.orDie)), - ); - const subscribed = yield* Deferred.make(); - yield* eventSink.stream({ eventType: "turn-item.updated" }).pipe( - Stream.runForEach(({ event }) => - event.type !== "turn-item.updated" - ? Effect.void - : event.payload.id === sentinel.id - ? Deferred.succeed(subscribed, undefined) - : onPublished(event.payload), - ), - Effect.forkChild, - ); - yield* write(sentinel); - yield* Deferred.await(subscribed); - // Every write statement succeeds; the deferred dangling reference fails only the COMMIT. - yield* sql`CREATE TABLE commit_fault_parent (id TEXT PRIMARY KEY)`; - yield* sql`CREATE TABLE commit_fault_child ( - parent_id TEXT REFERENCES commit_fault_parent (id) DEFERRABLE INITIALLY DEFERRED - )`; - yield* sql`CREATE TABLE commit_fault (enabled INTEGER NOT NULL)`; - yield* sql`INSERT INTO commit_fault VALUES (1)`; - yield* sql`CREATE TRIGGER commit_fault AFTER INSERT ON orchestration_events - WHEN (SELECT enabled FROM commit_fault) = 1 - BEGIN INSERT INTO commit_fault_child VALUES ('missing'); END`; - return { - eventSink, - write, - restore: sql`UPDATE commit_fault SET enabled = 0`.pipe(Effect.orDie), - persistedTexts: (itemId: TurnItemId) => - sql<{ readonly text: string }>` - SELECT json_extract(payload_json, '$.text') AS text FROM orchestration_events - WHERE event_type = 'turn-item.updated' AND json_extract(payload_json, '$.id') = ${itemId} - ORDER BY sequence - `.pipe( - Effect.map((rows) => rows.map((row) => row.text)), - Effect.orDie, - ), - }; -}); - -it.effect.each([ - { name: "before", afterRoot: false, batched: false, persisted: false }, - { name: "after", afterRoot: true, batched: false, persisted: false }, - { name: "before", afterRoot: false, batched: true, persisted: false }, - { name: "after", afterRoot: true, batched: true, persisted: false }, - // The idle write's body runs and only its COMMIT fails. - { name: "before", afterRoot: false, batched: true, persisted: true }, -])( - "retries a failed idle flush $name the root completes and keeps the run live (batched: $batched, persisted: $persisted)", - ({ afterRoot, batched, persisted }) => - Effect.gen(function* () { - const key = `idle-flush-retry-${afterRoot ? "after" : "before"}-root${batched ? "-batched" : ""}${persisted ? "-persisted" : ""}`; - const ids = backgroundScenarioIds(key); - const events = yield* Queue.unbounded(); - const subagentIngested = yield* Deferred.make(); - const rootFinalized = yield* Deferred.make(); - const firstWriteFailed = yield* Deferred.make(); - const failureHandled = yield* Deferred.make(); - const retried = yield* Deferred.make(); - const attempts = yield* Ref.make(0); - // What subscribers saw when persisted; otherwise what the write received. - const childWrites = yield* Ref.make>([]); - const interrupts = yield* Ref.make(0); - const recordChildWrite = Effect.fnUntraced(function* (turnItem: OrchestrationV2TurnItem) { - if (turnItem.type !== "assistant_message") return; - const { text, streaming } = turnItem; - yield* Ref.update(childWrites, (current) => [ - ...current, - `${streaming ? "streaming" : "final"}:${text}`, - ]); - if (text === heldChildText) yield* Deferred.succeed(retried, undefined); - }); - const snapshot = childAssistantSnapshot(ids); - assert(snapshot.type === "turn_item.updated"); - const persistence = persisted - ? yield* persistTurnItemsWithFailingCommit( - { ...snapshot.turnItem, id: TurnItemId.make(`${ids.childItemId}:sentinel`) }, - recordChildWrite, - ).pipe(Effect.provideContext(yield* Layer.build(layerPersistedEventSink))) - : undefined; - const commitFailure = yield* Ref.make(null); - const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { - batched: batched && persistence === undefined, - ...(persistence === undefined ? {} : { eventSink: persistence.eventSink }), - onIngest: (event) => - Effect.gen(function* () { - if (event.type === "subagent.updated") { - yield* Deferred.succeed(subagentIngested, undefined); - if ((yield* Ref.get(attempts)) > 0) - yield* Deferred.succeed(failureHandled, undefined); - return; - } - if (event.type !== "turn_item.updated" || event.turnItem.type !== "assistant_message") - return; - if ( - isHeldChildSnapshot(event) && - (yield* Ref.getAndUpdate(attempts, (n) => n + 1)) === 0 - ) { - yield* Deferred.succeed(firstWriteFailed, undefined); - yield* Effect.sleep("2 seconds"); - if (persistence === undefined) return yield* Effect.die(new Error("disk I/O error")); - return yield* persistence.write(event.turnItem).pipe( - Effect.tapError((error) => - Ref.set(commitFailure, isSqlError(error.cause) ? error.cause.reason._tag : null), - ), - Effect.orDie, - ); - } - if (persistence === undefined) return yield* recordChildWrite(event.turnItem); - yield* persistence.write(event.turnItem).pipe(Effect.orDie); - }), - onFinalized: () => Deferred.succeed(rootFinalized, undefined), - onInterruptTurn: () => Ref.update(interrupts, (n) => n + 1), - }).pipe(Effect.forkChild); - - yield* holdChildSnapshot(ids, events, subagentIngested); - if (afterRoot) { - yield* Queue.offer(events, rootTerminalEvent(ids, "completed")); - yield* Deferred.await(rootFinalized); - } - yield* TestClock.adjust("1 second"); - yield* Deferred.await(firstWriteFailed); - yield* TestClock.adjust("2 seconds"); - yield* Queue.offer(events, linkedSubagentEvent(ids, "running")); - yield* Deferred.await(failureHandled); - assert.equal(yield* Ref.get(attempts), 1); - assert.deepEqual(yield* Ref.get(childWrites), []); - if (persistence !== undefined) { - assert.equal(yield* Ref.get(commitFailure), "ConstraintError"); - assert.deepEqual(yield* persistence.persistedTexts(ids.childItemId), []); - yield* persistence.restore; - } - yield* TestClock.adjust("1 second"); - yield* Deferred.await(retried); - const childText = (text: string, streaming: boolean) => - ({ - ...snapshot, - turnItem: { ...snapshot.turnItem, text, streaming }, - }) as ProviderAdapter.ProviderAdapterV2Event; - yield* Queue.offerAll(events, [ - childText(`${heldChildText} and more`, true), - childText(`${heldChildText} and more.`, false), - ...(afterRoot ? [] : [rootTerminalEvent(ids, "completed")]), - linkedSubagentEvent(ids, "completed"), - ]); - const observed = yield* Fiber.join(scenario); - - assert.deepEqual(yield* Ref.get(childWrites), [ - `streaming:${heldChildText}`, - `final:${heldChildText} and more.`, - ]); - if (persistence !== undefined) { - assert.deepEqual(yield* persistence.persistedTexts(ids.childItemId), [ - heldChildText, - `${heldChildText} and more.`, - ]); - } - assert.equal(yield* Ref.get(interrupts), 0); - assert.include(observed, "root-finalized"); - assert.equal(observed.at(-1), "subagent:completed"); - }), -); - -it.effect("reschedules idle text when a newer snapshot precedes a failed write's retry", () => - Effect.gen(function* () { - const key = "idle-flush-earlier-deadline"; - const ids = backgroundScenarioIds(key); - const events = yield* Queue.unbounded(); - const subagentIngested = yield* Deferred.make(); - const markers = yield* Queue.unbounded(); - const newerWrite = yield* Deferred.make(); - const attempts = yield* Ref.make(0); - const childWrites = yield* Ref.make>([]); - const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { - onIngest: (event) => - Effect.gen(function* () { - if (event.type === "subagent.updated") - yield* Deferred.succeed(subagentIngested, undefined); - if (event.type === "message.updated" && event.message.role === "user") - yield* Queue.offer(markers, undefined); - if (event.type !== "turn_item.updated" || event.turnItem.type !== "assistant_message") - return; - if ((yield* Ref.getAndUpdate(attempts, (n) => n + 1)) < 2) - return yield* Effect.die(new Error("disk I/O error")); - const { text } = event.turnItem; - yield* Ref.update(childWrites, (current) => [...current, text]); - if (text === "New partial response") { - yield* Deferred.succeed(newerWrite, undefined); - } - }), - }).pipe(Effect.forkChild); - const marker = streamedMessageEvent({ - id: `message:${key}:marker`, - threadId: ids.threadId, - runId: ids.runId, - role: "user", - text: "marker", - streaming: false, - }); - - yield* holdChildSnapshot(ids, events, subagentIngested); - yield* TestClock.adjust("1 second"); - yield* Queue.offer(events, marker); - yield* Queue.take(markers); - assert.equal(yield* Ref.get(attempts), 1); - yield* TestClock.adjust("1 second"); - yield* Queue.offer(events, marker); - yield* Queue.take(markers); - assert.equal(yield* Ref.get(attempts), 2); - - const newer = childAssistantSnapshot(ids); - assert(newer.type === "turn_item.updated" && newer.turnItem.type === "assistant_message"); - yield* Queue.offerAll(events, [ - { ...newer, turnItem: { ...newer.turnItem, text: "New partial response" } }, - marker, - ]); - yield* Queue.take(markers); - yield* TestClock.adjust("800 millis"); - yield* Deferred.await(newerWrite); - assert.deepEqual(yield* Ref.get(childWrites), ["New partial response"]); - - yield* Queue.offerAll(events, [ - rootTerminalEvent(ids, "completed"), - linkedSubagentEvent(ids, "completed"), - ]); - yield* Fiber.join(scenario); - }), -); - -it.effect("keeps a failing idle write's backoff and caps warnings across other messages", () => - Effect.gen(function* () { - const key = "idle-flush-unrelated-success"; - const ids = backgroundScenarioIds(key); - const events = yield* Queue.unbounded(); - const markers = yield* Queue.unbounded(); - const failingAttempts = yield* Queue.unbounded(); - const otherWritten = yield* Deferred.make(); - const recovered = yield* Deferred.make(); - const anotherFailed = yield* Deferred.make(); - const attempts = yield* Ref.make(0); - let warnings = 0; - const message = (id: string, role: "assistant" | "user", text: string) => - streamedMessageEvent({ - id: `message:${key}:${id}`, - threadId: ids.threadId, - runId: ids.runId, - role, - text, - streaming: role === "assistant", - }); - const marker = message("marker", "user", "marker"); - const readPast = (...items: ReadonlyArray) => - Queue.offerAll(events, [...items, marker]).pipe(Effect.andThen(Queue.take(markers))); - const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { - onIngest: (event) => - Effect.gen(function* () { - if (event.type !== "message.updated") return; - if (event.message.role === "user") return yield* Queue.offer(markers, undefined); - if (event.message.text === "Other partial") - return yield* Deferred.succeed(otherWritten, undefined); - if (event.message.text === "Another failing partial") { - yield* Deferred.succeed(anotherFailed, undefined); - return yield* Effect.die(new Error("disk I/O error")); - } - const attempt = yield* Ref.updateAndGet(attempts, (n) => n + 1); - yield* Queue.offer(failingAttempts, attempt); - if (attempt <= 2) return yield* Effect.die(new Error("disk I/O error")); - yield* Deferred.succeed(recovered, undefined); - }), - }).pipe( - Effect.provide( - Logger.layer( - [ - Logger.make(({ logLevel }) => { - if (logLevel === "Warn") warnings += 1; - }), - ], - { mergeWithExisting: false }, - ), - ), - Effect.forkChild, - ); - - yield* readPast(message("failing", "assistant", "Failing partial")); - yield* TestClock.adjust("800 millis"); - assert.equal(yield* Queue.take(failingAttempts), 1); - // Due before the failing write's 1s retry. - yield* readPast(message("other", "assistant", "Other partial")); - yield* TestClock.adjust("800 millis"); - yield* Deferred.await(otherWritten); - yield* TestClock.adjust("200 millis"); - assert.equal(yield* Queue.take(failingAttempts), 2); - // A second failure backs off 2s; the other message's success must not restart at 1s. - yield* TestClock.adjust("1 second"); - yield* readPast(); - assert.equal(yield* Ref.get(attempts), 2); - assert.equal(warnings, 1); - yield* TestClock.adjust("1 second"); - yield* Deferred.await(recovered); - - yield* readPast(message("another", "assistant", "Another failing partial")); - yield* TestClock.adjust("800 millis"); - yield* Deferred.await(anotherFailed); - yield* readPast(); - assert.equal(warnings, 1); - - yield* Queue.offer(events, rootTerminalEvent(ids, "completed")); - yield* Fiber.join(scenario); - }), -); - function streamedMessageEvent(input: { readonly id: string; readonly threadId: ThreadId; @@ -4564,16 +4233,12 @@ function streamedMessageEvent(input: { } as ProviderAdapter.ProviderAdapterV2Event; } -it.effect.each([ - { name: "", failFirst: false }, - { name: " after its first idle write failed", failFirst: true }, -])("drops a child's held text once the child is handed to another run$name", ({ failFirst }) => +it.effect("drops a child's held text once the child is handed to another run", () => Effect.gen(function* () { - const key = `idle-child-handoff${failFirst ? "-after-failure" : ""}`; + const key = "idle-child-handoff"; const ids = backgroundScenarioIds(key); const events = yield* Queue.unbounded(); const subagentIngested = yield* Deferred.make(); - const firstWriteFailed = yield* Deferred.make(); const handoffRead = yield* Deferred.make(); const rootFlushed = yield* Deferred.make(); const staleWrites = yield* Ref.make>([]); @@ -4583,15 +4248,8 @@ it.effect.each([ if (event.type !== "message.updated" && event.type !== "turn_item.updated") return Effect.void; const message = event.type === "message.updated" ? event.message : event.turnItem; - if (message.threadId === ids.childThreadId) { - return Deferred.succeed(firstWriteFailed, undefined).pipe( - Effect.flatMap((first) => - failFirst && first - ? Effect.die(new Error("disk I/O error")) - : Ref.update(staleWrites, (current) => [...current, event.type]), - ), - ); - } + if (message.threadId === ids.childThreadId) + return Ref.update(staleWrites, (current) => [...current, event.type]); if (event.type === "message.updated" && event.message.role === "user") return Deferred.succeed(handoffRead, undefined); return Deferred.succeed(rootFlushed, undefined); @@ -4599,11 +4257,6 @@ it.effect.each([ }).pipe(Effect.forkChild); yield* holdChildSnapshot(ids, events, subagentIngested); - if (failFirst) { - // The held snapshot stays held for a retry that must not outlive the handoff. - yield* TestClock.adjust("1 second"); - yield* Deferred.await(firstWriteFailed); - } const handoff = linkedSubagentEvent(ids, "running"); yield* Queue.offerAll(events, [ streamedMessageEvent({ @@ -4698,49 +4351,3 @@ it.effect("flushes idle text only after the snapshots queued ahead of its wake", ]); }), ); - -it.effect("interrupts a blocked idle flush write when ingestion is cancelled", () => - Effect.gen(function* () { - const ids = backgroundScenarioIds("idle-flush-cancelled"); - const events = yield* Queue.unbounded(); - const subagentIngested = yield* Deferred.make(); - const writeStarted = yield* Deferred.make(); - const writeInterrupted = yield* Deferred.make(); - const failStart = yield* Deferred.make(); - const scenario = yield* runBackgroundItemScenario( - "idle-flush-cancelled", - () => Stream.fromQueue(events), - { - batched: true, - // A failed turn start cancels ingestion. - startTurn: Deferred.await(failStart).pipe( - Effect.andThen( - new ProviderAdapter.ProviderAdapterTurnStartError({ - driver, - threadId: ids.threadId, - providerThreadId: ids.providerThreadId, - runId: ids.runId, - cause: "start failed", - }), - ), - ), - onIngest: (event) => - event.type === "subagent.updated" - ? Deferred.succeed(subagentIngested, undefined) - : isHeldChildSnapshot(event) - ? Deferred.succeed(writeStarted, undefined).pipe( - Effect.andThen(Effect.never), - Effect.onInterrupt(() => Deferred.succeed(writeInterrupted, undefined)), - ) - : Effect.void, - }, - ).pipe(Effect.forkChild); - - yield* holdChildSnapshot(ids, events, subagentIngested); - yield* TestClock.adjust("1 second"); - yield* Deferred.await(writeStarted); - yield* Deferred.succeed(failStart, undefined); - yield* Deferred.await(writeInterrupted); - yield* Fiber.join(scenario); - }), -); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 4b3b0e8a942bd..8bdb4660f30d4 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1500,43 +1500,18 @@ export const layer: Layer.Layer< idleFlushWake = undefined; return wake === undefined ? Effect.void : Fiber.interrupt(wake.fiber); }); - // A failed idle write keeps its text held and retries with backoff while provider events - // keep ingesting. The provider's next update still writes, and fails ingestion, the normal way. - // Backoff belongs to each held snapshot; the warning cap covers the whole run. - const idleFlushFailures = new WeakMap(); - let nextIdleFlushWarningAt = Number.NEGATIVE_INFINITY; const flushIdleAssistantText = Effect.gen(function* () { const now = DateTime.toEpochMillis(yield* DateTime.now); // A wake due by now has fired; a later deadline needs a new one. if (idleFlushWake !== undefined && idleFlushWake.at <= now) idleFlushWake = undefined; if (idleFlushClosed) return; - const due = assistantStreaming.dueIdleFlushes(now); - for (const [index, flush] of due.entries()) { + for (const event of assistantStreaming.takeIdleFlushes(now)) { // The held text's child may have been handed to another run since it arrived. - if (!routeProviderEvent(flush.event, routeIdentity, yield* Ref.get(eventRouting))[0]) { - flush.discard(); + if (!routeProviderEvent(event, routeIdentity, yield* Ref.get(eventRouting))[0]) continue; - } - const written = yield* Effect.exit(ingestDelivered(flush.event, flush.event)); - if (Exit.isFailure(written)) { - if (Cause.hasInterruptsOnly(written.cause)) return yield* written; - const failures = (idleFlushFailures.get(flush.event) ?? 0) + 1; - const failedAt = DateTime.toEpochMillis(yield* DateTime.now); - if (failedAt >= nextIdleFlushWarningAt) { - yield* Effect.logWarning("orchestration V2 idle assistant text flush failed", { - runId: input.run.id, - failures, - cause: written.cause, - }); - nextIdleFlushWarningAt = failedAt + 10 * 60_000; - } - idleFlushFailures.set(flush.event, failures); - const retryAt = failedAt + Math.min(1_000 * 2 ** (failures - 1), 30_000); - for (const pending of due.slice(index)) pending.retryAt(retryAt); - break; - } - flush.delivered(now); - yield* trackChildLifecycle(flush.event, true); + // A failed write fails ingestion like any provider write. + yield* ingestDelivered(event, event); + yield* trackChildLifecycle(event, true); } yield* scheduleIdleFlush; }); diff --git a/apps/server/src/orchestration-v2/assistantStreaming.test.ts b/apps/server/src/orchestration-v2/assistantStreaming.test.ts index 08da5bf79eb38..d0d2a4d17a065 100644 --- a/apps/server/src/orchestration-v2/assistantStreaming.test.ts +++ b/apps/server/src/orchestration-v2/assistantStreaming.test.ts @@ -26,49 +26,35 @@ it("holds a streamed section heading until its content has a boundary", () => { }); }); -it("delivers held text once idle, never after its final, and never in turn mode", () => { - const snapshot = (text: string, streaming = true) => +it("delivers held text once idle without shrinking it, and never in turn mode", () => { + const snapshot = (text: string) => ({ type: "turn_item.updated", - turnItem: { id: "item", type: "reasoning", text, streaming }, + turnItem: { id: "item", type: "reasoning", text, streaming: true }, }) as unknown as ProviderAdapter.ProviderAdapterV2Event; - const text = (event: ProviderAdapter.ProviderAdapterV2Event | undefined) => + const text = (event: ProviderAdapter.ProviderAdapterV2Event | null | undefined) => event?.type === "turn_item.updated" && "text" in event.turnItem ? event.turnItem.text : null; - // Flushes everything due and reports what it delivered. const flush = (filter: ReturnType, now: number) => - filter.dueIdleFlushes(now).map((due) => { - due.delivered(now); - return text(due.event); - }); + filter.takeIdleFlushes(now).map(text); const paragraph = makeAssistantStreamingFilter("paragraph", 800); - assert.equal(text(paragraph.filter(snapshot("A\n\n```ts\nx()"), 0) ?? undefined), "A\n\n"); + assert.equal(text(paragraph.filter(snapshot("A\n\n```ts\nx()"), 0)), "A\n\n"); // Inside the 400ms throttle even though a boundary is ready. assert.isNull(paragraph.filter(snapshot("A\n\n```ts\nx()\n```\n\nB"), 100)); - assert.equal(paragraph.nextIdleFlushAt(), 900); assert.deepEqual(flush(paragraph, 899), []); - // A failed write keeps the snapshot held until its retry time. - paragraph.dueIdleFlushes(900)[0]?.retryAt(1_900); - assert.equal(paragraph.nextIdleFlushAt(), 1_900); - assert.deepEqual(flush(paragraph, 1_899), []); - assert.deepEqual(flush(paragraph, 1_900), ["A\n\n```ts\nx()\n```\n\nB"]); - assert.isNull(paragraph.nextIdleFlushAt()); - - paragraph.filter(snapshot("A\n\n```ts\nx()\n```\n\nB and C"), 2_500); - const superseded = paragraph.dueIdleFlushes(3_300); - assert.isNotNull(paragraph.filter(snapshot("A\n\n```ts\nx()\n```\n\nB and C.", false), 3_400)); - // An outcome for a snapshot the final replaced changes nothing. - superseded[0]?.retryAt(3_500); + assert.equal(paragraph.nextIdleFlushAt(), 900); + assert.deepEqual(flush(paragraph, 900), ["A\n\n```ts\nx()\n```\n\nB"]); assert.isNull(paragraph.nextIdleFlushAt()); - assert.deepEqual(flush(paragraph, 5_000), []); + // The next boundary is behind what the idle delivery showed. + assert.isNull(paragraph.filter(snapshot("A\n\n```ts\nx()\n```\n\nB and C"), 2_500)); // Provider rewrites replace delivered text: shorter at once, same-length after the throttle. const rewrite = makeAssistantStreamingFilter("paragraph", 800); assert.equal( - text(rewrite.filter(snapshot("Original long paragraph.\n\n"), 0) ?? undefined), + text(rewrite.filter(snapshot("Original long paragraph.\n\n"), 0)), "Original long paragraph.\n\n", ); - assert.equal(text(rewrite.filter(snapshot("New.\n\n"), 1_000) ?? undefined), "New.\n\n"); + assert.equal(text(rewrite.filter(snapshot("New.\n\n"), 1_000)), "New.\n\n"); assert.isNull(rewrite.filter(snapshot("Old.\n\n"), 1_100)); assert.deepEqual(flush(rewrite, 1_900), ["Old.\n\n"]); assert.isNull(rewrite.filter(snapshot("Old"), 2_500)); diff --git a/apps/server/src/orchestration-v2/assistantStreaming.ts b/apps/server/src/orchestration-v2/assistantStreaming.ts index 59f6aaa47dabf..ef6de3ab3f483 100644 --- a/apps/server/src/orchestration-v2/assistantStreaming.ts +++ b/apps/server/src/orchestration-v2/assistantStreaming.ts @@ -86,14 +86,6 @@ interface StreamTrack { readonly dueAt: number; } -/** A held snapshot due for delivery. Each outcome applies only while nothing newer replaced it. */ -interface IdleFlush { - readonly event: ProviderAdapter.ProviderAdapterV2Event; - readonly delivered: (now: number) => void; - readonly retryAt: (at: number) => void; - readonly discard: () => void; -} - function streamedMessage(event: ProviderAdapter.ProviderAdapterV2Event): StreamedMessage | null { if (event.type === "message.updated" && event.message.role === "assistant") return event.message; if ( @@ -127,6 +119,14 @@ export function makeAssistantStreamingFilter( idleFlushMs = ASSISTANT_IDLE_FLUSH_MS, ) { const tracks = new Map(); + // Earliest deadline of any held snapshot, or earlier: a newer snapshot only pushes its own + // deadline later, so the cache stays O(1) per event and is recomputed when it comes due. + let nextDueAt: number | null = null; + const track = (key: string, value: StreamTrack) => { + tracks.set(key, value); + if (value.held !== null && (nextDueAt === null || value.dueAt < nextDueAt)) + nextDueAt = value.dueAt; + }; return { filter( event: ProviderAdapter.ProviderAdapterV2Event, @@ -149,7 +149,7 @@ export function makeAssistantStreamingFilter( const previous = tracks.get(key); const delivered = previous?.text ?? ""; const hold = () => { - tracks.set(key, { + track(key, { text: delivered, at: previous?.at ?? Number.NEGATIVE_INFINITY, held: message.text !== delivered ? event : null, @@ -162,7 +162,7 @@ export function makeAssistantStreamingFilter( // An idle delivery may already show more than the next boundary; never shrink it. A // provider rewrite or truncation is not a prefix and replaces it. if (delivered.startsWith(ready)) return hold(); - tracks.set(key, { + track(key, { text: ready, at: now, held: ready !== message.text ? event : null, @@ -170,33 +170,24 @@ export function makeAssistantStreamingFilter( }); return withText(event, ready); }, - nextIdleFlushAt(): number | null { - let next: number | null = null; - for (const track of tracks.values()) { - if (track.held !== null && (next === null || track.dueAt < next)) next = track.dueAt; - } - return next; - }, - dueIdleFlushes(now: number): ReadonlyArray { - const due: Array = []; - for (const [key, track] of tracks) { - const event = track.held; - if (event === null || track.dueAt > now) continue; - const settle = (next: (current: StreamTrack) => StreamTrack) => { - const current = tracks.get(key); - if (current?.held === event) tracks.set(key, next(current)); - }; - due.push({ - event, - delivered: (at) => - settle((current) => ({ - text: streamedMessage(event)?.text ?? current.text, - at, - held: null, - dueAt: at, - })), - retryAt: (at) => settle((current) => ({ ...current, dueAt: at })), - discard: () => settle((current) => ({ ...current, held: null })), + nextIdleFlushAt: (): number | null => nextDueAt, + /** Returns the held snapshots due by `now` and counts them as delivered. */ + takeIdleFlushes(now: number): ReadonlyArray { + const due: Array = []; + nextDueAt = null; + for (const [key, current] of tracks) { + const event = current.held; + if (event === null) continue; + if (current.dueAt > now) { + if (nextDueAt === null || current.dueAt < nextDueAt) nextDueAt = current.dueAt; + continue; + } + due.push(event); + tracks.set(key, { + text: streamedMessage(event)?.text ?? current.text, + at: now, + held: null, + dueAt: now, }); } return due; From 75fc58e072221ab26c939931ebf6650078133edb Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Sat, 7 Nov 2026 18:52:35 +0000 Subject: [PATCH 07/11] test(server): type the stored-event stub Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/RunExecutionService.test.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index be3562920b485..23aede9f95f76 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -14,6 +14,7 @@ import { type OrchestrationV2ProviderTurn, type OrchestrationV2Run, type OrchestrationV2RunAttempt, + type OrchestrationV2StoredEvent, type OrchestrationV2Subagent, type OrchestrationV2TurnItem, ProviderDriverKind, @@ -3829,7 +3830,7 @@ function runBackgroundItemScenario( } // Only the count of stored events matters to the run. return event.type === "provider_thread.updated" - ? [{} as OrchestrationV2DomainEvent] + ? [{} as OrchestrationV2StoredEvent] : []; }); // A write outside a batch waits for queued snapshots, as the event sink's does. From 73f652053e93249b541145d639d9b3c0d2a69bf0 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Mon, 9 Nov 2026 10:07:47 +0000 Subject: [PATCH 08/11] fix(server): keep a settled root's held text from reopening it When a root ended mid-paragraph while child work kept ingestion open, the idle flush later wrote the root's held tail as a streaming snapshot after the run had settled. The flush now skips root-thread text once the root run is finalized; child text still flushes. Tests: the filter's omnibus case becomes a sequence table covering unfinished fences, lists and tables, appends after an idle delivery, completion at the idle deadline, fast paragraph bursts, two messages, empty and whitespace text, rewrites and turn mode. The queued-final ordering test folds into the idle-paragraph scenario, and the interrupted child asserts it stops streaming. Co-Authored-By: Claude Opus 5.5 --- .../RunExecutionService.test.ts | 132 +++++------- .../orchestration-v2/RunExecutionService.ts | 9 + .../assistantStreaming.test.ts | 199 +++++++++++++++--- 3 files changed, 231 insertions(+), 109 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 23aede9f95f76..e5797e9b76e87 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -4023,27 +4023,41 @@ it.effect("delivers a held paragraph after the provider goes idle with its promp }, ] as ReadonlyArray; + const blockerStarted = yield* Deferred.make(); + const releaseBlocker = yield* Deferred.make(); + const blocker = streamedMessageEvent({ + id: `message:${key}:blocker`, + threadId: ids.threadId, + runId: ids.runId, + role: "assistant", + text: "Other", + streaming: false, + }); const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(providerEvents), { batched: true, onIngest: (event, write) => - event.type === "message.updated" || event.type === "turn_item.updated" - ? Effect.gen(function* () { - const message = event.type === "message.updated" ? event.message : event.turnItem; - const text = "text" in message ? message.text : ""; - const streaming = "streaming" in message && message.streaming; - yield* Ref.update(delivered, (current) => [ - ...current, - `${event.type}:${streaming ? "streaming" : "final"}:${JSON.stringify(text)}${ - write.inSharedCommit ? " in commit" : "" - }`, - ]); - if (event.type === "turn_item.updated") { - yield* Deferred.succeed(paragraphDelivered, undefined); - if (streaming && text === "First.\n\nWaiting on the build") - yield* Deferred.succeed(idleDelivered, undefined); - } - }) - : Effect.void, + event.type === "message.updated" && event.message.text === "Other" + ? Deferred.succeed(blockerStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseBlocker)), + ) + : event.type === "message.updated" || event.type === "turn_item.updated" + ? Effect.gen(function* () { + const message = event.type === "message.updated" ? event.message : event.turnItem; + const text = "text" in message ? message.text : ""; + const streaming = "streaming" in message && message.streaming; + yield* Ref.update(delivered, (current) => [ + ...current, + `${event.type}:${streaming ? "streaming" : "final"}:${JSON.stringify(text)}${ + write.inSharedCommit ? " in commit" : "" + }`, + ]); + if (event.type === "turn_item.updated") { + yield* Deferred.succeed(paragraphDelivered, undefined); + if (streaming && text === "First.\n\nWaiting on the build") + yield* Deferred.succeed(idleDelivered, undefined); + } + }) + : Effect.void, }).pipe(Effect.forkChild); yield* Queue.offerAll(providerEvents, assistantText("First.\n\nWaiting on the build", true)); @@ -4053,12 +4067,18 @@ it.effect("delivers a held paragraph after the provider goes idle with its promp yield* Deferred.await(idleDelivered); yield* Queue.offerAll(providerEvents, [ ...assistantText("First.\n\nWaiting on the build to", true), + blocker, ...assistantText("First.\n\nWaiting on the build to finish.", false), - rootTerminalEvent(ids, "completed"), ]); + yield* Deferred.await(blockerStarted); + // The appended text comes due while its final snapshots wait behind the blocked write. + yield* TestClock.adjust("1 second"); + yield* Deferred.succeed(releaseBlocker, undefined); + yield* Queue.offer(providerEvents, rootTerminalEvent(ids, "completed")); yield* Fiber.join(scenario); // Snapshots share the batch commit; the idle write commits alone, before it counts as shown. + // The final snapshots win over the stale append that came due behind them. assert.deepEqual(yield* Ref.get(delivered), [ `message.updated:streaming:"First.\\n\\n" in commit`, `turn_item.updated:streaming:"First.\\n\\n" in commit`, @@ -4159,6 +4179,7 @@ it.effect("settles a child message only an idle flush wrote when the root is int assert.equal(settled.length, 1); assert.equal(settled[0]?.status, "interrupted"); assert.equal("text" in settled[0]! ? settled[0].text : null, heldChildText); + assert.isFalse("streaming" in settled[0]! && settled[0].streaming); }), ); @@ -4173,17 +4194,32 @@ it.effect.each([ const subagentIngested = yield* Deferred.make(); const rootFinalized = yield* Deferred.make(); const flushed = yield* Deferred.make(); + const rootWrites = yield* Ref.make>([]); const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { onIngest: (event) => event.type === "subagent.updated" ? Deferred.succeed(subagentIngested, undefined) : isHeldChildSnapshot(event) ? Deferred.succeed(flushed, undefined) - : Effect.void, + : event.type === "message.updated" && event.message.threadId === ids.threadId + ? Ref.update(rootWrites, (current) => [...current, event.message.text]) + : Effect.void, onFinalized: () => Deferred.succeed(rootFinalized, undefined), }).pipe(Effect.forkChild); yield* holdChildSnapshot(ids, events, subagentIngested); + // The root ends mid-paragraph: its tail is held, and the run settles without it. + yield* Queue.offer( + events, + streamedMessageEvent({ + id: `message:${key}:root`, + threadId: ids.threadId, + runId: ids.runId, + role: "assistant", + text: "Root.\n\nhalf", + streaming: true, + }), + ); const terminal = rootTerminalEvent(ids, terminalStatus); yield* Queue.offer( events, @@ -4202,6 +4238,8 @@ it.effect.each([ "turn_item:running", "subagent:completed", ]); + // A streaming write after the run settled would reopen the root's message. + assert.deepEqual(yield* Ref.get(rootWrites), ["Root.\n\n"]); }), ); @@ -4298,57 +4336,3 @@ it.effect("drops a child's held text once the child is handed to another run", ( assert.deepEqual(yield* Ref.get(staleWrites), []); }), ); - -it.effect("flushes idle text only after the snapshots queued ahead of its wake", () => - Effect.gen(function* () { - const key = "idle-flush-queued-snapshot"; - const ids = backgroundScenarioIds(key); - const events = yield* Queue.unbounded(); - const paragraphWritten = yield* Deferred.make(); - const blockerStarted = yield* Deferred.make(); - const releaseBlocker = yield* Deferred.make(); - const writes = yield* Ref.make>([]); - const message = (id: string, text: string, streaming: boolean) => - streamedMessageEvent({ - id: `message:${key}:${id}`, - threadId: ids.threadId, - runId: ids.runId, - role: "assistant", - text, - streaming, - }); - const scenario = yield* runBackgroundItemScenario(key, () => Stream.fromQueue(events), { - batched: true, - onIngest: (event) => { - if (event.type !== "message.updated") return Effect.void; - if (event.message.text === "Other") - return Deferred.succeed(blockerStarted, undefined).pipe( - Effect.andThen(Deferred.await(releaseBlocker)), - ); - const { streaming, text } = event.message; - return Ref.update(writes, (current) => [ - ...current, - `${streaming ? "streaming" : "final"}:${text}`, - ]).pipe(Effect.andThen(Deferred.succeed(paragraphWritten, undefined))); - }, - }).pipe(Effect.forkChild); - - yield* Queue.offer(events, message("text", "First.\n\nWaiting on the build", true)); - yield* Deferred.await(paragraphWritten); - yield* Queue.offerAll(events, [ - message("blocker", "Other", false), - message("text", "First.\n\nWaiting on the build to finish.", false), - ]); - yield* Deferred.await(blockerStarted); - // The held text comes due while its final snapshot waits behind the blocked write. - yield* TestClock.adjust("1 second"); - yield* Deferred.succeed(releaseBlocker, undefined); - yield* Queue.offer(events, rootTerminalEvent(ids, "completed")); - yield* Fiber.join(scenario); - - assert.deepEqual(yield* Ref.get(writes), [ - "streaming:First.\n\n", - "final:First.\n\nWaiting on the build to finish.", - ]); - }), -); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 6bc9a496ba72d..93bf2009809e6 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -176,6 +176,12 @@ function isRunOwnedSubagentTerminalStatus( return status === "interrupted" || status === "failed" || status === "cancelled"; } +function streamedThreadId(event: ProviderAdapter.ProviderAdapterV2Event): ThreadId | null { + if (event.type === "message.updated") return event.message.threadId; + if (event.type === "turn_item.updated") return event.turnItem.threadId; + return null; +} + function childTurnsContinue(terminal: ProviderTerminalEvent): boolean { return terminal.status === "failed" && terminal.childTurnsContinue === true; } @@ -1506,7 +1512,10 @@ export const layer: Layer.Layer< // A wake due by now has fired; a later deadline needs a new one. if (idleFlushWake !== undefined && idleFlushWake.at <= now) idleFlushWake = undefined; if (idleFlushClosed) return; + const rootSettled = yield* Ref.get(rootRunFinalized); for (const event of assistantStreaming.takeIdleFlushes(now)) { + // Settling the root run closed its messages; only children may still stream. + if (rootSettled && streamedThreadId(event) === input.run.threadId) continue; // The held text's child may have been handed to another run since it arrived. if (!routeProviderEvent(event, routeIdentity, yield* Ref.get(eventRouting))[0]) continue; diff --git a/apps/server/src/orchestration-v2/assistantStreaming.test.ts b/apps/server/src/orchestration-v2/assistantStreaming.test.ts index d0d2a4d17a065..0e8210ebd1f6a 100644 --- a/apps/server/src/orchestration-v2/assistantStreaming.test.ts +++ b/apps/server/src/orchestration-v2/assistantStreaming.test.ts @@ -1,4 +1,6 @@ +import { MessageId, ProviderDriverKind, ThreadId } from "@t3tools/contracts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as DateTime from "effect/DateTime"; import { assert, describe, expect, it } from "vite-plus/test"; import { makeAssistantStreamingFilter, splitBufferedAssistantText } from "./assistantStreaming.ts"; @@ -26,41 +28,168 @@ it("holds a streamed section heading until its content has a boundary", () => { }); }); -it("delivers held text once idle without shrinking it, and never in turn mode", () => { - const snapshot = (text: string) => - ({ - type: "turn_item.updated", - turnItem: { id: "item", type: "reasoning", text, streaming: true }, - }) as unknown as ProviderAdapter.ProviderAdapterV2Event; - const text = (event: ProviderAdapter.ProviderAdapterV2Event | null | undefined) => - event?.type === "turn_item.updated" && "text" in event.turnItem ? event.turnItem.text : null; - const flush = (filter: ReturnType, now: number) => - filter.takeIdleFlushes(now).map(text); +type Step = + // A snapshot at `at` ms, and the text it delivers at once (null when held back). + | { readonly at: number; readonly text: string; readonly shows: string | null } + | { readonly at: number; readonly final: string } + | { readonly id: string; readonly at: number; readonly text: string; readonly shows: string } + // Held snapshots due by `idle` ms, delivered whole. + | { readonly idle: number; readonly shows: ReadonlyArray } + | { readonly dueAt: number | null }; - const paragraph = makeAssistantStreamingFilter("paragraph", 800); - assert.equal(text(paragraph.filter(snapshot("A\n\n```ts\nx()"), 0)), "A\n\n"); - // Inside the 400ms throttle even though a boundary is ready. - assert.isNull(paragraph.filter(snapshot("A\n\n```ts\nx()\n```\n\nB"), 100)); - assert.deepEqual(flush(paragraph, 899), []); - assert.equal(paragraph.nextIdleFlushAt(), 900); - assert.deepEqual(flush(paragraph, 900), ["A\n\n```ts\nx()\n```\n\nB"]); - assert.isNull(paragraph.nextIdleFlushAt()); - // The next boundary is behind what the idle delivery showed. - assert.isNull(paragraph.filter(snapshot("A\n\n```ts\nx()\n```\n\nB and C"), 2_500)); +const createdAt = DateTime.makeUnsafe(0); +const snapshot = (id: string, text: string, streaming: boolean) => + ({ + type: "message.updated", + driver: ProviderDriverKind.make("codex"), + message: { + createdBy: "agent", + creationSource: "provider", + id: MessageId.make(id), + threadId: ThreadId.make("thread"), + runId: null, + nodeId: null, + role: "assistant", + text, + attachments: [], + streaming, + createdAt, + updatedAt: createdAt, + }, + }) satisfies ProviderAdapter.ProviderAdapterV2Event; +const shown = (event: ProviderAdapter.ProviderAdapterV2Event | null) => + event?.type === "message.updated" ? event.message.text : null; - // Provider rewrites replace delivered text: shorter at once, same-length after the throttle. - const rewrite = makeAssistantStreamingFilter("paragraph", 800); - assert.equal( - text(rewrite.filter(snapshot("Original long paragraph.\n\n"), 0)), - "Original long paragraph.\n\n", - ); - assert.equal(text(rewrite.filter(snapshot("New.\n\n"), 1_000)), "New.\n\n"); - assert.isNull(rewrite.filter(snapshot("Old.\n\n"), 1_100)); - assert.deepEqual(flush(rewrite, 1_900), ["Old.\n\n"]); - assert.isNull(rewrite.filter(snapshot("Old"), 2_500)); - assert.deepEqual(flush(rewrite, 3_300), ["Old"]); - - const turn = makeAssistantStreamingFilter("turn", 800); - assert.isNull(turn.filter(snapshot("A\n\nB"), 0)); - assert.isNull(turn.nextIdleFlushAt()); +it.each<{ readonly name: string; readonly mode?: "turn"; readonly steps: ReadonlyArray }>([ + { + name: "an unfinished code fence", + steps: [ + { at: 0, text: "A\n\n```ts\nx()", shows: "A\n\n" }, + { dueAt: 800 }, + { idle: 799, shows: [] }, + { idle: 800, shows: ["A\n\n```ts\nx()"] }, + { dueAt: null }, + { at: 1_300, text: "A\n\n```ts\nx()\n```\n\nB", shows: "A\n\n```ts\nx()\n```\n\n" }, + ], + }, + { + name: "a list's partial last item", + steps: [ + { at: 0, text: "Steps:\n\n- one\n- tw", shows: "Steps:\n\n- one\n" }, + { idle: 800, shows: ["Steps:\n\n- one\n- tw"] }, + { at: 1_300, text: "Steps:\n\n- one\n- two\n- th", shows: "Steps:\n\n- one\n- two\n" }, + ], + }, + { + name: "an incomplete table row", + steps: [ + { at: 0, text: "Intro.\n\n| a | b |\n|---|---|\n| 1 |", shows: "Intro.\n\n" }, + { idle: 800, shows: ["Intro.\n\n| a | b |\n|---|---|\n| 1 |"] }, + { + at: 1_300, + text: "Intro.\n\n| a | b |\n|---|---|\n| 1 | 2 |\n\nAfter", + shows: "Intro.\n\n| a | b |\n|---|---|\n| 1 | 2 |\n\n", + }, + ], + }, + { + name: "more of the same paragraph after an idle delivery", + steps: [ + { at: 0, text: "A\n\nB", shows: "A\n\n" }, + { idle: 800, shows: ["A\n\nB"] }, + { at: 900, text: "A\n\nB more", shows: null }, + { idle: 1_699, shows: [] }, + { idle: 1_700, shows: ["A\n\nB more"] }, + { idle: 5_000, shows: [] }, + // Its only boundary is behind what the idle delivery showed. + { at: 5_100, text: "A\n\nB more and", shows: null }, + { at: 5_200, text: "A\n\nB more and.\n\nC", shows: "A\n\nB more and.\n\n" }, + { at: 5_300, final: "A\n\nB more and.\n\nC." }, + { idle: 10_000, shows: [] }, + ], + }, + { + name: "completion at the idle deadline", + steps: [ + { at: 0, text: "A\n\nB", shows: "A\n\n" }, + { at: 800, final: "A\n\nB." }, + { idle: 800, shows: [] }, + ], + }, + { + name: "completion right after an idle delivery at the same instant", + steps: [ + { at: 0, text: "A\n\nB", shows: "A\n\n" }, + { idle: 800, shows: ["A\n\nB"] }, + { at: 800, final: "A\n\nB." }, + { idle: 5_000, shows: [] }, + ], + }, + { + name: "several fast paragraphs inside the throttle", + steps: [ + { at: 0, text: "A\n\nB", shows: "A\n\n" }, + { at: 100, text: "A\n\nB\n\nC", shows: null }, + { at: 200, text: "A\n\nB\n\nC\n\nD", shows: null }, + { at: 300, text: "A\n\nB\n\nC\n\nD\n\nE", shows: null }, + { idle: 1_099, shows: [] }, + { idle: 1_100, shows: ["A\n\nB\n\nC\n\nD\n\nE"] }, + ], + }, + { + name: "two messages due at different times", + steps: [ + { id: "a", at: 0, text: "A\n\nx", shows: "A\n\n" }, + { id: "b", at: 300, text: "B\n\ny", shows: "B\n\n" }, + { dueAt: 800 }, + { idle: 800, shows: ["A\n\nx"] }, + { dueAt: 1_100 }, + { idle: 1_100, shows: ["B\n\ny"] }, + ], + }, + { + name: "empty text", + steps: [{ at: 0, text: "", shows: null }, { dueAt: null }], + }, + { + name: "whitespace-only text", + steps: [ + { at: 0, text: " \n", shows: null }, + { idle: 800, shows: [" \n"] }, + { idle: 5_000, shows: [] }, + ], + }, + { + // Rewrites replace delivered text: shorter at once, same length after the throttle. + name: "provider rewrites and truncation", + steps: [ + { at: 0, text: "Original long paragraph.\n\n", shows: "Original long paragraph.\n\n" }, + { at: 1_000, text: "New.\n\n", shows: "New.\n\n" }, + { at: 1_100, text: "Old.\n\n", shows: null }, + { idle: 1_900, shows: ["Old.\n\n"] }, + { at: 2_500, text: "Old", shows: null }, + { idle: 3_300, shows: ["Old"] }, + { at: 4_000, text: "", shows: null }, + { idle: 4_800, shows: [""] }, + ], + }, + { + name: "turn mode", + mode: "turn", + steps: [{ at: 0, text: "A\n\nB", shows: null }, { dueAt: null }], + }, +])("paragraph streaming holds and idle-delivers $name", ({ mode, steps }) => { + const filter = makeAssistantStreamingFilter(mode ?? "paragraph", 800); + for (const step of steps) { + if ("dueAt" in step) assert.equal(filter.nextIdleFlushAt(), step.dueAt); + else if ("idle" in step) + assert.deepEqual(filter.takeIdleFlushes(step.idle).map(shown), [...step.shows]); + else if ("final" in step) + assert.equal(shown(filter.filter(snapshot("m", step.final, false), step.at)), step.final); + else + assert.equal( + shown(filter.filter(snapshot("id" in step ? step.id : "m", step.text, true), step.at)), + step.shows, + ); + } }); From 41e8ac4587ced0cd051cb2fbee8aa8f12c52ac41 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Mon, 9 Nov 2026 11:31:42 +0000 Subject: [PATCH 09/11] docs(server): explain untraced snapshot ingestion --- apps/server/src/orchestration-v2/RunExecutionService.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 93bf2009809e6..97b47ba105cca 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1416,6 +1416,7 @@ export const layer: Layer.Layer< } return { event, accepted }; }); + // Per-event ingestion avoids adding a span for every provider snapshot. const ingestDelivered = Effect.fnUntraced(function* ( event: ProviderAdapter.ProviderAdapterV2Event, deliveredEvent: ProviderAdapter.ProviderAdapterV2Event, From a829f031bd76bdf507f95e161ab365348299e1d7 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Mon, 9 Nov 2026 11:52:34 +0000 Subject: [PATCH 10/11] fix(server): keep idle flushing armed when a timer fires before the wall clock A timer can fire while Date.now() still reads a millisecond before its deadline. The idle-flush tick then kept the finished wake as pending, so no later wake was armed and held text stayed hidden for the rest of the run. Each tick now carries the deadline its timer slept until, and the flush treats that deadline as reached. Co-Authored-By: Claude Opus 5.5 --- .../RunExecutionService.test.ts | 57 +++++++++++++++---- .../orchestration-v2/RunExecutionService.ts | 16 ++++-- 2 files changed, 56 insertions(+), 17 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index e5797e9b76e87..e25b4c7ab556a 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -29,10 +29,12 @@ import { TurnItemId, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; @@ -3978,15 +3980,21 @@ it.effect("keeps a failed root's still-running native children live while the ru }), ); -it.effect("delivers a held paragraph after the provider goes idle with its prompt still open", () => +it.effect.each([ + { name: "on time", wakesEarly: false }, + // Timers run on the monotonic clock, so one can fire while the wall clock still reads a + // millisecond before its deadline. + { name: "before the wall clock reaches the deadline", wakesEarly: true }, +])("delivers held text while an open-prompt provider idles, waking $name", ({ wakesEarly }) => Effect.gen(function* () { - const key = "idle-paragraph"; + const key = `idle-paragraph-${wakesEarly}`; const ids = backgroundScenarioIds(key); const now = yield* DateTime.now; const providerEvents = yield* Queue.unbounded(); const delivered = yield* Ref.make>([]); const paragraphDelivered = yield* Deferred.make(); - const idleDelivered = yield* Deferred.make(); + const nextParagraphDelivered = yield* Deferred.make(); + const clock = yield* Clock.Clock; const assistantText = (text: string, streaming: boolean) => [ { @@ -4053,22 +4061,45 @@ it.effect("delivers a held paragraph after the provider goes idle with its promp ]); if (event.type === "turn_item.updated") { yield* Deferred.succeed(paragraphDelivered, undefined); - if (streaming && text === "First.\n\nWaiting on the build") - yield* Deferred.succeed(idleDelivered, undefined); + if (text === "First.\n\nWaiting on the build.\n\n") + yield* Deferred.succeed(nextParagraphDelivered, undefined); } }) : Effect.void, - }).pipe(Effect.forkChild); + }).pipe( + Effect.provideService(Clock.Clock, { + ...clock, + sleep: (duration) => + clock.sleep(wakesEarly ? Duration.subtract(duration, Duration.millis(1)) : duration), + }), + Effect.forkChild, + ); + // Waking a sleep runs ingestion until it suspends, so an idle write has landed by now. + const lastShown = Ref.get(delivered).pipe(Effect.map((shown) => shown.at(-1))); yield* Queue.offerAll(providerEvents, assistantText("First.\n\nWaiting on the build", true)); yield* Deferred.await(paragraphDelivered); // The provider keeps its prompt open while a backgrounded command runs. + yield* TestClock.adjust("2 seconds"); + assert.strictEqual( + yield* lastShown, + 'turn_item.updated:streaming:"First.\\n\\nWaiting on the build"', + ); + // The next paragraph's tail still flushes on a later pause. + yield* Queue.offerAll( + providerEvents, + assistantText("First.\n\nWaiting on the build.\n\nRunning", true), + ); + yield* Deferred.await(nextParagraphDelivered); yield* TestClock.adjust("1 second"); - yield* Deferred.await(idleDelivered); + assert.strictEqual( + yield* lastShown, + 'turn_item.updated:streaming:"First.\\n\\nWaiting on the build.\\n\\nRunning"', + ); yield* Queue.offerAll(providerEvents, [ - ...assistantText("First.\n\nWaiting on the build to", true), + ...assistantText("First.\n\nWaiting on the build.\n\nRunning tests", true), blocker, - ...assistantText("First.\n\nWaiting on the build to finish.", false), + ...assistantText("First.\n\nWaiting on the build.\n\nRunning tests passed.", false), ]); yield* Deferred.await(blockerStarted); // The appended text comes due while its final snapshots wait behind the blocked write. @@ -4084,8 +4115,12 @@ it.effect("delivers a held paragraph after the provider goes idle with its promp `turn_item.updated:streaming:"First.\\n\\n" in commit`, 'message.updated:streaming:"First.\\n\\nWaiting on the build"', 'turn_item.updated:streaming:"First.\\n\\nWaiting on the build"', - `message.updated:final:"First.\\n\\nWaiting on the build to finish." in commit`, - `turn_item.updated:final:"First.\\n\\nWaiting on the build to finish." in commit`, + `message.updated:streaming:"First.\\n\\nWaiting on the build.\\n\\n" in commit`, + `turn_item.updated:streaming:"First.\\n\\nWaiting on the build.\\n\\n" in commit`, + 'message.updated:streaming:"First.\\n\\nWaiting on the build.\\n\\nRunning"', + 'turn_item.updated:streaming:"First.\\n\\nWaiting on the build.\\n\\nRunning"', + `message.updated:final:"First.\\n\\nWaiting on the build.\\n\\nRunning tests passed." in commit`, + `turn_item.updated:final:"First.\\n\\nWaiting on the build.\\n\\nRunning tests passed." in commit`, ]); }), ); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 97b47ba105cca..400c3525e8455 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1479,8 +1479,10 @@ export const layer: Layer.Layer< // The timer only wakes ingestion. The flush runs as an ingestion step outside any provider // batch, so it stays ordered with routing and snapshots, and a write has committed when // it returns. - const idleFlushTick = { type: "idle_flush" } as const; - const idleFlushTicks = yield* Queue.dropping(1); + const idleFlushTicks = yield* Queue.dropping<{ + readonly type: "idle_flush"; + readonly at: number; + }>(1); let idleFlushWake: { readonly at: number; readonly fiber: Fiber.Fiber } | undefined; const scheduleIdleFlush = Effect.suspend(() => { const at = assistantStreaming.nextIdleFlushAt(); @@ -1495,7 +1497,7 @@ export const layer: Layer.Layer< if (previous !== undefined) yield* Fiber.interrupt(previous); const now = DateTime.toEpochMillis(yield* DateTime.now); const fiber = yield* Effect.sleep(Duration.millis(Math.max(0, at - now))).pipe( - Effect.andThen(Queue.offer(idleFlushTicks, idleFlushTick)), + Effect.andThen(Queue.offer(idleFlushTicks, { type: "idle_flush", at })), Effect.asVoid, Effect.forkDetach, ); @@ -1508,8 +1510,10 @@ export const layer: Layer.Layer< idleFlushWake = undefined; return wake === undefined ? Effect.void : Fiber.interrupt(wake.fiber); }); - const flushIdleAssistantText = Effect.gen(function* () { - const now = DateTime.toEpochMillis(yield* DateTime.now); + const flushIdleAssistantText = Effect.fnUntraced(function* (wokeAt: number) { + // A tick means its timer slept until `wokeAt`, but the wall clock can still read up + // to a millisecond earlier; trusting it would keep the finished wake and never re-arm. + const now = Math.max(DateTime.toEpochMillis(yield* DateTime.now), wokeAt); // A wake due by now has fired; a later deadline needs a new one. if (idleFlushWake !== undefined && idleFlushWake.at <= now) idleFlushWake = undefined; if (idleFlushClosed) return; @@ -1597,7 +1601,7 @@ export const layer: Layer.Layer< ), (event) => event.type === "idle_flush" - ? flushIdleAssistantText + ? flushIdleAssistantText(event.at) : Effect.gen(function* () { yield* ingestEvent(yield* routeEvent(event)); }), From 49dddb89b94b979a81e4a9fee90995590cbc35a1 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Mon, 9 Nov 2026 11:56:23 +0000 Subject: [PATCH 11/11] docs(server): explain untraced idle wake ingestion --- apps/server/src/orchestration-v2/RunExecutionService.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 400c3525e8455..07811415fa312 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1510,6 +1510,7 @@ export const layer: Layer.Layer< idleFlushWake = undefined; return wake === undefined ? Effect.void : Fiber.interrupt(wake.fiber); }); + // Per-tick ingestion avoids adding a span for every idle wake. const flushIdleAssistantText = Effect.fnUntraced(function* (wokeAt: number) { // A tick means its timer slept until `wokeAt`, but the wall clock can still read up // to a millisecond earlier; trusting it would keep the finished wake and never re-arm.