ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(core): replace background job service (#34559)

Browse source
This commit is contained in:
Kit Langton 2026-06-29 23:53:35 -04:00 • committed by GitHub
commit 461a1c3ab4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 593 additions and 673 deletions
Download patch file Download diff file Expand all files Collapse all files

275
packages/core/src/background-job.ts → packages/core/src/job.ts
View file

@ -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<Info>
backgrounded: Deferred.Deferred<Info>
scope: Scope.Closeable
token: object
pending: number
next: number
output?: { sequence: number; text: string }
tail: Deferred.Deferred<void>
promoted: Deferred.Deferred<Info>
onPromote?: Effect.Effect<void>
blockingSessions: Map<SessionSchema.ID, number>
isBackgrounded: boolean
}
type State = {
@ -42,36 +40,29 @@ type FinishResult = {
scope?: Scope.Closeable
}
type PromoteResult = {
type BackgroundResult = {
info?: Info
promoted?: Deferred.Deferred<Info>
onPromote?: Effect.Effect<void>
backgrounded?: Deferred.Deferred<Info>
}
type StartResult = { info: Info } | { info: Info; scope: Scope.Closeable; token: object }
type ExtendResult =
| { extended: false }
| {
extended: true
previous: Deferred.Deferred<void>
scope: Scope.Closeable
tail: Deferred.Deferred<void>
token: object
sequence: number
}
type BlockWait = {
done: Deferred.Deferred<Info>
backgrounded: Deferred.Deferred<Info>
}
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<string, unknown>
onPromote?: Effect.Effect<void>
run: Effect.Effect<string, unknown>
}
export type ExtendInput = {
id: string
run: Effect.Effect<string, unknown>
}
@ -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<Info[]>
readonly get: (id: string) => Effect.Effect<Info | undefined>
readonly start: (input: StartInput) => Effect.Effect<Info>
readonly extend: (input: ExtendInput) => Effect.Effect<boolean>
readonly wait: (input: WaitInput) => Effect.Effect<WaitResult>
readonly waitForPromotion: (id: string) => Effect.Effect<Info>
readonly promote: (id: string) => Effect.Effect<Info | undefined>
readonly block: (input: BlockInput) => Effect.Effect<BlockResult | undefined>
readonly background: (id: string) => Effect.Effect<Info | undefined>
readonly backgroundAll: (input: BackgroundAllInput) => Effect.Effect<Info[]>
readonly cancel: (id: string) => Effect.Effect<Info | undefined>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/BackgroundJob") {}
export class Service extends Context.Service<Service, Interface>()("@opencode/Job") {}
function snapshot(job: Active): Info {
return {
@ -110,6 +113,19 @@ function errorText(error: unknown) {
return String(error)
}
function incrementSession(input: Map<SessionSchema.ID, number>, sessionID: SessionSchema.ID) {
return new Map(input).set(sessionID, (input.get(sessionID) ?? 0) + 1)
}
function decrementSession(input: Map<SessionSchema.ID, number>, 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<string, unknown>,
) {
const settle = Effect.fn("Job.settle")(function* (id: string, token: object, exit: Exit.Exit<string, unknown>) {
const completed_at = yield* Clock.currentTimeMillis
const result = yield* SynchronizedRef.modify(state.jobs, (jobs): readonly [FinishResult, Map<string, Active>] => {
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<Status, "running"> = 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<SessionSchema.ID, number>(),
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<string, unknown>,
) {
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<Info>()
const promoted = yield* Deferred.make<Info>()
const tail = yield* Deferred.make<void>()
const backgrounded = yield* Deferred.make<Info>()
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<SessionSchema.ID, number>(),
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<void>()
const result = yield* SynchronizedRef.modify(
state.jobs,
(jobs): readonly [ExtendResult, Map<string, Active>] => {
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<string, Active>] => {
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<string, Active>] => {
const job = jobs.get(id)
if (!job || job.info.status !== "running") return [{}, jobs] as readonly [PromoteResult, Map<string, Active>]
if (job.info.metadata?.background === true)
return [{ info: snapshot(job) }, jobs] as readonly [PromoteResult, Map<string, Active>]
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<SessionSchema.ID, number>(),
}
return [
{ info: snapshot(next), onPromote: job.onPromote, promoted: job.promoted },
new Map(jobs).set(id, next),
] as readonly [PromoteResult, Map<string, Active>]
}),
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<string, Active>] => {
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<SessionSchema.ID, number>(),
}
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<string, Active>] => {
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<SessionSchema.ID, number>(),
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)

44
packages/core/src/tool/shell.ts
View file

@ -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<Schema.Schema.Type<typeof Output>, 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],
})

42
packages/core/src/tool/subagent.ts
View file

@ -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],
})

103
packages/core/test/background-job.test.ts
View file

@ -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<void>()
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<void>()
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<void>()
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")
}),
)
})

164
packages/core/test/job.test.ts Normal file
View file

@ -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<void>()
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<void>()
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<void>()
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<void>()
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<void>()
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")
}),
)
})

24
packages/core/test/tool-shell.test.ts
View file

@ -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 = <A, E, R>(
directory: string,
body: (registry: ToolRegistry.Interface) => Effect.Effect<A, E, R>,
) =>
const withSession = <A, E, R>(directory: string, body: (registry: ToolRegistry.Interface) => Effect.Effect<A, E, R>) =>
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.",

8
packages/core/test/tool-subagent.test.ts
View file

@ -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)