diff --git a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts index 6d5e0b92d87f..794cf3827d4f 100644 --- a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts +++ b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts @@ -14,12 +14,15 @@ import { ThreadId, TurnItemId, type OrchestrationV2DomainEvent, + type OrchestrationV2ProviderCapabilities, } from "@t3tools/contracts"; +import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; +import { ClaudeProviderCapabilitiesV2 } from "./Adapters/ClaudeAdapterV2.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as EffectWorker from "./EffectWorker.ts"; import * as EventSink from "./EventSink.ts"; @@ -34,6 +37,94 @@ const driver = ProviderDriverKind.make("codex"); const instanceId = ProviderInstanceId.make("codex"); const modelSelection = { instanceId, model: "test-model" }; +// A provider whose turns start running, then report only what a test offers +// on `events`. +const scriptedAdapter = (input: { + readonly cwd: string; + readonly events: Queue.Queue; + readonly started: ProviderAdapter.ProviderAdapterV2TurnInput[]; + readonly capabilities: OrchestrationV2ProviderCapabilities; + readonly interrupts?: ProviderAdapter.ProviderAdapterV2InterruptInput[]; +}): ProviderAdapter.ProviderAdapterV2["Service"] => ({ + instanceId, + driver, + getCapabilities: () => Effect.succeed(input.capabilities), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + openSession: (session) => + Effect.gen(function* () { + const now = yield* DateTime.now; + return { + instanceId, + driver, + providerSessionId: session.providerSessionId, + providerSession: { + id: session.providerSessionId, + driver, + providerInstanceId: instanceId, + status: "ready", + cwd: input.cwd, + model: modelSelection.model, + capabilities: input.capabilities, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.fromQueue(input.events), + ensureThread: ({ threadId }) => + Effect.succeed({ + id: ProviderThreadId.make(`provider-thread:codex:${threadId}`), + driver, + providerInstanceId: instanceId, + providerSessionId: session.providerSessionId, + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }), + resumeThread: ({ providerThread }) => Effect.succeed(providerThread), + startTurn: (turn) => + Effect.gen(function* () { + input.started.push(turn); + yield* Queue.offer(input.events, { + type: "provider_turn.updated", + driver, + providerTurn: { + id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), + providerThreadId: turn.providerThread.id, + nodeId: turn.rootNodeId, + runAttemptId: turn.attemptId, + nativeTurnRef: { + driver, + nativeId: `native:${turn.attemptId}`, + strength: "strong", + }, + ordinal: turn.providerTurnOrdinal, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + }), + steerTurn: () => Effect.die("unused"), + interruptTurn: (interrupt) => + Effect.sync(() => { + input.interrupts?.push(interrupt); + }), + respondToRuntimeRequest: () => Effect.die("unused"), + readThreadSnapshot: () => Effect.die("unused"), + rollbackThread: () => Effect.die("unused"), + forkThread: () => Effect.die("unused"), + }; + }), +}); + // Codex turns leave commands running, then the thread moves to another // provider thread (a provider switch). Stop must end all of the Codex work, // including when it selects an older resumed run on the other provider thread. @@ -59,85 +150,13 @@ const stopEarlierBackgroundWork = ({ const events = yield* Queue.unbounded(); const started: ProviderAdapter.ProviderAdapterV2TurnInput[] = []; const interrupts: ProviderAdapter.ProviderAdapterV2InterruptInput[] = []; - const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = { - instanceId, - driver, - getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), - planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), - openSession: (input) => - Effect.gen(function* () { - const now = yield* DateTime.now; - return { - instanceId, - driver, - providerSessionId: input.providerSessionId, - providerSession: { - id: input.providerSessionId, - driver, - providerInstanceId: instanceId, - status: "ready", - cwd, - model: modelSelection.model, - capabilities: CodexProviderCapabilitiesV2, - createdAt: now, - updatedAt: now, - lastError: null, - }, - events: Stream.fromQueue(events), - ensureThread: ({ threadId }) => - Effect.succeed({ - id: ProviderThreadId.make(`provider-thread:codex:${threadId}`), - driver, - providerInstanceId: instanceId, - providerSessionId: input.providerSessionId, - appThreadId: threadId, - ownerNodeId: null, - nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" }, - nativeConversationHeadRef: null, - status: "idle", - firstRunOrdinal: null, - lastRunOrdinal: null, - handoffIds: [], - forkedFrom: null, - createdAt: now, - updatedAt: now, - }), - resumeThread: ({ providerThread }) => Effect.succeed(providerThread), - startTurn: (turn) => - Effect.gen(function* () { - started.push(turn); - yield* Queue.offer(events, { - type: "provider_turn.updated", - driver, - providerTurn: { - id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), - providerThreadId: turn.providerThread.id, - nodeId: turn.rootNodeId, - runAttemptId: turn.attemptId, - nativeTurnRef: { - driver, - nativeId: `native:${turn.attemptId}`, - strength: "strong", - }, - ordinal: turn.providerTurnOrdinal, - status: "running", - startedAt: now, - completedAt: null, - }, - }); - }), - steerTurn: () => Effect.die("unused"), - interruptTurn: (interrupt) => - Effect.sync(() => { - interrupts.push(interrupt); - }), - respondToRuntimeRequest: () => Effect.die("unused"), - readThreadSnapshot: () => Effect.die("unused"), - rollbackThread: () => Effect.die("unused"), - forkThread: () => Effect.die("unused"), - }; - }), - }; + const adapter = scriptedAdapter({ + cwd, + events, + started, + capabilities: CodexProviderCapabilitiesV2, + interrupts, + }); yield* Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; @@ -759,6 +778,144 @@ const stopEarlierBackgroundWork = ({ }), ); +// `t3_worktree_handoff` moves the calling thread, so the workspace change +// detaches the session that is running that very tool call. Nothing will report +// on the call again, so it is cancelled with the run instead of waiting as background work. +it.effect("cancels a tool call cut off when a workspace change detaches its session", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("workspace-detach-tool-call"); + const events = yield* Queue.unbounded(); + const started: ProviderAdapter.ProviderAdapterV2TurnInput[] = []; + const adapter = scriptedAdapter({ + cwd, + events, + started, + capabilities: ClaudeProviderCapabilitiesV2, + }); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = ThreadId.make("thread:workspace-detach-tool-call"); + const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create"), + threadId, + projectId: ProjectId.make("project:workspace-detach-tool-call"), + title: "Workspace detach", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const running = yield* watch( + (event) => event.type === "provider-turn.updated" && event.payload.status === "running", + ); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("hand-off"), + threadId, + messageId: MessageId.make("message:hand-off"), + text: "Move this work to a worktree", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(running); + const turn = started[0]!; + const providerTurn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; + const toolCallId = TurnItemId.make("turn-item:worktree-handoff"); + const toolCallStarted = yield* watch( + (event) => event.type === "turn-item.updated" && event.payload.id === toolCallId, + ); + const now = yield* DateTime.now; + yield* Queue.offer(events, { + type: "turn_item.updated", + driver, + turnItem: { + id: toolCallId, + threadId, + runId: turn.runId, + nodeId: turn.rootNodeId, + providerThreadId: providerTurn.providerThreadId, + providerTurnId: providerTurn.id, + nativeItemRef: null, + parentItemId: null, + ordinal: 100, + status: "running", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + type: "dynamic_tool", + toolName: "mcp__t3-code__t3_worktree_handoff", + input: { branch: "t3code/handoff" }, + }, + }); + yield* Fiber.join(toolCallStarted); + + // Replays from here, so the failure commit cannot slip past the watch. + const sink = yield* EventSink.EventSinkV2; + const failed = yield* sink + .stream({ threadId, afterSequence: yield* sink.latestSequence() }) + .pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.id === turn.runId && + event.payload.status === "failed", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + // What the workspace change's detach effect runs. A Claude session hosts + // one provider thread, so detaching it releases the process. + const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* sessions.detach({ + providerSessionId: turn.providerThread.providerSessionId!, + threadId, + detail: "Workspace changed.", + }); + yield* Fiber.join(failed); + + const after = yield* orchestrator.getThreadProjection(threadId); + assert.equal(after.turnItems.find((item) => item.id === toolCallId)?.status, "cancelled"); + assert.deepEqual( + derivePendingBackgroundWork({ + latestRun: after.runs.at(-1), + providerThreads: after.providerThreads, + turnItems: after.turnItems, + activeProviderThreadId: after.thread.activeProviderThreadId, + runs: after.runs, + }), + [], + ); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { name: "workspace-detach-tool-call" }, + ProviderAdapterRegistry.layerSingle(adapter), + { runEffectWorker: false }, + ), + ), + ); + }), + ), +); + it.effect("Stop reaches background work an earlier provider thread still runs", () => stopEarlierBackgroundWork({}), ); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 708205663787..762d971b9f82 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -3450,6 +3450,128 @@ it.effect("refreshes pull requests after a provider stream exits with an error", }), ); +it.effect("cancels the run's open tool calls when its provider stream is lost", () => + Effect.gen(function* () { + const finishedItemId = TurnItemId.make("turn-item:lost-stream:finished"); + const { written } = yield* captureRootRunTermination({ + key: "lost-stream-open-tool", + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => + Stream.concat( + Stream.make( + backgroundTurnItemEvent(ids, "dynamic_tool", "running", 1), + backgroundTurnItemEvent(ids, "command_execution", "running", 2, finishedItemId), + backgroundTurnItemEvent(ids, "command_execution", "completed", 3, finishedItemId), + ), + // A workspace change detaches the session while the tool call runs. + Stream.fail( + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver, + providerSessionId: ProviderSessionId.make("session:detached"), + cause: "Workspace changed.", + }), + ), + ), + }); + assert.deepEqual( + written.map((item) => [item.type, item.status]), + [ + ["dynamic_tool", "cancelled"], + ["error", "failed"], + ], + ); + }), +); + +it.effect("strips unserved image bytes from a tool call it cancels", () => + Effect.gen(function* () { + const imageBase64 = Buffer.alloc(3_000, 7).toString("base64"); + const { written } = yield* captureRootRunTermination({ + key: "lost-stream-tool-image", + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => + Stream.concat( + Stream.make({ + type: "turn_item.updated", + driver, + turnItem: { + id: ids.itemId, + threadId: ids.threadId, + runId: ids.runId, + providerTurnId: ids.rootProviderTurnId, + ordinal: 1, + type: "dynamic_tool", + status: "running", + toolName: "Read", + output: { + type: "image", + file: { base64: imageBase64, type: "image/png", originalSize: 3_000 }, + }, + }, + } as ProviderAdapter.ProviderAdapterV2Event), + Stream.fail( + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver, + providerSessionId: ProviderSessionId.make("session:detached"), + cause: "Workspace changed.", + }), + ), + ), + }); + const cancelled = written.find((item) => item.type === "dynamic_tool"); + assert.deepEqual(cancelled?.type === "dynamic_tool" ? cancelled.output : null, { + type: "image", + file: { type: "image/png", originalSize: 3_000, sizeBytes: 3_000 }, + }); + }), +); + +it.effect("leaves open tool calls to the next run when only storing an event fails", () => + Effect.gen(function* () { + const { written } = yield* captureRootRunTermination({ + key: "ingest-failure-open-tool", + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => + Stream.make( + backgroundTurnItemEvent(ids, "dynamic_tool", "running", 1), + backgroundTurnItemEvent(ids, "dynamic_tool", "running", 2), + ), + // The provider is still running the tool; only this server failed to store an update. + ingestNormalized: (ingest) => + ingest.event.type === "turn_item.updated" && ingest.event.turnItem.ordinal === 2 + ? Effect.fail( + new ProviderEventIngestor.ProviderEventPublishError({ + providerSessionId: ingest.providerSessionId, + eventCount: 1, + }), + ) + : Effect.succeed([]), + }); + assert.deepEqual( + written.map((item) => [item.type, item.status]), + [["error", "failed"]], + ); + }), +); + +it.effect("leaves open tool calls to the next run when the provider reports the terminal", () => + Effect.gen(function* () { + const { written } = yield* captureRootRunTermination({ + key: "reported-terminal-open-tool", + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => + Stream.make( + backgroundTurnItemEvent(ids, "command_execution", "running", 1), + rootTerminalEvent(ids, "interrupted"), + ), + }); + assert.deepEqual( + written.map((item) => item.type), + ["run_interrupt_result"], + ); + }), +); + it.effect("refreshes pull requests only once when startup failure closes its event stream", () => Effect.gen(function* () { const ingestionStarted = yield* Deferred.make(); @@ -3506,6 +3628,7 @@ function captureRootRunTermination(input: { >; readonly startTurn?: ProviderAdapter.ProviderAdapterV2SessionRuntime["startTurn"]; readonly refreshAfterTurn?: Effect.Effect; + readonly ingestNormalized?: ProviderEventIngestor.ProviderEventIngestorV2["Service"]["ingestNormalized"]; }) { return Effect.gen(function* () { const ids = backgroundScenarioIds(input.key); @@ -3571,7 +3694,7 @@ function captureRootRunTermination(input: { }), IdAllocator.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ - ingestNormalized: () => Effect.succeed([]), + ingestNormalized: input.ingestNormalized ?? (() => Effect.succeed([])), }), ServerSettings.layerTest(), Layer.succeed(RunFinalizationService.RunFinalizationObserver, { diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index a316ce0ccd62..34a6fb549556 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -49,6 +49,7 @@ import { makeProviderFailureTurnItem, } from "@t3tools/provider-core/server/failure"; import * as RunFinalizationService from "./RunFinalizationService.ts"; +import { stripUnservedToolOutputImageBytes } from "./toolOutputImageBytes.ts"; export interface ProviderEventRoutingState { readonly ownedThreadIds: ReadonlySet; @@ -566,6 +567,13 @@ export const layer: Layer.Layer< readonly shouldFinalizeRun?: () => Effect.Effect; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection; + /** + * Open tool and command items of a run whose provider stream was lost. + * Nothing will report on them again, so they are cancelled with the run, + * as restart recovery does, instead of passing to the next run as + * background work. + */ + readonly orphanedTurnItems?: ReadonlyArray; readonly terminal: ProviderTerminalEvent; readonly failureItemPersisted: boolean; readonly refreshAfterTurn: Effect.Effect; @@ -636,6 +644,26 @@ export const layer: Layer.Layer< allocateEventId, }) : []; + const orphanedTurnItemEvents: Array = []; + for (const item of input.orphanedTurnItems ?? []) { + orphanedTurnItemEvents.push({ + id: yield* allocateEventId(), + type: "turn-item.updated", + threadId: item.threadId, + runId: input.run.id, + ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), + providerInstanceId: input.run.providerInstanceId, + occurredAt: completedAt, + // The snapshot is the adapter's item, so it gets the same image + // stripping the ingestor gives every stored update. + payload: stripUnservedToolOutputImageBytes({ + ...item, + status: "cancelled", + completedAt, + updatedAt: completedAt, + }), + }); + } const persistedStatus = input.terminal.status === "completed" ? "waiting" : input.terminal.status; // Completion cohorts are advanced by Orchestrator while a provider @@ -690,6 +718,7 @@ export const layer: Layer.Layer< // Terminalize open run-owned subagent rows before the root run // settles so projections never keep a forever-running subagent card. ...cascadedSubagentEvents, + ...orphanedTurnItemEvents, ...(finalizedAttempt === null ? [] : [ @@ -952,12 +981,17 @@ export const layer: Layer.Layer< ); const rootTerminalSeen = yield* Ref.make(false); const rootRunFinalized = yield* Ref.make(false); + // Set when the provider's stream itself fails, as when its session is + // released. A failure to store an event leaves the provider running. + const providerStreamLost = yield* Ref.make(false); const providerThreadOwnerLost = yield* Ref.make(false); const activeChildProviderTurns = yield* Ref.make>(new Set()); const activeChildSubagents = yield* Ref.make>(new Set()); + // Open background items with their latest update, if one arrived on + // this stream. A lost stream cancels this run's tool and command items. const activeBackgroundTurnItems = yield* Ref.make< - ReadonlySet - >(new Set(inheritedBackgroundTurnItemsById.keys())); + ReadonlyMap + >(new Map(Array.from(inheritedBackgroundTurnItemsById.keys(), (id) => [id, undefined]))); const openRunOwnedSubagents = yield* Ref.make(emptyOpenRunOwnedSubagentProjection()); const finalizeRootRun = (terminal: ProviderTerminalEvent) => Effect.gen(function* () { @@ -1092,11 +1126,11 @@ export const layer: Layer.Layer< belongsToInheritedBackgroundItem) ) { yield* Ref.update(activeBackgroundTurnItems, (current) => { - const next = new Set(current); + const next = new Map(current); if (isSettledTurnItemStatus(event.turnItem.status)) { next.delete(event.turnItem.id); } else { - next.add(event.turnItem.id); + next.set(event.turnItem.id, event.turnItem); } return next; }); @@ -1185,6 +1219,7 @@ export const layer: Layer.Layer< }); const filterAssistantEvent = makeAssistantStreamingFilter(responseStreamingMode); const providerEventFiber = yield* eventSubscription.events.pipe( + Stream.tapCause(() => Ref.set(providerStreamLost, true)), Stream.filterEffect((event) => Ref.modify(eventRouting, (state) => routeProviderEvent(event, routeIdentity, state)), ), @@ -1291,41 +1326,45 @@ export const layer: Layer.Layer< Effect.andThen( finalized ? Effect.void - : Ref.get(latestProviderThread).pipe( - Effect.flatMap((providerThread) => - Ref.get(latestTurnItemOrdinal).pipe( - Effect.flatMap((latestItemOrdinal) => - Ref.get(openRunOwnedSubagents).pipe( - Effect.flatMap((openSubagents) => - writeFinalRunEvents({ - run: input.run, - rootNode: input.rootNode, - checkpointScope: input.checkpointScope, - providerThread, - attempt: input.attempt, - // The failure may be the ownership - // read itself, so check in the write. - writeIfRunCurrent: { - activeAttemptId: input.attempt.id, - expectedStatus: "running", - }, - openRunOwnedSubagents: openSubagents, - terminal: makeFailedTerminalEvent( - makeProviderFailure({ - cause: Cause.squash(cause), - class: "unknown", - }), - latestItemOrdinal + 1, - ), - failureItemPersisted: false, - refreshAfterTurn, - }), - ), - ), - ), + : Effect.gen(function* () { + const latestItemOrdinal = yield* Ref.get(latestTurnItemOrdinal); + yield* writeFinalRunEvents({ + run: input.run, + rootNode: input.rootNode, + checkpointScope: input.checkpointScope, + providerThread: yield* Ref.get(latestProviderThread), + attempt: input.attempt, + // The failure may be the ownership + // read itself, so check in the write. + writeIfRunCurrent: { + activeAttemptId: input.attempt.id, + expectedStatus: "running", + }, + openRunOwnedSubagents: yield* Ref.get(openRunOwnedSubagents), + // The stream that would finish them is gone, for + // example when a workspace change detaches the + // session mid tool call. + orphanedTurnItems: (yield* Ref.get(providerStreamLost)) + ? Array.from( + (yield* Ref.get(activeBackgroundTurnItems)).values(), + ).filter( + (item): item is OrchestrationV2TurnItem => + item?.runId === input.run.id && + (item.type === "command_execution" || + item.type === "dynamic_tool"), + ) + : [], + terminal: makeFailedTerminalEvent( + makeProviderFailure({ + cause: Cause.squash(cause), + class: "unknown", + }), + latestItemOrdinal + 1, ), - ), - ), + failureItemPersisted: false, + refreshAfterTurn, + }); + }), ), Effect.mapError( (writeCause) =>