fix(core): recover durable held subagent reports after restart

packages/core/src/session/execution/restart.ts

@@ -100,10 +100,7 @@ export const layer = (options?: Options) =>
         return true
       })
 
-      const restartMessage = (
-        background: Job.Background,
-        recovery: Exclude<Job.Recovery, { kind: "subagent" }>,
-      ) => {
+      const restartMessage = (background: Job.Background, recovery: Exclude<Job.Recovery, { kind: "subagent" }>) => {
         const state = background.status === "running" ? "cancelled" : background.status
         if (recovery.kind === "shell") {
           const text =
@@ -165,17 +162,17 @@ export const layer = (options?: Options) =>
         yield* jobs.completeBackground(background.notificationID)
       })
 
-      // Returns true when this child's resume belongs to the deferred child pass.
       const recoverSubagent = Effect.fnUntraced(function* (
         background: Job.Background,
         recovery: Extract<Job.Recovery, { kind: "subagent" }>,
         suspended: ReadonlySet<SessionSchema.ID>,
-      ): Effect.fn.Return<boolean> {
+      ): Effect.fn.Return<"stale" | "owned" | "reported" | "exhausted" | "resume"> {
         const child = yield* store.get(recovery.childSessionID)
         if (!child || child.parentID !== recovery.parentSessionID || !(yield* store.get(recovery.parentSessionID))) {
           yield* jobs.completeBackground(background.notificationID)
-          return false
+          return "stale"
         }
+        if ((yield* jobs.get(background.id))?.status === "running") return "owned"
 
         // A silent child reports nothing, whether its settle ran or restart replays it, so only the
         // durable marker is left to clear.
@@ -196,16 +193,16 @@ export const layer = (options?: Options) =>
         // Finished before the crash: no settle survives to report it, so this replay is the report.
         if (background.status !== "running") {
           yield* notify(background)
-          return false
+          return "reported"
         }
-        if (yield* execution.isActive(recovery.childSessionID)) return false
+        if (yield* execution.isActive(recovery.childSessionID)) return "owned"
         if (!(yield* prepareResume(recovery.childSessionID))) {
           // prepareResume terminalizes on the bus rather than through a busy period, so no settle
           // report follows: this is the only delivery for the exhausted child.
           yield* notify({ status: "error", error: RESUME_EXHAUSTED.message })
-          return false
+          return "exhausted"
         }
-        return true
+        return "resume"
       })
 
       const resumeChild = Effect.fnUntraced(function* (
@@ -258,6 +255,9 @@ export const layer = (options?: Options) =>
             [...(yield* store.listSuspended()), ...children].filter((sessionID) => !active.has(sessionID)),
           )
           yield* store.releaseChildClaims(children)
+          const owned = new Set<SessionSchema.ID>()
+          const exhausted = new Set<SessionSchema.ID>()
+          const reported = new Set<string>()
           const resumable: Array<{
             readonly background: Job.Background
             readonly recovery: Extract<Job.Recovery, { kind: "subagent" }>
@@ -266,10 +266,16 @@ export const layer = (options?: Options) =>
             // Admit shell outcomes before a recovered child can start its first model request.
             pending.toSorted((a, b) => Number(a.recovery.kind === "subagent") - Number(b.recovery.kind === "subagent")),
             Effect.fnUntraced(function* (background) {
-              if ((yield* jobs.get(background.id))?.status === "running") return
               const recovery = background.recovery
-              if (recovery.kind !== "subagent") return yield* recoverToolJob(background, recovery)
-              if (yield* recoverSubagent(background, recovery, suspended)) resumable.push({ background, recovery })
+              if (recovery.kind !== "subagent") {
+                if ((yield* jobs.get(background.id))?.status === "running") return
+                return yield* recoverToolJob(background, recovery)
+              }
+              const result = yield* recoverSubagent(background, recovery, suspended)
+              if (result === "owned" || result === "resume") owned.add(recovery.childSessionID)
+              if (result === "exhausted") exhausted.add(recovery.childSessionID)
+              if (result === "reported") reported.add(background.notificationID)
+              if (result === "resume") resumable.push({ background, recovery })
             }),
             { discard: true },
           )
@@ -289,6 +295,22 @@ export const layer = (options?: Options) =>
           yield* Effect.forEach(resumable, ({ background, recovery }) => resumeChild(background, recovery), {
             discard: true,
           })
+          // Held direct prompts have no job marker. Wake them only after cancellations are admitted,
+          // parents are counted, and recovered jobs own their reports. A wake preserves the held
+          // answer boundary without retrying an interrupted model execution.
+          yield* Effect.forEach(
+            yield* store.listHeldChildren(),
+            Effect.fnUntraced(function* (held) {
+              if (owned.has(held.sessionID) || (yield* execution.isActive(held.sessionID))) return
+              if (exhausted.has(held.sessionID) || reported.has(held.report.notificationID))
+                return yield* store.releaseReport(held.sessionID)
+              const child = yield* store.get(held.sessionID)
+              if (!child || !SubagentMetadata.notify(child.metadata) || !(yield* store.get(held.parentID)))
+                return yield* store.releaseReport(held.sessionID)
+              yield* execution.wake(held.sessionID)
+            }),
+            { discard: true },
+          )
         }),
       })
     }),

