From bc9e40932254652b966dfde2a62ccc8e4188fc1b Mon Sep 17 00:00:00 2001 From: Spike Date: Mon, 2 Nov 2026 02:59:58 +0000 Subject: [PATCH 1/9] fix(server): tool calls a released session cuts off no longer keep the thread Waiting When T3 releases a provider session whose background work ends with its process, such as the detach after t3_worktree_handoff or a provider switch, the session cannot report how its open tool calls ended. Run finalization keeps command and dynamic tool items open for a late result, so these items stayed running. The thread then showed "Waiting on background task" until a restart. The release path already clears the dead session's background-task roster. It now also marks that session's open command_execution and dynamic_tool items as interrupted, as Stop does for items on a dead session. It reads the item ids when the session leaves the live map, so items of a replacement session stay open. Runtimes whose tasks can outlive them, and server shutdown, keep their items as before. Fixes #15993 Co-Authored-By: Claude Opus 5.5 --- .../ProviderSessionManager.test.ts | 123 ++++++++++++ .../ProviderSessionManager.ts | 99 +++++++-- .../ProviderSwitch.integration.test.ts | 188 +++++++++++++++++- 3 files changed, 391 insertions(+), 19 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 5bbb3926e2b66..4ddc1ce061f6e 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -10,6 +10,7 @@ import { type OrchestrationV2ProviderCapabilities, type OrchestrationV2ProviderSession, type OrchestrationV2ProviderThread, + type OrchestrationV2TurnItem, type Project, ProjectId, ProviderDriverKind, @@ -2512,6 +2513,128 @@ it.effect("ProviderSessionManagerV2 keeps the roster of a runtime whose tasks ca }), ); +const seedToolCalls = (input: { + readonly threadId: ThreadId; + readonly providerThreads: ReadonlyArray; +}) => + Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const [releasedThread, otherThread] = input.providerThreads; + const toolCall = ( + nativeItemId: string, + providerThread: OrchestrationV2ProviderThread, + status: "running" | "completed", + ordinal: number, + ) => + ({ + id: idAllocator.derive.turnItemFromProviderItem({ driver: CODEX_DRIVER, nativeItemId }), + threadId: input.threadId, + runId: null, + nodeId: null, + providerThreadId: providerThread.id, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status, + title: null, + startedAt: now, + completedAt: status === "completed" ? now : null, + updatedAt: now, + type: "dynamic_tool", + toolName: "mcp__t3-code__t3_worktree_handoff", + input: {}, + }) satisfies OrchestrationV2TurnItem; + const items = { + cutOff: toolCall("handoff-cut-off", releasedThread!, "running", 1), + finished: toolCall("finished-call", releasedThread!, "completed", 2), + otherSession: toolCall("other-session-call", otherThread!, "running", 3), + }; + yield* eventSink.write({ + events: yield* Effect.forEach(Object.values(items), (payload) => + Effect.gen(function* () { + return { + id: yield* idAllocator.allocate.event({ threadId: input.threadId }), + type: "turn-item.updated" as const, + threadId: input.threadId, + driver: CODEX_DRIVER, + occurredAt: now, + payload, + }; + }), + ), + }); + const statuses = Effect.gen(function* () { + const projection = yield* projectionStore.getThreadProjection(input.threadId); + const statusOf = (item: OrchestrationV2TurnItem) => { + const stored = projection.turnItems.find((candidate) => candidate.id === item.id); + return { status: stored?.status, completed: stored?.completedAt != null }; + }; + return { + cutOff: statusOf(items.cutOff), + finished: statusOf(items.finished), + otherSession: statusOf(items.otherSession), + }; + }); + return { statuses }; + }); + +// regression: #15993 a worktree handoff left its tool call running, so the thread stayed Waiting +it.effect( + "ProviderSessionManagerV2 interrupts the open tool calls of a session a detach releases", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const effect = Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const { threadId, providerSessionId, providerThreads } = + yield* openSessionWithBackgroundCommand("provider-session-manager-detach-tool-calls"); + const { statuses } = yield* seedToolCalls({ threadId, providerThreads }); + + yield* manager.detach({ providerSessionId, threadId, detail: "Workspace changed." }); + + assert.deepStrictEqual(yield* statuses, { + cutOff: { status: "interrupted", completed: true }, + finished: { status: "completed", completed: true }, + otherSession: { status: "running", completed: false }, + }); + }); + + yield* effect.pipe( + Effect.provide( + layerTest({ state, idleTimeoutMs: 60_000, capabilities: ExclusiveCapabilities }), + ), + ); + }), +); + +it.effect( + "ProviderSessionManagerV2 keeps the tool calls of a runtime whose tasks can outlive it", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const effect = Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const { threadId, providerSessionId, providerThreads } = + yield* openSessionWithBackgroundCommand("provider-session-manager-outliving-tool-calls"); + const { statuses } = yield* seedToolCalls({ threadId, providerThreads }); + + yield* manager.close(providerSessionId); + + assert.deepStrictEqual((yield* statuses).cutOff, { status: "running", completed: false }); + }); + + yield* effect.pipe( + Effect.provide( + layerTest({ state, idleTimeoutMs: 60_000, backgroundTasksEndWithProcess: false }), + ), + ); + }), +); + // regression: #19355 settled and idle-released threads stayed Waiting it.effect("ProviderSessionManagerV2 keeps a provider update that lands while release clears", () => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 03a7278774a64..d0c96263be121 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -8,6 +8,7 @@ import { OrchestrationV2ProviderSession, type OrchestrationV2ProviderThread, OrchestrationV2RuntimeRequest, + type OrchestrationV2TurnItem, ProviderInstanceId, ProviderSessionId, ThreadId, @@ -79,6 +80,17 @@ export const ProviderSessionReleaseReason = Schema.Literals([ ]); export type ProviderSessionReleaseReason = typeof ProviderSessionReleaseReason.Type; +// Tool calls that only a live provider process can end. +const releasedToolCallTypes: ReadonlyArray = [ + "command_execution", + "dynamic_tool", +]; +const activeTurnItemStatuses: ReadonlyArray = [ + "pending", + "running", + "waiting", +]; + export class ProviderSessionOpenError extends Schema.TaggedError()( "ProviderSessionOpenError", { @@ -741,8 +753,9 @@ export const layerWithOptions = ( } }); - // Read tasks as soon as the session leaves the live map, before a replacement reports its own. - // Keep OpenCode's roster: its background subagents share a server and can outlive the session. + // Read tasks and open tool calls as soon as the session leaves the live map, before a + // replacement reports its own. Keep OpenCode's roster: its background subagents share a + // server and can outlive the session. const readReleasedBackgroundTasks = Effect.fn( "ProviderSessionManager.readReleasedBackgroundTasks", )( @@ -762,19 +775,29 @@ export const layerWithOptions = ( if (input.detachedThreadId !== undefined) threadIds.add(input.detachedThreadId); const released = []; for (const threadId of threadIds) { - const projection = yield* projectionStore.getThreadRecords(threadId, [ - "providerThreads", - ]); + const projection = yield* projectionStore.getThreadRecords( + threadId, + ["providerThreads", "turnItems"], + { + turnItemTypes: releasedToolCallTypes, + turnItemStatuses: activeTurnItemStatuses, + }, + ); for (const providerThread of projection.providerThreads) { + if (providerThread.providerSessionId !== providerSessionId) continue; const tasks = providerThread.pendingBackgroundTasks ?? []; - if (providerThread.providerSessionId !== providerSessionId || tasks.length === 0) { - continue; - } + const turnItemIds = new Set( + projection.turnItems + .filter((item) => item.providerThreadId === providerThread.id) + .map((item) => item.id), + ); + if (tasks.length === 0 && turnItemIds.size === 0) continue; released.push({ threadId, providerSessionId, providerThreadId: providerThread.id, taskIds: new Set(tasks.map((task) => task.taskId)), + turnItemIds, }); } } @@ -795,13 +818,15 @@ export const layerWithOptions = ( >[number]; // Dead processes cannot report task endings; stale rosters keep threads Waiting. - const writeReleasedBackgroundWorkEvents = ( - released: ReadonlyArray, - ) => - Effect.forEach( - released, - ({ threadId, providerSessionId, providerThreadId, taskIds }) => - Effect.gen(function* () { + const writeReleasedRosterEvents = ({ + threadId, + providerSessionId, + providerThreadId, + taskIds, + }: ReleasedBackgroundTasks) => + taskIds.size === 0 + ? Effect.void + : Effect.gen(function* () { for (let attempt = 0; attempt < 3; attempt++) { const projection = yield* projectionStore.getThreadRecords(threadId, [ "providerThreads", @@ -840,7 +865,49 @@ export const layerWithOptions = ( providerThreadId, }, ); - }), + }); + + // Tool calls cut off by the dead process never report an end either. Run finalization + // keeps them open for a late result, so they would stay Waiting forever. + const writeReleasedToolCallEvents = ({ + threadId, + providerSessionId, + turnItemIds, + }: ReleasedBackgroundTasks) => + turnItemIds.size === 0 + ? Effect.void + : Effect.gen(function* () { + const projection = yield* projectionStore.getThreadRecords(threadId, ["turnItems"], { + turnItemTypes: releasedToolCallTypes, + turnItemStatuses: activeTurnItemStatuses, + }); + const stillOpen = projection.turnItems.filter((item) => turnItemIds.has(item.id)); + if (stillOpen.length === 0) return; + const now = yield* DateTime.now; + const events: Array = []; + for (const item of stillOpen) { + events.push({ + id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), + type: "turn-item.updated", + threadId, + ...(item.runId === null ? {} : { runId: item.runId }), + ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), + occurredAt: now, + payload: { ...item, status: "interrupted", completedAt: now, updatedAt: now }, + }); + } + yield* eventSink.write({ events }); + }); + + const writeReleasedBackgroundWorkEvents = ( + released: ReadonlyArray, + ) => + Effect.forEach( + released, + (entry) => + writeReleasedRosterEvents(entry).pipe( + Effect.andThen(writeReleasedToolCallEvents(entry)), + ), { discard: true }, ); diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index bb21d8ca9929a..83e7d9923f6de 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -8,6 +8,7 @@ import { MessageId, type ModelSelection, type OrchestrationV2Command, + type OrchestrationV2DomainEvent, type OrchestrationV2ProviderCapabilities, type OrchestrationV2ProviderSession, type OrchestrationV2ProviderThread, @@ -22,6 +23,7 @@ import { TurnItemId, ProviderDriverKind, } from "@t3tools/contracts"; +import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -119,6 +121,12 @@ function makeTestAdapter(input: { readonly subagentOnlyBackgroundWork?: boolean; readonly holdFirstTurn?: Deferred.Deferred; readonly releaseFirstTurn?: Deferred.Deferred; + /** A tool call the held turn reports running before it holds. */ + readonly heldRunningToolCall?: { + readonly id: TurnItemId; + readonly toolName: string; + }; + readonly backgroundTasksEndWithProcess?: true; }): ProviderAdapter.ProviderAdapterV2["Service"] { return { instanceId: input.instanceId, @@ -148,6 +156,9 @@ function makeTestAdapter(input: { providerSessionId: sessionInput.providerSessionId, providerSession, events: Stream.fromPubSub(events), + ...(input.backgroundTasksEndWithProcess === undefined + ? {} + : { backgroundTasksEndWithProcess: input.backgroundTasksEndWithProcess }), ...(input.pendingBackgroundWork === undefined ? {} : { @@ -263,18 +274,59 @@ function makeTestAdapter(input: { : { nativeThreadHasTurns: turnInput.nativeThreadHasTurns }), }, ]); + const providerTurnId = ProviderTurnId.make( + `provider-turn:${input.driver}:${turnInput.threadId}:${turnInput.runOrdinal}`, + ); if ( turnInput.runOrdinal === (input.holdRunOrdinal ?? 1) && input.holdFirstTurn !== undefined ) { + if (input.heldRunningToolCall !== undefined) { + const startedAt = yield* DateTime.now; + yield* PubSub.publish(events, { + type: "provider_turn.updated", + driver: input.driver, + providerTurn: { + id: providerTurnId, + providerThreadId: turnInput.providerThread.id, + nodeId: turnInput.rootNodeId, + runAttemptId: turnInput.attemptId, + nativeTurnRef: null, + ordinal: turnInput.runOrdinal, + status: "running", + startedAt, + completedAt: null, + }, + }); + yield* PubSub.publish(events, { + type: "turn_item.updated", + driver: input.driver, + turnItem: { + id: input.heldRunningToolCall.id, + threadId: turnInput.threadId, + runId: turnInput.runId, + nodeId: turnInput.rootNodeId, + providerThreadId: turnInput.providerThread.id, + providerTurnId, + nativeItemRef: null, + parentItemId: null, + ordinal: turnInput.runOrdinal * 100 + 1, + status: "running", + title: null, + startedAt, + completedAt: null, + updatedAt: startedAt, + type: "dynamic_tool", + toolName: input.heldRunningToolCall.toolName, + input: {}, + }, + }); + } yield* Deferred.succeed(input.holdFirstTurn, undefined); if (input.releaseFirstTurn === undefined) return; yield* Deferred.await(input.releaseFirstTurn); } const eventTime = yield* DateTime.now; - const providerTurnId = ProviderTurnId.make( - `provider-turn:${input.driver}:${turnInput.threadId}:${turnInput.runOrdinal}`, - ); const terminalStatus = input.failedRunOrdinals?.has(turnInput.runOrdinal) ? "failed" : input.interruptedRunOrdinals?.has(turnInput.runOrdinal) @@ -1760,6 +1812,136 @@ describe("orchestration v2 provider switching", () => { ), ); + // regression: #15993 the agent releases its own Claude session through a tool call, by + // handing the thread off to a worktree or by switching its provider. Nothing is left to + // report how the call ended, so it stayed running and the thread stayed Waiting. + it.live.each(["handoff", "switch"] as const)( + "interrupts a tool call that a %s cuts off by releasing its session", + (release) => + Effect.scoped( + Effect.gen(function* () { + const key = `${release}-cuts-off-tool-call`; + const worktreePath = yield* checkpointWorkspace(key); + const capturedTurns = yield* Ref.make>([]); + const started = yield* Deferred.make(); + const toolCallId = TurnItemId.make(`turn-item:${key}`); + const scenarioThreadId = ThreadId.make(`thread:${key}`); + const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([ + makeTestAdapter({ + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + holdFirstTurn: started, + heldRunningToolCall: { + id: toolCallId, + toolName: + release === "handoff" + ? "mcp__t3-code__t3_worktree_handoff" + : "mcp__t3-code__t3_thread_configure", + }, + backgroundTasksEndWithProcess: true, + }), + makeTestAdapter({ + instanceId: CODEX_MODEL_SELECTION.instanceId, + driver: CODEX_DRIVER, + capabilities: CodexProviderCapabilitiesV2, + modelSelection: CODEX_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + }), + ]); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const awaitEvent = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:${key}:create`), + threadId: scenarioThreadId, + projectId: ProjectId.make(`project:${key}`), + title: "Tool call releases its session", + modelSelection: CLAUDE_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + const toolCallRunning = yield* awaitEvent( + (event) => + event.type === "turn-item.updated" && + event.payload.id === toolCallId && + event.payload.status === "running", + ); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:${key}:message`), + threadId: scenarioThreadId, + messageId: MessageId.make(`message:${key}`), + text: "Release this session", + attachments: [], + modelSelection: CLAUDE_MODEL_SELECTION, + dispatchMode: { type: "start_immediately" }, + }); + yield* Deferred.await(started); + yield* Fiber.join(toolCallRunning); + + const runFailed = yield* awaitEvent( + (event) => event.type === "run.updated" && event.payload.status === "failed", + ); + yield* orchestrator.dispatch( + release === "handoff" + ? { + type: "thread.metadata.update", + commandId: CommandId.make(`command:${key}:release`), + threadId: scenarioThreadId, + branch: `t3/${key}`, + worktreePath, + expectedWorktreePath: null, + } + : { + type: "provider.switch", + commandId: CommandId.make(`command:${key}:release`), + threadId: scenarioThreadId, + modelSelection: CODEX_MODEL_SELECTION, + }, + ); + yield* worker.drain(); + yield* Fiber.join(runFailed); + + const projection = yield* orchestrator.getThreadProjection(scenarioThreadId); + const toolCall = projection.turnItems.find((item) => item.id === toolCallId); + assert.equal(toolCall?.status, "interrupted"); + assert.isNotNull(toolCall?.completedAt); + assert.deepEqual( + derivePendingBackgroundWork({ + latestRun: projection.runs.at(-1), + providerThreads: projection.providerThreads, + turnItems: projection.turnItems, + activeProviderThreadId: projection.thread.activeProviderThreadId, + runs: projection.runs, + }), + [], + ); + }).pipe( + Effect.provide(ProviderReplayHarness.layerWithRegistry({ name: key }, layerRegistry)), + ); + }), + ), + ); + it.live("checks the queued provider's capability while the current provider stays running", () => Effect.scoped( Effect.gen(function* () { From 4fea1ebc76c52c5361d4d1606b8c50cc72bc8ad1 Mon Sep 17 00:00:00 2001 From: Spike Date: Mon, 2 Nov 2026 03:14:43 +0000 Subject: [PATCH 2/9] fix(server): close cut-off tool calls without losing a late completion Review repairs: - Select the cut-off items at write time: open tool calls on the released provider threads that started at or before the release. This includes an item that the dead session's queued events stored after the release began. Items that a replacement session started later stay open. - Write the interrupt only if each item row is unchanged (EventSink.writeIfTurnItemsUnchanged), with three tries like the roster. A late completion from the dead session now wins over the interrupt. - Use Effect.fn spans for both release writers, and set driver on the events. - State in the backgroundTasksEndWithProcess comment that the flag also covers tool calls. - Extend the existing roster tests instead of adding new ones, add the server shutdown row, and keep one integration case (the handoff). Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/EventSink.ts | 60 +++++ .../src/orchestration-v2/ProjectionStore.ts | 2 +- .../ProviderSessionManager.test.ts | 252 +++++++++--------- .../ProviderSessionManager.ts | 211 +++++++-------- .../ProviderSwitch.integration.test.ts | 233 ++++++++-------- .../src/server/ProviderAdapter.ts | 6 +- 6 files changed, 404 insertions(+), 360 deletions(-) diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index b210d3c9840ae..f4d8dadb097d1 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -3,6 +3,7 @@ import { type OrchestrationV2Run, OrchestrationV2DomainEvent, type OrchestrationV2ProviderThread, + type OrchestrationV2TurnItem, OrchestrationV2StoredEvent, ProviderThreadId, RunAttemptId, @@ -135,6 +136,17 @@ export interface EventSinkV2Shape { }, EventSinkV2Error >; + /** Commits only if each expected turn item row still equals its expected payload. */ + readonly writeIfTurnItemsUnchanged: (input: { + readonly expected: ReadonlyArray; + readonly events: ReadonlyArray; + }) => Effect.Effect< + { + readonly committed: boolean; + readonly storedEvents: ReadonlyArray; + }, + EventSinkV2Error + >; readonly commitCommand: (input: { readonly commandId: CommandId; readonly threadId: ThreadId; @@ -616,6 +628,48 @@ const layerBase: Layer.Layer< ); }); + const writeIfTurnItemsUnchangedEffect = Effect.fn( + "orchestrationV2.EventSink.writeIfTurnItemsUnchanged", + )(function* (input: Parameters[0]) { + yield* Effect.annotateCurrentSpan({ + "orchestration_v2.event_count": input.events.length, + "orchestration_v2.turn_item_count": input.expected.length, + }); + const expected = yield* Effect.forEach(input.expected, (item) => + ProjectionStore.encodeTurnItemPayload(item).pipe( + Effect.map((payloadJson) => ({ id: item.id, payloadJson })), + ), + ); + + return yield* commitThenPublish( + Effect.gen(function* () { + for (const item of expected) { + const rows = yield* sql<{ readonly payload_json: string }>` + SELECT payload_json + FROM orchestration_v2_projection_turn_items + WHERE turn_item_id = ${item.id} + LIMIT 1 + `; + if (rows[0]?.payload_json !== item.payloadJson) { + return { + committed: false as const, + storedEvents: [] as ReadonlyArray, + }; + } + } + + const normalized = yield* normalizeEvents(input.events); + const storedEvents = yield* eventStore.append({ events: normalized }); + const outcomes = yield* applyStoredEvents(storedEvents); + return { committed: true as const, storedEvents, outcomes }; + }), + (result) => + result.committed + ? publishStoredEvents(result.storedEvents, result.outcomes) + : Effect.void, + ); + }); + const existingCommandResult = /* This helper runs for individual orchestration events; the enclosing operation owns the span. */ Effect.fnUntraced(function* (commandId: CommandId) { @@ -928,6 +982,12 @@ const layerBase: Layer.Layer< (cause) => new EventSinkWriteError({ eventCount: input.events.length, cause }), ), ), + writeIfTurnItemsUnchanged: (input) => + writeIfTurnItemsUnchangedEffect(input).pipe( + Effect.mapError( + (cause) => new EventSinkWriteError({ eventCount: input.events.length, cause }), + ), + ), commitCommand: (input) => commitCommandEffect(input).pipe( Effect.mapError( diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 975ce7b0b9c50..46463cca0df19 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1061,7 +1061,7 @@ const encodeMessagePayload = Schema.encodeEffect( const encodePlanPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2PlanArtifactSchema), ); -const encodeTurnItemPayload = Schema.encodeEffect( +export const encodeTurnItemPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2TurnItemJsonSchema), ); const encodeCheckpointScopePayload = Schema.encodeEffect( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 4ddc1ce061f6e..bb49a0c0b2a7f 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -145,7 +145,10 @@ const layerPausingAttachEventSink = (pause: { }), ).pipe(Layer.provide(layerTestEventSink)); -const layerBeforeUnchangedWriteEventSink = (before: Ref.Ref>) => +const layerBeforeUnchangedWriteEventSink = ( + before: Ref.Ref>, + beforeTurnItems: Ref.Ref> | undefined, +) => Layer.effect( EventSink.EventSinkV2, Effect.gen(function* () { @@ -157,6 +160,11 @@ const layerBeforeUnchangedWriteEventSink = (before: Ref.Ref> Effect.flatten, Effect.andThen(delegate.writeIfProviderThreadUnchanged(input)), ), + writeIfTurnItemsUnchanged: (input) => + (beforeTurnItems === undefined + ? Effect.void + : Ref.getAndSet(beforeTurnItems, Effect.void).pipe(Effect.flatten) + ).pipe(Effect.andThen(delegate.writeIfTurnItemsUnchanged(input))), }); }), ).pipe(Layer.provide(layerTestEventSink)); @@ -478,6 +486,7 @@ function layerTest(input: { readonly flakyReleaseWrites?: FlakyReleaseWrites; readonly pauseAttachWrite?: Parameters[0]; readonly beforeUnchangedWrite?: Ref.Ref>; + readonly beforeTurnItemsWrite?: Ref.Ref>; readonly pauseResolve?: { readonly armed: Ref.Ref; readonly paused: Deferred.Deferred; @@ -500,7 +509,10 @@ function layerTest(input: { : input.pauseAttachWrite !== undefined ? layerPausingAttachEventSink(input.pauseAttachWrite) : input.beforeUnchangedWrite !== undefined - ? layerBeforeUnchangedWriteEventSink(input.beforeUnchangedWrite) + ? layerBeforeUnchangedWriteEventSink( + input.beforeUnchangedWrite, + input.beforeTurnItemsWrite, + ) : input.failReleaseEventWrites ? layerFailingReleaseEventSink : layerTestEventSink; @@ -2451,155 +2463,115 @@ const openSessionWithBackgroundCommand = (fixtureName: string) => events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], }); yield* Effect.forEach(providerThreads, writeProviderThread); - yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); - const rosters = Effect.gen(function* () { - const projectionStore = yield* ProjectionStore.ProjectionStoreV2; - const projection = yield* projectionStore.getThreadProjection(threadId); - return providerThreads.map((seeded) => - ( - projection.providerThreads.find((thread) => thread.id === seeded.id) - ?.pendingBackgroundTasks ?? [] - ).map((task) => task.taskId), - ); - }); - return { threadId, providerSessionId, providerThreads, writeProviderThread, rosters }; - }); - -it.effect( - "ProviderSessionManagerV2 clears the background roster of a session a detach releases", - () => - Effect.gen(function* () { - const state = yield* Ref.make(emptyState); - const effect = Effect.gen(function* () { - const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; - const { threadId, providerSessionId, rosters } = yield* openSessionWithBackgroundCommand( - "provider-session-manager-detach-roster", - ); - assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]); - - yield* manager.detach({ providerSessionId, threadId, detail: "Thread settled." }); - - assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); - assert.deepStrictEqual(yield* rosters, [[], ["dev-server"]]); - }); - - yield* effect.pipe( - Effect.provide( - layerTest({ state, idleTimeoutMs: 60_000, capabilities: ExclusiveCapabilities }), - ), - ); - }), -); - -it.effect("ProviderSessionManagerV2 keeps the roster of a runtime whose tasks can outlive it", () => - Effect.gen(function* () { - const state = yield* Ref.make(emptyState); - const effect = Effect.gen(function* () { - const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; - const { providerSessionId, rosters } = yield* openSessionWithBackgroundCommand( - "provider-session-manager-outliving-roster", - ); - - yield* manager.close(providerSessionId); - - assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]); - }); - - yield* effect.pipe( - Effect.provide( - layerTest({ state, idleTimeoutMs: 60_000, backgroundTasksEndWithProcess: false }), - ), - ); - }), -); - -const seedToolCalls = (input: { - readonly threadId: ThreadId; - readonly providerThreads: ReadonlyArray; -}) => - Effect.gen(function* () { - const eventSink = yield* EventSink.EventSinkV2; - const idAllocator = yield* IdAllocator.IdAllocatorV2; - const projectionStore = yield* ProjectionStore.ProjectionStoreV2; - const now = yield* DateTime.now; - const [releasedThread, otherThread] = input.providerThreads; const toolCall = ( nativeItemId: string, providerThread: OrchestrationV2ProviderThread, status: "running" | "completed", - ordinal: number, + startedAt: DateTime.Utc, ) => ({ id: idAllocator.derive.turnItemFromProviderItem({ driver: CODEX_DRIVER, nativeItemId }), - threadId: input.threadId, + threadId, runId: null, nodeId: null, providerThreadId: providerThread.id, providerTurnId: null, nativeItemRef: null, parentItemId: null, - ordinal, + ordinal: 1, status, title: null, - startedAt: now, - completedAt: status === "completed" ? now : null, - updatedAt: now, + startedAt, + completedAt: status === "completed" ? startedAt : null, + updatedAt: startedAt, type: "dynamic_tool", toolName: "mcp__t3-code__t3_worktree_handoff", input: {}, }) satisfies OrchestrationV2TurnItem; - const items = { - cutOff: toolCall("handoff-cut-off", releasedThread!, "running", 1), - finished: toolCall("finished-call", releasedThread!, "completed", 2), - otherSession: toolCall("other-session-call", otherThread!, "running", 3), - }; - yield* eventSink.write({ - events: yield* Effect.forEach(Object.values(items), (payload) => - Effect.gen(function* () { - return { - id: yield* idAllocator.allocate.event({ threadId: input.threadId }), - type: "turn-item.updated" as const, - threadId: input.threadId, - driver: CODEX_DRIVER, - occurredAt: now, - payload, - }; - }), + const seededToolCalls = { + cutOff: toolCall("cut-off", providerThreads[0]!, "running", now), + finished: toolCall("finished", providerThreads[0]!, "completed", now), + otherSession: toolCall("other-session", providerThreads[1]!, "running", now), + // Started by a replacement session after the release. + replacement: toolCall( + "replacement", + providerThreads[0]!, + "running", + DateTime.add(now, { hours: 1 }), ), - }); - const statuses = Effect.gen(function* () { - const projection = yield* projectionStore.getThreadProjection(input.threadId); - const statusOf = (item: OrchestrationV2TurnItem) => { - const stored = projection.turnItems.find((candidate) => candidate.id === item.id); - return { status: stored?.status, completed: stored?.completedAt != null }; - }; + }; + const writeToolCall = (payload: OrchestrationV2TurnItem) => + Effect.gen(function* () { + yield* eventSink.write({ + events: [ + { + id: yield* idAllocator.allocate.event({ threadId }), + type: "turn-item.updated", + threadId, + driver: CODEX_DRIVER, + occurredAt: now, + payload, + }, + ], + }); + }); + yield* Effect.forEach(Object.values(seededToolCalls), writeToolCall); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + const toolCalls = Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const projection = yield* projectionStore.getThreadProjection(threadId); + const statusOf = (seeded: OrchestrationV2TurnItem) => + projection.turnItems.find((item) => item.id === seeded.id)?.status; return { - cutOff: statusOf(items.cutOff), - finished: statusOf(items.finished), - otherSession: statusOf(items.otherSession), + cutOff: statusOf(seededToolCalls.cutOff), + finished: statusOf(seededToolCalls.finished), + otherSession: statusOf(seededToolCalls.otherSession), + replacement: statusOf(seededToolCalls.replacement), }; }); - return { statuses }; + const rosters = Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const projection = yield* projectionStore.getThreadProjection(threadId); + return providerThreads.map((seeded) => + ( + projection.providerThreads.find((thread) => thread.id === seeded.id) + ?.pendingBackgroundTasks ?? [] + ).map((task) => task.taskId), + ); + }); + return { + threadId, + providerSessionId, + providerThreads, + writeProviderThread, + rosters, + seededToolCalls, + writeToolCall, + toolCalls, + }; }); // regression: #15993 a worktree handoff left its tool call running, so the thread stayed Waiting it.effect( - "ProviderSessionManagerV2 interrupts the open tool calls of a session a detach releases", + "ProviderSessionManagerV2 clears the background work of a session a detach releases", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); const effect = Effect.gen(function* () { const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; - const { threadId, providerSessionId, providerThreads } = - yield* openSessionWithBackgroundCommand("provider-session-manager-detach-tool-calls"); - const { statuses } = yield* seedToolCalls({ threadId, providerThreads }); + const { threadId, providerSessionId, rosters, toolCalls } = + yield* openSessionWithBackgroundCommand("provider-session-manager-detach-roster"); + assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]); yield* manager.detach({ providerSessionId, threadId, detail: "Workspace changed." }); - assert.deepStrictEqual(yield* statuses, { - cutOff: { status: "interrupted", completed: true }, - finished: { status: "completed", completed: true }, - otherSession: { status: "running", completed: false }, + assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); + assert.deepStrictEqual(yield* rosters, [[], ["dev-server"]]); + assert.deepStrictEqual(yield* toolCalls, { + cutOff: "interrupted", + finished: "completed", + otherSession: "running", + replacement: "running", }); }); @@ -2611,39 +2583,58 @@ it.effect( }), ); -it.effect( - "ProviderSessionManagerV2 keeps the tool calls of a runtime whose tasks can outlive it", - () => +it.effect.each(["a runtime whose tasks can outlive it", "server shutdown"] as const)( + "ProviderSessionManagerV2 keeps the background work after the release of %s", + (release) => Effect.gen(function* () { const state = yield* Ref.make(emptyState); const effect = Effect.gen(function* () { const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; - const { threadId, providerSessionId, providerThreads } = - yield* openSessionWithBackgroundCommand("provider-session-manager-outliving-tool-calls"); - const { statuses } = yield* seedToolCalls({ threadId, providerThreads }); + const { providerSessionId, rosters, toolCalls } = yield* openSessionWithBackgroundCommand( + "provider-session-manager-outliving-roster", + ); - yield* manager.close(providerSessionId); + yield* release === "server shutdown" + ? manager.release({ providerSessionId, reason: "server_shutdown" }) + : manager.close(providerSessionId); - assert.deepStrictEqual((yield* statuses).cutOff, { status: "running", completed: false }); + assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]); + assert.equal((yield* toolCalls).cutOff, "running"); }); yield* effect.pipe( Effect.provide( - layerTest({ state, idleTimeoutMs: 60_000, backgroundTasksEndWithProcess: false }), + layerTest({ + state, + idleTimeoutMs: 60_000, + ...(release === "server shutdown" ? {} : { backgroundTasksEndWithProcess: false }), + }), ), ); }), ); // regression: #19355 settled and idle-released threads stayed Waiting -it.effect("ProviderSessionManagerV2 keeps a provider update that lands while release clears", () => +it.effect("ProviderSessionManagerV2 keeps provider updates that land while release clears", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); const beforeUnchangedWrite = yield* Ref.make>(Effect.void); + const beforeTurnItemsWrite = yield* Ref.make>(Effect.void); const effect = Effect.gen(function* () { const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; - const { providerSessionId, providerThreads, writeProviderThread, rosters } = - yield* openSessionWithBackgroundCommand("provider-session-manager-raced-roster"); + const { + providerSessionId, + providerThreads, + writeProviderThread, + rosters, + seededToolCalls, + writeToolCall, + toolCalls, + } = yield* openSessionWithBackgroundCommand("provider-session-manager-raced-roster"); + yield* Ref.set( + beforeTurnItemsWrite, + writeToolCall({ ...seededToolCalls.cutOff, status: "completed" }).pipe(Effect.orDie), + ); const seeded = providerThreads[0]!; yield* Ref.set( beforeUnchangedWrite, @@ -2659,10 +2650,13 @@ it.effect("ProviderSessionManagerV2 keeps a provider update that lands while rel yield* manager.close(providerSessionId); assert.deepStrictEqual(yield* rosters, [["replacement-task"], ["dev-server"]]); + assert.equal((yield* toolCalls).cutOff, "completed"); }); yield* effect.pipe( - Effect.provide(layerTest({ state, idleTimeoutMs: 60_000, beforeUnchangedWrite })), + Effect.provide( + layerTest({ state, idleTimeoutMs: 60_000, beforeUnchangedWrite, beforeTurnItemsWrite }), + ), ); }), ); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index d0c96263be121..d87cbd9042bfe 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -753,11 +753,11 @@ export const layerWithOptions = ( } }); - // Read tasks and open tool calls as soon as the session leaves the live map, before a - // replacement reports its own. Keep OpenCode's roster: its background subagents share a - // server and can outlive the session. - const readReleasedBackgroundTasks = Effect.fn( - "ProviderSessionManager.readReleasedBackgroundTasks", + // Read the session's provider threads as soon as it leaves the live map, before a + // replacement reports its own tasks. Keep OpenCode's roster: its background subagents share + // a server and can outlive the session. + const readReleasedProviderThreads = Effect.fn( + "ProviderSessionManager.readReleasedProviderThreads", )( function* (input: { readonly entry: LiveSessionEntry; @@ -775,29 +775,19 @@ export const layerWithOptions = ( if (input.detachedThreadId !== undefined) threadIds.add(input.detachedThreadId); const released = []; for (const threadId of threadIds) { - const projection = yield* projectionStore.getThreadRecords( - threadId, - ["providerThreads", "turnItems"], - { - turnItemTypes: releasedToolCallTypes, - turnItemStatuses: activeTurnItemStatuses, - }, - ); + const projection = yield* projectionStore.getThreadRecords(threadId, [ + "providerThreads", + ]); for (const providerThread of projection.providerThreads) { if (providerThread.providerSessionId !== providerSessionId) continue; - const tasks = providerThread.pendingBackgroundTasks ?? []; - const turnItemIds = new Set( - projection.turnItems - .filter((item) => item.providerThreadId === providerThread.id) - .map((item) => item.id), - ); - if (tasks.length === 0 && turnItemIds.size === 0) continue; released.push({ threadId, providerSessionId, providerThreadId: providerThread.id, - taskIds: new Set(tasks.map((task) => task.taskId)), - turnItemIds, + driver: providerThread.driver, + taskIds: new Set( + (providerThread.pendingBackgroundTasks ?? []).map((task) => task.taskId), + ), }); } } @@ -813,100 +803,109 @@ export const layerWithOptions = ( ), ), ); - type ReleasedBackgroundTasks = Effect.Success< - ReturnType + type ReleasedProviderThread = Effect.Success< + ReturnType >[number]; // Dead processes cannot report task endings; stale rosters keep threads Waiting. - const writeReleasedRosterEvents = ({ + const writeReleasedRosterEvents = Effect.fn( + "ProviderSessionManager.writeReleasedRosterEvents", + )(function* ({ threadId, providerSessionId, providerThreadId, taskIds, - }: ReleasedBackgroundTasks) => - taskIds.size === 0 - ? Effect.void - : Effect.gen(function* () { - for (let attempt = 0; attempt < 3; attempt++) { - const projection = yield* projectionStore.getThreadRecords(threadId, [ - "providerThreads", - ]); - const providerThread = projection.providerThreads.find( - (candidate) => candidate.id === providerThreadId, - ); - const tasks = providerThread?.pendingBackgroundTasks ?? []; - const remaining = tasks.filter((task) => !taskIds.has(task.taskId)); - if (providerThread === undefined || remaining.length === tasks.length) return; - const now = yield* DateTime.now; - const { committed } = yield* eventSink.writeIfProviderThreadUnchanged({ - expected: providerThread, - events: [ - { - id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), - type: "provider-thread.updated", - threadId, - driver: providerThread.driver, - providerInstanceId: providerThread.providerInstanceId, - occurredAt: now, - payload: { - ...providerThread, - pendingBackgroundTasks: remaining, - updatedAt: now, - }, - }, - ], - }); - if (committed) return; - } - yield* Effect.logWarning( - "orchestration-v2.provider-session-release-roster-conflict", - { - providerSessionId, - providerThreadId, + }: ReleasedProviderThread) { + if (taskIds.size === 0) return; + for (let attempt = 0; attempt < 3; attempt++) { + const projection = yield* projectionStore.getThreadRecords(threadId, ["providerThreads"]); + const providerThread = projection.providerThreads.find( + (candidate) => candidate.id === providerThreadId, + ); + const tasks = providerThread?.pendingBackgroundTasks ?? []; + const remaining = tasks.filter((task) => !taskIds.has(task.taskId)); + if (providerThread === undefined || remaining.length === tasks.length) return; + const now = yield* DateTime.now; + const { committed } = yield* eventSink.writeIfProviderThreadUnchanged({ + expected: providerThread, + events: [ + { + id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), + type: "provider-thread.updated", + threadId, + driver: providerThread.driver, + providerInstanceId: providerThread.providerInstanceId, + occurredAt: now, + payload: { + ...providerThread, + pendingBackgroundTasks: remaining, + updatedAt: now, }, - ); - }); + }, + ], + }); + if (committed) return; + } + yield* Effect.logWarning("orchestration-v2.provider-session-release-roster-conflict", { + providerSessionId, + providerThreadId, + }); + }); - // Tool calls cut off by the dead process never report an end either. Run finalization - // keeps them open for a late result, so they would stay Waiting forever. - const writeReleasedToolCallEvents = ({ - threadId, - providerSessionId, - turnItemIds, - }: ReleasedBackgroundTasks) => - turnItemIds.size === 0 - ? Effect.void - : Effect.gen(function* () { - const projection = yield* projectionStore.getThreadRecords(threadId, ["turnItems"], { - turnItemTypes: releasedToolCallTypes, - turnItemStatuses: activeTurnItemStatuses, - }); - const stillOpen = projection.turnItems.filter((item) => turnItemIds.has(item.id)); - if (stillOpen.length === 0) return; - const now = yield* DateTime.now; - const events: Array = []; - for (const item of stillOpen) { - events.push({ - id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), - type: "turn-item.updated", - threadId, - ...(item.runId === null ? {} : { runId: item.runId }), - ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), - occurredAt: now, - payload: { ...item, status: "interrupted", completedAt: now, updatedAt: now }, - }); - } - yield* eventSink.write({ events }); + // Tool calls the dead process started never report an end either, and run finalization + // keeps them open for a late result. Items started after the release belong to a + // replacement session. Subagent rows are left out: run finalization ends run-owned ones. + const writeReleasedToolCallEvents = Effect.fn( + "ProviderSessionManager.writeReleasedToolCallEvents", + )(function* (released: ReleasedProviderThread, releasedAt: DateTime.Utc) { + const { threadId, providerSessionId, providerThreadId, driver } = released; + for (let attempt = 0; attempt < 3; attempt++) { + const projection = yield* projectionStore.getThreadRecords(threadId, ["turnItems"], { + turnItemTypes: releasedToolCallTypes, + turnItemStatuses: activeTurnItemStatuses, + }); + const cutOff = projection.turnItems.filter( + (item) => + item.providerThreadId === providerThreadId && + DateTime.isLessThanOrEqualTo(item.startedAt ?? item.updatedAt, releasedAt), + ); + if (cutOff.length === 0) return; + const now = yield* DateTime.now; + const events: Array = []; + for (const item of cutOff) { + events.push({ + id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), + type: "turn-item.updated", + threadId, + ...(item.runId === null ? {} : { runId: item.runId }), + ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), + driver, + occurredAt: now, + payload: { ...item, status: "interrupted", completedAt: now, updatedAt: now }, }); + } + // A late completion from the dead session's queued events wins over the interrupt. + const { committed } = yield* eventSink.writeIfTurnItemsUnchanged({ + expected: cutOff, + events, + }); + if (committed) return; + } + yield* Effect.logWarning("orchestration-v2.provider-session-release-tool-call-conflict", { + providerSessionId, + providerThreadId, + }); + }); const writeReleasedBackgroundWorkEvents = ( - released: ReadonlyArray, + released: ReadonlyArray, + releasedAt: DateTime.Utc, ) => Effect.forEach( released, - (entry) => - writeReleasedRosterEvents(entry).pipe( - Effect.andThen(writeReleasedToolCallEvents(entry)), + (providerThread) => + writeReleasedRosterEvents(providerThread).pipe( + Effect.andThen(writeReleasedToolCallEvents(providerThread, releasedAt)), ), { discard: true }, ); @@ -915,7 +914,7 @@ export const layerWithOptions = ( readonly entry: LiveSessionEntry; readonly reason: ProviderSessionReleaseReason; readonly detail?: string; - readonly releasedBackgroundTasks: ReadonlyArray; + readonly releasedProviderThreads: ReadonlyArray; readonly releasedAt: DateTime.Utc; readonly replaced: boolean; }) => @@ -929,7 +928,9 @@ export const layerWithOptions = ( input.entry.requestEventPermit.withPermits(1), ), ), - Effect.exit(writeReleasedBackgroundWorkEvents(input.releasedBackgroundTasks)), + Effect.exit( + writeReleasedBackgroundWorkEvents(input.releasedProviderThreads, input.releasedAt), + ), ], { concurrency: 1 }, ).pipe(Effect.flatMap(Exit.asVoidAll)); @@ -1086,7 +1087,7 @@ export const layerWithOptions = ( onNone: () => Effect.void, onSome: (entry) => Effect.gen(function* () { - const releasedBackgroundTasks = yield* readReleasedBackgroundTasks({ + const releasedProviderThreads = yield* readReleasedProviderThreads({ entry, reason: input.reason, ...(input.detachedThreadId === undefined @@ -1114,7 +1115,7 @@ export const layerWithOptions = ( entry, reason: input.reason, ...(input.detail === undefined ? {} : { detail: input.detail }), - releasedBackgroundTasks, + releasedProviderThreads, releasedAt, }; const recorded = yield* Effect.exit( diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 83e7d9923f6de..d9adead60d86d 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -1812,134 +1812,119 @@ describe("orchestration v2 provider switching", () => { ), ); - // regression: #15993 the agent releases its own Claude session through a tool call, by - // handing the thread off to a worktree or by switching its provider. Nothing is left to - // report how the call ended, so it stayed running and the thread stayed Waiting. - it.live.each(["handoff", "switch"] as const)( - "interrupts a tool call that a %s cuts off by releasing its session", - (release) => - Effect.scoped( - Effect.gen(function* () { - const key = `${release}-cuts-off-tool-call`; - const worktreePath = yield* checkpointWorkspace(key); - const capturedTurns = yield* Ref.make>([]); - const started = yield* Deferred.make(); - const toolCallId = TurnItemId.make(`turn-item:${key}`); - const scenarioThreadId = ThreadId.make(`thread:${key}`); - const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([ - makeTestAdapter({ - instanceId: CLAUDE_MODEL_SELECTION.instanceId, - driver: CLAUDE_DRIVER, - capabilities: ClaudeProviderCapabilitiesV2, - modelSelection: CLAUDE_MODEL_SELECTION, - responseByRunOrdinal: {}, - capturedTurns, - holdFirstTurn: started, - heldRunningToolCall: { - id: toolCallId, - toolName: - release === "handoff" - ? "mcp__t3-code__t3_worktree_handoff" - : "mcp__t3-code__t3_thread_configure", - }, - backgroundTasksEndWithProcess: true, - }), - makeTestAdapter({ - instanceId: CODEX_MODEL_SELECTION.instanceId, - driver: CODEX_DRIVER, - capabilities: CodexProviderCapabilitiesV2, - modelSelection: CODEX_MODEL_SELECTION, - responseByRunOrdinal: {}, - capturedTurns, - }), - ]); - yield* Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; - const awaitEvent = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => - orchestrator.streamDomainEvents.pipe( - Stream.filter(predicate), - Stream.take(1), - Stream.runDrain, - Effect.forkScoped, - ); - yield* orchestrator.dispatch({ - type: "thread.create", - createdBy: "user", - creationSource: "web", - commandId: CommandId.make(`command:${key}:create`), - threadId: scenarioThreadId, - projectId: ProjectId.make(`project:${key}`), - title: "Tool call releases its session", - modelSelection: CLAUDE_MODEL_SELECTION, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - }); - const toolCallRunning = yield* awaitEvent( - (event) => - event.type === "turn-item.updated" && - event.payload.id === toolCallId && - event.payload.status === "running", + // regression: #15993 t3_worktree_handoff releases the Claude session that runs it. Nothing + // was left to report how the call ended, so it stayed running and the thread stayed Waiting. + it.live("interrupts the handoff tool call that releases its own session", () => + Effect.scoped( + Effect.gen(function* () { + const key = "handoff-cuts-off-tool-call"; + const worktreePath = yield* checkpointWorkspace(key); + const capturedTurns = yield* Ref.make>([]); + const started = yield* Deferred.make(); + const toolCallId = TurnItemId.make(`turn-item:${key}`); + const scenarioThreadId = ThreadId.make(`thread:${key}`); + const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([ + makeTestAdapter({ + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + holdFirstTurn: started, + heldRunningToolCall: { + id: toolCallId, + toolName: "mcp__t3-code__t3_worktree_handoff", + }, + backgroundTasksEndWithProcess: true, + }), + makeTestAdapter({ + instanceId: CODEX_MODEL_SELECTION.instanceId, + driver: CODEX_DRIVER, + capabilities: CodexProviderCapabilitiesV2, + modelSelection: CODEX_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + }), + ]); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const awaitEvent = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, ); - yield* orchestrator.dispatch({ - type: "message.dispatch", - createdBy: "user", - creationSource: "web", - commandId: CommandId.make(`command:${key}:message`), - threadId: scenarioThreadId, - messageId: MessageId.make(`message:${key}`), - text: "Release this session", - attachments: [], - modelSelection: CLAUDE_MODEL_SELECTION, - dispatchMode: { type: "start_immediately" }, - }); - yield* Deferred.await(started); - yield* Fiber.join(toolCallRunning); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:${key}:create`), + threadId: scenarioThreadId, + projectId: ProjectId.make(`project:${key}`), + title: "Tool call releases its session", + modelSelection: CLAUDE_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + const toolCallRunning = yield* awaitEvent( + (event) => + event.type === "turn-item.updated" && + event.payload.id === toolCallId && + event.payload.status === "running", + ); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:${key}:message`), + threadId: scenarioThreadId, + messageId: MessageId.make(`message:${key}`), + text: "Release this session", + attachments: [], + modelSelection: CLAUDE_MODEL_SELECTION, + dispatchMode: { type: "start_immediately" }, + }); + yield* Deferred.await(started); + yield* Fiber.join(toolCallRunning); - const runFailed = yield* awaitEvent( - (event) => event.type === "run.updated" && event.payload.status === "failed", - ); - yield* orchestrator.dispatch( - release === "handoff" - ? { - type: "thread.metadata.update", - commandId: CommandId.make(`command:${key}:release`), - threadId: scenarioThreadId, - branch: `t3/${key}`, - worktreePath, - expectedWorktreePath: null, - } - : { - type: "provider.switch", - commandId: CommandId.make(`command:${key}:release`), - threadId: scenarioThreadId, - modelSelection: CODEX_MODEL_SELECTION, - }, - ); - yield* worker.drain(); - yield* Fiber.join(runFailed); + const runFailed = yield* awaitEvent( + (event) => event.type === "run.updated" && event.payload.status === "failed", + ); + yield* orchestrator.dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make(`command:${key}:release`), + threadId: scenarioThreadId, + branch: `t3/${key}`, + worktreePath, + expectedWorktreePath: null, + }); + yield* worker.drain(); + yield* Fiber.join(runFailed); - const projection = yield* orchestrator.getThreadProjection(scenarioThreadId); - const toolCall = projection.turnItems.find((item) => item.id === toolCallId); - assert.equal(toolCall?.status, "interrupted"); - assert.isNotNull(toolCall?.completedAt); - assert.deepEqual( - derivePendingBackgroundWork({ - latestRun: projection.runs.at(-1), - providerThreads: projection.providerThreads, - turnItems: projection.turnItems, - activeProviderThreadId: projection.thread.activeProviderThreadId, - runs: projection.runs, - }), - [], - ); - }).pipe( - Effect.provide(ProviderReplayHarness.layerWithRegistry({ name: key }, layerRegistry)), + const projection = yield* orchestrator.getThreadProjection(scenarioThreadId); + const toolCall = projection.turnItems.find((item) => item.id === toolCallId); + assert.equal(toolCall?.status, "interrupted"); + assert.isNotNull(toolCall?.completedAt); + assert.deepEqual( + derivePendingBackgroundWork({ + latestRun: projection.runs.at(-1), + providerThreads: projection.providerThreads, + turnItems: projection.turnItems, + activeProviderThreadId: projection.thread.activeProviderThreadId, + runs: projection.runs, + }), + [], ); - }), - ), + }).pipe( + Effect.provide(ProviderReplayHarness.layerWithRegistry({ name: key }, layerRegistry)), + ); + }), + ), ); it.live("checks the queued provider's capability while the current provider stays running", () => diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index a92256d7dd791..3560b1a4041b3 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -585,7 +585,11 @@ export interface ProviderAdapterV2SessionRuntime { providerThread: OrchestrationV2ProviderThread, options?: { readonly includeSubagents?: boolean }, ) => Effect.Effect; - /** Set only if all session background tasks end with its process; otherwise threads must keep Waiting. */ + /** + * Set only if all session background tasks and tool calls end with its process. Releasing the + * session then clears its task roster and interrupts its open tool calls; otherwise threads + * must keep Waiting. + */ readonly backgroundTasksEndWithProcess?: true; readonly getModelContextWindow?: ( modelSelection: ModelSelection, From ecf406398f48050402034e3f3f77eeeb412b4b39 Mon Sep 17 00:00:00 2001 From: Spike Date: Mon, 2 Nov 2026 03:26:20 +0000 Subject: [PATCH 3/9] fix(server): interrupt each cut-off tool call with its own conditional write Review repair: one conditional write for all cut-off items let three late completions use up the three tries, and the other open items stayed running. Each item now gets its own conditional write. A late completion rejects only the interrupt of its own item, and that item is then no longer open. Items that are still open after a conflict, such as after a late output update, are read again and tried again, up to three times. EventSink.writeIfTurnItemUnchanged replaces the batch write, and both conditional writes share one helper. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/EventSink.ts | 103 ++++++++---------- .../ProviderSessionManager.test.ts | 66 ++++++++--- .../ProviderSessionManager.ts | 66 ++++++----- 3 files changed, 133 insertions(+), 102 deletions(-) diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index f4d8dadb097d1..0af8f1453155c 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -136,9 +136,9 @@ export interface EventSinkV2Shape { }, EventSinkV2Error >; - /** Commits only if each expected turn item row still equals its expected payload. */ - readonly writeIfTurnItemsUnchanged: (input: { - readonly expected: ReadonlyArray; + /** Commits only if the stored turn item still equals expected. */ + readonly writeIfTurnItemUnchanged: (input: { + readonly expected: OrchestrationV2TurnItem; readonly events: ReadonlyArray; }) => Effect.Effect< { @@ -592,23 +592,15 @@ const layerBase: Layer.Layer< ); }); - const writeIfProviderThreadUnchangedEffect = Effect.fn( - "orchestrationV2.EventSink.writeIfProviderThreadUnchanged", - )(function* (input: Parameters[0]) { - yield* Effect.annotateCurrentSpan({ - "orchestration_v2.event_count": input.events.length, - "orchestration_v2.provider_thread_id": input.expected.id, - }); - const expectedJson = yield* ProjectionStore.encodeProviderThreadPayload(input.expected); - - return yield* commitThenPublish( + // Appends the events only if the stored payload still matches, in one transaction. + const writeIfStoredPayloadMatches = ( + readStoredPayload: Effect.Effect, E>, + expectedJson: string, + events: ReadonlyArray, + ) => + commitThenPublish( Effect.gen(function* () { - const rows = yield* sql<{ readonly payload_json: string }>` - SELECT payload_json - FROM orchestration_v2_projection_provider_threads - WHERE provider_thread_id = ${input.expected.id} - LIMIT 1 - `; + const rows = yield* readStoredPayload; if (rows[0]?.payload_json !== expectedJson) { return { committed: false as const, @@ -616,7 +608,7 @@ const layerBase: Layer.Layer< }; } - const normalized = yield* normalizeEvents(input.events); + const normalized = yield* normalizeEvents(events); const storedEvents = yield* eventStore.append({ events: normalized }); const outcomes = yield* applyStoredEvents(storedEvents); return { committed: true as const, storedEvents, outcomes }; @@ -626,47 +618,44 @@ const layerBase: Layer.Layer< ? publishStoredEvents(result.storedEvents, result.outcomes) : Effect.void, ); - }); - const writeIfTurnItemsUnchangedEffect = Effect.fn( - "orchestrationV2.EventSink.writeIfTurnItemsUnchanged", - )(function* (input: Parameters[0]) { + const writeIfProviderThreadUnchangedEffect = Effect.fn( + "orchestrationV2.EventSink.writeIfProviderThreadUnchanged", + )(function* (input: Parameters[0]) { yield* Effect.annotateCurrentSpan({ "orchestration_v2.event_count": input.events.length, - "orchestration_v2.turn_item_count": input.expected.length, + "orchestration_v2.provider_thread_id": input.expected.id, }); - const expected = yield* Effect.forEach(input.expected, (item) => - ProjectionStore.encodeTurnItemPayload(item).pipe( - Effect.map((payloadJson) => ({ id: item.id, payloadJson })), - ), + const expectedJson = yield* ProjectionStore.encodeProviderThreadPayload(input.expected); + return yield* writeIfStoredPayloadMatches( + sql<{ readonly payload_json: string }>` + SELECT payload_json + FROM orchestration_v2_projection_provider_threads + WHERE provider_thread_id = ${input.expected.id} + LIMIT 1 + `, + expectedJson, + input.events, ); + }); - return yield* commitThenPublish( - Effect.gen(function* () { - for (const item of expected) { - const rows = yield* sql<{ readonly payload_json: string }>` - SELECT payload_json - FROM orchestration_v2_projection_turn_items - WHERE turn_item_id = ${item.id} - LIMIT 1 - `; - if (rows[0]?.payload_json !== item.payloadJson) { - return { - committed: false as const, - storedEvents: [] as ReadonlyArray, - }; - } - } - - const normalized = yield* normalizeEvents(input.events); - const storedEvents = yield* eventStore.append({ events: normalized }); - const outcomes = yield* applyStoredEvents(storedEvents); - return { committed: true as const, storedEvents, outcomes }; - }), - (result) => - result.committed - ? publishStoredEvents(result.storedEvents, result.outcomes) - : Effect.void, + const writeIfTurnItemUnchangedEffect = Effect.fn( + "orchestrationV2.EventSink.writeIfTurnItemUnchanged", + )(function* (input: Parameters[0]) { + yield* Effect.annotateCurrentSpan({ + "orchestration_v2.event_count": input.events.length, + "orchestration_v2.turn_item_id": input.expected.id, + }); + const expectedJson = yield* ProjectionStore.encodeTurnItemPayload(input.expected); + return yield* writeIfStoredPayloadMatches( + sql<{ readonly payload_json: string }>` + SELECT payload_json + FROM orchestration_v2_projection_turn_items + WHERE turn_item_id = ${input.expected.id} + LIMIT 1 + `, + expectedJson, + input.events, ); }); @@ -982,8 +971,8 @@ const layerBase: Layer.Layer< (cause) => new EventSinkWriteError({ eventCount: input.events.length, cause }), ), ), - writeIfTurnItemsUnchanged: (input) => - writeIfTurnItemsUnchangedEffect(input).pipe( + writeIfTurnItemUnchanged: (input) => + writeIfTurnItemUnchangedEffect(input).pipe( Effect.mapError( (cause) => new EventSinkWriteError({ eventCount: input.events.length, cause }), ), diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index bb49a0c0b2a7f..246fdd19daea7 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -17,6 +17,7 @@ import { ProviderInstanceId, type ProviderSessionId, ThreadId, + TurnItemId, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; @@ -147,7 +148,8 @@ const layerPausingAttachEventSink = (pause: { const layerBeforeUnchangedWriteEventSink = ( before: Ref.Ref>, - beforeTurnItems: Ref.Ref> | undefined, + // One effect runs before each turn-item write, in order. + beforeTurnItems: Ref.Ref>> | undefined, ) => Layer.effect( EventSink.EventSinkV2, @@ -160,11 +162,13 @@ const layerBeforeUnchangedWriteEventSink = ( Effect.flatten, Effect.andThen(delegate.writeIfProviderThreadUnchanged(input)), ), - writeIfTurnItemsUnchanged: (input) => + writeIfTurnItemUnchanged: (input) => (beforeTurnItems === undefined ? Effect.void - : Ref.getAndSet(beforeTurnItems, Effect.void).pipe(Effect.flatten) - ).pipe(Effect.andThen(delegate.writeIfTurnItemsUnchanged(input))), + : Ref.modify(beforeTurnItems, ([next, ...rest]) => [next ?? Effect.void, rest]).pipe( + Effect.flatten, + ) + ).pipe(Effect.andThen(delegate.writeIfTurnItemUnchanged(input))), }); }), ).pipe(Layer.provide(layerTestEventSink)); @@ -486,7 +490,7 @@ function layerTest(input: { readonly flakyReleaseWrites?: FlakyReleaseWrites; readonly pauseAttachWrite?: Parameters[0]; readonly beforeUnchangedWrite?: Ref.Ref>; - readonly beforeTurnItemsWrite?: Ref.Ref>; + readonly beforeTurnItemsWrite?: Ref.Ref>>; readonly pauseResolve?: { readonly armed: Ref.Ref; readonly paused: Deferred.Deferred; @@ -2517,16 +2521,19 @@ const openSessionWithBackgroundCommand = (fixtureName: string) => }); yield* Effect.forEach(Object.values(seededToolCalls), writeToolCall); yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + const toolCallStatus = (id: OrchestrationV2TurnItem["id"]) => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const projection = yield* projectionStore.getThreadProjection(threadId); + return projection.turnItems.find((item) => item.id === id)?.status; + }); const toolCalls = Effect.gen(function* () { - const projectionStore = yield* ProjectionStore.ProjectionStoreV2; - const projection = yield* projectionStore.getThreadProjection(threadId); - const statusOf = (seeded: OrchestrationV2TurnItem) => - projection.turnItems.find((item) => item.id === seeded.id)?.status; + const statusOf = (seeded: OrchestrationV2TurnItem) => toolCallStatus(seeded.id); return { - cutOff: statusOf(seededToolCalls.cutOff), - finished: statusOf(seededToolCalls.finished), - otherSession: statusOf(seededToolCalls.otherSession), - replacement: statusOf(seededToolCalls.replacement), + cutOff: yield* statusOf(seededToolCalls.cutOff), + finished: yield* statusOf(seededToolCalls.finished), + otherSession: yield* statusOf(seededToolCalls.otherSession), + replacement: yield* statusOf(seededToolCalls.replacement), }; }); const rosters = Effect.gen(function* () { @@ -2548,6 +2555,7 @@ const openSessionWithBackgroundCommand = (fixtureName: string) => seededToolCalls, writeToolCall, toolCalls, + toolCallStatus, }; }); @@ -2619,7 +2627,7 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea Effect.gen(function* () { const state = yield* Ref.make(emptyState); const beforeUnchangedWrite = yield* Ref.make>(Effect.void); - const beforeTurnItemsWrite = yield* Ref.make>(Effect.void); + const beforeTurnItemsWrite = yield* Ref.make>>([]); const effect = Effect.gen(function* () { const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; const { @@ -2629,11 +2637,30 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea rosters, seededToolCalls, writeToolCall, - toolCalls, + toolCallStatus, } = yield* openSessionWithBackgroundCommand("provider-session-manager-raced-roster"); + // Before each cleanup write, one of these calls completes late. The completion must win, + // and the conflicts must not stop the open calls from being interrupted. + const lateCompleted = ["late-1", "late-2", "late-3", "late-4"].map((name) => ({ + ...seededToolCalls.cutOff, + id: TurnItemId.make(`turn-item:${name}`), + })); + // Read after the late calls, so one write for all items would run out of tries first. + const lastOpen = { ...seededToolCalls.cutOff, id: TurnItemId.make("turn-item:last-open") }; + yield* Effect.forEach([...lateCompleted, lastOpen], writeToolCall); + // The first hook also updates the cut-off call without ending it, so it needs a second try. yield* Ref.set( beforeTurnItemsWrite, - writeToolCall({ ...seededToolCalls.cutOff, status: "completed" }).pipe(Effect.orDie), + lateCompleted.map((item, index) => + writeToolCall({ ...item, status: "completed" }).pipe( + Effect.andThen( + index === 0 + ? writeToolCall({ ...seededToolCalls.cutOff, title: "late output" }) + : Effect.void, + ), + Effect.orDie, + ), + ), ); const seeded = providerThreads[0]!; yield* Ref.set( @@ -2650,7 +2677,12 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea yield* manager.close(providerSessionId); assert.deepStrictEqual(yield* rosters, [["replacement-task"], ["dev-server"]]); - assert.equal((yield* toolCalls).cutOff, "completed"); + assert.deepStrictEqual( + yield* Effect.forEach([seededToolCalls.cutOff, ...lateCompleted, lastOpen], (item) => + toolCallStatus(item.id), + ), + ["interrupted", "completed", "completed", "completed", "completed", "interrupted"], + ); }); yield* effect.pipe( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index d87cbd9042bfe..3c3f16b1c202a 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -754,8 +754,8 @@ export const layerWithOptions = ( }); // Read the session's provider threads as soon as it leaves the live map, before a - // replacement reports its own tasks. Keep OpenCode's roster: its background subagents share - // a server and can outlive the session. + // replacement reports its own tasks. Keep OpenCode's background work: its background + // subagents share a server and can outlive the session. const readReleasedProviderThreads = Effect.fn( "ProviderSessionManager.readReleasedProviderThreads", )( @@ -859,38 +859,48 @@ export const layerWithOptions = ( "ProviderSessionManager.writeReleasedToolCallEvents", )(function* (released: ReleasedProviderThread, releasedAt: DateTime.Utc) { const { threadId, providerSessionId, providerThreadId, driver } = released; - for (let attempt = 0; attempt < 3; attempt++) { - const projection = yield* projectionStore.getThreadRecords(threadId, ["turnItems"], { + const readCutOff = projectionStore + .getThreadRecords(threadId, ["turnItems"], { turnItemTypes: releasedToolCallTypes, turnItemStatuses: activeTurnItemStatuses, - }); - const cutOff = projection.turnItems.filter( - (item) => - item.providerThreadId === providerThreadId && - DateTime.isLessThanOrEqualTo(item.startedAt ?? item.updatedAt, releasedAt), + }) + .pipe( + Effect.map((projection) => + projection.turnItems.filter( + (item) => + item.providerThreadId === providerThreadId && + DateTime.isLessThanOrEqualTo(item.startedAt ?? item.updatedAt, releasedAt), + ), + ), ); - if (cutOff.length === 0) return; - const now = yield* DateTime.now; - const events: Array = []; - for (const item of cutOff) { - events.push({ - id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), - type: "turn-item.updated", - threadId, - ...(item.runId === null ? {} : { runId: item.runId }), - ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), - driver, - occurredAt: now, - payload: { ...item, status: "interrupted", completedAt: now, updatedAt: now }, + // One write per item: a late completion from the dead session's queued events rejects + // only the interrupt of its own item, and that item then is no longer open. + let pending = yield* readCutOff; + for (let attempt = 0; attempt < 3 && pending.length > 0; attempt++) { + const conflicted: Array = []; + for (const item of pending) { + const now = yield* DateTime.now; + const { committed } = yield* eventSink.writeIfTurnItemUnchanged({ + expected: item, + events: [ + { + id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), + type: "turn-item.updated", + threadId, + ...(item.runId === null ? {} : { runId: item.runId }), + ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), + driver, + occurredAt: now, + payload: { ...item, status: "interrupted", completedAt: now, updatedAt: now }, + }, + ], }); + if (!committed) conflicted.push(item.id); } - // A late completion from the dead session's queued events wins over the interrupt. - const { committed } = yield* eventSink.writeIfTurnItemsUnchanged({ - expected: cutOff, - events, - }); - if (committed) return; + if (conflicted.length === 0) return; + pending = (yield* readCutOff).filter((item) => conflicted.includes(item.id)); } + if (pending.length === 0) return; yield* Effect.logWarning("orchestration-v2.provider-session-release-tool-call-conflict", { providerSessionId, providerThreadId, From f3a749568461a30aec82a4efb4516f6ad5664488 Mon Sep 17 00:00:00 2001 From: Spike Date: Mon, 2 Nov 2026 03:33:34 +0000 Subject: [PATCH 4/9] fix(server): read all cut-off tool calls again after a conflict Review repair: the retry read only the items that conflicted. A conflict shows that the dead session's queued events are still landing, so they can also store more cut-off items. The retry now reads all open cut-off items again. Items that the earlier pass interrupted are no longer open. Co-Authored-By: Claude Opus 5.5 --- .../ProviderSessionManager.test.ts | 26 +++++++++++++++---- .../ProviderSessionManager.ts | 11 ++++---- 2 files changed, 27 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 246fdd19daea7..33c0055a9e79a 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -2648,14 +2648,21 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea // Read after the late calls, so one write for all items would run out of tries first. const lastOpen = { ...seededToolCalls.cutOff, id: TurnItemId.make("turn-item:last-open") }; yield* Effect.forEach([...lateCompleted, lastOpen], writeToolCall); - // The first hook also updates the cut-off call without ending it, so it needs a second try. + // The first hook also updates the cut-off call without ending it, so it needs a second + // try, and stores one more call that the dead session started before the release. + const storedLate = { + ...seededToolCalls.cutOff, + id: TurnItemId.make("turn-item:stored-late"), + }; yield* Ref.set( beforeTurnItemsWrite, lateCompleted.map((item, index) => writeToolCall({ ...item, status: "completed" }).pipe( Effect.andThen( index === 0 - ? writeToolCall({ ...seededToolCalls.cutOff, title: "late output" }) + ? writeToolCall({ ...seededToolCalls.cutOff, title: "late output" }).pipe( + Effect.andThen(writeToolCall(storedLate)), + ) : Effect.void, ), Effect.orDie, @@ -2678,10 +2685,19 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea assert.deepStrictEqual(yield* rosters, [["replacement-task"], ["dev-server"]]); assert.deepStrictEqual( - yield* Effect.forEach([seededToolCalls.cutOff, ...lateCompleted, lastOpen], (item) => - toolCallStatus(item.id), + yield* Effect.forEach( + [seededToolCalls.cutOff, ...lateCompleted, lastOpen, storedLate], + (item) => toolCallStatus(item.id), ), - ["interrupted", "completed", "completed", "completed", "completed", "interrupted"], + [ + "interrupted", + "completed", + "completed", + "completed", + "completed", + "interrupted", + "interrupted", + ], ); }); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 3c3f16b1c202a..05b611c4c591e 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -874,10 +874,11 @@ export const layerWithOptions = ( ), ); // One write per item: a late completion from the dead session's queued events rejects - // only the interrupt of its own item, and that item then is no longer open. + // only the interrupt of its own item, and that item then is no longer open. After a + // conflict, read again: queued events can also store more cut-off items. let pending = yield* readCutOff; for (let attempt = 0; attempt < 3 && pending.length > 0; attempt++) { - const conflicted: Array = []; + let conflicted = false; for (const item of pending) { const now = yield* DateTime.now; const { committed } = yield* eventSink.writeIfTurnItemUnchanged({ @@ -895,10 +896,10 @@ export const layerWithOptions = ( }, ], }); - if (!committed) conflicted.push(item.id); + if (!committed) conflicted = true; } - if (conflicted.length === 0) return; - pending = (yield* readCutOff).filter((item) => conflicted.includes(item.id)); + if (!conflicted) return; + pending = yield* readCutOff; } if (pending.length === 0) return; yield* Effect.logWarning("orchestration-v2.provider-session-release-tool-call-conflict", { From 4a50e856976f11fc19dab35961d9373338af2a1e Mon Sep 17 00:00:00 2001 From: Spike Date: Mon, 2 Nov 2026 03:40:16 +0000 Subject: [PATCH 5/9] fix(server): read the cut-off tool calls again after every pass Review repair: a pass with no conflict returned at once, so an item that the dead session's queued events stored during that pass was never read. Each pass now ends with a new read, and cleanup stops only when that read finds no open cut-off item, or after three passes. The warning then names the count of open tool calls. The raced test checks statuses by item id. Co-Authored-By: Claude Opus 5.5 --- .../ProviderSessionManager.test.ts | 58 +++++++++++-------- .../ProviderSessionManager.ts | 12 ++-- 2 files changed, 38 insertions(+), 32 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 33c0055a9e79a..093fe5985dc2e 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -2648,26 +2648,36 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea // Read after the late calls, so one write for all items would run out of tries first. const lastOpen = { ...seededToolCalls.cutOff, id: TurnItemId.make("turn-item:last-open") }; yield* Effect.forEach([...lateCompleted, lastOpen], writeToolCall); - // The first hook also updates the cut-off call without ending it, so it needs a second - // try, and stores one more call that the dead session started before the release. const storedLate = { ...seededToolCalls.cutOff, id: TurnItemId.make("turn-item:stored-late"), }; + const storedInCleanPass = { + ...seededToolCalls.cutOff, + id: TurnItemId.make("turn-item:stored-in-clean-pass"), + }; + const complete = (item: OrchestrationV2TurnItem) => + writeToolCall({ ...item, status: "completed" }); + // One hook runs before each cleanup write. The first pass writes the cut-off call, the + // four late calls and the last open call; the second pass writes the cut-off call again + // and the call stored late. yield* Ref.set( beforeTurnItemsWrite, - lateCompleted.map((item, index) => - writeToolCall({ ...item, status: "completed" }).pipe( - Effect.andThen( - index === 0 - ? writeToolCall({ ...seededToolCalls.cutOff, title: "late output" }).pipe( - Effect.andThen(writeToolCall(storedLate)), - ) - : Effect.void, - ), - Effect.orDie, + [ + // The cut-off call gets an update that keeps it open, so it needs a second pass, and + // the dead session stores one more call. + complete(lateCompleted[0]!).pipe( + Effect.andThen(writeToolCall({ ...seededToolCalls.cutOff, title: "late output" })), + Effect.andThen(writeToolCall(storedLate)), ), - ), + complete(lateCompleted[1]!), + complete(lateCompleted[2]!), + complete(lateCompleted[3]!), + Effect.void, + Effect.void, + // The second pass has no conflict, but one more call lands during it. + writeToolCall(storedInCleanPass), + ].map(Effect.orDie), ); const seeded = providerThreads[0]!; yield* Ref.set( @@ -2684,21 +2694,19 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea yield* manager.close(providerSessionId); assert.deepStrictEqual(yield* rosters, [["replacement-task"], ["dev-server"]]); - assert.deepStrictEqual( + const statuses = Object.fromEntries( yield* Effect.forEach( - [seededToolCalls.cutOff, ...lateCompleted, lastOpen, storedLate], - (item) => toolCallStatus(item.id), + [seededToolCalls.cutOff, ...lateCompleted, lastOpen, storedLate, storedInCleanPass], + (item) => toolCallStatus(item.id).pipe(Effect.map((status) => [item.id, status])), ), - [ - "interrupted", - "completed", - "completed", - "completed", - "completed", - "interrupted", - "interrupted", - ], ); + assert.deepStrictEqual(statuses, { + [seededToolCalls.cutOff.id]: "interrupted", + ...Object.fromEntries(lateCompleted.map((item) => [item.id, "completed"])), + [lastOpen.id]: "interrupted", + [storedLate.id]: "interrupted", + [storedInCleanPass.id]: "interrupted", + }); }); yield* effect.pipe( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 05b611c4c591e..d7613ec24ac4d 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -874,14 +874,13 @@ export const layerWithOptions = ( ), ); // One write per item: a late completion from the dead session's queued events rejects - // only the interrupt of its own item, and that item then is no longer open. After a - // conflict, read again: queued events can also store more cut-off items. + // only the interrupt of its own item, and that item then is no longer open. Read again + // after each pass: queued events can also store more cut-off items. let pending = yield* readCutOff; - for (let attempt = 0; attempt < 3 && pending.length > 0; attempt++) { - let conflicted = false; + for (let pass = 0; pass < 3 && pending.length > 0; pass++) { for (const item of pending) { const now = yield* DateTime.now; - const { committed } = yield* eventSink.writeIfTurnItemUnchanged({ + yield* eventSink.writeIfTurnItemUnchanged({ expected: item, events: [ { @@ -896,15 +895,14 @@ export const layerWithOptions = ( }, ], }); - if (!committed) conflicted = true; } - if (!conflicted) return; pending = yield* readCutOff; } if (pending.length === 0) return; yield* Effect.logWarning("orchestration-v2.provider-session-release-tool-call-conflict", { providerSessionId, providerThreadId, + openToolCalls: pending.length, }); }); From 6b0412db8d2d9a1b6f13f2bc9a07c9cfa115c4d5 Mon Sep 17 00:00:00 2001 From: Spike Date: Mon, 2 Nov 2026 03:45:58 +0000 Subject: [PATCH 6/9] test(server): pin the three-pass bound of tool-call cleanup Review nit: the raced test stores one more cut-off call during the third pass and checks that it stays open, so the pass bound is tested. The comment on the late calls now says that only the first four writes complete one. Co-Authored-By: Claude Opus 5.5 --- .../ProviderSessionManager.test.ts | 23 +++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 093fe5985dc2e..74882950ebf5b 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -2639,8 +2639,8 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea writeToolCall, toolCallStatus, } = yield* openSessionWithBackgroundCommand("provider-session-manager-raced-roster"); - // Before each cleanup write, one of these calls completes late. The completion must win, - // and the conflicts must not stop the open calls from being interrupted. + // Before each of the first four cleanup writes, one of these calls completes late. The + // completion must win, and the conflicts must not stop the open calls from being interrupted. const lateCompleted = ["late-1", "late-2", "late-3", "late-4"].map((name) => ({ ...seededToolCalls.cutOff, id: TurnItemId.make(`turn-item:${name}`), @@ -2656,11 +2656,15 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea ...seededToolCalls.cutOff, id: TurnItemId.make("turn-item:stored-in-clean-pass"), }; + const storedAfterLastPass = { + ...seededToolCalls.cutOff, + id: TurnItemId.make("turn-item:stored-after-last-pass"), + }; const complete = (item: OrchestrationV2TurnItem) => writeToolCall({ ...item, status: "completed" }); // One hook runs before each cleanup write. The first pass writes the cut-off call, the // four late calls and the last open call; the second pass writes the cut-off call again - // and the call stored late. + // and the call stored late; the third pass writes the call stored in the clean pass. yield* Ref.set( beforeTurnItemsWrite, [ @@ -2677,6 +2681,9 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea Effect.void, // The second pass has no conflict, but one more call lands during it. writeToolCall(storedInCleanPass), + Effect.void, + // Cleanup stops after three passes, so a call that lands during the third stays open. + writeToolCall(storedAfterLastPass), ].map(Effect.orDie), ); const seeded = providerThreads[0]!; @@ -2696,7 +2703,14 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea assert.deepStrictEqual(yield* rosters, [["replacement-task"], ["dev-server"]]); const statuses = Object.fromEntries( yield* Effect.forEach( - [seededToolCalls.cutOff, ...lateCompleted, lastOpen, storedLate, storedInCleanPass], + [ + seededToolCalls.cutOff, + ...lateCompleted, + lastOpen, + storedLate, + storedInCleanPass, + storedAfterLastPass, + ], (item) => toolCallStatus(item.id).pipe(Effect.map((status) => [item.id, status])), ), ); @@ -2706,6 +2720,7 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea [lastOpen.id]: "interrupted", [storedLate.id]: "interrupted", [storedInCleanPass.id]: "interrupted", + [storedAfterLastPass.id]: "running", }); }); From b19b6d3759d7d7f025377232474ae4f1acdad180 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 3 Nov 2026 04:54:46 +0000 Subject: [PATCH 7/9] fix(server): clean released tool calls after subscriber drain --- .../ProviderSessionManager.test.ts | 61 +++++++++++++++++-- .../ProviderSessionManager.ts | 53 ++++++++++++---- 2 files changed, 96 insertions(+), 18 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index a47aa76e1cee0..a535a89446892 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -2657,18 +2657,69 @@ const openSessionWithBackgroundCommand = (fixtureName: string) => }); // regression: #15993 a worktree handoff left its tool call running, so the thread stayed Waiting -it.effect( - "ProviderSessionManagerV2 clears the background work of a session a detach releases", - () => +it.effect.each(["detach", "stream failure", "provider stop", "early close"] as const)( + "ProviderSessionManagerV2 clears background work after queued events drain on %s", + (release) => Effect.gen(function* () { const state = yield* Ref.make(emptyState); const effect = Effect.gen(function* () { const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; - const { threadId, providerSessionId, rosters, toolCalls } = + const { threadId, providerSessionId, rosters, toolCalls, seededToolCalls, writeToolCall } = yield* openSessionWithBackgroundCommand("provider-session-manager-detach-roster"); + const runtime = Option.getOrThrow(yield* manager.get(providerSessionId)); + const subscription = yield* runtime.subscribeEvents!; + const consuming = yield* Deferred.make(); + const resume = yield* Deferred.make(); + const consumer = yield* subscription.events.pipe( + Stream.tap(() => + Deferred.succeed(consuming, undefined).pipe( + Effect.andThen(Deferred.await(resume)), + Effect.andThen(writeToolCall(seededToolCalls.cutOff)), + ), + ), + (events) => (release === "early close" ? events.pipe(Stream.take(1)) : events), + Stream.runDrain, + Effect.exit, + Effect.forkScoped, + ); + const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId))!; + yield* Queue.offer(queue, { + type: "provider_session.updated", + driver: CODEX_DRIVER, + providerSession: runtime.providerSession, + }); + yield* Deferred.await(consuming); assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]); - yield* manager.detach({ providerSessionId, threadId, detail: "Workspace changed." }); + const eventSink = yield* EventSink.EventSinkV2; + const afterSequence = yield* eventSink.latestSequence({ threadId }); + if (release === "detach" || release === "early close") { + yield* manager.detach({ providerSessionId, threadId, detail: "Workspace changed." }); + } else { + if (release === "provider stop") { + yield* Queue.offer(queue, { + type: "provider_session.updated", + driver: CODEX_DRIVER, + providerSession: { ...runtime.providerSession, status: "stopped" }, + }); + } + yield* Queue.end(queue); + } + yield* eventSink.stream({ threadId, afterSequence }).pipe( + Stream.filter( + ({ event }) => + event.type === "turn-item.updated" && + event.payload.id === seededToolCalls.cutOff.id && + event.payload.status === "interrupted", + ), + Stream.take(1), + Stream.runDrain, + ); + yield* Deferred.succeed(resume, undefined); + assert.equal( + Exit.isFailure(yield* Fiber.join(consumer)), + release !== "provider stop" && release !== "early close", + ); assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); assert.deepStrictEqual(yield* rosters, [[], ["dev-server"]]); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 30e208c89a3e9..a9a0e989daeb1 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -529,6 +529,11 @@ export const layerWithOptions = ( ), ); + const subscriberReleaseCleanup = new WeakMap< + Queue.Queue, + Effect.Effect + >(); + const publishToSubscribers = ( subscribers: Ref.Ref< ReadonlyMap> @@ -546,6 +551,8 @@ export const layerWithOptions = ( const failSubscribers = Effect.fn("ProviderSessionManager.failSubscribers")(function* ( entry: LiveSessionEntry, detail: string, + afterDrain: Effect.Effect, + cause?: Cause.Cause, ) { const error = new ProviderAdapter.ProviderAdapterEventStreamError({ driver: entry.runtime.driver, @@ -555,11 +562,13 @@ export const layerWithOptions = ( const subscribers = yield* Ref.getAndSet(entry.eventSubscribers, new Map()); yield* Effect.forEach( subscribers.values(), - (queue) => - Queue.offer(queue, { + (queue) => { + subscriberReleaseCleanup.set(queue, afterDrain); + return Queue.offer(queue, { type: "failure", - cause: Cause.fail(error), - }), + cause: cause ?? Cause.fail(error), + }); + }, { discard: true }, ); }); @@ -582,11 +591,17 @@ export const layerWithOptions = ( // must let consumers drain them before the stream completes. const endSubscribers = Effect.fn("ProviderSessionManager.endSubscribers")(function* ( entry: LiveSessionEntry, + afterDrain: Effect.Effect, ) { const subscribers = yield* Ref.getAndSet(entry.eventSubscribers, new Map()); - yield* Effect.forEach(subscribers.values(), (queue) => Queue.end(queue), { - discard: true, - }); + yield* Effect.forEach( + subscribers.values(), + (queue) => { + subscriberReleaseCleanup.set(queue, afterDrain); + return Queue.end(queue); + }, + { discard: true }, + ); }); const cancelIdleFiber = (fiber: Fiber.Fiber | null) => @@ -1088,6 +1103,7 @@ export const layerWithOptions = ( readonly onlyIfIdleGeneration?: number; readonly keepIfBusyOtherThan?: ReadonlyArray; readonly gracefulSubscribers?: boolean; + readonly eventStreamCause?: Cause.Cause; }) => Effect.acquireUseRelease( removeLiveEntry(input), @@ -1106,14 +1122,22 @@ export const layerWithOptions = ( if (input.cancelIdleFiber !== false) { yield* cancelIdleFiber(entry.idleFiber); } + // Queued events can persist tool calls after the release-time scan. + const afterDrain = Effect.forEach( + releasedProviderThreads, + (thread) => writeReleasedToolCallEvents(thread, releasedAt), + { discard: true }, + ).pipe(logReleaseFailure(input.providerSessionId)); if (input.gracefulSubscribers === true) { - yield* endSubscribers(entry); + yield* endSubscribers(entry, afterDrain); } else if (input.reason === "server_shutdown") { yield* interruptSubscribers(entry); } else { yield* failSubscribers( entry, input.detail ?? `Provider session released: ${input.reason}.`, + afterDrain, + input.eventStreamCause, ); } const closeExit = yield* closeScopeWithin(entry.scope, { @@ -1754,6 +1778,13 @@ export const layerWithOptions = ( ? Queue.clear(queue).pipe(Effect.andThen(Queue.end(queue)), Effect.asVoid) : Effect.void, ), + Effect.andThen( + Effect.suspend(() => { + const cleanup = subscriberReleaseCleanup.get(queue); + subscriberReleaseCleanup.delete(queue); + return cleanup ?? Effect.void; + }), + ), ); const events = Stream.fromQueue(queue).pipe( Stream.mapEffect((signal) => @@ -2143,11 +2174,6 @@ export const layerWithOptions = ( cause: "Provider event stream ended unexpectedly.", }), ); - yield* publishToSubscribers(entry.eventSubscribers, { - type: "failure", - cause, - }); - yield* Ref.set(entry.eventSubscribers, new Map()); // lastError is the thread banner, so it gets the same bounded // message as a failed turn; the raw cause survives only in the log. yield* Effect.logWarning("orchestration-v2.provider-session-runtime-failed", { @@ -2158,6 +2184,7 @@ export const layerWithOptions = ( providerSessionId: entry.runtime.providerSessionId, reason: "runtime_error", detail: makeProviderFailure({ cause }).message, + eventStreamCause: cause, }).pipe(logReleaseFailure(entry.runtime.providerSessionId)); }), ), From 30fbbcd203d58cc79981f76d4cd4ce88776b7b11 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 10 Nov 2026 11:21:25 +0000 Subject: [PATCH 8/9] fix(server): run release tool-call cleanup off the subscriber's fiber The cleanup ran in the subscriber stream's finalizer, so a released run's failure write waited for it and a handoff continuation could queue behind the still-running run. --- .../ProviderSessionManager.test.ts | 23 +++++++++++-------- .../ProviderSessionManager.ts | 3 ++- 2 files changed, 15 insertions(+), 11 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index b626a1344ea64..2362120af419f 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -3016,21 +3016,24 @@ it.effect.each(["detach", "stream failure", "provider stop", "early close"] as c } yield* Queue.end(queue); } - yield* eventSink.stream({ threadId, afterSequence }).pipe( - Stream.filter( - ({ event }) => - event.type === "turn-item.updated" && - event.payload.id === seededToolCalls.cutOff.id && - event.payload.status === "interrupted", - ), - Stream.take(1), - Stream.runDrain, - ); + const interrupts = (count: number) => + eventSink.stream({ threadId, afterSequence }).pipe( + Stream.filter( + ({ event }) => + event.type === "turn-item.updated" && + event.payload.id === seededToolCalls.cutOff.id && + event.payload.status === "interrupted", + ), + Stream.take(count), + Stream.runDrain, + ); + yield* interrupts(1); yield* Deferred.succeed(resume, undefined); assert.equal( Exit.isFailure(yield* Fiber.join(consumer)), release !== "provider stop" && release !== "early close", ); + yield* interrupts(2); assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); assert.deepStrictEqual(yield* rosters, [[], ["dev-server"]]); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index e4af5105308ac..e1c2257f842cb 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -2013,11 +2013,12 @@ export const layerWithOptions = ( ? Queue.clear(queue).pipe(Effect.andThen(Queue.end(queue)), Effect.asVoid) : Effect.void, ), + // Detached so the consumer's run failure is not ordered behind release cleanup. Effect.andThen( Effect.suspend(() => { const cleanup = subscriberReleaseCleanup.get(queue); subscriberReleaseCleanup.delete(queue); - return cleanup ?? Effect.void; + return cleanup === undefined ? Effect.void : Effect.forkDetach(cleanup); }), ), ); From 115e50849e220c0443af02918b9f4bc5fccaafcf Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 21:02:55 +0000 Subject: [PATCH 9/9] test(server): drop the assertion that pins the cleanup pass limit AI-assisted; verified with the tests and tools listed above. --- .../ProviderSessionManager.test.ts | 17 +---------------- 1 file changed, 1 insertion(+), 16 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 22061a80cab31..40c2d28a590a9 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -3431,10 +3431,6 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea ...seededToolCalls.cutOff, id: TurnItemId.make("turn-item:stored-in-clean-pass"), }; - const storedAfterLastPass = { - ...seededToolCalls.cutOff, - id: TurnItemId.make("turn-item:stored-after-last-pass"), - }; const complete = (item: OrchestrationV2TurnItem) => writeToolCall({ ...item, status: "completed" }); // One hook runs before each cleanup write. The first pass writes the cut-off call, the @@ -3456,9 +3452,6 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea Effect.void, // The second pass has no conflict, but one more call lands during it. writeToolCall(storedInCleanPass), - Effect.void, - // Cleanup stops after three passes, so a call that lands during the third stays open. - writeToolCall(storedAfterLastPass), ].map(Effect.orDie), ); const seeded = providerThreads[0]!; @@ -3478,14 +3471,7 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea assert.deepStrictEqual(yield* rosters, [["replacement-task"], ["dev-server"]]); const statuses = Object.fromEntries( yield* Effect.forEach( - [ - seededToolCalls.cutOff, - ...lateCompleted, - lastOpen, - storedLate, - storedInCleanPass, - storedAfterLastPass, - ], + [seededToolCalls.cutOff, ...lateCompleted, lastOpen, storedLate, storedInCleanPass], (item) => toolCallStatus(item.id).pipe(Effect.map((status) => [item.id, status])), ), ); @@ -3495,7 +3481,6 @@ it.effect("ProviderSessionManagerV2 keeps provider updates that land while relea [lastOpen.id]: "interrupted", [storedLate.id]: "interrupted", [storedInCleanPass.id]: "interrupted", - [storedAfterLastPass.id]: "running", }); });