diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index b210d3c9840ae..0af8f1453155c 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 the stored turn item still equals expected. */ + readonly writeIfTurnItemUnchanged: (input: { + readonly expected: OrchestrationV2TurnItem; + readonly events: ReadonlyArray; + }) => Effect.Effect< + { + readonly committed: boolean; + readonly storedEvents: ReadonlyArray; + }, + EventSinkV2Error + >; readonly commitCommand: (input: { readonly commandId: CommandId; readonly threadId: ThreadId; @@ -580,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, @@ -604,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 }; @@ -614,6 +618,45 @@ const layerBase: Layer.Layer< ? publishStoredEvents(result.storedEvents, result.outcomes) : Effect.void, ); + + 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* 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, + ); + }); + + 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, + ); }); const existingCommandResult = @@ -928,6 +971,12 @@ const layerBase: Layer.Layer< (cause) => new EventSinkWriteError({ eventCount: input.events.length, cause }), ), ), + writeIfTurnItemUnchanged: (input) => + writeIfTurnItemUnchangedEffect(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 3cffce15b70d6..8247493808bc4 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1078,7 +1078,7 @@ const encodeMessagePayload = Schema.encodeEffect( const encodePlanPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2PlanArtifactSchema), ); -const encodeTurnItemPayload = Schema.encodeEffect( +export const encodeTurnItemPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2TurnItemJsonSchema), ); // The codec every *Json payload schema uses for timestamps, so index columns match payload_json. diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index d8c046c018651..a535a89446892 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -11,12 +11,14 @@ import { type OrchestrationV2ProviderCapabilities, type OrchestrationV2ProviderSession, type OrchestrationV2ProviderThread, + type OrchestrationV2TurnItem, type Project, ProjectId, ProviderDriverKind, ProviderInstanceId, type ProviderSessionId, ThreadId, + TurnItemId, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; @@ -145,7 +147,11 @@ const layerPausingAttachEventSink = (pause: { }), ).pipe(Layer.provide(layerTestEventSink)); -const layerBeforeUnchangedWriteEventSink = (before: Ref.Ref>) => +const layerBeforeUnchangedWriteEventSink = ( + before: Ref.Ref>, + // One effect runs before each turn-item write, in order. + beforeTurnItems: Ref.Ref>> | undefined, +) => Layer.effect( EventSink.EventSinkV2, Effect.gen(function* () { @@ -157,6 +163,13 @@ const layerBeforeUnchangedWriteEventSink = (before: Ref.Ref> Effect.flatten, Effect.andThen(delegate.writeIfProviderThreadUnchanged(input)), ), + writeIfTurnItemUnchanged: (input) => + (beforeTurnItems === undefined + ? Effect.void + : Ref.modify(beforeTurnItems, ([next, ...rest]) => [next ?? Effect.void, rest]).pipe( + Effect.flatten, + ) + ).pipe(Effect.andThen(delegate.writeIfTurnItemUnchanged(input))), }); }), ).pipe(Layer.provide(layerTestEventSink)); @@ -574,6 +587,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; @@ -596,7 +610,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; @@ -2547,7 +2564,75 @@ const openSessionWithBackgroundCommand = (fixtureName: string) => events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], }); yield* Effect.forEach(providerThreads, writeProviderThread); + const toolCall = ( + nativeItemId: string, + providerThread: OrchestrationV2ProviderThread, + status: "running" | "completed", + startedAt: DateTime.Utc, + ) => + ({ + id: idAllocator.derive.turnItemFromProviderItem({ driver: CODEX_DRIVER, nativeItemId }), + threadId, + runId: null, + nodeId: null, + providerThreadId: providerThread.id, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status, + title: null, + startedAt, + completedAt: status === "completed" ? startedAt : null, + updatedAt: startedAt, + type: "dynamic_tool", + toolName: "mcp__t3-code__t3_worktree_handoff", + input: {}, + }) satisfies OrchestrationV2TurnItem; + 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 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 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 statusOf = (seeded: OrchestrationV2TurnItem) => toolCallStatus(seeded.id); + return { + 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* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; const projection = yield* projectionStore.getThreadProjection(threadId); @@ -2558,25 +2643,92 @@ const openSessionWithBackgroundCommand = (fixtureName: string) => ).map((task) => task.taskId), ); }); - return { threadId, providerSessionId, providerThreads, writeProviderThread, rosters }; + return { + threadId, + providerSessionId, + providerThreads, + writeProviderThread, + rosters, + seededToolCalls, + writeToolCall, + toolCalls, + toolCallStatus, + }; }); -it.effect( - "ProviderSessionManagerV2 clears the background roster of a session a detach releases", - () => +// regression: #15993 a worktree handoff left its tool call running, so the thread stayed Waiting +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 } = yield* openSessionWithBackgroundCommand( - "provider-session-manager-detach-roster", + 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: "Thread settled." }); + 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"]]); + assert.deepStrictEqual(yield* toolCalls, { + cutOff: "interrupted", + finished: "completed", + otherSession: "running", + replacement: "running", + }); }); yield* effect.pipe( @@ -2587,37 +2739,101 @@ it.effect( }), ); -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", - ); +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 { 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* rosters, [["dev-server"], ["dev-server"]]); - }); + 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 }), - ), - ); - }), + yield* effect.pipe( + Effect.provide( + 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>>([]); 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, + toolCallStatus, + } = yield* openSessionWithBackgroundCommand("provider-session-manager-raced-roster"); + // 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}`), + })); + // 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); + 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 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; the third pass writes the call stored in the clean pass. + yield* Ref.set( + beforeTurnItemsWrite, + [ + // 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), + 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]!; yield* Ref.set( beforeUnchangedWrite, @@ -2633,10 +2849,33 @@ it.effect("ProviderSessionManagerV2 keeps a provider update that lands while rel yield* manager.close(providerSessionId); assert.deepStrictEqual(yield* rosters, [["replacement-task"], ["dev-server"]]); + const statuses = Object.fromEntries( + yield* Effect.forEach( + [ + seededToolCalls.cutOff, + ...lateCompleted, + lastOpen, + storedLate, + storedInCleanPass, + storedAfterLastPass, + ], + (item) => toolCallStatus(item.id).pipe(Effect.map((status) => [item.id, status])), + ), + ); + assert.deepStrictEqual(statuses, { + [seededToolCalls.cutOff.id]: "interrupted", + ...Object.fromEntries(lateCompleted.map((item) => [item.id, "completed"])), + [lastOpen.id]: "interrupted", + [storedLate.id]: "interrupted", + [storedInCleanPass.id]: "interrupted", + [storedAfterLastPass.id]: "running", + }); }); 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 f06011676c667..a9a0e989daeb1 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", { @@ -517,6 +529,11 @@ export const layerWithOptions = ( ), ); + const subscriberReleaseCleanup = new WeakMap< + Queue.Queue, + Effect.Effect + >(); + const publishToSubscribers = ( subscribers: Ref.Ref< ReadonlyMap> @@ -534,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, @@ -543,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 }, ); }); @@ -570,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) => @@ -741,10 +768,11 @@ 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. - 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 background work: its background + // subagents share a server and can outlive the session. + const readReleasedProviderThreads = Effect.fn( + "ProviderSessionManager.readReleasedProviderThreads", )( function* (input: { readonly entry: LiveSessionEntry; @@ -766,15 +794,15 @@ export const layerWithOptions = ( "providerThreads", ]); for (const providerThread of projection.providerThreads) { - const tasks = providerThread.pendingBackgroundTasks ?? []; - if (providerThread.providerSessionId !== providerSessionId || tasks.length === 0) { - continue; - } + if (providerThread.providerSessionId !== providerSessionId) continue; released.push({ threadId, providerSessionId, providerThreadId: providerThread.id, - taskIds: new Set(tasks.map((task) => task.taskId)), + driver: providerThread.driver, + taskIds: new Set( + (providerThread.pendingBackgroundTasks ?? []).map((task) => task.taskId), + ), }); } } @@ -790,57 +818,119 @@ 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 = Effect.fn( + "ProviderSessionManager.writeReleasedRosterEvents", + )(function* ({ + threadId, + providerSessionId, + providerThreadId, + taskIds, + }: 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 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; + const readCutOff = projectionStore + .getThreadRecords(threadId, ["turnItems"], { + turnItemTypes: releasedToolCallTypes, + turnItemStatuses: activeTurnItemStatuses, + }) + .pipe( + Effect.map((projection) => + projection.turnItems.filter( + (item) => + item.providerThreadId === providerThreadId && + DateTime.isLessThanOrEqualTo(item.startedAt ?? item.updatedAt, releasedAt), + ), + ), + ); + // 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. Read again + // after each pass: queued events can also store more cut-off items. + let pending = yield* readCutOff; + for (let pass = 0; pass < 3 && pending.length > 0; pass++) { + for (const item of pending) { + const now = yield* DateTime.now; + 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 }, + }, + ], + }); + } + pending = yield* readCutOff; + } + if (pending.length === 0) return; + yield* Effect.logWarning("orchestration-v2.provider-session-release-tool-call-conflict", { + providerSessionId, + providerThreadId, + openToolCalls: pending.length, + }); + }); + const writeReleasedBackgroundWorkEvents = ( - released: ReadonlyArray, + released: ReadonlyArray, + releasedAt: DateTime.Utc, ) => Effect.forEach( released, - ({ threadId, providerSessionId, providerThreadId, taskIds }) => - 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, - }, - ); - }), + (providerThread) => + writeReleasedRosterEvents(providerThread).pipe( + Effect.andThen(writeReleasedToolCallEvents(providerThread, releasedAt)), + ), { discard: true }, ); @@ -848,7 +938,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; }) => @@ -862,7 +952,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)); @@ -1011,6 +1103,7 @@ export const layerWithOptions = ( readonly onlyIfIdleGeneration?: number; readonly keepIfBusyOtherThan?: ReadonlyArray; readonly gracefulSubscribers?: boolean; + readonly eventStreamCause?: Cause.Cause; }) => Effect.acquireUseRelease( removeLiveEntry(input), @@ -1019,7 +1112,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 @@ -1029,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, { @@ -1047,7 +1148,7 @@ export const layerWithOptions = ( entry, reason: input.reason, ...(input.detail === undefined ? {} : { detail: input.detail }), - releasedBackgroundTasks, + releasedProviderThreads, releasedAt, }; const recorded = yield* Effect.exit( @@ -1677,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) => @@ -2066,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", { @@ -2081,6 +2184,7 @@ export const layerWithOptions = ( providerSessionId: entry.runtime.providerSessionId, reason: "runtime_error", detail: makeProviderFailure({ cause }).message, + eventStreamCause: cause, }).pipe(logReleaseFailure(entry.runtime.providerSessionId)); }), ), 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 f36e463079f99..44bcb449aa116 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; readonly backgroundWork?: Ref.Ref; readonly closedSessions?: Ref.Ref>; }): ProviderAdapter.ProviderAdapterV2["Service"] { @@ -156,6 +164,9 @@ function makeTestAdapter(input: { providerSessionId: sessionInput.providerSessionId, providerSession, events: Stream.fromPubSub(events), + ...(input.backgroundTasksEndWithProcess === undefined + ? {} + : { backgroundTasksEndWithProcess: input.backgroundTasksEndWithProcess }), ...(input.pendingBackgroundWork === undefined ? {} : { @@ -279,18 +290,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) @@ -1767,6 +1819,121 @@ describe("orchestration v2 provider switching", () => { ), ); + // 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: "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({ + 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)), + ); + }), + ), + ); + it.live("checks the queued provider's capability while the current provider stays running", () => Effect.scoped( Effect.gen(function* () { diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 9d83da489b306..b818bf677bf81 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -587,7 +587,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,