From d8eb4abf772b761f40991e1f5874ca7442db048a Mon Sep 17 00:00:00 2001 From: 99Gaoxiaoqi <72881245+99Gaoxiaoqi@users.noreply.github.com> Date: Mon, 7 Sep 2026 23:24:16 +0800 Subject: [PATCH] fix(runtime-host): defer Graph wakes while Session execution is paused Preserve durable checkpoints during Plan mode or missing connection identity, then resume after explicit configuration or Plan execution changes. Keep foreground Plan admission ahead of background recovery. Fixes #4991 Generated-by: OpenAI Codex --- .../core/src/agent-graph-supervisor-wake.ts | 5 +- .../__tests__/execution-composition.test.ts | 197 +++++++++++ .../src/__tests__/plan-coordinator.test.ts | 319 ++++++++++++++++++ .../__tests__/root-turn-coordinator.test.ts | 56 +++ .../session-catalog-coordinator.test.ts | 105 ++++++ .../agent-graph-execution-coordinator.ts | 12 + .../src/server/execution-composition.ts | 14 + .../src/server/host-session-availability.ts | 5 +- .../src/server/plan-coordinator.ts | 27 +- .../src/server/session-catalog-coordinator.ts | 15 +- .../agent-graph-supervisor-wake.test.ts | 206 +++++++++++ .../src/agent-graph-supervisor-wake.ts | 61 +++- packages/runtime/src/goal-turn-lifecycle.ts | 2 +- .../src/sqlite-session-metadata-store.ts | 12 +- 14 files changed, 1024 insertions(+), 12 deletions(-) create mode 100644 packages/runtime-host/src/__tests__/plan-coordinator.test.ts diff --git a/packages/core/src/agent-graph-supervisor-wake.ts b/packages/core/src/agent-graph-supervisor-wake.ts index d8d02f7ba3..692b237cea 100644 --- a/packages/core/src/agent-graph-supervisor-wake.ts +++ b/packages/core/src/agent-graph-supervisor-wake.ts @@ -109,6 +109,9 @@ export interface AgentGraphSupervisorWakeStore { wakeId: string, ): Promise; listUnsettledAgentGraphSupervisorWakes(): Promise; - listRetryableAgentGraphSupervisorWakes(): Promise; + /** Pending or failed wakes eligible for a new attempt, optionally scoped to one Session. */ + listRetryableAgentGraphSupervisorWakes( + rootSessionId?: string, + ): Promise; recoverAgentGraphSupervisorWakes(): Promise; } diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index f20fa37f49..50be007552 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -47,7 +47,9 @@ import { SessionManager, type BackendFactory } from '@maka/runtime/session-manag import { workHubDirectStopAbortSource } from '@maka/runtime/session-manager'; import { fingerprintAgentGraphRunnableIntent } from '@maka/runtime/stream-graph-admission'; import type { AgentGraphRunnableIntent } from '@maka/runtime/stream-graph-readiness'; +import { agentGraphIdForRootSession } from '@maka/runtime/stream-graph-coordinator'; import { createAgentGraphControlStore } from '@maka/storage/agent-graph-control-store'; +import { openInteractivePlanStoreForWrite } from '@maka/storage/plan-authority'; import { openInteractiveExecutionStoresForWrite } from '@maka/storage/execution-stores'; import { createSessionStore } from '@maka/storage/session-store'; import { @@ -1984,6 +1986,201 @@ test('production composition validates graph stop before aborting a claimed chil }); }); +for (const resumeVia of [ + 'configuration', + 'plan.control', + 'plan.turn.start', + 'account-selection', +] as const) { + test(`production Graph recovery keeps a paused wake durable through startup and resumes once via ${resumeVia}`, async () => { + await withCompositionRoot(async ({ root, owner }) => { + const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + const connectionId = await configureFakeDefaultTarget(owner); + const missingConnection = resumeVia === 'account-selection'; + const session = await stores.sessionStore.create({ + cwd: root, + ...(missingConnection ? {} : { llmConnectionId: connectionId }), + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'explore', + collaborationMode: missingConnection ? 'agent' : 'plan', + }); + const graphId = agentGraphIdForRootSession(session.id); + let graphStore = createAgentGraphControlStore(root); + await graphStore.commitAgentGraphScheduleUpdate({ + schemaVersion: 1, + graphId, + updateId: `graph_update_${'a'.repeat(32)}`, + updateFingerprint: `sha256:${'a'.repeat(64)}`, + source: { + sessionId: session.id, + runId: 'historical-run', + turnId: 'historical-turn', + toolCallId: 'historical-tool', + }, + addWork: [], + stop: [{ targetId: 'historical-work', reason: 'historical checkpoint' }], + }); + const client = { + hostEpoch: 'execution-composition-test', + connectionId: 'plan-graph-test', + principal: 'local_os_user' as const, + acquireResidency: () => ({ release() {} }), + }; + let drains = 0; + const context = { + ...compositionContext(owner), + requestDrain: () => { + drains += 1; + }, + }; + const overrides = { + primaryBackendFactory: (backendContext: ConstructorParameters[0]) => + new FakeBackend(backendContext), + }; + let restartedOwner: InteractiveRootOwner | undefined; + let composition = await createExecutionRuntimeHostComposition(context, {}, overrides); + try { + await composition.recover(); + const queried = await composition.handlers['agent.graph.query']( + { rootSessionId: session.id }, + client, + ); + assert.ok(queried.ok); + const snapshot = queried.result; + const wakeId = `${graphId}:${snapshot.snapshotVersion}`; + await graphStore.claimAgentGraphSupervisorWake({ + schemaVersion: 1, + graphId, + wakeId, + snapshotVersion: snapshot.snapshotVersion, + rootSessionId: session.id, + }); + await graphStore.beginAgentGraphSupervisorWakeAttempt({ + graphId, + wakeId, + attemptId: 'failed-plan-attempt', + turnId: 'failed-plan-turn', + }); + await graphStore.completeAgentGraphSupervisorWakeAttempt({ + graphId, + wakeId, + attemptId: 'failed-plan-attempt', + status: 'retryable_failed', + failureReason: missingConnection + ? 'This Session requires an explicit account selection before it can run.' + : 'Background and delegated roots cannot execute while the Session is in Plan mode.', + }); + graphStore.close(); + await composition.close(); + await owner.close(); + restartedOwner = await tryAcquireInteractiveRootOwner(owner.capability); + assert.ok(restartedOwner); + composition = await createExecutionRuntimeHostComposition( + { ...context, owner: restartedOwner }, + {}, + overrides, + ); + await composition.recover(); + graphStore = createAgentGraphControlStore(root); + // A live query after recovery verifies the Graph authority was not drained. + assert.ok( + (await composition.handlers['agent.graph.query']({ rootSessionId: session.id }, client)) + .ok, + ); + const recoveredStores = await openInteractiveExecutionStoresForWrite(restartedOwner.lease); + assert.equal( + (await graphStore.readAgentGraphSupervisorWake(graphId, wakeId))?.attemptCount, + 1, + ); + assert.equal( + (await recoveredStores.sessionStore.readHeaderSnapshot(session.id)).collaborationMode, + missingConnection ? 'agent' : 'plan', + ); + assert.equal( + (await recoveredStores.runtimeEventStore.listSessionInvocations(session.id)).length, + 0, + ); + let replay: (() => Promise) | undefined; + if (resumeVia === 'configuration' || missingConnection) { + const header = await recoveredStores.sessionStore.readHeaderRecordSnapshot(session.id); + const changed = await composition.handlers['session.configuration.update']( + { + sessionId: session.id, + expectedRevision: header.revision, + patch: missingConnection + ? { + modelTarget: { + kind: 'explicit', + connectionId, + connectionSlug: 'fake', + model: 'fake-model', + }, + } + : { collaborationMode: 'agent' }, + }, + client, + ); + assert.ok(changed.ok); + } else { + const plans = await openInteractivePlanStoreForWrite(restartedOwner.lease); + const submitted = await plans.submitProposal({ + sessionId: session.id, + operationId: 'submit-plan', + turnId: 'proposal-turn', + title: 'Resume graph work', + steps: [ + { + id: 'step-1', + title: 'Review checkpoint', + description: 'Inspect the pending graph checkpoint', + }, + ], + }); + assert.equal(submitted.event.type, 'plan_submitted'); + if (submitted.event.type !== 'plan_submitted') throw new Error('Missing proposal'); + const approval = { + kind: 'approve_proposal' as const, + sessionId: session.id, + proposalId: submitted.event.proposal.proposalId, + expectedRevision: submitted.event.proposal.revision, + expectedStoreVersion: submitted.event.storeVersion, + }; + if (resumeVia === 'plan.control') { + const request = { ...approval, operationId: 'approve-plan' }; + assert.ok((await composition.handlers['plan.control'](request, client)).ok); + replay = () => composition.handlers['plan.control'](request, client); + } else { + const request = { ...approval, turnId: 'foreground-plan-turn' }; + assert.ok((await composition.handlers['plan.turn.start'](request, client)).ok); + replay = () => composition.handlers['plan.turn.start'](request, client); + } + } + await waitFor( + async () => + (await graphStore.readAgentGraphSupervisorWake(graphId, wakeId))?.status === + 'delivered', + 10_000, + ); + assert.equal( + (await graphStore.readAgentGraphSupervisorWake(graphId, wakeId))?.attemptCount, + 2, + ); + await replay?.(); + assert.equal( + (await recoveredStores.runtimeEventStore.listSessionInvocations(session.id)).length, + resumeVia === 'plan.turn.start' ? 2 : 1, + ); + assert.equal(drains, 0); + } finally { + graphStore.close(); + await composition.close(); + await restartedOwner?.close(); + } + }); + }); +} + function compositionContext(owner: InteractiveRootOwner) { return { owner, diff --git a/packages/runtime-host/src/__tests__/plan-coordinator.test.ts b/packages/runtime-host/src/__tests__/plan-coordinator.test.ts new file mode 100644 index 0000000000..757e96d592 --- /dev/null +++ b/packages/runtime-host/src/__tests__/plan-coordinator.test.ts @@ -0,0 +1,319 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import test from 'node:test'; +import { PLAN_USER_ABANDON_REASON, PLAN_USER_CANCEL_REASON } from '@maka/core/plan'; +import { openInteractiveExecutionStoresForWrite } from '@maka/storage/execution-stores'; +import { openInteractivePlanStoreForWrite } from '@maka/storage/plan-authority'; +import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority'; +import type { PlanControlInput } from '../protocol/index.js'; +import type { ConnectionContext } from '../server/operation-dispatcher.js'; +import { HostPlanCoordinator } from '../server/plan-coordinator.js'; +import { SessionAdmissionGate } from '../server/session-admission-gate.js'; + +const context: ConnectionContext = { + hostEpoch: 'plan-test-epoch', + connectionId: 'plan-test-client', + principal: 'local_os_user', + acquireResidency: () => ({ release: () => undefined }), +}; + +test('ordinary approval and resume wake only once after durable control and refresh', async () => { + const fixture = await createFixture(); + try { + const approval = await fixture.submit(); + const approved = await fixture.control(approval); + assert.ok(approved.ok); + assert.deepEqual(fixture.observed, ['projection', 'refreshed', 'resumed']); + const executionId = approved.result.executionId!; + + await fixture.store.interruptActiveExecution(fixture.sessionId, 'test pause'); + fixture.observed.length = 0; + assert.deepEqual(await fixture.control(approval), approved); + assert.deepEqual(fixture.observed, ['projection', 'refreshed']); + assert.equal((await fixture.store.readState(fixture.sessionId)).activeExecutionId, undefined); + + fixture.observed.length = 0; + const resume = { + kind: 'resume_execution' as const, + sessionId: fixture.sessionId, + executionId, + operationId: 'resume-operation', + }; + const resumed = await fixture.control(resume); + assert.ok(resumed.ok); + assert.deepEqual(fixture.observed, ['projection', 'refreshed', 'resumed']); + await fixture.store.interruptActiveExecution(fixture.sessionId, 'test pause again'); + + fixture.observed.length = 0; + const cancelled = await fixture.control({ + kind: 'cancel_execution', + sessionId: fixture.sessionId, + executionId, + operationId: 'cancel-operation', + }); + assert.ok(cancelled.ok); + assert.deepEqual(fixture.observed, ['projection', 'refreshed']); + fixture.observed.length = 0; + assert.deepEqual(await fixture.control(resume), resumed); + assert.deepEqual(fixture.observed, ['projection', 'refreshed']); + assert.equal( + (await fixture.store.readState(fixture.sessionId)).executions[0]?.status, + 'cancelled', + ); + await fixture.archive(true); + fixture.observed.length = 0; + assert.deepEqual(await fixture.control(approval), approved); + assert.deepEqual(fixture.observed, ['projection', 'refreshed']); + } finally { + await fixture.close(); + } +}); + +test('archived, conflicting, and abandoned Plan controls do not wake execution', async () => { + const fixture = await createFixture(); + try { + const approval = await fixture.submit(); + const stale = await fixture.control({ ...approval, expectedRevision: 999 }); + assert.equal(stale.ok, false); + assert.deepEqual(fixture.observed, []); + await fixture.archive(true); + const archived = await fixture.control(approval); + assert.equal(archived.ok ? null : archived.error.code, 'session_archived'); + assert.deepEqual(fixture.observed, []); + await fixture.archive(false); + const abandoned = await fixture.control({ + kind: 'abandon_proposal', + sessionId: fixture.sessionId, + proposalId: approval.proposalId, + operationId: 'abandon-operation', + }); + assert.ok(abandoned.ok); + assert.deepEqual(fixture.observed, ['projection', 'refreshed']); + } finally { + await fixture.close(); + } +}); + +test('foreground Plan turns notify recovery after admission for approval or resume', async () => { + const fixture = await createFixture(); + try { + const approval = await fixture.submit(); + const { operationId: _operationId, ...turnApproval } = approval; + const approved = await fixture.coordinator.handlers['plan.turn.start']( + { ...turnApproval, turnId: 'approval-turn' }, + context, + ); + assert.ok(approved.ok); + assert.deepEqual(fixture.observed, ['projection', 'refreshed', 'admitted', 'resumed']); + await fixture.store.interruptActiveExecution(fixture.sessionId, 'test pause'); + fixture.observed.length = 0; + const resumed = await fixture.coordinator.handlers['plan.turn.start']( + { + kind: 'resume_execution', + sessionId: fixture.sessionId, + executionId: approved.result.plan.executionId!, + turnId: 'resume-turn', + }, + context, + ); + assert.ok(resumed.ok); + assert.deepEqual(fixture.observed, ['projection', 'refreshed', 'admitted', 'resumed']); + fixture.observed.length = 0; + const replay = await fixture.coordinator.handlers['plan.turn.start']( + { ...turnApproval, turnId: 'approval-turn' }, + context, + ); + assert.ok(replay.ok); + assert.deepEqual(fixture.observed, ['projection', 'refreshed', 'admitted']); + } finally { + await fixture.close(); + } +}); + +test('Plan control commit still notifies recovery when subsequent foreground admission is rejected', async () => { + const fixture = await createFixture({ rejectTurn: true }); + try { + const { operationId: _operationId, ...approval } = await fixture.submit(); + const outcome = await fixture.coordinator.handlers['plan.turn.start']( + { ...approval, turnId: 'rejected-turn' }, + context, + ); + assert.equal(outcome.ok ? null : outcome.error.code, 'operation_conflict'); + assert.ok((await fixture.store.readState(fixture.sessionId)).activeExecutionId); + assert.deepEqual(fixture.observed, ['projection', 'refreshed', 'admitted', 'resumed']); + } finally { + await fixture.close(); + } +}); + +test('failed refresh after approval does not wake execution or replay a stale wake', async () => { + const options = { failRefresh: true }; + const fixture = await createFixture(options); + try { + const approval = await fixture.submit(); + const result = await fixture.control(approval); + assert.equal(result.ok ? null : result.error.code, 'persistence_failed'); + assert.deepEqual(fixture.observed, ['projection', 'drain']); + assert.ok((await fixture.store.readState(fixture.sessionId)).activeExecutionId); + options.failRefresh = false; + fixture.observed.length = 0; + assert.ok((await fixture.control(approval)).ok); + assert.deepEqual(fixture.observed, ['projection', 'refreshed']); + } finally { + await fixture.close(); + } +}); + +async function createFixture(options: { failRefresh?: boolean; rejectTurn?: boolean } = {}) { + const root = await mkdtemp(join(tmpdir(), 'maka-plan-coordinator-')); + const capability = await resolveStorageRoot({ path: root, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + const { sessionStore: sessions } = await openInteractiveExecutionStoresForWrite(owner.lease); + const store = await openInteractivePlanStoreForWrite(owner.lease); + const session = await sessions.create({ + cwd: root, + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'explore', + collaborationMode: 'plan', + }); + const observed: string[] = []; + const admission = new SessionAdmissionGate(); + let foregroundAdmissionPending = false; + const coordinator = new HostPlanCoordinator({ + store, + sessions, + sessionAdmission: admission, + runtime: { + approvePlan: (input) => store.approveProposal(input), + resumePlanExecution: (sessionId, executionId, operationId) => + store.resumeExecution(sessionId, executionId, operationId), + cancelPlanExecution: (sessionId, executionId, operationId) => + store.cancelExecution({ + sessionId, + executionId, + operationId, + reason: PLAN_USER_CANCEL_REASON, + }), + abandonPlanProposal: (sessionId, proposalId, operationId) => + store.abandonProposal({ + sessionId, + proposalId, + operationId, + reason: PLAN_USER_ABANDON_REASON, + }), + requestPlanRevision: (sessionId, proposalId, operationId) => + store.requestRevision({ sessionId, proposalId, operationId }), + }, + isSessionActive: () => false, + onProjectionChanged: () => observed.push('projection'), + refreshContinuity: async (sessionId) => { + if (options.failRefresh) throw new Error('Refresh unavailable'); + assert.ok((await store.readState(sessionId)).storeVersion > 1); + observed.push('refreshed'); + }, + onExecutionResumed: (sessionId) => { + assert.equal(sessionId, session.id); + assert.equal(foregroundAdmissionPending, false); + assert.ok(['refreshed', 'admitted'].includes(observed.at(-1)!)); + observed.push('resumed'); + }, + requestDrain: () => observed.push('drain'), + root: { + startHostedExternalTransition: async (input) => { + foregroundAdmissionPending = true; + const outcome = await admission.run< + Awaited< + ReturnType< + NonNullable< + ConstructorParameters[0]['root'] + >['startHostedExternalTransition'] + > + > + >(input.sessionId, async (lease) => { + const prepared = await input.prepareContent(lease); + if (prepared.kind === 'rejected') return prepared.outcome; + observed.push('admitted'); + if (options.rejectTurn) { + return { + ok: false, + error: { code: 'operation_conflict', message: 'Admission rejected' }, + }; + } + return { + ok: true, + result: { + sessionId: input.sessionId, + turnId: input.turnId, + runId: 'foreground-run', + status: 'running', + }, + }; + }); + foregroundAdmissionPending = false; + return outcome; + }, + }, + }); + return { + coordinator, + store, + sessions, + observed, + sessionId: session.id, + control: (input: PlanControlInput) => coordinator.handlers['plan.control'](input, context), + archive: async (isArchived: boolean) => { + const { revision } = await sessions.readHeaderRecordSnapshot(session.id); + await sessions.setSessionsArchivedVersioned( + [{ sessionId: session.id, expectedVersion: revision }], + isArchived, + ); + }, + submit: async () => { + const submitted = await store.submitProposal({ + sessionId: session.id, + turnId: 'proposal-turn', + operationId: 'submit-operation', + title: 'Test Plan', + steps: [{ id: 'step-1', title: 'Run work', description: 'Run pending work' }], + }); + assert.equal(submitted.event.type, 'plan_submitted'); + if (submitted.event.type !== 'plan_submitted') throw new Error('Missing proposal'); + return { + kind: 'approve_proposal' as const, + sessionId: session.id, + proposalId: submitted.event.proposal.proposalId, + expectedRevision: submitted.event.proposal.revision, + expectedStoreVersion: submitted.state.storeVersion, + operationId: 'approve-operation', + }; + }, + close: async () => { + store.close(); + await owner.close(); + await rm(root, { recursive: true, force: true }); + }, + }; +} diff --git a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts index b2e30264b0..5ab2278f85 100644 --- a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts @@ -2138,6 +2138,60 @@ test('Agent Graph supervisor wake waits for root idle and binds one durable exec } }); +for (const pausedBy of ['plan', 'missing-account'] as const) { + test(`Agent Graph final admission paused by ${pausedBy} creates no root admission or Host drain`, async () => { + const fixture = await createFailureFixture({ + collaborationMode: pausedBy === 'plan' ? 'plan' : 'agent', + legacyConnectionIdentity: pausedBy === 'missing-account', + registerBackend: (backends) => + backends.register('ai-sdk', () => { + throw new Error('Paused admission must not construct a backend'); + }), + }); + try { + const graphId = agentGraphIdForRootSession(fixture.sessionId); + const turnId = 'plan-blocked-graph-turn'; + const outcome = await graphExecutions(fixture).run( + fixture.sessionId, + { + turnId, + text: 'Inspect the durable graph.', + turnOrchestration: { mode: 'graph', source: 'host_api' }, + origin: { + kind: 'agent_graph', + graphId, + wakeId: `${graphId}:plan`, + attemptId: 'plan-attempt', + }, + }, + new AbortController().signal, + async () => true, + ); + assert.deepEqual(outcome, { + kind: 'paused', + turnId, + reason: + pausedBy === 'plan' + ? 'Background and delegated roots cannot execute while the Session is in Plan mode.' + : 'This Session requires an explicit account selection before it can run.', + }); + assert.equal( + await fixture.stores.agentRunStore.readRootTurnAdmission(fixture.sessionId, turnId), + undefined, + ); + assert.equal( + (await fixture.stores.runtimeEventStore.listSessionInvocations(fixture.sessionId)).length, + 0, + ); + assert.equal(fixture.drainRequested(), false); + } finally { + await fixture.coordinator.close(); + await fixture.messages.close(); + await fixture.dispose(); + } + }); +} + test('Agent Graph supervisor wake preserves structured context-overflow outcomes', async () => { const fixture = await createFailureFixture({ registerBackend: (backends) => @@ -5565,6 +5619,7 @@ async function createFailureFixture(options: { afterHandoffSeal?(): Promise; directoryHostId?: string; corruptSessionRole?: boolean; + collaborationMode?: 'agent' | 'plan'; legacyConnectionIdentity?: boolean; childTools?: MakaTool[]; wrapAdmissionStore?(store: RootTurnAdmissionStore): RootTurnAdmissionStore; @@ -5608,6 +5663,7 @@ async function createFailureFixture(options: { : undefined; const session = await stores.sessionStore.create({ cwd: capability.canonicalPath, + collaborationMode: options.collaborationMode, ...(options.legacyConnectionIdentity ? {} : { llmConnectionId: 'cccccccc-cccc-4ccc-8ccc-cccccccccccc' }), diff --git a/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts index 578a4f9c30..ba44d23472 100644 --- a/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts @@ -983,6 +983,109 @@ test('configuration update admits Plan mode through Runtime authority', async () assert.equal(fixture.drainRequests(), 0); }); +test('configuration wakes execution only after a committed Plan to Agent transition is refreshed', async () => { + const observed: string[] = []; + const fixture = createFixture({ + header: { collaborationMode: 'plan' }, + continuity: { + refreshCanonical: async () => { + assert.equal(fixture.header().collaborationMode, 'agent'); + observed.push('refreshed'); + }, + }, + onExecutionResumed: (sessionId) => observed.push(sessionId), + }); + const input = { + sessionId: fixture.sessionId, + expectedRevision: fixture.revision(), + patch: { collaborationMode: 'agent' as const }, + }; + const outcome = await fixture.coordinator.handlers['session.configuration.update']( + input, + context, + ); + assert.equal(outcome.ok && outcome.result.kind, 'committed'); + assert.deepEqual(observed, ['refreshed', fixture.sessionId]); + + observed.length = 0; + const conflict = await fixture.coordinator.handlers['session.configuration.update']( + input, + context, + ); + assert.equal(conflict.ok && conflict.result.kind, 'revision_conflict'); + const noop = await fixture.coordinator.handlers['session.configuration.update']( + { ...input, expectedRevision: fixture.revision() }, + context, + ); + assert.equal(noop.ok && noop.result.kind, 'committed'); + assert.deepEqual(observed, []); +}); + +test('explicit account selection resumes a legacy Session only after the binding commits', async () => { + const resumed: string[] = []; + const fixture = createFixture({ + legacyConnectionIdentity: true, + onExecutionResumed: (sessionId) => { + assert.equal(fixture.header().llmConnectionId, 'connection-1'); + resumed.push(sessionId); + }, + }); + const input = configurationInput(fixture.sessionId, fixture.revision()); + const outcome = await fixture.coordinator.handlers['session.configuration.update']( + input, + context, + ); + assert.equal(outcome.ok && outcome.result.kind, 'committed'); + assert.deepEqual(resumed, [fixture.sessionId]); + await fixture.coordinator.handlers['session.configuration.update']( + { ...input, expectedRevision: fixture.revision() }, + context, + ); + assert.deepEqual(resumed, [fixture.sessionId]); +}); + +test('configuration does not wake on model changes, archival, or rejected transitions', async () => { + for (const scenario of ['model', 'archived', 'rejected', 'refresh_failed'] as const) { + const resumed: string[] = []; + const fixture = createFixture({ + header: { + collaborationMode: 'plan', + isArchived: scenario === 'archived', + model: 'old-model', + }, + ...(scenario === 'rejected' + ? { + manager: { + transitionSessionConfiguration: async () => { + throw new SessionConfigurationTransitionError('operation_conflict', 'Busy Session'); + }, + }, + } + : {}), + ...(scenario === 'refresh_failed' + ? { + continuity: { + refreshCanonical: async () => { + throw new Error('Read unavailable'); + }, + }, + } + : {}), + onExecutionResumed: (sessionId) => resumed.push(sessionId), + }); + const input = configurationInput(fixture.sessionId, fixture.revision()); + const outcome = await fixture.coordinator.handlers['session.configuration.update']( + { + ...input, + patch: { ...input.patch, collaborationMode: scenario === 'model' ? 'plan' : 'agent' }, + }, + context, + ); + assert.equal(outcome.ok, scenario === 'model', scenario); + assert.deepEqual(resumed, [], scenario); + } +}); + test('configuration update never rebinds a bound Session through a reused slug', async () => { let observedRef: unknown; const fixture = createFixture({ @@ -1602,6 +1705,7 @@ function createFixture( readonly runtimePolicy?: RuntimePolicy; readonly projectCatalog?: ProjectCatalog; readonly onProjectChanged?: () => void; + readonly onExecutionResumed?: (sessionId: string) => void; readonly legacyConnectionIdentity?: boolean; readonly header?: Partial; } = {}, @@ -1686,6 +1790,7 @@ function createFixture( new HostProjectMembershipGate(), options.onProjectChanged ?? (() => undefined), ), + onExecutionResumed: options.onExecutionResumed, requestDrain: () => { drains += 1; }, diff --git a/packages/runtime-host/src/server/agent-graph-execution-coordinator.ts b/packages/runtime-host/src/server/agent-graph-execution-coordinator.ts index a1449be86e..86e4bfd28f 100644 --- a/packages/runtime-host/src/server/agent-graph-execution-coordinator.ts +++ b/packages/runtime-host/src/server/agent-graph-execution-coordinator.ts @@ -28,6 +28,7 @@ import { } from '@maka/runtime/agent-graph-supervisor-wake'; import { RuntimeHostedRootConflictError, + RuntimeHostedRootUnavailableError, RuntimeMessageAuthorityInvariantError, } from '@maka/runtime/message-authority'; import { type SessionManager } from '@maka/runtime/session-manager'; @@ -37,6 +38,10 @@ import type { HostedExecutionAuthority, HostedExecutionSnapshot, } from './hosted-execution-authority.js'; +import { + LEGACY_CONNECTION_IDENTITY_EXECUTION_UNAVAILABLE_REASON, + PLAN_BACKGROUND_EXECUTION_UNAVAILABLE_REASON, +} from './host-session-availability.js'; import { waitForHostedExecutionIdleOrAbort, waitForHostedExecutionTerminal, @@ -133,6 +138,13 @@ export class HostAgentGraphExecutionCoordinator { break; } catch (error) { if (gateCancelled) return superseded(input.turnId); + if ( + error instanceof RuntimeHostedRootUnavailableError && + (error.message === PLAN_BACKGROUND_EXECUTION_UNAVAILABLE_REASON || + error.message === LEGACY_CONNECTION_IDENTITY_EXECUTION_UNAVAILABLE_REASON) + ) { + return { kind: 'paused', turnId: input.turnId, reason: error.message }; + } if (!(error instanceof RuntimeHostedRootConflictError)) throw error; const whenIdle = this.#executions.whenIdle(sessionId); if (whenIdle) await waitForHostedExecutionIdleOrAbort(whenIdle, abortSignal); diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index c8f4af9ae9..b8d6b94fad 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -1303,6 +1303,14 @@ export async function createExecutionRuntimeHostComposition( throw error; } }, + whenSessionExecutionIdle: (sessionId) => coordinator.whenIdle(sessionId), + isSessionPaused: async (sessionId) => { + const header = await stores.sessionStore.readHeaderSnapshot(sessionId); + return ( + header.collaborationMode === 'plan' || + (header.llmConnectionId === undefined && header.backend !== 'fake') + ); + }, acquireResidency: () => context.acquireResidency('agent-graph-supervisor'), onError: () => context.requestDrain(), }); @@ -1347,6 +1355,9 @@ export async function createExecutionRuntimeHostComposition( onCommittedMutation: registerConfigurationMutation, }); const sessionCatalog = new HostSessionCatalogCoordinator({ + onExecutionResumed: (sessionId) => { + void requireGraphSupervisorWake(graphSupervisorWake).notifySessionResumed(sessionId); + }, stores: stores.sessionStore, runtimePolicy: runtimePolicyStores, manager, @@ -1691,6 +1702,9 @@ export async function createExecutionRuntimeHostComposition( requestDrain: context.requestDrain, }); const plans = new HostPlanCoordinator({ + onExecutionResumed: (sessionId) => { + void requireGraphSupervisorWake(graphSupervisorWake).notifySessionResumed(sessionId); + }, store: openedPlanStore, sessions: stores.sessionStore, runtime: manager, diff --git a/packages/runtime-host/src/server/host-session-availability.ts b/packages/runtime-host/src/server/host-session-availability.ts index cd69d98945..46ec8aaa1c 100644 --- a/packages/runtime-host/src/server/host-session-availability.ts +++ b/packages/runtime-host/src/server/host-session-availability.ts @@ -26,6 +26,9 @@ import { type SessionToolProfile, } from '@maka/core/session'; +export const PLAN_BACKGROUND_EXECUTION_UNAVAILABLE_REASON = + 'Background and delegated roots cannot execute while the Session is in Plan mode.'; + const WORKTREE_CHILD_UNAVAILABLE_REASON = 'Worktree child Sessions must be continued through their parent agent.'; const CHILD_CONTINUATION_UNAVAILABLE_REASON = @@ -116,7 +119,7 @@ export function runtimeHostExecutionUnavailableReason( execution.kind !== 'regenerate' && execution.kind !== 'context_compact' && execution.kind !== 'safe_boundary_continuation' - ? 'Background and delegated roots cannot execute while the Session is in Plan mode.' + ? PLAN_BACKGROUND_EXECUTION_UNAVAILABLE_REASON : undefined) ?? (header.subagentWorkspace && !isManagedWorktreeChildExecution(execution) ? WORKTREE_CHILD_UNAVAILABLE_REASON diff --git a/packages/runtime-host/src/server/plan-coordinator.ts b/packages/runtime-host/src/server/plan-coordinator.ts index 8bd777fbd6..b8cac84141 100644 --- a/packages/runtime-host/src/server/plan-coordinator.ts +++ b/packages/runtime-host/src/server/plan-coordinator.ts @@ -60,7 +60,11 @@ type PlanRuntime = Pick< type AdmittedPlanControlRequest = | { readonly kind: 'ordinary'; readonly input: PlanControlInput } - | { readonly kind: 'plan_turn'; readonly input: PlanTurnStartInput }; + | { + readonly kind: 'plan_turn'; + readonly input: PlanTurnStartInput; + readonly onExecutionResumed: () => void; + }; export interface HostPlanCoordinatorInput { readonly store: InteractivePlanStoreWriter; @@ -69,6 +73,7 @@ export interface HostPlanCoordinatorInput { readonly sessionAdmission: SessionAdmissionGate; readonly isSessionActive: (sessionId: string) => boolean; readonly refreshContinuity: (sessionId: string, lease: SessionAdmissionLease) => Promise; + readonly onExecutionResumed?: (sessionId: string) => void; readonly onProjectionChanged: (sessionId: string) => void; readonly requestDrain: () => void; readonly root: Pick; @@ -91,6 +96,7 @@ export class HostPlanCoordinator { readonly #sessionAdmission: SessionAdmissionGate; readonly #isSessionActive: (sessionId: string) => boolean; readonly #refreshContinuity: HostPlanCoordinatorInput['refreshContinuity']; + readonly #onExecutionResumed: HostPlanCoordinatorInput['onExecutionResumed']; readonly #onProjectionChanged: HostPlanCoordinatorInput['onProjectionChanged']; readonly #requestDrain: () => void; readonly #root: HostPlanCoordinatorInput['root']; @@ -102,6 +108,7 @@ export class HostPlanCoordinator { this.#sessionAdmission = input.sessionAdmission; this.#isSessionActive = input.isSessionActive; this.#refreshContinuity = input.refreshContinuity; + this.#onExecutionResumed = input.onExecutionResumed; this.#onProjectionChanged = input.onProjectionChanged; this.#requestDrain = input.requestDrain; this.#root = input.root; @@ -111,6 +118,7 @@ export class HostPlanCoordinator { input: PlanTurnStartInput, context: ConnectionContext, ): Promise> { + let executionResumed = false; let plan: PlanControlResult | undefined; let planFailure: Extract, { ok: false }> | undefined; const turn = await this.#root.startHostedExternalTransition( @@ -120,7 +128,16 @@ export class HostPlanCoordinator { inputDigest: planTurnInputDigest(input), archivedMessage: 'Cannot start Plan execution in an archived Session', prepareContent: async (lease) => { - const outcome = await this.#control({ kind: 'plan_turn', input }, lease); + const outcome = await this.#control( + { + kind: 'plan_turn', + input, + onExecutionResumed: () => { + executionResumed = true; + }, + }, + lease, + ); if (!outcome.ok) { planFailure = outcome; return { @@ -137,6 +154,8 @@ export class HostPlanCoordinator { }, context, ); + // Foreground admission must finish before recovery can schedule background work. + if (executionResumed) this.#onExecutionResumed?.(input.sessionId); if (planFailure) return { ok: false, error: planFailure.error }; if (!turn.ok) return { ok: false, error: turn.error }; if (!plan) { @@ -221,6 +240,10 @@ export class HostPlanCoordinator { const result = await this.#applyControl(input); this.#onProjectionChanged(input.sessionId); await this.#refreshContinuity(input.sessionId, lease); + if (!replay && (input.kind === 'approve_proposal' || input.kind === 'resume_execution')) { + if (request.kind === 'plan_turn') request.onExecutionResumed(); + else this.#onExecutionResumed?.(input.sessionId); + } return { ok: true, result: projectControlResult(result) }; } catch (error) { return this.#controlFailure(error); diff --git a/packages/runtime-host/src/server/session-catalog-coordinator.ts b/packages/runtime-host/src/server/session-catalog-coordinator.ts index 0c08a2f029..4f8fdffa1a 100644 --- a/packages/runtime-host/src/server/session-catalog-coordinator.ts +++ b/packages/runtime-host/src/server/session-catalog-coordinator.ts @@ -172,6 +172,7 @@ export interface HostSessionCatalogCoordinatorOptions { readonly continuity: SessionContinuity; readonly workspaceResolver: HostWorkspaceResolver; readonly requestDrain: () => void; + readonly onExecutionResumed?: (sessionId: string) => void; readonly sessionAccessAuthority?: Pick< RuntimeHostAccessAuthority, 'activeSessionGrantForPrincipal' @@ -282,6 +283,7 @@ export class HostSessionCatalogCoordinator { readonly #continuity: SessionContinuity; readonly #workspaceResolver: HostWorkspaceResolver; readonly #requestDrain: () => void; + readonly #onExecutionResumed: HostSessionCatalogCoordinatorOptions['onExecutionResumed']; readonly #sessionAccessAuthority: | Pick | undefined; @@ -294,6 +296,7 @@ export class HostSessionCatalogCoordinator { this.#continuity = options.continuity; this.#workspaceResolver = options.workspaceResolver; this.#requestDrain = options.requestDrain; + this.#onExecutionResumed = options.onExecutionResumed; this.#sessionAccessAuthority = options.sessionAccessAuthority; } @@ -723,7 +726,17 @@ export class HostSessionCatalogCoordinator { clearConnectionBlock: input.patch.modelTarget !== undefined, configuration, }); - return configurationSuccess(await this.#committedUpdate(input.sessionId, lease)); + const result = await this.#committedUpdate(input.sessionId, lease); + if ( + (current.header.collaborationMode === 'plan' && + configuration.collaborationMode === 'agent') || + (current.header.llmConnectionId === undefined && + current.header.backend !== 'fake' && + configuration.llmConnectionId !== undefined) + ) { + this.#onExecutionResumed?.(input.sessionId); + } + return configurationSuccess(result); } catch (error) { if ( !commitAttempted && diff --git a/packages/runtime/src/__tests__/agent-graph-supervisor-wake.test.ts b/packages/runtime/src/__tests__/agent-graph-supervisor-wake.test.ts index 04bd4436b2..4820d28e70 100644 --- a/packages/runtime/src/__tests__/agent-graph-supervisor-wake.test.ts +++ b/packages/runtime/src/__tests__/agent-graph-supervisor-wake.test.ts @@ -26,6 +26,7 @@ import { AgentGraphSupervisorWakeCoordinator, recoverAgentGraphSupervisorContextOverflow, type AgentGraphSupervisorWakeDiagnostic, + type AgentGraphSupervisorWakeInput, type AgentGraphSupervisorTurnOutcome, } from '../agent-graph-supervisor-wake.js'; import { SessionActivityRegistry, type GoalTurnOutcome } from '../goal-turn-lifecycle.js'; @@ -900,6 +901,211 @@ describe('Agent Graph supervisor wake delivery', () => { }); }); +describe('Plan pauses durable Graph delivery', () => { + async function fixture(t: import('node:test').TestContext) { + const store = createSqliteSessionMetadataStore(':memory:'); + const activityRegistry = new SessionActivityRegistry(); + const state = { paused: true, available: true, starts: 0, errors: [] as unknown[] }; + const input: AgentGraphSupervisorWakeInput = { + activityRegistry, + wakeStore: store, + readSnapshot: async () => snapshot(), + isSessionPaused: async () => state.paused, + isSessionDeliverable: async () => state.available, + startTurn: async (_sessionId, turn) => { + state.starts += 1; + return { kind: 'completed', turnId: turn.turnId }; + }, + inspectAttempt: async () => 'missing', + onError: (_sessionId, error) => { + state.errors.push(error); + }, + newId: sequentialIds(), + }; + const coordinator = new AgentGraphSupervisorWakeCoordinator(input); + t.after(async () => { + await coordinator.close(); + store.close(); + }); + const wake = () => store.readAgentGraphSupervisorWake('graph-1', 'graph-1:snapshot-1'); + return { store, activityRegistry, state, input, coordinator, wake }; + } + + test('persists a new milestone in Plan and resumes it once without a new Graph event', async (t) => { + const f = await fixture(t); + await f.coordinator.notify('root-session', reconciliation()); + assert.equal((await f.wake())?.status, 'pending'); + assert.equal((await f.wake())?.attemptCount, 0); + // Graph (unlike Swarm) cannot reconstruct a milestone from notify(session) alone. + await f.coordinator.notify('root-session'); + assert.equal(f.state.starts, 0); + f.state.paused = false; + await Promise.all([ + f.coordinator.notifySessionResumed('root-session'), + f.coordinator.notifySessionResumed('root-session'), + ]); + await f.coordinator.waitForIdle(); + await f.coordinator.notifySessionResumed('root-session'); + await f.coordinator.waitForIdle(); + assert.equal(f.state.starts, 1); + assert.equal((await f.wake())?.status, 'delivered'); + assert.deepEqual(f.state.errors, []); + }); + + test('startup preserves a failed Plan wake without adding attempts', async (t) => { + const f = await fixture(t); + await createRunningAttempt(f.store); + await f.store.completeAgentGraphSupervisorWakeAttempt({ + graphId: 'graph-1', + wakeId: 'graph-1:snapshot-1', + attemptId: 'crashed-attempt', + status: 'retryable_failed', + failureReason: 'plan mode', + }); + await f.coordinator.recover(); + await f.coordinator.waitForIdle(); + assert.equal((await f.wake())?.attemptCount, 1); + assert.equal((await f.wake())?.status, 'retryable_failed'); + assert.equal(f.state.starts, 0); + assert.deepEqual(f.state.errors, []); + f.state.paused = false; + await f.coordinator.notifySessionResumed('root-session'); + await f.coordinator.waitForIdle(); + assert.equal(f.state.starts, 1); + }); + + test('rechecks Plan after waiting for activity before creating an attempt', async (t) => { + const f = await fixture(t); + f.state.paused = false; + const lease = f.activityRegistry.reserve('root-session'); + const acquiring = deferred(); + const acquire = f.activityRegistry.acquire.bind(f.activityRegistry); + f.activityRegistry.acquire = async (...args) => { + acquiring.resolve(); + return acquire(...args); + }; + const delivery = f.coordinator.notify('root-session', reconciliation()); + await acquiring.promise; + f.state.paused = true; + lease.release(); + await delivery; + assert.equal((await f.wake())?.attemptCount, 0); + assert.equal(f.state.starts, 0); + assert.deepEqual(f.state.errors, []); + }); + + test('does not lose a resume while the previous pause check is still unwinding', async (t) => { + const f = await fixture(t); + const checked = deferred(); + const releaseCheck = deferred(); + let first = true; + f.input.isSessionPaused = async () => { + const paused = f.state.paused; + if (first) { + first = false; + checked.resolve(); + await releaseCheck.promise; + } + return paused; + }; + const delivery = f.coordinator.notify('root-session', reconciliation()); + await checked.promise; + f.state.paused = false; + const resumed = f.coordinator.notifySessionResumed('root-session'); + releaseCheck.resolve(); + await Promise.all([delivery, resumed]); + await f.coordinator.waitForIdle(); + assert.equal(f.state.starts, 1); + assert.equal((await f.wake())?.status, 'delivered'); + }); + + for (const stillPaused of [true, false]) { + test(`final admission pause remains retryable without Host error (still Plan: ${stillPaused})`, async (t) => { + const f = await fixture(t); + f.state.paused = false; + f.input.startTurn = async (_sessionId, turn) => { + f.state.starts += 1; + if (f.state.starts > 1) return { kind: 'completed', turnId: turn.turnId }; + f.state.paused = stillPaused; + return { kind: 'paused', turnId: turn.turnId, reason: 'plan mode' }; + }; + await f.coordinator.notify('root-session', reconciliation()); + await f.coordinator.waitForIdle(); + assert.equal(f.state.starts, stillPaused ? 1 : 2); + assert.equal((await f.wake())?.status, stillPaused ? 'retryable_failed' : 'delivered'); + assert.deepEqual(f.state.errors, []); + }); + } + + test('foreground Plan execution settles before its deferred wake can create an attempt', async (t) => { + const f = await fixture(t); + await f.coordinator.notify('root-session', reconciliation()); + const foreground = deferred(); + const waiting = deferred(); + f.input.whenSessionExecutionIdle = () => { + waiting.resolve(); + return foreground.promise; + }; + f.state.paused = false; + const resumed = f.coordinator.notifySessionResumed('root-session'); + await waiting.promise; + assert.equal((await f.wake())?.attemptCount, 0); + assert.equal(f.state.starts, 0); + foreground.resolve(); + await resumed; + await f.coordinator.waitForIdle(); + assert.equal(f.state.starts, 1); + }); + + test('stop during deferred foreground recovery supersedes the wake without restarting', async (t) => { + const f = await fixture(t); + await f.coordinator.notify('root-session', reconciliation()); + const foreground = deferred(); + const waiting = deferred(); + f.input.whenSessionExecutionIdle = () => { + waiting.resolve(); + return foreground.promise; + }; + f.state.paused = false; + void f.coordinator.notifySessionResumed('root-session'); + await waiting.promise; + await f.coordinator.runWithSessionWakesSuppressed('root-session', async () => {}); + await f.coordinator.waitForIdle(); + assert.equal((await f.wake())?.status, 'superseded'); + assert.equal(f.state.starts, 0); + assert.deepEqual(f.state.errors, []); + }); + + test('resume only delivers its Session and archive still supersedes paused wakes', async (t) => { + const f = await fixture(t); + await f.coordinator.notify('root-session', reconciliation()); + await f.store.claimAgentGraphSupervisorWake({ + schemaVersion: 1, + graphId: 'graph-2', + wakeId: 'graph-2:snapshot-1', + snapshotVersion: 'snapshot-1', + rootSessionId: 'other-session', + }); + f.state.paused = false; + await f.coordinator.notifySessionResumed('root-session'); + await f.coordinator.waitForIdle(); + assert.equal(f.state.starts, 1); + assert.equal( + (await f.store.readAgentGraphSupervisorWake('graph-2', 'graph-2:snapshot-1'))?.status, + 'pending', + ); + f.state.available = false; + f.state.paused = true; + await f.coordinator.notifySessionResumed('other-session'); + await f.coordinator.waitForIdle(); + assert.equal( + (await f.store.readAgentGraphSupervisorWake('graph-2', 'graph-2:snapshot-1'))?.status, + 'superseded', + ); + assert.deepEqual(f.state.errors, []); + }); +}); + async function createRunningAttempt( store: ReturnType, ): Promise { diff --git a/packages/runtime/src/agent-graph-supervisor-wake.ts b/packages/runtime/src/agent-graph-supervisor-wake.ts index 30d088a552..6d0982f9ae 100644 --- a/packages/runtime/src/agent-graph-supervisor-wake.ts +++ b/packages/runtime/src/agent-graph-supervisor-wake.ts @@ -26,6 +26,7 @@ import type { ContextCompactionOutcome } from '@maka/core/events'; import { type SessionEvent } from '@maka/core/events'; import { type UserMessageInput } from '@maka/core/runtime-inputs'; import type { RuntimeInvocationOutcome } from '@maka/core/runtime-invocation'; +import { waitForIdleOrAbort } from './goal-turn-lifecycle.js'; import type { GoalTurnOutcome, SessionActivityLease, @@ -98,7 +99,8 @@ export interface AgentGraphSupervisorContextRecoveryDiagnostic { export type AgentGraphSupervisorTurnOutcome = | GoalTurnOutcome | { kind: 'context_overflow'; turnId: string; reason: string } - | { kind: 'superseded'; turnId: string; reason: string }; + | { kind: 'superseded'; turnId: string; reason: string } + | { kind: 'paused'; turnId: string; reason: string }; export async function recoverAgentGraphSupervisorContextOverflow(input: { rootSessionId: string; @@ -235,6 +237,10 @@ export interface AgentGraphSupervisorWakeInput { ): Promise; newId(): string; isSessionDeliverable?(rootSessionId: string): Promise; + /** Temporary execution policy: preserve the durable wake without admitting an attempt. */ + isSessionPaused?(rootSessionId: string): Promise; + /** Foreground Plan execution owns the root before background recovery resumes. */ + whenSessionExecutionIdle?(rootSessionId: string): Promise | undefined; /** Keep an external host alive while a durable wake is admitted or delivered. */ acquireResidency?(rootSessionId: string): SessionActivityLease; maxDeliveryAttempts?: number; @@ -327,6 +333,32 @@ export class AgentGraphSupervisorWakeCoordinator { return recovered; } + /** Resume durable checkpoints after an explicit Session execution-mode transition. */ + notifySessionResumed(rootSessionId: string): Promise | undefined { + if (this.#closed || this.#sessionWakesSuppressed(rootSessionId)) return undefined; + // Capture before tracking this task. A pause check may already have read the + // old mode; waiting for its task to exit prevents pendingWakeIds from eating + // the resume edge. Never wait for this task itself. + const pending = [...(this.#tasksBySession.get(rootSessionId) ?? [])]; + return this.#runTracked(rootSessionId, async (abortSignal) => { + try { + await Promise.all(pending); + const foreground = this.#input.whenSessionExecutionIdle?.(rootSessionId); + if (foreground) await waitForIdleOrAbort(foreground, abortSignal); + abortSignal.throwIfAborted(); + for (const wake of await this.#input.wakeStore.listRetryableAgentGraphSupervisorWakes( + rootSessionId, + )) { + this.#scheduleRecoveredWake(wake); + } + } catch (error) { + if (!this.#closed && !isAbortError(error)) { + await notifyError(this.#input.onError, rootSessionId, error); + } + } + }); + } + async waitForIdle(): Promise { while (this.#tasks.size > 0) await Promise.all([...this.#tasks]); } @@ -412,7 +444,12 @@ export class AgentGraphSupervisorWakeCoordinator { } #scheduleRecoveredWake(wake: AgentGraphSupervisorWakeRecord): void { - if (this.#closed || this.#pendingWakeIds.has(wake.wakeId)) return; + if ( + this.#closed || + this.#sessionWakesSuppressed(wake.rootSessionId) || + this.#pendingWakeIds.has(wake.wakeId) + ) + return; this.#pendingWakeIds.add(wake.wakeId); void this.#runTracked(wake.rootSessionId, async (abortSignal) => { try { @@ -453,6 +490,7 @@ export class AgentGraphSupervisorWakeCoordinator { await this.#supersedeSession(wake.rootSessionId, 'session_unavailable'); return; } + if (await this.#isSessionPaused(wake.rootSessionId)) return; const snapshot = await this.#input.readSnapshot(wake.rootSessionId); if (snapshot.graphId !== wake.graphId) { await this.#input.wakeStore.supersedeAgentGraphSupervisorWakes({ @@ -488,10 +526,16 @@ export class AgentGraphSupervisorWakeCoordinator { await this.#supersedeSession(wake.rootSessionId, 'session_unavailable'); return; } + if (await this.#isSessionPaused(wake.rootSessionId)) return; let overflowAttempt: { attemptId: string; turnId: string; failureReason: string } | undefined; const activity = await this.#input.activityRegistry.acquire(wake.rootSessionId, abortSignal); try { if (this.#closed || this.#sessionWakesSuppressed(wake.rootSessionId)) return; + if (!(await this.#isSessionDeliverable(wake.rootSessionId))) { + await this.#supersedeSession(wake.rootSessionId, 'session_unavailable'); + return; + } + if (await this.#isSessionPaused(wake.rootSessionId)) return; const attemptId = this.#input.newId(); const turnId = this.#input.newId(); const admission = await this.#input.wakeStore.beginAgentGraphSupervisorWakeAttempt({ @@ -546,6 +590,13 @@ export class AgentGraphSupervisorWakeCoordinator { }); return; } + if (outcome.kind === 'paused') { + await this.#markRetryable(wake.graphId, wake.wakeId, attemptId, outcome.reason); + if (await this.#isSessionPaused(wake.rootSessionId)) return; + // The mode may have changed back while final admission unwound. + void this.notifySessionResumed(wake.rootSessionId); + return; + } if (outcome.kind === 'suspended') { await this.#input.wakeStore.completeAgentGraphSupervisorWakeAttempt({ graphId: wake.graphId, @@ -674,6 +725,10 @@ export class AgentGraphSupervisorWakeCoordinator { }); } + async #isSessionPaused(rootSessionId: string): Promise { + return (await this.#input.isSessionPaused?.(rootSessionId)) ?? false; + } + async #isSessionDeliverable(rootSessionId: string): Promise { return (await this.#input.isSessionDeliverable?.(rootSessionId)) ?? true; } @@ -835,7 +890,7 @@ export function isAgentGraphSupervisorMilestone( function wakeOutcomeFailure( outcome: Exclude, ): string { - if (outcome.kind === 'context_overflow') return outcome.reason; + if (outcome.kind === 'context_overflow' || outcome.kind === 'paused') return outcome.reason; if (outcome.kind === 'errored' || outcome.kind === 'suspended') { return `${outcome.kind}: ${outcome.reason}`; } diff --git a/packages/runtime/src/goal-turn-lifecycle.ts b/packages/runtime/src/goal-turn-lifecycle.ts index 6e6c82ffaa..2bfdb95c37 100644 --- a/packages/runtime/src/goal-turn-lifecycle.ts +++ b/packages/runtime/src/goal-turn-lifecycle.ts @@ -96,7 +96,7 @@ function throwIfAborted(abortSignal?: AbortSignal): void { throw new DOMException('Session activity acquisition was aborted', 'AbortError'); } -async function waitForIdleOrAbort( +export async function waitForIdleOrAbort( whenIdle: Promise, abortSignal?: AbortSignal, ): Promise { diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index f42fa09387..f2792874ce 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -4022,7 +4022,9 @@ export class SqliteSessionMetadataStore { return rows.map(decodeAgentGraphSupervisorWakeAttemptRow); } - async listRetryableAgentGraphSupervisorWakes(): Promise { + async listRetryableAgentGraphSupervisorWakes( + rootSessionId?: string, + ): Promise { this.assertOpen(); const rows = this.db .prepare( @@ -4041,11 +4043,15 @@ export class SqliteSessionMetadataStore { created_at AS createdAt, updated_at AS updatedAt FROM agent_graph_supervisor_wakes - WHERE status = 'retryable_failed' + WHERE status IN ('pending', 'retryable_failed') + AND (? IS NULL OR root_session_id = ?) ORDER BY updated_at ASC, graph_id ASC, wake_id ASC `, ) - .all() as unknown as AgentGraphSupervisorWakeRow[]; + .all( + rootSessionId ?? null, + rootSessionId ?? null, + ) as unknown as AgentGraphSupervisorWakeRow[]; return rows.map(decodeAgentGraphSupervisorWakeRow); }