diff --git a/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts index ed61c44fa2..87d08839f9 100644 --- a/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts @@ -741,6 +741,32 @@ describe('Host Session retirement coordinator', () => { }); }); + test('archives a Session whose Agent Graph is stopped but remains open', async () => { + await withHarness(async (harness) => { + harness.quiescentGraphs.add(harness.rootId); + harness.blockers.graphWake.add(harness.rootId); + + const busy = await harness.coordinator.handlers['session.lifecycle.set']( + { sessionId: harness.revisionId, state: 'archived' }, + CONNECTION_CONTEXT, + ); + assert.equal(busy.ok, false); + if (busy.ok) assert.fail('An active supervisor wake must block Session retirement'); + assert.equal(busy.error.code, 'session_busy'); + await assertFamilyLifecycle(harness, false); + + harness.blockers.graphWake.clear(); + + const outcome = await harness.coordinator.handlers['session.lifecycle.set']( + { sessionId: harness.revisionId, state: 'archived' }, + CONNECTION_CONTEXT, + ); + + assert.equal(outcome.ok, true); + await assertFamilyLifecycle(harness, true); + }); + }); + test('retires a bound child worktree only after the Session tombstone commits', async () => { await withHarness(async (harness) => { const binding = { @@ -1235,6 +1261,7 @@ async function withHarness( graphWake: new Set(), scheduledTasks: new Set(), }; + const quiescentGraphs = new Set(); const memoryExtractionLane = new MemoryExtractionSessionLane(); const admission = new SessionAdmissionGate(); const harness: RetirementHarness = { @@ -1246,6 +1273,7 @@ async function withHarness( familyIds: [rootSession.id, revision.id], actions, blockers, + quiescentGraphs, admission, memoryExtractionLane, failRemoveCommit: false, @@ -1325,7 +1353,12 @@ async function withHarness( hasLiveSessionState: (sessionId) => blockers.effect.has(sessionId), }, graph: { - hasLiveSessionState: async (sessionId) => blockers.graph.has(sessionId), + readRetirementDisposition: async (sessionId) => + blockers.graph.has(sessionId) + ? ({ kind: 'busy', status: 'active' } as const) + : quiescentGraphs.has(sessionId) + ? ({ kind: 'quiescent_open' } as const) + : ({ kind: 'clear' } as const), listGraphIds: async (sessionId) => [agentGraphIdForRootSession(sessionId)], }, graphWake: { @@ -1429,6 +1462,7 @@ interface RetirementHarness { readonly graphWake: Set; readonly scheduledTasks: Set; }; + readonly quiescentGraphs: Set; readonly admission: SessionAdmissionGate; readonly memoryExtractionLane: MemoryExtractionSessionLane; coordinator: HostSessionRetirementCoordinator; diff --git a/packages/runtime-host/src/server/session-retirement-coordinator.ts b/packages/runtime-host/src/server/session-retirement-coordinator.ts index 52b672de96..8d6d0d1851 100644 --- a/packages/runtime-host/src/server/session-retirement-coordinator.ts +++ b/packages/runtime-host/src/server/session-retirement-coordinator.ts @@ -31,6 +31,7 @@ import { type SessionHeaderSnapshot, } from '@maka/storage/execution-stores'; import { type SessionManager } from '@maka/runtime/session-manager'; +import type { AgentGraphRetirementDisposition } from '@maka/runtime/stream-graph-coordinator'; import type { InteractiveSessionTodoWriter } from '@maka/storage/session-todo-authority'; import type { InteractiveContextOffloadWriter } from '@maka/storage/context-offload-store'; import { @@ -85,7 +86,7 @@ type RetirementSessionEffects = { hasLiveSessionState(sessionId: string): boolean; }; type RetirementGraph = { - hasLiveSessionState(sessionId: string): Promise; + readRetirementDisposition(sessionId: string): Promise; listGraphIds(sessionId: string): Promise; }; type RetirementGraphWake = { @@ -660,8 +661,13 @@ export class HostSessionRetirementCoordinator { throw new SessionRetirementBusyError(`Session ${sessionId} has a live derived effect`); } const header = requireFamilyRecord(family, sessionId).header; - if (!header.subagentParent && (await this.#graph.hasLiveSessionState(sessionId))) { - throw new SessionRetirementBusyError(`Session ${sessionId} has a live Agent Graph`); + if (!header.subagentParent) { + const graph = await this.#graph.readRetirementDisposition(sessionId); + if (graph.kind === 'busy') { + throw new SessionRetirementBusyError( + `Session ${sessionId} has an Agent Graph that is ${graph.status}`, + ); + } } if (!header.subagentParent && this.#graphWake.hasLiveSessionState(sessionId)) { throw new SessionRetirementBusyError( diff --git a/packages/runtime/src/__tests__/stream-graph-coordinator.test.ts b/packages/runtime/src/__tests__/stream-graph-coordinator.test.ts index 4201b6c807..e35b0e2d65 100644 --- a/packages/runtime/src/__tests__/stream-graph-coordinator.test.ts +++ b/packages/runtime/src/__tests__/stream-graph-coordinator.test.ts @@ -1364,7 +1364,7 @@ describe('host-managed agent graph coordinator', () => { await coordinator.close(); }); - test('fences retirement from durable open and closing graph state, not a stale client projection', async () => { + test('classifies retirement from durable graph state, not a stale client projection', async () => { const rootSessionId = 'root-session'; const childSessionId = 'child-session'; const graphId = agentGraphIdForRootSession(rootSessionId); @@ -1390,6 +1390,19 @@ describe('host-managed agent graph coordinator', () => { committedAt: 10, }; const workId = addUpdate.addWork[0]!.workId; + const stopRequest = compileAgentGraphScheduleUpdate({ + graphId, + input: { + operation: 'stop', + stop: [{ target_id: workId, reason: 'The work was stopped by the user.' }], + }, + context: toolContext(rootSessionId, 'root-run', 'root-turn', 'stop-work'), + }); + const stopUpdate: AgentGraphScheduleUpdate = { + ...stopRequest, + revision: 2, + committedAt: 20, + }; const finishRequest = compileAgentGraphScheduleUpdate({ graphId, input: { @@ -1538,16 +1551,74 @@ describe('host-managed agent graph coordinator', () => { try { assert.equal((await coordinator.getSnapshot(rootSessionId)).scheduleRevision, 0); assert.equal(await coordinator.readSessionState(rootSessionId), 'absent'); + assert.deepEqual(await coordinator.readRetirementDisposition(rootSessionId), { + kind: 'clear', + }); scheduleUpdates = [addUpdate]; assert.equal(await coordinator.readSessionState(rootSessionId), 'live'); assert.equal(await coordinator.hasLiveSessionState(rootSessionId), true); + assert.deepEqual(await coordinator.readRetirementDisposition(rootSessionId), { + kind: 'busy', + status: 'waiting', + }); - scheduleUpdates = [addUpdate, finishUpdate]; provisions = [provision]; claims = [claim]; + assert.deepEqual(await coordinator.readRetirementDisposition(rootSessionId), { + kind: 'busy', + status: 'active', + }); + + scheduleUpdates = [addUpdate, stopUpdate]; + runs = [ + { + ...runningRun, + terminalEvent: { + id: 'child-aborted-terminal', + sessionId: childSessionId, + invocationId: 'child-invocation', + runId, + turnId, + ts: 14, + partial: false, + role: 'system', + author: 'system', + status: 'aborted', + }, + }, + ]; + runtimeEvents = [ + runningEvent, + { + id: 'child-aborted', + invocationId: 'child-invocation', + sessionId: childSessionId, + runId, + turnId, + ts: 14, + role: 'system', + author: 'system', + partial: false, + status: 'aborted', + actions: { endInvocation: true }, + }, + ]; + assert.equal(await coordinator.readSessionState(rootSessionId), 'live'); + assert.equal(await coordinator.hasLiveSessionState(rootSessionId), true); + assert.deepEqual(await coordinator.readRetirementDisposition(rootSessionId), { + kind: 'quiescent_open', + }); + + scheduleUpdates = [addUpdate, finishUpdate]; + runs = [runningRun]; + runtimeEvents = [runningEvent]; assert.equal(await coordinator.readSessionState(rootSessionId), 'live'); assert.equal(await coordinator.hasLiveSessionState(rootSessionId), true); + assert.deepEqual(await coordinator.readRetirementDisposition(rootSessionId), { + kind: 'busy', + status: 'closing', + }); runs = [ { @@ -1584,6 +1655,47 @@ describe('host-managed agent graph coordinator', () => { ]; assert.equal(await coordinator.readSessionState(rootSessionId), 'terminal'); assert.equal(await coordinator.hasLiveSessionState(rootSessionId), false); + assert.deepEqual(await coordinator.readRetirementDisposition(rootSessionId), { + kind: 'clear', + }); + + scheduleUpdates = [addUpdate]; + runs = [ + { + ...runningRun, + terminalEvent: { + id: 'child-failed-terminal', + sessionId: childSessionId, + invocationId: 'child-invocation', + runId, + turnId, + ts: 15, + partial: false, + role: 'system', + author: 'system', + status: 'failed', + }, + }, + ]; + runtimeEvents = [ + runningEvent, + { + id: 'child-failed', + invocationId: 'child-invocation', + sessionId: childSessionId, + runId, + turnId, + ts: 15, + role: 'system', + author: 'system', + partial: false, + status: 'failed', + actions: { endInvocation: true }, + }, + ]; + assert.deepEqual(await coordinator.readRetirementDisposition(rootSessionId), { + kind: 'quiescent_open', + }); } finally { await coordinator.close(); } diff --git a/packages/runtime/src/stream-graph-coordinator.ts b/packages/runtime/src/stream-graph-coordinator.ts index 8ebb0885ca..3eb3378e11 100644 --- a/packages/runtime/src/stream-graph-coordinator.ts +++ b/packages/runtime/src/stream-graph-coordinator.ts @@ -139,6 +139,11 @@ export interface AgentGraphExecutionStopInput { withSupervisorWakesSuppressed(operation: () => Promise): Promise; } +export type AgentGraphRetirementDisposition = + | { readonly kind: 'clear' } + | { readonly kind: 'quiescent_open' } + | { readonly kind: 'busy'; readonly status: 'active' | 'waiting' | 'closing' }; + interface GraphDriver { rootSessionId: string; graphId: string; @@ -391,6 +396,32 @@ export class AgentGraphCoordinator { return (await this.readSessionState(rootSessionId)) === 'live'; } + /** + * Classify durable graph state for Session retirement without changing the + * broader live-state semantics used by recovery and graph epoch selection. + */ + async readRetirementDisposition(rootSessionId: string): Promise { + const snapshot = buildAgentGraphClientSnapshot( + await this.#readClientModelInputForGraph( + rootSessionId, + await this.currentGraphId(rootSessionId), + ), + ); + if (snapshot.scheduleRevision === 0) return { kind: 'clear' }; + switch (snapshot.status) { + case 'empty': + case 'completed': + return { kind: 'clear' }; + case 'stopped': + case 'failed': + return { kind: 'quiescent_open' }; + case 'active': + case 'waiting': + case 'closing': + return { kind: 'busy', status: snapshot.status }; + } + } + /** * Reconstruct one stable, reference-only control/data-plane timeline page. *