openAgent

1import { describe, expect } from "bun:test"
2import { Cause, Deferred, Effect, Exit, Fiber, Latch, Ref, Scope } from "effect"
3import { Runner } from "@/effect/runner"
4import { it } from "../lib/effect"
5 
6const waitForState = <A, E>(runner: Runner.Runner<A, E>, tag: Runner.State<A, E>["_tag"]) =>
7  Effect.gen(function* () {
8    while (runner.state._tag !== tag) yield* Effect.yieldNow
9  }).pipe(Effect.timeout("1 second"))
10 
11describe("Runner", () => {
12  // --- ensureRunning semantics ---
13 
14  it.live(
15    "ensureRunning starts work and returns result",
16    Effect.gen(function* () {
17      const s = yield* Scope.Scope
18      const runner = Runner.make<string>(s)
19      const result = yield* runner.ensureRunning(Effect.succeed("hello"))
20      expect(result).toBe("hello")
21      expect(runner.state._tag).toBe("Idle")
22      expect(runner.busy).toBe(false)
23    }),
24  )
25 
26  it.live(
27    "ensureRunning propagates work failures",
28    Effect.gen(function* () {
29      const s = yield* Scope.Scope
30      const runner = Runner.make<string, string>(s)
31      const exit = yield* runner.ensureRunning(Effect.fail("boom")).pipe(Effect.exit)
32      expect(Exit.isFailure(exit)).toBe(true)
33      expect(runner.state._tag).toBe("Idle")
34    }),
35  )
36 
37  it.live(
38    "concurrent callers share the same run",
39    Effect.gen(function* () {
40      const s = yield* Scope.Scope
41      const runner = Runner.make<string>(s)
42      const calls = yield* Ref.make(0)
43      const work = Effect.gen(function* () {
44        yield* Ref.update(calls, (n) => n + 1)
45        yield* Effect.sleep("10 millis")
46        return "shared"
47      })
48 
49      const [a, b] = yield* Effect.all([runner.ensureRunning(work), runner.ensureRunning(work)], {
50        concurrency: "unbounded",
51      })
52 
53      expect(a).toBe("shared")
54      expect(b).toBe("shared")
55      expect(yield* Ref.get(calls)).toBe(1)
56    }),
57  )
58 
59  it.live(
60    "concurrent callers all receive same error",
61    Effect.gen(function* () {
62      const s = yield* Scope.Scope
63      const runner = Runner.make<string, string>(s)
64      const work = Effect.gen(function* () {
65        yield* Effect.sleep("10 millis")
66        return yield* Effect.fail("boom")
67      })
68 
69      const [a, b] = yield* Effect.all(
70        [runner.ensureRunning(work).pipe(Effect.exit), runner.ensureRunning(work).pipe(Effect.exit)],
71        { concurrency: "unbounded" },
72      )
73 
74      expect(Exit.isFailure(a)).toBe(true)
75      expect(Exit.isFailure(b)).toBe(true)
76    }),
77  )
78 
79  it.live(
80    "ensureRunning can be called again after previous run completes",
81    Effect.gen(function* () {
82      const s = yield* Scope.Scope
83      const runner = Runner.make<string>(s)
84      expect(yield* runner.ensureRunning(Effect.succeed("first"))).toBe("first")
85      expect(yield* runner.ensureRunning(Effect.succeed("second"))).toBe("second")
86    }),
87  )
88 
89  it.live(
90    "second ensureRunning ignores new work if already running",
91    Effect.gen(function* () {
92      const s = yield* Scope.Scope
93      const runner = Runner.make<string>(s)
94      const ran = yield* Ref.make<string[]>([])
95 
96      const first = Effect.gen(function* () {
97        yield* Ref.update(ran, (a) => [...a, "first"])
98        yield* Effect.sleep("50 millis")
99        return "first-result"
100      })
101      const second = Effect.gen(function* () {
102        yield* Ref.update(ran, (a) => [...a, "second"])
103        return "second-result"
104      })
105 
106      const [a, b] = yield* Effect.all([runner.ensureRunning(first), runner.ensureRunning(second)], {
107        concurrency: "unbounded",
108      })
109 
110      expect(a).toBe("first-result")
111      expect(b).toBe("first-result")
112      expect(yield* Ref.get(ran)).toEqual(["first"])
113    }),
114  )
115 
116  // --- cancel semantics ---
117 
118  it.live(
119    "cancel interrupts running work",
120    Effect.gen(function* () {
121      const s = yield* Scope.Scope
122      const runner = Runner.make<string>(s)
123      const started = yield* Deferred.make<void>()
124      const fiber = yield* runner
125        .ensureRunning(
126          Effect.gen(function* () {
127            yield* Deferred.succeed(started, void 0)
128            return yield* Effect.never.pipe(Effect.as("never"))
129          }),
130        )
131        .pipe(Effect.forkChild)
132      yield* Deferred.await(started)
133      expect(runner.busy).toBe(true)
134      expect(runner.state._tag).toBe("Running")
135 
136      yield* runner.cancel
137      expect(runner.busy).toBe(false)
138 
139      const exit = yield* Fiber.await(fiber)
140      expect(Exit.isFailure(exit)).toBe(true)
141    }),
142  )
143 
144  it.live(
145    "cancel on idle is a no-op",
146    Effect.gen(function* () {
147      const s = yield* Scope.Scope
148      const runner = Runner.make<string>(s)
149      yield* runner.cancel
150      expect(runner.busy).toBe(false)
151    }),
152  )
153 
154  it.live(
155    "cancel with onInterrupt resolves callers gracefully",
156    Effect.gen(function* () {
157      const s = yield* Scope.Scope
158      const runner = Runner.make<string>(s, { onInterrupt: Effect.succeed("fallback") })
159      const fiber = yield* runner.ensureRunning(Effect.never.pipe(Effect.as("never"))).pipe(Effect.forkChild)
160      yield* waitForState(runner, "Running")
161 
162      yield* runner.cancel
163 
164      const exit = yield* Fiber.await(fiber)
165      expect(Exit.isSuccess(exit)).toBe(true)
166      if (Exit.isSuccess(exit)) expect(exit.value).toBe("fallback")
167    }),
168  )
169 
170  it.live(
171    "cancel with queued callers resolves all",
172    Effect.gen(function* () {
173      const s = yield* Scope.Scope
174      const runner = Runner.make<string>(s, { onInterrupt: Effect.succeed("fallback") })
175 
176      const a = yield* runner.ensureRunning(Effect.never.pipe(Effect.as("x"))).pipe(Effect.forkChild)
177      yield* waitForState(runner, "Running")
178      const b = yield* runner.ensureRunning(Effect.succeed("y")).pipe(Effect.forkChild)
179      yield* Effect.yieldNow
180 
181      yield* runner.cancel
182 
183      const [exitA, exitB] = yield* Effect.all([Fiber.await(a), Fiber.await(b)])
184      expect(Exit.isSuccess(exitA)).toBe(true)
185      expect(Exit.isSuccess(exitB)).toBe(true)
186      if (Exit.isSuccess(exitA)) expect(exitA.value).toBe("fallback")
187      if (Exit.isSuccess(exitB)) expect(exitB.value).toBe("fallback")
188    }),
189  )
190 
191  it.live(
192    "work can be started after cancel",
193    Effect.gen(function* () {
194      const s = yield* Scope.Scope
195      const runner = Runner.make<string>(s)
196      const fiber = yield* runner.ensureRunning(Effect.never.pipe(Effect.as("x"))).pipe(Effect.forkChild)
197      yield* waitForState(runner, "Running")
198      yield* runner.cancel
199      yield* Fiber.await(fiber)
200 
201      const result = yield* runner.ensureRunning(Effect.succeed("after-cancel"))
202      expect(result).toBe("after-cancel")
203    }),
204  )
205 
206  it.live(
207    "cancel does not deadlock when replacement work starts before interrupted run exits",
208    Effect.gen(function* () {
209      const s = yield* Scope.Scope
210      const hit = yield* Deferred.make<void>()
211      const hold = yield* Deferred.make<void>()
212      const done = yield* Deferred.make<void>()
213 
214      yield* Effect.gen(function* () {
215        const runner = Runner.make<string>(s)
216        const first = Effect.never.pipe(
217          Effect.onInterrupt(() => Deferred.succeed(hit, undefined)),
218          Effect.ensuring(Deferred.await(hold)),
219          Effect.as("first"),
220        )
221 
222        const a = yield* runner.ensureRunning(first).pipe(Effect.exit, Effect.forkChild)
223        yield* waitForState(runner, "Running")
224 
225        const stop = yield* runner.cancel.pipe(Effect.forkChild)
226        yield* Deferred.await(hit).pipe(Effect.timeout("250 millis"))
227 
228        const b = yield* runner.ensureRunning(Deferred.await(done).pipe(Effect.as("second"))).pipe(Effect.forkChild)
229        yield* Effect.yieldNow
230        expect(runner.busy).toBe(true)
231 
232        yield* Deferred.succeed(hold, undefined)
233        const stopExit = yield* Fiber.await(stop).pipe(Effect.timeout("250 millis"))
234        expect(Exit.isSuccess(stopExit)).toBe(true)
235 
236        expect(runner.busy).toBe(true)
237        yield* Deferred.succeed(done, undefined)
238        expect(yield* Fiber.join(b).pipe(Effect.timeout("250 millis"))).toBe("second")
239        expect(runner.busy).toBe(false)
240 
241        const exit = yield* Fiber.join(a)
242        expect(Exit.isFailure(exit)).toBe(true)
243      }).pipe(
244        Effect.ensuring(
245          Effect.all([Deferred.succeed(hold, undefined), Deferred.succeed(done, undefined)], { discard: true }).pipe(
246            Effect.ignore,
247          ),
248        ),
249      )
250    }),
251  )
252 
253  // --- shell semantics ---
254 
255  it.live(
256    "shell runs exclusively",
257    Effect.gen(function* () {
258      const s = yield* Scope.Scope
259      const runner = Runner.make<string>(s)
260      const result = yield* runner.startShell(Effect.succeed("shell-done"))
261      expect(result).toBe("shell-done")
262      expect(runner.busy).toBe(false)
263    }),
264  )
265 
266  it.live(
267    "shell rejects when run is active",
268    Effect.gen(function* () {
269      const s = yield* Scope.Scope
270      const runner = Runner.make<string>(s)
271      const started = yield* Deferred.make<void>()
272      const fiber = yield* runner
273        .ensureRunning(
274          Effect.gen(function* () {
275            yield* Deferred.succeed(started, undefined)
276            return yield* Effect.never.pipe(Effect.as("x"))
277          }),
278        )
279        .pipe(Effect.forkChild)
280      yield* Deferred.await(started).pipe(Effect.timeout("250 millis"))
281      yield* Effect.gen(function* () {
282        while (runner.state._tag !== "Running") yield* Effect.yieldNow
283      }).pipe(Effect.timeout("250 millis"))
284 
285      const exit = yield* runner.startShell(Effect.succeed("nope")).pipe(Effect.exit)
286      expect(Exit.isFailure(exit)).toBe(true)
287 
288      yield* runner.cancel
289      yield* Fiber.await(fiber).pipe(Effect.timeout("250 millis"))
290    }),
291  )
292 
293  it.live(
294    "shell rejects when another shell is running",
295    Effect.gen(function* () {
296      const s = yield* Scope.Scope
297      const runner = Runner.make<string>(s)
298      const gate = yield* Deferred.make<void>()
299 
300      const sh = yield* runner.startShell(Deferred.await(gate).pipe(Effect.as("first"))).pipe(Effect.forkChild)
301      yield* waitForState(runner, "Shell")
302 
303      const exit = yield* runner.startShell(Effect.succeed("second")).pipe(Effect.exit)
304      expect(Exit.isFailure(exit)).toBe(true)
305      if (Exit.isFailure(exit)) expect(Cause.squash(exit.cause)).toBeInstanceOf(Runner.Busy)
306 
307      yield* Deferred.succeed(gate, undefined)
308      yield* Fiber.await(sh)
309    }),
310  )
311 
312  it.live(
313    "cancel interrupts shell",
314    Effect.gen(function* () {
315      const s = yield* Scope.Scope
316      const runner = Runner.make<string>(s)
317      const gate = yield* Deferred.make<void>()
318 
319      const sh = yield* runner.startShell(Deferred.await(gate).pipe(Effect.as("ignored"))).pipe(Effect.forkChild)
320      yield* waitForState(runner, "Shell")
321 
322      const stop = yield* runner.cancel.pipe(Effect.forkChild)
323      const stopExit = yield* Fiber.await(stop).pipe(Effect.timeout("250 millis"))
324      expect(Exit.isSuccess(stopExit)).toBe(true)
325      expect(runner.busy).toBe(false)
326 
327      const shellExit = yield* Fiber.await(sh)
328      expect(Exit.isFailure(shellExit)).toBe(true)
329 
330      yield* Deferred.succeed(gate, undefined).pipe(Effect.ignore)
331    }),
332  )
333 
334  it.live(
335    "cancel does not mask shell defects",
336    Effect.gen(function* () {
337      const s = yield* Scope.Scope
338      const runner = Runner.make<string>(s, { onInterrupt: Effect.succeed("interrupted") })
339      const ready = yield* Latch.make()
340 
341      const sh = yield* runner
342        .startShell(
343          Effect.gen(function* () {
344            yield* ready.open
345            return yield* Effect.never.pipe(Effect.as("ignored"))
346          }).pipe(Effect.ensuring(Effect.die("boom"))),
347          ready,
348        )
349        .pipe(Effect.forkChild)
350      yield* ready.await.pipe(Effect.timeout("250 millis"))
351 
352      yield* runner.cancel
353      expect(Exit.isFailure(yield* Fiber.await(sh))).toBe(true)
354    }),
355  )
356 
357  // --- shell→run handoff ---
358 
359  it.live(
360    "ensureRunning queues behind shell then runs after",
361    Effect.gen(function* () {
362      const s = yield* Scope.Scope
363      const runner = Runner.make<string>(s)
364      const gate = yield* Deferred.make<void>()
365 
366      const sh = yield* runner.startShell(Deferred.await(gate).pipe(Effect.as("shell-result"))).pipe(Effect.forkChild)
367      yield* waitForState(runner, "Shell")
368      expect(runner.state._tag).toBe("Shell")
369 
370      const run = yield* runner.ensureRunning(Effect.succeed("run-result")).pipe(Effect.forkChild)
371      yield* waitForState(runner, "ShellThenRun")
372      expect(runner.state._tag).toBe("ShellThenRun")
373 
374      yield* Deferred.succeed(gate, undefined)
375      yield* Fiber.await(sh)
376 
377      const exit = yield* Fiber.await(run)
378      expect(Exit.isSuccess(exit)).toBe(true)
379      if (Exit.isSuccess(exit)) expect(exit.value).toBe("run-result")
380      expect(runner.state._tag).toBe("Idle")
381    }),
382  )
383 
384  it.live(
385    "multiple ensureRunning callers share the queued run behind shell",
386    Effect.gen(function* () {
387      const s = yield* Scope.Scope
388      const runner = Runner.make<string>(s)
389      const calls = yield* Ref.make(0)
390      const gate = yield* Deferred.make<void>()
391 
392      const sh = yield* runner.startShell(Deferred.await(gate).pipe(Effect.as("shell"))).pipe(Effect.forkChild)
393      yield* waitForState(runner, "Shell")
394 
395      const work = Effect.gen(function* () {
396        yield* Ref.update(calls, (n) => n + 1)
397        return "run"
398      })
399      const a = yield* runner.ensureRunning(work).pipe(Effect.forkChild)
400      const b = yield* runner.ensureRunning(work).pipe(Effect.forkChild)
401      yield* waitForState(runner, "ShellThenRun")
402 
403      yield* Deferred.succeed(gate, undefined)
404      yield* Fiber.await(sh)
405 
406      const [exitA, exitB] = yield* Effect.all([Fiber.await(a), Fiber.await(b)])
407      expect(Exit.isSuccess(exitA)).toBe(true)
408      expect(Exit.isSuccess(exitB)).toBe(true)
409      expect(yield* Ref.get(calls)).toBe(1)
410    }),
411  )
412 
413  it.live(
414    "cancel during shell_then_run cancels both",
415    Effect.gen(function* () {
416      const s = yield* Scope.Scope
417      const runner = Runner.make<string>(s)
418 
419      const sh = yield* runner.startShell(Effect.never.pipe(Effect.as("aborted"))).pipe(Effect.forkChild)
420      yield* waitForState(runner, "Shell")
421 
422      const run = yield* runner.ensureRunning(Effect.succeed("y")).pipe(Effect.forkChild)
423      yield* waitForState(runner, "ShellThenRun")
424      expect(runner.state._tag).toBe("ShellThenRun")
425 
426      yield* runner.cancel
427      expect(runner.busy).toBe(false)
428 
429      yield* Fiber.await(sh)
430      const exit = yield* Fiber.await(run)
431      expect(Exit.isFailure(exit)).toBe(true)
432    }),
433  )
434 
435  // --- lifecycle callbacks ---
436 
437  it.live(
438    "onIdle fires when returning to idle from running",
439    Effect.gen(function* () {
440      const s = yield* Scope.Scope
441      const count = yield* Ref.make(0)
442      const runner = Runner.make<string>(s, {
443        onIdle: Ref.update(count, (n) => n + 1),
444      })
445      yield* runner.ensureRunning(Effect.succeed("ok"))
446      expect(yield* Ref.get(count)).toBe(1)
447    }),
448  )
449 
450  it.live(
451    "onIdle fires on cancel",
452    Effect.gen(function* () {
453      const s = yield* Scope.Scope
454      const count = yield* Ref.make(0)
455      const runner = Runner.make<string>(s, {
456        onIdle: Ref.update(count, (n) => n + 1),
457      })
458      const fiber = yield* runner.ensureRunning(Effect.never.pipe(Effect.as("x"))).pipe(Effect.forkChild)
459      yield* waitForState(runner, "Running")
460      yield* runner.cancel
461      yield* Fiber.await(fiber)
462      expect(yield* Ref.get(count)).toBeGreaterThanOrEqual(1)
463    }),
464  )
465 
466  it.live(
467    "onBusy fires when shell starts",
468    Effect.gen(function* () {
469      const s = yield* Scope.Scope
470      const count = yield* Ref.make(0)
471      const runner = Runner.make<string>(s, {
472        onBusy: Ref.update(count, (n) => n + 1),
473      })
474      yield* runner.startShell(Effect.succeed("done"))
475      expect(yield* Ref.get(count)).toBe(1)
476    }),
477  )
478 
479  // --- busy flag ---
480 
481  it.live(
482    "busy is true during run",
483    Effect.gen(function* () {
484      const s = yield* Scope.Scope
485      const runner = Runner.make<string>(s)
486      const gate = yield* Deferred.make<void>()
487 
488      const fiber = yield* runner.ensureRunning(Deferred.await(gate).pipe(Effect.as("ok"))).pipe(Effect.forkChild)
489      yield* waitForState(runner, "Running")
490      expect(runner.busy).toBe(true)
491 
492      yield* Deferred.succeed(gate, undefined)
493      yield* Fiber.await(fiber)
494      expect(runner.busy).toBe(false)
495    }),
496  )
497 
498  it.live(
499    "busy is true during shell",
500    Effect.gen(function* () {
501      const s = yield* Scope.Scope
502      const runner = Runner.make<string>(s)
503      const gate = yield* Deferred.make<void>()
504 
505      const fiber = yield* runner.startShell(Deferred.await(gate).pipe(Effect.as("ok"))).pipe(Effect.forkChild)
506      yield* waitForState(runner, "Shell")
507      expect(runner.busy).toBe(true)
508 
509      yield* Deferred.succeed(gate, undefined)
510      yield* Fiber.await(fiber)
511      expect(runner.busy).toBe(false)
512    }),
513  )
514})
515