From 461a1c3ab428345008cfe9f11e23c6e4b9d1878b Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Mon, 29 Jun 2026 23:53:35 -0400 Subject: [PATCH] refactor(core): replace background job service (#34559) --- .../core/src/{background-job.ts => job.ts} | 275 +++++++++--------- packages/core/src/tool/shell.ts | 44 +-- packages/core/src/tool/subagent.ts | 42 +-- packages/core/test/background-job.test.ts | 103 ------- packages/core/test/job.test.ts | 164 +++++++++++ packages/core/test/tool-shell.test.ts | 24 +- packages/core/test/tool-subagent.test.ts | 8 +- packages/opencode/src/effect/app-runtime.ts | 4 +- packages/opencode/src/{background => }/job.ts | 24 +- .../instance/httpapi/handlers/experimental.ts | 14 +- .../server/routes/instance/httpapi/server.ts | 4 +- packages/opencode/src/session/run-state.ts | 28 +- packages/opencode/src/session/session.ts | 23 +- packages/opencode/src/tool/registry.ts | 6 +- packages/opencode/src/tool/task.ts | 53 ++-- packages/opencode/test/AGENTS.md | 2 +- packages/opencode/test/background/job.test.ts | 243 ---------------- packages/opencode/test/job.test.ts | 131 +++++++++ .../opencode/test/server/session-list.test.ts | 4 +- packages/opencode/test/session/prompt.test.ts | 4 +- .../opencode/test/session/session.test.ts | 4 +- packages/opencode/test/tool/task.test.ts | 58 ++-- specs/v2/schema-changelog.md | 2 +- specs/v2/todo.md | 2 +- 24 files changed, 593 insertions(+), 673 deletions(-) rename packages/core/src/{background-job.ts => job.ts} (53%) delete mode 100644 packages/core/test/background-job.test.ts create mode 100644 packages/core/test/job.test.ts rename packages/opencode/src/{background => }/job.ts (55%) delete mode 100644 packages/opencode/test/background/job.test.ts create mode 100644 packages/opencode/test/job.test.ts diff --git a/packages/core/src/background-job.ts b/packages/core/src/job.ts similarity index 53% rename from packages/core/src/background-job.ts rename to packages/core/src/job.ts index 4287830bf1..1662ea2b8b 100644 --- a/packages/core/src/background-job.ts +++ b/packages/core/src/job.ts @@ -1,8 +1,9 @@ -export * as BackgroundJob from "./background-job" +export * as Job from "./job" import { Cause, Clock, Context, Deferred, Effect, Exit, Layer, Scope, SynchronizedRef } from "effect" -import { Identifier } from "./id/id" import { makeGlobalNode } from "./effect/app-node" +import { Identifier } from "./id/id" +import { SessionSchema } from "./session/schema" export type Status = "running" | "completed" | "error" | "cancelled" @@ -21,14 +22,11 @@ export type Info = { type Active = { info: Info done: Deferred.Deferred + backgrounded: Deferred.Deferred scope: Scope.Closeable token: object - pending: number - next: number - output?: { sequence: number; text: string } - tail: Deferred.Deferred - promoted: Deferred.Deferred - onPromote?: Effect.Effect + blockingSessions: Map + isBackgrounded: boolean } type State = { @@ -42,36 +40,29 @@ type FinishResult = { scope?: Scope.Closeable } -type PromoteResult = { +type BackgroundResult = { info?: Info - promoted?: Deferred.Deferred - onPromote?: Effect.Effect + backgrounded?: Deferred.Deferred } type StartResult = { info: Info } | { info: Info; scope: Scope.Closeable; token: object } -type ExtendResult = - | { extended: false } - | { - extended: true - previous: Deferred.Deferred - scope: Scope.Closeable - tail: Deferred.Deferred - token: object - sequence: number - } +type BlockWait = { + done: Deferred.Deferred + backgrounded: Deferred.Deferred +} + +type BlockStart = + | { type: "missing" } + | { type: "finished"; info: Info } + | { type: "backgrounded"; info: Info } + | { type: "wait"; wait: BlockWait } export type StartInput = { id?: string type: string title?: string metadata?: Record - onPromote?: Effect.Effect - run: Effect.Effect -} - -export type ExtendInput = { - id: string run: Effect.Effect } @@ -85,18 +76,30 @@ export type WaitResult = { timedOut: boolean } +export type BlockInput = { + id: string + sessionID: SessionSchema.ID +} + +export type BlockResult = { type: "finished"; info: Info } | { type: "backgrounded"; info: Info } + +export type BackgroundAllInput = { + sessionID: SessionSchema.ID + type?: string +} + export interface Interface { readonly list: () => Effect.Effect readonly get: (id: string) => Effect.Effect readonly start: (input: StartInput) => Effect.Effect - readonly extend: (input: ExtendInput) => Effect.Effect readonly wait: (input: WaitInput) => Effect.Effect - readonly waitForPromotion: (id: string) => Effect.Effect - readonly promote: (id: string) => Effect.Effect + readonly block: (input: BlockInput) => Effect.Effect + readonly background: (id: string) => Effect.Effect + readonly backgroundAll: (input: BackgroundAllInput) => Effect.Effect readonly cancel: (id: string) => Effect.Effect } -export class Service extends Context.Service()("@opencode/BackgroundJob") {} +export class Service extends Context.Service()("@opencode/Job") {} function snapshot(job: Active): Info { return { @@ -110,6 +113,19 @@ function errorText(error: unknown) { return String(error) } +function incrementSession(input: Map, sessionID: SessionSchema.ID) { + return new Map(input).set(sessionID, (input.get(sessionID) ?? 0) + 1) +} + +function decrementSession(input: Map, sessionID: SessionSchema.ID) { + const count = input.get(sessionID) + if (count === undefined) return input + const next = new Map(input) + if (count <= 1) next.delete(sessionID) + else next.set(sessionID, count - 1) + return next +} + /** * Makes one scoped, process-local registry. Entries are intentionally not * durable: process restart or owner-scope closure loses status and interrupts @@ -123,26 +139,13 @@ export const make = Effect.gen(function* () { scope: yield* Scope.Scope, } - const settle = Effect.fn("BackgroundJob.settle")(function* ( - id: string, - token: object, - sequence: number, - exit: Exit.Exit, - ) { + const settle = Effect.fn("Job.settle")(function* (id: string, token: object, exit: Exit.Exit) { const completed_at = yield* Clock.currentTimeMillis const result = yield* SynchronizedRef.modify(state.jobs, (jobs): readonly [FinishResult, Map] => { const job = jobs.get(id) if (!job) return [{}, jobs] if (job.token !== token) return [{}, jobs] if (job.info.status !== "running") return [{ info: snapshot(job) }, jobs] - const pending = job.pending - 1 - const output = - Exit.isSuccess(exit) && (!job.output || sequence > job.output.sequence) - ? { sequence, text: exit.value } - : job.output - if (Exit.isSuccess(exit) && pending > 0) { - return [{}, new Map(jobs).set(id, { ...job, pending, output })] - } const status: Exclude = Exit.isSuccess(exit) ? "completed" : Cause.hasInterruptsOnly(exit.cause) @@ -150,14 +153,12 @@ export const make = Effect.gen(function* () { : "error" const next = { ...job, - onPromote: undefined, - pending: 0, - output, + blockingSessions: new Map(), info: { ...job.info, status, completed_at, - ...(output ? { output: output.text } : {}), + ...(Exit.isSuccess(exit) ? { output: exit.value } : {}), ...(Exit.isFailure(exit) ? { error: errorText(Cause.squash(exit.cause)) } : {}), }, } @@ -170,43 +171,41 @@ export const make = Effect.gen(function* () { return result.info }) - const fork = Effect.fn("BackgroundJob.fork")(function* ( + const fork = Effect.fn("Job.fork")(function* ( scope: Scope.Scope, id: string, token: object, - sequence: number, run: Effect.Effect, ) { return yield* run.pipe( Effect.matchCauseEffect({ - onSuccess: (output) => settle(id, token, sequence, Exit.succeed(output)), - onFailure: (cause) => settle(id, token, sequence, Exit.failCause(cause)), + onSuccess: (output) => settle(id, token, Exit.succeed(output)), + onFailure: (cause) => settle(id, token, Exit.failCause(cause)), }), Effect.asVoid, Effect.forkIn(scope, { startImmediately: true }), ) }) - const list: Interface["list"] = Effect.fn("BackgroundJob.list")(function* () { + const list: Interface["list"] = Effect.fn("Job.list")(function* () { return Array.from((yield* SynchronizedRef.get(state.jobs)).values()) .map(snapshot) .toSorted((a, b) => a.started_at - b.started_at) }) - const get: Interface["get"] = Effect.fn("BackgroundJob.get")(function* (id) { + const get: Interface["get"] = Effect.fn("Job.get")(function* (id) { const job = (yield* SynchronizedRef.get(state.jobs)).get(id) - if (!job) return + if (!job) return undefined return snapshot(job) }) - const start: Interface["start"] = Effect.fn("BackgroundJob.start")(function* (input) { + const start: Interface["start"] = Effect.fn("Job.start")(function* (input) { return yield* Effect.uninterruptibleMask((restore) => Effect.gen(function* () { const id = input.id ?? Identifier.ascending("job") const started_at = yield* Clock.currentTimeMillis const done = yield* Deferred.make() - const promoted = yield* Deferred.make() - const tail = yield* Deferred.make() + const backgrounded = yield* Deferred.make() const result = yield* SynchronizedRef.modifyEffect( state.jobs, Effect.fnUntraced(function* (jobs) { @@ -226,13 +225,11 @@ export const make = Effect.gen(function* () { metadata: input.metadata, }, done, + backgrounded, scope, token, - pending: 1, - next: 1, - tail, - promoted, - onPromote: input.onPromote, + blockingSessions: new Map(), + isBackgrounded: false, } return [{ info: snapshot(job), scope, token }, new Map(jobs).set(id, job)] as readonly [ StartResult, @@ -240,56 +237,13 @@ export const make = Effect.gen(function* () { ] }), ) - if ("scope" in result) - yield* fork( - result.scope, - id, - result.token, - 0, - restore(input.run).pipe(Effect.ensuring(Deferred.succeed(tail, undefined))), - ) + if ("scope" in result) yield* fork(result.scope, id, result.token, restore(input.run)) return result.info }), ) }) - const extend: Interface["extend"] = Effect.fn("BackgroundJob.extend")(function* (input) { - return yield* Effect.uninterruptibleMask((restore) => - Effect.gen(function* () { - const tail = yield* Deferred.make() - const result = yield* SynchronizedRef.modify( - state.jobs, - (jobs): readonly [ExtendResult, Map] => { - const job = jobs.get(input.id) - if (!job || job.info.status !== "running") return [{ extended: false }, jobs] - return [ - { extended: true, previous: job.tail, scope: job.scope, tail, token: job.token, sequence: job.next }, - new Map(jobs).set(input.id, { - ...job, - pending: job.pending + 1, - next: job.next + 1, - tail, - }), - ] - }, - ) - if (!result.extended) return false - yield* fork( - result.scope, - input.id, - result.token, - result.sequence, - Deferred.await(result.previous).pipe( - Effect.andThen(restore(input.run)), - Effect.ensuring(Deferred.succeed(result.tail, undefined)), - ), - ) - return true - }), - ) - }) - - const wait: Interface["wait"] = Effect.fn("BackgroundJob.wait")(function* (input) { + const wait: Interface["wait"] = Effect.fn("Job.wait")(function* (input) { const job = (yield* SynchronizedRef.get(state.jobs)).get(input.id) if (!job) return { timedOut: false } if (job.info.status !== "running") return { info: snapshot(job), timedOut: false } @@ -300,41 +254,91 @@ export const make = Effect.gen(function* () { return { info: snapshot(job), timedOut: true } }) - const waitForPromotion: Interface["waitForPromotion"] = Effect.fn("BackgroundJob.waitForPromotion")(function* (id) { - const job = (yield* SynchronizedRef.get(state.jobs)).get(id) - if (!job || job.info.status !== "running") return yield* Effect.never - if (job.info.metadata?.background === true) return snapshot(job) - return yield* Deferred.await(job.promoted) + const removeBlock = Effect.fn("Job.removeBlock")(function* (input: BlockInput) { + yield* SynchronizedRef.update(state.jobs, (jobs) => { + const job = jobs.get(input.id) + if (!job || job.info.status !== "running" || job.isBackgrounded) return jobs + return new Map(jobs).set(input.id, { + ...job, + blockingSessions: decrementSession(job.blockingSessions, input.sessionID), + }) + }) }) - const promote: Interface["promote"] = Effect.fn("BackgroundJob.promote")(function* (id) { - const result = yield* SynchronizedRef.modifyEffect( + const block: Interface["block"] = Effect.fn("Job.block")(function* (input) { + const result = yield* SynchronizedRef.modify(state.jobs, (jobs): readonly [BlockStart, Map] => { + const job = jobs.get(input.id) + if (!job) return [{ type: "missing" }, jobs] + if (job.info.status !== "running") return [{ type: "finished", info: snapshot(job) }, jobs] + if (job.isBackgrounded) return [{ type: "backgrounded", info: snapshot(job) }, jobs] + return [ + { type: "wait", wait: { done: job.done, backgrounded: job.backgrounded } }, + new Map(jobs).set(input.id, { + ...job, + blockingSessions: incrementSession(job.blockingSessions, input.sessionID), + }), + ] + }) + if (result.type === "missing") return undefined + if (result.type === "finished") return { type: "finished", info: result.info } + if (result.type === "backgrounded") return { type: "backgrounded", info: result.info } + return yield* Effect.raceFirst( + Deferred.await(result.wait.done).pipe(Effect.map((info) => ({ type: "finished" as const, info }))), + Deferred.await(result.wait.backgrounded).pipe(Effect.map((info) => ({ type: "backgrounded" as const, info }))), + ).pipe(Effect.ensuring(removeBlock(input))) + }) + + const background: Interface["background"] = Effect.fn("Job.background")(function* (id) { + const result = yield* SynchronizedRef.modify( state.jobs, - Effect.fnUntraced(function* (jobs) { + (jobs): readonly [BackgroundResult, Map] => { const job = jobs.get(id) - if (!job || job.info.status !== "running") return [{}, jobs] as readonly [PromoteResult, Map] - if (job.info.metadata?.background === true) - return [{ info: snapshot(job) }, jobs] as readonly [PromoteResult, Map] + if (!job || job.info.status !== "running") return [{}, jobs] + if (job.isBackgrounded) return [{ info: snapshot(job) }, jobs] const next = { ...job, - onPromote: undefined, - info: { - ...job.info, - metadata: { ...job.info.metadata, background: true }, - }, + isBackgrounded: true, + blockingSessions: new Map(), } - return [ - { info: snapshot(next), onPromote: job.onPromote, promoted: job.promoted }, - new Map(jobs).set(id, next), - ] as readonly [PromoteResult, Map] - }), + return [{ info: snapshot(next), backgrounded: job.backgrounded }, new Map(jobs).set(id, next)] + }, ) - if (result.info && result.promoted) yield* Deferred.succeed(result.promoted, result.info).pipe(Effect.ignore) - if (result.onPromote) yield* result.onPromote.pipe(Effect.ignore) + if (result.info && result.backgrounded) + yield* Deferred.succeed(result.backgrounded, result.info).pipe(Effect.ignore) return result.info }) - const cancel: Interface["cancel"] = Effect.fn("BackgroundJob.cancel")(function* (id) { + const backgroundAll: Interface["backgroundAll"] = Effect.fn("Job.backgroundAll")(function* (input) { + const result = yield* SynchronizedRef.modify( + state.jobs, + (jobs): readonly [BackgroundResult[], Map] => { + const results: BackgroundResult[] = [] + const next = new Map(jobs) + for (const [id, job] of jobs) { + if (job.info.status !== "running") continue + if (job.isBackgrounded) continue + if (input.type !== undefined && job.info.type !== input.type) continue + if (!job.blockingSessions.has(input.sessionID)) continue + const updated = { + ...job, + isBackgrounded: true, + blockingSessions: new Map(), + } + results.push({ info: snapshot(updated), backgrounded: job.backgrounded }) + next.set(id, updated) + } + return [results, next] + }, + ) + yield* Effect.forEach( + result, + (item) => (item.info && item.backgrounded ? Deferred.succeed(item.backgrounded, item.info) : Effect.void), + { discard: true }, + ) + return result.flatMap((item) => (item.info ? [item.info] : [])) + }) + + const cancel: Interface["cancel"] = Effect.fn("Job.cancel")(function* (id) { const completed_at = yield* Clock.currentTimeMillis const result = yield* SynchronizedRef.modify(state.jobs, (jobs): readonly [FinishResult, Map] => { const job = jobs.get(id) @@ -342,8 +346,7 @@ export const make = Effect.gen(function* () { if (job.info.status !== "running") return [{ info: snapshot(job) }, jobs] const next = { ...job, - onPromote: undefined, - pending: 0, + blockingSessions: new Map(), info: { ...job.info, status: "cancelled" as const, @@ -357,7 +360,7 @@ export const make = Effect.gen(function* () { return result.info }) - return Service.of({ list, get, start, extend, wait, waitForPromotion, promote, cancel }) + return Service.of({ list, get, start, wait, block, background, backgroundAll, cancel }) }) export const layer = Layer.effect(Service, make) diff --git a/packages/core/src/tool/shell.ts b/packages/core/src/tool/shell.ts index 9371a51d4f..b56438a80b 100644 --- a/packages/core/src/tool/shell.ts +++ b/packages/core/src/tool/shell.ts @@ -3,8 +3,8 @@ export * as ShellTool from "./shell" import path from "path" import { ToolFailure } from "@opencode-ai/llm" import { Effect, Layer, Schema, Scope } from "effect" -import { BackgroundJob } from "../background-job" import { FSUtil } from "../fs-util" +import { Job } from "../job" import { LocationMutation } from "../location-mutation" import { LocationServiceMap } from "../location-service-map" import { PermissionV2 } from "../permission" @@ -74,8 +74,8 @@ const modelOutput = (output: Output): string | undefined => { // TODO: Restore PowerShell and cmd-specific invocation/path handling on Windows. // TODO: Add plugin shell.env environment augmentation once V2 plugin hooks exist. // TODO: Add durable/live progress metadata streaming for long-running commands once V2 tool invocation progress context is wired. -// TODO: Persist background job status and define restart recovery before exposing remote observation. -// TODO: Add HTTP background-job observation only after durable status, restart recovery, and authorization are defined. +// TODO: Persist job status and define restart recovery before exposing remote observation. +// TODO: Add HTTP job observation only after durable status, restart recovery, and authorization are defined. // TODO: Revisit process-group cleanup and platform coverage with shell-specific tests if current AppProcess semantics do not fully cover it. // TODO: Revisit binary output handling if stdout/stderr decoding is text-only. // TODO: Stream full shell output into managed storage while retaining only a bounded in-memory preview. @@ -98,12 +98,12 @@ export const layer = Layer.effectDiscard( Effect.gen(function* () { const tools = yield* ApplicationTools.Service const sessions = yield* SessionV2.Service - const jobs = yield* BackgroundJob.Service + const jobs = yield* Job.Service const locations = yield* LocationServiceMap.Service const scope = yield* Scope.Scope const fsUtil = yield* FSUtil.Service - const injectWhenDone = Effect.fn("ShellTool.injectWhenDone")(function* ( + const notifyWhenDone = Effect.fn("ShellTool.notifyWhenDone")(function* ( sessionID: SessionSchema.ID, callID: string, command: string, @@ -121,9 +121,9 @@ export const layer = Layer.effectDiscard( if (state === undefined) return Effect.void const text = state === "completed" - ? result.info!.output ?? "" + ? (result.info!.output ?? "") : state === "error" - ? result.info!.error ?? "Command failed" + ? (result.info!.error ?? "Command failed") : "Command cancelled" return sessions.synthetic({ sessionID, @@ -156,9 +156,7 @@ export const layer = Layer.effectDiscard( Effect.gen(function* () { const parent = yield* sessions .get(context.sessionID) - .pipe( - Effect.mapError(() => new ToolFailure({ message: `Session not found: ${context.sessionID}` })), - ) + .pipe(Effect.mapError(() => new ToolFailure({ message: `Session not found: ${context.sessionID}` }))) return yield* Effect.gen(function* () { const mutation = yield* LocationMutation.Service const shell = yield* Shell.Service @@ -203,16 +201,18 @@ export const layer = Layer.effectDiscard( timeout, metadata: { sessionID: context.sessionID }, }) - const final = yield* shell.wait(info.id) - const page = yield* shell.output(info.id, { limit: MAX_CAPTURE_BYTES }) + return yield* Effect.gen(function* () { + const final = yield* shell.wait(info.id) + const page = yield* shell.output(info.id, { limit: MAX_CAPTURE_BYTES }) - if (final.status === "timeout") - return `Command exceeded timeout of ${timeout} ms. Retry with a larger timeout if the command is expected to take longer.` + if (final.status === "timeout") + return `Command exceeded timeout of ${timeout} ms. Retry with a larger timeout if the command is expected to take longer.` - const truncated = page.size > page.cursor - const body = page.output || "(no output)" - const notice = truncated ? `\n\n[output truncated; full output saved to: ${final.file}]` : "" - return `${body}${notice}` + const truncated = page.size > page.cursor + const body = page.output || "(no output)" + const notice = truncated ? `\n\n[output truncated; full output saved to: ${final.file}]` : "" + return `${body}${notice}` + }).pipe(Effect.onInterrupt(() => shell.remove(info.id).pipe(Effect.ignore))) }) const info = yield* jobs.start({ @@ -220,10 +220,10 @@ export const layer = Layer.effectDiscard( type: name, title: input.command, metadata: { sessionID: context.sessionID }, - onPromote: injectWhenDone(context.sessionID, context.toolCallID, input.command), run: run(), }) - yield* injectWhenDone(context.sessionID, context.toolCallID, input.command) + yield* jobs.background(info.id) + yield* notifyWhenDone(context.sessionID, context.toolCallID, input.command) return { output: BACKGROUND_STARTED, truncated: false, @@ -262,7 +262,7 @@ export const layer = Layer.effectDiscard( status: "completed" as const, ...(warnings.length ? { warnings } : {}), } - }).pipe(Effect.provide(locations.get(parent.location))) as Effect.Effect, ToolFailure> + }).pipe(Effect.provide(locations.get(parent.location))) }).pipe(Effect.mapError(() => new ToolFailure({ message: `Unable to execute command: ${input.command}` }))), }), }) @@ -273,5 +273,5 @@ export const layer = Layer.effectDiscard( export const node = makeGlobalNode({ name: "shell-tool", layer, - deps: [ApplicationTools.node, SessionV2.node, BackgroundJob.node, LocationServiceMap.node, FSUtil.node], + deps: [ApplicationTools.node, SessionV2.node, Job.node, LocationServiceMap.node, FSUtil.node], }) diff --git a/packages/core/src/tool/subagent.ts b/packages/core/src/tool/subagent.ts index f5e8ebcd84..e32987aad3 100644 --- a/packages/core/src/tool/subagent.ts +++ b/packages/core/src/tool/subagent.ts @@ -3,7 +3,7 @@ export * as SubagentTool from "./subagent" import { ToolFailure } from "@opencode-ai/llm" import { Effect, Layer, Schema, Scope } from "effect" import { AgentV2 } from "../agent" -import { BackgroundJob } from "../background-job" +import { Job } from "../job" import { LocationServiceMap } from "../location-service-map" import { SessionV2 } from "../session" import { SessionSchema } from "../session/schema" @@ -44,7 +44,7 @@ export const layer = Layer.effectDiscard( Effect.gen(function* () { const tools = yield* ApplicationTools.Service const sessions = yield* SessionV2.Service - const jobs = yield* BackgroundJob.Service + const jobs = yield* Job.Service const locations = yield* LocationServiceMap.Service const scope = yield* Scope.Scope @@ -77,7 +77,7 @@ export const layer = Layer.effectDiscard( }) }) - const injectWhenDone = Effect.fn("SubagentTool.injectWhenDone")(function* ( + const notifyWhenDone = Effect.fn("SubagentTool.notifyWhenDone")(function* ( parentID: SessionSchema.ID, childID: SessionSchema.ID, description: string, @@ -138,37 +138,37 @@ export const layer = Layer.effectDiscard( yield* sessions.prompt({ sessionID: child.id, prompt: { text: input.prompt }, resume: false }) yield* sessions.resume(child.id) return yield* latestAssistantText(child.id) - }) + }).pipe(Effect.onInterrupt(() => sessions.interrupt(child.id))) const info = yield* jobs.start({ id: child.id, type: name, title: input.description, metadata: {}, - onPromote: injectWhenDone(context.sessionID, child.id, input.description), run, }) if (background) { - if ((yield* jobs.promote(info.id)) === undefined) - yield* injectWhenDone(context.sessionID, child.id, input.description) + yield* jobs.background(info.id) + yield* notifyWhenDone(context.sessionID, child.id, input.description) return { sessionID: child.id, status: "running" as const, output: BACKGROUND_STARTED } } - const result = yield* Effect.raceFirst( - jobs.wait({ id: child.id }).pipe(Effect.map((waited) => waited.info)), - jobs.waitForPromotion(child.id), - ).pipe( - Effect.onInterrupt(() => - Effect.all([sessions.interrupt(child.id), jobs.cancel(child.id)], { discard: true }), - ), - ) - if (result?.metadata?.background === true) + const result = yield* jobs + .block({ id: child.id, sessionID: context.sessionID }) + .pipe( + Effect.onInterrupt(() => + Effect.all([sessions.interrupt(child.id), jobs.cancel(child.id)], { discard: true }), + ), + ) + if (result?.type === "backgrounded") { + yield* notifyWhenDone(context.sessionID, child.id, input.description) return { sessionID: child.id, status: "running" as const, output: BACKGROUND_STARTED } - if (result?.status === "error") - return yield* new ToolFailure({ message: result.error ?? "Subagent failed" }) - if (result?.status === "cancelled") return yield* new ToolFailure({ message: "Subagent cancelled" }) - return { sessionID: child.id, status: "completed" as const, output: result?.output ?? NO_TEXT } + } + if (result?.info.status === "error") + return yield* new ToolFailure({ message: result.info.error ?? "Subagent failed" }) + if (result?.info.status === "cancelled") return yield* new ToolFailure({ message: "Subagent cancelled" }) + return { sessionID: child.id, status: "completed" as const, output: result?.info.output ?? NO_TEXT } }), }), }) @@ -182,5 +182,5 @@ export const layer = Layer.effectDiscard( export const node = makeGlobalNode({ name: "subagent-tool", layer, - deps: [ApplicationTools.node, SessionV2.node, BackgroundJob.node, LocationServiceMap.node], + deps: [ApplicationTools.node, SessionV2.node, Job.node, LocationServiceMap.node], }) diff --git a/packages/core/test/background-job.test.ts b/packages/core/test/background-job.test.ts deleted file mode 100644 index 1c4f93f019..0000000000 --- a/packages/core/test/background-job.test.ts +++ /dev/null @@ -1,103 +0,0 @@ -import { describe, expect } from "bun:test" -import { BackgroundJob } from "@opencode-ai/core/background-job" -import { Deferred, Effect, Exit, Scope } from "effect" -import { it } from "./lib/effect" - -describe("BackgroundJob", () => { - it.live("tracks process-local work through explicit observation", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.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" }, - }) - }).pipe(Effect.provide(BackgroundJob.layer)), - ) - - it.live("publishes jobs before starting immediately settling work", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.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}` }, - }) - }) - }) - }).pipe(Effect.provide(BackgroundJob.layer)), - ) - - it.live("increments pending work before starting immediately settling extensions", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - - yield* Effect.forEach(Array.from({ length: 100 }), (_, index) => - Effect.gen(function* () { - const first = yield* Deferred.make() - const job = yield* jobs.start({ - type: "test", - run: Deferred.await(first).pipe(Effect.as(`first-${index}`)), - }) - - expect(yield* jobs.extend({ id: job.id, run: Effect.succeed(`second-${index}`) })).toBe(true) - expect((yield* jobs.get(job.id))?.status).toBe("running") - - yield* Deferred.succeed(first, undefined) - expect(yield* jobs.wait({ id: job.id })).toMatchObject({ - timedOut: false, - info: { status: "completed", output: `second-${index}` }, - }) - }), - ) - }).pipe(Effect.provide(BackgroundJob.layer)), - ) - - 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* BackgroundJob.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") - }), - ) -}) diff --git a/packages/core/test/job.test.ts b/packages/core/test/job.test.ts new file mode 100644 index 0000000000..8c35d51210 --- /dev/null +++ b/packages/core/test/job.test.ts @@ -0,0 +1,164 @@ +import { describe, expect } from "bun:test" +import { Job } from "@opencode-ai/core/job" +import { Deferred, Effect, Exit, Fiber, Scope } from "effect" +import { SessionSchema } from "@opencode-ai/core/session/schema" +import { testEffect } from "./lib/effect" + +const it = testEffect(Job.layer) + +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") + }), + ) +}) diff --git a/packages/core/test/tool-shell.test.ts b/packages/core/test/tool-shell.test.ts index d9243ad0c5..42617c8f75 100644 --- a/packages/core/test/tool-shell.test.ts +++ b/packages/core/test/tool-shell.test.ts @@ -17,12 +17,11 @@ import { ModelV2 } from "@opencode-ai/core/model" import { ProviderV2 } from "@opencode-ai/core/provider" import { AbsolutePath } from "@opencode-ai/core/schema" import { AgentV2 } from "@opencode-ai/core/agent" -import { BackgroundJob } from "@opencode-ai/core/background-job" +import { Job } from "@opencode-ai/core/job" import { SessionV2 } from "@opencode-ai/core/session" import { SessionEvent } from "@opencode-ai/core/session/event" import { SessionExecution } from "@opencode-ai/core/session/execution" import { SessionMessage } from "@opencode-ai/core/session/message" -import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model" import { SessionStore } from "@opencode-ai/core/session/store" import { PermissionV2 } from "@opencode-ai/core/permission" import { ShellTool } from "@opencode-ai/core/tool/shell" @@ -120,7 +119,7 @@ const layer = AppNodeBuilder.build( LayerNode.group([ Database.node, EventV2.node, - BackgroundJob.node, + Job.node, ToolOutputStore.cleanupNode, SessionV2.node, ShellTool.node, @@ -155,10 +154,7 @@ const overflowCommand = (bytes: number) => ? `[Console]::Out.Write(('x' * ${bytes})); Start-Sleep -Milliseconds 100` : `head -c ${bytes} /dev/zero | tr '\\0' 'x'` -const withSession = ( - directory: string, - body: (registry: ToolRegistry.Interface) => Effect.Effect, -) => +const withSession = (directory: string, body: (registry: ToolRegistry.Interface) => Effect.Effect) => Effect.gen(function* () { const sessions = yield* SessionV2.Service const location = Location.Ref.make({ directory: AbsolutePath.make(directory) }) @@ -214,9 +210,7 @@ describe("ShellTool", () => { reset() return Effect.promise(() => fs.mkdir(path.join(tmp.path, "src"))).pipe( Effect.andThen( - withSession(tmp.path, (registry) => - settleTool(registry, call({ command: cwdCommand, workdir: "src" })), - ), + withSession(tmp.path, (registry) => settleTool(registry, call({ command: cwdCommand, workdir: "src" }))), ), Effect.andThen((settled) => Effect.sync(() => @@ -247,9 +241,7 @@ describe("ShellTool", () => { : Effect.void return Effect.promise(() => fs.mkdir(workdir)).pipe( Effect.andThen( - withSession(tmp.path, (registry) => - executeTool(registry, call({ command: cwdCommand, workdir: "src" })), - ), + withSession(tmp.path, (registry) => executeTool(registry, call({ command: cwdCommand, workdir: "src" }))), ), Effect.andThen(Effect.sync(() => expect(assertions.map((input) => input.action)).toEqual(["shell"]))), ) @@ -314,9 +306,7 @@ describe("ShellTool", () => { reset() denyAction = "external_directory" const target = path.join(outside.path, "secret.txt") - return withSession(active.path, (registry) => - settleTool(registry, call({ command: `cat ${target}` })), - ).pipe( + return withSession(active.path, (registry) => settleTool(registry, call({ command: `cat ${target}` }))).pipe( Effect.andThen((settled) => Effect.sync(() => { expect(assertions.map((item) => item.action)).toEqual(["shell"]) @@ -417,7 +407,7 @@ test("keeps locked deferred parity TODOs visible", async () => { "Restore PowerShell and cmd-specific invocation/path handling on Windows.", "Add plugin shell.env environment augmentation once V2 plugin hooks exist.", "Add durable/live progress metadata streaming for long-running commands once V2 tool invocation progress context is wired.", - "Persist background job status and define restart recovery before exposing remote observation.", + "Persist job status and define restart recovery before exposing remote observation.", "Revisit process-group cleanup and platform coverage with shell-specific tests if current AppProcess semantics do not fully cover it.", "Revisit binary output handling if stdout/stderr decoding is text-only.", "Stream full shell output into managed storage while retaining only a bounded in-memory preview.", diff --git a/packages/core/test/tool-subagent.test.ts b/packages/core/test/tool-subagent.test.ts index ae50ecc9d5..70dd693244 100644 --- a/packages/core/test/tool-subagent.test.ts +++ b/packages/core/test/tool-subagent.test.ts @@ -10,7 +10,7 @@ import { ModelV2 } from "@opencode-ai/core/model" import { ProviderV2 } from "@opencode-ai/core/provider" import { AbsolutePath } from "@opencode-ai/core/schema" import { AgentV2 } from "@opencode-ai/core/agent" -import { BackgroundJob } from "@opencode-ai/core/background-job" +import { Job } from "@opencode-ai/core/job" import { LocationServiceMap } from "@opencode-ai/core/location-service-map" import { SessionV2 } from "@opencode-ai/core/session" import { SessionEvent } from "@opencode-ai/core/session/event" @@ -95,7 +95,7 @@ const layer = AppNodeBuilder.build( LayerNode.group([ Database.node, EventV2.node, - BackgroundJob.node, + Job.node, ToolOutputStore.cleanupNode, SessionV2.node, SubagentTool.node, @@ -242,7 +242,7 @@ describe("SubagentTool", () => { ), ) - it.live("promotes background work and injects one synthetic parent completion", () => + it.live("notifies once when background work completes", () => Effect.acquireRelease( Effect.promise(() => tmpdir()), (dir) => Effect.promise(() => dir[Symbol.asyncDispose]()), @@ -251,7 +251,6 @@ describe("SubagentTool", () => { Effect.gen(function* () { const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) }) const sessions = yield* SessionV2.Service - const jobs = yield* BackgroundJob.Service const parent = yield* sessions.create({ location }) yield* withSubagent(parent.location) const locations = yield* LocationServiceMap.Service @@ -270,7 +269,6 @@ describe("SubagentTool", () => { const childID = outputSessionID(settled.output?.structured) expect(settled.output?.structured).toMatchObject({ status: "running" }) - yield* jobs.promote(childID) yield* Effect.yieldNow const synthetic = (yield* sessions.context(parent.id)).filter((message) => message.type === "synthetic") expect(synthetic).toHaveLength(1) diff --git a/packages/opencode/src/effect/app-runtime.ts b/packages/opencode/src/effect/app-runtime.ts index 30dbb4c880..b6060279e9 100644 --- a/packages/opencode/src/effect/app-runtime.ts +++ b/packages/opencode/src/effect/app-runtime.ts @@ -48,7 +48,7 @@ import { ShareNext } from "@/share/share-next" import { SessionShare } from "@/share/session" import { Npm } from "@opencode-ai/core/npm" import { memoMap } from "@opencode-ai/core/effect/memo-map" -import { BackgroundJob } from "@/background/job" +import { Job } from "@/job" import { RuntimeFlags } from "@/effect/runtime-flags" import { EventV2Bridge } from "@/event-v2-bridge" @@ -74,7 +74,7 @@ export const AppLayer = Layer.mergeAll( Todo.defaultLayer, Session.defaultLayer, SessionStatus.defaultLayer, - BackgroundJob.defaultLayer, + Job.defaultLayer, RuntimeFlags.defaultLayer, EventV2Bridge.defaultLayer, SessionRunState.defaultLayer, diff --git a/packages/opencode/src/background/job.ts b/packages/opencode/src/job.ts similarity index 55% rename from packages/opencode/src/background/job.ts rename to packages/opencode/src/job.ts index 8095826efc..0e8ea7825a 100644 --- a/packages/opencode/src/background/job.ts +++ b/packages/opencode/src/job.ts @@ -1,32 +1,34 @@ import { LayerNode } from "@opencode-ai/core/effect/layer-node" -import { BackgroundJob as CoreBackgroundJob } from "@opencode-ai/core/background-job" +import { Service, make } from "@opencode-ai/core/job" import { InstanceState } from "@/effect/instance-state" import { Effect, Layer } from "effect" export { Service, - type ExtendInput, + type BackgroundAllInput, + type BlockInput, + type BlockResult, type Info, type Interface, type StartInput, type Status, type WaitInput, type WaitResult, -} from "@opencode-ai/core/background-job" +} from "@opencode-ai/core/job" /** Keeps the legacy service instance-scoped while sharing the core registry engine. */ export const layer = Layer.effect( - CoreBackgroundJob.Service, + Service, Effect.gen(function* () { - const state = yield* InstanceState.make(() => CoreBackgroundJob.make) - return CoreBackgroundJob.Service.of({ + const state = yield* InstanceState.make(() => make) + return Service.of({ list: () => InstanceState.useEffect(state, (jobs) => jobs.list()), get: (id) => InstanceState.useEffect(state, (jobs) => jobs.get(id)), start: (input) => InstanceState.useEffect(state, (jobs) => jobs.start(input)), - extend: (input) => InstanceState.useEffect(state, (jobs) => jobs.extend(input)), wait: (input) => InstanceState.useEffect(state, (jobs) => jobs.wait(input)), - waitForPromotion: (id) => InstanceState.useEffect(state, (jobs) => jobs.waitForPromotion(id)), - promote: (id) => InstanceState.useEffect(state, (jobs) => jobs.promote(id)), + block: (input) => InstanceState.useEffect(state, (jobs) => jobs.block(input)), + background: (id) => InstanceState.useEffect(state, (jobs) => jobs.background(id)), + backgroundAll: (input) => InstanceState.useEffect(state, (jobs) => jobs.backgroundAll(input)), cancel: (id) => InstanceState.useEffect(state, (jobs) => jobs.cancel(id)), }) }), @@ -34,6 +36,6 @@ export const layer = Layer.effect( export const defaultLayer = layer -export const node = LayerNode.make({ service: CoreBackgroundJob.Service, layer, deps: [] }) +export const node = LayerNode.make({ service: Service, layer, deps: [] }) -export * as BackgroundJob from "./job" +export * as Job from "./job" diff --git a/packages/opencode/src/server/routes/instance/httpapi/handlers/experimental.ts b/packages/opencode/src/server/routes/instance/httpapi/handlers/experimental.ts index caed544005..1433c2041a 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/handlers/experimental.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/handlers/experimental.ts @@ -1,6 +1,6 @@ import { Account } from "@/account/account" import { Agent } from "@/agent/agent" -import { BackgroundJob } from "@/background/job" +import { Job } from "@/job" import { Config } from "@/config/config" import { InstanceState } from "@/effect/instance-state" import { RuntimeFlags } from "@/effect/runtime-flags" @@ -33,7 +33,7 @@ export const experimentalHandlers = HttpApiBuilder.group(InstanceHttpApi, "exper const registry = yield* ToolRegistry.Service const worktreeSvc = yield* Worktree.Service const sessions = yield* Session.Service - const background = yield* BackgroundJob.Service + const jobs = yield* Job.Service const flags = yield* RuntimeFlags.Service const capabilities = Effect.fn("ExperimentalHttpApi.capabilities")(function* () { @@ -159,15 +159,7 @@ export const experimentalHandlers = HttpApiBuilder.group(InstanceHttpApi, "exper params: { sessionID: SessionID } }) { if (!flags.experimentalBackgroundSubagents) return false - const jobs = (yield* background.list()).filter( - (job) => - job.type === "task" && - job.status === "running" && - job.metadata?.parentSessionId === ctx.params.sessionID && - job.metadata.background !== true, - ) - const promoted = yield* Effect.forEach(jobs, (job) => background.promote(job.id), { concurrency: "unbounded" }) - return promoted.some((job) => job !== undefined) + return (yield* jobs.backgroundAll({ sessionID: ctx.params.sessionID, type: "task" })).length > 0 }) const resource = Effect.fn("ExperimentalHttpApi.resource")(function* () { diff --git a/packages/opencode/src/server/routes/instance/httpapi/server.ts b/packages/opencode/src/server/routes/instance/httpapi/server.ts index c3c253812f..1795c2b4da 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/server.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/server.ts @@ -7,7 +7,7 @@ import * as Observability from "@opencode-ai/core/observability" import { Account } from "@/account/account" import { Agent } from "@/agent/agent" import { Auth } from "@/auth" -import { BackgroundJob } from "@/background/job" +import { Job } from "@/job" import { Command } from "@/command" import { Config } from "@/config/config" import { Workspace } from "@/control-plane/workspace" @@ -233,7 +233,7 @@ const app = LayerNode.group([ Session.node, SessionProjector.node, SessionStatus.node, - BackgroundJob.node, + Job.node, RuntimeFlags.node, EventV2Bridge.node, SessionRunState.node, diff --git a/packages/opencode/src/session/run-state.ts b/packages/opencode/src/session/run-state.ts index 17b0efaeaa..f37ceff97e 100644 --- a/packages/opencode/src/session/run-state.ts +++ b/packages/opencode/src/session/run-state.ts @@ -2,7 +2,7 @@ import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { InstanceState } from "@/effect/instance-state" import { SessionV1 } from "@opencode-ai/core/v1/session" import { Runner } from "@/effect/runner" -import { BackgroundJob } from "@/background/job" +import { Job } from "@/job" import { Effect, Latch, Layer, Scope, Context } from "effect" import { Session } from "./session" import { SessionID } from "./schema" @@ -29,7 +29,7 @@ export class Service extends Context.Service()("@opencode/Se export const layer = Layer.effect( Service, Effect.gen(function* () { - const background = yield* BackgroundJob.Service + const jobs = yield* Job.Service const status = yield* SessionStatus.Service const state = yield* InstanceState.make( @@ -75,7 +75,7 @@ export const layer = Layer.effect( }) const cancel = Effect.fn("SessionRunState.cancel")(function* (sessionID: SessionID) { - yield* cancelBackgroundJobs(background, sessionID) + yield* cancelJobs(jobs, sessionID) const data = yield* InstanceState.get(state) const existing = data.runners.get(sessionID) if (!existing) { @@ -108,31 +108,25 @@ export const layer = Layer.effect( }), ) -export const defaultLayer = layer.pipe( - Layer.provide(BackgroundJob.defaultLayer), - Layer.provide(SessionStatus.defaultLayer), -) +export const defaultLayer = layer.pipe(Layer.provide(Job.defaultLayer), Layer.provide(SessionStatus.defaultLayer)) -const cancelBackgroundJobs = Effect.fn("SessionRunState.cancelBackgroundJobs")(function* ( - background: BackgroundJob.Interface, - sessionID: SessionID, -) { - const jobs = yield* background.list() +const cancelJobs = Effect.fn("SessionRunState.cancelJobs")(function* (jobs: Job.Interface, sessionID: SessionID) { + const running = yield* jobs.list() const pending = new Set([sessionID]) const cancelled = new Set() - const matches = (job: BackgroundJob.Info) => { + const matches = (job: Job.Info) => { if (job.status !== "running") return false if (cancelled.has(job.id)) return false if (pending.has(job.id)) return true if (typeof job.metadata?.sessionId === "string" && pending.has(job.metadata.sessionId)) return true return typeof job.metadata?.parentSessionId === "string" && pending.has(job.metadata.parentSessionId) } - let batch = jobs.filter(matches) + let batch = running.filter(matches) while (batch.length > 0) { yield* Effect.forEach( batch, (job) => - background.cancel(job.id).pipe( + jobs.cancel(job.id).pipe( Effect.tap(() => Effect.sync(() => { cancelled.add(job.id) @@ -143,7 +137,7 @@ const cancelBackgroundJobs = Effect.fn("SessionRunState.cancelBackgroundJobs")(f ), { concurrency: "unbounded", discard: true }, ) - batch = jobs.filter(matches) + batch = running.filter(matches) } }) @@ -151,6 +145,6 @@ function busyError(sessionID: SessionID) { return new Session.BusyError({ sessionID }) } -export const node = LayerNode.make({ service: Service, layer: layer, deps: [BackgroundJob.node, SessionStatus.node] }) +export const node = LayerNode.make({ service: Service, layer: layer, deps: [Job.node, SessionStatus.node] }) export * as SessionRunState from "./run-state" diff --git a/packages/opencode/src/session/session.ts b/packages/opencode/src/session/session.ts index f4edf50ed6..1d3252bb09 100644 --- a/packages/opencode/src/session/session.ts +++ b/packages/opencode/src/session/session.ts @@ -4,7 +4,7 @@ import { Slug } from "@opencode-ai/core/util/slug" import { SessionV1 } from "@opencode-ai/core/v1/session" import { serviceUse } from "@opencode-ai/core/effect/service-use" import path from "path" -import { BackgroundJob } from "@/background/job" +import { Job } from "@/job" import { Decimal } from "decimal.js" import type { ProviderMetadata, Usage } from "@opencode-ai/llm" import { InstallationVersion } from "@opencode-ai/core/installation/version" @@ -491,13 +491,13 @@ export type Patch = Omit, "time" | "share" | "summary" | "revert" export const layer: Layer.Layer< Service, never, - BackgroundJob.Service | RuntimeFlags.Service | Database.Service | EventV2Bridge.Service + Job.Service | RuntimeFlags.Service | Database.Service | EventV2Bridge.Service > = Layer.effect( Service, Effect.gen(function* () { const { db } = yield* Database.Service const database = yield* Database.Service - const background = yield* BackgroundJob.Service + const jobs = yield* Job.Service const events = yield* EventV2Bridge.Service const flags = yield* RuntimeFlags.Service @@ -618,7 +618,7 @@ export const layer: Layer.Layer< Effect.catchCause(() => Effect.succeed(false)), ) - if (hasInstance) yield* cancelBackgroundJobs(background, sessionID) + if (hasInstance) yield* cancelJobs(jobs, sessionID) const kids = yield* children(sessionID) for (const child of kids) { yield* remove(child.id) @@ -941,7 +941,7 @@ export const layer: Layer.Layer< ) export const defaultLayer = layer.pipe( - Layer.provide(BackgroundJob.defaultLayer), + Layer.provide(Job.defaultLayer), Layer.provide(Database.defaultLayer), Layer.provide(EventV2Bridge.defaultLayer), Layer.provide( @@ -953,19 +953,16 @@ export const defaultLayer = layer.pipe( Layer.provide(RuntimeFlags.defaultLayer), ) -const cancelBackgroundJobs = Effect.fn("Session.cancelBackgroundJobs")(function* ( - background: BackgroundJob.Interface, - sessionID: SessionID, -) { - const jobs = yield* background.list() +const cancelJobs = Effect.fn("Session.cancelJobs")(function* (jobs: Job.Interface, sessionID: SessionID) { + const running = yield* jobs.list() yield* Effect.forEach( - jobs.filter((job) => { + running.filter((job) => { if (job.status !== "running") return false if (job.id === sessionID) return true if (job.metadata?.sessionId === sessionID) return true return job.metadata?.parentSessionId === sessionID }), - (job) => background.cancel(job.id), + (job) => jobs.cancel(job.id), { concurrency: "unbounded", discard: true }, ) }) @@ -1098,7 +1095,7 @@ export function* listGlobal(input?: { export const node = LayerNode.make({ service: Service, layer: layer, - deps: [BackgroundJob.node, RuntimeFlags.node, Database.node, EventV2Bridge.node], + deps: [Job.node, RuntimeFlags.node, Database.node, EventV2Bridge.node], }) export * as Session from "./session" diff --git a/packages/opencode/src/tool/registry.ts b/packages/opencode/src/tool/registry.ts index 553fa1ebaf..2b64d24eee 100644 --- a/packages/opencode/src/tool/registry.ts +++ b/packages/opencode/src/tool/registry.ts @@ -48,7 +48,7 @@ import { EventV2Bridge } from "@/event-v2-bridge" import { Agent } from "../agent/agent" import { Skill } from "../skill" import { Permission } from "@/permission" -import { BackgroundJob } from "@/background/job" +import { Job } from "@/job" import { RuntimeFlags } from "@/effect/runtime-flags" import { ProviderV2 } from "@opencode-ai/core/provider" import { ModelV2 } from "@opencode-ai/core/model" @@ -325,7 +325,7 @@ export const defaultLayer = Layer.suspend(() => Layer.provide(Skill.defaultLayer), Layer.provide(Agent.defaultLayer), Layer.provide(Session.defaultLayer), - Layer.provide(BackgroundJob.defaultLayer), + Layer.provide(Job.defaultLayer), Layer.provide(Provider.defaultLayer), Layer.provide(LSP.defaultLayer), Layer.provide(Instruction.defaultLayer), @@ -426,7 +426,7 @@ export const node = LayerNode.make({ Agent.node, Skill.node, Session.node, - BackgroundJob.node, + Job.node, Provider.node, LSP.node, Instruction.node, diff --git a/packages/opencode/src/tool/task.ts b/packages/opencode/src/tool/task.ts index b0a866c90e..f61cea1048 100644 --- a/packages/opencode/src/tool/task.ts +++ b/packages/opencode/src/tool/task.ts @@ -2,7 +2,7 @@ import * as Tool from "./tool" import DESCRIPTION from "./task.txt" import { ToolJsonSchema } from "./json-schema" import { SessionV1 } from "@opencode-ai/core/v1/session" -import { BackgroundJob } from "@/background/job" +import { Job } from "@/job" import { Session } from "@/session/session" import { SessionID, MessageID } from "../session/schema" import { MessageV2 } from "../session/message-v2" @@ -33,11 +33,11 @@ const BACKGROUND_STARTED = [ "DO NOT sleep, poll for progress, ask the task for status, or duplicate this task's work — avoid working with the same files or topics it is using.", "Work on non-overlapping tasks, or briefly tell the user what you launched and end your response.", ].join("\n") -const BACKGROUND_UPDATED = [ - "Additional context sent to the running background task.", +const BACKGROUND_ALREADY_RUNNING = [ + "The task is already working in the background.", "The task is still working in the background. You will be notified automatically when it finishes.", "DO NOT sleep, poll for progress, ask the task for status, or duplicate this task's work — avoid working with the same files or topics it is using.", - "Work on non-overlapping tasks, or briefly tell the user what you sent and end your response.", + "Work on non-overlapping tasks, or briefly tell the user it is still running and end your response.", ].join("\n") const BaseParameterFields = { @@ -82,7 +82,7 @@ export const TaskTool = Tool.define( id, Effect.gen(function* () { const agent = yield* Agent.Service - const background = yield* BackgroundJob.Service + const jobs = yield* Job.Service const config = yield* Config.Service const sessions = yield* Session.Service const scope = yield* Scope.Scope @@ -229,7 +229,7 @@ export const TaskTool = Tool.define( }) const notify = Effect.fn("TaskTool.notifyBackgroundResult")(function* (jobID: string) { - yield* background.wait({ id: jobID }).pipe( + yield* jobs.wait({ id: jobID }).pipe( Effect.flatMap((result) => { if (result.info?.status === "completed") return inject("completed", result.info.output ?? "") if (result.info?.status === "error") return inject("error", result.info.error ?? "") @@ -239,7 +239,8 @@ export const TaskTool = Tool.define( ) }) - if (yield* background.extend({ id: nextSession.id, run: runTask() })) { + const existing = yield* jobs.get(nextSession.id) + if (existing?.status === "running") { return { title: params.description, metadata: { @@ -250,24 +251,17 @@ export const TaskTool = Tool.define( output: renderOutput({ sessionID: nextSession.id, state: "running", - summary: "Background task updated", - text: BACKGROUND_UPDATED, + summary: "Background task already running", + text: BACKGROUND_ALREADY_RUNNING, }), } } - const info = yield* background.start({ + const info = yield* jobs.start({ id: nextSession.id, type: id, title: params.description, metadata, - onPromote: Effect.all([ - ctx.metadata({ - title: params.description, - metadata: { ...metadata, background: true, jobId: nextSession.id }, - }), - notify(nextSession.id), - ]), run: runTask().pipe(Effect.onInterrupt(() => ops.cancel(nextSession.id))), }) @@ -289,6 +283,7 @@ export const TaskTool = Tool.define( } if (runInBackground) { + yield* jobs.background(info.id) yield* notify(info.id) return backgroundResult() } @@ -306,23 +301,27 @@ export const TaskTool = Tool.define( }), () => Effect.gen(function* () { - const result = yield* Effect.raceFirst( - background.wait({ id: nextSession.id }).pipe(Effect.map((waited) => waited.info)), - background.waitForPromotion(nextSession.id), - ) - if (result?.metadata?.background === true) return backgroundResult() - if (result?.status === "error") return yield* Effect.fail(new Error(result.error ?? "Task failed")) - if (result?.status === "cancelled") return yield* Effect.fail(new Error("Task cancelled")) + const result = yield* jobs.block({ id: nextSession.id, sessionID: ctx.sessionID }) + if (result?.type === "backgrounded") { + yield* ctx.metadata({ + title: params.description, + metadata: { ...metadata, background: true, jobId: nextSession.id }, + }) + yield* notify(nextSession.id) + return backgroundResult() + } + if (result?.info.status === "error") + return yield* Effect.fail(new Error(result.info.error ?? "Task failed")) + if (result?.info.status === "cancelled") return yield* Effect.fail(new Error("Task cancelled")) return { title: params.description, metadata, - output: renderOutput({ sessionID: nextSession.id, state: "completed", text: result?.output ?? "" }), + output: renderOutput({ sessionID: nextSession.id, state: "completed", text: result?.info.output ?? "" }), } }), (_, exit) => Effect.gen(function* () { - if (Exit.hasInterrupts(exit)) - yield* Effect.all([cancel, background.cancel(nextSession.id)], { discard: true }) + if (Exit.hasInterrupts(exit)) yield* Effect.all([cancel, jobs.cancel(nextSession.id)], { discard: true }) }).pipe( Effect.ensuring( Effect.sync(() => { diff --git a/packages/opencode/test/AGENTS.md b/packages/opencode/test/AGENTS.md index 464b75cd82..33b4e9c143 100644 --- a/packages/opencode/test/AGENTS.md +++ b/packages/opencode/test/AGENTS.md @@ -172,7 +172,7 @@ Wait on a **published readiness signal**, not wall-clock time. Available afforda - `awaitWithTimeout(effect, message, duration?)` from `test/lib/effect.ts` — wrap any effect with `Effect.timeoutOrElse` and a custom error message. - `llm.wait(n)` from `test/lib/llm-server.ts` — wait until the mock LLM has received `n` HTTP calls. - `SessionStatus.Service` `.get(sessionID)` — observable per-session state (`{ type: "busy" | "idle" | ... }`). -- `BackgroundJob.wait({ id, timeout })` from `src/background/job.ts` — wait for a background job to complete. +- `Job.wait({ id, timeout })` from `src/job.ts` — wait for a job to complete. - Bus subscriptions — fork `Stream.runForEach(bus.subscribe(Event), ...)` and open a `Latch` inside the callback to signal first-event readiness. - `Deferred.await(deferred).pipe(Effect.timeoutOrElse(...))` for one-shot signals. diff --git a/packages/opencode/test/background/job.test.ts b/packages/opencode/test/background/job.test.ts deleted file mode 100644 index dbcb484dc6..0000000000 --- a/packages/opencode/test/background/job.test.ts +++ /dev/null @@ -1,243 +0,0 @@ -import { describe, expect } from "bun:test" -import { Deferred, Effect } from "effect" -import { BackgroundJob } from "@/background/job" -import { testEffect } from "../lib/effect" - -const it = testEffect(BackgroundJob.defaultLayer) - -describe("background.job", () => { - it.instance("tracks started jobs through completion", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const latch = yield* Deferred.make() - const job = yield* jobs.start({ - type: "test", - title: "test job", - run: Deferred.await(latch).pipe(Effect.as("done")), - }) - - expect(job.id.startsWith("job_")).toBe(true) - expect(job.status).toBe("running") - expect(job.title).toBe("test job") - - yield* Deferred.succeed(latch, undefined) - const done = yield* jobs.wait({ id: job.id }) - - expect(done.timedOut).toBe(false) - expect(done.info?.status).toBe("completed") - expect(done.info?.output).toBe("done") - expect((yield* jobs.list()).map((item) => item.id)).toEqual([job.id]) - }), - ) - - it.instance("returns a running snapshot when wait times out", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const job = yield* jobs.start({ - type: "test", - run: Effect.never, - }) - - const result = yield* jobs.wait({ id: job.id, timeout: 1 }) - - expect(result.timedOut).toBe(true) - expect(result.info?.status).toBe("running") - }), - ) - - it.instance("deduplicates concurrent starts for a running id", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const started = yield* Deferred.make() - const id = "job_test" - const [first, second] = yield* Effect.all( - [ - jobs.start({ - id, - type: "test", - run: Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)), - }), - jobs.start({ - id, - type: "test", - run: Effect.fail(new Error("duplicate started")), - }), - ], - { concurrency: "unbounded" }, - ) - - yield* Deferred.await(started) - - expect(first.id).toBe(id) - expect(second.id).toBe(id) - expect(first.status).toBe("running") - expect(second.status).toBe("running") - expect((yield* jobs.list()).map((item) => item.id)).toEqual([id]) - - yield* jobs.cancel(id) - }), - ) - - it.instance("waits for extensions before completing a running job", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const first = yield* Deferred.make() - const second = yield* Deferred.make() - const job = yield* jobs.start({ - type: "test", - run: Deferred.await(first).pipe(Effect.as("first")), - }) - - expect(yield* jobs.extend({ id: job.id, run: Deferred.await(second).pipe(Effect.as("second")) })).toBe(true) - yield* Deferred.succeed(first, undefined) - expect((yield* jobs.get(job.id))?.status).toBe("running") - - yield* Deferred.succeed(second, undefined) - const done = yield* jobs.wait({ id: job.id }) - expect(done.info?.status).toBe("completed") - expect(done.info?.output).toBe("second") - }), - ) - - it.instance("runs extensions after earlier work completes", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const first = yield* Deferred.make() - const order: string[] = [] - const job = yield* jobs.start({ - type: "test", - run: Effect.sync(() => order.push("start")).pipe(Effect.andThen(Deferred.await(first)), Effect.as("first")), - }) - - expect( - yield* jobs.extend({ - id: job.id, - run: Effect.sync(() => order.push("extend")).pipe(Effect.as("second")), - }), - ).toBe(true) - yield* Effect.yieldNow - expect(order).toEqual(["start"]) - - yield* Deferred.succeed(first, undefined) - expect((yield* jobs.wait({ id: job.id })).info?.output).toBe("second") - expect(order).toEqual(["start", "extend"]) - }), - ) - - it.instance("rejects extensions after a job completes", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const job = yield* jobs.start({ type: "test", run: Effect.succeed("done") }) - yield* jobs.wait({ id: job.id }) - - expect(yield* jobs.extend({ id: job.id, run: Effect.succeed("late") })).toBe(false) - expect((yield* jobs.get(job.id))?.output).toBe("done") - }), - ) - - it.instance("records failed jobs", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const job = yield* jobs.start({ - type: "test", - run: Effect.fail(new Error("boom")), - }) - - const result = yield* jobs.wait({ id: job.id }) - - expect(result.info?.status).toBe("error") - expect(result.info?.error).toBe("boom") - }), - ) - - it.instance("ignores stale settlements after restarting a failed job", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const fail = yield* Deferred.make() - const interrupted = yield* Deferred.make() - const release = yield* Deferred.make() - const id = "job_test" - yield* jobs.start({ - id, - type: "test", - run: Deferred.await(fail).pipe(Effect.andThen(Effect.fail(new Error("boom")))), - }) - yield* jobs.extend({ - id, - run: Effect.never.pipe( - Effect.ensuring(Deferred.succeed(interrupted, undefined).pipe(Effect.andThen(Deferred.await(release)))), - ), - }) - - yield* Deferred.succeed(fail, undefined) - expect((yield* jobs.wait({ id })).info?.status).toBe("error") - yield* Deferred.await(interrupted) - yield* jobs.start({ id, type: "test", run: Effect.never }) - - yield* Deferred.succeed(release, undefined) - yield* Effect.yieldNow - expect((yield* jobs.get(id))?.status).toBe("running") - yield* jobs.cancel(id) - }), - ) - - it.instance("can cancel running jobs", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const interrupted = yield* Deferred.make() - const job = yield* jobs.start({ - type: "test", - run: Effect.never.pipe(Effect.ensuring(Deferred.succeed(interrupted, undefined))), - }) - yield* jobs.extend({ - id: job.id, - run: Effect.never, - }) - - const cancelled = yield* jobs.cancel(job.id) - - expect(cancelled?.status).toBe("cancelled") - yield* Deferred.await(interrupted).pipe(Effect.timeout("1 second")) - expect((yield* jobs.get(job.id))?.status).toBe("cancelled") - }), - ) - - it.instance("promotes running jobs without interrupting them", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const latch = yield* Deferred.make() - const promoted = yield* Deferred.make() - const job = yield* jobs.start({ - type: "test", - metadata: { parentSessionId: "parent" }, - onPromote: Deferred.succeed(promoted, undefined).pipe(Effect.asVoid), - run: Deferred.await(latch).pipe(Effect.as("done")), - }) - - const info = yield* jobs.promote(job.id) - - expect(info?.status).toBe("running") - expect(info?.metadata?.background).toBe(true) - yield* Deferred.await(promoted) - expect((yield* jobs.get(job.id))?.status).toBe("running") - - yield* Deferred.succeed(latch, undefined) - expect((yield* jobs.wait({ id: job.id })).info?.output).toBe("done") - }), - ) - - it.instance("returns immutable snapshots", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const job = yield* jobs.start({ - type: "test", - metadata: { value: "initial" }, - run: Effect.succeed("done"), - }) - - if (job.metadata) job.metadata.value = "changed" - - expect((yield* jobs.get(job.id))?.metadata?.value).toBe("initial") - }), - ) -}) diff --git a/packages/opencode/test/job.test.ts b/packages/opencode/test/job.test.ts new file mode 100644 index 0000000000..28da9c43bd --- /dev/null +++ b/packages/opencode/test/job.test.ts @@ -0,0 +1,131 @@ +import { describe, expect } from "bun:test" +import { Deferred, Effect, Fiber } from "effect" +import { Job } from "@/job" +import { SessionSchema } from "@opencode-ai/core/session/schema" +import { testEffect } from "./lib/effect" + +const it = testEffect(Job.defaultLayer) + +describe("job", () => { + it.instance("tracks started jobs through completion", () => + Effect.gen(function* () { + const jobs = yield* Job.Service + const latch = yield* Deferred.make() + const job = yield* jobs.start({ + type: "test", + title: "test job", + run: Deferred.await(latch).pipe(Effect.as("done")), + }) + + expect(job.id.startsWith("job_")).toBe(true) + expect(job.status).toBe("running") + expect(job.title).toBe("test job") + + yield* Deferred.succeed(latch, undefined) + const done = yield* jobs.wait({ id: job.id }) + + expect(done.timedOut).toBe(false) + expect(done.info?.status).toBe("completed") + expect(done.info?.output).toBe("done") + expect((yield* jobs.list()).map((item) => item.id)).toEqual([job.id]) + }), + ) + + it.instance("returns a running snapshot when wait times out", () => + Effect.gen(function* () { + const jobs = yield* Job.Service + const job = yield* jobs.start({ type: "test", run: Effect.never }) + + const result = yield* jobs.wait({ id: job.id, timeout: 1 }) + + expect(result.timedOut).toBe(true) + expect(result.info?.status).toBe("running") + }), + ) + + it.instance("deduplicates concurrent starts for a running id", () => + Effect.gen(function* () { + const jobs = yield* Job.Service + const started = yield* Deferred.make() + const id = "job_test" + const [first, second] = yield* Effect.all( + [ + jobs.start({ + id, + type: "test", + run: Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)), + }), + jobs.start({ id, type: "test", run: Effect.fail(new Error("duplicate started")) }), + ], + { concurrency: "unbounded" }, + ) + + yield* Deferred.await(started) + + expect(first.id).toBe(id) + expect(second.id).toBe(id) + expect(first.status).toBe("running") + expect(second.status).toBe("running") + expect((yield* jobs.list()).map((item) => item.id)).toEqual([id]) + + yield* jobs.cancel(id) + }), + ) + + it.instance("records failed jobs", () => + Effect.gen(function* () { + const jobs = yield* Job.Service + const job = yield* jobs.start({ type: "test", run: Effect.fail(new Error("boom")) }) + + const result = yield* jobs.wait({ id: job.id }) + + expect(result.info?.status).toBe("error") + expect(result.info?.error).toBe("boom") + }), + ) + + it.instance("can cancel running jobs", () => + Effect.gen(function* () { + const jobs = yield* Job.Service + const interrupted = yield* Deferred.make() + const job = yield* jobs.start({ + type: "test", + run: Effect.never.pipe(Effect.ensuring(Deferred.succeed(interrupted, undefined))), + }) + + const cancelled = yield* jobs.cancel(job.id) + + expect(cancelled?.status).toBe("cancelled") + yield* Deferred.await(interrupted).pipe(Effect.timeout("1 second")) + expect((yield* jobs.get(job.id))?.status).toBe("cancelled") + }), + ) + + it.instance("releases blocking waits when backgrounded", () => + 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.forkChild) + + expect(yield* jobs.background(job.id)).toMatchObject({ id: job.id, status: "running" }) + expect(yield* Fiber.join(waiting)).toMatchObject({ type: "backgrounded", info: { id: job.id } }) + + yield* Deferred.succeed(latch, undefined) + expect((yield* jobs.wait({ id: job.id })).info?.output).toBe("done") + }), + ) + + it.instance("returns immutable snapshots", () => + Effect.gen(function* () { + const jobs = yield* Job.Service + const job = yield* jobs.start({ type: "test", metadata: { value: "initial" }, run: Effect.succeed("done") }) + + if (job.metadata) job.metadata.value = "changed" + + expect((yield* jobs.get(job.id))?.metadata?.value).toBe("initial") + }), + ) +}) diff --git a/packages/opencode/test/server/session-list.test.ts b/packages/opencode/test/server/session-list.test.ts index 213e3cdce3..7a345da433 100644 --- a/packages/opencode/test/server/session-list.test.ts +++ b/packages/opencode/test/server/session-list.test.ts @@ -12,7 +12,7 @@ import { testEffect } from "../lib/effect" import { EventV2Bridge } from "@/event-v2-bridge" import { Storage } from "@/storage/storage" import { RuntimeFlags } from "@/effect/runtime-flags" -import { BackgroundJob } from "@/background/job" +import { Job } from "@/job" const layer = (experimentalWorkspaces: boolean) => Layer.mergeAll( @@ -24,7 +24,7 @@ const layer = (experimentalWorkspaces: boolean) => Layer.provide(EventV2Bridge.defaultLayer), Layer.provide(SessionProjector.defaultLayer), Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces })), - Layer.provide(BackgroundJob.defaultLayer), + Layer.provide(Job.defaultLayer), ), ) const it = testEffect(layer(false)) diff --git a/packages/opencode/test/session/prompt.test.ts b/packages/opencode/test/session/prompt.test.ts index dbaa724f76..d726deea85 100644 --- a/packages/opencode/test/session/prompt.test.ts +++ b/packages/opencode/test/session/prompt.test.ts @@ -11,7 +11,7 @@ import path from "path" import { fileURLToPath } from "url" import { NamedError } from "@opencode-ai/core/util/error" import { Agent as AgentSvc } from "../../src/agent/agent" -import { BackgroundJob } from "@/background/job" +import { Job } from "@/job" import { Command } from "../../src/command" import { Config } from "@/config/config" import { LSP } from "@/lsp/lsp" @@ -183,7 +183,7 @@ function makePrompt(input?: { mcpInstructions?: MCP.ServerInstructions[]; proces lsp, makeMcp(input?.mcpInstructions), FSUtil.defaultLayer, - BackgroundJob.defaultLayer, + Job.defaultLayer, status, Database.defaultLayer, EventV2Bridge.defaultLayer, diff --git a/packages/opencode/test/session/session.test.ts b/packages/opencode/test/session/session.test.ts index c82f713d2b..375f9078d2 100644 --- a/packages/opencode/test/session/session.test.ts +++ b/packages/opencode/test/session/session.test.ts @@ -12,7 +12,7 @@ import { provideInstance, testInstanceStoreLayer, tmpdirScoped } from "../fixtur import { testEffect } from "../lib/effect" import { Storage } from "@/storage/storage" import { RuntimeFlags } from "@/effect/runtime-flags" -import { BackgroundJob } from "@/background/job" +import { Job } from "@/job" import { EventV2Bridge } from "@/event-v2-bridge" import { GlobalBus } from "@/bus/global" @@ -24,7 +24,7 @@ const it = testEffect( Layer.provideMerge(EventV2Bridge.defaultLayer), Layer.provide(SessionProjector.defaultLayer), Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces: false })), - Layer.provide(BackgroundJob.defaultLayer), + Layer.provide(Job.defaultLayer), ), CrossSpawnSpawner.defaultLayer, testInstanceStoreLayer, diff --git a/packages/opencode/test/tool/task.test.ts b/packages/opencode/test/tool/task.test.ts index 97bb7db065..8f0d95d74e 100644 --- a/packages/opencode/test/tool/task.test.ts +++ b/packages/opencode/test/tool/task.test.ts @@ -3,7 +3,7 @@ import { SessionV1 } from "@opencode-ai/core/v1/session" import { Database } from "@opencode-ai/core/database/database" import { Deferred, Effect, Exit, Fiber, Layer } from "effect" import { Agent } from "../../src/agent/agent" -import { BackgroundJob } from "@/background/job" +import { Job } from "@/job" import { EventV2Bridge } from "@/event-v2-bridge" import { Config } from "@/config/config" import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" @@ -19,7 +19,7 @@ import { Truncate } from "@/tool/truncate" import { ToolRegistry } from "@/tool/registry" import { RuntimeFlags } from "@/effect/runtime-flags" import { disposeAllInstances } from "../fixture/fixture" -import { testEffect } from "../lib/effect" +import { pollWithTimeout, testEffect } from "../lib/effect" import { ProviderV2 } from "@opencode-ai/core/provider" import { ModelV2 } from "@opencode-ai/core/model" @@ -35,7 +35,7 @@ const ref = { const layer = (flags: Partial = {}) => Layer.mergeAll( Agent.defaultLayer, - BackgroundJob.defaultLayer, + Job.defaultLayer, EventV2Bridge.defaultLayer, Config.defaultLayer, CrossSpawnSpawner.defaultLayer, @@ -480,9 +480,9 @@ describe("tool.task", () => { }), ) - it.instance("promotes a running foreground task without restarting it", () => + it.instance("backgrounds a running foreground task without restarting it", () => Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service + const jobs = yield* Job.Service const { chat, assistant } = yield* seed() const tool = yield* TaskTool const def = yield* tool.init() @@ -531,7 +531,12 @@ describe("tool.task", () => { expect(job).toBeDefined() if (!job) throw new Error("task job not found") expect(job.metadata?.parentSessionId).toBe(chat.id) - yield* jobs.promote(job.id) + yield* pollWithTimeout( + jobs + .backgroundAll({ sessionID: chat.id, type: "task" }) + .pipe(Effect.map((backgrounded) => (backgrounded.length > 0 ? backgrounded : undefined))), + "task never blocked the parent session", + ) const result = yield* Fiber.join(fiber) expect(result.metadata.background).toBe(true) @@ -548,7 +553,7 @@ describe("tool.task", () => { background.instance("execute launches background tasks without waiting for completion", () => Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service + const jobs = yield* Job.Service const { chat, assistant } = yield* seed() const tool = yield* TaskTool const def = yield* tool.init() @@ -584,15 +589,13 @@ describe("tool.task", () => { }), ) - background.instance("background task completion waits for running updates", () => + background.instance("running task_id reports the existing background task", () => Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service + const jobs = yield* Job.Service const { chat, assistant } = yield* seed() const tool = yield* TaskTool const def = yield* tool.init() const first = defer() - const second = defer() - const updated = defer() const injected = defer() let prompts = 0 const promptOps: TaskPromptOps = { @@ -603,9 +606,7 @@ describe("tool.task", () => { return Effect.succeed(reply(input, "done")) } prompts++ - if (prompts === 1) return Effect.promise(() => first.promise).pipe(Effect.as(reply(input, "first done"))) - updated.resolve(input) - return Effect.promise(() => second.promise).pipe(Effect.as(reply(input, "second done"))) + return Effect.promise(() => first.promise).pipe(Effect.as(reply(input, "first done"))) }, } const context = { @@ -640,27 +641,22 @@ describe("tool.task", () => { expect(result.metadata.sessionId).toBe(started.metadata.sessionId) expect(result.metadata.background).toBe(true) - expect(result.output).toContain("Background task updated") + expect(result.output).toContain("Background task already running") + expect(prompts).toBe(1) first.resolve() - expect((yield* jobs.get(started.metadata.sessionId))?.status).toBe("running") - expect((yield* Effect.promise(() => updated.promise)).parts).toEqual([ - { type: "text", text: "also inspect cancellation" }, - ]) - - second.resolve() const waited = yield* jobs.wait({ id: started.metadata.sessionId, timeout: 1_000 }) expect(waited.info?.status).toBe("completed") - expect(waited.info?.output).toBe("second done") + expect(waited.info?.output).toBe("first done") const notification = yield* Effect.promise(() => injected.promise) expect(notification.variant).toBe("xhigh") expect(notification.parts[0]?.type).toBe("text") - if (notification.parts[0]?.type === "text") expect(notification.parts[0].text).toContain("second done") + if (notification.parts[0]?.type === "text") expect(notification.parts[0].text).toContain("first done") }), ) - background.instance("background tasks complete through the background job service", () => + background.instance("background tasks complete through the job service", () => Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service + const jobs = yield* Job.Service const { chat, assistant } = yield* seed() const tool = yield* TaskTool const def = yield* tool.init() @@ -693,7 +689,7 @@ describe("tool.task", () => { background.instance("background task completion does not wait for the parent async prompt", () => Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service + const jobs = yield* Job.Service const { chat, assistant } = yield* seed() const tool = yield* TaskTool const def = yield* tool.init() @@ -731,7 +727,7 @@ describe("tool.task", () => { background.instance("removing the parent session cancels running background tasks", () => Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service + const jobs = yield* Job.Service const sessions = yield* Session.Service const { chat, assistant } = yield* seed() const tool = yield* TaskTool @@ -770,7 +766,7 @@ describe("tool.task", () => { background.instance("removing the child task session cancels its running background task", () => Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service + const jobs = yield* Job.Service const sessions = yield* Session.Service const { chat, assistant } = yield* seed() const tool = yield* TaskTool @@ -809,7 +805,7 @@ describe("tool.task", () => { background.instance("cancelling the parent run cancels running background tasks", () => Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service + const jobs = yield* Job.Service const runState = yield* SessionRunState.Service const { chat, assistant } = yield* seed() const tool = yield* TaskTool @@ -848,7 +844,7 @@ describe("tool.task", () => { it.instance("cancelling a child run cancels its own pre-runner task job", () => Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service + const jobs = yield* Job.Service const runState = yield* SessionRunState.Service const sessions = yield* Session.Service const { chat } = yield* seed() @@ -869,7 +865,7 @@ describe("tool.task", () => { it.instance("cancelling a parent run recursively cancels descendant background tasks", () => Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service + const jobs = yield* Job.Service const runState = yield* SessionRunState.Service const sessions = yield* Session.Service const { chat } = yield* seed() diff --git a/specs/v2/schema-changelog.md b/specs/v2/schema-changelog.md index 379c675d3d..e9ea22e522 100644 --- a/specs/v2/schema-changelog.md +++ b/specs/v2/schema-changelog.md @@ -690,7 +690,7 @@ Affected schema: Change: - Remove the optional `background` bash parameter and process-local background settlement shape from the shipped tool. -- Retain the internal `BackgroundJob` prototype for a later integration slice. +- Retain the internal `Job` prototype for a later integration slice. Reason: diff --git a/specs/v2/todo.md b/specs/v2/todo.md index b774c51c6b..ee7f9dff2a 100644 --- a/specs/v2/todo.md +++ b/specs/v2/todo.md @@ -47,7 +47,7 @@ Next reviewed slices: remaining one-turn native-adapter use with a narrow typed dispatcher - batch streamed deltas and add covering context indexes - expose replayable Session event cursors over HTTP and the generated SDK where remote consumers need them -- integrate the new BackgroundJob service with V2 tool execution: support background +- integrate the new Job service with V2 tool execution: support background bash jobs and background agent dispatch with durable status observation, completion delivery, and explicit cancellation / continuation semantics - add durable/clustered interruption, retries, and stale-owner fencing only as