fix(runner): queue work when already running instead of discarding

packages/opencode/src/effect/runner.ts

@@ -35,6 +35,7 @@ export type State<A, E> =
   | { readonly _tag: "Running"; readonly run: RunHandle<A, E> }
   | { readonly _tag: "Shell"; readonly shell: ShellHandle<A, E> }
   | { readonly _tag: "ShellThenRun"; readonly shell: ShellHandle<A, E>; readonly run: PendingHandle<A, E> }
+  | { readonly _tag: "RunningThenRun"; readonly run: RunHandle<A, E>; readonly pending: PendingHandle<A, E> }
 
 export const make = <A, E = never>(
   scope: Scope.Scope,
@@ -67,20 +68,35 @@ export const make = <A, E = never>(
   const idleIfCurrent = () =>
     SynchronizedRef.modify(ref, (st) => [st._tag === "Idle" ? idle : Effect.void, st] as const).pipe(Effect.flatten)
 
-  const finishRun = (id: number, done: Deferred.Deferred<A, E | Cancelled>, exit: Exit.Exit<A, E>) =>
-    SynchronizedRef.modify(
+  const finishRun = (
+    id: number,
+    done: Deferred.Deferred<A, E | Cancelled>,
+    exit: Exit.Exit<A, E>,
+  ): Effect.Effect<void> =>
+    SynchronizedRef.modifyEffect(
       ref,
-      (st) =>
-        [
-          Effect.gen(function* () {
-            if (st._tag === "Running" && st.run.id === id) yield* idle
-            yield* complete(done, exit)
-          }),
-          st._tag === "Running" && st.run.id === id ? ({ _tag: "Idle" } as const) : st,
-        ] as const,
+      Effect.fnUntraced(function* (st) {
+        if (st._tag === "Running" && st.run.id === id) {
+          return [
+            Effect.gen(function* () {
+              yield* idle
+              yield* complete(done, exit)
+            }),
+            { _tag: "Idle" } as const,
+          ] as const
+        }
+        if (st._tag === "RunningThenRun" && st.run.id === id) {
+          const run = yield* startRun(st.pending.work, st.pending.done)
+          return [complete(done, exit), { _tag: "Running", run } as const] as const
+        }
+        return [complete(done, exit), st] as const
+      }),
     ).pipe(Effect.flatten)
 
-  const startRun = (work: Effect.Effect<A, E>, done: Deferred.Deferred<A, E | Cancelled>) =>
+  const startRun = (
+    work: Effect.Effect<A, E>,
+    done: Deferred.Deferred<A, E | Cancelled>,
+  ): Effect.Effect<RunHandle<A, E>> =>
     Effect.gen(function* () {
       const id = next()
       const fiber = yield* work.pipe(
@@ -117,9 +133,18 @@ export const make = <A, E = never>(
       ref,
       Effect.fnUntraced(function* (st) {
         switch (st._tag) {
-          case "Running":
           case "ShellThenRun":
             return [awaitDone(st.run.done), st] as const
+          case "RunningThenRun":
+            return [awaitDone(st.pending.done), st] as const
+          case "Running": {
+            const pending = {
+              id: next(),
+              done: yield* Deferred.make<A, E | Cancelled>(),
+              work,
+            } satisfies PendingHandle<A, E>
+            return [awaitDone(pending.done), { _tag: "RunningThenRun", run: st.run, pending }] as const
+          }
           case "Shell": {
             const run = {
               id: next(),
@@ -198,6 +223,16 @@ export const make = <A, E = never>(
           }),
           { _tag: "Idle" } as const,
         ] as const
+      case "RunningThenRun":
+        return [
+          Effect.gen(function* () {
+            yield* Fiber.interrupt(st.run.fiber)
+            yield* Deferred.fail(st.run.done, new Cancelled()).pipe(Effect.asVoid)
+            yield* Deferred.fail(st.pending.done, new Cancelled()).pipe(Effect.asVoid)
+            yield* idleIfCurrent()
+          }),
+          { _tag: "Idle" } as const,
+        ] as const
     }
   }).pipe(Effect.flatten)
 

packages/opencode/test/effect/runner.test.ts

@@ -52,7 +52,7 @@ describe("Runner", () => {
 
       expect(a).toBe("shared")
       expect(b).toBe("shared")
-      expect(yield* Ref.get(calls)).toBe(1)
+      expect(yield* Ref.get(calls)).toBe(2) // queued run also increments
     }),
   )
 
@@ -87,7 +87,7 @@ describe("Runner", () => {
   )
 
   it.live(
-    "second ensureRunning ignores new work if already running",
+    "second ensureRunning queues behind running and executes after",
     Effect.gen(function* () {
       const s = yield* Scope.Scope
       const runner = Runner.make<string>(s)
@@ -108,13 +108,99 @@ describe("Runner", () => {
       })
 
       expect(a).toBe("first-result")
-      expect(b).toBe("first-result")
-      expect(yield* Ref.get(ran)).toEqual(["first"])
+      expect(b).toBe("second-result")
+      expect(yield* Ref.get(ran)).toEqual(["first", "second"])
     }),
   )
 
   // --- cancel semantics ---
 
+  it.live(
+    "ensureRunning queues behind running and transitions through RunningThenRun",
+    Effect.gen(function* () {
+      const s = yield* Scope.Scope
+      const runner = Runner.make<string>(s)
+      const gate = yield* Deferred.make<void>()
+
+      const first = Deferred.await(gate).pipe(Effect.as("first"))
+      const second = Effect.succeed("second")
+
+      const a = yield* runner.ensureRunning(first).pipe(Effect.forkChild)
+      yield* waitForState(runner, "Running")
+
+      const b = yield* runner.ensureRunning(second).pipe(Effect.forkChild)
+      yield* waitForState(runner, "RunningThenRun")
+      expect(runner.state._tag).toBe("RunningThenRun")
+
+      yield* Deferred.succeed(gate, undefined)
+      yield* Fiber.await(a)
+
+      const exitB = yield* Fiber.await(b)
+      expect(Exit.isSuccess(exitB)).toBe(true)
+      if (Exit.isSuccess(exitB)) expect(exitB.value).toBe("second")
+      expect(runner.state._tag).toBe("Idle")
+    }),
+  )
+
+  it.live(
+    "third ensureRunning shares pending in RunningThenRun",
+    Effect.gen(function* () {
+      const s = yield* Scope.Scope
+      const runner = Runner.make<string>(s)
+      const gate = yield* Deferred.make<void>()
+      const calls = yield* Ref.make(0)
+
+      const first = Deferred.await(gate).pipe(Effect.as("first"))
+      const work = Effect.gen(function* () {
+        yield* Ref.update(calls, (n) => n + 1)
+        return "queued"
+      })
+
+      const a = yield* runner.ensureRunning(first).pipe(Effect.forkChild)
+      yield* waitForState(runner, "Running")
+
+      const b = yield* runner.ensureRunning(work).pipe(Effect.forkChild)
+      yield* waitForState(runner, "RunningThenRun")
+
+      const c = yield* runner.ensureRunning(work).pipe(Effect.forkChild)
+      yield* Effect.yieldNow
+      expect(runner.state._tag).toBe("RunningThenRun")
+
+      yield* Deferred.succeed(gate, undefined)
+      yield* Fiber.await(a)
+
+      const [exitB, exitC] = yield* Effect.all([Fiber.await(b), Fiber.await(c)])
+      expect(Exit.isSuccess(exitB)).toBe(true)
+      expect(Exit.isSuccess(exitC)).toBe(true)
+      if (Exit.isSuccess(exitB)) expect(exitB.value).toBe("queued")
+      if (Exit.isSuccess(exitC)) expect(exitC.value).toBe("queued")
+    }),
+  )
+
+  it.live(
+    "cancel in RunningThenRun cancels both current and pending",
+    Effect.gen(function* () {
+      const s = yield* Scope.Scope
+      const runner = Runner.make<string>(s, { onInterrupt: Effect.succeed("fallback") })
+      const gate = yield* Deferred.make<void>()
+
+      const a = yield* runner.ensureRunning(Deferred.await(gate).pipe(Effect.as("first"))).pipe(Effect.forkChild)
+      yield* waitForState(runner, "Running")
+
+      const b = yield* runner.ensureRunning(Effect.succeed("second")).pipe(Effect.forkChild)
+      yield* waitForState(runner, "RunningThenRun")
+
+      yield* runner.cancel
+      expect(runner.busy).toBe(false)
+
+      const [exitA, exitB] = yield* Effect.all([Fiber.await(a), Fiber.await(b)])
+      expect(Exit.isSuccess(exitA)).toBe(true)
+      expect(Exit.isSuccess(exitB)).toBe(true)
+      if (Exit.isSuccess(exitA)) expect(exitA.value).toBe("fallback")
+      if (Exit.isSuccess(exitB)) expect(exitB.value).toBe("fallback")
+    }),
+  )
+
   it.live(
     "cancel interrupts running work",
     Effect.gen(function* () {