openAgent

1export * as BackgroundJob from "./background-job"
2 
3import { Cause, Clock, Context, Deferred, Effect, Exit, Layer, Scope, SynchronizedRef } from "effect"
4import { Identifier } from "./id/id"
5import { makeGlobalNode } from "./effect/app-node"
6 
7export type Status = "running" | "completed" | "error" | "cancelled"
8 
9export type Info = {
10  id: string
11  type: string
12  title?: string
13  status: Status
14  started_at: number
15  completed_at?: number
16  output?: string
17  error?: string
18  metadata?: Record<string, unknown>
19}
20 
21type Active = {
22  info: Info
23  done: Deferred.Deferred<Info>
24  scope: Scope.Closeable
25  token: object
26  pending: number
27  next: number
28  output?: { sequence: number; text: string }
29  tail: Deferred.Deferred<void>
30  promoted: Deferred.Deferred<Info>
31  onPromote?: Effect.Effect<void>
32}
33 
34type State = {
35  jobs: SynchronizedRef.SynchronizedRef<Map<string, Active>>
36  scope: Scope.Scope
37}
38 
39type FinishResult = {
40  info?: Info
41  done?: Deferred.Deferred<Info>
42  scope?: Scope.Closeable
43}
44 
45type PromoteResult = {
46  info?: Info
47  promoted?: Deferred.Deferred<Info>
48  onPromote?: Effect.Effect<void>
49}
50 
51type StartResult = { info: Info } | { info: Info; scope: Scope.Closeable; token: object }
52 
53type ExtendResult =
54  | { extended: false }
55  | {
56      extended: true
57      previous: Deferred.Deferred<void>
58      scope: Scope.Closeable
59      tail: Deferred.Deferred<void>
60      token: object
61      sequence: number
62    }
63 
64export type StartInput = {
65  id?: string
66  type: string
67  title?: string
68  metadata?: Record<string, unknown>
69  onPromote?: Effect.Effect<void>
70  run: Effect.Effect<string, unknown>
71}
72 
73export type ExtendInput = {
74  id: string
75  run: Effect.Effect<string, unknown>
76}
77 
78export type WaitInput = {
79  id: string
80  timeout?: number
81}
82 
83export type WaitResult = {
84  info?: Info
85  timedOut: boolean
86}
87 
88export interface Interface {
89  readonly list: () => Effect.Effect<Info[]>
90  readonly get: (id: string) => Effect.Effect<Info | undefined>
91  readonly start: (input: StartInput) => Effect.Effect<Info>
92  readonly extend: (input: ExtendInput) => Effect.Effect<boolean>
93  readonly wait: (input: WaitInput) => Effect.Effect<WaitResult>
94  readonly waitForPromotion: (id: string) => Effect.Effect<Info>
95  readonly promote: (id: string) => Effect.Effect<Info | undefined>
96  readonly cancel: (id: string) => Effect.Effect<Info | undefined>
97}
98 
99export class Service extends Context.Service<Service, Interface>()("@opencode/BackgroundJob") {}
100 
101function snapshot(job: Active): Info {
102  return {
103    ...job.info,
104    ...(job.info.metadata ? { metadata: { ...job.info.metadata } } : {}),
105  }
106}
107 
108function errorText(error: unknown) {
109  if (error instanceof Error) return error.message
110  return String(error)
111}
112 
113/**
114 * Makes one scoped, process-local registry. Entries are intentionally not
115 * durable: process restart or owner-scope closure loses status and interrupts
116 * live work. Persisted observation, restart recovery, and remote workers need a
117 * separate durable ownership slice rather than pretending this registry has
118 * those semantics.
119 */
120export const make = Effect.gen(function* () {
121  const state: State = {
122    jobs: yield* SynchronizedRef.make(new Map()),
123    scope: yield* Scope.Scope,
124  }
125 
126  const settle = Effect.fn("BackgroundJob.settle")(function* (
127    id: string,
128    token: object,
129    sequence: number,
130    exit: Exit.Exit<string, unknown>,
131  ) {
132    const completed_at = yield* Clock.currentTimeMillis
133    const result = yield* SynchronizedRef.modify(state.jobs, (jobs): readonly [FinishResult, Map<string, Active>] => {
134      const job = jobs.get(id)
135      if (!job) return [{}, jobs]
136      if (job.token !== token) return [{}, jobs]
137      if (job.info.status !== "running") return [{ info: snapshot(job) }, jobs]
138      const pending = job.pending - 1
139      const output =
140        Exit.isSuccess(exit) && (!job.output || sequence > job.output.sequence)
141          ? { sequence, text: exit.value }
142          : job.output
143      if (Exit.isSuccess(exit) && pending > 0) {
144        return [{}, new Map(jobs).set(id, { ...job, pending, output })]
145      }
146      const status: Exclude<Status, "running"> = Exit.isSuccess(exit)
147        ? "completed"
148        : Cause.hasInterruptsOnly(exit.cause)
149          ? "cancelled"
150          : "error"
151      const next = {
152        ...job,
153        onPromote: undefined,
154        pending: 0,
155        output,
156        info: {
157          ...job.info,
158          status,
159          completed_at,
160          ...(output ? { output: output.text } : {}),
161          ...(Exit.isFailure(exit) ? { error: errorText(Cause.squash(exit.cause)) } : {}),
162        },
163      }
164      return [{ info: snapshot(next), done: job.done, scope: job.scope }, new Map(jobs).set(id, next)]
165    })
166    if (result.info && result.done) yield* Deferred.succeed(result.done, result.info).pipe(Effect.ignore)
167    if (result.scope) {
168      yield* Scope.close(result.scope, Exit.void).pipe(Effect.forkIn(state.scope, { startImmediately: true }))
169    }
170    return result.info
171  })
172 
173  const fork = Effect.fn("BackgroundJob.fork")(function* (
174    scope: Scope.Scope,
175    id: string,
176    token: object,
177    sequence: number,
178    run: Effect.Effect<string, unknown>,
179  ) {
180    return yield* run.pipe(
181      Effect.matchCauseEffect({
182        onSuccess: (output) => settle(id, token, sequence, Exit.succeed(output)),
183        onFailure: (cause) => settle(id, token, sequence, Exit.failCause(cause)),
184      }),
185      Effect.asVoid,
186      Effect.forkIn(scope, { startImmediately: true }),
187    )
188  })
189 
190  const list: Interface["list"] = Effect.fn("BackgroundJob.list")(function* () {
191    return Array.from((yield* SynchronizedRef.get(state.jobs)).values())
192      .map(snapshot)
193      .toSorted((a, b) => a.started_at - b.started_at)
194  })
195 
196  const get: Interface["get"] = Effect.fn("BackgroundJob.get")(function* (id) {
197    const job = (yield* SynchronizedRef.get(state.jobs)).get(id)
198    if (!job) return
199    return snapshot(job)
200  })
201 
202  const start: Interface["start"] = Effect.fn("BackgroundJob.start")(function* (input) {
203    return yield* Effect.uninterruptibleMask((restore) =>
204      Effect.gen(function* () {
205        const id = input.id ?? Identifier.ascending("job")
206        const started_at = yield* Clock.currentTimeMillis
207        const done = yield* Deferred.make<Info>()
208        const promoted = yield* Deferred.make<Info>()
209        const tail = yield* Deferred.make<void>()
210        const result = yield* SynchronizedRef.modifyEffect(
211          state.jobs,
212          Effect.fnUntraced(function* (jobs) {
213            const existing = jobs.get(id)
214            if (existing?.info.status === "running") {
215              return [{ info: snapshot(existing) }, jobs] as readonly [StartResult, Map<string, Active>]
216            }
217            const scope = yield* Scope.fork(state.scope, "parallel")
218            const token = {}
219            const job = {
220              info: {
221                id,
222                type: input.type,
223                title: input.title,
224                status: "running" as const,
225                started_at,
226                metadata: input.metadata,
227              },
228              done,
229              scope,
230              token,
231              pending: 1,
232              next: 1,
233              tail,
234              promoted,
235              onPromote: input.onPromote,
236            }
237            return [{ info: snapshot(job), scope, token }, new Map(jobs).set(id, job)] as readonly [
238              StartResult,
239              Map<string, Active>,
240            ]
241          }),
242        )
243        if ("scope" in result)
244          yield* fork(
245            result.scope,
246            id,
247            result.token,
248            0,
249            restore(input.run).pipe(Effect.ensuring(Deferred.succeed(tail, undefined))),
250          )
251        return result.info
252      }),
253    )
254  })
255 
256  const extend: Interface["extend"] = Effect.fn("BackgroundJob.extend")(function* (input) {
257    return yield* Effect.uninterruptibleMask((restore) =>
258      Effect.gen(function* () {
259        const tail = yield* Deferred.make<void>()
260        const result = yield* SynchronizedRef.modify(
261          state.jobs,
262          (jobs): readonly [ExtendResult, Map<string, Active>] => {
263            const job = jobs.get(input.id)
264            if (!job || job.info.status !== "running") return [{ extended: false }, jobs]
265            return [
266              { extended: true, previous: job.tail, scope: job.scope, tail, token: job.token, sequence: job.next },
267              new Map(jobs).set(input.id, {
268                ...job,
269                pending: job.pending + 1,
270                next: job.next + 1,
271                tail,
272              }),
273            ]
274          },
275        )
276        if (!result.extended) return false
277        yield* fork(
278          result.scope,
279          input.id,
280          result.token,
281          result.sequence,
282          Deferred.await(result.previous).pipe(
283            Effect.andThen(restore(input.run)),
284            Effect.ensuring(Deferred.succeed(result.tail, undefined)),
285          ),
286        )
287        return true
288      }),
289    )
290  })
291 
292  const wait: Interface["wait"] = Effect.fn("BackgroundJob.wait")(function* (input) {
293    const job = (yield* SynchronizedRef.get(state.jobs)).get(input.id)
294    if (!job) return { timedOut: false }
295    if (job.info.status !== "running") return { info: snapshot(job), timedOut: false }
296    if (input.timeout === undefined) return { info: yield* Deferred.await(job.done), timedOut: false }
297    if (input.timeout <= 0) return { info: snapshot(job), timedOut: true }
298    const info = yield* Deferred.await(job.done).pipe(Effect.timeoutOption(input.timeout))
299    if (info._tag === "Some") return { info: info.value, timedOut: false }
300    return { info: snapshot(job), timedOut: true }
301  })
302 
303  const waitForPromotion: Interface["waitForPromotion"] = Effect.fn("BackgroundJob.waitForPromotion")(function* (id) {
304    const job = (yield* SynchronizedRef.get(state.jobs)).get(id)
305    if (!job || job.info.status !== "running") return yield* Effect.never
306    if (job.info.metadata?.background === true) return snapshot(job)
307    return yield* Deferred.await(job.promoted)
308  })
309 
310  const promote: Interface["promote"] = Effect.fn("BackgroundJob.promote")(function* (id) {
311    const result = yield* SynchronizedRef.modifyEffect(
312      state.jobs,
313      Effect.fnUntraced(function* (jobs) {
314        const job = jobs.get(id)
315        if (!job || job.info.status !== "running") return [{}, jobs] as readonly [PromoteResult, Map<string, Active>]
316        if (job.info.metadata?.background === true)
317          return [{ info: snapshot(job) }, jobs] as readonly [PromoteResult, Map<string, Active>]
318        const next = {
319          ...job,
320          onPromote: undefined,
321          info: {
322            ...job.info,
323            metadata: { ...job.info.metadata, background: true },
324          },
325        }
326        return [
327          { info: snapshot(next), onPromote: job.onPromote, promoted: job.promoted },
328          new Map(jobs).set(id, next),
329        ] as readonly [PromoteResult, Map<string, Active>]
330      }),
331    )
332    if (result.info && result.promoted) yield* Deferred.succeed(result.promoted, result.info).pipe(Effect.ignore)
333    if (result.onPromote) yield* result.onPromote.pipe(Effect.ignore)
334    return result.info
335  })
336 
337  const cancel: Interface["cancel"] = Effect.fn("BackgroundJob.cancel")(function* (id) {
338    const completed_at = yield* Clock.currentTimeMillis
339    const result = yield* SynchronizedRef.modify(state.jobs, (jobs): readonly [FinishResult, Map<string, Active>] => {
340      const job = jobs.get(id)
341      if (!job) return [{}, jobs]
342      if (job.info.status !== "running") return [{ info: snapshot(job) }, jobs]
343      const next = {
344        ...job,
345        onPromote: undefined,
346        pending: 0,
347        info: {
348          ...job.info,
349          status: "cancelled" as const,
350          completed_at,
351        },
352      }
353      return [{ info: snapshot(next), done: job.done, scope: job.scope }, new Map(jobs).set(id, next)]
354    })
355    if (result.info && result.done) yield* Deferred.succeed(result.done, result.info).pipe(Effect.ignore)
356    if (result.scope) yield* Scope.close(result.scope, Exit.void)
357    return result.info
358  })
359 
360  return Service.of({ list, get, start, extend, wait, waitForPromotion, promote, cancel })
361})
362 
363const layer = Layer.effect(Service, make)
364 
365export const node = makeGlobalNode({ service: Service, layer, deps: [] })
366