From b51547cc9d6b69812f5e3602cbc6a8af37c58c61 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 09:37:22 +0000 Subject: [PATCH 1/8] test(server): reproduce #32891 restart racing turn end or stop A restart_active message.dispatch whose target run already completed, or that a Stop has reached before the provider stopped, fails with "cannot be steered" instead of starting the message as a new turn. Co-Authored-By: Claude Opus 5.5 --- .../SteeringCompletion.integration.test.ts | 108 ++++++++++++++++++ 1 file changed, 108 insertions(+) diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 2b08c0b8345b6..0624f2d00f3cd 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1441,6 +1441,114 @@ it.effect("keeps a queued message queued when promoted into a run that Stop reac ), ); +// regression: #32891 a restart that Stop already reached is lost +it.effect("starts a restart sent into a run that Stop reached once the stop settles", () => + Effect.scoped( + Effect.gen(function* () { + const { started, steered, layer, startFirstTurn } = + yield* nextTurnSelectionHarness("restart-after-stop"); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const first = started[0]!; + // The worker has not delivered the interrupt yet, so the run still reads as running. + yield* orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make("stop"), + threadId, + runId: first.runId, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("restart"), + threadId, + messageId: MessageId.make("message:restart"), + text: "restart", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "restart_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + + assert.deepEqual(steered, []); + assert.deepEqual( + started.map((turn) => turn.message.text), + ["first", "restart"], + ); + }).pipe(Effect.provide(layer)); + }), + ), +); + +// regression: #32891 a restart that arrives after the turn ended is lost +it.effect("starts a restart that missed the turn as a new turn", () => + Effect.scoped( + Effect.gen(function* () { + const { events, started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( + "restart-after-completion", + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const first = started[0]!; + const settled = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === first.runId && + event.payload.status === "waiting", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...turn, status: "completed", completedAt: yield* DateTime.now }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: turn.providerThreadId, + providerTurnId: turn.id, + runOrdinal: first.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(settled); + yield* worker.drain(); + + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("restart"), + threadId, + messageId: MessageId.make("message:restart"), + text: "restart", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "restart_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + + assert.deepEqual(steered, []); + assert.deepEqual( + started.map((turn) => turn.message.text), + ["first", "restart"], + ); + }).pipe(Effect.provide(layer)); + }), + ), +); + it.effect("queues a plain send behind the stopped queue and releases its hold", () => Effect.scoped( Effect.gen(function* () { From 024aab1acfe28d38e0f4595b4aae3c197c478254 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 09:44:01 +0000 Subject: [PATCH 2/8] fix(server): a restart that misses its turn starts as a new turn A restart_active dispatch whose target run already ended, whose provider turn completed, or that a Stop already reached failed as an unsteerable run and lost the message. Restart means interrupt and start, so when the interrupt has already happened only the start is left: the dispatch now becomes start_immediately, which waits behind a stopping run and keeps a stopped queue's hold. Explicit steers are unchanged. Fixes pingdotgg/t3code#32891 Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/Orchestrator.ts | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 5d3c4391b1fe8..12de01bc1dc0a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -5831,7 +5831,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command.dispatchMode, command.deliveryIntent, ); - if (dispatchMode.type === "steer_active") { + if (dispatchMode.type === "steer_active" || dispatchMode.type === "restart_active") { const targetRunId = dispatchMode.targetRunId; const target = projection.runs.find((run) => run.id === targetRunId); const turn = latestProviderTurnForAttempt( @@ -5843,6 +5843,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if ( target !== undefined && (target.status === "completed" || + (dispatchMode.type === "restart_active" && !isActiveRun(target)) || ((target.status === "running" || target.status === "waiting") && turn !== undefined && turn.status === "completed")) @@ -5853,7 +5854,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // Stop leaves the run reading as running until its provider stops. A plain // send in that window queues behind the stopped queue instead of steering // text the agent never receives, and like any user send it releases the - // hold; an explicit steer or restart still fails below. + // hold; an explicit steer still fails below. if ( command.deliveryIntent === "auto" && (dispatchMode.type === "steer_active" || dispatchMode.type === "restart_active") && @@ -5861,6 +5862,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) { dispatchMode = { type: "queue_after_active" }; } + // An explicit restart that Stop reached has only its start left; it starts + // once the stop settles and leaves the stopped queue's hold in place. + if ( + dispatchMode.type === "restart_active" && + (yield* stopReachedRun(command, command.threadId, dispatchMode.targetRunId)) + ) { + dispatchMode = { type: "start_immediately" }; + } const freshContext = command.freshContext === true; if ( freshContext && From 8f4808547707464d3627a0f90bf343ecf3b7276f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 09:58:57 +0000 Subject: [PATCH 3/8] fix(server): cover a restart into an interrupted run; correct the hold comment Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 4 +- .../SteeringCompletion.integration.test.ts | 122 +++++++++--------- 2 files changed, 64 insertions(+), 62 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 12de01bc1dc0a..ccf4d659b89f7 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -5862,8 +5862,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) { dispatchMode = { type: "queue_after_active" }; } - // An explicit restart that Stop reached has only its start left; it starts - // once the stop settles and leaves the stopped queue's hold in place. + // An explicit restart that Stop reached has only its start left, so it + // queues behind the stopping run like any other start. if ( dispatchMode.type === "restart_active" && (yield* stopReachedRun(command, command.threadId, dispatchMode.targetRunId)) diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 0624f2d00f3cd..7db6cdfeee4f5 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1484,69 +1484,71 @@ it.effect("starts a restart sent into a run that Stop reached once the stop sett ); // regression: #32891 a restart that arrives after the turn ended is lost -it.effect("starts a restart that missed the turn as a new turn", () => - Effect.scoped( - Effect.gen(function* () { - const { events, started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( - "restart-after-completion", - ); - yield* Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; - const threadId = yield* startFirstTurn; - const first = started[0]!; - const settled = yield* orchestrator.streamDomainEvents.pipe( - Stream.filter( - (event) => - event.type === "run.updated" && - event.payload.id === first.runId && - event.payload.status === "waiting", - ), - Stream.take(1), - Stream.runDrain, - Effect.forkScoped, +it.effect.each(["completed", "interrupted"] as const)( + "starts a restart that missed the turn (%s) as a new turn", + (status) => + Effect.scoped( + Effect.gen(function* () { + const { events, started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( + `restart-after-${status}`, ); - const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; - yield* Queue.offer(events, { - type: "provider_turn.updated", - driver, - providerTurn: { ...turn, status: "completed", completedAt: yield* DateTime.now }, - }); - yield* Queue.offer(events, { - type: "turn.terminal", - driver, - providerThreadId: turn.providerThreadId, - providerTurnId: turn.id, - runOrdinal: first.runOrdinal, - status: "completed", - failure: null, - threadDisposition: "reusable", - }); - yield* Fiber.join(settled); - yield* worker.drain(); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const first = started[0]!; + const settled = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === first.runId && + event.payload.status === (status === "completed" ? "waiting" : "interrupted"), + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...turn, status, completedAt: yield* DateTime.now }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: turn.providerThreadId, + providerTurnId: turn.id, + runOrdinal: first.runOrdinal, + status, + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(settled); + yield* worker.drain(); - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make("restart"), - threadId, - messageId: MessageId.make("message:restart"), - text: "restart", - attachments: [], - modelSelection: runSelection, - dispatchMode: { type: "restart_active", targetRunId: first.runId }, - createdBy: "user", - creationSource: "web", - }); - yield* worker.drain(); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("restart"), + threadId, + messageId: MessageId.make("message:restart"), + text: "restart", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "restart_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); - assert.deepEqual(steered, []); - assert.deepEqual( - started.map((turn) => turn.message.text), - ["first", "restart"], - ); - }).pipe(Effect.provide(layer)); - }), - ), + assert.deepEqual(steered, []); + assert.deepEqual( + started.map((turn) => turn.message.text), + ["first", "restart"], + ); + }).pipe(Effect.provide(layer)); + }), + ), ); it.effect("queues a plain send behind the stopped queue and releases its hold", () => From d3ff1efef01c1e2c1bea8e28e2971340c9ef8225 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 10:43:33 +0000 Subject: [PATCH 4/8] fix(server): a restart into a turn that ended before its run settled starts as a new turn Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 5 +- .../SteeringCompletion.integration.test.ts | 80 ++++++++++++++----- 2 files changed, 63 insertions(+), 22 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 149512402f6f7..a95e698a1fc93 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6079,7 +6079,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (dispatchMode.type === "restart_active" && !isActiveRun(target)) || ((target.status === "running" || target.status === "waiting") && turn !== undefined && - turn.status === "completed")) + (turn.status === "completed" || + (dispatchMode.type === "restart_active" && + turn.status !== "pending" && + turn.status !== "running")))) ) { dispatchMode = { type: "start_immediately" }; } diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index edc60574d7c0f..ac7db072d57be 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1816,48 +1816,70 @@ it.effect("starts a restart sent into a run that Stop reached once the stop sett ); // regression: #32891 a restart that arrives after the turn ended is lost -it.effect.each(["completed", "interrupted"] as const)( - "starts a restart that missed the turn (%s) as a new turn", - (status) => +it.effect.each([ + { status: "completed", settled: true, starts: true }, + { status: "interrupted", settled: true, starts: true }, + { status: "interrupted", settled: false, starts: true }, + // A provider failure holds the queue, so the restart waits like any queued message. + { status: "failed", settled: false, starts: false }, + { status: "cancelled", settled: false, starts: true }, +] as const)( + "keeps a restart that missed the turn ($status, run settled: $settled) as a new turn", + ({ status, settled: runSettled, starts }) => Effect.scoped( Effect.gen(function* () { const { events, started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( - `restart-after-${status}`, + `restart-after-${status}-${runSettled ? "settled" : "unsettled"}`, ); yield* Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; const threadId = yield* startFirstTurn; const first = started[0]!; - const settled = yield* orchestrator.streamDomainEvents.pipe( + const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; + const turnEnded = yield* orchestrator.streamDomainEvents.pipe( Stream.filter( (event) => - event.type === "run.updated" && - event.payload.id === first.runId && - event.payload.status === (status === "completed" ? "waiting" : "interrupted"), + event.type === "provider-turn.updated" && + event.payload.id === turn.id && + event.payload.status === status, ), Stream.take(1), Stream.runDrain, Effect.forkScoped, ); - const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; yield* Queue.offer(events, { type: "provider_turn.updated", driver, providerTurn: { ...turn, status, completedAt: yield* DateTime.now }, }); - yield* Queue.offer(events, { - type: "turn.terminal", - driver, - providerThreadId: turn.providerThreadId, - providerTurnId: turn.id, - runOrdinal: first.runOrdinal, - status, - failure: null, - threadDisposition: "reusable", + yield* Fiber.join(turnEnded); + const settleRun = Effect.gen(function* () { + const settled = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === first.runId && + event.payload.status !== "running", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: turn.providerThreadId, + providerTurnId: turn.id, + runOrdinal: first.runOrdinal, + status, + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(settled); + yield* worker.drain(); }); - yield* Fiber.join(settled); - yield* worker.drain(); + if (runSettled) yield* settleRun; yield* orchestrator.dispatch({ type: "message.dispatch", @@ -1872,11 +1894,27 @@ it.effect.each(["completed", "interrupted"] as const)( creationSource: "web", }); yield* worker.drain(); + if (!runSettled) { + const restartLeftQueue = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.userMessageId === "message:restart" && + (starts ? event.payload.status !== "queued" : event.payload.queueHeld === true), + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* settleRun; + yield* Fiber.join(restartLeftQueue); + yield* worker.drain(); + } assert.deepEqual(steered, []); assert.deepEqual( started.map((turn) => turn.message.text), - ["first", "restart"], + starts ? ["first", "restart"] : ["first"], ); }).pipe(Effect.provide(layer)); }), From c987223f2417e25b5061e71ea405a56d781ceeef Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 10:58:59 +0000 Subject: [PATCH 5/8] test(server): drop the failed-turn restart row, which needs a provider failure the fixture cannot build Co-Authored-By: Claude Opus 5.5 --- .../SteeringCompletion.integration.test.ts | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index ac7db072d57be..c286f54a5f9ac 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1817,15 +1817,13 @@ it.effect("starts a restart sent into a run that Stop reached once the stop sett // regression: #32891 a restart that arrives after the turn ended is lost it.effect.each([ - { status: "completed", settled: true, starts: true }, - { status: "interrupted", settled: true, starts: true }, - { status: "interrupted", settled: false, starts: true }, - // A provider failure holds the queue, so the restart waits like any queued message. - { status: "failed", settled: false, starts: false }, - { status: "cancelled", settled: false, starts: true }, + { status: "completed", settled: true }, + { status: "interrupted", settled: true }, + { status: "interrupted", settled: false }, + { status: "cancelled", settled: false }, ] as const)( - "keeps a restart that missed the turn ($status, run settled: $settled) as a new turn", - ({ status, settled: runSettled, starts }) => + "starts a restart that missed the turn ($status, run settled: $settled) as a new turn", + ({ status, settled: runSettled }) => Effect.scoped( Effect.gen(function* () { const { events, started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( @@ -1900,7 +1898,7 @@ it.effect.each([ (event) => event.type === "run.updated" && event.payload.userMessageId === "message:restart" && - (starts ? event.payload.status !== "queued" : event.payload.queueHeld === true), + event.payload.status !== "queued", ), Stream.take(1), Stream.runDrain, @@ -1914,7 +1912,7 @@ it.effect.each([ assert.deepEqual(steered, []); assert.deepEqual( started.map((turn) => turn.message.text), - starts ? ["first", "restart"] : ["first"], + ["first", "restart"], ); }).pipe(Effect.provide(layer)); }), From 48782caff66ceaf2a289a23dac415b6427bdc723 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 11:04:32 +0000 Subject: [PATCH 6/8] test(server): cover a restart into a completed turn whose run has not settled Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/SteeringCompletion.integration.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index c286f54a5f9ac..04a9dd66848ca 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1818,6 +1818,7 @@ it.effect("starts a restart sent into a run that Stop reached once the stop sett // regression: #32891 a restart that arrives after the turn ended is lost it.effect.each([ { status: "completed", settled: true }, + { status: "completed", settled: false }, { status: "interrupted", settled: true }, { status: "interrupted", settled: false }, { status: "cancelled", settled: false }, From 57c9f29ea38266be28fed197d6c12790694a010d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 11:40:00 +0000 Subject: [PATCH 7/8] test(server): cover a restart into a failed provider turn whose run has not settled Co-Authored-By: Claude Opus 5.5 --- .../SteeringCompletion.integration.test.ts | 56 +++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 04a9dd66848ca..703c9ef7a8002 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1920,6 +1920,62 @@ it.effect.each([ ), ); +// regression: #32891 a restart into a failed provider turn is lost +it.effect("queues a restart sent into a provider turn that failed before its run settled", () => + Effect.scoped( + Effect.gen(function* () { + const { events, started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( + "restart-after-failed-unsettled", + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const first = started[0]!; + const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; + const turnFailed = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && + event.payload.id === turn.id && + event.payload.status === "failed", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...turn, status: "failed", completedAt: yield* DateTime.now }, + }); + yield* Fiber.join(turnFailed); + + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("restart"), + threadId, + messageId: MessageId.make("message:restart"), + text: "restart", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "restart_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + + assert.deepEqual(steered, []); + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + projection.runs.find((run) => run.userMessageId === "message:restart")?.status, + "queued", + ); + }).pipe(Effect.provide(layer)); + }), + ), +); + it.effect("queues a plain send behind the stopped queue and releases its hold", () => Effect.scoped( Effect.gen(function* () { From abf9f927869681e66c2ac8f4a7ee9aaa314f19d2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 11:51:09 +0000 Subject: [PATCH 8/8] test(server): an explicit steer into a run Stop reached is still refused Co-Authored-By: Claude Opus 5.5 --- .../SteeringCompletion.integration.test.ts | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 703c9ef7a8002..37b2ab259f840 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1774,7 +1774,7 @@ it.effect("keeps a queued message queued when promoted into a run that Stop reac ); // regression: #32891 a restart that Stop already reached is lost -it.effect("starts a restart sent into a run that Stop reached once the stop settles", () => +it.effect("refuses a steer but starts a restart sent into a run that Stop reached", () => Effect.scoped( Effect.gen(function* () { const { started, steered, layer, startFirstTurn } = @@ -1791,6 +1791,22 @@ it.effect("starts a restart sent into a run that Stop reached once the stop sett threadId, runId: first.runId, }); + const steerError = yield* orchestrator + .dispatch({ + type: "message.dispatch", + commandId: CommandId.make("steer"), + threadId, + messageId: MessageId.make("message:steer"), + text: "steer", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "steer_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }) + .pipe(Effect.flip); + assert.equal(steerError._tag, "OrchestratorDispatchError"); + assert.include(String(steerError.cause), "is stopping and cannot be steered"); yield* orchestrator.dispatch({ type: "message.dispatch", commandId: CommandId.make("restart"),