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* () {