fix(core): await child background work before subagent completion

packages/core/src/job.ts

@@ -62,6 +62,7 @@ type Active = {
 
 type State = {
   jobs: SynchronizedRef.SynchronizedRef<Map<string, Active>>
+  deliveries: SynchronizedRef.SynchronizedRef<Map<SessionMessage.ID, Deferred.Deferred<void>>>
   scope: Scope.Scope
 }
 
@@ -131,6 +132,7 @@ export interface Interface {
   readonly cancel: (id: string) => Effect.Effect<Info | undefined>
   readonly pendingBackground: Effect.Effect<readonly Background[]>
   readonly completeBackground: (notificationID: SessionMessage.ID) => Effect.Effect<void>
+  readonly awaitBackground: (notificationID: SessionMessage.ID) => Effect.Effect<void>
 }
 
 export class Service extends Context.Service<Service, Interface>()("@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<void>()
+        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,
   })
 })
 

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(

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<Session.Interface, "resume" | "messages">
+  jobs: Pick<Job.Interface, "pendingBackground" | "awaitBackground">
+  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<Session.Interface, "synthetic">,
   jobs: Pick<Job.Interface, "completeBackground">,

packages/core/src/session/subagent-job.ts

@@ -41,15 +41,7 @@ export const make: Effect.Effect<Runner, never, Session.Service | Job.Service |
         title: recovery.description,
         metadata: {},
         recovery,
-        run: Effect.gen(function* () {
-          yield* sessions.resume(recovery.childSessionID)
-          const messages = yield* sessions.messages({ sessionID: recovery.childSessionID, order: "desc", limit: 20 })
-          const assistant = messages.find(
-            (message) =>
-              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)

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

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<void>()
+          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(`<subagent sessionID="${child.id}" state="completed"`)
+          expect((yield* jobs.get(child.id))?.status).toBe("completed")
+        }),
+      ),
+    ),
+  )
 })