From a998e6c80a747c5599c2c68223922ea3b0cd099e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 20:32:08 +0000 Subject: [PATCH 1/5] test(server): workspace handoff during a run must not interrupt it Co-Authored-By: Claude Opus 5.5 --- .../SelectionRestart.integration.test.ts | 116 ++++++++++++++++++ 1 file changed, 116 insertions(+) diff --git a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts index 03cdf5e85a642..e30252dc19c17 100644 --- a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts +++ b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts @@ -438,6 +438,122 @@ it.live("retries a failed selection restart session open before completing", () ), ); +it.live("a workspace handoff during a run does not leave the run interrupted", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("workspace-handoff-lifecycle"); + const handoffCwd = yield* checkpointWorkspace("workspace-handoff-lifecycle-target"); + const threadId = ThreadId.make("thread:workspace-handoff-lifecycle"); + const state = yield* Ref.make({ + activeTurn: null, + opened: [], + started: [], + closedSessionCount: 0, + failedReplacementOpen: false, + }); + const layerRegistry = ProviderAdapterRegistry.layerSingle(makeRestartAdapter(state)); + + const projection = yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const eventSink = yield* EventSink.EventSinkV2; + + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:workspace-handoff:create"), + threadId, + projectId: ProjectId.make("project:workspace-handoff"), + title: "Workspace handoff", + modelSelection: initialSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + }); + const running = yield* eventSink + .stream({ + threadId, + afterSequence: yield* eventSink.latestSequence({ threadId }), + }) + .pipe( + Stream.map((stored) => stored.event), + Stream.filter( + (event) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:workspace-handoff:first"), + threadId, + messageId: MessageId.make("message:workspace-handoff:first"), + text: "move to a worktree", + attachments: [], + modelSelection: initialSelection, + dispatchMode: { type: "start_immediately" }, + }); + yield* Fiber.join(running); + yield* worker.drain(); + + const settled = yield* eventSink + .stream({ + threadId, + afterSequence: yield* eventSink.latestSequence({ threadId }), + }) + .pipe( + Stream.map((stored) => stored.event), + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.status !== "running" && + event.payload.status !== "queued" && + event.payload.status !== "starting", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + // What t3_worktree_handoff dispatches while the calling turn is still running. + yield* orchestrator.dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make("command:workspace-handoff:rebind"), + threadId, + branch: "repro-handoff", + worktreePath: handoffCwd, + }); + yield* Fiber.join(settled); + yield* worker.drain(); + return yield* orchestrator.getThreadProjection(threadId); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { name: "workspace-handoff-lifecycle" }, + layerRegistry, + ), + ), + ); + + assert.equal(projection.thread.worktreePath, handoffCwd); + assert.notInclude( + ["interrupted", "failed"], + projection.runs[0]?.status, + "an intentional workspace handoff must not settle the handing-off run as interrupted or failed", + ); + assert.isFalse( + projection.turnItems.some((item) => item.type === "run_interrupt_result"), + "an intentional workspace handoff must not project an interrupt result", + ); + }), + ), +); + it.live( "restarts the live session on a model change when a newer stopped session record exists", () => From 0b002c7f45efaff82c5f808536a40d517438f3a4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 20:38:27 +0000 Subject: [PATCH 2/5] fix(server): a worktree handoff no longer marks the handing-off run interrupted Moving a thread to another workspace detaches its provider session, which interrupts the turn that asked for the move. Finalization now settles that turn as completed when the thread's workspace moved and nobody pressed Stop. Co-Authored-By: Claude Opus 5.5 --- .../ProviderTurnStartService.ts | 8 ++++++++ .../orchestration-v2/RunExecutionService.ts | 20 ++++++++++++++++--- 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 4eb988ec43089..27b41c94d0ce5 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -140,6 +140,7 @@ export const layer: Layer.Layer< readonly attemptId: OrchestrationV2RunAttempt["id"]; readonly providerThreadId: OrchestrationV2ProviderThread["id"]; readonly runOrdinal: number; + readonly worktreePath: string | null; readonly inheritedBackgroundTurnItems: ReturnType< typeof RunExecutionService.selectInheritedBackgroundTurnItems >; @@ -191,6 +192,11 @@ export const layer: Layer.Layer< }), ) .pipe(Effect.catchCause(() => Effect.succeed(false))), + hasWorkspaceMoved: () => + projectionStore.getRuntimeRecoveryProjection(input.threadId).pipe( + Effect.map((current) => current.thread.worktreePath !== input.worktreePath), + Effect.catchCause(() => Effect.succeed(false)), + ), loadRunInterruptReason: () => projectionStore .getThreadRecords(input.threadId, ["turnItems"], { @@ -542,6 +548,7 @@ export const layer: Layer.Layer< attemptId: attempt.id, providerThreadId: providerThread.id, runOrdinal: run.ordinal, + worktreePath: projection.thread.worktreePath, inheritedBackgroundTurnItems, }); const { isCurrentAttemptInStatus } = runControls; @@ -1333,6 +1340,7 @@ export const layer: Layer.Layer< shouldFinalizeRun: runControls.shouldFinalizeRun, hasUnpairedRunInterruptRequest: runControls.hasUnpairedRunInterruptRequest, loadRunInterruptReason: runControls.loadRunInterruptReason, + hasWorkspaceMoved: runControls.hasWorkspaceMoved, message: { messageId: message.id, text: userText, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 11ab94cefa69b..652dba58a78b7 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -574,6 +574,7 @@ export interface RunExecutionServiceV2StartRootRunInput { readonly shouldFinalizeRun?: () => Effect.Effect; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly loadRunInterruptReason?: () => Effect.Effect; + readonly hasWorkspaceMoved?: () => Effect.Effect; readonly message: ProviderAdapter.ProviderAdapterV2TurnMessage; readonly modelSelection: ModelSelection; readonly runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy; @@ -1048,6 +1049,16 @@ export const layer: Layer.Layer< if (yield* Ref.get(rootRunFinalized)) { return; } + // Moving the thread to another workspace detaches its provider + // session, which stops the turn that asked for the move. That ends + // the turn as the handoff intended; only a Stop makes it an interrupt. + const settledTerminal: ProviderTerminalEvent = + terminal.status === "interrupted" && + input.hasWorkspaceMoved !== undefined && + !(yield* input.hasUnpairedRunInterruptRequest?.() ?? Effect.succeed(false)) && + (yield* input.hasWorkspaceMoved()) + ? { ...terminal, status: "completed" } + : terminal; const providerThread = yield* Ref.get(latestProviderThread); const openSubagents = yield* Ref.get(openRunOwnedSubagents); yield* writeFinalRunEvents({ @@ -1076,13 +1087,16 @@ export const layer: Layer.Layer< ? {} : { loadRunInterruptReason: input.loadRunInterruptReason }), openRunOwnedSubagents: openSubagents, - terminal, - failureItemPersisted: terminal.status === "failed", + terminal: settledTerminal, + failureItemPersisted: settledTerminal.status === "failed", refreshAfterTurn, }).pipe( Effect.mapError((cause) => new RunExecutionIngestError({ runId: input.run.id, cause })), ); - if (isRunOwnedSubagentTerminalStatus(terminal.status) && !childTurnsContinue(terminal)) { + if ( + isRunOwnedSubagentTerminalStatus(settledTerminal.status) && + !childTurnsContinue(settledTerminal) + ) { yield* Ref.set(openRunOwnedSubagents, emptyOpenRunOwnedSubagentProjection()); } yield* Ref.set(rootRunFinalized, true); From b1d8a170f3722da1fc4520be0b53552f2ed4dd58 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 20:49:18 +0000 Subject: [PATCH 3/5] fix(server): a worktree handoff still settles the run's native subagents Rewriting the handing-off run's interrupted terminal to completed also skipped the subagent cascade, so native subagents that died with the detached session stayed running forever. Cascade them from the raw interrupted terminal while the root run still completes. Co-Authored-By: Claude Opus 5.5 --- .../RunExecutionService.test.ts | 133 ++++++++++++++++++ .../orchestration-v2/RunExecutionService.ts | 16 ++- .../SelectionRestart.integration.test.ts | 7 +- 3 files changed, 144 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index e7937f8a1fc9f..7dcfbafebb61a 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -2349,6 +2349,139 @@ it.effect( }), ); +it.effect("settles native subagents when a workspace handoff completes the root run", () => + Effect.gen(function* () { + const ids = backgroundScenarioIds("subagent-workspace-handoff"); + const providerInstanceId = ProviderInstanceId.make("codex"); + const written = yield* Ref.make>([]); + const ingestionDone = yield* Deferred.make(); + const layerTest = RunExecutionService.layer.pipe( + Layer.provide( + Layer.mergeAll( + McpAppModelContext.layerEmpty, + Layer.mock(CheckpointService.CheckpointServiceV2)({ + captureBaseline: () => Effect.void, + }), + layerTurnItemPositionsTest, + Layer.mock(EventSink.EventSinkV2)({ + write: () => Effect.succeed([]), + writeWithEffects: (input) => + Ref.update(written, (current) => [...current, ...input.events]).pipe(Effect.as([])), + writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }), + }), + IdAllocator.layer, + Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + ingestNormalized: () => Effect.succeed([]), + }), + ServerSettings.layerTest(), + ), + ), + ); + const fixture = { + ids, + providerInstanceId, + childThreadId: ids.childThreadId, + driver, + status: "running", + } as const; + const runningSubagent = makeRunOwnedSubagentFixture(fixture); + const runningTurnItem = makeRunOwnedSubagentTurnItemFixture(fixture); + const runningNode = makeRunOwnedSubagentNodeFixture({ ids, status: "running" }); + const runningChildNode = makeRunOwnedSubagentChildNodeFixture({ ids, status: "running" }); + + yield* Effect.gen(function* () { + const runExecution = yield* RunExecutionService.RunExecutionServiceV2; + yield* runExecution.startRootRun({ + commandId: CommandId.make("command:subagent-workspace-handoff"), + appThread: { id: ids.threadId } as OrchestrationV2AppThread, + providerSessionId: ProviderSessionId.make("session:subagent-workspace-handoff"), + session: { + events: Stream.empty, + subscribeEvents: Effect.succeed({ + // The provider event stream stays open: only settled children may end ingestion. + events: Stream.concat( + Stream.fromIterable([ + childThreadCreatedEvent(ids), + { type: "subagent.updated", driver, subagent: runningSubagent }, + { type: "node.updated", driver, node: runningNode }, + { type: "node.updated", driver, node: runningChildNode }, + { type: "turn_item.updated", driver, turnItem: runningTurnItem }, + rootTerminalEvent(ids, "interrupted"), + ] satisfies ReadonlyArray), + Stream.never, + ), + close: Deferred.succeed(ingestionDone, undefined), + }), + startTurn: () => Effect.void, + } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime, + run: { + id: ids.runId, + threadId: ids.threadId, + ordinal: 1, + providerInstanceId, + } as OrchestrationV2Run, + rootNode: { id: ids.rootNodeId } as OrchestrationV2ExecutionNode, + checkpointScope: { + id: CheckpointScopeId.make("checkpoint-scope:subagent-workspace-handoff"), + } as OrchestrationV2CheckpointScope, + providerThread: { id: ids.providerThreadId, driver } as OrchestrationV2ProviderThread, + attempt: { + id: ids.attemptId, + providerTurnId: ids.rootProviderTurnId, + } as OrchestrationV2RunAttempt, + attemptId: ids.attemptId, + providerTurnOrdinal: 1, + hasUnpairedRunInterruptRequest: () => Effect.succeed(false), + hasWorkspaceMoved: () => Effect.succeed(true), + message: { + messageId: MessageId.make("message:subagent-workspace-handoff:user"), + text: "Spawn a subagent then move to a worktree.", + attachments: [], + createdBy: "user", + creationSource: "web", + }, + modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" }, + runtimePolicy: { + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }); + }).pipe(Effect.provide(layerTest)); + + const closed = yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds")); + assert.isTrue(Option.isSome(closed), "settled children must end event ingestion"); + + const events = yield* Ref.get(written); + const statusOf = (type: OrchestrationV2DomainEvent["type"], id: string) => + events.flatMap((event) => { + if (event.type !== type) return []; + const payload = ( + event as { readonly payload: { readonly id: string; readonly status: string } } + ).payload; + return payload.id === id ? [payload.status] : []; + }); + assert.deepEqual(statusOf("subagent.updated", runningSubagent.id), ["interrupted"]); + assert.deepEqual(statusOf("turn-item.updated", runningTurnItem.id), ["interrupted"]); + assert.deepEqual(statusOf("node.updated", runningNode.id), ["interrupted"]); + assert.deepEqual(statusOf("node.updated", runningChildNode.id), ["interrupted"]); + assert.deepEqual(statusOf("run.updated", ids.runId), ["waiting"]); + assert.deepEqual(statusOf("run-attempt.updated", ids.attemptId), ["completed"]); + assert.isFalse( + events.some( + (event) => + event.type === "turn-item.updated" && event.payload.type === "run_interrupt_result", + ), + ); + }), +); + it.effect("omits interrupt results and subagent cascade for a superseded attempt", () => Effect.gen(function* () { const { written, observed } = yield* captureRootRunTermination({ diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 652dba58a78b7..692d9f2b916b2 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -625,6 +625,7 @@ export const layer: Layer.Layer< readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly loadRunInterruptReason?: () => Effect.Effect; readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection; + readonly openRunOwnedSubagentsTerminal?: ProviderTerminalEvent; readonly interruptedProviderTurn?: OrchestrationV2ProviderTurn; readonly terminal: ProviderTerminalEvent; readonly completedAt?: DateTime.Utc; @@ -692,14 +693,15 @@ export const layer: Layer.Layer< open.turnItems.size > 0 || open.childTurnItems.size > 0 || open.nodes.size > 0; + const subagentTerminal = input.openRunOwnedSubagentsTerminal ?? input.terminal; const cascadedSubagentEvents = - isRunOwnedSubagentTerminalStatus(input.terminal.status) && - !childTurnsContinue(input.terminal) && + isRunOwnedSubagentTerminalStatus(subagentTerminal.status) && + !childTurnsContinue(subagentTerminal) && hasOpenSubagentProjection ? yield* cascadeTerminalizeRunOwnedSubagents({ run: input.run, open, - status: input.terminal.status, + status: subagentTerminal.status, completedAt, allocateEventId, }) @@ -1052,6 +1054,8 @@ export const layer: Layer.Layer< // Moving the thread to another workspace detaches its provider // session, which stops the turn that asked for the move. That ends // the turn as the handoff intended; only a Stop makes it an interrupt. + // Its native subagents died with the session, so they still settle as + // interrupted from the raw terminal. const settledTerminal: ProviderTerminalEvent = terminal.status === "interrupted" && input.hasWorkspaceMoved !== undefined && @@ -1087,16 +1091,14 @@ export const layer: Layer.Layer< ? {} : { loadRunInterruptReason: input.loadRunInterruptReason }), openRunOwnedSubagents: openSubagents, + openRunOwnedSubagentsTerminal: terminal, terminal: settledTerminal, failureItemPersisted: settledTerminal.status === "failed", refreshAfterTurn, }).pipe( Effect.mapError((cause) => new RunExecutionIngestError({ runId: input.run.id, cause })), ); - if ( - isRunOwnedSubagentTerminalStatus(settledTerminal.status) && - !childTurnsContinue(settledTerminal) - ) { + if (isRunOwnedSubagentTerminalStatus(terminal.status) && !childTurnsContinue(terminal)) { yield* Ref.set(openRunOwnedSubagents, emptyOpenRunOwnedSubagentProjection()); } yield* Ref.set(rootRunFinalized, true); diff --git a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts index e30252dc19c17..0f7cee35dae8e 100644 --- a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts +++ b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts @@ -541,11 +541,8 @@ it.live("a workspace handoff during a run does not leave the run interrupted", ( ); assert.equal(projection.thread.worktreePath, handoffCwd); - assert.notInclude( - ["interrupted", "failed"], - projection.runs[0]?.status, - "an intentional workspace handoff must not settle the handing-off run as interrupted or failed", - ); + assert.equal(projection.runs[0]?.status, "completed"); + assert.isNotNull(projection.runs[0]?.checkpointId ?? null); assert.isFalse( projection.turnItems.some((item) => item.type === "run_interrupt_result"), "an intentional workspace handoff must not project an interrupt result", From ca238a6237247def03c551ca1f797f4bac7a855c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 20:59:55 +0000 Subject: [PATCH 4/5] fix(server): a Stop racing a worktree handoff keeps the run interrupted The handoff rewrite read the interrupt request before writing the completed terminal, so a Stop that committed in between was overwritten by a completed run. The completed write now refuses in the same transaction when an unpaired interrupt request is stored, and finalization falls back to the raw interrupted terminal. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/EventSink.ts | 24 ++++ .../FoundationPersistence.test.ts | 101 ++++++++++++++++ .../RunExecutionService.test.ts | 42 ++++++- .../orchestration-v2/RunExecutionService.ts | 109 +++++++++++------- 4 files changed, 231 insertions(+), 45 deletions(-) diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index 3d2bc8e4b45a8..ec375c32e0230 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, @@ -99,6 +100,11 @@ export interface EventSinkV2Shape { readonly runId: RunId; readonly activeAttemptId: RunAttemptId; readonly expectedStatus: OrchestrationV2Run["status"]; + /** Also refuse while this run interrupt request is stored without its result. */ + readonly unlessInterruptRequested?: { + readonly requestId: OrchestrationV2TurnItem["id"]; + readonly resultId: OrchestrationV2TurnItem["id"]; + }; readonly events: ReadonlyArray; readonly effects?: ReadonlyArray; }) => Effect.Effect< @@ -500,6 +506,24 @@ const layerBase: Layer.Layer< storedEvents: [] as ReadonlyArray, }; } + if (input.unlessInterruptRequested !== undefined) { + const { requestId, resultId } = input.unlessInterruptRequested; + const pending = yield* sql<{ readonly pending: number }>` + SELECT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_turn_items + WHERE thread_id = ${input.threadId} AND turn_item_id = ${requestId} + ) AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_turn_items + WHERE thread_id = ${input.threadId} AND turn_item_id = ${resultId} + ) AS pending + `; + if (pending[0]?.pending === 1) { + return { + committed: false as const, + storedEvents: [] as ReadonlyArray, + }; + } + } const normalized = yield* normalizeEvents( input.guardPendingUserInputCancellations === true diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index a37bcf8f2cd86..67c4526023f6d 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -1481,6 +1481,107 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { }).pipe(Effect.provide(Layer.fresh(layerTest))), ); + it.effect("refuses a run-current write once an unpaired interrupt request is stored", () => + Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:foundation-interrupt-guard"); + const runId = RunId.make("run:foundation-interrupt-guard"); + const attemptId = RunAttemptId.make("run-attempt:foundation-interrupt-guard"); + const requestId = TurnItemId.make("turn-item:foundation-interrupt-guard:request"); + const resultId = TurnItemId.make("turn-item:foundation-interrupt-guard:result"); + const run: OrchestrationV2Run = { + id: runId, + threadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make("message:foundation-interrupt-guard"), + rootNodeId: null, + activeAttemptId: attemptId, + status: "running", + queuePosition: null, + requestedAt: now, + startedAt: now, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + yield* eventSink.write({ + events: [ + threadCreatedEvent({ + id: "event:foundation-interrupt-guard:thread", + thread: makeThread(threadId, now), + now, + }), + { + id: EventId.make("event:foundation-interrupt-guard:run"), + type: "run.created", + threadId, + runId, + providerInstanceId, + occurredAt: now, + payload: run, + }, + ], + }); + const signalItem = (id: TurnItemId, type: "run_interrupt_request" | "run_interrupt_result") => + ({ + id: EventId.make(`event:${id}`), + type: "turn-item.updated", + threadId, + runId, + occurredAt: now, + payload: { + id, + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: "Interrupt", + startedAt: now, + completedAt: now, + updatedAt: now, + type, + message: "Stopped", + }, + }) as OrchestrationV2DomainEvent; + const completeRun = (suffix: string) => + eventSink.writeIfRunCurrent({ + threadId, + runId, + activeAttemptId: attemptId, + expectedStatus: "running", + unlessInterruptRequested: { requestId, resultId }, + events: [ + { + id: EventId.make(`event:foundation-interrupt-guard:waiting:${suffix}`), + type: "run.updated", + threadId, + runId, + providerInstanceId, + occurredAt: now, + payload: { ...run, status: "waiting" }, + }, + ], + }); + + yield* eventSink.write({ events: [signalItem(requestId, "run_interrupt_request")] }); + const refused = yield* completeRun("refused"); + assert.isFalse(refused.committed); + assert.deepEqual(refused.storedEvents, []); + + yield* eventSink.write({ events: [signalItem(resultId, "run_interrupt_result")] }); + assert.isTrue((yield* completeRun("paired")).committed); + }).pipe(Effect.provide(Layer.fresh(layerTest))), + ); + it.effect("guards post-terminal provider-thread writes by attempt and run ordinal", () => Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 7dcfbafebb61a..042e92abb3efb 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -2367,7 +2367,10 @@ it.effect("settles native subagents when a workspace handoff completes the root write: () => Effect.succeed([]), writeWithEffects: (input) => Ref.update(written, (current) => [...current, ...input.events]).pipe(Effect.as([])), - writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }), + writeIfRunCurrent: (input) => + Ref.update(written, (current) => [...current, ...input.events]).pipe( + Effect.as({ committed: true, storedEvents: [] }), + ), }), IdAllocator.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ @@ -2431,7 +2434,7 @@ it.effect("settles native subagents when a workspace handoff completes the root } as OrchestrationV2RunAttempt, attemptId: ids.attemptId, providerTurnOrdinal: 1, - hasUnpairedRunInterruptRequest: () => Effect.succeed(false), + shouldFinalizeRun: () => Effect.succeed(true), hasWorkspaceMoved: () => Effect.succeed(true), message: { messageId: MessageId.make("message:subagent-workspace-handoff:user"), @@ -2521,6 +2524,37 @@ it.effect("emits run_interrupt_result when superseded attempt still has a hard-s }), ); +it.effect("keeps a Stop that commits while a workspace handoff completes the run", () => + Effect.gen(function* () { + const stopCommitted = yield* Ref.make(false); + const { written, finalEvents, committedEffects } = yield* captureRootRunTermination({ + key: "stop-races-workspace-handoff", + shouldFinalizeRun: () => Effect.succeed(true), + // Stop commits after the handoff is detected and before the terminal write. + hasWorkspaceMoved: () => Ref.set(stopCommitted, true).pipe(Effect.as(true)), + // The sink refuses a guarded write once the Stop request is stored. + beforeTerminalWrite: (request) => + Ref.get(stopCommitted).pipe( + Effect.map((stopped) => !(stopped && request.unlessInterruptRequested !== undefined)), + ), + }); + const statuses = (type: OrchestrationV2DomainEvent["type"]) => + finalEvents.flatMap((event) => + event.type === type ? [(event.payload as { readonly status: string }).status] : [], + ); + assert.deepEqual(statuses("run.updated"), ["interrupted"]); + assert.deepEqual(statuses("run-attempt.updated"), ["interrupted"]); + assert.deepEqual( + written.map((item) => item.type), + ["run_interrupt_result"], + ); + assert.deepEqual( + committedEffects.map((effect) => effect.request.type), + ["checkpoint.capture"], + ); + }), +); + it.effect("does not overwrite Stop when ownership changes after the finalization read", () => Effect.gen(function* () { const { written, observed, submittedEffects, committedEffects } = @@ -2835,6 +2869,7 @@ function captureRootRunTermination(input: { readonly shouldFinalizeRun: () => Effect.Effect; readonly rejectTerminalWrite?: boolean; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; + readonly hasWorkspaceMoved?: () => Effect.Effect; readonly seedOpenSubagent?: boolean; readonly events?: ( ids: BackgroundScenarioIds, @@ -3022,6 +3057,9 @@ function captureRootRunTermination(input: { : { hasUnpairedRunInterruptRequest: input.hasUnpairedRunInterruptRequest, }), + ...(input.hasWorkspaceMoved === undefined + ? {} + : { hasWorkspaceMoved: input.hasWorkspaceMoved }), message: { messageId: MessageId.make(`message:${input.key}`), text: "interrupt projection", diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 692d9f2b916b2..2d4f197cb9230 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -634,6 +634,7 @@ export const layer: Layer.Layer< readonly writeIfRunCurrent?: { readonly activeAttemptId: RunAttemptId; readonly expectedStatus: OrchestrationV2Run["status"]; + readonly unlessInterruptRequested?: boolean; }; }) { const completedAt = input.completedAt ?? (yield* DateTime.now); @@ -680,7 +681,7 @@ export const layer: Layer.Layer< yield* input.refreshAfterTurn; } } - return; + return false; } const allocateEventId = () => idAllocator.allocate.event({ threadId: input.run.threadId }); const reason = @@ -871,16 +872,31 @@ export const layer: Layer.Layer< runId: input.run.id, activeAttemptId: input.writeIfRunCurrent.activeAttemptId, expectedStatus: input.writeIfRunCurrent.expectedStatus, + ...(input.writeIfRunCurrent.unlessInterruptRequested === true + ? { + unlessInterruptRequested: { + requestId: idAllocator.derive.runSignalTurnItem({ + runId: input.run.id, + signal: "interrupt-request", + }), + resultId: idAllocator.derive.runSignalTurnItem({ + runId: input.run.id, + signal: "interrupt-result", + }), + }, + } + : {}), events: finalization.events, effects: finalization.effects, }); if (!result.committed) { - return; + return false; } } else { yield* eventSink.writeWithEffects(finalization); } yield* input.refreshAfterTurn; + return true; }, ); @@ -1051,53 +1067,60 @@ export const layer: Layer.Layer< if (yield* Ref.get(rootRunFinalized)) { return; } + const providerThread = yield* Ref.get(latestProviderThread); + const openSubagents = yield* Ref.get(openRunOwnedSubagents); + const writeTerminal = (settledTerminal: ProviderTerminalEvent) => + writeFinalRunEvents({ + run: input.run, + rootNode: input.rootNode, + checkpointScope: input.checkpointScope, + providerThread, + attempt: input.attempt, + ...(input.shouldFinalizeRun === undefined + ? {} + : { + shouldFinalizeRun: input.shouldFinalizeRun, + // Stop may commit between the ownership read and this + // terminal write. Gate the events and checkpoint together. + writeIfRunCurrent: { + activeAttemptId: input.attempt.id, + expectedStatus: "running" as const, + unlessInterruptRequested: settledTerminal !== terminal, + }, + }), + ...(input.hasUnpairedRunInterruptRequest === undefined + ? {} + : { + hasUnpairedRunInterruptRequest: input.hasUnpairedRunInterruptRequest, + }), + ...(input.loadRunInterruptReason === undefined + ? {} + : { loadRunInterruptReason: input.loadRunInterruptReason }), + openRunOwnedSubagents: openSubagents, + openRunOwnedSubagentsTerminal: terminal, + terminal: settledTerminal, + failureItemPersisted: settledTerminal.status === "failed", + refreshAfterTurn, + }).pipe( + Effect.mapError( + (cause) => new RunExecutionIngestError({ runId: input.run.id, cause }), + ), + ); // Moving the thread to another workspace detaches its provider // session, which stops the turn that asked for the move. That ends // the turn as the handoff intended; only a Stop makes it an interrupt. // Its native subagents died with the session, so they still settle as - // interrupted from the raw terminal. - const settledTerminal: ProviderTerminalEvent = + // interrupted from the raw terminal. The completed write refuses in + // the same transaction as a committed Stop request, so a Stop racing + // the handoff falls back to the interrupted terminal. + const handedOff = terminal.status === "interrupted" && + input.shouldFinalizeRun !== undefined && input.hasWorkspaceMoved !== undefined && - !(yield* input.hasUnpairedRunInterruptRequest?.() ?? Effect.succeed(false)) && - (yield* input.hasWorkspaceMoved()) - ? { ...terminal, status: "completed" } - : terminal; - const providerThread = yield* Ref.get(latestProviderThread); - const openSubagents = yield* Ref.get(openRunOwnedSubagents); - yield* writeFinalRunEvents({ - run: input.run, - rootNode: input.rootNode, - checkpointScope: input.checkpointScope, - providerThread, - attempt: input.attempt, - ...(input.shouldFinalizeRun === undefined - ? {} - : { - shouldFinalizeRun: input.shouldFinalizeRun, - // Stop may commit between the ownership read and this - // terminal write. Gate the events and checkpoint together. - writeIfRunCurrent: { - activeAttemptId: input.attempt.id, - expectedStatus: "running" as const, - }, - }), - ...(input.hasUnpairedRunInterruptRequest === undefined - ? {} - : { - hasUnpairedRunInterruptRequest: input.hasUnpairedRunInterruptRequest, - }), - ...(input.loadRunInterruptReason === undefined - ? {} - : { loadRunInterruptReason: input.loadRunInterruptReason }), - openRunOwnedSubagents: openSubagents, - openRunOwnedSubagentsTerminal: terminal, - terminal: settledTerminal, - failureItemPersisted: settledTerminal.status === "failed", - refreshAfterTurn, - }).pipe( - Effect.mapError((cause) => new RunExecutionIngestError({ runId: input.run.id, cause })), - ); + (yield* input.hasWorkspaceMoved()); + if (!handedOff || !(yield* writeTerminal({ ...terminal, status: "completed" }))) { + yield* writeTerminal(terminal); + } if (isRunOwnedSubagentTerminalStatus(terminal.status) && !childTurnsContinue(terminal)) { yield* Ref.set(openRunOwnedSubagents, emptyOpenRunOwnedSubagentProjection()); } From cc1cd4d68cadfc4df69895f243b8394da828d1a5 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 01:14:41 +0000 Subject: [PATCH 5/5] test(server): align workspace handoff test with main's run execution layer Provide PreviewAgentActivity, use mockEventSink and rootAppThread like the neighbouring tests, and trim the handoff comment to the comment length cap. AI-assisted; verified with the tests and tools listed above. --- .../src/orchestration-v2/RunExecutionService.test.ts | 5 +++-- .../src/orchestration-v2/RunExecutionService.ts | 11 ++++------- 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 74b4f82307124..beba39a06a63e 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -2590,11 +2590,12 @@ it.effect("settles native subagents when a workspace handoff completes the root Layer.provide( Layer.mergeAll( McpAppModelContext.layerEmpty, + PreviewAgentActivity.layer, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void, }), layerTurnItemPositionsTest, - Layer.mock(EventSink.EventSinkV2)({ + mockEventSink({ write: () => Effect.succeed([]), writeWithEffects: (input) => Ref.update(written, (current) => [...current, ...input.events]).pipe(Effect.as([])), @@ -2627,7 +2628,7 @@ it.effect("settles native subagents when a workspace handoff completes the root const runExecution = yield* RunExecutionService.RunExecutionServiceV2; yield* runExecution.startRootRun({ commandId: CommandId.make("command:subagent-workspace-handoff"), - appThread: { id: ids.threadId } as OrchestrationV2AppThread, + appThread: rootAppThread(ids.threadId), providerSessionId: ProviderSessionId.make("session:subagent-workspace-handoff"), session: { events: Stream.empty, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index d43534d04e972..c8fe7a6577d79 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1153,13 +1153,10 @@ export const layer: Layer.Layer< (cause) => new RunExecutionIngestError({ runId: input.run.id, cause }), ), ); - // Moving the thread to another workspace detaches its provider - // session, which stops the turn that asked for the move. That ends - // the turn as the handoff intended; only a Stop makes it an interrupt. - // Its native subagents died with the session, so they still settle as - // interrupted from the raw terminal. The completed write refuses in - // the same transaction as a committed Stop request, so a Stop racing - // the handoff falls back to the interrupted terminal. + // A workspace handoff detaches the session and stops the turn that asked + // for it: the run completes, while its native subagents still settle + // as interrupted. The completed write refuses in the same transaction + // as a stored Stop request, so a racing Stop stays authoritative. const handedOff = terminal.status === "interrupted" && input.shouldFinalizeRun !== undefined &&