diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index 0c42ac73ff9cb..81ec4019f2c0f 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, @@ -127,6 +128,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< @@ -525,6 +531,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 acf3cc27f6211..76a8460c2b624 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -1984,6 +1984,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/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 98652036118fc..98a9807e634a9 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -233,6 +233,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 >; @@ -284,6 +285,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"], { @@ -695,6 +701,7 @@ export const layer: Layer.Layer< attemptId: attempt.id, providerThreadId: providerThread.id, runOrdinal: run.ordinal, + worktreePath: projection.thread.worktreePath, inheritedBackgroundTurnItems, }); const { isCurrentAttemptInStatus } = runControls; @@ -1709,6 +1716,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.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index e25b4c7ab556a..beba39a06a63e 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -2580,6 +2580,143 @@ 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, + PreviewAgentActivity.layer, + Layer.mock(CheckpointService.CheckpointServiceV2)({ + captureBaseline: () => Effect.void, + }), + layerTurnItemPositionsTest, + mockEventSink({ + write: () => Effect.succeed([]), + writeWithEffects: (input) => + Ref.update(written, (current) => [...current, ...input.events]).pipe(Effect.as([])), + writeIfRunCurrent: (input) => + Ref.update(written, (current) => [...current, ...input.events]).pipe( + Effect.as({ 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: rootAppThread(ids.threadId), + 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, + shouldFinalizeRun: () => Effect.succeed(true), + 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({ @@ -2619,6 +2756,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 } = @@ -3180,6 +3348,7 @@ function captureRootRunTermination(input: { readonly shouldFinalizeRun: () => Effect.Effect; readonly rejectTerminalWrite?: boolean; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; + readonly hasWorkspaceMoved?: () => Effect.Effect; readonly loadRunInterruptReason?: () => Effect.Effect; readonly hasPendingBackgroundWorkForThread?: ProviderAdapter.ProviderAdapterV2SessionRuntime["hasPendingBackgroundWorkForThread"]; readonly seedOpenSubagent?: boolean; @@ -3376,6 +3545,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 e981ab7362d2d..c8fe7a6577d79 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -586,6 +586,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; @@ -639,6 +640,7 @@ export const layer: Layer.Layer< readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly loadRunInterruptReason?: () => Effect.Effect; readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection; + readonly openRunOwnedSubagentsTerminal?: ProviderTerminalEvent; // Root turn snapshot written with the run when T3 ends the run itself. readonly abandonedProviderTurn?: OrchestrationV2ProviderTurn; readonly terminal: ProviderTerminalEvent; @@ -648,6 +650,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); @@ -694,7 +697,7 @@ export const layer: Layer.Layer< yield* input.refreshAfterTurn; } } - return; + return false; } const allocateEventId = () => idAllocator.allocate.event({ threadId: input.run.threadId }); const reason = @@ -709,14 +712,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, }) @@ -874,11 +878,25 @@ 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); @@ -890,6 +908,7 @@ export const layer: Layer.Layer< childrenRunning: hasOpenSubagentProjection && cascadedSubagentEvents.length === 0, }); yield* input.refreshAfterTurn; + return true; }, ); @@ -1097,38 +1116,55 @@ export const layer: Layer.Layer< } 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, - terminal, - failureItemPersisted: terminal.status === "failed", - refreshAfterTurn, - }).pipe( - Effect.mapError((cause) => new RunExecutionIngestError({ runId: input.run.id, cause })), - ); + 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 }), + ), + ); + // 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 && + input.hasWorkspaceMoved !== undefined && + (yield* input.hasWorkspaceMoved()); + if (!handedOff || !(yield* writeTerminal({ ...terminal, status: "completed" }))) { + yield* writeTerminal(terminal); + } if (isRunOwnedSubagentTerminalStatus(terminal.status) && !childTurnsContinue(terminal)) { yield* Ref.set(openRunOwnedSubagents, emptyOpenRunOwnedSubagentProjection()); idleFlushClosed = true; diff --git a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts index 4352d8b2d6477..ba49f0bcfba53 100644 --- a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts +++ b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts @@ -446,6 +446,119 @@ 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.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", + ); + }), + ), +); + it.live("Stop during a restart interrupts the turn the restart could not stop", () => Effect.scoped( Effect.gen(function* () {