From 3439564e193dc71bcecbb4f62de6fc3e9530dad3 Mon Sep 17 00:00:00 2001 From: Xin Huang Date: Sat, 11 Jul 2026 15:06:45 +0800 Subject: [PATCH] fix(runner): queue work when already running instead of discarding When a background subagent completes and calls ops.prompt() to notify the parent session, the parent's Runner is often still Running (processing the previous turn). The old ensureRunning discarded the new work item and just awaited the current run's completion, so the notification was lost and the parent session would hang until the user typed 'continue'. Add a RunningThenRun state to the Runner state machine. When ensureRunning is called while Running, the new work is stored as pending and the runner transitions to RunningThenRun. When the current run finishes, finishRun starts the pending work and transitions back to Running. Cancel in the RunningThenRun state interrupts both the current and pending work. This fixes the root cause of the parent-not-notified bug. The alternative approach in PR #35041 moves the notification to the session completion path but still races through ensureRunning and can lose the notification when the parent is busy. Refs: #35066 --- packages/opencode/src/effect/runner.ts | 59 +++++++++--- packages/opencode/test/effect/runner.test.ts | 94 +++++++++++++++++++- 2 files changed, 137 insertions(+), 16 deletions(-) diff --git a/packages/opencode/src/effect/runner.ts b/packages/opencode/src/effect/runner.ts index f21a61c97e57..7e2f60d57cea 100644 --- a/packages/opencode/src/effect/runner.ts +++ b/packages/opencode/src/effect/runner.ts @@ -35,6 +35,7 @@ export type State = | { readonly _tag: "Running"; readonly run: RunHandle } | { readonly _tag: "Shell"; readonly shell: ShellHandle } | { readonly _tag: "ShellThenRun"; readonly shell: ShellHandle; readonly run: PendingHandle } + | { readonly _tag: "RunningThenRun"; readonly run: RunHandle; readonly pending: PendingHandle } export const make = ( scope: Scope.Scope, @@ -67,20 +68,35 @@ export const make = ( 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, exit: Exit.Exit) => - SynchronizedRef.modify( + const finishRun = ( + id: number, + done: Deferred.Deferred, + exit: Exit.Exit, + ): Effect.Effect => + 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, done: Deferred.Deferred) => + const startRun = ( + work: Effect.Effect, + done: Deferred.Deferred, + ): Effect.Effect> => Effect.gen(function* () { const id = next() const fiber = yield* work.pipe( @@ -117,9 +133,18 @@ export const make = ( 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(), + work, + } satisfies PendingHandle + 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 = ( }), { _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) diff --git a/packages/opencode/test/effect/runner.test.ts b/packages/opencode/test/effect/runner.test.ts index 27fe9e02542e..88e0b14d9e2f 100644 --- a/packages/opencode/test/effect/runner.test.ts +++ b/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(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(s) + const gate = yield* Deferred.make() + + 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(s) + const gate = yield* Deferred.make() + 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(s, { onInterrupt: Effect.succeed("fallback") }) + const gate = yield* Deferred.make() + + 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* () {