mirror of
https://github.com/anomalyco/opencode.git
synced 2026-11-15 20:52:01 +03:00
refactor(core): remove leftover foreground subagent and reattach code (#63392)
This commit is contained in:
committed by
GitHub
Unverified
parent
41cc8d6b9d
commit
23dd370ab5
@@ -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<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>
|
||||
@@ -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<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.
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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<Session.Interface, "synthetic">
|
||||
|
||||
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<Job.Recovery, { kind: "shell" }>,
|
||||
notices: Session.Interface,
|
||||
notices: Notices,
|
||||
reattached: Set<string>,
|
||||
) {
|
||||
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<string>,
|
||||
) {
|
||||
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<Job.Recovery, { kind: "subagent" }>,
|
||||
result: Pick<Job.Background, "status" | "output" | "error">,
|
||||
@@ -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))) {
|
||||
@@ -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 },
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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>
|
||||
}
|
||||
|
||||
@@ -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
|
||||
@@ -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 },
|
||||
|
||||
@@ -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
|
||||
@@ -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)),
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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<Session.Interface, "synthetic">,
|
||||
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<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({
|
||||
@@ -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<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")
|
||||
|
||||
@@ -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<void>()
|
||||
@@ -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<string>()
|
||||
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<void>()
|
||||
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<void>()
|
||||
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
|
||||
|
||||
@@ -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" },
|
||||
])
|
||||
}),
|
||||
|
||||
@@ -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<void>()
|
||||
const released = yield* Deferred.make<void>()
|
||||
const cancelStarted = yield* Deferred.make<void>()
|
||||
const cancelReleased = yield* Deferred.make<void>()
|
||||
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", () => {
|
||||
|
||||
Reference in New Issue
Block a user