diff --git a/AGENTS.md b/AGENTS.md index 15f27a31245d..7ce72dcb52c9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -216,7 +216,7 @@ const table = sqliteTable("session", { - Keep `SessionExecution` process-global and Session-ID based. Its local implementation owns the process-local Session coordinator and discovers placement through `SessionStore` plus `LocationServiceMap.get(session.location)` only when a drain starts; no layer should take a Session ID. Interruption targets the active process-local ownership chain for that Session; interruption of a known but idle or locally unowned Session is a no-op, while the public API rejects an unknown Session. - Keep `SessionRunner`, model resolution, tool registry, permissions, and filesystem Location-scoped. Omitted `Location.workspaceID` means implicit-local placement; explicit workspace identity remains reserved for future placement semantics. - Each step is one explicit `llm.stream` call that continues from reloaded history. Do not add an in-memory tool loop. -- Keep local Session drains process-local until clustering is implemented. `SessionRunCoordinator` joins explicit same-Session resumes, coalesces prompt wakeups, and allows different Sessions to run concurrently. A write-ahead execution claim marks a process-local busy period for restart recovery: terminal completion, failure, or user interruption releases it, while shutdown interruption and process death preserve it. Startup recovery resumes claimed top-level Sessions with durable per-execution attempt accounting. The claim is a recovery marker, not clustered ownership, fencing, or an exactly-once guarantee. +- Keep local Session drains process-local until clustering is implemented. `SessionRunCoordinator` joins explicit same-Session resumes, coalesces prompt wakeups, and allows different Sessions to run concurrently. A write-ahead execution claim marks a process-local busy period for restart recovery: terminal completion, failure, or user interruption releases it, while shutdown interruption and process death preserve it. Startup recovery resumes claimed Sessions with durable per-execution attempt accounting. Running subagent Jobs own recovery of the children they represent. The claim is a recovery marker, not clustered ownership, fencing, or an exactly-once guarantee. - Keep delivery vocabulary explicit. Prompts steer by default. At safe step boundaries, steered compaction takes priority up to the first steered move control; other steers retain enqueue order. At an idle boundary, steers take priority; otherwise exactly one queued item delivers before the runner reevaluates continuation. Inbox items may be cancelled or changed between queue and steer before delivery. Promoting new user input resets the selected agent's step allowance; a batch of steers resets it once. - One step is one logical LLM call; its durable record covers only the model-visible span. Do not write "provider turn", and do not use bare "turn" for a single call: "turn" is reserved for the future assistant-turn unit containing all steps from prompt promotion until the session would go idle. - Keep event replay ownership separate from clustered Session execution ownership. diff --git a/packages/core/src/session/execution/restart.ts b/packages/core/src/session/execution/restart.ts index af2f126443a9..541f22611c97 100644 --- a/packages/core/src/session/execution/restart.ts +++ b/packages/core/src/session/execution/restart.ts @@ -354,7 +354,7 @@ export const layer = (options?: Options) => }) return false } - // Defer the start until every suspended root has counted its attempt; a fast + // Defer the start until every suspended session has counted its attempt; a fast // resumed child settling in Phase 1 would wake its parent past that budget. return true }) @@ -368,11 +368,23 @@ export const layer = (options?: Options) => resumeSuspendedSessions: Effect.gen(function* () { const active = yield* execution.active const pending = yield* jobs.pendingBackground - const children = pending.flatMap((background) => - background.status === "running" && background.recovery.kind === "subagent" - ? [background.recovery.childSessionID] - : [], + const children = new Set( + pending.flatMap((background) => + background.status === "running" && background.recovery.kind === "subagent" + ? [background.recovery.childSessionID] + : [], + ), ) + yield* store.releaseOrphanedChildClaims() + // Job.cancel persists the cancellation before interrupting the execution. A crash + // in between leaves its claim behind. Settlement writes a formatted error to the + // outbox atomically with release; such an older report can coexist with a new claim. + for (const { status, recovery, error } of pending) { + if (status !== "cancelled" || recovery.kind !== "subagent" || error !== undefined) continue + if (!active.has(recovery.childSessionID) && !children.has(recovery.childSessionID)) { + yield* store.release(recovery.childSessionID) + } + } // Early notices wait for recovery's accounting, including Sessions that exhaust their budget. const suspended = new Set( [...(yield* store.listSuspended()), ...children].filter((sessionID) => !active.has(sessionID)), @@ -386,7 +398,7 @@ export const layer = (options?: Options) => suspended.clear() }), ) - // Hold notice wakes until every suspended root has counted its attempt, so a woken + // Hold notice wakes until every suspended session has counted its attempt, so a woken // child cannot settle mid-sweep and wake its parent past that budget. const notices: Notices = { synthetic: (input) => @@ -400,7 +412,6 @@ export const layer = (options?: Options) => Effect.uninterruptible, ), } - yield* store.releaseChildClaims(children) const recoverable = Effect.fnUntraced(function* (background: Job.Background) { return (yield* jobs.get(background.id))?.status !== "running" }) @@ -438,12 +449,16 @@ export const layer = (options?: Options) => { discard: true }, ) - // Count every suspended root's attempt before a resumed child can settle and wake it. + // A directly driven child can also settle and wake its parent. Account for every + // claimed execution before starting any of them, leaving running jobs to SubagentJob. const resumed = yield* execution.active + const direct = yield* Effect.filter( + (yield* store.listSuspended()).filter((sessionID) => !resumed.has(sessionID) && !children.has(sessionID)), + prepareResume, + ) yield* Effect.forEach( - (yield* store.listSuspended()).filter((sessionID) => !resumed.has(sessionID)), + direct, Effect.fnUntraced(function* (sessionID) { - if (!(yield* prepareResume(sessionID))) return suspended.delete(sessionID) yield* execution.resume(sessionID).pipe(Effect.ignore, Effect.forkIn(scope)) }), diff --git a/packages/core/src/session/store.ts b/packages/core/src/session/store.ts index d38f979112ce..94e13d2717df 100644 --- a/packages/core/src/session/store.ts +++ b/packages/core/src/session/store.ts @@ -75,8 +75,7 @@ export interface Interface { messageID: SessionMessage.ID, ) => Effect.Effect<{ readonly sessionID: Session.ID; readonly message: SessionMessage.MediaInfo } | undefined> /** - * Top-level Sessions holding an execution claim. Recoverable background - * children are resumed separately through their durable Job records. + * Sessions holding an execution claim, including directly driven children. */ readonly listSuspended: () => Effect.Effect> /** @@ -89,10 +88,9 @@ export interface Interface { /** Releases the claim and resets resume accounting. Terminal events call this on commit. */ readonly release: (sessionID: Session.ID) => Effect.Effect /** - * Clears orphaned child claims except children owned by recoverable - * background subagent jobs. + * Release abandoned claims when the child's parent was deleted. */ - readonly releaseChildClaims: (recoverable: ReadonlyArray) => Effect.Effect + readonly releaseOrphanedChildClaims: () => Effect.Effect /** * Durably counts one more resume of an orphaned claim, returning the new * total — or undefined when the Session no longer exists. @@ -243,7 +241,7 @@ const layer = Layer.effect( return yield* db .select({ sessionID: SessionTable.id }) .from(SessionTable) - .where(and(isNotNull(SessionTable.time_suspended), isNull(SessionTable.parent_id))) + .where(isNotNull(SessionTable.time_suspended)) .all() .pipe( Effect.orDie, @@ -270,15 +268,14 @@ const layer = Layer.effect( .run() .pipe(Effect.orDie) }), - releaseChildClaims: Effect.fn("SessionStore.releaseChildClaims")((recoverable) => + releaseOrphanedChildClaims: Effect.fn("SessionStore.releaseOrphanedChildClaims")(() => db .update(SessionTable) .set({ time_suspended: null, resume_attempts: 0, time_updated: sql`${SessionTable.time_updated}` }) .where( and( isNotNull(SessionTable.time_suspended), - isNotNull(SessionTable.parent_id), - recoverable.length > 0 ? notInArray(SessionTable.id, Array.from(recoverable)) : undefined, + notInArray(SessionTable.parent_id, db.select({ id: SessionTable.id }).from(SessionTable)), ), ) .run() diff --git a/packages/core/test/session-execution.test.ts b/packages/core/test/session-execution.test.ts index f1787d1e6b11..710d42367d84 100644 --- a/packages/core/test/session-execution.test.ts +++ b/packages/core/test/session-execution.test.ts @@ -60,23 +60,63 @@ describe("SessionExecution lifecycle", () => { expect(SessionExecution.terminal(interrupted, "user")).toEqual({ type: "interrupted", reason: "user" }) }) - it.effect("the sweep only lists claimed top-level Sessions", () => + it.effect("resumes a directly driven child after execution teardown", () => Effect.gen(function* () { const database = yield* Database.Service const store = yield* SessionStore.Service const parent = Session.ID.make("ses_recover_parent") const child = Session.ID.make("ses_recover_child") const idle = Session.ID.make("ses_recover_idle") - yield* seedSessions(database, [parent], { time_suspended: Date.now() }) - yield* seedSessions(database, [idle]) - // Children recover through background Job records, never through the root claim sweep. - yield* seedSessions(database, [child], { time_suspended: Date.now(), parent_id: parent }) + yield* seedSessions(database, [parent]) + yield* seedSessions(database, [child, idle], { parent_id: parent }) + const scope = yield* Scope.fork(yield* Effect.scope) + const running = yield* Deferred.make() + const before = yield* buildExecution(scope, () => + Deferred.succeed(running, undefined).pipe(Effect.andThen(Effect.never)), + ) + yield* Context.get(before, SessionExecution.Service).resume(child).pipe(Effect.forkIn(scope)) + yield* Deferred.await(running) + yield* Scope.close(scope, Exit.void) + expect((yield* claims(database))[child]).toBe(true) - expect(yield* store.listSuspended()).toEqual([parent]) + const drained: Session.ID[] = [] + const resumed = yield* Deferred.make() + const after = yield* buildExecution(yield* Scope.fork(yield* Effect.scope), ({ sessionID }) => + Effect.sync(() => void drained.push(sessionID)).pipe(Effect.andThen(Deferred.succeed(resumed, undefined))), + ) + const restart = Context.get(after, SessionRestart.Service) + yield* restart.resumeSuspendedSessions + expect((yield* store.context(child)).filter((message) => message.type === "synthetic")).toMatchObject([ + { metadata: { notice: "restart" } }, + ]) + yield* Deferred.await(resumed) + yield* Context.get(after, SessionExecution.Service).awaitIdle(child) + yield* restart.resumeSuspendedSessions + expect(drained).toEqual([child]) + expect(yield* claims(database)).toEqual({ [parent]: false, [child]: false, [idle]: false }) + }), + ) - // The sweep clears orphaned child claims outright; parents keep theirs. - yield* store.releaseChildClaims([]) - expect(yield* claims(database)).toEqual({ [parent]: true, [child]: false, [idle]: false }) + it.effect("leaves a child parked when its parent is gone or its restart budget is exhausted", () => + Effect.gen(function* () { + const database = yield* Database.Service + const parent = Session.ID.make("ses_child_budget_parent") + const exhausted = Session.ID.make("ses_child_budget_exhausted") + const orphan = Session.ID.make("ses_child_budget_orphan") + yield* seedSessions(database, [parent]) + yield* seedSessions(database, [exhausted], { parent_id: parent, time_suspended: Date.now(), resume_attempts: 2 }) + yield* seedSessions(database, [orphan], { parent_id: Session.ID.make("ses_deleted"), time_suspended: Date.now() }) + const drained: Session.ID[] = [] + const context = yield* buildExecution( + yield* Scope.fork(yield* Effect.scope), + ({ sessionID }) => Effect.sync(() => void drained.push(sessionID)), + { maxAttempts: 2 }, + ) + yield* Context.get(context, SessionRestart.Service).resumeSuspendedSessions + yield* Effect.forEach([exhausted, orphan], Context.get(context, SessionExecution.Service).awaitIdle) + expect(drained).toEqual([]) + expect(yield* claims(database)).toEqual({ [parent]: false, [exhausted]: false, [orphan]: false }) + expect(yield* attempts(database, exhausted)).toBe(0) }), ) @@ -279,6 +319,50 @@ describe("SessionExecution lifecycle", () => { }), ) + it.effect("honors a job cancellation persisted before its child could release the claim", () => + Effect.gen(function* () { + const database = yield* Database.Service + const jobs = yield* Job.Service + const parent = Session.ID.make("ses_cancel_parent") + const child = Session.ID.make("ses_cancel_child") + yield* seedSessions(database, [parent]) + yield* seedSessions(database, [child], { parent_id: parent, time_suspended: Date.now(), resume_attempts: 1 }) + yield* jobs.start({ + id: child, + type: "subagent", + recovery: { + kind: "subagent", + childSessionID: child, + parentSessionID: parent, + agent: "general", + description: "Inspect", + }, + background: true, + run: Effect.never, + }) + yield* jobs.cancel(child) + const scope = yield* Scope.fork(yield* Effect.scope) + const restarted = yield* Job.make.pipe(Scope.provide(scope)) + const drained: Session.ID[] = [] + const context = yield* buildExecution( + scope, + ({ sessionID }) => Effect.sync(() => void drained.push(sessionID)), + undefined, + restarted, + ) + const restart = Context.get(context, SessionRestart.Service) + yield* restart.resumeSuspendedSessions + yield* Context.get(context, SessionExecution.Service).awaitIdle(parent) + yield* restart.resumeSuspendedSessions + expect(drained).toEqual([parent]) + expect(yield* SessionInbox.list(database.db, parent)).toMatchObject([ + { payload: { metadata: { state: "cancelled" } } }, + ]) + expect((yield* claims(database))[child]).toBe(false) + expect(yield* attempts(database, child)).toBe(0) + }), + ) + it.effect("starts every claimed execution without waiting for earlier drains to finish", () => Effect.gen(function* () { const database = yield* Database.Service @@ -448,7 +532,7 @@ describe("SessionRestart background recovery", () => { const parent = Session.ID.make("ses_background_recovery_parent") const child = Session.ID.make("ses_background_recovery_child") yield* seedSessions(database, [parent]) - yield* seedSessions(database, [child], { parent_id: parent, time_suspended: Date.now() }) + yield* seedSessions(database, [child], { parent_id: parent }) yield* seedBackground(jobs, parent, [ { id: "sh_background_orphan", shellID: "sh_background_orphan", command: "sleep 60" }, ]) @@ -1243,7 +1327,8 @@ describe("SessionRestart background recovery", () => { const child = Session.ID.make("ses_subagent_recovery_child") const unrelated = Session.ID.make("ses_subagent_unrelated_child") yield* seedSessions(database, [parent], { time_suspended: Date.now(), resume_attempts: 1 }) - yield* seedSessions(database, [child, unrelated], { parent_id: parent, time_suspended: Date.now() }) + yield* seedSessions(database, [child], { parent_id: parent, time_suspended: Date.now() }) + yield* seedSessions(database, [unrelated], { parent_id: parent }) yield* jobs.start({ id: child, type: "subagent", diff --git a/specs/v2/session.md b/specs/v2/session.md index 8a320616f4b0..631f1e49f14e 100644 --- a/specs/v2/session.md +++ b/specs/v2/session.md @@ -65,7 +65,7 @@ The public interrupt operation verifies that the durable Session exists. An unkn `sessions.active()` snapshots busy periods currently owned by this process. Durable execution events and claims are historical and recovery records, not proof that this process is still live. -Execution commits a write-ahead claim when a process-local busy period starts. Success, failure, and user interruption release the claim; shutdown interruption and unclean process death preserve it. On startup, managed Node and fetch runtimes resume claimed top-level Sessions, append a durable continuation instruction, and count recovery attempts. Recovery is bounded per claimed execution but does not guarantee exactly-once provider requests or tool effects. +Execution commits a write-ahead claim when a process-local busy period starts. Success, failure, and user interruption release the claim; shutdown interruption and unclean process death preserve it. On startup, managed Node and fetch runtimes resume claimed Sessions, append a durable continuation instruction, and count recovery attempts. A running subagent Job owns its child's recovery; a directly driven child resumes through its own claim. Recovery is bounded per claimed execution but does not guarantee exactly-once provider requests or tool effects. ## One Step May Have Several Physical Attempts @@ -113,6 +113,6 @@ There is no separate finite Session-history endpoint. Request/response consumers ## Recovery Boundaries Stay Explicit -An advisory wake is not itself crash recovery. Crash recovery is driven by a write-ahead execution claim that survives without a releasing terminal. Startup recovery resumes claimed top-level Sessions from durable projected history with bounded attempt accounting. It fails stale running tool projections before continuing, but it cannot prove whether an interrupted external operation already took effect and does not guarantee exactly-once provider or tool behavior. +An advisory wake is not itself crash recovery. Crash recovery is driven by a write-ahead execution claim that survives without a releasing terminal. Startup recovery resumes claimed Sessions from durable projected history with bounded attempt accounting. It fails stale running tool projections before continuing, but it cannot prove whether an interrupted external operation already took effect and does not guarantee exactly-once provider or tool behavior. Event replay ownership is separate from Session execution ownership. Local execution remains process-owned until clustering introduces an explicit placement and fencing protocol.