agent-radar
7be535059e
Dependency audit / audit (push) Canceled after 0s
CI / plan (push) Canceled after 0s
Gitoxide helper admission / macos-latest (push) Canceled after 0s
Gitoxide helper admission / ubuntu-latest (push) Canceled after 0s
Gitoxide helper admission / windows-latest (push) Canceled after 0s
Runtime Host peer admission / quality (push) Canceled after 0s
Windows recovery / windows_recovery (push) Canceled after 0s
CI / heavy (push) Canceled after 0s
CI / test (push) Canceled after 0s
923 lines
36 KiB
TypeScript
923 lines
36 KiB
TypeScript
/*
|
|
* 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 { isWorkHubCoordinationSessionTarget, sessionRevisionFamilyId } from '@maka/core/session';
|
|
import type {
|
|
SubagentWorkspaceBinding,
|
|
SubagentWorktreeExecutor,
|
|
} from '@maka/core/subagent-workspace';
|
|
import type { InteractiveArtifactStoreWriter } from '@maka/storage/artifact-stores';
|
|
import {
|
|
isSessionNotFoundError,
|
|
SessionMetadataConflictError,
|
|
SessionMetadataVersionConflictError,
|
|
type ExecutionSessionWriter,
|
|
type SessionHeaderSnapshot,
|
|
} from '@maka/storage/execution-stores';
|
|
import { type SessionManager } from '@maka/runtime/session-manager';
|
|
import type { InteractiveSessionTodoWriter } from '@maka/storage/session-todo-authority';
|
|
import {
|
|
type OperationOutcome,
|
|
type SessionCatalogItem,
|
|
type SessionLifecycleSetInput,
|
|
type SessionRemoveInput,
|
|
type SessionRemovePreviewInput,
|
|
type SessionRemoveResult,
|
|
} from '../protocol/index.js';
|
|
import {
|
|
HostScheduledTaskSessionBusyError,
|
|
type HostScheduledTaskSessionRetirement,
|
|
} from './scheduled-task-coordinator.js';
|
|
import type { HostClientCapabilityCoordinator } from './client-capability-coordinator.js';
|
|
import type { HostGoalCoordinator, HostGoalSessionRetirement } from './goal-coordinator.js';
|
|
import type { HostInteractionCoordinator } from './interaction-coordinator.js';
|
|
import type { HostMessageCoordinator } from './message-coordinator.js';
|
|
import type { SessionRetirementOperationHandlerMap } from './operation-dispatcher.js';
|
|
import { projectSessionCatalogRecord } from './session-catalog-coordinator.js';
|
|
import { type SessionAdmissionLease, SessionAdmissionGate } from './session-admission-gate.js';
|
|
import type { SessionContinuityCoordinator } from './session-continuity-coordinator.js';
|
|
import type { RootTurnCoordinator } from './root-turn-coordinator.js';
|
|
import type { HostRuntimeResourceCoordinator } from './runtime-resource-coordinator.js';
|
|
import { purgeSessionSidecars } from './session-sidecar-purge.js';
|
|
import type { MemoryExtractionSessionLane } from './memory-extraction-session-lane.js';
|
|
|
|
const FAMILY_STABILIZATION_ATTEMPTS = 4;
|
|
|
|
type RetirementStores = Pick<
|
|
ExecutionSessionWriter,
|
|
| 'listHeaders'
|
|
| 'probeSessionRemoval'
|
|
| 'readCatalogRecord'
|
|
| 'readHeaderRecordSnapshot'
|
|
| 'reconcileOrphanedAgentGraphRetirements'
|
|
| 'listPendingSessionRetirementCleanupIds'
|
|
| 'completeSessionRetirementCleanup'
|
|
| 'removeSessionsVersioned'
|
|
| 'setSessionsArchivedVersioned'
|
|
>;
|
|
|
|
type RetirementRoot = Pick<RootTurnCoordinator, 'readRootState'>;
|
|
type RetirementMessages = Pick<HostMessageCoordinator, 'hasLiveSessionState' | 'retireSessions'>;
|
|
type RetirementInteractions = Pick<HostInteractionCoordinator, 'hasPendingSession'>;
|
|
type RetirementGoals = Pick<
|
|
HostGoalCoordinator,
|
|
'beginSessionRetirement' | 'hasLiveGoal' | 'unarchiveSessions'
|
|
>;
|
|
type RetirementResources = Pick<HostRuntimeResourceCoordinator, 'hasLiveSessionResources'>;
|
|
type RetirementSessionEffects = {
|
|
hasLiveSessionState(sessionId: string): boolean;
|
|
};
|
|
type RetirementGraph = {
|
|
hasLiveSessionState(sessionId: string): Promise<boolean>;
|
|
listGraphIds(sessionId: string): Promise<readonly string[]>;
|
|
};
|
|
type RetirementGraphWake = {
|
|
hasLiveSessionState(sessionId: string): boolean;
|
|
retireSessions(sessionIds: readonly string[]): Promise<number>;
|
|
};
|
|
type RetirementManager = Pick<
|
|
SessionManager,
|
|
'disposeSessionBackend' | 'finalizeChildWorkspacePatches'
|
|
>;
|
|
type RetirementCapabilities = Pick<HostClientCapabilityCoordinator, 'retireSessions'>;
|
|
type RetirementContinuity = Pick<
|
|
SessionContinuityCoordinator,
|
|
'refreshCanonical' | 'retireSessions'
|
|
>;
|
|
|
|
export interface HostSessionRetirementCoordinatorOptions {
|
|
readonly stores: RetirementStores;
|
|
readonly admission: SessionAdmissionGate;
|
|
readonly root: RetirementRoot;
|
|
readonly messages: RetirementMessages;
|
|
readonly interactions: RetirementInteractions;
|
|
readonly goals: RetirementGoals;
|
|
readonly scheduledTasks: {
|
|
beginSessionRetirement(
|
|
sessionIds: readonly string[],
|
|
): Promise<HostScheduledTaskSessionRetirement>;
|
|
};
|
|
readonly resources: RetirementResources;
|
|
readonly sessionEffects: RetirementSessionEffects;
|
|
readonly graph: RetirementGraph;
|
|
readonly graphWake: RetirementGraphWake;
|
|
readonly manager: RetirementManager;
|
|
readonly capabilities: RetirementCapabilities;
|
|
readonly continuity: RetirementContinuity;
|
|
readonly artifacts: Pick<InteractiveArtifactStoreWriter, 'purgeSessionArtifacts'>;
|
|
readonly sessionTodo: Pick<InteractiveSessionTodoWriter, 'purgeSessionState'>;
|
|
readonly assertNoContextOffloadReferences?: (sessionIds: readonly string[]) => Promise<void>;
|
|
readonly purgeOperationalState: (sessionId: string) => Promise<void>;
|
|
readonly purgeAgentGraphState: (sessionId: string) => Promise<void>;
|
|
readonly worktrees?: Pick<SubagentWorktreeExecutor, 'retire'>;
|
|
readonly requestDrain: () => void;
|
|
readonly memoryExtractionLane: MemoryExtractionSessionLane;
|
|
}
|
|
|
|
interface StableFamily {
|
|
readonly sessionIds: readonly string[];
|
|
readonly records: ReadonlyMap<string, SessionHeaderSnapshot>;
|
|
readonly admission: SessionAdmissionLease;
|
|
}
|
|
|
|
interface StableRemovalPlan {
|
|
readonly remove: StableFamily;
|
|
readonly archive: StableFamily;
|
|
}
|
|
|
|
interface RetirementHandles {
|
|
readonly goal: HostGoalSessionRetirement;
|
|
readonly scheduledTasks: HostScheduledTaskSessionRetirement;
|
|
}
|
|
|
|
class RetryFamilyResolution extends Error {
|
|
readonly name = 'RetryFamilyResolution';
|
|
|
|
constructor(readonly sessionIds: readonly string[]) {
|
|
super('Session revision family changed before retirement admission');
|
|
}
|
|
}
|
|
|
|
class RetryRemovalPlanResolution extends Error {
|
|
readonly name = 'RetryRemovalPlanResolution';
|
|
|
|
constructor(
|
|
readonly removeSessionIds: readonly string[],
|
|
readonly archiveSessionIds: readonly string[],
|
|
readonly archiveGuardSessionIds: readonly string[],
|
|
) {
|
|
super('Session removal plan changed before retirement admission');
|
|
}
|
|
}
|
|
|
|
class SessionRetirementBusyError extends Error {
|
|
readonly name = 'SessionRetirementBusyError';
|
|
}
|
|
|
|
/** Host-owned archive, unarchive, remove, and revision-family commit authority. */
|
|
export class HostSessionRetirementCoordinator {
|
|
readonly handlers: SessionRetirementOperationHandlerMap = {
|
|
'session.lifecycle.set': (input) => this.#setLifecycle(input),
|
|
'session.remove': (input) => this.#remove(input),
|
|
'session.remove.preview': (input) => this.#previewRemoval(input),
|
|
};
|
|
|
|
readonly #stores: RetirementStores;
|
|
readonly #admission: SessionAdmissionGate;
|
|
readonly #root: RetirementRoot;
|
|
readonly #messages: RetirementMessages;
|
|
readonly #interactions: RetirementInteractions;
|
|
readonly #goals: RetirementGoals;
|
|
readonly #scheduledTasks: HostSessionRetirementCoordinatorOptions['scheduledTasks'];
|
|
readonly #resources: RetirementResources;
|
|
readonly #sessionEffects: RetirementSessionEffects;
|
|
readonly #graph: RetirementGraph;
|
|
readonly #graphWake: RetirementGraphWake;
|
|
readonly #manager: RetirementManager;
|
|
readonly #capabilities: RetirementCapabilities;
|
|
readonly #continuity: RetirementContinuity;
|
|
readonly #artifacts: HostSessionRetirementCoordinatorOptions['artifacts'];
|
|
readonly #sessionTodo: HostSessionRetirementCoordinatorOptions['sessionTodo'];
|
|
readonly #assertNoContextOffloadReferences: HostSessionRetirementCoordinatorOptions['assertNoContextOffloadReferences'];
|
|
readonly #purgeOperationalState: HostSessionRetirementCoordinatorOptions['purgeOperationalState'];
|
|
readonly #purgeAgentGraphState: HostSessionRetirementCoordinatorOptions['purgeAgentGraphState'];
|
|
readonly #worktrees: HostSessionRetirementCoordinatorOptions['worktrees'];
|
|
readonly #requestDrain: () => void;
|
|
readonly #memoryExtractionLane: MemoryExtractionSessionLane;
|
|
readonly #cleanupQueue = new Set<string>();
|
|
readonly #retiredWorktrees = new Map<string, SubagentWorkspaceBinding>();
|
|
#cleanupWorker: Promise<void> | null = null;
|
|
#closing = false;
|
|
|
|
constructor(options: HostSessionRetirementCoordinatorOptions) {
|
|
this.#stores = options.stores;
|
|
this.#admission = options.admission;
|
|
this.#root = options.root;
|
|
this.#messages = options.messages;
|
|
this.#interactions = options.interactions;
|
|
this.#goals = options.goals;
|
|
this.#scheduledTasks = options.scheduledTasks;
|
|
this.#resources = options.resources;
|
|
this.#sessionEffects = options.sessionEffects;
|
|
this.#graph = options.graph;
|
|
this.#graphWake = options.graphWake;
|
|
this.#manager = options.manager;
|
|
this.#capabilities = options.capabilities;
|
|
this.#continuity = options.continuity;
|
|
this.#artifacts = options.artifacts;
|
|
this.#sessionTodo = options.sessionTodo;
|
|
this.#assertNoContextOffloadReferences = options.assertNoContextOffloadReferences;
|
|
this.#purgeOperationalState = options.purgeOperationalState;
|
|
this.#purgeAgentGraphState = options.purgeAgentGraphState;
|
|
this.#worktrees = options.worktrees;
|
|
this.#requestDrain = options.requestDrain;
|
|
this.#memoryExtractionLane = options.memoryExtractionLane;
|
|
}
|
|
|
|
async recover(): Promise<void> {
|
|
await this.#stores.reconcileOrphanedAgentGraphRetirements();
|
|
await this.#reconcileOrphanedSubagentArchives();
|
|
this.#scheduleCleanup(await this.#stores.listPendingSessionRetirementCleanupIds());
|
|
}
|
|
|
|
async close(): Promise<void> {
|
|
this.#closing = true;
|
|
await this.#cleanupWorker;
|
|
}
|
|
|
|
async #setLifecycle(
|
|
input: SessionLifecycleSetInput,
|
|
): Promise<OperationOutcome<'session.lifecycle.set'>> {
|
|
try {
|
|
return await this.#withStableFamily(input.sessionId, async (family) => {
|
|
const target = requireFamilyRecord(family, input.sessionId);
|
|
const archived = input.state === 'archived';
|
|
if ([...family.records.values()].every(({ header }) => header.isArchived === archived)) {
|
|
return lifecycleSuccess(
|
|
projectSessionCatalogRecord(await this.#stores.readCatalogRecord(input.sessionId)),
|
|
);
|
|
}
|
|
|
|
if (!archived) {
|
|
let committed = false;
|
|
try {
|
|
await this.#stores.setSessionsArchivedVersioned(versionedFamily(family), false);
|
|
committed = true;
|
|
this.#goals.unarchiveSessions(family.sessionIds);
|
|
await this.#refreshFamily(family);
|
|
return lifecycleSuccess(
|
|
projectSessionCatalogRecord(await this.#stores.readCatalogRecord(input.sessionId)),
|
|
);
|
|
} catch (error) {
|
|
if (committed) return this.#uncertainLifecycle('unarchive');
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
let handles: RetirementHandles | undefined;
|
|
let committed = false;
|
|
try {
|
|
handles = await this.#prepareRetirement(family, 'archive');
|
|
await this.#finalizeWorkspacePatches(family.sessionIds);
|
|
await this.#disposeBackends(family.sessionIds);
|
|
const committable = await this.#refreshFamilyRecords(family);
|
|
await this.#stores.setSessionsArchivedVersioned(versionedFamily(committable), true);
|
|
committed = true;
|
|
handles.goal.commit();
|
|
handles.scheduledTasks.commit();
|
|
await this.#graphWake.retireSessions(family.sessionIds);
|
|
this.#capabilities.retireSessions(family.sessionIds);
|
|
this.#messages.retireSessions(family.sessionIds);
|
|
await this.#refreshFamily(family);
|
|
return lifecycleSuccess(
|
|
projectSessionCatalogRecord(await this.#stores.readCatalogRecord(target.header.id)),
|
|
);
|
|
} catch (error) {
|
|
if (committed) return this.#uncertainLifecycle('archive');
|
|
handles?.goal.rollback();
|
|
handles?.scheduledTasks.rollback();
|
|
throw error;
|
|
}
|
|
});
|
|
} catch (error) {
|
|
return this.#lifecycleFailure(error);
|
|
}
|
|
}
|
|
|
|
async #remove(input: SessionRemoveInput): Promise<OperationOutcome<'session.remove'>> {
|
|
let probe;
|
|
try {
|
|
probe = await this.#stores.probeSessionRemoval(input.sessionId);
|
|
} catch {
|
|
return removeFailure('persistence_failed', 'Session removal state is unavailable');
|
|
}
|
|
if (probe.kind === 'removed') {
|
|
try {
|
|
this.#scheduleCleanup(
|
|
await this.#stores.listPendingSessionRetirementCleanupIds(input.sessionId),
|
|
);
|
|
} catch {
|
|
this.#requestDrain();
|
|
}
|
|
return removeSuccess(input.sessionId);
|
|
}
|
|
if (probe.kind === 'absent') return removeFailure('not_found', 'Session does not exist');
|
|
|
|
try {
|
|
return await this.#withStableRemovalPlan(input.sessionId, async (plan) => {
|
|
const target = requireFamilyRecord(plan.remove, input.sessionId);
|
|
if (target.revision !== input.expectedRevision) {
|
|
return removeOutcome({
|
|
kind: 'revision_conflict',
|
|
expectedRevision: input.expectedRevision,
|
|
actualRevision: target.revision,
|
|
});
|
|
}
|
|
|
|
let removeHandles: RetirementHandles | undefined;
|
|
let archiveHandles: RetirementHandles | undefined;
|
|
let committed = false;
|
|
try {
|
|
removeHandles = await this.#prepareRetirement(plan.remove, 'remove');
|
|
if (plan.archive.sessionIds.length > 0) {
|
|
archiveHandles = await this.#prepareRetirement(plan.archive, 'archive');
|
|
}
|
|
await this.#assertNoContextOffloadReferences?.(plan.remove.sessionIds);
|
|
const allSessionIds = [...plan.remove.sessionIds, ...plan.archive.sessionIds];
|
|
await this.#finalizeWorkspacePatches(allSessionIds);
|
|
await this.#disposeBackends(allSessionIds);
|
|
const committableRemove = await this.#refreshFamilyRecords(plan.remove);
|
|
const committableArchive = await this.#refreshFamilyRecords(plan.archive);
|
|
const removedSessionIds = await this.#stores.removeSessionsVersioned(
|
|
versionedFamily(committableRemove),
|
|
versionedFamily(committableArchive),
|
|
);
|
|
committed = true;
|
|
removeHandles.goal.commit();
|
|
removeHandles.scheduledTasks.commit();
|
|
archiveHandles?.goal.commit();
|
|
archiveHandles?.scheduledTasks.commit();
|
|
await this.#graphWake.retireSessions(allSessionIds);
|
|
this.#rememberRetiredWorktrees(committableRemove, removedSessionIds);
|
|
this.#scheduleCleanup(removedSessionIds);
|
|
this.#capabilities.retireSessions(allSessionIds);
|
|
this.#messages.retireSessions(allSessionIds);
|
|
await this.#continuity.retireSessions(plan.remove.sessionIds, plan.remove.admission);
|
|
await this.#refreshFamily(plan.archive);
|
|
// What the confirm warned about, now executed: the distinct subtasks
|
|
// (deduplicated by revision family) this removal moved to the archive.
|
|
// The renderer reports this verbatim rather than re-deriving the plan.
|
|
const archivedSubtaskCount = new Set(
|
|
plan.archive.sessionIds.map((id) =>
|
|
sessionRevisionFamilyId(requireFamilyRecord(plan.archive, id).header),
|
|
),
|
|
).size;
|
|
return removeSuccess(input.sessionId, archivedSubtaskCount);
|
|
} catch (error) {
|
|
if (committed) return this.#uncertainRemove();
|
|
archiveHandles?.goal.rollback();
|
|
archiveHandles?.scheduledTasks.rollback();
|
|
removeHandles?.goal.rollback();
|
|
removeHandles?.scheduledTasks.rollback();
|
|
throw error;
|
|
}
|
|
});
|
|
} catch (error) {
|
|
return this.#removeFailure(error, input);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Read-only preview of how many subtasks a delete of this parent would move
|
|
* to the archive — the confirm warns off this so the renderer never has to
|
|
* re-derive the plan from a catalog projection that lacks the operator marker
|
|
* and the copy state. Absent or already-removed targets, and Agent Graph
|
|
* operators (which retire with their root rather than archive), preview zero.
|
|
*/
|
|
async #previewRemoval(
|
|
input: SessionRemovePreviewInput,
|
|
): Promise<OperationOutcome<'session.remove.preview'>> {
|
|
let probe;
|
|
try {
|
|
probe = await this.#stores.probeSessionRemoval(input.sessionId);
|
|
} catch {
|
|
return previewFailure('persistence_failed', 'Session removal state is unavailable');
|
|
}
|
|
if (probe.kind !== 'present') return previewSuccess(0);
|
|
try {
|
|
const plan = await this.#readRemovalPlanSessionIds(input.sessionId);
|
|
return previewSuccess(plan.archivableSubtaskCount);
|
|
} catch (error) {
|
|
// A graph operator has no independent delete and archives nothing; a
|
|
// target that vanished mid-read has nothing left to archive either.
|
|
if (
|
|
error instanceof SessionMetadataConflictError ||
|
|
error instanceof SessionRetirementMissingSessionError
|
|
) {
|
|
return previewSuccess(0);
|
|
}
|
|
return previewFailure('persistence_failed', 'Session removal plan is unavailable');
|
|
}
|
|
}
|
|
|
|
async #withStableRemovalPlan<T>(
|
|
sessionId: string,
|
|
operation: (plan: StableRemovalPlan) => Promise<T>,
|
|
): Promise<T> {
|
|
// Only the id sets stabilize here; the archivable-subtask count is a
|
|
// preview-only read, so it is intentionally not threaded through the retry.
|
|
let planIds: {
|
|
removeSessionIds: readonly string[];
|
|
archiveSessionIds: readonly string[];
|
|
archiveGuardSessionIds: readonly string[];
|
|
} = await this.#readRemovalPlanSessionIds(sessionId);
|
|
for (let attempt = 0; attempt < FAMILY_STABILIZATION_ATTEMPTS; attempt += 1) {
|
|
const allSessionIds = [
|
|
...planIds.removeSessionIds,
|
|
...planIds.archiveSessionIds,
|
|
...planIds.archiveGuardSessionIds,
|
|
].sort();
|
|
try {
|
|
return await this.#memoryExtractionLane.runMany(allSessionIds, () =>
|
|
this.#admission.runMany(allSessionIds, async (admission) => {
|
|
const stableIds = await this.#readRemovalPlanSessionIds(sessionId);
|
|
if (
|
|
!sameIds(planIds.removeSessionIds, stableIds.removeSessionIds) ||
|
|
!sameIds(planIds.archiveSessionIds, stableIds.archiveSessionIds) ||
|
|
!sameIds(planIds.archiveGuardSessionIds, stableIds.archiveGuardSessionIds)
|
|
) {
|
|
throw new RetryRemovalPlanResolution(
|
|
stableIds.removeSessionIds,
|
|
stableIds.archiveSessionIds,
|
|
stableIds.archiveGuardSessionIds,
|
|
);
|
|
}
|
|
const snapshots = await Promise.all(
|
|
allSessionIds.map((id) => this.#stores.readHeaderRecordSnapshot(id)),
|
|
);
|
|
const records = new Map(
|
|
allSessionIds.map((id, index) => [id, snapshots[index]!] as const),
|
|
);
|
|
return operation({
|
|
remove: stableFamily(planIds.removeSessionIds, records, admission),
|
|
archive: stableFamily(planIds.archiveSessionIds, records, admission),
|
|
});
|
|
}),
|
|
);
|
|
} catch (error) {
|
|
if (!(error instanceof RetryRemovalPlanResolution)) throw error;
|
|
planIds = {
|
|
removeSessionIds: [...error.removeSessionIds],
|
|
archiveSessionIds: [...error.archiveSessionIds],
|
|
archiveGuardSessionIds: [...error.archiveGuardSessionIds],
|
|
};
|
|
}
|
|
}
|
|
throw new SessionMetadataConflictError(
|
|
'Session removal plan kept changing during retirement admission',
|
|
);
|
|
}
|
|
|
|
async #withStableFamily<T>(
|
|
sessionId: string,
|
|
operation: (family: StableFamily) => Promise<T>,
|
|
): Promise<T> {
|
|
let sessionIds = await this.#readFamilySessionIds(sessionId);
|
|
for (let attempt = 0; attempt < FAMILY_STABILIZATION_ATTEMPTS; attempt += 1) {
|
|
try {
|
|
return await this.#memoryExtractionLane.runMany(sessionIds, () =>
|
|
this.#admission.runMany(sessionIds, async (admission) => {
|
|
const stableIds = await this.#readFamilySessionIds(sessionId);
|
|
if (!sameIds(sessionIds, stableIds)) throw new RetryFamilyResolution(stableIds);
|
|
const snapshots = await Promise.all(
|
|
stableIds.map((id) => this.#stores.readHeaderRecordSnapshot(id)),
|
|
);
|
|
return operation({
|
|
sessionIds: stableIds,
|
|
records: new Map(stableIds.map((id, index) => [id, snapshots[index]!])),
|
|
admission,
|
|
});
|
|
}),
|
|
);
|
|
} catch (error) {
|
|
if (!(error instanceof RetryFamilyResolution)) throw error;
|
|
sessionIds = [...error.sessionIds];
|
|
}
|
|
}
|
|
throw new SessionMetadataConflictError(
|
|
'Session revision family kept changing during retirement admission',
|
|
);
|
|
}
|
|
|
|
async #readFamilySessionIds(sessionId: string): Promise<string[]> {
|
|
const target = await this.#stores.probeSessionRemoval(sessionId);
|
|
if (target.kind !== 'present') {
|
|
throw new SessionRetirementMissingSessionError(target.kind);
|
|
}
|
|
if (isWorkHubCoordinationSessionTarget(target.record.header)) {
|
|
throw new SessionMetadataConflictError(
|
|
'WorkHub Coordination Session lifecycle is owned by WorkHub',
|
|
);
|
|
}
|
|
if (target.record.header.subagentParent?.graph) {
|
|
throw new SessionMetadataConflictError(
|
|
'Agent Graph operator Sessions retire with their root Session',
|
|
);
|
|
}
|
|
const familyId = sessionRevisionFamilyId(target.record.header);
|
|
const headers = await this.#stores.listHeaders();
|
|
const roots = headers.filter(
|
|
(header) =>
|
|
header.conversationCopy?.state !== 'preparing' &&
|
|
!header.subagentParent &&
|
|
sessionRevisionFamilyId(header) === familyId,
|
|
);
|
|
const graphRoots = new Map(
|
|
await Promise.all(
|
|
roots.map(
|
|
async (header) => [header.id, await this.#graph.listGraphIds(header.id)] as const,
|
|
),
|
|
),
|
|
);
|
|
const members = headers
|
|
.filter((header) => {
|
|
if (header.conversationCopy?.state === 'preparing') return false;
|
|
if (sessionRevisionFamilyId(header) === familyId) return true;
|
|
const parent = header.subagentParent;
|
|
return (
|
|
parent?.graph !== undefined &&
|
|
graphRoots.get(parent.parentSessionId)?.includes(parent.graph.graphId) === true
|
|
);
|
|
})
|
|
.map((header) => header.id);
|
|
if (!members.includes(sessionId)) members.push(sessionId);
|
|
return [...new Set(members)].sort();
|
|
}
|
|
|
|
async #readRemovalPlanSessionIds(sessionId: string): Promise<{
|
|
removeSessionIds: readonly string[];
|
|
archiveSessionIds: readonly string[];
|
|
archiveGuardSessionIds: readonly string[];
|
|
archivableSubtaskCount: number;
|
|
}> {
|
|
const removeSessionIds = await this.#readFamilySessionIds(sessionId);
|
|
const removeIds = new Set(removeSessionIds);
|
|
const headers = await this.#stores.listHeaders();
|
|
const ordinaryParentIds = new Set(
|
|
headers
|
|
.filter((header) => removeIds.has(header.id) && !header.subagentParent?.graph)
|
|
.map((header) => header.id),
|
|
);
|
|
const childFamilyIds = new Set(
|
|
headers
|
|
.filter(
|
|
(header) =>
|
|
header.conversationCopy?.state !== 'preparing' &&
|
|
header.subagentParent !== undefined &&
|
|
header.subagentParent.graph === undefined &&
|
|
ordinaryParentIds.has(header.subagentParent.parentSessionId),
|
|
)
|
|
.map(sessionRevisionFamilyId),
|
|
);
|
|
const childSessionHeaders = headers.filter(
|
|
(header) =>
|
|
header.conversationCopy?.state !== 'preparing' &&
|
|
!removeIds.has(header.id) &&
|
|
childFamilyIds.has(sessionRevisionFamilyId(header)),
|
|
);
|
|
const archiveHeaders = childSessionHeaders.filter((header) => !header.isArchived);
|
|
const archiveSessionIds = archiveHeaders.map((header) => header.id);
|
|
const archiveGuardSessionIds = childSessionHeaders
|
|
.filter((header) => header.isArchived)
|
|
.map((header) => header.id);
|
|
return {
|
|
removeSessionIds: [...removeIds].sort(),
|
|
archiveSessionIds: [...new Set(archiveSessionIds)].sort(),
|
|
archiveGuardSessionIds: [...new Set(archiveGuardSessionIds)].sort(),
|
|
// Distinct subtasks (by revision family) that a delete would move to the
|
|
// archive — the count the confirm warns off, matching what `#remove`
|
|
// reports afterwards.
|
|
archivableSubtaskCount: new Set(archiveHeaders.map(sessionRevisionFamilyId)).size,
|
|
};
|
|
}
|
|
|
|
async #reconcileOrphanedSubagentArchives(): Promise<void> {
|
|
const headers = await this.#stores.listHeaders();
|
|
const liveSessionIds = new Set(headers.map((header) => header.id));
|
|
const orphanFamilyIds = new Set(
|
|
headers
|
|
.filter(
|
|
(header) =>
|
|
!header.isArchived &&
|
|
header.subagentParent !== undefined &&
|
|
header.subagentParent.graph === undefined &&
|
|
!liveSessionIds.has(header.subagentParent.parentSessionId),
|
|
)
|
|
.map(sessionRevisionFamilyId),
|
|
);
|
|
const orphanSessionIds = headers
|
|
.filter(
|
|
(header) =>
|
|
header.conversationCopy?.state !== 'preparing' &&
|
|
orphanFamilyIds.has(sessionRevisionFamilyId(header)),
|
|
)
|
|
.map((header) => header.id)
|
|
.sort();
|
|
if (orphanSessionIds.length === 0) return;
|
|
const records = await Promise.all(
|
|
orphanSessionIds.map((sessionId) => this.#stores.readHeaderRecordSnapshot(sessionId)),
|
|
);
|
|
await this.#stores.setSessionsArchivedVersioned(
|
|
records.map((record) => ({
|
|
sessionId: record.header.id,
|
|
expectedVersion: record.revision,
|
|
})),
|
|
true,
|
|
);
|
|
}
|
|
|
|
async #prepareRetirement(
|
|
family: StableFamily,
|
|
kind: 'archive' | 'remove',
|
|
): Promise<RetirementHandles> {
|
|
for (const sessionId of family.sessionIds) {
|
|
if (this.#root.readRootState(sessionId).kind !== 'idle') {
|
|
throw new SessionRetirementBusyError(
|
|
`Session ${sessionId} has an active or reserved root Turn`,
|
|
);
|
|
}
|
|
if (this.#messages.hasLiveSessionState(sessionId)) {
|
|
throw new SessionRetirementBusyError(
|
|
`Session ${sessionId} has queued or in-flight Messages`,
|
|
);
|
|
}
|
|
if (await this.#interactions.hasPendingSession(sessionId)) {
|
|
throw new SessionRetirementBusyError(`Session ${sessionId} has a pending Interaction`);
|
|
}
|
|
if (this.#goals.hasLiveGoal(sessionId)) {
|
|
throw new SessionRetirementBusyError(`Session ${sessionId} has a live Goal`);
|
|
}
|
|
if (await this.#resources.hasLiveSessionResources(sessionId)) {
|
|
throw new SessionRetirementBusyError(`Session ${sessionId} has a live Runtime Resource`);
|
|
}
|
|
if (this.#sessionEffects.hasLiveSessionState(sessionId)) {
|
|
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 && this.#graphWake.hasLiveSessionState(sessionId)) {
|
|
throw new SessionRetirementBusyError(
|
|
`Session ${sessionId} has an active Agent Graph supervisor wake`,
|
|
);
|
|
}
|
|
}
|
|
|
|
const scheduledTasks = await this.#scheduledTasks.beginSessionRetirement(family.sessionIds);
|
|
try {
|
|
const goal = await this.#goals.beginSessionRetirement(family.sessionIds, kind);
|
|
return { goal, scheduledTasks };
|
|
} catch (error) {
|
|
scheduledTasks.rollback();
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async #disposeBackends(sessionIds: readonly string[]): Promise<void> {
|
|
const outcomes = await Promise.allSettled(
|
|
sessionIds.map((sessionId) => this.#manager.disposeSessionBackend(sessionId)),
|
|
);
|
|
const failures = outcomes.flatMap((outcome) =>
|
|
outcome.status === 'rejected' ? [outcome.reason] : [],
|
|
);
|
|
if (failures.length === 0) return;
|
|
this.#requestDrain();
|
|
throw new AggregateError(failures, 'Session backend disposal failed during retirement');
|
|
}
|
|
|
|
async #finalizeWorkspacePatches(sessionIds: readonly string[]): Promise<void> {
|
|
for (const sessionId of sessionIds) {
|
|
await this.#manager.finalizeChildWorkspacePatches(sessionId);
|
|
}
|
|
}
|
|
|
|
#scheduleCleanup(sessionIds: readonly string[]): void {
|
|
if (this.#closing) return;
|
|
for (const sessionId of sessionIds) this.#cleanupQueue.add(sessionId);
|
|
if (this.#cleanupWorker || this.#cleanupQueue.size === 0) return;
|
|
const worker = this.#drainCleanup();
|
|
this.#cleanupWorker = worker;
|
|
void worker.then(
|
|
() => this.#finishCleanup(worker),
|
|
() => this.#finishCleanup(worker),
|
|
);
|
|
}
|
|
|
|
async #drainCleanup(): Promise<void> {
|
|
while (this.#cleanupQueue.size > 0) {
|
|
const batch = [...this.#cleanupQueue];
|
|
this.#cleanupQueue.clear();
|
|
await Promise.allSettled(batch.map((sessionId) => this.#cleanupRetiredSession(sessionId)));
|
|
}
|
|
}
|
|
|
|
async #cleanupRetiredSession(sessionId: string): Promise<void> {
|
|
const worktree = this.#retiredWorktrees.get(sessionId);
|
|
const outcomes = await Promise.allSettled([
|
|
purgeSessionSidecars(
|
|
{
|
|
artifacts: this.#artifacts,
|
|
sessionTodo: this.#sessionTodo,
|
|
purgeOperationalState: this.#purgeOperationalState,
|
|
},
|
|
sessionId,
|
|
),
|
|
this.#purgeAgentGraphState(sessionId),
|
|
...(worktree && this.#worktrees ? [this.#worktrees.retire(worktree)] : []),
|
|
]);
|
|
const failures = outcomes.flatMap((outcome) =>
|
|
outcome.status === 'rejected' ? [outcome.reason] : [],
|
|
);
|
|
if (failures.length > 0) {
|
|
throw new AggregateError(failures, `Session ${sessionId} retirement cleanup failed`);
|
|
}
|
|
await this.#stores.completeSessionRetirementCleanup(sessionId);
|
|
this.#retiredWorktrees.delete(sessionId);
|
|
}
|
|
|
|
#rememberRetiredWorktrees(family: StableFamily, removedSessionIds: readonly string[]): void {
|
|
for (const sessionId of removedSessionIds) {
|
|
const binding = family.records.get(sessionId)?.header.subagentWorkspace;
|
|
if (binding) this.#retiredWorktrees.set(sessionId, binding);
|
|
}
|
|
}
|
|
|
|
#finishCleanup(worker: Promise<void>): void {
|
|
if (this.#cleanupWorker !== worker) return;
|
|
this.#cleanupWorker = null;
|
|
if (!this.#closing && this.#cleanupQueue.size > 0) this.#scheduleCleanup([]);
|
|
}
|
|
|
|
async #refreshFamilyRecords(family: StableFamily): Promise<StableFamily> {
|
|
const snapshots = await Promise.all(
|
|
family.sessionIds.map((sessionId) => this.#stores.readHeaderRecordSnapshot(sessionId)),
|
|
);
|
|
return {
|
|
...family,
|
|
records: new Map(family.sessionIds.map((sessionId, index) => [sessionId, snapshots[index]!])),
|
|
};
|
|
}
|
|
|
|
async #refreshFamily(family: StableFamily): Promise<void> {
|
|
for (const sessionId of family.sessionIds) {
|
|
await this.#continuity.refreshCanonical(sessionId, family.admission);
|
|
}
|
|
}
|
|
|
|
#uncertainLifecycle(kind: 'archive' | 'unarchive') {
|
|
this.#requestDrain();
|
|
return lifecycleFailure(
|
|
'commit_outcome_unknown',
|
|
`Session ${kind} committed but publication is uncertain`,
|
|
);
|
|
}
|
|
|
|
#lifecycleFailure(error: unknown): OperationOutcome<'session.lifecycle.set'> {
|
|
if (error instanceof SessionRetirementMissingSessionError || isSessionNotFoundError(error)) {
|
|
return lifecycleFailure('not_found', 'Session does not exist');
|
|
}
|
|
if (
|
|
error instanceof SessionRetirementBusyError ||
|
|
error instanceof HostScheduledTaskSessionBusyError
|
|
) {
|
|
return lifecycleFailure('session_busy', error.message);
|
|
}
|
|
if (error instanceof SessionMetadataConflictError) {
|
|
return lifecycleFailure('operation_conflict', error.message);
|
|
}
|
|
return lifecycleFailure('persistence_failed', 'Session lifecycle could not be committed');
|
|
}
|
|
|
|
#uncertainRemove(): OperationOutcome<'session.remove'> {
|
|
this.#requestDrain();
|
|
return removeFailure(
|
|
'commit_outcome_unknown',
|
|
'Session remove committed but publication is uncertain',
|
|
);
|
|
}
|
|
|
|
#removeFailure(error: unknown, input: SessionRemoveInput): OperationOutcome<'session.remove'> {
|
|
if (error instanceof SessionRetirementMissingSessionError) {
|
|
return error.state === 'removed'
|
|
? removeSuccess(input.sessionId)
|
|
: removeFailure('not_found', 'Session does not exist');
|
|
}
|
|
if (isSessionNotFoundError(error)) {
|
|
return removeFailure('not_found', 'Session does not exist');
|
|
}
|
|
if (
|
|
error instanceof SessionRetirementBusyError ||
|
|
error instanceof HostScheduledTaskSessionBusyError
|
|
) {
|
|
return removeFailure('session_busy', error.message);
|
|
}
|
|
if (error instanceof SessionMetadataVersionConflictError) {
|
|
if (error.sessionId !== input.sessionId || error.expectedVersion !== input.expectedRevision) {
|
|
return removeFailure(
|
|
'operation_conflict',
|
|
'Session revision family changed during removal',
|
|
);
|
|
}
|
|
return removeOutcome({
|
|
kind: 'revision_conflict',
|
|
expectedRevision: error.expectedVersion,
|
|
actualRevision: error.actualVersion,
|
|
});
|
|
}
|
|
if (error instanceof SessionMetadataConflictError) {
|
|
return removeFailure('operation_conflict', error.message);
|
|
}
|
|
return removeFailure('persistence_failed', 'Session remove could not be committed');
|
|
}
|
|
}
|
|
|
|
class SessionRetirementMissingSessionError extends Error {
|
|
readonly name = 'SessionRetirementMissingSession';
|
|
|
|
constructor(readonly state: 'removed' | 'absent') {
|
|
super(`Session is ${state}`);
|
|
}
|
|
}
|
|
|
|
function sameIds(left: readonly string[], right: readonly string[]): boolean {
|
|
return left.length === right.length && left.every((id, index) => id === right[index]);
|
|
}
|
|
|
|
function requireFamilyRecord(family: StableFamily, sessionId: string): SessionHeaderSnapshot {
|
|
const record = family.records.get(sessionId);
|
|
if (!record) throw new SessionMetadataConflictError('Session left its revision family');
|
|
return record;
|
|
}
|
|
|
|
function stableFamily(
|
|
sessionIds: readonly string[],
|
|
records: ReadonlyMap<string, SessionHeaderSnapshot>,
|
|
admission: SessionAdmissionLease,
|
|
): StableFamily {
|
|
return {
|
|
sessionIds,
|
|
records: new Map(sessionIds.map((sessionId) => [sessionId, records.get(sessionId)!] as const)),
|
|
admission,
|
|
};
|
|
}
|
|
|
|
function versionedFamily(family: StableFamily) {
|
|
return family.sessionIds.map((sessionId) => ({
|
|
sessionId,
|
|
expectedVersion: requireFamilyRecord(family, sessionId).revision,
|
|
}));
|
|
}
|
|
|
|
function lifecycleSuccess(result: SessionCatalogItem): OperationOutcome<'session.lifecycle.set'> {
|
|
return { ok: true, result };
|
|
}
|
|
|
|
function lifecycleFailure(
|
|
code: Extract<OperationOutcome<'session.lifecycle.set'>, { ok: false }>['error']['code'],
|
|
message: string,
|
|
): Extract<OperationOutcome<'session.lifecycle.set'>, { ok: false }> {
|
|
return { ok: false, error: { code, message } };
|
|
}
|
|
|
|
function removeSuccess(
|
|
sessionId: string,
|
|
archivedSubtaskCount = 0,
|
|
): OperationOutcome<'session.remove'> {
|
|
return removeOutcome(
|
|
archivedSubtaskCount > 0
|
|
? { kind: 'removed', sessionId, archivedSubtaskCount }
|
|
: { kind: 'removed', sessionId },
|
|
);
|
|
}
|
|
|
|
function removeOutcome(result: SessionRemoveResult): OperationOutcome<'session.remove'> {
|
|
return { ok: true, result };
|
|
}
|
|
|
|
function removeFailure(
|
|
code: Extract<OperationOutcome<'session.remove'>, { ok: false }>['error']['code'],
|
|
message: string,
|
|
): Extract<OperationOutcome<'session.remove'>, { ok: false }> {
|
|
return { ok: false, error: { code, message } };
|
|
}
|
|
|
|
function previewSuccess(
|
|
archivableSubtaskCount: number,
|
|
): OperationOutcome<'session.remove.preview'> {
|
|
return { ok: true, result: { archivableSubtaskCount } };
|
|
}
|
|
|
|
function previewFailure(
|
|
code: Extract<OperationOutcome<'session.remove.preview'>, { ok: false }>['error']['code'],
|
|
message: string,
|
|
): Extract<OperationOutcome<'session.remove.preview'>, { ok: false }> {
|
|
return { ok: false, error: { code, message } };
|
|
}
|