openAgent

1import { describe, expect } from "bun:test"
2import { LayerNode } from "@opencode-ai/core/effect/layer-node"
3import { Deferred, Effect } from "effect"
4import { BackgroundJob } from "@/background/job"
5import { testEffect } from "../lib/effect"
6 
7const it = testEffect(LayerNode.compile(BackgroundJob.node))
8 
9describe("background.job", () => {
10  it.instance("tracks started jobs through completion", () =>
11    Effect.gen(function* () {
12      const jobs = yield* BackgroundJob.Service
13      const latch = yield* Deferred.make<void>()
14      const job = yield* jobs.start({
15        type: "test",
16        title: "test job",
17        run: Deferred.await(latch).pipe(Effect.as("done")),
18      })
19 
20      expect(job.id.startsWith("job_")).toBe(true)
21      expect(job.status).toBe("running")
22      expect(job.title).toBe("test job")
23 
24      yield* Deferred.succeed(latch, undefined)
25      const done = yield* jobs.wait({ id: job.id })
26 
27      expect(done.timedOut).toBe(false)
28      expect(done.info?.status).toBe("completed")
29      expect(done.info?.output).toBe("done")
30      expect((yield* jobs.list()).map((item) => item.id)).toEqual([job.id])
31    }),
32  )
33 
34  it.instance("returns a running snapshot when wait times out", () =>
35    Effect.gen(function* () {
36      const jobs = yield* BackgroundJob.Service
37      const job = yield* jobs.start({
38        type: "test",
39        run: Effect.never,
40      })
41 
42      const result = yield* jobs.wait({ id: job.id, timeout: 1 })
43 
44      expect(result.timedOut).toBe(true)
45      expect(result.info?.status).toBe("running")
46    }),
47  )
48 
49  it.instance("deduplicates concurrent starts for a running id", () =>
50    Effect.gen(function* () {
51      const jobs = yield* BackgroundJob.Service
52      const started = yield* Deferred.make<void>()
53      const id = "job_test"
54      const [first, second] = yield* Effect.all(
55        [
56          jobs.start({
57            id,
58            type: "test",
59            run: Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)),
60          }),
61          jobs.start({
62            id,
63            type: "test",
64            run: Effect.fail(new Error("duplicate started")),
65          }),
66        ],
67        { concurrency: "unbounded" },
68      )
69 
70      yield* Deferred.await(started)
71 
72      expect(first.id).toBe(id)
73      expect(second.id).toBe(id)
74      expect(first.status).toBe("running")
75      expect(second.status).toBe("running")
76      expect((yield* jobs.list()).map((item) => item.id)).toEqual([id])
77 
78      yield* jobs.cancel(id)
79    }),
80  )
81 
82  it.instance("waits for extensions before completing a running job", () =>
83    Effect.gen(function* () {
84      const jobs = yield* BackgroundJob.Service
85      const first = yield* Deferred.make<void>()
86      const second = yield* Deferred.make<void>()
87      const job = yield* jobs.start({
88        type: "test",
89        run: Deferred.await(first).pipe(Effect.as("first")),
90      })
91 
92      expect(yield* jobs.extend({ id: job.id, run: Deferred.await(second).pipe(Effect.as("second")) })).toBe(true)
93      yield* Deferred.succeed(first, undefined)
94      expect((yield* jobs.get(job.id))?.status).toBe("running")
95 
96      yield* Deferred.succeed(second, undefined)
97      const done = yield* jobs.wait({ id: job.id })
98      expect(done.info?.status).toBe("completed")
99      expect(done.info?.output).toBe("second")
100    }),
101  )
102 
103  it.instance("runs extensions after earlier work completes", () =>
104    Effect.gen(function* () {
105      const jobs = yield* BackgroundJob.Service
106      const first = yield* Deferred.make<void>()
107      const order: string[] = []
108      const job = yield* jobs.start({
109        type: "test",
110        run: Effect.sync(() => order.push("start")).pipe(Effect.andThen(Deferred.await(first)), Effect.as("first")),
111      })
112 
113      expect(
114        yield* jobs.extend({
115          id: job.id,
116          run: Effect.sync(() => order.push("extend")).pipe(Effect.as("second")),
117        }),
118      ).toBe(true)
119      yield* Effect.yieldNow
120      expect(order).toEqual(["start"])
121 
122      yield* Deferred.succeed(first, undefined)
123      expect((yield* jobs.wait({ id: job.id })).info?.output).toBe("second")
124      expect(order).toEqual(["start", "extend"])
125    }),
126  )
127 
128  it.instance("rejects extensions after a job completes", () =>
129    Effect.gen(function* () {
130      const jobs = yield* BackgroundJob.Service
131      const job = yield* jobs.start({ type: "test", run: Effect.succeed("done") })
132      yield* jobs.wait({ id: job.id })
133 
134      expect(yield* jobs.extend({ id: job.id, run: Effect.succeed("late") })).toBe(false)
135      expect((yield* jobs.get(job.id))?.output).toBe("done")
136    }),
137  )
138 
139  it.instance("records failed jobs", () =>
140    Effect.gen(function* () {
141      const jobs = yield* BackgroundJob.Service
142      const job = yield* jobs.start({
143        type: "test",
144        run: Effect.fail(new Error("boom")),
145      })
146 
147      const result = yield* jobs.wait({ id: job.id })
148 
149      expect(result.info?.status).toBe("error")
150      expect(result.info?.error).toBe("boom")
151    }),
152  )
153 
154  it.instance("ignores stale settlements after restarting a failed job", () =>
155    Effect.gen(function* () {
156      const jobs = yield* BackgroundJob.Service
157      const fail = yield* Deferred.make<void>()
158      const interrupted = yield* Deferred.make<void>()
159      const release = yield* Deferred.make<void>()
160      const id = "job_test"
161      yield* jobs.start({
162        id,
163        type: "test",
164        run: Deferred.await(fail).pipe(Effect.andThen(Effect.fail(new Error("boom")))),
165      })
166      yield* jobs.extend({
167        id,
168        run: Effect.never.pipe(
169          Effect.ensuring(Deferred.succeed(interrupted, undefined).pipe(Effect.andThen(Deferred.await(release)))),
170        ),
171      })
172 
173      yield* Deferred.succeed(fail, undefined)
174      expect((yield* jobs.wait({ id })).info?.status).toBe("error")
175      yield* Deferred.await(interrupted)
176      yield* jobs.start({ id, type: "test", run: Effect.never })
177 
178      yield* Deferred.succeed(release, undefined)
179      yield* Effect.yieldNow
180      expect((yield* jobs.get(id))?.status).toBe("running")
181      yield* jobs.cancel(id)
182    }),
183  )
184 
185  it.instance("can cancel running jobs", () =>
186    Effect.gen(function* () {
187      const jobs = yield* BackgroundJob.Service
188      const interrupted = yield* Deferred.make<void>()
189      const job = yield* jobs.start({
190        type: "test",
191        run: Effect.never.pipe(Effect.ensuring(Deferred.succeed(interrupted, undefined))),
192      })
193      yield* jobs.extend({
194        id: job.id,
195        run: Effect.never,
196      })
197 
198      const cancelled = yield* jobs.cancel(job.id)
199 
200      expect(cancelled?.status).toBe("cancelled")
201      yield* Deferred.await(interrupted).pipe(Effect.timeout("1 second"))
202      expect((yield* jobs.get(job.id))?.status).toBe("cancelled")
203    }),
204  )
205 
206  it.instance("promotes running jobs without interrupting them", () =>
207    Effect.gen(function* () {
208      const jobs = yield* BackgroundJob.Service
209      const latch = yield* Deferred.make<void>()
210      const promoted = yield* Deferred.make<void>()
211      const job = yield* jobs.start({
212        type: "test",
213        metadata: { parentSessionId: "parent" },
214        onPromote: Deferred.succeed(promoted, undefined).pipe(Effect.asVoid),
215        run: Deferred.await(latch).pipe(Effect.as("done")),
216      })
217 
218      const info = yield* jobs.promote(job.id)
219 
220      expect(info?.status).toBe("running")
221      expect(info?.metadata?.background).toBe(true)
222      yield* Deferred.await(promoted)
223      expect((yield* jobs.get(job.id))?.status).toBe("running")
224 
225      yield* Deferred.succeed(latch, undefined)
226      expect((yield* jobs.wait({ id: job.id })).info?.output).toBe("done")
227    }),
228  )
229 
230  it.instance("returns immutable snapshots", () =>
231    Effect.gen(function* () {
232      const jobs = yield* BackgroundJob.Service
233      const job = yield* jobs.start({
234        type: "test",
235        metadata: { value: "initial" },
236        run: Effect.succeed("done"),
237      })
238 
239      if (job.metadata) job.metadata.value = "changed"
240 
241      expect((yield* jobs.get(job.id))?.metadata?.value).toBe("initial")
242    }),
243  )
244})
245