From d65cd33a03983751598f6bdce76b402ac7bf2cca Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sat, 3 Oct 2026 16:20:17 -0700 Subject: [PATCH] fix(server): Stop ends a dev server left running before a provider switch (#15355) Co-authored-by: Claude Opus 5.5 (1M context) --- .../BackgroundWorkStop.integration.test.ts | 446 ++++++++++++++++++ .../src/orchestration-v2/EffectWorker.ts | 4 +- .../src/orchestration-v2/Orchestrator.ts | 59 +++ .../ProjectionSettlement.test.ts | 2 +- .../ThreadSettlementService.test.ts | 15 +- .../ThreadSettlementService.ts | 6 +- 6 files changed, 527 insertions(+), 5 deletions(-) create mode 100644 apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts diff --git a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..2b1c529ba17775ef6e3af337540e6ea72a21bb7e --- /dev/null +++ b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts @@ -0,0 +1,446 @@ +import { assert, it } from "@effect/vitest"; +import { + CommandId, + EventId, + MessageId, + NodeId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ProviderThreadId, + ProviderTurnId, + RunAttemptId, + RunId, + ThreadId, + TurnItemId, + type OrchestrationV2DomainEvent, +} from "@t3tools/contracts"; +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 { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; +import * as EffectWorker from "./EffectWorker.ts"; +import * as EventSink from "./EventSink.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import type { + ProviderAdapterV2Event, + ProviderAdapterV2InterruptInput, + ProviderAdapterV2Shape, + ProviderAdapterV2TurnInput, +} from "./ProviderAdapter.ts"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import { makeOrchestratorV2ReplayLayerWithRegistry } from "./testkit/ProviderReplayHarness.ts"; +import { checkpointWorkspace } from "./testkit/ReplayFixtureWorkspace.ts"; + +const driver = ProviderDriverKind.make("codex"); +const instanceId = ProviderInstanceId.make("codex"); +const modelSelection = { instanceId, model: "test-model" }; + +// Codex turns leave commands running, then the thread moves to another +// provider thread (a provider switch). Stop on the newer, settled run must +// reach both provider threads and end all of the Codex work. +it.effect("Stop reaches background work an earlier provider thread still runs", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("background-work-stop"); + const events = yield* Queue.unbounded(); + const started: ProviderAdapterV2TurnInput[] = []; + const interrupts: ProviderAdapterV2InterruptInput[] = []; + const adapter: ProviderAdapterV2Shape = { + 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"), + }; + }), + }; + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const sink = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make("thread:background-work-stop"); + 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:background-work-stop"), + title: "Background work stop", + 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("start-dev-server"), + threadId, + messageId: MessageId.make("message:start-dev-server"), + text: "Start the dev server", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(running); + const first = started[0]!; + const codexTurn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; + const devServerId = TurnItemId.make("turn-item:dev-server"); + const now = yield* DateTime.now; + yield* sink.write({ + events: [ + { + id: EventId.make("dev-server"), + type: "turn-item.updated", + threadId, + runId: first.runId, + occurredAt: now, + payload: { + id: devServerId, + threadId, + runId: first.runId, + nodeId: first.rootNodeId, + providerThreadId: codexTurn.providerThreadId, + providerTurnId: codexTurn.id, + nativeItemRef: null, + parentItemId: null, + ordinal: 100, + status: "running", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + type: "command_execution", + input: "vp run dev --share", + }, + }, + ], + }); + const settled = yield* watch( + (event) => + event.type === "run.updated" && + event.payload.id === first.runId && + event.payload.status === "waiting", + ); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...codexTurn, status: "completed", completedAt: now }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: codexTurn.providerThreadId, + providerTurnId: codexTurn.id, + runOrdinal: first.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(settled); + yield* worker.drain(); + + const codexProviderThread = (yield* orchestrator.getThreadProjection(threadId)) + .providerThreads[0]!; + // A settled run with its root node, attempt, provider turn and, + // optionally, a command or native subagent it left running. + const settledRun = (input: { + readonly ordinal: number; + readonly providerThreadId: ProviderThreadId; + readonly runningItem?: { readonly id: TurnItemId; readonly kind: "command" | "subagent" }; + }) => { + const runId = RunId.make(`run:${input.ordinal}`); + const attemptId = RunAttemptId.make(`attempt:${input.ordinal}`); + const nodeId = NodeId.make(`node:${input.ordinal}`); + const providerTurnId = ProviderTurnId.make(`provider-turn:${input.ordinal}`); + const events: Array = [ + { + id: EventId.make(`run:${input.ordinal}`), + type: "run.created", + threadId, + occurredAt: now, + payload: { + id: runId, + threadId, + ordinal: input.ordinal, + providerInstanceId: instanceId, + modelSelection, + providerThreadId: input.providerThreadId, + userMessageId: MessageId.make(`message:${input.ordinal}`), + rootNodeId: nodeId, + activeAttemptId: attemptId, + status: "completed", + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }, + { + id: EventId.make(`node:${input.ordinal}`), + type: "node.updated", + threadId, + runId, + occurredAt: now, + payload: { + id: nodeId, + threadId, + runId, + parentNodeId: null, + rootNodeId: nodeId, + kind: "root_turn", + status: "completed", + countsForRun: true, + providerThreadId: input.providerThreadId, + providerTurnId, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: now, + }, + }, + { + id: EventId.make(`attempt:${input.ordinal}`), + type: "run-attempt.created", + threadId, + occurredAt: now, + payload: { + id: attemptId, + runId, + attemptOrdinal: 1, + rootNodeId: nodeId, + providerInstanceId: instanceId, + providerThreadId: input.providerThreadId, + providerTurnId, + reason: "initial", + status: "completed", + startedAt: now, + completedAt: now, + }, + }, + { + id: EventId.make(`provider-turn:${input.ordinal}`), + type: "provider-turn.updated", + threadId, + occurredAt: now, + payload: { + id: providerTurnId, + providerThreadId: input.providerThreadId, + nodeId, + runAttemptId: attemptId, + nativeTurnRef: null, + ordinal: input.ordinal, + status: "completed", + startedAt: now, + completedAt: now, + }, + }, + ]; + const item = input.runningItem; + if (item !== undefined) { + const base = { + id: item.id, + threadId, + runId, + nodeId, + providerThreadId: input.providerThreadId, + providerTurnId, + nativeItemRef: null, + parentItemId: null, + ordinal: input.ordinal * 100, + status: "running", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + } as const; + events.push({ + id: EventId.make(`item:${input.ordinal}`), + type: "turn-item.updated", + threadId, + runId, + occurredAt: now, + payload: + item.kind === "command" + ? { ...base, type: "command_execution", input: "vp run test --watch" } + : { + ...base, + // A native subagent item names its own provider thread + // but its parent's provider turn. + providerThreadId: ProviderThreadId.make("provider-thread:codex-subagent"), + type: "subagent", + subagentId: NodeId.make(`subagent:${input.ordinal}`), + origin: "provider_native", + driver, + providerInstanceId: instanceId, + childThreadId: null, + prompt: "Review the change", + result: null, + }, + }); + } + return { runId, providerTurnId, events }; + }; + + // Later Codex runs leave a second command and a native subagent. Then + // the thread moves on to another provider thread, which also has a + // live session. + const watcherId = TurnItemId.make("turn-item:watcher"); + const watcherRun = settledRun({ + ordinal: 2, + providerThreadId: codexProviderThread.id, + runningItem: { id: watcherId, kind: "command" }, + }); + const reviewerId = TurnItemId.make("turn-item:reviewer"); + const reviewerRun = settledRun({ + ordinal: 3, + providerThreadId: codexProviderThread.id, + runningItem: { id: reviewerId, kind: "subagent" }, + }); + const otherProviderThreadId = ProviderThreadId.make("provider-thread:other"); + const latestRun = settledRun({ ordinal: 4, providerThreadId: otherProviderThreadId }); + yield* sink.write({ + events: [ + ...watcherRun.events, + ...reviewerRun.events, + { + id: EventId.make("provider-thread:other"), + type: "provider-thread.updated", + threadId, + occurredAt: now, + payload: { + ...codexProviderThread, + id: otherProviderThreadId, + firstRunOrdinal: 4, + lastRunOrdinal: 4, + }, + }, + ...latestRun.events, + ], + }); + + yield* orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make("stop-background-work"), + threadId, + runId: latestRun.runId, + }); + yield* worker.drain(); + + // Stop reaches both provider threads. The Codex one is interrupted at + // its latest pending work, the subagent's parent turn, so its settle + // covers all three Codex runs. + assert.sameDeepMembers( + interrupts.map((interrupt) => [interrupt.providerThread.id, interrupt.providerTurnId]), + [ + [otherProviderThreadId, latestRun.providerTurnId], + [codexProviderThread.id, reviewerRun.providerTurnId], + ], + ); + const after = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + [devServerId, watcherId, reviewerId].map( + (id) => after.turnItems.find((item) => item.id === id)?.status, + ), + ["interrupted", "interrupted", "interrupted"], + ); + }).pipe( + Effect.provide( + makeOrchestratorV2ReplayLayerWithRegistry( + { name: "background-work-stop" }, + ProviderAdapterRegistry.makeSingleLayer(adapter), + { runEffectWorker: false }, + ), + ), + ); + }), + ), +); diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 1abf070b87ec77d05036878a549f51bce3afc6b1..093806059981d5342577241f07ccbb097dd7e75a 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -169,10 +169,12 @@ export const executorLayer: Layer.Layer< // The provider has stopped what it still ran and reported it. // Whatever the thread still shows on that provider thread is // work no process will report on, so the Stop ends it too. + // One Stop can interrupt several provider threads, so the + // settle is keyed by effect, not by the Stop command. Effect.andThen( threads.dispatch({ type: "thread.background-work.settle", - commandId: CommandId.make(`${effect.commandId}:background-work-settled`), + commandId: CommandId.make(`${effect.id}:background-work-settled`), threadId: effect.threadId, providerThreadId: effect.request.providerThreadId, providerTurnId: effect.request.providerTurnId, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 2707fe5b83441797150e922183f7e12df2d2ce84..1fd4d7a7b8260486920b55101d24374441de6074 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -8108,6 +8108,63 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Run ${command.runId} is not interruptible.`, }); } + // Background work can outlive a provider switch, such as a Codex dev + // server left running when the thread moved to Claude. Stop also reaches + // each other live provider thread that owns pending work; that + // interrupt's settle follow-up ends what its provider leaves behind. + // Work on a dead session is settled with this run's below. + const otherProviderInterrupts: Array = []; + if (hasBackgroundWork) { + const runOrdinals = new Map( + projection.runs.map((candidate) => [candidate.id, candidate.ordinal]), + ); + const runOrdinalOf = (item: (typeof projection.turnItems)[number]) => + item.runId === null ? -1 : (runOrdinals.get(item.runId) ?? -1); + // Each provider thread is interrupted at its latest pending work: that + // turn's run bounds the settle follow-up, which must cover all of the + // thread's work. The target comes from the item's provider turn, since + // a native subagent item names its own provider thread but its + // parent's turn; interrupting the parent's turn reaches the subagent. + const latestTurnByProviderThread = new Map< + OrchestrationV2ProviderThread["id"], + { readonly turn: (typeof projection.providerTurns)[number]; readonly runOrdinal: number } + >(); + for (const item of pendingBackgroundTurnItems({ + turnItems: projection.turnItems, + runs: projection.runs, + })) { + const turn = projection.providerTurns.find( + (candidate) => candidate.id === item.providerTurnId, + ); + if (turn === undefined || turn.providerThreadId === providerThread.id) continue; + const runOrdinal = runOrdinalOf(item); + const latest = latestTurnByProviderThread.get(turn.providerThreadId); + if (latest === undefined || runOrdinal > latest.runOrdinal) { + latestTurnByProviderThread.set(turn.providerThreadId, { turn, runOrdinal }); + } + } + for (const { turn } of latestTurnByProviderThread.values()) { + const owner = projection.providerThreads.find( + (candidate) => candidate.id === turn.providerThreadId, + ); + if (owner === undefined || owner.providerSessionId === null) continue; + const ownerSession = yield* providerSessions + .get(owner.providerSessionId) + .pipe(Effect.orElseSucceed(() => Option.none())); + if (Option.isNone(ownerSession)) continue; + otherProviderInterrupts.push({ + id: `effect:${command.commandId}:provider-turn.interrupt:${turn.id}`, + commandId: command.commandId, + threadId: command.threadId, + request: { + type: "provider-turn.interrupt", + providerSessionId: owner.providerSessionId, + providerThreadId: owner.id, + providerTurnId: turn.id, + }, + }); + } + } // Stop on a settled thread's background work. Its process may be gone // (released, restarted) and only the projection still shows the work; // the settle follow-up ends whatever no provider reports ending. @@ -8139,6 +8196,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio throughRunOrdinal: run.ordinal, now, }); + yield* Ref.update(effects, (existing) => [...existing, ...otherProviderInterrupts]); return undefined; } if (providerThread.providerSessionId === null) { @@ -8200,6 +8258,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio providerTurnId: providerTurn.id, }, } satisfies PendingOrchestrationEffectV2, + ...otherProviderInterrupts, ]); return undefined; }); diff --git a/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts b/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts index 60149fe8a292a2183bca80f28121dd48851f3a76..03b67cc4e4ceead8b80e113d408941552d653f27 100644 --- a/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts @@ -247,7 +247,7 @@ it.effect.each([ forkedFrom: null, createdAt: old, updatedAt: old, - pendingBackgroundTasks: [{ taskId: "running-task", kind: "command" }], + pendingBackgroundTasks: [{ taskId: "running-task", kind: "subagent" }], }, }); const candidates = yield* store.getSettlementCandidates(); diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index 03933d9d004897b805f37e2024e21d86510a6a76..1c0671ddff45de0aa1b339c96696087d7f9f31cb 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -138,12 +138,25 @@ describe("isAutoSettlementCandidate", () => { ).toBe(false); expect( ThreadSettlementService.isAutoSettlementCandidate( - shell({ pendingBackgroundTasks: [{ label: "task" }] as never }), + shell({ pendingBackgroundTasks: [{ taskId: "review", kind: "subagent" }] }), NOW_MS, ), ).toBe(false); }); + it("settles a thread whose only background work is a command left running", () => { + expect( + ThreadSettlementService.isAutoSettlementCandidate( + shell({ + pendingBackgroundTasks: [ + { taskId: "dev", kind: "command", description: "vp run dev --share" }, + ], + }), + NOW_MS, + ), + ).toBe(true); + }); + it("keeps snoozed threads parked until they wake early on error or completion", () => { const snoozed = shell({ snoozedUntil: at(60 * 60 * 1_000), diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index 74c9de8d6b377568d97851003a80cce3fc1d26d0..bcd4f7b1ebc943be30cb156e431b3ae523308d14 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -1,3 +1,4 @@ +import { backgroundWorkHoldsCompletion } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; import { @@ -138,9 +139,10 @@ export function isAutoSettlementCandidate( if (thread.pinnedAt != null || thread.autoSettleDisabledAt != null) return false; // Blocked-on-you work must never park behind a settled override. if (thread.pendingRuntimeRequest !== null) return false; - // A live run — or post-settlement background work — is not staleness. + // A live run, or background work that will wake the agent, is not + // staleness. A dev server left running is: the agent is done. if (thread.activityRunStatus != null) return false; - if ((thread.pendingBackgroundTasks?.length ?? 0) > 0) return false; + if (backgroundWorkHoldsCompletion(thread.pendingBackgroundTasks ?? [])) return false; if (threadHasQueuedTurnStart(thread, nowMs)) return false; const snoozedUntilMs = toMillis(thread.snoozedUntil); if (snoozedUntilMs === null || snoozedUntilMs <= nowMs) return true; -- 2.54.0