From 23dd370ab5aded4a6bd3de26de35177b24fcaeba Mon Sep 17 00:00:00 2001 From: James Long Date: Wed, 11 Nov 2026 15:52:50 -0600 Subject: [PATCH] refactor(core): remove leftover foreground subagent and reattach code (#63392) --- packages/core/src/job.ts | 22 +--- packages/core/src/session/background.ts | 9 +- packages/core/src/session/execution.ts | 3 +- .../core/src/session/execution/restart.ts | 37 ++----- .../src/session/execution/subagent-report.ts | 6 -- .../core/src/session/subagent-completion.ts | 3 - packages/core/src/session/subagent-job.ts | 52 ++++----- .../core/src/tool/plugin/image-generate.ts | 6 +- packages/core/src/tool/plugin/shell.ts | 15 +-- packages/core/test/job.test.ts | 102 +++--------------- packages/core/test/session-background.test.ts | 27 +---- packages/core/test/subagent-report.test.ts | 91 +--------------- 12 files changed, 70 insertions(+), 303 deletions(-) diff --git a/packages/core/src/job.ts b/packages/core/src/job.ts index 1491ed22de6..94252e3fa6d 100644 --- a/packages/core/src/job.ts +++ b/packages/core/src/job.ts @@ -38,7 +38,6 @@ export type Background = typeof Background.Type export type Status = Background["status"] export type SubagentReportState = - | { readonly mode: "foreground" } | { readonly mode: "background" readonly notificationID: SessionMessage.ID @@ -107,7 +106,7 @@ type BackgroundResult = { backgrounded?: Deferred.Deferred } -type StartResult = { info: Info } | { info: Info; job: Active } | { info: Info; backgrounded: Deferred.Deferred } +type StartResult = { info: Info } | { info: Info; job: Active } type BlockWait = { done: Deferred.Deferred @@ -363,14 +362,7 @@ export const make = Effect.gen(function* () { const existing = jobs.get(input.id) // A settling generation is already reporting; new work starts a fresh generation. if (existing?.info.status === "running" && !existing.settling) { - if (!input.background || existing.isBackgrounded) return [{ info: snapshot(existing) }, jobs] - // Joining live work after its start must register the durable marker before a fast - // drain can settle and classify the period as foreground. - const updated = yield* markBackground(existing) - return [ - { info: snapshot(updated), backgrounded: existing.backgrounded }, - new Map(jobs).set(input.id, updated), - ] + return [{ info: snapshot(existing) }, jobs] } const scope = yield* Scope.fork(state.scope, "parallel") // Closing the registry interrupts the work without settling it; interrupt its observers too. @@ -406,7 +398,6 @@ export const make = Effect.gen(function* () { return [{ info: snapshot(job), job }, new Map(jobs).set(input.id, job)] }), ) - if ("backgrounded" in result) yield* Deferred.succeed(result.backgrounded, result.info) if ("job" in result) yield* restore(input.run).pipe( Effect.exit, @@ -498,8 +489,6 @@ export const make = Effect.gen(function* () { const job = jobs.get(id) // Recoverable work may finish before the caller backgrounds it. if (!job || (job.info.status !== "running" && !job.recovery)) return [{}, jobs] - // A settling generation already reported as foreground; a late Ctrl+B must not revive it. - if (job.settling) return [{}, jobs] if (job.isBackgrounded) return [{ info: snapshot(job) }, jobs] const next = yield* markBackground(job) return [{ info: snapshot(next), backgrounded: job.backgrounded }, new Map(jobs).set(id, next)] @@ -519,7 +508,6 @@ export const make = Effect.gen(function* () { const next = new Map(jobs) for (const [id, job] of jobs) { if (job.info.status !== "running") continue - if (job.settling) continue if (job.isBackgrounded) continue if (input.type !== undefined && job.info.type !== input.type) continue if (!job.blockingSessions.has(input.sessionID)) continue @@ -649,7 +637,9 @@ export const make = Effect.gen(function* () { return yield* SynchronizedRef.modify(state.jobs, (jobs): readonly [SubagentReportState, Map] => { const job = jobs.get(id) // A generation reports at most once, even when an earlier attempt only observed a cancel. - if (!job || job.reported) return [{ mode: "none" }, jobs] + if (!job?.isBackgrounded || job.reported || !job.info.notificationID || job.recovery?.kind !== "subagent") { + return [{ mode: "none" }, jobs] + } const live = job.info.status === "running" && !job.settling // A cancel within this busy period already consumed the live entry, but it still belongs // here. A creation-tick interruption never bound a generation, so any cancel matches. @@ -662,8 +652,6 @@ export const make = Effect.gen(function* () { ...(job.info.status === "running" ? { settling: true } : {}), } const next = new Map(jobs).set(id, updated) - if (!job.isBackgrounded) return [{ mode: "foreground" }, next] - if (!job.info.notificationID || job.recovery?.kind !== "subagent") return [{ mode: "none" }, next] return [ { mode: "background", diff --git a/packages/core/src/session/background.ts b/packages/core/src/session/background.ts index 2471ee883b0..f0a7c920a3f 100644 --- a/packages/core/src/session/background.ts +++ b/packages/core/src/session/background.ts @@ -82,14 +82,9 @@ export const outstanding = Effect.fnUntraced(function* ( const children = new Set(items.flatMap((item) => (item.kind === "subagent" ? [item.childSessionID] : []))) for (const activeID of active) { if (activeID === sessionID || children.has(activeID)) continue - // Foreground-blocking work is the caller's inline result, and cancelled work is - // acknowledged elsewhere: neither is still running for this report. + // A cancelled child may still be draining while it cleans up. const tracked = jobs.get ? yield* jobs.get(activeID) : undefined - if ( - tracked && - ((tracked.status === "running" && tracked.notificationID === undefined) || tracked.status === "cancelled") - ) - continue + if (tracked?.status === "cancelled") continue if (!(yield* descends(store, parents, activeID, sessionID))) continue const info = yield* store.get(activeID) if (!info?.parentID) continue diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index f59066ab668..382f1f9db35 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -158,8 +158,7 @@ export const layer = Layer.effect( yield* bus.publish(SessionEvent.Execution.Interrupted, { sessionID, reason: "shutdown" }) return } - // Classify and claim the Job generation before cancelling it: the settlement commit - // owns the outbox row, and a cancel must not race a foreground classification. + // Claim the Job generation before cancelling it so the settlement commit owns its outbox row. const settlement = yield* subagentReport.prepare(sessionID, outcome) if (outcome.type === "interrupted") yield* jobs.cancel(sessionID) yield* publishTerminal(bus, sessionID, outcome, settlement.commit) diff --git a/packages/core/src/session/execution/restart.ts b/packages/core/src/session/execution/restart.ts index da8892f02bb..db69616d9c9 100644 --- a/packages/core/src/session/execution/restart.ts +++ b/packages/core/src/session/execution/restart.ts @@ -19,6 +19,7 @@ import { SessionStore } from "../store.js" import { ImageNotice } from "../../image/notice.js" import { ShellResult } from "../../shell/result.js" import { SubagentCompletion } from "../subagent-completion.js" +import { SubagentJob } from "../subagent-job.js" import { SubagentMetadata } from "../subagent-metadata.js" const CONTINUE_AFTER_SERVER_RESTART = @@ -41,6 +42,8 @@ export interface Options { const DEFAULT_MAX_ATTEMPTS = 10 +type Notices = Pick + export interface Interface { /** * Resumes Sessions whose execution claim was never released — turns orphaned @@ -214,7 +217,7 @@ export const layer = (options?: Options) => const recoverShellJob = Effect.fnUntraced(function* ( background: Job.Background, recovery: Extract, - notices: Session.Interface, + notices: Notices, reattached: Set, ) { const shell = background.status === "running" ? yield* shellOf(recovery.sessionID) : undefined @@ -238,7 +241,7 @@ export const layer = (options?: Options) => // command still running in the daemon. Settle the call as moved to the background and watch the command. const settleBackgroundedCalls = Effect.fnUntraced(function* ( sessionID: SessionSchema.ID, - notices: Session.Interface, + notices: Notices, reattached: ReadonlySet, ) { const calls = (yield* SessionHistory.incompleteAssistants(db, sessionID)).flatMap((message) => @@ -282,7 +285,7 @@ export const layer = (options?: Options) => }) const deliverSubagentSafe = Effect.fnUntraced(function* ( - notices: Session.Interface, + notices: Notices, notificationID: Job.Background["notificationID"], recovery: Extract, result: Pick, @@ -328,7 +331,7 @@ export const layer = (options?: Options) => const recoverSubagent = Effect.fnUntraced(function* ( background: Job.Background, recovery: Extract, - notices: Session.Interface, + notices: Notices, ): Effect.fn.Return { const child = yield* store.get(recovery.childSessionID) if (!child || child.parentID !== recovery.parentSessionID || !(yield* store.get(recovery.parentSessionID))) { @@ -396,12 +399,11 @@ export const layer = (options?: Options) => ) // Hold notice wakes until every suspended root has counted its attempt, so a woken // child cannot settle mid-sweep and wake its parent past that budget. - const notices: Session.Interface = { - ...sessions, + const notices: Notices = { synthetic: (input) => sessions.synthetic({ ...input, resume: false }).pipe( Effect.tap(() => { - if (input.resume === false || suspended.has(input.sessionID)) return Effect.void + if (suspended.has(input.sessionID)) return Effect.void if (!state.defer) return wake(input.sessionID) wakes.add(input.sessionID) return Effect.void @@ -465,26 +467,7 @@ export const layer = (options?: Options) => yield* Effect.forEach( resumable, Effect.fnUntraced(function* ({ background, recovery }) { - yield* jobs.start({ - id: background.id, - type: "subagent", - title: recovery.description, - notificationID: background.notificationID, - recovery, - background: true, - run: execution.resume(recovery.childSessionID).pipe( - Effect.onInterrupt(() => - jobs - .isCancelled(background.id) - .pipe( - Effect.flatMap((cancelled) => - cancelled ? execution.interrupt(recovery.childSessionID) : Effect.void, - ), - ), - ), - Effect.as(""), - ), - }) + yield* SubagentJob.start(jobs, execution, recovery, background.notificationID) suspended.delete(recovery.childSessionID) }), { discard: true }, diff --git a/packages/core/src/session/execution/subagent-report.ts b/packages/core/src/session/execution/subagent-report.ts index 2bbb96bd46b..25f90e91d1b 100644 --- a/packages/core/src/session/execution/subagent-report.ts +++ b/packages/core/src/session/execution/subagent-report.ts @@ -135,12 +135,6 @@ export const make = (deps: Deps) => { const period = periods.get(sessionID) periods.delete(sessionID) const jobState = yield* deps.jobs.prepareSubagentReport(sessionID, period?.boundGeneration) - if (jobState.mode === "foreground") { - return { - commit: () => deps.store.release(sessionID), - deliver: Effect.void, - } - } const initial = yield* deps.store.get(sessionID) if (!initial?.parentID && jobState.mode === "none") { return { diff --git a/packages/core/src/session/subagent-completion.ts b/packages/core/src/session/subagent-completion.ts index 3f03509d0e6..a06e5fdc4df 100644 --- a/packages/core/src/session/subagent-completion.ts +++ b/packages/core/src/session/subagent-completion.ts @@ -18,7 +18,6 @@ export interface SyntheticSession { readonly text: string readonly description?: string readonly metadata?: SessionSchema.Metadata - readonly resume?: boolean }) => Effect.Effect } @@ -56,7 +55,6 @@ export const deliver = Effect.fnUntraced(function* ( jobs: Pick, input: Pick & { recovery: Extract - resume?: boolean }, ) { if (input.status === "running") return @@ -70,7 +68,6 @@ export const deliver = Effect.fnUntraced(function* ( yield* sessions.synthetic({ ...(input.notificationID ? { id: input.notificationID } : {}), sessionID: recovery.parentSessionID, - ...(input.resume === false ? { resume: false } : {}), description: recovery.description, text: `\n${body}\n`, metadata: { source: "subagent", childID: recovery.childSessionID, agent: recovery.agent, state: input.status }, diff --git a/packages/core/src/session/subagent-job.ts b/packages/core/src/session/subagent-job.ts index 43bbf4e5cbf..3e0d40a982e 100644 --- a/packages/core/src/session/subagent-job.ts +++ b/packages/core/src/session/subagent-job.ts @@ -14,6 +14,32 @@ export interface Interface { export class Service extends Context.Service()("@opencode/SubagentJob") {} +export function start( + jobs: Job.Interface, + execution: SessionExecution.Interface, + recovery: Recovery, + notificationID?: Job.Background["notificationID"], +) { + const id = recovery.childSessionID + return jobs.start({ + id, + type: "subagent", + title: recovery.description, + metadata: notificationID === undefined ? {} : undefined, + recovery, + notificationID, + background: true, + run: execution.resume(id).pipe( + // Only an explicit caller cancel releases the child's claim. A server shutdown + // interrupts this job but must leave the child resumable on the next start. + Effect.onInterrupt(() => + jobs.isCancelled(id).pipe(Effect.flatMap((cancelled) => (cancelled ? execution.interrupt(id) : Effect.void))), + ), + Effect.as(""), + ), + }) +} + const make = Effect.gen(function* () { const execution = yield* SessionExecution.Service const jobs = yield* Job.Service @@ -22,31 +48,7 @@ const make = Effect.gen(function* () { const context = yield* Effect.context() return Service.of({ - start: (recovery) => - jobs - .start({ - id: recovery.childSessionID, - type: "subagent", - title: recovery.description, - metadata: {}, - recovery, - background: true, - run: execution.resume(recovery.childSessionID).pipe( - // Only an explicit caller cancel releases the child's claim. A server shutdown - // interrupts this job but must leave the child resumable on the next start. - Effect.onInterrupt(() => - jobs - .isCancelled(recovery.childSessionID) - .pipe( - Effect.flatMap((cancelled) => - cancelled ? execution.interrupt(recovery.childSessionID) : Effect.void, - ), - ), - ), - Effect.as(""), - ), - }) - .pipe(Effect.setContext(context)), + start: (recovery) => start(jobs, execution, recovery).pipe(Effect.setContext(context)), }) }) diff --git a/packages/core/src/tool/plugin/image-generate.ts b/packages/core/src/tool/plugin/image-generate.ts index e03d51f23f5..726e35c9b39 100644 --- a/packages/core/src/tool/plugin/image-generate.ts +++ b/packages/core/src/tool/plugin/image-generate.ts @@ -345,9 +345,9 @@ export const Plugin = { title: input.prompt, metadata: { sessionID: context.sessionID, path: stem, model: ref }, recovery: { kind: "image", sessionID: context.sessionID, path: stem, model: ref }, + background: true, run, }) - yield* jobs.background(id) yield* notifyWhenDone({ sessionID: context.sessionID, id, notificationID, file: stem }) return { output: { id, path: stem, model: ref }, @@ -359,9 +359,7 @@ export const Plugin = { ].join("\n"), metadata: { path: stem, model: ref, jobID: id }, } - }).pipe( - Effect.mapError(toolFailure("Unable to generate image")), - ), + }).pipe(Effect.mapError(toolFailure("Unable to generate image"))), }) }) .pipe(Effect.orDie) diff --git a/packages/core/src/tool/plugin/shell.ts b/packages/core/src/tool/plugin/shell.ts index 584ba46af65..a7f215ca210 100644 --- a/packages/core/src/tool/plugin/shell.ts +++ b/packages/core/src/tool/plugin/shell.ts @@ -122,7 +122,7 @@ const backgroundResult = (info: ShellResult.Result["info"], timeout?: number) => // Tells the session how a background command ended, once its job settles. const notify = Effect.fnUntraced(function* ( - sessions: Session.Interface, + sessions: Pick, jobs: Job.Interface, input: { readonly sessionID: SessionSchema.ID @@ -130,15 +130,12 @@ const notify = Effect.fnUntraced(function* ( readonly shellID: string readonly command: string readonly settled: Deferred.Deferred - // Read at delivery; `false` admits the notice without waking the session. - readonly resume?: () => false | undefined }, ) { const info = (yield* jobs.wait({ id: input.id })).info if (!info || info.status === "running") return const output = info.status === "completed" ? yield* Deferred.await(input.settled) : undefined const text = output ? (info.output ?? "") : info.status === "error" ? (info.error ?? "Command failed") : "Cancelled" - const resume = input.resume?.() // A session deleted while its command ran has no one left to tell. yield* sessions .synthetic({ @@ -153,7 +150,6 @@ const notify = Effect.fnUntraced(function* ( text, output, }), - ...(resume === false ? { resume } : {}), }) .pipe(Effect.catchTag("Session.NotFoundError", () => Effect.void)) if (info.notificationID) yield* jobs.completeBackground(info.notificationID) @@ -166,6 +162,7 @@ const startJob = Effect.fnUntraced(function* ( input: { readonly sessionID: SessionSchema.ID readonly info: ShellSchema.Info + readonly background?: boolean readonly notificationID?: SessionMessage.ID readonly timeout?: () => number }, @@ -185,6 +182,7 @@ const startJob = Effect.fnUntraced(function* ( command: info.command, }, ...(input.notificationID ? { notificationID: input.notificationID } : {}), + background: input.background, run: Effect.gen(function* () { const result = yield* shell.result(info) if (!result.capture) return yield* new Shell.NotFoundError({ id: info.id }) @@ -225,22 +223,19 @@ export const backgroundToolResult = (info: ShellSchema.Info) => { export const reattach = Effect.fnUntraced(function* (input: { readonly shell: Shell.Interface readonly jobs: Job.Interface - readonly sessions: Session.Interface + readonly sessions: Pick readonly sessionID: SessionSchema.ID readonly info: ShellSchema.Info // The job's existing durable row, if the previous server had backgrounded it. readonly notificationID?: SessionMessage.ID - readonly resume?: () => false | undefined }) { - const { job, settled } = yield* startJob(input.jobs, input.shell, input) - yield* input.jobs.background(job.id) + const { job, settled } = yield* startJob(input.jobs, input.shell, { ...input, background: true }) const notifying = notify(input.sessions, input.jobs, { sessionID: input.sessionID, id: job.id, shellID: input.info.id, command: input.info.command, settled, - resume: input.resume, }) const current = yield* input.shell.get(input.info.id).pipe(Effect.option) if (Option.isSome(current) && current.value.status !== "running") diff --git a/packages/core/test/job.test.ts b/packages/core/test/job.test.ts index 7db1ff4e455..b9679427857 100644 --- a/packages/core/test/job.test.ts +++ b/packages/core/test/job.test.ts @@ -594,15 +594,15 @@ describe("Job", () => { type: "subagent", title: "Report work", recovery, + background: true, run: Deferred.await(latch).pipe(Effect.as("done")), }) - const background = yield* jobs.background(job.id) - if (!background?.notificationID) return yield* Effect.die("background marker missing") + if (!job.notificationID) return yield* Effect.die("background marker missing") expect(yield* jobs.currentGeneration(job.id)).toMatchObject({ running: true }) expect(yield* jobs.prepareSubagentReport(job.id)).toEqual({ mode: "background", - notificationID: background.notificationID, + notificationID: job.notificationID, recovery, startedAt: job.started_at, }) @@ -614,24 +614,22 @@ describe("Job", () => { yield* jobs.wait({ id: job.id }) // The settlement commit owns the terminal row while this generation is settling. // Settling must not overwrite it with a stale running-period value. - expect(yield* jobs.pendingBackground).toMatchObject([ - { notificationID: background.notificationID, status: "running" }, - ]) + expect(yield* jobs.pendingBackground).toMatchObject([{ notificationID: job.notificationID, status: "running" }]) yield* jobs.recordSubagentReport({ - notificationID: background.notificationID, + notificationID: job.notificationID, recovery, status: "completed", output: "done", startedAt: job.started_at, }) expect(yield* jobs.pendingBackground).toMatchObject([ - { notificationID: background.notificationID, status: "completed", output: "done" }, + { notificationID: job.notificationID, status: "completed", output: "done" }, ]) - yield* jobs.completeBackground(background.notificationID) + yield* jobs.completeBackground(job.notificationID) }), ) - it.live("refuses a late background handoff for settling foreground work and starts a fresh generation", () => + it.live("starts a fresh generation while the previous subagent report is settling", () => Effect.gen(function* () { const jobs = yield* Job.Service const latch = yield* Deferred.make() @@ -646,21 +644,22 @@ describe("Job", () => { id: recovery.childSessionID, type: "subagent", recovery, + background: true, run: Deferred.await(latch).pipe(Effect.as("old")), }) const generation = (yield* jobs.currentGeneration(job.id))?.generation if (generation === undefined) return yield* Effect.die("generation missing") const oldWait = yield* jobs.wait({ id: job.id }).pipe(Effect.forkChild({ startImmediately: true })) - expect(yield* jobs.prepareSubagentReport(job.id)).toEqual({ mode: "foreground" }) + expect(yield* jobs.prepareSubagentReport(job.id)).toMatchObject({ mode: "background" }) expect(yield* jobs.currentGeneration(job.id)).toEqual({ generation, running: false }) - expect(yield* jobs.background(job.id)).toBeUndefined() const replacementLatch = yield* Deferred.make() yield* jobs.start({ id: job.id, type: "subagent", recovery, + background: true, run: Deferred.await(replacementLatch), }) const current = yield* jobs.currentGeneration(job.id) @@ -680,40 +679,6 @@ describe("Job", () => { }), ) - it.live("keeps settling work out of a session-wide background handoff", () => - Effect.gen(function* () { - const jobs = yield* Job.Service - const parent = SessionSchema.ID.make("ses_settling_all") - const latch = yield* Deferred.make() - const first = yield* jobs.start({ - id: "job_settling_all_first", - type: "test", - run: Deferred.await(latch).pipe(Effect.as("first")), - }) - const second = yield* jobs.start({ - id: "job_settling_all_second", - type: "test", - run: Deferred.await(latch).pipe(Effect.as("second")), - }) - 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: parent }) - .pipe(Effect.forkIn(scope, { startImmediately: true })) - - expect(yield* jobs.prepareSubagentReport(first.id)).toEqual({ mode: "foreground" }) - expect(yield* jobs.backgroundAll({ sessionID: parent })).toMatchObject([{ id: second.id }]) - expect(yield* jobs.get(first.id)).toMatchObject({ status: "running" }) - expect((yield* jobs.get(first.id))?.notificationID).toBeUndefined() - - yield* Deferred.succeed(latch, undefined) - expect(yield* Fiber.join(firstWait)).toMatchObject({ type: "finished", info: { id: first.id } }) - expect(yield* Fiber.join(secondWait)).toMatchObject({ type: "backgrounded", info: { id: second.id } }) - }), - ) - it.live("matches an external cancel to its busy period through the bound generation", () => Effect.gen(function* () { const jobs = yield* Job.Service @@ -728,6 +693,7 @@ describe("Job", () => { id: recovery.childSessionID, type: "subagent", recovery, + background: true, run: Effect.never, }) const generation = (yield* jobs.currentGeneration(job.id))?.generation @@ -736,8 +702,8 @@ describe("Job", () => { yield* jobs.cancel(job.id) expect(yield* jobs.get(job.id)).toMatchObject({ status: "cancelled" }) // The cancel consumed the live entry; its matching generation still belongs to this busy period. - expect(yield* jobs.prepareSubagentReport(job.id, generation)).toEqual({ mode: "foreground" }) expect(yield* jobs.prepareSubagentReport(job.id, generation - 1)).toEqual({ mode: "none" }) + expect(yield* jobs.prepareSubagentReport(job.id, generation)).toMatchObject({ mode: "background" }) expect(yield* jobs.prepareSubagentReport(job.id)).toEqual({ mode: "none" }) }), ) @@ -819,48 +785,6 @@ describe("Job", () => { }), ) - it.live("promotes joined foreground work to background while it is still live", () => - Effect.gen(function* () { - const jobs = yield* Job.Service - const parent = SessionSchema.ID.make("ses_promote_parent") - const recovery = { - kind: "subagent" as const, - parentSessionID: parent, - childSessionID: SessionSchema.ID.make("ses_promote_child"), - agent: "explore", - description: "Promote work", - } - const latch = yield* Deferred.make() - const foreground = yield* jobs.start({ - id: recovery.childSessionID, - type: "subagent", - recovery, - run: Deferred.await(latch).pipe(Effect.as("foreground")), - }) - const waiting = yield* jobs - .block({ id: foreground.id, sessionID: parent }) - .pipe(Effect.forkChild({ startImmediately: true })) - - const promoted = yield* jobs.start({ - id: foreground.id, - type: "subagent", - recovery, - background: true, - run: Effect.die("Joined work must not run twice"), - }) - expect(promoted.notificationID).toEqual(expect.any(String)) - expect(yield* Fiber.join(waiting)).toMatchObject({ type: "backgrounded", info: { id: foreground.id } }) - expect(yield* jobs.pendingBackground).toMatchObject([{ id: foreground.id, status: "running" }]) - - yield* Deferred.succeed(latch, undefined) - yield* jobs.wait({ id: foreground.id }) - const marker = (yield* jobs.pendingBackground)[0] - if (!marker) return yield* Effect.die("background marker missing") - expect(marker.status).toBe("completed") - yield* jobs.completeBackground(marker.notificationID) - }), - ) - it.live("distinguishes caller cancellation from teardown", () => Effect.gen(function* () { const jobs = yield* Job.Service diff --git a/packages/core/test/session-background.test.ts b/packages/core/test/session-background.test.ts index 76e24e77060..14b1a7da567 100644 --- a/packages/core/test/session-background.test.ts +++ b/packages/core/test/session-background.test.ts @@ -71,38 +71,17 @@ describe("SessionBackground.outstanding", () => { }), ) - it.effect("skips created foreground-blocking and cancelled descendants when job state is available", () => + it.effect("skips cancelled descendants when job state is available", () => Effect.gen(function* () { const jobs = { runningBackground: Effect.succeed([]), get: (sessionID: string) => Effect.succeed( - sessionID === grandchild - ? info(grandchild, "running") - : sessionID === other - ? info(other, "cancelled") - : undefined, + sessionID === child ? info(child, "cancelled") : info(grandchild, "running", SessionMessage.ID.create()), ), } - // A foreground subagent's inline caller owns its result, and a cancelled one is - // acknowledged elsewhere: neither is still running for this report. - expect(yield* SessionBackground.outstanding(jobs, store, root, new Set([grandchild, other]))).toEqual([]) - - expect( - yield* SessionBackground.outstanding( - { - runningBackground: Effect.succeed([]), - get: (sessionID) => - Effect.succeed( - sessionID === grandchild ? info(grandchild, "running", SessionMessage.ID.create()) : undefined, - ), - }, - store, - root, - new Set([grandchild]), - ), - ).toEqual([ + expect(yield* SessionBackground.outstanding(jobs, store, root, new Set([child, grandchild]))).toEqual([ { kind: "subagent", sessionID: child, childSessionID: grandchild, agent: "subagent", description: "Subagent" }, ]) }), diff --git a/packages/core/test/subagent-report.test.ts b/packages/core/test/subagent-report.test.ts index 741bbd613e2..f210d5fba29 100644 --- a/packages/core/test/subagent-report.test.ts +++ b/packages/core/test/subagent-report.test.ts @@ -82,12 +82,11 @@ describe("SubagentReport settlement", () => { const parent = sessionID("ses_bgreport_parent") const child = sessionID("ses_bgreport_child") const active = sessionID("ses_bgreport_active") - const foreground = sessionID("ses_bgreport_foreground") const cancelled = sessionID("ses_bgreport_cancelled") const stopping = sessionID("ses_bgreport_stopping") yield* seedSessions(database, [parent]) yield* seedSessions(database, [child], { parent_id: parent, agent: "explore", title: "Background audit" }) - yield* seedSessions(database, [active, foreground, cancelled, stopping], { parent_id: child }) + yield* seedSessions(database, [active, cancelled, stopping], { parent_id: child }) const shell = yield* jobs.start({ id: "sh_bgreport_shell", @@ -103,18 +102,6 @@ describe("SubagentReport settlement", () => { background: true, run: Effect.never, }) - yield* jobs.start({ - id: foreground, - type: "subagent", - recovery: { - kind: "subagent", - parentSessionID: child, - childSessionID: foreground, - agent: "explore", - description: "Foreground descendant", - }, - run: Effect.never, - }) yield* jobs.start({ id: cancelled, type: "subagent", @@ -138,7 +125,7 @@ describe("SubagentReport settlement", () => { const scope = yield* Scope.make() yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)) const context = yield* buildExecution(scope, ({ sessionID }) => { - if (sessionID === active || sessionID === foreground || sessionID === cancelled) return gate + if (sessionID === active || sessionID === cancelled) return gate if (sessionID === stopping) return Deferred.succeed(stoppingStarted, undefined).pipe( Effect.andThen(Effect.never), @@ -148,7 +135,6 @@ describe("SubagentReport settlement", () => { }) const execution = Context.get(context, SessionExecution.Service) yield* execution.resume(active).pipe(Effect.forkIn(scope)) - yield* execution.resume(foreground).pipe(Effect.forkIn(scope)) yield* execution.resume(cancelled).pipe(Effect.forkIn(scope)) yield* execution.resume(stopping).pipe(Effect.forkIn(scope)) yield* Deferred.await(started) @@ -488,79 +474,6 @@ describe("SubagentReport settlement", () => { expect(yield* jobs.get(child)).toBeUndefined() }), ) - - it.effect("suppresses foreground job reports and still reports a later non-job period", () => - Effect.gen(function* () { - const database = yield* Database.Service - const jobs = yield* Job.Service - const parent = sessionID("ses_foreground_parent") - const child = sessionID("ses_foreground_child") - const cancelled = sessionID("ses_foreground_cancelled") - yield* seedSessions(database, [parent]) - yield* seedSessions(database, [child, cancelled], { parent_id: parent, agent: "explore" }) - - const started = yield* Deferred.make() - const released = yield* Deferred.make() - const cancelStarted = yield* Deferred.make() - const cancelReleased = yield* Deferred.make() - const scope = yield* Scope.make() - yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)) - const context = yield* buildExecution(scope, ({ sessionID }) => { - if (sessionID === child) - return Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(released))) - if (sessionID === cancelled) - return Deferred.succeed(cancelStarted, undefined).pipe(Effect.andThen(Deferred.await(cancelReleased))) - return Effect.void - }) - const execution = Context.get(context, SessionExecution.Service) - - // Normal foreground completion returns to the inline caller, not the parent's inbox. - const job = yield* jobs.start({ - id: child, - type: "subagent", - recovery: { - kind: "subagent", - parentSessionID: parent, - childSessionID: child, - agent: "explore", - description: "Foreground audit", - }, - run: execution.resume(child).pipe(Effect.as("inline")), - }) - yield* Deferred.await(started) - yield* seedAssistant(database, child, 100, "Inline answer") - yield* Deferred.succeed(released, undefined) - yield* jobs.wait({ id: job.id }) - expect(yield* SessionInbox.list(database.db, parent)).toEqual([]) - - // The next non-job period is a fresh busy period and reports normally. - yield* seedAssistant(database, child, 200, "Follow-up after foreground") - yield* execution.wake(child) - yield* execution.awaitIdle(child) - const reports = yield* SessionInbox.list(database.db, parent) - expect(reports).toHaveLength(1) - expect(text(reports[0])).toContain("Follow-up after foreground") - - // A cancelled foreground generation is still the inline caller's to report. - yield* jobs.start({ - id: cancelled, - type: "subagent", - recovery: { - kind: "subagent", - parentSessionID: parent, - childSessionID: cancelled, - agent: "explore", - description: "Foreground cancellation", - }, - run: execution.resume(cancelled).pipe(Effect.as("")), - }) - yield* Deferred.await(cancelStarted) - expect(yield* jobs.cancel(cancelled)).toMatchObject({ status: "cancelled" }) - expect(yield* execution.interrupt(cancelled)).toBe(true) - yield* execution.awaitIdle(cancelled) - expect(yield* SessionInbox.list(database.db, parent)).toHaveLength(1) - }), - ) }) describe("SubagentReport outbox", () => {