From 7d598b56109987c6c1db23714db72cc6f2f7856d Mon Sep 17 00:00:00 2001
From: Kit Langton <kit.langton@gmail.com>
Date: Thu, 2 Jul 2026 21:15:15 -0400
Subject: [PATCH] refactor(core): drop synchronous drain start from the
 coordinator

Drains now uniformly start one tick after wake or run, removing the
ready-deferred release that ran the first drain synchronously inside
the caller. The synchronous start was observable only by tests that
asserted state immediately after fire-and-forget calls, coupling them
to scheduler internals; those tests now synchronize explicitly through
session.wait or stream-start signals.
---
 packages/core/src/session/run-coordinator.ts | 14 ++++++--------
 packages/core/test/session-runner.test.ts    |  5 ++++-
 2 files changed, 10 insertions(+), 9 deletions(-)

diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts
index 518f264dd1..c8af3de69b 100644
--- a/packages/core/src/session/run-coordinator.ts
+++ b/packages/core/src/session/run-coordinator.ts
@@ -49,29 +49,27 @@ export const make = <Key, E>(options: {
         ),
       )
 
-    const start = (key: Key, force: boolean, defer = false) => {
+    const start = (key: Key, force: boolean) => {
       const execution: Execution<E> = { done: Deferred.makeUnsafe<void, E>(), pendingWake: false, stopping: false }
       executions.set(key, execution)
-      const ready = Deferred.makeUnsafe<void>()
+      // The leading yield lets `owner` be assigned before the drain can settle, and keeps
+      // failing self-waking executions from growing the stack across successor starts.
+      // Drains start one tick after wake; callers observe progress through events or run.
       execution.owner = fork(
-        (defer ? Effect.yieldNow : Deferred.await(ready)).pipe(
+        Effect.yieldNow.pipe(
           Effect.andThen(loop(key, execution, force)),
           Effect.onExit((exit) => Effect.sync(() => settle(key, execution, exit))),
           Effect.exit,
           Effect.asVoid,
         ),
       )
-      // Releasing after `owner` is assigned starts the drain synchronously (a prompt's wake
-      // reaches the provider in the same tick) without racing ownership. Successor starts
-      // defer one tick instead so failing self-waking executions cannot grow the stack.
-      if (!defer) Deferred.doneUnsafe(ready, Effect.void)
       return execution
     }
 
     // A doorbell that survives the execution loop (rung after the loop decided to end, or
     // during failure or interruption cleanup) starts a fresh execution for the remaining work.
     const settle = (key: Key, execution: Execution<E>, exit: Exit.Exit<void, E>) => {
-      if (execution.pendingWake) start(key, false, true)
+      if (execution.pendingWake) start(key, false)
       else executions.delete(key)
       Deferred.doneUnsafe(execution.done, exit)
     }
diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts
index c07946c788..bce33797f6 100644
--- a/packages/core/test/session-runner.test.ts
+++ b/packages/core/test/session-runner.test.ts
@@ -655,6 +655,7 @@ describe("SessionRunnerLLM", () => {
       response = []
 
       const message = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Run automatically" }) })
+      yield* session.wait(sessionID)
 
       expect(requests).toHaveLength(1)
       expect(yield* session.messages({ sessionID })).toMatchObject([
@@ -714,6 +715,7 @@ describe("SessionRunnerLLM", () => {
 
       systemUnavailable = false
       yield* session.prompt({ id: messageID, sessionID, prompt: Prompt.make({ text: "First" }) })
+      yield* session.wait(sessionID)
 
       expect(requests).toHaveLength(1)
       expect(requests[0]?.messages.map((message) => message.role)).toEqual(["user"])
@@ -2537,8 +2539,9 @@ describe("SessionRunnerLLM", () => {
 
       const first = yield* session.resume(sessionID).pipe(Effect.forkChild)
       yield* Deferred.await(streamStarted)
+      streamStarted = yield* Deferred.make<void>()
       const second = yield* session.resume(otherSessionID).pipe(Effect.forkChild)
-      yield* Effect.yieldNow
+      yield* Deferred.await(streamStarted)
 
       expect(requests).toHaveLength(2)
       expect(requests.map((request) => request.providerOptions?.openai?.promptCacheKey)).toEqual([