openAgent

1import { describe, expect } from "bun:test"
2import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect"
3import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator"
4import { testEffect } from "./lib/effect"
5 
6const it = testEffect(Layer.empty)
7 
8describe("SessionRunCoordinator", () => {
9  it.effect("joins concurrent resumes for one key", () =>
10    Effect.scoped(
11      Effect.gen(function* () {
12        const gate = yield* Deferred.make<void>()
13        let runs = 0
14        const coordinator = yield* SessionRunCoordinator.make({
15          drain: () => Effect.sync(() => runs++).pipe(Effect.andThen(Deferred.await(gate))),
16        })
17 
18        const first = yield* coordinator.run("session").pipe(Effect.forkChild)
19        yield* Effect.yieldNow
20        const second = yield* coordinator.run("session").pipe(Effect.forkChild)
21        yield* Effect.yieldNow
22 
23        expect(runs).toBe(1)
24        yield* Deferred.succeed(gate, undefined)
25        yield* Effect.all([Fiber.join(first), Fiber.join(second)])
26        expect(runs).toBe(1)
27      }),
28    ),
29  )
30 
31  it.effect("joins a wake-started execution without forcing a successor", () =>
32    Effect.scoped(
33      Effect.gen(function* () {
34        const started = yield* Deferred.make<void>()
35        const gate = yield* Deferred.make<void>()
36        const forces: boolean[] = []
37        const coordinator = yield* SessionRunCoordinator.make<string, never>({
38          drain: (_key, force) =>
39            Effect.sync(() => forces.push(force)).pipe(
40              Effect.andThen(Deferred.succeed(started, undefined)),
41              Effect.andThen(Deferred.await(gate)),
42            ),
43        })
44 
45        yield* coordinator.wake("session")
46        yield* Deferred.await(started)
47        const resumed = yield* coordinator.run("session").pipe(Effect.forkChild)
48        yield* Effect.yieldNow
49        yield* Deferred.succeed(gate, undefined)
50        yield* Fiber.join(resumed)
51 
52        expect(forces).toEqual([false])
53      }),
54    ),
55  )
56 
57  it.effect("starts execution when woken while idle", () =>
58    Effect.scoped(
59      Effect.gen(function* () {
60        const drained = yield* Deferred.make<void>()
61        const coordinator = yield* SessionRunCoordinator.make({ drain: () => Deferred.succeed(drained, undefined) })
62 
63        yield* coordinator.wake("session")
64        yield* Deferred.await(drained)
65      }),
66    ),
67  )
68 
69  it.effect("snapshots only active executions", () =>
70    Effect.scoped(
71      Effect.gen(function* () {
72        const firstStarted = yield* Deferred.make<void>()
73        const secondStarted = yield* Deferred.make<void>()
74        const firstGate = yield* Deferred.make<void>()
75        const secondGate = yield* Deferred.make<void>()
76        const coordinator = yield* SessionRunCoordinator.make({
77          drain: (key: string) =>
78            Deferred.succeed(key === "first" ? firstStarted : secondStarted, undefined).pipe(
79              Effect.andThen(Deferred.await(key === "first" ? firstGate : secondGate)),
80            ),
81        })
82 
83        expect(Array.from(yield* coordinator.active)).toEqual([])
84        const first = yield* coordinator.run("first").pipe(Effect.forkChild)
85        yield* Deferred.await(firstStarted)
86        expect(Array.from(yield* coordinator.active)).toEqual(["first"])
87 
88        const second = yield* coordinator.run("second").pipe(Effect.forkChild)
89        yield* Deferred.await(secondStarted)
90        expect(Array.from(yield* coordinator.active)).toEqual(["first", "second"])
91 
92        yield* Deferred.succeed(firstGate, undefined)
93        yield* Fiber.join(first)
94        expect(Array.from(yield* coordinator.active)).toEqual(["second"])
95        yield* Deferred.succeed(secondGate, undefined)
96        yield* Fiber.join(second)
97        expect(Array.from(yield* coordinator.active)).toEqual([])
98      }),
99    ),
100  )
101 
102  it.effect("cleans active executions after failure and defect", () =>
103    Effect.scoped(
104      Effect.gen(function* () {
105        const failure = new Error("failed")
106        const defect = new Error("defect")
107        const coordinator = yield* SessionRunCoordinator.make({
108          drain: (key: string) => (key === "failure" ? Effect.fail(failure) : Effect.die(defect)),
109        })
110 
111        const failed = yield* coordinator.run("failure").pipe(Effect.exit)
112        expect(Exit.isFailure(failed) && Cause.hasFails(failed.cause)).toBeTrue()
113        expect(Array.from(yield* coordinator.active)).toEqual([])
114 
115        const died = yield* coordinator.run("defect").pipe(Effect.exit)
116        expect(Exit.isFailure(died) && Cause.hasDies(died.cause)).toBeTrue()
117        expect(Array.from(yield* coordinator.active)).toEqual([])
118      }),
119    ),
120  )
121 
122  it.effect("cleans active executions when its scope closes", () =>
123    Effect.gen(function* () {
124      const started = yield* Deferred.make<void>()
125      const coordinator = yield* Effect.scoped(
126        Effect.gen(function* () {
127          const coordinator = yield* SessionRunCoordinator.make({
128            drain: () => Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)),
129          })
130          yield* coordinator.wake("session")
131          yield* Deferred.await(started)
132          expect(Array.from(yield* coordinator.active)).toEqual(["session"])
133          return coordinator
134        }),
135      )
136 
137      expect(Array.from(yield* coordinator.active)).toEqual([])
138    }),
139  )
140 
141  it.effect("coalesces wakes received during active execution", () =>
142    Effect.scoped(
143      Effect.gen(function* () {
144        const firstStarted = yield* Deferred.make<void>()
145        const firstGate = yield* Deferred.make<void>()
146        const secondStarted = yield* Deferred.make<void>()
147        let runs = 0
148        const coordinator = yield* SessionRunCoordinator.make({
149          drain: () =>
150            Effect.sync(() => ++runs).pipe(
151              Effect.flatMap((run) =>
152                run === 1
153                  ? Deferred.succeed(firstStarted, undefined).pipe(Effect.andThen(Deferred.await(firstGate)))
154                  : Deferred.succeed(secondStarted, undefined),
155              ),
156            ),
157        })
158 
159        const resumed = yield* coordinator.run("session").pipe(Effect.forkChild)
160        yield* Deferred.await(firstStarted)
161        yield* Effect.all([coordinator.wake("session"), coordinator.wake("session"), coordinator.wake("session")], {
162          concurrency: "unbounded",
163        })
164        yield* Deferred.succeed(firstGate, undefined)
165        yield* Deferred.await(secondStarted)
166        yield* Fiber.join(resumed)
167 
168        expect(runs).toBe(2)
169      }),
170    ),
171  )
172 
173  it.effect("runs again when woken during the follow-up", () =>
174    Effect.scoped(
175      Effect.gen(function* () {
176        const firstGate = yield* Deferred.make<void>()
177        const secondStarted = yield* Deferred.make<void>()
178        const secondGate = yield* Deferred.make<void>()
179        const thirdStarted = yield* Deferred.make<void>()
180        let runs = 0
181        const coordinator = yield* SessionRunCoordinator.make({
182          drain: () =>
183            Effect.sync(() => ++runs).pipe(
184              Effect.flatMap((run) =>
185                run === 1
186                  ? Deferred.await(firstGate)
187                  : run === 2
188                    ? Deferred.succeed(secondStarted, undefined).pipe(Effect.andThen(Deferred.await(secondGate)))
189                    : Deferred.succeed(thirdStarted, undefined),
190              ),
191            ),
192        })
193 
194        const resumed = yield* coordinator.run("session").pipe(Effect.forkChild)
195        yield* Effect.yieldNow
196        yield* coordinator.wake("session")
197        yield* Deferred.succeed(firstGate, undefined)
198        yield* Deferred.await(secondStarted)
199        yield* coordinator.wake("session")
200        yield* Deferred.succeed(secondGate, undefined)
201        yield* Deferred.await(thirdStarted)
202        yield* Fiber.join(resumed)
203 
204        expect(runs).toBe(3)
205      }),
206    ),
207  )
208 
209  it.effect("does nothing when interrupted while idle", () =>
210    Effect.scoped(
211      Effect.gen(function* () {
212        const coordinator = yield* SessionRunCoordinator.make({ drain: () => Effect.void })
213        yield* coordinator.interrupt("session")
214      }),
215    ),
216  )
217 
218  it.effect("interrupts active execution and clears its pending wake", () =>
219    Effect.scoped(
220      Effect.gen(function* () {
221        const started = yield* Deferred.make<void>()
222        const interrupted = yield* Deferred.make<void>()
223        let runs = 0
224        const coordinator = yield* SessionRunCoordinator.make({
225          drain: () =>
226            Effect.sync(() => ++runs).pipe(
227              Effect.andThen(Deferred.succeed(started, undefined)),
228              Effect.andThen(Effect.never),
229              Effect.onInterrupt(() => Deferred.succeed(interrupted, undefined)),
230            ),
231        })
232 
233        const resumed = yield* coordinator.run("session").pipe(Effect.forkChild)
234        yield* Deferred.await(started)
235        yield* coordinator.wake("session")
236        yield* coordinator.interrupt("session")
237        yield* Deferred.await(interrupted)
238 
239        const exit = yield* Fiber.await(resumed)
240        expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBeTrue()
241        expect(Array.from(yield* coordinator.active)).toEqual([])
242        expect(runs).toBe(1)
243      }),
244    ),
245  )
246 
247  it.effect("runs a wake registered during interruption cleanup", () =>
248    Effect.scoped(
249      Effect.gen(function* () {
250        const firstStarted = yield* Deferred.make<void>()
251        const cleanupStarted = yield* Deferred.make<void>()
252        const cleanupGate = yield* Deferred.make<void>()
253        const secondStarted = yield* Deferred.make<void>()
254        let runs = 0
255        const coordinator = yield* SessionRunCoordinator.make({
256          drain: () =>
257            Effect.sync(() => ++runs).pipe(
258              Effect.flatMap((run) =>
259                run === 1
260                  ? Deferred.succeed(firstStarted, undefined).pipe(
261                      Effect.andThen(Effect.never),
262                      Effect.onInterrupt(() =>
263                        Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
264                      ),
265                    )
266                  : Deferred.succeed(secondStarted, undefined),
267              ),
268            ),
269        })
270 
271        yield* coordinator.wake("session")
272        yield* Deferred.await(firstStarted)
273        const interrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
274        yield* Deferred.await(cleanupStarted)
275        yield* coordinator.wake("session")
276        yield* Deferred.succeed(cleanupGate, undefined)
277        yield* Fiber.join(interrupt)
278        yield* Deferred.await(secondStarted)
279 
280        expect(runs).toBe(2)
281      }),
282    ),
283  )
284 
285  it.effect("starts a resume registered during interruption cleanup", () =>
286    Effect.scoped(
287      Effect.gen(function* () {
288        const firstStarted = yield* Deferred.make<void>()
289        const cleanupStarted = yield* Deferred.make<void>()
290        const cleanupGate = yield* Deferred.make<void>()
291        const secondStarted = yield* Deferred.make<void>()
292        const forces: boolean[] = []
293        const coordinator = yield* SessionRunCoordinator.make<string, never>({
294          drain: (_key, force) => {
295            forces.push(force)
296            return forces.length === 1
297              ? Deferred.succeed(firstStarted, undefined).pipe(
298                  Effect.andThen(Effect.never),
299                  Effect.onInterrupt(() =>
300                    Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
301                  ),
302                )
303              : Deferred.succeed(secondStarted, undefined)
304          },
305        })
306 
307        yield* coordinator.wake("session")
308        yield* Deferred.await(firstStarted)
309        const interrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
310        yield* Deferred.await(cleanupStarted)
311        const resumed = yield* coordinator.run("session").pipe(Effect.forkChild)
312        yield* Deferred.succeed(cleanupGate, undefined)
313        yield* Effect.all([Fiber.join(interrupt), Fiber.join(resumed)])
314        yield* Deferred.await(secondStarted)
315 
316        expect(forces).toEqual([false, true])
317      }),
318    ),
319  )
320 
321  it.effect("starts one follow-up when a wake races with failure", () =>
322    Effect.scoped(
323      Effect.gen(function* () {
324        const gate = yield* Deferred.make<void>()
325        const secondStarted = yield* Deferred.make<void>()
326        const failure = new Error("failed")
327        let runs = 0
328        const coordinator = yield* SessionRunCoordinator.make({
329          drain: () =>
330            Effect.sync(() => ++runs).pipe(
331              Effect.flatMap((run) =>
332                run === 1
333                  ? Deferred.await(gate).pipe(Effect.andThen(Effect.fail(failure)))
334                  : Deferred.succeed(secondStarted, undefined),
335              ),
336            ),
337        })
338 
339        const resumed = yield* coordinator.run("session").pipe(Effect.forkChild)
340        yield* Effect.yieldNow
341        yield* coordinator.wake("session")
342        yield* Deferred.succeed(gate, undefined)
343 
344        expect(yield* Fiber.join(resumed).pipe(Effect.flip)).toBe(failure)
345        yield* Deferred.await(secondStarted)
346        expect(runs).toBe(2)
347      }),
348    ),
349  )
350 
351  it.effect("does not cancel execution when a joined waiter is interrupted", () =>
352    Effect.scoped(
353      Effect.gen(function* () {
354        const gate = yield* Deferred.make<void>()
355        let runs = 0
356        const coordinator = yield* SessionRunCoordinator.make({
357          drain: () => Effect.sync(() => runs++).pipe(Effect.andThen(Deferred.await(gate))),
358        })
359 
360        const first = yield* coordinator.run("session").pipe(Effect.forkChild)
361        yield* Effect.yieldNow
362        const second = yield* coordinator.run("session").pipe(Effect.forkChild)
363        yield* Fiber.interrupt(second)
364        yield* Deferred.succeed(gate, undefined)
365        yield* Fiber.join(first)
366 
367        expect(runs).toBe(1)
368      }),
369    ),
370  )
371 
372  it.effect("runs different keys concurrently", () =>
373    Effect.scoped(
374      Effect.gen(function* () {
375        const gate = yield* Deferred.make<void>()
376        const bothStarted = yield* Deferred.make<void>()
377        let active = 0
378        const coordinator = yield* SessionRunCoordinator.make({
379          drain: () =>
380            Effect.sync(() => ++active).pipe(
381              Effect.tap(() => (active === 2 ? Deferred.succeed(bothStarted, undefined) : Effect.void)),
382              Effect.andThen(Deferred.await(gate)),
383            ),
384        })
385 
386        const first = yield* coordinator.run("first").pipe(Effect.forkChild)
387        const second = yield* coordinator.run("second").pipe(Effect.forkChild)
388        yield* Deferred.await(bothStarted)
389        yield* Deferred.succeed(gate, undefined)
390        yield* Effect.all([Fiber.join(first), Fiber.join(second)])
391      }),
392    ),
393  )
394 
395  it.effect("trampolines synchronous self-waking execution", () =>
396    Effect.scoped(
397      Effect.gen(function* () {
398        const limit = 20_000
399        const completed = yield* Deferred.make<void>()
400        let runs = 0
401        let wake: (key: string) => Effect.Effect<void> = () => Effect.void
402        const coordinator = yield* SessionRunCoordinator.make<string, never>({
403          drain: (key) =>
404            Effect.sync(() => ++runs).pipe(
405              Effect.tap((run) => (run < limit ? wake(key) : Deferred.succeed(completed, undefined))),
406              Effect.asVoid,
407            ),
408        })
409        wake = coordinator.wake
410 
411        yield* coordinator.wake("session")
412        yield* Deferred.await(completed)
413 
414        expect(runs).toBe(limit)
415      }),
416    ),
417  )
418})
419