Skip to content

Commit 80da76e

Browse files
cevrcevr
committed
fix(core): a subagent waiting on its own background work is not finished
A subagent that backgrounds a shell (or a subagent of its own) and ends its turn to wait went idle, and the subagent job settled on that first idle: the parent got the 'waiting' answer and never the one written after the notice woke the child. Job now tracks each backgrounded job's notice until its owner admits it (undelivered / awaitDelivered; a removed Session releases its notices), and the subagent job (live and restart-recovered) waits until the child owns no undelivered notice and the turns they woke have gone idle.
1 parent ccd7252 commit 80da76e

6 files changed

Lines changed: 400 additions & 2 deletions

File tree

‎packages/core/src/job.ts‎

Lines changed: 55 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -68,8 +68,15 @@ type Active = {
6868
recovery?: Recovery
6969
}
7070

71+
type Notice = {
72+
recovery: Recovery
73+
delivered: Deferred.Deferred<void>
74+
}
75+
7176
type State = {
7277
jobs: SynchronizedRef.SynchronizedRef<Map<string, Active>>
78+
/** Backgrounded work whose completion notice its owner has not admitted yet, by notice ID. */
79+
notices: Map<SessionMessage.ID, Notice>
7380
scope: Scope.Scope
7481
}
7582

@@ -126,6 +133,11 @@ export type BlockInput = {
126133

127134
export type BlockResult = { type: "finished"; info: Info } | { type: "backgrounded"; info: Info }
128135

136+
export type Undelivered = {
137+
notificationID: SessionMessage.ID
138+
recovery: Recovery
139+
}
140+
129141
export type BackgroundAllInput = {
130142
sessionID: SessionSchema.ID
131143
type?: string
@@ -142,6 +154,13 @@ export interface Interface {
142154
readonly cancelSession: (sessionID: SessionSchema.ID) => Effect.Effect<Info[]>
143155
readonly pendingBackground: Effect.Effect<readonly Background[]>
144156
readonly runningBackground: Effect.Effect<readonly Recovery[]>
157+
/**
158+
* Backgrounded work whose completion notice has not been admitted yet, running or not:
159+
* each notice will still wake its owner Session.
160+
*/
161+
readonly undelivered: Effect.Effect<readonly Undelivered[]>
162+
/** Resolves once the notice is admitted or its Session is removed; at once for an unknown notice. */
163+
readonly awaitDelivered: (notificationID: SessionMessage.ID) => Effect.Effect<void>
145164
readonly completeBackground: (notificationID: SessionMessage.ID) => Effect.Effect<void>
146165
}
147166

@@ -205,6 +224,7 @@ export const make = Effect.gen(function* () {
205224
const db = (yield* Database.Service).db
206225
const state: State = {
207226
jobs: yield* SynchronizedRef.make(new Map()),
227+
notices: new Map(),
208228
scope: yield* Scope.Scope,
209229
}
210230

@@ -403,13 +423,16 @@ export const make = Effect.gen(function* () {
403423
})
404424

405425
const markBackground = Effect.fnUntraced(function* (job: Active) {
426+
const notificationID = job.recovery ? (job.info.notificationID ?? SessionMessage.ID.create()) : undefined
427+
if (job.recovery && notificationID && !state.notices.has(notificationID))
428+
state.notices.set(notificationID, { recovery: job.recovery, delivered: yield* Deferred.make<void>() })
406429
const next = {
407430
...job,
408431
isBackgrounded: true,
409432
blockingSessions: new Map<SessionSchema.ID, number>(),
410433
info: {
411434
...job.info,
412-
...(job.recovery ? { notificationID: job.info.notificationID ?? SessionMessage.ID.create() } : {}),
435+
...(notificationID ? { notificationID } : {}),
413436
},
414437
}
415438
yield* persistBackground(next)
@@ -498,6 +521,16 @@ export const make = Effect.gen(function* () {
498521
.where(or(eq(JobTable.session_id, sessionID), eq(JobTable.child_session_id, sessionID)))
499522
.run()
500523
.pipe(Effect.orDie)
524+
// A removed Session admits no notice, so nothing may keep waiting on one.
525+
yield* Effect.forEach(
526+
[...state.notices].filter(
527+
([, notice]) =>
528+
ownerSessionID(notice.recovery) === sessionID ||
529+
(notice.recovery.kind === "subagent" && notice.recovery.childSessionID === sessionID),
530+
),
531+
([notificationID]) => deliver(notificationID),
532+
{ discard: true },
533+
)
501534
yield* SynchronizedRef.update(state.jobs, (jobs) => {
502535
const next = new Map(jobs)
503536
for (const [id, job] of jobs) {
@@ -536,6 +569,24 @@ export const make = Effect.gen(function* () {
536569
Effect.withSpan("Job.runningBackground"),
537570
)
538571

572+
const undelivered: Interface["undelivered"] = Effect.sync(() =>
573+
[...state.notices].map(([notificationID, notice]) => ({ notificationID, recovery: notice.recovery })),
574+
).pipe(Effect.withSpan("Job.undelivered"))
575+
576+
const awaitDelivered: Interface["awaitDelivered"] = (notificationID) =>
577+
Effect.suspend(() => {
578+
const notice = state.notices.get(notificationID)
579+
return notice ? Deferred.await(notice.delivered) : Effect.void
580+
})
581+
582+
const deliver = (notificationID: SessionMessage.ID) =>
583+
Effect.suspend(() => {
584+
const notice = state.notices.get(notificationID)
585+
if (!notice) return Effect.void
586+
state.notices.delete(notificationID)
587+
return Deferred.succeed(notice.delivered, undefined)
588+
})
589+
539590
const completeBackground: Interface["completeBackground"] = Effect.fn("Job.completeBackground")((notificationID) =>
540591
SynchronizedRef.updateEffect(state.jobs, (jobs) =>
541592
Effect.gen(function* () {
@@ -546,7 +597,7 @@ export const make = Effect.gen(function* () {
546597
next.delete(entry[0])
547598
return next
548599
}),
549-
),
600+
).pipe(Effect.andThen(deliver(notificationID))),
550601
)
551602

552603
return Service.of({
@@ -560,6 +611,8 @@ export const make = Effect.gen(function* () {
560611
cancelSession,
561612
pendingBackground,
562613
runningBackground,
614+
undelivered,
615+
awaitDelivered,
563616
completeBackground,
564617
})
565618
})

‎packages/core/src/session/background.ts‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,32 @@ export const outstanding = Effect.fnUntraced(function* (
3838
})
3939
})
4040

41+
/**
42+
* Waits until nothing `sessionID` launched in the background will wake it again: every
43+
* notice it owns is admitted and the turn each one wakes has gone idle. A notice owned by a
44+
* descendant wakes that descendant, not this Session, so it is not waited on here.
45+
*/
46+
export const awaitQuiet = Effect.fnUntraced(function* <E>(
47+
jobs: Pick<Job.Interface, "undelivered" | "awaitDelivered">,
48+
awaitIdle: (sessionID: SessionSchema.ID) => Effect.Effect<void, E>,
49+
sessionID: SessionSchema.ID,
50+
) {
51+
while (true) {
52+
const owned = (yield* jobs.undelivered).filter(
53+
(notice) =>
54+
(notice.recovery.kind === "subagent" ? notice.recovery.parentSessionID : notice.recovery.sessionID) ===
55+
sessionID,
56+
)
57+
if (owned.length === 0) return
58+
yield* Effect.forEach(owned, (notice) => jobs.awaitDelivered(notice.notificationID), {
59+
concurrency: "unbounded",
60+
discard: true,
61+
})
62+
// Admission rings the owner's doorbell before it is acknowledged, so the woken turn is already active.
63+
yield* awaitIdle(sessionID)
64+
}
65+
})
66+
4167
const descends = Effect.fnUntraced(function* (
4268
store: Pick<SessionStore.Interface, "get">,
4369
from: SessionSchema.ID,

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import { SessionSchema } from "../schema.js"
1111
import { SessionStore } from "../store.js"
1212
import { ImageNotice } from "../../image/notice.js"
1313
import { ShellResult } from "../../shell/result.js"
14+
import { SessionBackground } from "../background.js"
1415
import { SubagentCompletion } from "../subagent-completion.js"
1516

1617
const CONTINUE_AFTER_SERVER_RESTART =
@@ -201,6 +202,7 @@ export const layer = (options?: Options) =>
201202
notificationID: background.notificationID,
202203
recovery,
203204
run: execution.resume(recovery.childSessionID).pipe(
205+
Effect.andThen(SessionBackground.awaitQuiet(jobs, execution.awaitIdle, recovery.childSessionID)),
204206
Effect.andThen(store.context(recovery.childSessionID)),
205207
Effect.map((messages) => {
206208
const assistant = messages.findLast(

‎packages/core/src/session/subagent-job.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import { Node } from "@opencode/util/effect/app-node"
55
import type { LayerNode } from "@opencode/util/effect/layer-node"
66
import { Job } from "../job.js"
77
import { Session } from "../session.js"
8+
import { SessionBackground } from "./background.js"
89
import { SubagentCompletion } from "./subagent-completion.js"
910

1011
type Recovery = Extract<Job.Recovery, { kind: "subagent" }>
@@ -52,6 +53,8 @@ const make: Effect.Effect<Interface, never, Session.Service | Job.Service | Scop
5253
recovery,
5354
run: Effect.gen(function* () {
5455
yield* sessions.resume(recovery.childSessionID)
56+
// Idle is not finished while the child's own background work will still wake it.
57+
yield* SessionBackground.awaitQuiet(jobs, sessions.wait, recovery.childSessionID)
5558
const messages = yield* sessions.messages({ sessionID: recovery.childSessionID, order: "desc", limit: 20 })
5659
const assistant = messages.find(
5760
(message) =>

‎packages/core/test/job.test.ts‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -492,6 +492,51 @@ describe("Job", () => {
492492
}),
493493
)
494494

495+
it.live("keeps a background notice undelivered until its owner admits it or is removed", () =>
496+
Effect.gen(function* () {
497+
const jobs = yield* Job.Service
498+
const sessionID = SessionSchema.ID.make("ses_notice_owner")
499+
const removedID = SessionSchema.ID.make("ses_notice_removed")
500+
const finished = yield* jobs.start({
501+
id: "job_notice_finished",
502+
type: "shell",
503+
recovery: { kind: "shell", sessionID, shellID: "job_notice_finished", command: "bun test" },
504+
run: Effect.succeed("done"),
505+
})
506+
const running = yield* jobs.start({
507+
id: "job_notice_running",
508+
type: "shell",
509+
recovery: { kind: "shell", sessionID: removedID, shellID: "job_notice_running", command: "bun run dev" },
510+
run: Effect.never,
511+
})
512+
const notice = (yield* jobs.background(finished.id))?.notificationID
513+
yield* jobs.background(running.id)
514+
if (notice === undefined) return yield* Effect.die("Expected a notice ID")
515+
yield* jobs.wait({ id: finished.id })
516+
517+
// A settled job still owes its owner the notice that will wake it.
518+
expect((yield* jobs.undelivered).map((item) => item.recovery)).toMatchObject([
519+
{ sessionID, shellID: "job_notice_finished" },
520+
{ sessionID: removedID, shellID: "job_notice_running" },
521+
])
522+
const admitted = yield* jobs.awaitDelivered(notice).pipe(Effect.forkScoped({ startImmediately: true }))
523+
expect(admitted.pollUnsafe()).toBeUndefined()
524+
yield* jobs.completeBackground(notice)
525+
yield* Fiber.join(admitted).pipe(Effect.timeout("1 second"))
526+
527+
const pendingRunning = (yield* jobs.undelivered)[0]
528+
expect(pendingRunning?.recovery).toMatchObject({ sessionID: removedID })
529+
if (pendingRunning === undefined) return
530+
const released = yield* jobs
531+
.awaitDelivered(pendingRunning.notificationID)
532+
.pipe(Effect.forkScoped({ startImmediately: true }))
533+
yield* jobs.cancelSession(removedID)
534+
yield* Fiber.join(released).pipe(Effect.timeout("1 second"))
535+
expect(yield* jobs.undelivered).toEqual([])
536+
yield* jobs.awaitDelivered(SessionMessage.ID.create()).pipe(Effect.timeout("1 second"))
537+
}),
538+
)
539+
495540
it.live("migrates legacy kv job.background/* markers into the job table", () =>
496541
Effect.gen(function* () {
497542
const db = (yield* Database.Service).db

0 commit comments

Comments
 (0)