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