diff --git a/packages/core/src/job.ts b/packages/core/src/job.ts index 026fa89c9863..78b25da8fdb2 100644 --- a/packages/core/src/job.ts +++ b/packages/core/src/job.ts @@ -62,6 +62,7 @@ type Active = { type State = { jobs: SynchronizedRef.SynchronizedRef> + deliveries: SynchronizedRef.SynchronizedRef>> scope: Scope.Scope } @@ -131,6 +132,7 @@ export interface Interface { readonly cancel: (id: string) => Effect.Effect readonly pendingBackground: Effect.Effect readonly completeBackground: (notificationID: SessionMessage.ID) => Effect.Effect + readonly awaitBackground: (notificationID: SessionMessage.ID) => Effect.Effect } export class Service extends Context.Service()("@opencode/Job") {} @@ -173,6 +175,7 @@ export const make = Effect.gen(function* () { const kv = yield* KV.Service const state: State = { jobs: yield* SynchronizedRef.make(new Map()), + deliveries: yield* SynchronizedRef.make(new Map()), scope: yield* Scope.Scope, } @@ -413,10 +416,41 @@ export const make = Effect.gen(function* () { return recovered }).pipe(Effect.withSpan("Job.pendingBackground")) - const completeBackground: Interface["completeBackground"] = Effect.fn("Job.completeBackground")((notificationID) => - kv.remove(`${backgroundPrefix}${notificationID}`), + const completeBackground: Interface["completeBackground"] = Effect.fn("Job.completeBackground")( + function* (notificationID) { + yield* kv.remove(`${backgroundPrefix}${notificationID}`) + const waiter = yield* SynchronizedRef.modify(state.deliveries, (waiters) => { + const deferred = waiters.get(notificationID) + if (!deferred) return [undefined, waiters] as const + const next = new Map(waiters) + next.delete(notificationID) + return [deferred, next] as const + }) + if (waiter) yield* Deferred.succeed(waiter, undefined) + }, ) + /** + * Resolves once a durable background notification has been admitted, which is + * when its marker clears. Waiting on the job itself only observes settlement; + * the wake-up notification is admitted asynchronously after that. + */ + const awaitBackground: Interface["awaitBackground"] = Effect.fn("Job.awaitBackground")(function* (notificationID) { + const waiter = yield* SynchronizedRef.modifyEffect( + state.deliveries, + Effect.fnUntraced(function* (waiters) { + // Marker read and waiter registration are atomic here; completeBackground + // clears the marker before resolving, so no separate recheck is needed. + if (!(yield* kv.get(`${backgroundPrefix}${notificationID}`))) return [undefined, waiters] as const + const existing = waiters.get(notificationID) + if (existing) return [existing, waiters] as const + const deferred = Deferred.makeUnsafe() + return [deferred, new Map(waiters).set(notificationID, deferred)] as const + }), + ) + if (waiter) yield* Deferred.await(waiter) + }) + return Service.of({ get, start, @@ -427,6 +461,7 @@ export const make = Effect.gen(function* () { cancel, pendingBackground, completeBackground, + awaitBackground, }) }) diff --git a/packages/core/src/session/execution/restart.ts b/packages/core/src/session/execution/restart.ts index 7038349350c3..017f4dc581aa 100644 --- a/packages/core/src/session/execution/restart.ts +++ b/packages/core/src/session/execution/restart.ts @@ -170,16 +170,7 @@ export const layer = (options?: Options) => title: recovery.description, notificationID: background.notificationID, recovery, - run: execution.resume(recovery.childSessionID).pipe( - Effect.andThen(store.context(recovery.childSessionID)), - Effect.map((messages) => { - const assistant = messages.findLast( - (message) => - message.type === "assistant" && message.time.completed !== undefined && message.error === undefined, - ) - return SubagentCompletion.text(assistant) - }), - ), + run: SubagentCompletion.finalText({ sessions, jobs, sessionID: recovery.childSessionID }), }) yield* jobs.background(background.id) yield* jobs.wait({ id: background.id }).pipe( diff --git a/packages/core/src/session/subagent-completion.ts b/packages/core/src/session/subagent-completion.ts index 746bc5170ee5..1834912e9477 100644 --- a/packages/core/src/session/subagent-completion.ts +++ b/packages/core/src/session/subagent-completion.ts @@ -4,6 +4,7 @@ import { Effect } from "effect" import type { Job } from "../job.js" import type { Session } from "../session.js" import type { SessionMessage } from "./message.js" +import { SessionSchema } from "./schema.js" export const NO_TEXT = "Subagent completed without a text response." @@ -17,6 +18,37 @@ export function text(message: SessionMessage.Info | undefined) { ) } +/** + * Runs the child session to quiescence and returns its final completed response. + * A child can end a turn while its own shell or nested subagent is still running; + * that work admits a wake-up notification and resumes the child, so the response + * is only final once no pending notification will wake it again. + */ +export const finalText = Effect.fnUntraced(function* (input: { + sessions: Pick + jobs: Pick + sessionID: SessionSchema.ID +}) { + while (true) { + yield* input.sessions.resume(input.sessionID) + const pending = (yield* input.jobs.pendingBackground).filter((job) => + job.recovery.kind === "shell" + ? job.recovery.sessionID === input.sessionID + : job.recovery.parentSessionID === input.sessionID, + ) + if (pending.length === 0) break + yield* Effect.forEach(pending, (job) => input.jobs.awaitBackground(job.notificationID), { + concurrency: "unbounded", + discard: true, + }) + } + const messages = yield* input.sessions.messages({ sessionID: input.sessionID, order: "desc", limit: 20 }) + const assistant = messages.find( + (message) => message.type === "assistant" && message.time.completed !== undefined && message.error === undefined, + ) + return text(assistant) +}) + export const deliver = Effect.fnUntraced(function* ( sessions: Pick, jobs: Pick, diff --git a/packages/core/src/session/subagent-job.ts b/packages/core/src/session/subagent-job.ts index f8e7503fbd0a..5a21263269ac 100644 --- a/packages/core/src/session/subagent-job.ts +++ b/packages/core/src/session/subagent-job.ts @@ -41,15 +41,7 @@ export const make: Effect.Effect - message.type === "assistant" && message.time.completed !== undefined && message.error === undefined, - ) - return SubagentCompletion.text(assistant) - }), + run: SubagentCompletion.finalText({ sessions, jobs, sessionID: recovery.childSessionID }), }), background: Effect.fn("SubagentJob.background")(function* (recovery: Recovery) { const info = yield* jobs.background(recovery.childSessionID) diff --git a/packages/core/test/job.test.ts b/packages/core/test/job.test.ts index 85d066e7995e..e178851b1493 100644 --- a/packages/core/test/job.test.ts +++ b/packages/core/test/job.test.ts @@ -275,6 +275,34 @@ describe("Job", () => { }), ) + it.live("waits for background acknowledgment before releasing an observer", () => + Effect.gen(function* () { + const jobs = yield* Job.Service + const job = yield* jobs.start({ + type: "shell", + recovery: { + kind: "shell", + sessionID: SessionSchema.ID.make("ses_background_await"), + shellID: "shell_background_await", + command: "echo done", + }, + run: Effect.succeed("done"), + }) + const background = yield* jobs.background(job.id) + if (!background?.notificationID) return yield* Effect.die("background marker missing") + + // Settlement alone must not release an observer; only acknowledgment does. + const waiting = yield* jobs + .awaitBackground(background.notificationID) + .pipe(Effect.forkIn(yield* Scope.Scope, { startImmediately: true })) + expect(yield* Fiber.await(waiting).pipe(Effect.timeoutOption("20 millis"))).toMatchObject({ _tag: "None" }) + + yield* jobs.completeBackground(background.notificationID) + yield* Fiber.join(waiting) + expect(yield* jobs.pendingBackground).toEqual([]) + }), + ) + it.live("persists backgroundAll ownership before releasing a blocked subagent", () => Effect.gen(function* () { const jobs = yield* Job.Service diff --git a/packages/core/test/tool-subagent.test.ts b/packages/core/test/tool-subagent.test.ts index c0e82c789394..bc47579cd99a 100644 --- a/packages/core/test/tool-subagent.test.ts +++ b/packages/core/test/tool-subagent.test.ts @@ -783,4 +783,86 @@ describe("SubagentTool", () => { ), ), ) + + it.live("waits for pending child background work before notifying the parent", () => + Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (dir) => Effect.promise(() => dir[Symbol.asyncDispose]()), + ).pipe( + Effect.flatMap((dir) => + Effect.gen(function* () { + const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) }) + const sessions = yield* Session.Service + const parent = yield* sessions.create({ location }) + const child = yield* sessions.create({ + parentID: parent.id, + title: "review", + agent: Agent.ID.make("reviewer"), + model: childModel, + }) + yield* withSubagent(parent.location) + const locations = yield* LocationServiceMap.Service + const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location))) + const jobs = yield* Job.Service + const bus = yield* Bus.Service + + // The child ends its turn while its own background shell is still running. + const shellGate = yield* Deferred.make() + const shell = yield* jobs.start({ + id: "sh_child_pending", + type: "shell", + title: "sleep", + recovery: { kind: "shell", sessionID: child.id, shellID: "sh_child_pending", command: "sleep" }, + run: Deferred.await(shellGate).pipe(Effect.as("")), + }) + const marker = yield* jobs.background(shell.id) + if (!marker?.notificationID) return yield* Effect.die("Expected a pending shell marker") + + const childTurn = yield* bus.subscribe(SessionEvent.Text.Ended).pipe( + Stream.filter((event) => event.data.sessionID === child.id), + Stream.take(1), + Stream.runCollect, + Effect.forkScoped({ startImmediately: true }), + ) + const notified = yield* bus.subscribe(SessionEvent.InboxEnqueued).pipe( + Stream.filter((event) => event.data.sessionID === parent.id && event.data.item.type === "synthetic"), + Stream.take(1), + Stream.runCollect, + Effect.forkScoped({ startImmediately: true }), + ) + + const settled = yield* executeTool(registry, { + sessionID: parent.id, + ...toolIdentity, + call: { + type: "tool-call", + id: "call-child-pending", + name: SubagentTool.name, + input: { + agent: "reviewer", + description: "background review", + prompt: "review", + sessionID: child.id, + background: true, + }, + }, + }) + expect(settled.metadata).toMatchObject({ sessionID: child.id, status: "running" }) + + yield* Fiber.join(childTurn) + yield* Effect.sleep("20 millis") + expect((yield* jobs.get(child.id))?.status).toBe("running") + expect((yield* sessions.inbox(parent.id)).filter((item) => item.type === "synthetic")).toEqual([]) + + // Acknowledging the shell notification clears its marker; only then may the child settle. + yield* jobs.completeBackground(marker.notificationID) + const admission = Array.from(yield* Fiber.join(notified))[0] + expect(admission?.data.item.type).toBe("synthetic") + if (admission?.data.item.type !== "synthetic") return yield* Effect.die("Expected a synthetic inbox item") + expect(admission.data.item.payload.text).toContain(`