Visitar URL original
refactor(core): remove leftover foreground subagent and reattach code by jlongster · Pull Request #63392 · anomalyco/opencode · GitHub
Skip to content

refactor(core): remove leftover foreground subagent and reattach code #63392

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rekram1-node merged 1 commit into from
Nov 11, 2026
Merged

refactor(core): remove leftover foreground subagent and reattach code #63392

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 5 additions & 17 deletions packages/core/src/job.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -107,7 +106,7 @@ type BackgroundResult = {
backgrounded?: Deferred.Deferred<Info>
}

type StartResult = { info: Info } | { info: Info; job: Active } | { info: Info; backgrounded: Deferred.Deferred<Info> }
type StartResult = { info: Info } | { info: Info; job: Active }

type BlockWait = {
done: Deferred.Deferred<Info>
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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)]
Expand All @@ -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
Expand Down Expand Up @@ -649,7 +637,9 @@ export const make = Effect.gen(function* () {
return yield* SynchronizedRef.modify(state.jobs, (jobs): readonly [SubagentReportState, Map<string, Active>] => {
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.
Expand All @@ -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",
Expand Down
9 changes: 2 additions & 7 deletions packages/core/src/session/background.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
3 changes: 1 addition & 2 deletions packages/core/src/session/execution.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
37 changes: 10 additions & 27 deletions packages/core/src/session/execution/restart.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 =
Expand All @@ -41,6 +42,8 @@ export interface Options {

const DEFAULT_MAX_ATTEMPTS = 10

type Notices = Pick<Session.Interface, "synthetic">

export interface Interface {
/**
* Resumes Sessions whose execution claim was never released — turns orphaned
Expand Down Expand Up @@ -214,7 +217,7 @@ export const layer = (options?: Options) =>
const recoverShellJob = Effect.fnUntraced(function* (
background: Job.Background,
recovery: Extract<Job.Recovery, { kind: "shell" }>,
notices: Session.Interface,
notices: Notices,
reattached: Set<string>,
) {
const shell = background.status === "running" ? yield* shellOf(recovery.sessionID) : undefined
Expand All @@ -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<string>,
) {
const calls = (yield* SessionHistory.incompleteAssistants(db, sessionID)).flatMap((message) =>
Expand Down Expand Up @@ -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<Job.Recovery, { kind: "subagent" }>,
result: Pick<Job.Background, "status" | "output" | "error">,
Expand Down Expand Up @@ -328,7 +331,7 @@ export const layer = (options?: Options) =>
const recoverSubagent = Effect.fnUntraced(function* (
background: Job.Background,
recovery: Extract<Job.Recovery, { kind: "subagent" }>,
notices: Session.Interface,
notices: Notices,
): Effect.fn.Return<boolean> {
const child = yield* store.get(recovery.childSessionID)
if (!child || child.parentID !== recovery.parentSessionID || !(yield* store.get(recovery.parentSessionID))) {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 },
Expand Down
6 changes: 0 additions & 6 deletions packages/core/src/session/execution/subagent-report.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
3 changes: 0 additions & 3 deletions packages/core/src/session/subagent-completion.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@ export interface SyntheticSession<E = never> {
readonly text: string
readonly description?: string
readonly metadata?: SessionSchema.Metadata
readonly resume?: boolean
}) => Effect.Effect<unknown, E>
}

Expand Down Expand Up @@ -56,7 +55,6 @@ export const deliver = Effect.fnUntraced(function* <E>(
jobs: Pick<Job.Interface, "completeBackground">,
input: Pick<Job.Info, "status" | "output" | "error" | "notificationID"> & {
recovery: Extract<Job.Recovery, { kind: "subagent" }>
resume?: boolean
},
) {
if (input.status === "running") return
Expand All @@ -70,7 +68,6 @@ export const deliver = Effect.fnUntraced(function* <E>(
yield* sessions.synthetic({
...(input.notificationID ? { id: input.notificationID } : {}),
sessionID: recovery.parentSessionID,
...(input.resume === false ? { resume: false } : {}),
description: recovery.description,
text: `<subagent sessionID="${recovery.childSessionID}" state="${input.status}" description="${recovery.description}">\n${body}\n</subagent>`,
metadata: { source: "subagent", childID: recovery.childSessionID, agent: recovery.agent, state: input.status },
Expand Down
52 changes: 27 additions & 25 deletions packages/core/src/session/subagent-job.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,32 @@ export interface Interface {

export class Service extends Context.Service<Service, Interface>()("@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
Expand All @@ -22,31 +48,7 @@ const make = Effect.gen(function* () {
const context = yield* Effect.context<never>()

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)),
})
})

Expand Down
6 changes: 2 additions & 4 deletions packages/core/src/tool/plugin/image-generate.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 },
Expand All @@ -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)
Expand Down
15 changes: 5 additions & 10 deletions packages/core/src/tool/plugin/shell.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,23 +122,20 @@ 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<Session.Interface, "synthetic">,
jobs: Job.Interface,
input: {
readonly sessionID: SessionSchema.ID
readonly id: string
readonly shellID: string
readonly command: string
readonly settled: Deferred.Deferred<Output>
// 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({
Expand All @@ -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)
Expand All @@ -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
},
Expand All @@ -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 })
Expand Down Expand Up @@ -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<Session.Interface, "synthetic">
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")
Expand Down
Loading