packages/core/test/session-execution.test.ts

@@ -1042,6 +1042,7 @@ describe("SessionRestart background recovery", () => {
   it.effect("resumes a background subagent and notifies its parent exactly once", () =>
     Effect.gen(function* () {
       const database = yield* Database.Service
+      const store = yield* SessionStore.Service
       const jobs = yield* Job.Service
       const bus = yield* Bus.Service
       const parent = Session.ID.make("ses_subagent_recovery_parent")
@@ -1064,6 +1065,8 @@ describe("SessionRestart background recovery", () => {
       yield* jobs.background(child)
       const marker = (yield* jobs.pendingBackground)[0]
       if (!marker) return yield* Effect.die("background record missing")
+      const heldID = SessionMessage.ID.create()
+      yield* store.holdReport(child, { notificationID: heldID, since: undefined })
 
       const resumed = yield* Deferred.make<void>()
       const release = yield* Deferred.make<void>()
@@ -1119,11 +1122,193 @@ describe("SessionRestart background recovery", () => {
         },
       ])
       yield* pendingEmpty(restarted)
+      expect(yield* store.listHeldChildren()).toEqual([])
       yield* restart.resumeSuspendedSessions
       expect(yield* SessionInbox.list(database.db, parent)).toHaveLength(1)
     }),
   )
 
+  for (const state of ["cancellation", "no-answer", "stale", "finished", "silent", "orphan"] as const) {
+    it.effect(`recovers a held child independently of its ${state} marker state`, () =>
+      Effect.gen(function* () {
+        const database = yield* Database.Service
+        const store = yield* SessionStore.Service
+        const jobs = yield* Job.Service
+        const bus = yield* Bus.Service
+        const parent = Session.ID.make("ses_held_parent")
+        const child = Session.ID.make("ses_held_child")
+        const other = Session.ID.make("ses_held_other")
+        if (state !== "orphan")
+          yield* seedSessions(
+            database,
+            [parent],
+            state === "cancellation" ? { time_suspended: Date.now(), resume_attempts: 2 } : {},
+          )
+        yield* seedSessions(database, [child], {
+          parent_id: parent,
+          time_suspended: Date.now(),
+          metadata: state === "silent" ? { "subagent.notify": false } : {},
+        })
+        yield* answer(bus, child, "Previously reported answer")
+        const old = (yield* store.context(child)).findLast((message) => message.type === "assistant")
+        if (!old) return yield* Effect.die("old answer missing")
+        const held = { notificationID: SessionMessage.ID.create(), since: { at: 0, answer: old.id } }
+        yield* store.holdReport(child, held)
+        if (state === "stale" || state === "finished") {
+          if (state === "stale") yield* seedSessions(database, [other])
+          yield* jobs.start({
+            id: child,
+            type: "subagent",
+            recovery: {
+              kind: "subagent",
+              parentSessionID: state === "stale" ? other : parent,
+              childSessionID: child,
+              agent: "explore",
+              description: "Earlier inspection",
+            },
+            run: state === "stale" ? Effect.never : Effect.succeed("Previously reported answer"),
+          })
+          if (state === "finished") yield* jobs.wait({ id: child })
+          yield* jobs.background(child)
+          yield* answer(bus, child, "Owed answer")
+        }
+        if (state === "cancellation")
+          yield* seedBackground(jobs, child, [{ id: "held-shell", shellID: "held-shell", command: "bun dev" }])
+
+        const failed: Session.ID[] = []
+        yield* bus.project(SessionEvent.Execution.Failed, (event) =>
+          Effect.sync(() => void failed.push(event.data.sessionID)),
+        )
+        const drained: Array<{ sessionID: Session.ID; force: boolean }> = []
+        const scope = yield* Scope.make()
+        yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
+        const restarted = yield* Job.make.pipe(Scope.provide(scope))
+        const context = yield* buildExecution(
+          scope,
+          (input) =>
+            Effect.gen(function* () {
+              drained.push({ sessionID: input.sessionID, force: input.force })
+              if (input.sessionID !== child) return
+              expect(yield* store.heldReport(child)).toEqual(held)
+              if (state !== "cancellation") return
+              // Parent budget accounting and admit-only cancellation both precede the held wake.
+              expect(failed).toEqual([parent])
+              expect(yield* SessionInbox.list(database.db, child)).toMatchObject([
+                { payload: { text: expect.stringContaining("server restarted") } },
+              ])
+              expect(yield* SessionInbox.promote(database.db, bus, child, "input")).toBe(1)
+              yield* answer(bus, child, "Owed answer")
+            }),
+          { maxAttempts: 2 },
+          restarted,
+        )
+        const restart = Context.get(context, SessionRestart.Service)
+        const execution = Context.get(context, SessionExecution.Service)
+        yield* restart.resumeSuspendedSessions
+        yield* execution.awaitIdle(child)
+        if (state !== "orphan") yield* execution.awaitIdle(parent)
+        expect(drained.filter((input) => input.sessionID === child)).toEqual(
+          state === "silent" || state === "orphan" ? [] : [{ sessionID: child, force: false }],
+        )
+        expect(yield* store.listHeldChildren()).toEqual([])
+        expect(yield* attempts(database, child)).toBe(0)
+        expect(yield* restarted.pendingBackground).toEqual([])
+        const notices = yield* SessionInbox.list(database.db, parent)
+        expect(notices).toHaveLength(state === "finished" ? 2 : state === "stale" || state === "cancellation" ? 1 : 0)
+        if (state === "stale" || state === "finished" || state === "cancellation") {
+          expect(notices.find((notice) => notice.id === held.notificationID)).toMatchObject({
+            payload: { text: expect.stringContaining("Owed answer") },
+          })
+          expect(notices.find((notice) => notice.id === held.notificationID)?.payload).not.toMatchObject({
+            text: expect.stringContaining("Previously reported answer"),
+          })
+        }
+        if (state === "stale") expect(yield* SessionInbox.list(database.db, other)).toEqual([])
+        const count = drained.length
+        yield* restart.resumeSuspendedSessions
+        expect(drained).toHaveLength(count)
+        expect(yield* SessionInbox.list(database.db, parent)).toEqual(notices)
+      }),
+    )
+  }
+
+  for (const owner of ["job", "execution"] as const) {
+    it.effect(`leaves a held child's report to its live ${owner}`, () =>
+      Effect.gen(function* () {
+        const database = yield* Database.Service
+        const store = yield* SessionStore.Service
+        const bus = yield* Bus.Service
+        const parent = Session.ID.make("ses_held_owned_parent")
+        const child = Session.ID.make("ses_held_owned_child")
+        yield* seedSessions(database, [parent])
+        yield* seedSessions(database, [child], { parent_id: parent })
+        const held = { notificationID: SessionMessage.ID.create(), since: undefined }
+        yield* store.holdReport(child, held)
+        const scope = yield* Scope.make()
+        yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
+        const jobs = yield* Job.make.pipe(Scope.provide(scope))
+        const release = yield* Deferred.make<void>()
+        const started = yield* Deferred.make<void>()
+        const drained: Session.ID[] = []
+        const context = yield* buildExecution(
+          scope,
+          ({ sessionID }) =>
+            Effect.gen(function* () {
+              drained.push(sessionID)
+              if (sessionID !== child) return
+              yield* Deferred.succeed(started, undefined)
+              yield* Deferred.await(release)
+              yield* answer(bus, child, "Owned answer")
+            }),
+          undefined,
+          jobs,
+        )
+        const execution = Context.get(context, SessionExecution.Service)
+        const restart = Context.get(context, SessionRestart.Service)
+        const notificationID = owner === "job" ? SessionMessage.ID.create() : held.notificationID
+        if (owner === "job") {
+          yield* jobs.start({
+            id: child,
+            type: "subagent",
+            notificationID,
+            recovery: {
+              kind: "subagent",
+              parentSessionID: parent,
+              childSessionID: child,
+              agent: "explore",
+              description: "Live inspection",
+            },
+            // A registered job owns the report even before it starts an execution.
+            run: Deferred.await(release).pipe(Effect.andThen(execution.resume(child)), Effect.as("Owned answer")),
+          })
+          yield* jobs.background(child)
+        }
+        if (owner === "execution") {
+          yield* execution.wake(child)
+          yield* Deferred.await(started)
+        }
+        yield* restart.resumeSuspendedSessions
+        expect(yield* store.heldReport(child)).toEqual(held)
+        expect(drained).toEqual(owner === "job" ? [] : [child])
+        expect(yield* SessionInbox.list(database.db, parent)).toEqual([])
+        yield* Deferred.succeed(release, undefined)
+        if (owner === "job") yield* jobs.wait({ id: child })
+        yield* execution.awaitIdle(child)
+        yield* execution.awaitIdle(parent)
+        expect(drained.filter((id) => id === child)).toHaveLength(1)
+        expect(yield* store.listHeldChildren()).toEqual([])
+        expect(yield* SessionInbox.list(database.db, parent)).toMatchObject([
+          { id: notificationID, payload: { text: expect.stringContaining("Owned answer") } },
+        ])
+        // Replay the now-finished job's marker with the same ID as settlement.
+        yield* restart.resumeSuspendedSessions
+        yield* execution.awaitIdle(parent)
+        expect(yield* SessionInbox.list(database.db, parent)).toHaveLength(1)
+        expect(yield* jobs.pendingBackground).toEqual([])
+      }),
+    )
+  }
+
   it.effect("keeps a resumed child's marker when shutdown interrupts it again", () =>
     Effect.gen(function* () {
       const database = yield* Database.Service
@@ -1275,6 +1460,7 @@ describe("SessionRestart background recovery", () => {
   it.effect("delivers a subagent result persisted before restart without rerunning the child", () =>
     Effect.gen(function* () {
       const database = yield* Database.Service
+      const store = yield* SessionStore.Service
       const jobs = yield* Job.Service
       const parent = Session.ID.make("ses_subagent_completed_parent")
       const child = Session.ID.make("ses_subagent_completed_child")
@@ -1296,6 +1482,9 @@ describe("SessionRestart background recovery", () => {
       yield* jobs.background(child)
       yield* Deferred.succeed(complete, "Recovered result")
       yield* jobs.wait({ id: child })
+      const marker = (yield* jobs.pendingBackground)[0]
+      if (!marker) return yield* Effect.die("background record missing")
+      yield* store.holdReport(child, { notificationID: marker.notificationID, since: undefined })
 
       const parentWoken = yield* Deferred.make<void>()
       const drained: Session.ID[] = []
@@ -1316,20 +1505,28 @@ describe("SessionRestart background recovery", () => {
 
       expect(drained).toEqual([parent])
       expect(yield* SessionInbox.list(database.db, parent)).toMatchObject([
-        { payload: { text: expect.stringContaining("Recovered result"), metadata: { state: "completed" } } },
+        {
+          id: marker.notificationID,
+          payload: { text: expect.stringContaining("Recovered result"), metadata: { state: "completed" } },
+        },
       ])
       expect(yield* restarted.pendingBackground).toEqual([])
+      expect(yield* store.listHeldChildren()).toEqual([])
+      yield* Context.get(context, SessionRestart.Service).resumeSuspendedSessions
+      expect(yield* SessionInbox.list(database.db, parent)).toHaveLength(1)
     }),
   )
 
   it.effect("clears a finished silent child's marker without reporting it", () =>
     Effect.gen(function* () {
       const database = yield* Database.Service
+      const store = yield* SessionStore.Service
       const jobs = yield* Job.Service
       const parent = Session.ID.make("ses_subagent_silent_parent")
       const child = Session.ID.make("ses_subagent_silent_child")
       yield* seedSessions(database, [parent])
       yield* seedSessions(database, [child], { parent_id: parent, metadata: { "subagent.notify": false } })
+      yield* store.holdReport(child, { notificationID: SessionMessage.ID.create(), since: undefined })
       const complete = yield* Deferred.make<string>()
       yield* jobs.start({
         id: child,
@@ -1363,6 +1560,7 @@ describe("SessionRestart background recovery", () => {
       expect(drained).toEqual([])
       expect(yield* SessionInbox.list(database.db, parent)).toEqual([])
       expect(yield* restarted.pendingBackground).toEqual([])
+      expect(yield* store.listHeldChildren()).toEqual([])
     }),
   )
 
@@ -1588,11 +1786,13 @@ describe("SessionRestart background recovery", () => {
   it.effect("terminalizes a recovered subagent that exhausts its resume budget", () =>
     Effect.gen(function* () {
       const database = yield* Database.Service
+      const store = yield* SessionStore.Service
       const jobs = yield* Job.Service
       const parent = Session.ID.make("ses_subagent_exhausted_parent")
       const child = Session.ID.make("ses_subagent_exhausted_child")
       yield* seedSessions(database, [parent])
       yield* seedSessions(database, [child], { parent_id: parent, time_suspended: Date.now(), resume_attempts: 2 })
+      yield* store.holdReport(child, { notificationID: SessionMessage.ID.create(), since: undefined })
       yield* jobs.start({
         id: child,
         type: "subagent",
@@ -1635,6 +1835,9 @@ describe("SessionRestart background recovery", () => {
         },
       ])
       expect(yield* restarted.pendingBackground).toEqual([])
+      expect(yield* store.listHeldChildren()).toEqual([])
+      yield* Context.get(context, SessionRestart.Service).resumeSuspendedSessions
+      expect(yield* SessionInbox.list(database.db, parent)).toHaveLength(1)
     }),
   )
 })

packages/core/test/subagent-lifetime.test.ts

@@ -1,13 +1,16 @@
 import { describe, expect } from "bun:test"
-import { Deferred, Effect, Layer, Stream } from "effect"
+import { Context, Deferred, Effect, Exit, Layer, Scope, Stream } from "effect"
 import { LanguageModel } from "@opencode/ai/effect"
 import { OpenAIChat } from "@opencode/ai/protocols"
 import { TestLLM } from "@opencode/ai/testing"
 import { Agent } from "@opencode/core/agent"
+import { Bus } from "@opencode/core/bus"
+import { Database } from "@opencode/core/database/database"
 import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder"
 import { LayerNodePlatform } from "@opencode/core/effect/app-node-platform"
 import { Watcher } from "@opencode/core/filesystem/watcher"
 import { Job } from "@opencode/core/job"
+import { Instance } from "@opencode/core/instance/service"
 import { LocationServiceMap } from "@opencode/core/location-service-map"
 import { Model } from "@opencode/core/model"
 import { Plugin } from "@opencode/core/plugin"
@@ -15,6 +18,8 @@ import { Provider } from "@opencode/core/provider"
 import { AbsolutePath } from "@opencode/core/schema"
 import { Session } from "@opencode/core/session"
 import { SessionExecution } from "@opencode/core/session/execution"
+import { SessionRestart } from "@opencode/core/session/execution/restart"
+import { SessionInbox } from "@opencode/core/session/inbox"
 import { SessionMessage } from "@opencode/core/session/message"
 import { SessionRunnerModel } from "@opencode/core/session/runner/model"
 import { SessionSchema } from "@opencode/core/session/schema"
@@ -41,6 +46,10 @@ const it = testEffect(
         SubagentJob.node,
         SessionExecution.node,
         LocationServiceMap.node,
+        Bus.node,
+        Database.node,
+        Instance.node,
+        SessionInbox.node,
       ]),
       [
         Global.node.replace(tempGlobalLayer),
@@ -399,6 +408,68 @@ describe("subagent observer lifetime", () => {
     )
   }
 
+  it.live("restart recovers a directly prompted child's durable hold after cancelling its shell", () =>
+    Effect.gen(function* () {
+      const store = yield* SessionStore.Service
+      const llm = yield* TestLLM.Test
+      const f = yield* family()
+      const before = yield* Scope.make()
+      yield* Effect.addFinalizer(() => Scope.close(before, Exit.void))
+      const oldJobs = yield* Job.make.pipe(Scope.provide(before))
+      const old = yield* Layer.buildWithScope(
+        SessionExecution.layer.pipe(Layer.provide(Layer.succeed(Job.Service, oldJobs))),
+        before,
+      )
+      const oldExecution = Context.get(old, SessionExecution.Service)
+      yield* oldJobs.start({
+        id: "restart-shell",
+        type: "shell",
+        recovery: { kind: "shell", sessionID: f.child.id, shellID: "restart-shell", command: "bun dev" },
+        run: Effect.never,
+      })
+      yield* oldJobs.background("restart-shell")
+      yield* llm.push(TestLLM.text("first", "first"), TestLLM.text("after restart", "after"))
+      yield* f.sessions.prompt({ sessionID: f.child.id, text: "Start the dev server", resume: false })
+      yield* oldExecution.wake(f.child.id)
+      yield* oldExecution.awaitIdle(f.child.id)
+      const held = yield* store.heldReport(f.child.id)
+      expect(held).toBeDefined()
+      expect(yield* reports(f.parent.id)).toEqual([])
+      yield* Scope.close(before, Exit.void)
+      expect(yield* store.heldReport(f.child.id)).toEqual(held)
+
+      const after = yield* Scope.make()
+      yield* Effect.addFinalizer(() => Scope.close(after, Exit.void))
+      const jobs = yield* Job.make.pipe(Scope.provide(after))
+      const context = yield* Layer.buildWithScope(
+        SessionRestart.layer().pipe(
+          Layer.provideMerge(SessionExecution.layer),
+          Layer.provide(Layer.succeed(Job.Service, jobs)),
+        ),
+        after,
+      )
+      const restart = Context.get(context, SessionRestart.Service)
+      const execution = Context.get(context, SessionExecution.Service)
+      yield* restart.resumeSuspendedSessions
+      yield* execution.awaitIdle(f.child.id)
+      yield* execution.awaitIdle(f.parent.id)
+      const notices = yield* reports(f.parent.id)
+      expect(notices).toHaveLength(1)
+      expect(notices[0]).toMatchObject({ id: held?.notificationID, metadata: { childID: f.child.id } })
+      expect(notices[0]?.text).toContain("after restart")
+      expect(notices[0]?.text).not.toContain("first")
+      expect(yield* store.listHeldChildren()).toEqual([])
+      expect(yield* jobs.pendingBackground).toEqual([])
+      expect(yield* llm.requests()).toHaveLength(3)
+      expect((yield* f.sessions.context(f.child.id)).filter((message) => message.type === "synthetic")).toMatchObject([
+        { text: expect.stringContaining("server restarted") },
+      ])
+      yield* restart.resumeSuspendedSessions
+      expect(yield* reports(f.parent.id)).toEqual(notices)
+      expect(yield* llm.requests()).toHaveLength(3)
+    }),
+  )
+
   it.live("a silent background grandchild job releases its parent's held report", () =>
     Effect.gen(function* () {
       const jobs = yield* Job.Service