diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 884683979fb6..1c4638003cdb 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -419,6 +419,7 @@ const composerSelection = { const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* ( name: string, canRestart = false, + interruptTurn: Effect.Effect = Effect.void, ) { const cwd = yield* checkpointWorkspace(name); const events = yield* Queue.unbounded(); @@ -503,7 +504,7 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* Effect.sync(() => { steered.push(turn.message.text); }), - interruptTurn: () => Effect.void, + interruptTurn: () => interruptTurn, respondToRuntimeRequest: () => Effect.void, readThreadSnapshot: () => Effect.die("unused"), rollbackThread: () => Effect.die("unused"), @@ -878,8 +879,15 @@ it.effect("keeps a queued message queued when promoted into a run that Stop reac it.effect("queues a plain send behind the stopped queue and releases its hold", () => Effect.scoped( Effect.gen(function* () { - const { started, steered, layer, startFirstTurn } = - yield* nextTurnSelectionHarness("plain-send-after-stop"); + const interruptReached = yield* Deferred.make(); + const releaseInterrupt = yield* Deferred.make(); + const { started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( + "plain-send-after-stop", + false, + Deferred.succeed(interruptReached, undefined).pipe( + Effect.andThen(Deferred.await(releaseInterrupt)), + ), + ); yield* Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; @@ -909,10 +917,12 @@ it.effect("queues a plain send behind the stopped queue and releases its hold", }); // The provider has not stopped yet, so a plain send still resolves to a steer. yield* send("sent", "auto"); - yield* worker.drain(); + // Once the stop settles, the run-update reactor promotes the queue on + // its own fiber, so read the queue while the provider is still stopping. + const draining = yield* worker.drain().pipe(Effect.forkScoped); + yield* Deferred.await(interruptReached); const projection = yield* orchestrator.getThreadProjection(threadId); - assert.deepEqual(steered, []); assert.equal(started.length, 1); const queued = projection.runs .filter((run) => run.status === "queued") @@ -925,6 +935,9 @@ it.effect("queues a plain send behind the stopped queue and releases its hold", ); // The user sent it, so the queue they stopped runs again once the provider stops. assert.isFalse(queued.some((run) => run.queueHeld === true)); + yield* Deferred.succeed(releaseInterrupt, undefined); + yield* Fiber.join(draining); + assert.deepEqual(steered, []); }).pipe(Effect.provide(layer)); }), ),