From 570d6549e6fe7bf494355e763a09e5f92b9dfd07 Mon Sep 17 00:00:00 2001 From: Yanhao Zhu Date: Tue, 11 Aug 2026 21:17:10 +0800 Subject: [PATCH 1/3] fix(session): interrupt a running prompt when a new one is submitted When the user submits a new prompt while a run is active, the Runner's ensureRunning simply awaited the current run's completion and dropped the new work. If the assistant was blocked on a long-running tool like `sleep`, the new prompt was not answered until the tool finished. Now ensureRunning interrupts the active run and starts the new work, so a new prompt is handled promptly and the current tool (e.g. a shell command) is cancelled. The interrupted caller still resolves through onInterrupt. Adds a regression test that submits a second prompt while a bash `sleep` is running and asserts the second prompt completes immediately. --- packages/opencode/src/effect/runner.ts | 17 ++- packages/opencode/test/effect/runner.test.ts | 68 +++++++----- packages/opencode/test/session/prompt.test.ts | 104 ++++++++++++++++-- 3 files changed, 148 insertions(+), 41 deletions(-) diff --git a/packages/opencode/src/effect/runner.ts b/packages/opencode/src/effect/runner.ts index f21a61c97e57..67d16a753a5d 100644 --- a/packages/opencode/src/effect/runner.ts +++ b/packages/opencode/src/effect/runner.ts @@ -117,7 +117,22 @@ export const make = ( ref, Effect.fnUntraced(function* (st) { switch (st._tag) { - case "Running": + case "Running": { + // A new run is requested while one is active — e.g. the user submits + // a new prompt while the assistant is blocked on a long-running tool + // like `sleep`. Interrupt the current run and start the new work so + // the new prompt is handled promptly instead of waiting for the tool + // to finish. The interrupted caller resolves through `onInterrupt`. + // The fiber interrupt runs on a separate fiber: the interrupted run's + // `finishRun` acquires this same ref, so interrupting it while holding + // the lock would deadlock. + const old = st.run + yield* Deferred.fail(old.done, new Cancelled()).pipe(Effect.asVoid) + const done = yield* Deferred.make() + const run = yield* startRun(work, done) + yield* Effect.suspend(() => Fiber.interrupt(old.fiber)).pipe(Effect.forkIn(scope)) + return [awaitDone(done), { _tag: "Running", run }] as const + } case "ShellThenRun": return [awaitDone(st.run.done), st] as const case "Shell": { diff --git a/packages/opencode/test/effect/runner.test.ts b/packages/opencode/test/effect/runner.test.ts index 27fe9e02542e..4201cfc5eab7 100644 --- a/packages/opencode/test/effect/runner.test.ts +++ b/packages/opencode/test/effect/runner.test.ts @@ -34,28 +34,6 @@ describe("Runner", () => { }), ) - it.live( - "concurrent callers share the same run", - Effect.gen(function* () { - const s = yield* Scope.Scope - const runner = Runner.make(s) - const calls = yield* Ref.make(0) - const work = Effect.gen(function* () { - yield* Ref.update(calls, (n) => n + 1) - yield* Effect.sleep("10 millis") - return "shared" - }) - - const [a, b] = yield* Effect.all([runner.ensureRunning(work), runner.ensureRunning(work)], { - concurrency: "unbounded", - }) - - expect(a).toBe("shared") - expect(b).toBe("shared") - expect(yield* Ref.get(calls)).toBe(1) - }), - ) - it.live( "concurrent callers all receive same error", Effect.gen(function* () { @@ -87,29 +65,59 @@ describe("Runner", () => { ) it.live( - "second ensureRunning ignores new work if already running", + "second ensureRunning interrupts the active run and replaces it", Effect.gen(function* () { const s = yield* Scope.Scope const runner = Runner.make(s) + const started = yield* Deferred.make() const ran = yield* Ref.make([]) const first = Effect.gen(function* () { yield* Ref.update(ran, (a) => [...a, "first"]) - yield* Effect.sleep("50 millis") - return "first-result" + yield* Deferred.succeed(started, void 0) + return yield* Effect.never.pipe(Effect.as("first-result")) }) const second = Effect.gen(function* () { yield* Ref.update(ran, (a) => [...a, "second"]) return "second-result" }) - const [a, b] = yield* Effect.all([runner.ensureRunning(first), runner.ensureRunning(second)], { - concurrency: "unbounded", + const a = yield* runner.ensureRunning(first).pipe(Effect.forkChild) + yield* Deferred.await(started) + const b = yield* runner.ensureRunning(second).pipe(Effect.forkChild) + + // The first caller is interrupted; the second caller gets the new result. + const exitA = yield* Fiber.await(a) + const exitB = yield* Fiber.await(b) + expect(Exit.isFailure(exitA)).toBe(true) + expect(Exit.isSuccess(exitB)).toBe(true) + if (Exit.isSuccess(exitB)) expect(exitB.value).toBe("second-result") + expect(yield* Ref.get(ran)).toEqual(["first", "second"]) + }), + ) + + it.live( + "interrupted first caller resolves through onInterrupt", + Effect.gen(function* () { + const s = yield* Scope.Scope + const runner = Runner.make(s, { onInterrupt: Effect.succeed("interrupted") }) + const started = yield* Deferred.make() + + const first = Effect.gen(function* () { + yield* Deferred.succeed(started, void 0) + return yield* Effect.never.pipe(Effect.as("first-result")) }) - expect(a).toBe("first-result") - expect(b).toBe("first-result") - expect(yield* Ref.get(ran)).toEqual(["first"]) + const a = yield* runner.ensureRunning(first).pipe(Effect.forkChild) + yield* Deferred.await(started) + const b = yield* runner.ensureRunning(Effect.succeed("second-result")).pipe(Effect.forkChild) + + const exitA = yield* Fiber.await(a) + const exitB = yield* Fiber.await(b) + expect(Exit.isSuccess(exitA)).toBe(true) + if (Exit.isSuccess(exitA)) expect(exitA.value).toBe("interrupted") + expect(Exit.isSuccess(exitB)).toBe(true) + if (Exit.isSuccess(exitB)) expect(exitB.value).toBe("second-result") }), ) diff --git a/packages/opencode/test/session/prompt.test.ts b/packages/opencode/test/session/prompt.test.ts index 5a0176abc9b0..25f062fe654e 100644 --- a/packages/opencode/test/session/prompt.test.ts +++ b/packages/opencode/test/session/prompt.test.ts @@ -1340,7 +1340,7 @@ it.instance( ) it.instance( - "cancel with queued callers resolves all cleanly", + "cancel resolves concurrent loop callers cleanly", () => Effect.gen(function* () { const { llm } = yield* useServerConfig(providerCfg) @@ -1357,11 +1357,10 @@ it.instance( yield* prompt.cancel(chat.id) const [exitA, exitB] = yield* Effect.all([Fiber.await(a), Fiber.await(b)]) + // The second caller interrupts the first and runs its own loop; cancel + // must still resolve both callers without hanging. expect(Exit.isSuccess(exitA)).toBe(true) expect(Exit.isSuccess(exitB)).toBe(true) - if (Exit.isSuccess(exitA) && Exit.isSuccess(exitB)) { - expect(exitA.value.info.id).toBe(exitB.value.info.id) - } }), { git: true }, 10_000, @@ -1384,7 +1383,7 @@ noLLMServer.instance("concurrent loop callers get same result", () => }), ) -it.instance("concurrent loop callers all receive same error result", () => +it.instance("concurrent loop callers both resolve when LLM fails", () => Effect.gen(function* () { const { llm } = yield* useServerConfig(providerCfg) const prompt = yield* SessionPrompt.Service @@ -1394,11 +1393,18 @@ it.instance("concurrent loop callers all receive same error result", () => yield* llm.fail("boom") yield* user(chat.id, "hello") - const [a, b] = yield* Effect.all([prompt.loop({ sessionID: chat.id }), prompt.loop({ sessionID: chat.id })], { - concurrency: "unbounded", - }) - expect(a.info.id).toBe(b.info.id) - expect(a.info.role).toBe("assistant") + const a = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild) + // Wait for the first run to pass its model lookup so the second run reads + // the cached instance state instead of sharing the first run's in-flight + // lookup (which is interrupted with the first run). + yield* llm.wait(1) + const b = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild) + + const [exitA, exitB] = yield* Effect.all([Fiber.await(a), Fiber.await(b)]) + // The second caller replaces the first, so both resolve to an assistant + // message rather than sharing a single run. + expect(Exit.isSuccess(exitA)).toBe(true) + expect(Exit.isSuccess(exitB)).toBe(true) }), ) @@ -1956,6 +1962,84 @@ unix( 30_000, ) +unix( + "a new prompt takes over a running bash tool instead of waiting for it", + () => + Effect.gen(function* () { + const { dir, llm } = yield* useServerConfig(providerCfg) + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const afs = yield* FSUtil.Service + const chat = yield* sessions.create({ + title: "Prompt while sleep", + permission: [{ permission: "*", pattern: "*", action: "allow" }], + }) + const ready = path.join(dir, ".bash-ready") + + const lastUserIncludes = (text: string) => (hit: { body: Record }) => { + const messages = Array.isArray(hit.body.messages) ? hit.body.messages : [] + for (let i = messages.length - 1; i >= 0; i--) { + const msg = messages[i] + if (!msg || typeof msg !== "object") continue + if (!("role" in msg) || msg.role !== "user") continue + const content = "content" in msg ? msg.content : undefined + if (typeof content === "string") return content.includes(text) + if (Array.isArray(content)) { + return content.some( + (part) => + part !== null && + typeof part === "object" && + "text" in part && + typeof part.text === "string" && + part.text.includes(text), + ) + } + return false + } + return false + } + + yield* prompt.prompt({ + sessionID: chat.id, + agent: "build", + noReply: true, + parts: [{ type: "text", text: "run bash" }], + }) + yield* llm.pushMatch( + lastUserIncludes("run bash"), + reply() + .tool("bash", { command: `touch "${ready}"; sleep 30`, timeout: 30_000, workdir: path.resolve(dir) }) + .item(), + ) + yield* llm.pushMatch(lastUserIncludes("second"), reply().text("second done").stop().item()) + + const first = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild) + yield* pollWithTimeout( + afs.existsSafe(ready).pipe(Effect.map((exists) => (exists ? (true as const) : undefined))), + "bash tool never started", + ) + + yield* prompt.prompt({ + sessionID: chat.id, + agent: "build", + noReply: true, + parts: [{ type: "text", text: "second" }], + }) + + const exit = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.timeout("10 seconds"), Effect.exit) + expect(Exit.isSuccess(exit)).toBe(true) + if (Exit.isSuccess(exit)) { + const parts = exit.value.parts.filter((part) => part.type === "text") + expect(parts.some((part) => part.type === "text" && part.text === "second done")).toBe(true) + } + + yield* prompt.cancel(chat.id) + yield* Fiber.await(first).pipe(Effect.timeout("5 seconds"), Effect.exit) + }), + { git: true }, + 30_000, +) + unixNoLLMServer( "cancel interrupts loop queued behind shell", () => From ae7ef3e9288e768cf9e06366c0d70026ee5a3980 Mon Sep 17 00:00:00 2001 From: Yanhao Zhu Date: Thu, 13 Aug 2026 12:56:08 +0800 Subject: [PATCH 2/3] fix(session): wait for interruption cleanup --- packages/opencode/src/effect/runner.ts | 4 +- packages/opencode/test/effect/runner.test.ts | 87 ++++++++++++++++++++ 2 files changed, 89 insertions(+), 2 deletions(-) diff --git a/packages/opencode/src/effect/runner.ts b/packages/opencode/src/effect/runner.ts index 67d16a753a5d..674fd0d560aa 100644 --- a/packages/opencode/src/effect/runner.ts +++ b/packages/opencode/src/effect/runner.ts @@ -125,9 +125,9 @@ export const make = ( // to finish. The interrupted caller resolves through `onInterrupt`. // The fiber interrupt runs on a separate fiber: the interrupted run's // `finishRun` acquires this same ref, so interrupting it while holding - // the lock would deadlock. + // the lock would deadlock. `finishRun` also completes the old deferred + // after interruption cleanup, so the old caller cannot return early. const old = st.run - yield* Deferred.fail(old.done, new Cancelled()).pipe(Effect.asVoid) const done = yield* Deferred.make() const run = yield* startRun(work, done) yield* Effect.suspend(() => Fiber.interrupt(old.fiber)).pipe(Effect.forkIn(scope)) diff --git a/packages/opencode/test/effect/runner.test.ts b/packages/opencode/test/effect/runner.test.ts index 4201cfc5eab7..aa0dc4c4ade0 100644 --- a/packages/opencode/test/effect/runner.test.ts +++ b/packages/opencode/test/effect/runner.test.ts @@ -121,6 +121,93 @@ describe("Runner", () => { }), ) + it.live( + "interrupted caller waits for cleanup before resolving through onInterrupt", + Effect.gen(function* () { + const s = yield* Scope.Scope + const started = yield* Deferred.make() + const cleanupStarted = yield* Deferred.make() + const releaseCleanup = yield* Deferred.make() + const releaseSecond = yield* Deferred.make() + const runner = Runner.make(s, { onInterrupt: Effect.succeed("interrupted") }) + + const first = Effect.gen(function* () { + yield* Deferred.succeed(started, undefined) + return yield* Effect.never.pipe( + Effect.onInterrupt(() => Deferred.succeed(cleanupStarted, undefined)), + Effect.ensuring(Deferred.await(releaseCleanup)), + Effect.as("first-result"), + ) + }) + + const a = yield* runner.ensureRunning(first).pipe(Effect.forkChild) + yield* Deferred.await(started) + const observer = yield* Effect.gen(function* () { + yield* Deferred.await(cleanupStarted) + return a.pollUnsafe() + }).pipe( + Effect.ensuring( + Effect.all([Deferred.succeed(releaseCleanup, undefined), Deferred.succeed(releaseSecond, undefined)], { + discard: true, + }), + ), + Effect.forkChild, + ) + const b = yield* runner + .ensureRunning(Deferred.await(releaseSecond).pipe(Effect.as("second-result"))) + .pipe(Effect.forkChild) + + const early = yield* Fiber.join(observer).pipe(Effect.timeout("250 millis")) + const exitA = yield* Fiber.await(a).pipe(Effect.timeout("250 millis")) + const exitB = yield* Fiber.await(b).pipe(Effect.timeout("250 millis")) + expect(Exit.isSuccess(exitA)).toBe(true) + if (Exit.isSuccess(exitA)) expect(exitA.value).toBe("interrupted") + expect(Exit.isSuccess(exitB)).toBe(true) + if (Exit.isSuccess(exitB)) expect(exitB.value).toBe("second-result") + expect(early).toBeUndefined() + }), + ) + + it.live( + "replaced run does not publish idle while replacement is active", + Effect.gen(function* () { + const s = yield* Scope.Scope + const started = yield* Deferred.make() + const cleanupStarted = yield* Deferred.make() + const releaseCleanup = yield* Deferred.make() + const releaseSecond = yield* Deferred.make() + const events = yield* Ref.make([]) + const runner = Runner.make(s, { + onIdle: Ref.update(events, (items) => [...items, "idle"]), + }) + + const first = Effect.gen(function* () { + yield* Deferred.succeed(started, undefined) + return yield* Effect.never.pipe( + Effect.onInterrupt(() => Deferred.succeed(cleanupStarted, undefined)), + Effect.ensuring(Deferred.await(releaseCleanup)), + Effect.as("first-result"), + ) + }) + + const a = yield* runner.ensureRunning(first).pipe(Effect.exit, Effect.forkChild) + yield* Deferred.await(started) + const b = yield* runner + .ensureRunning(Deferred.await(releaseSecond).pipe(Effect.as("second-result"))) + .pipe(Effect.forkChild) + yield* Deferred.await(cleanupStarted) + yield* Deferred.succeed(releaseCleanup, undefined) + yield* Fiber.join(a) + + expect(runner.busy).toBe(true) + expect(yield* Ref.get(events)).toEqual([]) + + yield* Deferred.succeed(releaseSecond, undefined) + expect(yield* Fiber.join(b)).toBe("second-result") + expect(yield* Ref.get(events)).toEqual(["idle"]) + }), + ) + // --- cancel semantics --- it.live( From 602f8acd7d58d95e61ff77bcde463714b27b5425 Mon Sep 17 00:00:00 2001 From: Yanhao Zhu Date: Thu, 13 Aug 2026 14:14:41 +0800 Subject: [PATCH 3/3] fix(session): serialize prompt replacements --- packages/opencode/src/effect/runner.ts | 21 ++++-- packages/opencode/test/effect/runner.test.ts | 68 ++++++++++++++++++-- 2 files changed, 79 insertions(+), 10 deletions(-) diff --git a/packages/opencode/src/effect/runner.ts b/packages/opencode/src/effect/runner.ts index 674fd0d560aa..cdf9150064ed 100644 --- a/packages/opencode/src/effect/runner.ts +++ b/packages/opencode/src/effect/runner.ts @@ -123,14 +123,23 @@ export const make = ( // like `sleep`. Interrupt the current run and start the new work so // the new prompt is handled promptly instead of waiting for the tool // to finish. The interrupted caller resolves through `onInterrupt`. - // The fiber interrupt runs on a separate fiber: the interrupted run's - // `finishRun` acquires this same ref, so interrupting it while holding - // the lock would deadlock. `finishRun` also completes the old deferred - // after interruption cleanup, so the old caller cannot return early. + // Interrupt the old run on an independent fiber because its `finishRun` + // acquires this same ref. The replacement waits on a gate so its session + // writes cannot overlap the old run's cleanup, while cancelling the + // replacement cannot prevent the old caller from being completed. const old = st.run + const stopped = yield* Deferred.make() const done = yield* Deferred.make() - const run = yield* startRun(work, done) - yield* Effect.suspend(() => Fiber.interrupt(old.fiber)).pipe(Effect.forkIn(scope)) + const run = yield* startRun( + Effect.uninterruptibleMask((restore) => + Deferred.await(stopped).pipe(Effect.andThen(restore(Effect.yieldNow.pipe(Effect.andThen(work))))), + ), + done, + ) + yield* Fiber.interrupt(old.fiber).pipe( + Effect.ensuring(Deferred.succeed(stopped, undefined)), + Effect.forkIn(scope), + ) return [awaitDone(done), { _tag: "Running", run }] as const } case "ShellThenRun": diff --git a/packages/opencode/test/effect/runner.test.ts b/packages/opencode/test/effect/runner.test.ts index aa0dc4c4ade0..2562cc9c8e2b 100644 --- a/packages/opencode/test/effect/runner.test.ts +++ b/packages/opencode/test/effect/runner.test.ts @@ -128,6 +128,7 @@ describe("Runner", () => { const started = yield* Deferred.make() const cleanupStarted = yield* Deferred.make() const releaseCleanup = yield* Deferred.make() + const secondStarted = yield* Deferred.make() const releaseSecond = yield* Deferred.make() const runner = Runner.make(s, { onInterrupt: Effect.succeed("interrupted") }) @@ -144,7 +145,10 @@ describe("Runner", () => { yield* Deferred.await(started) const observer = yield* Effect.gen(function* () { yield* Deferred.await(cleanupStarted) - return a.pollUnsafe() + return { + old: a.pollUnsafe(), + replacementStarted: yield* Deferred.isDone(secondStarted), + } }).pipe( Effect.ensuring( Effect.all([Deferred.succeed(releaseCleanup, undefined), Deferred.succeed(releaseSecond, undefined)], { @@ -154,7 +158,12 @@ describe("Runner", () => { Effect.forkChild, ) const b = yield* runner - .ensureRunning(Deferred.await(releaseSecond).pipe(Effect.as("second-result"))) + .ensureRunning( + Deferred.succeed(secondStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseSecond)), + Effect.as("second-result"), + ), + ) .pipe(Effect.forkChild) const early = yield* Fiber.join(observer).pipe(Effect.timeout("250 millis")) @@ -164,7 +173,8 @@ describe("Runner", () => { if (Exit.isSuccess(exitA)) expect(exitA.value).toBe("interrupted") expect(Exit.isSuccess(exitB)).toBe(true) if (Exit.isSuccess(exitB)) expect(exitB.value).toBe("second-result") - expect(early).toBeUndefined() + expect(early.old).toBeUndefined() + expect(early.replacementStarted).toBe(false) }), ) @@ -208,6 +218,55 @@ describe("Runner", () => { }), ) + it.live( + "successive replacements wait for the original run cleanup", + Effect.gen(function* () { + const s = yield* Scope.Scope + const started = yield* Deferred.make() + const cleanupStarted = yield* Deferred.make() + const releaseCleanup = yield* Deferred.make() + const secondStarted = yield* Deferred.make() + const thirdStarted = yield* Deferred.make() + const releaseThird = yield* Deferred.make() + const runner = Runner.make(s, { onInterrupt: Effect.succeed("interrupted") }) + + const first = Effect.gen(function* () { + yield* Deferred.succeed(started, undefined) + return yield* Effect.never.pipe( + Effect.onInterrupt(() => Deferred.succeed(cleanupStarted, undefined)), + Effect.ensuring(Deferred.await(releaseCleanup)), + Effect.as("first"), + ) + }) + + const a = yield* runner.ensureRunning(first).pipe(Effect.forkChild) + yield* Deferred.await(started) + const b = yield* runner + .ensureRunning(Deferred.succeed(secondStarted, undefined).pipe(Effect.as("second"))) + .pipe(Effect.forkChild) + yield* Deferred.await(cleanupStarted) + const c = yield* runner + .ensureRunning( + Deferred.succeed(thirdStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseThird)), + Effect.as("third"), + ), + ) + .pipe(Effect.forkChild) + + yield* Effect.yieldNow + expect(yield* Deferred.isDone(secondStarted)).toBe(false) + expect(yield* Deferred.isDone(thirdStarted)).toBe(false) + + yield* Deferred.succeed(releaseCleanup, undefined) + expect(yield* Fiber.join(a)).toBe("interrupted") + expect(yield* Fiber.join(b)).toBe("interrupted") + yield* Deferred.await(thirdStarted).pipe(Effect.timeout("250 millis")) + yield* Deferred.succeed(releaseThird, undefined) + expect(yield* Fiber.join(c)).toBe("third") + }), + ) + // --- cancel semantics --- it.live( @@ -275,7 +334,8 @@ describe("Runner", () => { yield* runner.cancel - const [exitA, exitB] = yield* Effect.all([Fiber.await(a), Fiber.await(b)]) + const exitA = yield* Fiber.await(a).pipe(Effect.timeout("250 millis")) + const exitB = yield* Fiber.await(b).pipe(Effect.timeout("250 millis")) expect(Exit.isSuccess(exitA)).toBe(true) expect(Exit.isSuccess(exitB)).toBe(true) if (Exit.isSuccess(exitA)) expect(exitA.value).toBe("fallback")