From 23eb65cc19c9aa7125ef6bed4e50ba428aa6d785 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 5 Nov 2026 00:23:30 -0800 Subject: [PATCH] fix(core): recover durable held subagent reports after restart --- .../core/src/session/execution/restart.ts | 50 +++-- packages/core/test/session-execution.test.ts | 205 +++++++++++++++++- packages/core/test/subagent-lifetime.test.ts | 73 ++++++- 3 files changed, 312 insertions(+), 16 deletions(-) diff --git a/packages/core/src/session/execution/restart.ts b/packages/core/src/session/execution/restart.ts index 56e6d622e15..aa5dc8d65ac 100644 --- a/packages/core/src/session/execution/restart.ts +++ b/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, - ) => { + const restartMessage = (background: Job.Background, recovery: Exclude) => { 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, suspended: ReadonlySet, - ): Effect.fn.Return { + ): 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() + const exhausted = new Set() + const reported = new Set() const resumable: Array<{ readonly background: Job.Background readonly recovery: Extract @@ -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 }, + ) }), }) }), diff --git a/packages/core/test/session-execution.test.ts b/packages/core/test/session-execution.test.ts index 73d2d8ada43..cfbee9069da 100644 --- a/packages/core/test/session-execution.test.ts +++ b/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() const release = yield* Deferred.make() @@ -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() + const started = yield* Deferred.make() + 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() 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() 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) }), ) }) diff --git a/packages/core/test/subagent-lifetime.test.ts b/packages/core/test/subagent-lifetime.test.ts index da905661b94..f6516a11c9e 100644 --- a/packages/core/test/subagent-lifetime.test.ts +++ b/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