import { describe, expect } from "bun:test" import { Job } from "@opencode-ai/core/job" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { Deferred, Effect, Exit, Fiber, Scope } from "effect" import { SessionSchema } from "@opencode-ai/core/session/schema" import { testEffect } from "./lib/effect" const it = testEffect(AppNodeBuilder.build(Job.node)) describe("Job", () => { it.live("tracks process-local work through explicit observation", () => Effect.gen(function* () { const jobs = yield* Job.Service const latch = yield* Deferred.make() const job = yield* jobs.start({ type: "test", metadata: { durable: false }, run: Deferred.await(latch).pipe(Effect.as("done")), }) expect(job).toMatchObject({ type: "test", status: "running", metadata: { durable: false } }) expect(yield* jobs.wait({ id: job.id, timeout: 0 })).toMatchObject({ timedOut: true, info: { status: "running" }, }) yield* Deferred.succeed(latch, undefined) expect(yield* jobs.wait({ id: job.id })).toMatchObject({ timedOut: false, info: { status: "completed", output: "done" }, }) }), ) it.live("publishes jobs before starting immediately settling work", () => Effect.gen(function* () { const jobs = yield* Job.Service yield* Effect.forEach(Array.from({ length: 100 }), (_, index) => { const id = `job_immediate_start_${index}` return Effect.gen(function* () { const job = yield* jobs.start({ id, type: "test", run: jobs .get(id) .pipe( Effect.flatMap((info) => info?.status === "running" ? Effect.succeed(`done-${index}`) : Effect.fail("job started before publish"), ), ), }) expect(yield* jobs.wait({ id: job.id })).toMatchObject({ timedOut: false, info: { status: "completed", output: `done-${index}` }, }) }) }) }), ) it.live("returns finished from a blocking wait when completion wins", () => Effect.gen(function* () { const jobs = yield* Job.Service const latch = yield* Deferred.make() const job = yield* jobs.start({ type: "test", run: Deferred.await(latch).pipe(Effect.as("done")) }) const waiting = yield* jobs .block({ id: job.id, sessionID: SessionSchema.ID.make("ses_parent") }) .pipe(Effect.forkIn(yield* Scope.Scope, { startImmediately: true })) yield* Deferred.succeed(latch, undefined) expect(yield* Fiber.join(waiting)).toMatchObject({ type: "finished", info: { status: "completed", output: "done" }, }) expect(yield* jobs.background(job.id)).toBeUndefined() }), ) it.live("returns backgrounded from a blocking wait when background wins", () => Effect.gen(function* () { const jobs = yield* Job.Service const latch = yield* Deferred.make() const job = yield* jobs.start({ type: "test", run: Deferred.await(latch).pipe(Effect.as("done")) }) const waiting = yield* jobs .block({ id: job.id, sessionID: SessionSchema.ID.make("ses_parent") }) .pipe(Effect.forkIn(yield* Scope.Scope, { startImmediately: true })) expect(yield* jobs.background(job.id)).toMatchObject({ id: job.id, status: "running" }) expect(yield* Fiber.join(waiting)).toMatchObject({ type: "backgrounded", info: { id: job.id, status: "running" }, }) yield* Deferred.succeed(latch, undefined) expect(yield* jobs.wait({ id: job.id })).toMatchObject({ timedOut: false, info: { status: "completed", output: "done" }, }) }), ) it.live("backgrounds only jobs actively blocking a session", () => Effect.gen(function* () { const jobs = yield* Job.Service const parent = SessionSchema.ID.make("ses_parent") const other = SessionSchema.ID.make("ses_other") const latch = yield* Deferred.make() const first = yield* jobs.start({ id: "job_first", type: "test", run: Deferred.await(latch).pipe(Effect.as("first")), }) const second = yield* jobs.start({ id: "job_second", type: "test", run: Deferred.await(latch).pipe(Effect.as("second")), }) const third = yield* jobs.start({ id: "job_third", type: "other", run: Deferred.await(latch).pipe(Effect.as("third")), }) const scope = yield* Scope.Scope const firstWait = yield* jobs .block({ id: first.id, sessionID: parent }) .pipe(Effect.forkIn(scope, { startImmediately: true })) const secondWait = yield* jobs .block({ id: second.id, sessionID: other }) .pipe(Effect.forkIn(scope, { startImmediately: true })) const thirdWait = yield* jobs .block({ id: third.id, sessionID: parent }) .pipe(Effect.forkIn(scope, { startImmediately: true })) expect(yield* jobs.backgroundAll({ sessionID: parent, type: "test" })).toMatchObject([{ id: first.id }]) expect(yield* Fiber.join(firstWait)).toMatchObject({ type: "backgrounded", info: { id: first.id } }) yield* Deferred.succeed(latch, undefined) expect(yield* Fiber.join(secondWait)).toMatchObject({ type: "finished", info: { id: second.id } }) expect(yield* Fiber.join(thirdWait)).toMatchObject({ type: "finished", info: { id: third.id } }) }), ) it.live("interrupts live work without promising settlement after the owning process-local scope closes", () => Effect.gen(function* () { const scope = yield* Scope.make() const interrupted = yield* Deferred.make() const jobs = yield* Job.make.pipe(Scope.provide(scope)) const job = yield* jobs.start({ type: "test", run: Effect.never.pipe(Effect.ensuring(Deferred.succeed(interrupted, undefined))), }) yield* Scope.close(scope, Exit.void) yield* Deferred.await(interrupted).pipe(Effect.timeout("1 second")) // The abandoned in-memory registry is not a durable observation channel. expect((yield* jobs.get(job.id))?.status).toBe("running") }), ) })