diff --git a/packages/agent-core-v2/src/_base/di/cascadeEngine.ts b/packages/agent-core-v2/src/_base/di/cascadeEngine.ts new file mode 100644 index 00000000000..34d7d9d6bed --- /dev/null +++ b/packages/agent-core-v2/src/_base/di/cascadeEngine.ts @@ -0,0 +1,836 @@ +/** + * `di` domain — cascade engine + wait scheduler (L2), one per container, with + * tree-wide orchestration (D9: cascades propagate along instance edges across + * scopes). + * + * The dependency graph, request queue, in-flight set, and settle waiters are + * shared by the whole scope tree (`CascadeTree`, owned by the root). Every + * change (provide / unprovide / update) runs as a single transaction + * orchestrated by the engine of the scope where the change was submitted: + * ① compute the contagion set from the tree-global graph; + * ② broadcast WillCascade to the orchestrator's abort hook (bounded wait, + * then forced; failures are best-effort, never a veto); + * ③ tear the contagion set down in global reverse topological order, serially + * (each scope's engine executes its own units; Active → Unloading → + * Pending, or removed for an unprovided token; a descendant scope that dies + * mid-transaction is skipped idempotently); + * ④ apply the change in its own scope (a replace never passes through the + * waiting area); + * ⑤ recheck the waiting area across scopes and rebuild satisfied units in + * global topological order; + * ⑥ append the transaction to the orchestrator's history ring. + * + * Requests serialize through the tree queue; requests queued together merge + * their contagion sets (deduped by scope+token) into one transaction. This is + * one transaction across the tree but not a distributed transaction: a single + * orchestrator, a deterministic order, local execution per scope. Like the + * Ledger, the engine has a sync fast path: with no async abort wait and no + * async disposers, a transaction completes within the tick. + */ + +import { onUnexpectedError } from '../errors/unexpectedError'; +import { isPromiseLike } from '../lifecycle/disposer'; +import type { SyncDescriptor } from './descriptors'; +import { + PairIndex, + type DependencyGraph, + type ScopedToken, +} from './dependencyGraph'; +import { CascadeConflictError } from './errors'; +import type { ServiceIdentifier } from './instantiation'; + +export type UnitState = 'Pending' | 'Activating' | 'Active' | 'Unloading' | 'Failed'; + +export type CascadeAction = 'provide' | 'unprovide' | 'update'; + +/** Eager units activate as soon as their dependencies are satisfied; on-demand units wait for their first resolution (but cascade-torn units always rebuild). */ +export type UnitActivation = 'eager' | 'ondemand'; + +export interface CascadeChange { + readonly action: CascadeAction; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + readonly token: ServiceIdentifier; + readonly descriptor?: SyncDescriptor; + /** Pre-materialized value for a `provide` change (mutually exclusive with `descriptor`). */ + readonly instance?: unknown; + readonly pinned?: boolean; + readonly activation?: UnitActivation; + readonly reason: string; +} + +export interface CascadeHistoryEntry { + readonly seq: number; + readonly reason: string; + readonly changes: ReadonlyArray<{ token: string; action: CascadeAction }>; + readonly affected: readonly string[]; + readonly tornDown: readonly string[]; + readonly rebuilt: readonly string[]; + readonly failed: readonly string[]; + readonly abortWaited: boolean; + readonly abortTimedOut: boolean; + readonly durationMs: number; +} + +export interface CascadeEngineOptions { + /** + * Abort hook (§4.5): invoked at transaction step ② with the contagion set. + * A returned promise is awaited up to `abortWaitMs` (best-effort), then the + * cascade proceeds anyway (forced teardown). + */ + onWillCascade?: ( + affected: readonly ScopedToken[], + reason: string, + ) => void | Promise; + /** Bounded wait for in-flight work to abort (default 5000ms). */ + readonly abortWaitMs?: number; + /** Suspended-resolution timeout (default 30000ms). */ + readonly resolveTimeoutMs?: number; + /** History ring capacity (default 200). */ + readonly historyCapacity?: number; + readonly now?: () => number; +} + +/** Container operations one engine drives for its own scope's units. */ +export interface CascadeHost { + /** Registered in this container or an ancestor. */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + isRegistered(token: ServiceIdentifier): boolean; + /** The container owning this token in this container's chain, if any. */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + ownerScopeOf(token: ServiceIdentifier): object | undefined; + /** Has a live materialized instance in this container. */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + isMaterialized(token: ServiceIdentifier): boolean; + /** Create + cache the instance (throws on construction failure). */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + materialize(token: ServiceIdentifier): unknown; + /** Tear the live instance down and reset the entry to its recipe. */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + retire(token: ServiceIdentifier): void | Promise; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + applyProvide( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + token: ServiceIdentifier, + descriptor: SyncDescriptor, + pinned: boolean | undefined, + ): number; + /** Register a pre-materialized instance (a new generation). */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + applyProvideInstance( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + token: ServiceIdentifier, + instance: unknown, + pinned: boolean | undefined, + ): number; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + applyUnprovide(token: ServiceIdentifier): void; + /** The unit's recipe: the pending descriptor, or the retained one of a live instance. */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + recipeOf(token: ServiceIdentifier): SyncDescriptor | undefined; + /** Constructor-declared (instance-edge) dependencies of a recipe. */ + dependenciesOf( + recipe: SyncDescriptor, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + ): Array>; +} + +/** Structural handle a scoped token's `scope` provides to the engine. */ +export interface CascadeScopeHandle { + readonly cascade: CascadeEngine; + readonly cascadeDisposed: boolean; + /** Distance from the tree root (root = 0); parents always sort shallower. */ + readonly cascadeDepth: number; +} + +interface UnitRecord { + state: UnitState; + error?: unknown; + activation: UnitActivation; + /** True once the unit has had a live instance (torn-down units always rebuild). */ + everActive: boolean; +} + +interface QueuedRequest { + readonly engine: CascadeEngine; + readonly change: CascadeChange; + readonly resolve: () => void; + readonly reject: (error: unknown) => void; +} + +const DEFAULT_ABORT_WAIT_MS = 5000; +const DEFAULT_RESOLVE_TIMEOUT_MS = 30000; +const DEFAULT_HISTORY_CAPACITY = 200; + +/** + * Tree-wide cascade runtime, owned by the root container and shared by every + * engine of the scope tree: the persistent graph, the serialized request + * queue, the in-flight contagion set of the running transaction, and the + * settle waiters for suspended resolutions. + */ +export class CascadeTree { + readonly graph: DependencyGraph; + readonly queue: QueuedRequest[] = []; + /** Every live engine of the tree (engines register at construction). */ + readonly engines = new Set(); + running = false; + /** The scope orchestrating the running transaction (for label rendering). */ + orchestrator: object | undefined; + private readonly _inFlight = new PairIndex(); + private _settleWaiters: Array<() => void> = []; + private readonly _scopeSeq = new Map(); + private _nextScopeSeq = 0; + + constructor(graph: DependencyGraph) { + this.graph = graph; + } + + inFlightSet(ref: ScopedToken, on: boolean): void { + if (on) { + this._inFlight.set(ref.scope, ref.token, true); + } else { + this._inFlight.delete(ref.scope, ref.token); + } + } + + inFlightHas(ref: ScopedToken): boolean { + return this._inFlight.get(ref.scope, ref.token) !== undefined; + } + + inFlightClear(refs: Iterable): void { + for (const ref of refs) { + this._inFlight.delete(ref.scope, ref.token); + } + } + + /** Stable per-scope sequence used to render cross-scope labels (`#n:token`). */ + seqOf(scope: object): number { + let seq = this._scopeSeq.get(scope); + if (seq === undefined) { + seq = this._nextScopeSeq++; + this._scopeSeq.set(scope, seq); + } + return seq; + } + + addSettleWaiter(waiter: () => void): void { + this._settleWaiters.push(waiter); + } + + fireSettleWaiters(): void { + const waiters = this._settleWaiters.splice(0); + for (const waiter of waiters) { + waiter(); + } + } +} + +export class CascadeEngine { + private readonly _units = new Map< + // eslint-disable-next-line @typescript-eslint/no-explicit-any + ServiceIdentifier, + UnitRecord + >(); + /** missing dependency token → waiting unit tokens (§5.5 wake intersection). */ + private readonly _pendingIndex = new Map< + // eslint-disable-next-line @typescript-eslint/no-explicit-any + ServiceIdentifier, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + Set> + >(); + private readonly _history: CascadeHistoryEntry[] = []; + private _historySeq = 0; + private _disposed = false; + + constructor( + private readonly _host: CascadeHost, + private readonly _scope: CascadeScopeHandle, + private readonly _tree: CascadeTree, + private _options: CascadeEngineOptions = {}, + ) { + this._tree.engines.add(this); + } + + /** Merge new options (tests configure hooks/timeouts per scenario). */ + configure(options: CascadeEngineOptions): void { + this._options = { ...this._options, ...options }; + } + + /** State of an engine-tracked unit; undefined for tokens the engine never saw. */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + unitState(token: ServiceIdentifier): UnitState | undefined { + return this._units.get(token)?.state; + } + + /** The sticky failure of a Failed unit. */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + unitFailure(token: ServiceIdentifier): unknown { + const unit = this._units.get(token); + return unit?.state === 'Failed' ? unit.error : undefined; + } + + /** True while the scoped token sits inside the running transaction's contagion set. */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + isInFlight(token: ServiceIdentifier): boolean { + const owner = this._host.ownerScopeOf(token) ?? this._scope; + return this._tree.inFlightHas({ scope: owner, token }); + } + + history(): readonly CascadeHistoryEntry[] { + return this._history; + } + + /** Waiting-area snapshot for introspection: waiting unit → missing tokens. */ + pendingSnapshot(): ReadonlyMap { + const snapshot = new Map(); + for (const [token, unit] of this._units) { + if (unit.state !== 'Pending') continue; + snapshot.set( + token.toString(), + this._missingDeps(token).map((dep) => dep.toString()), + ); + } + return snapshot; + } + + /** + * Queue a change on the tree. Queued requests merge (deduped by + * scope+token) into the next transaction. The returned promise settles when + * the transaction that applied the change completes; with the sync fast + * path the change is already applied when `submit` returns. + */ + submit(change: CascadeChange): Promise { + if (this._disposed) { + return Promise.resolve(); + } + return new Promise((resolve, reject) => { + this._tree.queue.push({ engine: this, change, resolve, reject }); + this._pump(); + }); + } + + /** Explicit reload of a unit (D5): a replace-self transaction. */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + update(token: ServiceIdentifier, reason?: string): Promise { + return this.submit({ + action: 'update', + token, + reason: reason ?? `update ${String(token)}`, + }); + } + + /** Settles when no transaction is running and the tree queue is empty. */ + whenIdle(): Promise { + if (!this._tree.running && this._tree.queue.length === 0) { + return Promise.resolve(); + } + return new Promise((resolve) => { + this._tree.addSettleWaiter(() => { + void this.whenIdle().then(resolve); + }); + }); + } + + /** + * Async resolution path (§4.3): a token inside the running transaction's + * contagion set suspends until the transaction completes (then resolves); + * anything else resolves immediately. Times out with `CascadeConflictError`. + */ + resolveWhenAvailable( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + token: ServiceIdentifier, + timeoutMs?: number, + ): Promise { + if (!this.isInFlight(token)) { + try { + return Promise.resolve(this._host.materialize(token) as T); + } catch (error) { + return Promise.reject(error); + } + } + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + reject( + new CascadeConflictError( + String(token), + 'timed out waiting for the in-flight cascade to settle', + ), + ); + }, timeoutMs ?? this._options.resolveTimeoutMs ?? DEFAULT_RESOLVE_TIMEOUT_MS); + this._tree.addSettleWaiter(() => { + clearTimeout(timer); + try { + resolve(this._host.materialize(token) as T); + } catch (error) { + reject(error); + } + }); + }); + } + + /** Container-side notification: a unit was materialized outside activation. */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + observedMaterialization(token: ServiceIdentifier): void { + const unit = this._units.get(token); + if (unit !== undefined && unit.state === 'Pending') { + unit.state = 'Active'; + unit.everActive = true; + unit.error = undefined; + } + } + + dispose(): void { + this._disposed = true; + this._tree.engines.delete(this); + this._units.clear(); + this._pendingIndex.clear(); + // Only this engine's queued requests are withdrawn; the tree queue keeps + // serving the other scopes. + const remaining: QueuedRequest[] = []; + for (const request of this._tree.queue) { + if (request.engine === this) { + request.resolve(); + } else { + remaining.push(request); + } + } + this._tree.queue.length = 0; + this._tree.queue.push(...remaining); + } + + // ------------------------------------------------------ orchestrator ops + + /** + * Orchestrator-driven: tear down one of THIS engine's live units + * (Active → Unloading → Pending). Idempotently skipped when this engine's + * scope died mid-transaction. `parkAsPending` is false for the unprovide + * target itself (removal is not a state). + */ + _teardownForCascade( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + token: ServiceIdentifier, + tornDown: string[], + parkAsPending: boolean, + ): void | Promise { + if (this._disposed || !this._host.isMaterialized(token)) { + return undefined; + } + this._unitFor(token).state = 'Unloading'; + const out = this._host.retire(token); + tornDown.push(this._label({ scope: this._scope, token })); + if (parkAsPending) { + // Dependents and replaced units go back to the waiting area with their + // recipe retained; they were live, so they always rebuild. + this._markPending(token, undefined, true); + } + return out; + } + + /** Orchestrator-driven: recheck THIS engine's waiting area (transaction ⑤). */ + _recheckForCascade(rebuilt: string[], failed: string[]): void { + if (this._disposed) { + return; + } + this._recheckPending(rebuilt, failed); + } + + /** Orchestrator-driven: apply THIS engine's own change (transaction ④). */ + _applyChangeForCascade(change: CascadeChange): void { + if (this._disposed) { + return; + } + switch (change.action) { + case 'provide': + if (change.descriptor !== undefined) { + this._host.applyProvide(change.token, change.descriptor, change.pinned); + this._markPending(change.token, change.activation ?? 'eager', false); + } else { + // Pre-materialized instance: a new generation that is already + // live — no activation to schedule, dependents rebuild below. + this._host.applyProvideInstance(change.token, change.instance, change.pinned); + const unit = this._unitFor(change.token); + unit.state = 'Active'; + unit.everActive = true; + unit.error = undefined; + } + break; + case 'unprovide': + this._host.applyUnprovide(change.token); + this._units.delete(change.token); + break; + case 'update': + // Reload of a live-or-failed unit: it rebuilds like a torn one. + this._markPending(change.token, undefined, true); + break; + } + } + + // ------------------------------------------------------------------ queue + + private _pump(): void { + if (this._tree.running) { + return; + } + const batch = this._tree.queue.splice(0); + if (batch.length === 0) { + return; + } + this._tree.running = true; + const finish = (error?: unknown): void => { + this._tree.running = false; + for (const request of batch) { + if (error === undefined) { + request.resolve(); + } else { + request.reject(error); + } + } + this._tree.fireSettleWaiters(); + this._pump(); + }; + // The first request's engine orchestrates the merged transaction. + const orchestrator = batch[0]!.engine; + try { + const out = orchestrator._transact(batch); + if (isPromiseLike(out)) { + Promise.resolve(out).then( + () => { finish(); }, + (error: unknown) => { finish(error); }, + ); + } else { + finish(); + } + } catch (error) { + finish(error); + } + } + + // ------------------------------------------------------------ transaction + + private _transact(batch: QueuedRequest[]): void | Promise { + const changes = mergeBatch(batch); + const started = this._options.now?.() ?? Date.now(); + const reason = changes.map(({ change }) => change.reason).join('; '); + // ① contagion set from the tree-global graph (includes the changed tokens). + const affected = this._tree.graph.affectedSet( + changes.map(({ engine, change }) => ({ scope: engine._scope, token: change.token })), + ); + for (const ref of affected) { + this._tree.inFlightSet(ref, true); + } + this._tree.orchestrator = this._scope; + const complete = (abort: { waited: boolean; timedOut: boolean }): void | Promise => { + const clear = (): void => { + this._tree.inFlightClear(affected); + this._tree.orchestrator = undefined; + }; + let out: void | Promise; + try { + out = this._applyTransaction(changes, affected, reason, started, abort); + } catch (error) { + clear(); + throw error; + } + if (isPromiseLike(out)) { + // The contagion set stays in flight until the transaction settles. + return Promise.resolve(out).then(clear, (error: unknown) => { + clear(); + throw error; + }); + } + clear(); + return undefined; + }; + // ② WillCascade broadcast → abort hook (bounded wait, best-effort). + const wait = this._waitForAbort(affected, reason); + if (isPromiseLike(wait)) { + return Promise.resolve(wait).then(complete); + } + return complete(wait); + } + + private _waitForAbort( + affected: readonly ScopedToken[], + reason: string, + ): { waited: boolean; timedOut: boolean } | Promise<{ waited: boolean; timedOut: boolean }> { + const hook = this._options.onWillCascade; + if (hook === undefined) { + return { waited: false, timedOut: false }; + } + let out: void | Promise; + try { + out = hook(affected, reason); + } catch (error) { + onUnexpectedError(error); + return { waited: false, timedOut: false }; + } + if (!isPromiseLike(out)) { + return { waited: false, timedOut: false }; + } + const waitMs = this._options.abortWaitMs ?? DEFAULT_ABORT_WAIT_MS; + return Promise.race([ + Promise.resolve(out).then( + () => ({ waited: true, timedOut: false }), + (error: unknown) => { + // Best-effort (§4.5): an async abort failure is logged, never a veto. + onUnexpectedError(error); + return { waited: true, timedOut: false }; + }, + ), + new Promise<{ waited: boolean; timedOut: boolean }>((resolve) => { + setTimeout(() => { resolve({ waited: true, timedOut: true }); }, waitMs); + }), + ]); + } + + private _applyTransaction( + changes: QueuedRequest[], + affected: readonly ScopedToken[], + reason: string, + started: number, + abort: { waited: boolean; timedOut: boolean }, + ): void | Promise { + const tornDown: string[] = []; + const rebuilt: string[] = []; + const failed: string[] = []; + + // ③ tear the contagion set down in global reverse topological order, + // serially; each scope's engine executes its own units. + const teardownOrder = this._tree.graph.reverseTopoOrder(affected); + let index = 0; + const step = (): void | Promise => { + while (index < teardownOrder.length) { + const ref = teardownOrder[index]!; + index += 1; + const owner = engineOf(ref); + if (owner === undefined || owner._disposed) { + continue; // descendant scope died mid-transaction: skip idempotently + } + // The unprovide target itself is removed in ④, not parked as Pending. + const removed = changes.some( + ({ engine, change }) => + engine === owner && change.token === ref.token && change.action === 'unprovide', + ); + const out = owner._teardownForCascade(ref.token, tornDown, !removed); + if (isPromiseLike(out)) { + return Promise.resolve(out).then(step); + } + } + return undefined; + }; + + const after = (): void => { + // ④ apply each change in its own scope. + for (const { engine, change } of changes) { + engine._applyChangeForCascade(change); + } + // ⑤ recheck the waiting area across the tree: every live engine, in + // shallow-first order (dependencies only point upward, so depth order + // is a valid global topological order across scopes), iterated to a + // fixpoint — activating one unit may satisfy the next. + const enginesInOrder = [...this._tree.engines] + .filter((engine) => !engine._disposed) + .sort((a, b) => a._scope.cascadeDepth - b._scope.cascadeDepth); + for (;;) { + let progress = false; + for (const engine of enginesInOrder) { + const before = rebuilt.length + failed.length; + engine._recheckForCascade(rebuilt, failed); + if (rebuilt.length + failed.length > before) { + progress = true; + } + } + if (!progress) { + break; + } + } + // ⑥ history ring (orchestrator-local). + this._pushHistory({ + seq: ++this._historySeq, + reason, + changes: changes.map(({ engine, change }) => ({ + token: this._label({ scope: engine._scope, token: change.token }), + action: change.action, + })), + affected: affected.map((ref) => this._label(ref)), + tornDown, + rebuilt, + failed, + abortWaited: abort.waited, + abortTimedOut: abort.timedOut, + durationMs: (this._options.now?.() ?? Date.now()) - started, + }); + }; + + const drained = step(); + if (isPromiseLike(drained)) { + return Promise.resolve(drained).then(after); + } + after(); + return undefined; + } + + // ------------------------------------------------------------------ units + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + private _unitFor(token: ServiceIdentifier): UnitRecord { + let unit = this._units.get(token); + if (unit === undefined) { + unit = { state: 'Pending', activation: 'eager', everActive: false }; + this._units.set(token, unit); + } + return unit; + } + + /** Back to the waiting area; `everActive` marks a cascade-torn unit (always rebuilds). */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + private _markPending( + token: ServiceIdentifier, + activation?: UnitActivation, + everActive?: boolean, + ): void { + const unit = this._unitFor(token); + unit.state = 'Pending'; + unit.error = undefined; + if (activation !== undefined) { + unit.activation = activation; + } + if (everActive !== undefined) { + unit.everActive = everActive; + } + } + + /** + * ⑤ for one engine: rebuild the missing-token index from scratch (cheap: + * one sweep of the waiting area), then activate every satisfied unit in + * topological order, iterating to a fixpoint. Satisfaction consults the + * dependency's OWNING engine across scopes (D9). + */ + private _recheckPending(rebuilt: string[], failed: string[]): void { + for (;;) { + this._pendingIndex.clear(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const satisfied: ServiceIdentifier[] = []; + for (const [token, unit] of this._units) { + if (unit.state !== 'Pending') continue; + const missing = this._missingDeps(token); + if (missing.length === 0) { + // On-demand units that were never live wait for their first + // resolution instead of auto-activating. + if (unit.everActive || unit.activation === 'eager') { + satisfied.push(token); + } + } else { + for (const dep of missing) { + let waiters = this._pendingIndex.get(dep); + if (waiters === undefined) { + waiters = new Set(); + this._pendingIndex.set(dep, waiters); + } + waiters.add(token); + } + } + } + if (satisfied.length === 0) { + return; + } + const ordered = this._tree.graph.topoOrder( + satisfied.map((token) => ({ scope: this._scope, token })), + ); + for (const ref of ordered) { + this._activate(ref.token, rebuilt, failed); + } + // Materialization pulls descriptor dependencies transitively; sweep the + // units that became materialized as a side effect. + for (const [token, unit] of this._units) { + if ( + (unit.state === 'Pending' || unit.state === 'Activating') && + this._host.isMaterialized(token) + ) { + unit.state = 'Active'; + unit.everActive = true; + } + } + } + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + private _activate(token: ServiceIdentifier, rebuilt: string[], failed: string[]): void { + const unit = this._unitFor(token); + unit.state = 'Activating'; + unit.error = undefined; + try { + this._host.materialize(token); + unit.state = 'Active'; + unit.everActive = true; + rebuilt.push(this._label({ scope: this._scope, token })); + } catch (error) { + // D5: Failed is sticky — no automatic retry; explicit update() reloads. + unit.state = 'Failed'; + unit.error = error; + failed.push(this._label({ scope: this._scope, token })); + } + } + + /** Missing dependencies of a Pending unit (empty = ready to activate). */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + private _missingDeps(token: ServiceIdentifier): Array> { + const recipe = this._host.recipeOf(token); + if (recipe === undefined) { + // No recipe to rebuild from (e.g. a foreign-seeded instance that was + // torn down): the unit can never become satisfied on its own. + return [token]; + } + return this._host + .dependenciesOf(recipe) + .filter((dep) => !this._isAvailable(dep)); + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + private _isAvailable(dep: ServiceIdentifier): boolean { + if (!this._host.isRegistered(dep)) { + return false; + } + // The dependency's state is owned by the engine of the scope that + // registered it (an ancestor's engine for a cross-scope injection). + const owner = this._host.ownerScopeOf(dep); + const engine = owner === undefined ? undefined : engineOf({ scope: owner, token: dep }); + const state = engine?.unitState(dep); + return state === undefined || state === 'Active'; + } + + /** Render a scoped token for history: plain for the orchestrator's scope, `#n:token` for others. */ + private _label(ref: ScopedToken): string { + if (ref.scope === this._tree.orchestrator) { + return ref.token.toString(); + } + return `#${this._tree.seqOf(ref.scope)}:${ref.token.toString()}`; + } + + /** Merge-queue dedupe key for this engine's scope. */ + _mergeKey(): string { + return String(this._tree.seqOf(this._scope)); + } + + private _pushHistory(entry: CascadeHistoryEntry): void { + this._history.push(entry); + const capacity = this._options.historyCapacity ?? DEFAULT_HISTORY_CAPACITY; + if (this._history.length > capacity) { + this._history.splice(0, this._history.length - capacity); + } + } +} + +/** Resolve a scoped token to its scope's engine (structural handle). */ +function engineOf(ref: ScopedToken): CascadeEngine | undefined { + return (ref.scope as Partial).cascade; +} + +/** Queued requests merge: one change per scope+token (latest wins), order preserved. */ +function mergeBatch(batch: QueuedRequest[]): QueuedRequest[] { + const byKey = new Map(); + for (const request of batch) { + const key = `${request.engine._mergeKey()}:${request.change.token.toString()}`; + byKey.set(key, request); + } + return [...byKey.values()]; +} diff --git a/packages/agent-core-v2/src/_base/di/dependencyGraph.ts b/packages/agent-core-v2/src/_base/di/dependencyGraph.ts new file mode 100644 index 00000000000..b52c1439138 --- /dev/null +++ b/packages/agent-core-v2/src/_base/di/dependencyGraph.ts @@ -0,0 +1,270 @@ +/** + * `di` domain — persistent dependency graph (L2 substrate), tree-global. + * + * One graph is shared by every container of a scope tree. Edges are recorded + * when a service's constructor dependencies are resolved and removed when the + * consumer is torn down, so the graph always mirrors the live containers. + * Both ends of an edge are scope-tagged: a consumer in a child scope may bind + * a token owned by an ancestor scope (child → parent only — a parent can never + * resolve a child's token, so cross-tree cycles are impossible by + * construction). Instance edges bind a consumer to its dependency's + * generation (the dependency changes → the consumer is torn down and rebuilt, + * across scopes); collection edges (Phase 3) are recorded for introspection + * but never join a cascade contagion set. + */ + +import type { ServiceIdentifier } from './instantiation'; + +/** A token as seen from the tree: the owning container plus the identifier. */ +export interface ScopedToken { + /** The container whose collection owns the registration. */ + readonly scope: object; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + readonly token: ServiceIdentifier; +} + +export type DependencyEdgeKind = 'instance' | 'collection'; + +export interface DependencyEdge { + readonly consumer: ScopedToken; + readonly dependency: ScopedToken; + readonly kind: DependencyEdgeKind; +} + +/** Nested scope → token maps, so scoped tokens stay structural (no interning). */ +export class PairIndex { + private readonly _map = new Map, + V + >>(); + + get(scope: object, token: ServiceIdentifier): V | undefined { + return this._map.get(scope)?.get(token); + } + + set(scope: object, token: ServiceIdentifier, value: V): void { + let inner = this._map.get(scope); + if (inner === undefined) { + inner = new Map(); + this._map.set(scope, inner); + } + inner.set(token, value); + } + + delete(scope: object, token: ServiceIdentifier): void { + const inner = this._map.get(scope); + if (inner === undefined) return; + inner.delete(token); + if (inner.size === 0) { + this._map.delete(scope); + } + } + + entries(): Array<[ScopedToken, V]> { + const out: Array<[ScopedToken, V]> = []; + for (const [scope, inner] of this._map) { + for (const [token, value] of inner) { + out.push([{ scope, token }, value]); + } + } + return out; + } + + clear(): void { + this._map.clear(); + } +} + +export class DependencyGraph { + /** live instance → its scoped token */ + private readonly _refByInstance = new Map(); + /** scoped token → live instance */ + private readonly _instanceByRef = new PairIndex(); + /** consumer instance → (dependency scoped token → edge kind) */ + private readonly _out = new Map>(); + /** dependency scoped token → (consumer instance → edge kind) */ + private readonly _in = new PairIndex>(); + + /** Register a materialized service instance so its edges can be tracked. */ + addInstance( + instance: object, + scope: object, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + token: ServiceIdentifier, + ): void { + const ref: ScopedToken = { scope, token }; + this._refByInstance.set(instance, ref); + this._instanceByRef.set(scope, token, instance); + } + + /** Drop a consumer: its outbound edges and token mapping (inbound edges stay). */ + removeInstance(instance: object): void { + const ref = this._refByInstance.get(instance); + if (ref !== undefined) { + this._refByInstance.delete(instance); + if (this._instanceByRef.get(ref.scope, ref.token) === instance) { + this._instanceByRef.delete(ref.scope, ref.token); + } + } + const out = this._out.get(instance); + if (out !== undefined) { + for (const [dependency] of out.entries()) { + this._in.get(dependency.scope, dependency.token)?.delete(instance); + } + this._out.delete(instance); + } + } + + addEdge( + consumerInstance: object, + dependency: ScopedToken, + kind: DependencyEdgeKind = 'instance', + ): void { + let out = this._out.get(consumerInstance); + if (out === undefined) { + out = new PairIndex(); + this._out.set(consumerInstance, out); + } + out.set(dependency.scope, dependency.token, kind); + + let inbound = this._in.get(dependency.scope, dependency.token); + if (inbound === undefined) { + inbound = new Map(); + this._in.set(dependency.scope, dependency.token, inbound); + } + inbound.set(consumerInstance, kind); + } + + /** + * The contagion set: the changed scoped tokens plus every scoped token whose + * live instance transitively depends on them through instance edges — + * computed across the whole tree (dependents always live in the changed + * scope's subtree, since edges point child → parent). + */ + affectedSet(changed: Iterable): ScopedToken[] { + const seen = new PairIndex(); + const queue: ScopedToken[] = []; + const push = (ref: ScopedToken): void => { + if (seen.get(ref.scope, ref.token) !== undefined) return; + seen.set(ref.scope, ref.token, true); + queue.push(ref); + }; + for (const ref of changed) { + push(ref); + } + const affected: ScopedToken[] = []; + while (queue.length > 0) { + const ref = queue.pop()!; + affected.push(ref); + const inbound = this._in.get(ref.scope, ref.token); + if (inbound === undefined) continue; + for (const [consumerInstance, kind] of inbound) { + if (kind !== 'instance') continue; + const consumer = this._refByInstance.get(consumerInstance); + if (consumer === undefined) continue; + push(consumer); + } + } + return affected; + } + + /** Dependencies-first order over the given scoped-token subset (instance edges). */ + topoOrder(tokens: Iterable): ScopedToken[] { + const subset = new PairIndex(); + for (const ref of tokens) { + subset.set(ref.scope, ref.token, ref); + } + const ordered: ScopedToken[] = []; + const done = new PairIndex(); + const visit = (ref: ScopedToken): void => { + if (done.get(ref.scope, ref.token) !== undefined) return; + done.set(ref.scope, ref.token, true); + for (const dependency of this._dependenciesOf(ref, subset)) { + visit(dependency); + } + ordered.push(ref); + }; + for (const [, ref] of subset.entries()) { + visit(ref); + } + return ordered; + } + + /** Dependents-first order (teardown order) over the given scoped-token subset. */ + reverseTopoOrder(tokens: Iterable): ScopedToken[] { + return this.topoOrder(tokens).toReversed(); + } + + /** Instance-edge cycle check over the live graph; returns the cycle path or null. */ + findCycle(label: (ref: ScopedToken) => string): string[] | null { + const state = new PairIndex<'visiting' | 'done'>(); + const path: ScopedToken[] = []; + const visit = (ref: ScopedToken): string[] | null => { + state.set(ref.scope, ref.token, 'visiting'); + path.push(ref); + for (const dependency of this._dependenciesOf(ref, undefined)) { + const mark = state.get(dependency.scope, dependency.token); + if (mark === 'done') continue; + if (mark === 'visiting') { + const start = path.findIndex( + (entry) => + entry.scope === dependency.scope && entry.token === dependency.token, + ); + return [...path.slice(start), dependency].map(label); + } + const cycle = visit(dependency); + if (cycle !== null) return cycle; + } + path.pop(); + state.set(ref.scope, ref.token, 'done'); + return null; + }; + for (const [ref] of this._instanceByRef.entries()) { + if (state.get(ref.scope, ref.token) !== undefined) continue; + const cycle = visit(ref); + if (cycle !== null) return cycle; + } + return null; + } + + /** Introspection: every live edge (both kinds), scoped on both ends. */ + edges(): DependencyEdge[] { + const edges: DependencyEdge[] = []; + for (const [consumerInstance, out] of this._out) { + const consumer = this._refByInstance.get(consumerInstance); + if (consumer === undefined) continue; + for (const [dependency, kind] of out.entries()) { + edges.push({ consumer, dependency, kind }); + } + } + return edges; + } + + clear(): void { + this._refByInstance.clear(); + this._instanceByRef.clear(); + this._out.clear(); + this._in.clear(); + } + + /** In-subset instance-edge dependencies of a scoped token's live instance. */ + private _dependenciesOf( + ref: ScopedToken, + subset: PairIndex | undefined, + ): ScopedToken[] { + const instance = this._instanceByRef.get(ref.scope, ref.token); + if (instance === undefined) return []; + const out = this._out.get(instance); + if (out === undefined) return []; + const result: ScopedToken[] = []; + for (const [dependency, kind] of out.entries()) { + if (kind !== 'instance') continue; + if (subset !== undefined && subset.get(dependency.scope, dependency.token) === undefined) { + continue; + } + result.push(dependency); + } + return result; + } +} diff --git a/packages/agent-core-v2/src/_base/di/errors.ts b/packages/agent-core-v2/src/_base/di/errors.ts index 5d49dafc055..d1de9d4153d 100644 --- a/packages/agent-core-v2/src/_base/di/errors.ts +++ b/packages/agent-core-v2/src/_base/di/errors.ts @@ -24,3 +24,19 @@ export class CyclicDependencyError extends Error { this.name = 'CyclicDependencyError'; } } + +/** + * Raised when a resolution hits a token inside an in-flight cascade + * transaction's contagion set. The async resolution path suspends instead + * (see `CascadeEngine.resolveWhenAvailable`); the sync path cannot suspend, + * so it fails fast with this error. + */ +export class CascadeConflictError extends Error { + constructor( + readonly token: string, + readonly detail: string, + ) { + super(`Cascade conflict resolving '${token}': ${detail}`); + this.name = 'CascadeConflictError'; + } +} diff --git a/packages/agent-core-v2/src/_base/di/instantiation.ts b/packages/agent-core-v2/src/_base/di/instantiation.ts index 673c06e4239..5e9e41cc632 100644 --- a/packages/agent-core-v2/src/_base/di/instantiation.ts +++ b/packages/agent-core-v2/src/_base/di/instantiation.ts @@ -2,8 +2,9 @@ * `di` domain — service identifiers, `createDecorator`, and the `IInstantiationService` contract. */ -import type { SyncDescriptor0 } from './descriptors'; -import type { DisposableStore } from './lifecycle'; +import type { SyncDescriptor, SyncDescriptor0 } from './descriptors'; +import type { CascadeEngine } from './cascadeEngine'; +import type { DisposableStore, IDisposable } from './lifecycle'; import type { ServiceCollection } from './serviceCollection'; // eslint-disable-next-line @typescript-eslint/no-namespace @@ -109,9 +110,32 @@ export interface ServicesAccessor { get(id: ServiceIdentifier): T; } +export interface ProvideOptions { + /** Cascade-line metadata (L4): a pinned unit never joins a cascade. */ + readonly pinned?: boolean; + /** + * `eager` (default): the unit activates as soon as its dependencies are + * satisfied. `ondemand`: it materializes at first resolution (a cascade-torn + * unit always rebuilds regardless). + */ + readonly activation?: 'eager' | 'ondemand'; +} + +/** + * Handle to one `provide` registration: it is an entry in the provider's + * ledger, so disposing the handle unprovides the token. (Grows into the full + * FiberHandle — thenable / state / update — in Phase 3.) + */ +export interface ProvideHandle extends IDisposable { + readonly uid: number; +} + export interface IInstantiationService { readonly _serviceBrand: undefined; + /** Cascade engine (L2): per-container facade over the tree-wide orchestrated transactions. */ + readonly cascade: CascadeEngine; + invokeFunction( fn: (accessor: ServicesAccessor, ...args: TS) => R, ...args: TS @@ -129,6 +153,17 @@ export interface IInstantiationService { ...args: GetLeadingNonServiceArgs> ): R; createChild(services: ServiceCollection, store?: DisposableStore): IInstantiationService; + /** + * Register (or replace) a token at runtime. Replacing retires the previous + * materialized instance before the new generation becomes visible. + */ + provide( + id: ServiceIdentifier, + instanceOrDescriptor: T | SyncDescriptor, + options?: ProvideOptions, + ): ProvideHandle; + /** Remove a token, retiring its materialized instance. No-op when absent. */ + unprovide(id: ServiceIdentifier): void; dispose(): void; } diff --git a/packages/agent-core-v2/src/_base/di/instantiationService.ts b/packages/agent-core-v2/src/_base/di/instantiationService.ts index dd4a6a30ff2..23503170673 100644 --- a/packages/agent-core-v2/src/_base/di/instantiationService.ts +++ b/packages/agent-core-v2/src/_base/di/instantiationService.ts @@ -3,21 +3,22 @@ */ import { SyncDescriptor } from './descriptors'; -import { CyclicDependencyError } from './errors'; +import { CascadeEngine, CascadeTree, type CascadeChange, type CascadeHost } from './cascadeEngine'; +import { DependencyGraph } from './dependencyGraph'; +import { CascadeConflictError, CyclicDependencyError } from './errors'; import { Graph } from './graph'; import { IInstantiationService as IInstantiationServiceDecorator, _util, type IInstantiationService, + type ProvideHandle, + type ProvideOptions, type ServiceIdentifier, type ServicesAccessor, } from './instantiation'; -import { - dispose, - isDisposable, - type DisposableStore, - type IDisposable, -} from './lifecycle'; +import { isDisposable, type DisposableStore } from './lifecycle'; +import { onUnexpectedError } from '../errors/unexpectedError'; +import { Ledger, type LedgerEntry } from '../lifecycle/ledger'; import { ServiceCollection } from './serviceCollection'; // eslint-disable-next-line @typescript-eslint/no-unused-vars @@ -113,8 +114,29 @@ export class InstantiationService implements IInstantiationService { protected readonly _parent?: InstantiationService; + protected readonly _ledger = new Ledger('InstantiationService'); + + /** Tree-global persistent dependency graph (shared by the whole scope tree). */ + get dependencyGraph(): DependencyGraph { + return this._tree.graph; + } + + private readonly _tree: CascadeTree; + + /** Cascade engine (L2): one per container; tree-wide orchestrated transactions. */ + readonly cascade: CascadeEngine; + + private _parentLedgerEntry: LedgerEntry | undefined; + + /** Materialized instance → its ledger entry (for individual retirement). */ + private readonly _instanceEntries = new Map(); + + /** Token → the ledger entry of its latest provide (generation-guarded). */ // eslint-disable-next-line @typescript-eslint/no-explicit-any - protected readonly _constructionOrder: any[] = []; + private readonly _provideEntries = new Map, LedgerEntry>(); + + /** Set while the cascade engine itself resolves — bypasses the in-flight guard. */ + private _cascadeResolving = false; protected readonly _children = new Set(); @@ -124,9 +146,6 @@ export class InstantiationService implements IInstantiationService { // eslint-disable-next-line @typescript-eslint/no-explicit-any private readonly _activeInstantiations = new Set>(); - // eslint-disable-next-line @typescript-eslint/no-explicit-any - private readonly _servicesToMaybeDispose = new Set(); - private _disposed = false; constructor( @@ -138,6 +157,65 @@ export class InstantiationService implements IInstantiationService { this._parent = parent; this._globalGraph = _enableTracing ? parent?._globalGraph ?? new Graph(e => e) : undefined; this._services.set(IInstantiationServiceDecorator, this); + this._tree = parent?._tree ?? new CascadeTree(new DependencyGraph()); + const host: CascadeHost = { + isRegistered: (token) => this._getServiceInstanceOrDescriptor(token) !== undefined, + ownerScopeOf: (token) => this._ownerOf(token), + isMaterialized: (token) => { + const value = this._services.get(token); + return value !== undefined && !(value instanceof SyncDescriptor); + }, + materialize: (token) => { + this._cascadeResolving = true; + try { + return this._getOrCreateServiceInstance( + token, + Trace.traceCreation(false, CascadeEngine), + ); + } finally { + this._cascadeResolving = false; + } + }, + retire: (token) => this._retireUnit(token), + applyProvide: (token, descriptor, pinned) => { + this._services.set(token, descriptor, { pinned }); + return this._services.uidOf(token)!; + }, + applyProvideInstance: (token, instance, pinned) => { + this._services.set(token, instance, { pinned }); + return this._services.uidOf(token)!; + }, + applyUnprovide: (token) => { + this._services.delete(token); + }, + recipeOf: (token) => { + const entry = this._services.entry(token); + if (entry === undefined) return undefined; + return entry.value instanceof SyncDescriptor ? entry.value : entry.recipe; + }, + dependenciesOf: (recipe) => + _util.getServiceDependencies(recipe.ctor).map((dependency) => dependency.id), + }; + this.cascade = new CascadeEngine(host, this, this._tree); + } + + /** Structural handle for the cascade engine's scoped tokens. */ + get cascadeDisposed(): boolean { + return this._disposed; + } + + /** Distance from the tree root (root = 0). */ + get cascadeDepth(): number { + return (this._parent?.cascadeDepth ?? -1) + 1; + } + + /** The container owning a token in this container's resolution chain. */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + private _ownerOf(id: ServiceIdentifier): InstantiationService | undefined { + if (this._services.has(id)) { + return this; + } + return this._parent?._ownerOf(id); } invokeFunction( @@ -169,6 +247,125 @@ export class InstantiationService implements IInstantiationService { } } + provide( + id: ServiceIdentifier, + instanceOrDescriptor: T | SyncDescriptor, + options?: ProvideOptions, + ): ProvideHandle { + this._assertNotDisposed(); + this._releaseProvideEntry(id); + + if ( + !(instanceOrDescriptor instanceof SyncDescriptor) && + this._services.get(id) === instanceOrDescriptor + ) { + // Re-affirming the very instance already materialized under this token: + // refresh the registration, no retirement, no cascade. + this._services.set(id, instanceOrDescriptor, { pinned: options?.pinned }); + const uid = this._services.uidOf(id)!; + const entry = this._ledger.register(() => { + if (this._services.uidOf(id) === uid) { + this.unprovide(id); + } + }, `provide:${String(id)}`); + this._provideEntries.set(id, entry); + return { + uid, + dispose: () => { + void entry.dispose(); + }, + }; + } + + // Everything else — a recipe or a replacing instance — is one cascade + // transaction, so live dependents are torn down and rebuilt (D1/D4). + const beforeUid = this._services.uidOf(id); + let appliedUid: number | undefined; + const noteApplied = (): void => { + const uid = this._services.uidOf(id); + if (uid !== undefined && uid !== beforeUid) { + appliedUid = uid; + } + }; + const change: CascadeChange = + instanceOrDescriptor instanceof SyncDescriptor + ? { + action: 'provide', + token: id, + descriptor: instanceOrDescriptor, + pinned: options?.pinned, + activation: options?.activation, + reason: `provide ${String(id)}`, + } + : { + action: 'provide', + token: id, + instance: instanceOrDescriptor, + pinned: options?.pinned, + reason: `provide ${String(id)}`, + }; + noteApplied(); + this.cascade.submit(change).then(noteApplied, onUnexpectedError); + noteApplied(); // the sync fast path has already applied the change + const entry = this._ledger.register(() => { + // Generation guard: only unprovide the generation this entry provided. + if (appliedUid !== undefined && this._services.uidOf(id) === appliedUid) { + this.unprovide(id); + } + }, `provide:${String(id)}`); + this._provideEntries.set(id, entry); + return { + get uid(): number { + if (appliedUid === undefined) { + throw new Error( + `provide of '${String(id)}' has not been applied yet (cascade in flight)`, + ); + } + return appliedUid; + }, + dispose: () => { + void entry.dispose(); + }, + }; + } + + unprovide(id: ServiceIdentifier): void { + if (this._disposed) { + return; + } + this._releaseProvideEntry(id); + if (this._services.get(id) === undefined) { + return; + } + this.cascade + .submit({ action: 'unprovide', token: id, reason: `unprovide ${String(id)}` }) + .catch(onUnexpectedError); + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + private _releaseProvideEntry(id: ServiceIdentifier): void { + const entry = this._provideEntries.get(id); + if (entry !== undefined) { + this._provideEntries.delete(id); + entry.release(); + } + } + + /** Retire the live instance of a token and reset its entry to the recipe. */ + private _retireUnit(id: ServiceIdentifier): void | Promise { + const instance = this._services.get(id); + if (instance === undefined || instance instanceof SyncDescriptor) { + return undefined; + } + this._services.unmaterialize(id); + const entry = this._instanceEntries.get(instance); + if (entry === undefined) { + return undefined; + } + this._instanceEntries.delete(instance); + return entry.dispose(); + } + // eslint-disable-next-line @typescript-eslint/no-explicit-any createInstance(descriptor: SyncDescriptor, ...rest: any[]): T; // eslint-disable-next-line @typescript-eslint/no-explicit-any @@ -204,42 +401,40 @@ export class InstantiationService implements IInstantiationService { 'createChild requires a ServiceCollection instance (got something else)', ); } - const child = new InstantiationService(services, this._strict, this, this._enableTracing); + const child = this._createChildService(services); this._children.add(child); + child._parentLedgerEntry = this._ledger.register(() => { + child.dispose(); + }, 'child-instantiation'); store?.add(child); return child; } + protected _createChildService(services: ServiceCollection): InstantiationService { + return new InstantiationService(services, this._strict, this, this._enableTracing); + } + dispose(): void { if (this._disposed) { return; } this._disposed = true; - const childSnapshot = Array.from(this._children); - this._children.clear(); - - const ownInstances: IDisposable[] = []; - for (let i = this._constructionOrder.length - 1; i >= 0; i--) { - const instance = this._constructionOrder[i]!; - if (isDisposable(instance)) { - ownInstances.push(instance); - this._servicesToMaybeDispose.delete(instance); - } - } - - const remainingInstances: IDisposable[] = []; - for (const candidate of this._servicesToMaybeDispose) { - if (isDisposable(candidate)) { - remainingInstances.push(candidate); - } - } - try { - dispose([...childSnapshot, ...ownInstances, ...remainingInstances]); + // Children first (forward creation order): their services may depend on + // this container's instances, so they must die before them. Each child + // releases its ledger entry, so the ledger teardown below skips them. + for (const child of Array.from(this._children)) { + child.dispose(); + } + this._children.clear(); + void this._ledger.teardown('scope-close'); + this._services.dispose(); + this.cascade.dispose(); } finally { - this._constructionOrder.length = 0; - this._servicesToMaybeDispose.clear(); + this._children.clear(); + this._parentLedgerEntry?.release(); + this._parentLedgerEntry = undefined; if (this._parent) { this._parent._children.delete(this); } @@ -281,6 +476,21 @@ export class InstantiationService implements IInstantiationService { } protected _getOrCreateServiceInstance(id: ServiceIdentifier, _trace: Trace): T { + if (!this._cascadeResolving) { + if (this.cascade.isInFlight(id)) { + // The sync resolution path cannot suspend; the async path + // (cascade.resolveWhenAvailable) waits for the transaction instead. + throw new CascadeConflictError( + String(id), + 'token is inside an in-flight cascade transaction', + ); + } + const failure = this.cascade.unitFailure(id); + if (failure !== undefined) { + // D5: Failed is sticky — resolving a failed unit rethrows its error. + throw failure as Error; + } + } const entry = this._getServiceInstanceOrDescriptor(id); if (entry instanceof SyncDescriptor) { @@ -397,13 +607,7 @@ export class InstantiationService implements IInstantiationService { _trace: Trace, ): T { if (this._services.get(id) instanceof SyncDescriptor) { - return this._createServiceInstance( - id, - ctor, - args, - _trace, - this._servicesToMaybeDispose, - ); + return this._createServiceInstance(id, ctor, args, _trace); } if (this._parent) { return this._parent._createServiceInstanceWithOwner( @@ -422,15 +626,39 @@ export class InstantiationService implements IInstantiationService { ctor: any, args: ReadonlyArray = [], _trace: Trace, - // eslint-disable-next-line @typescript-eslint/no-explicit-any - disposeBucket: Set, ): T { const root = this._root(); root._inProgress.push(id); try { const result = this._createInstance(ctor, args.slice(), _trace); - disposeBucket.add(result); - this._constructionOrder.push(result); + // Persistent tree-global graph: record the instance and its + // constructor-injection (instance) edges, both ends scope-tagged; the + // ledger entry removes them again at teardown. Edges point child → + // parent (a dependency's owner is always this container or an ancestor). + this.dependencyGraph.addInstance(result as object, this, id); + for (const dependency of _util.getServiceDependencies(ctor)) { + const owner = this._ownerOf(dependency.id); + if (owner !== undefined) { + this.dependencyGraph.addEdge( + result as object, + { scope: owner, token: dependency.id }, + 'instance', + ); + } + } + const entry = this._ledger.register(() => { + this._instanceEntries.delete(result); + this.dependencyGraph.removeInstance(result as object); + if (isDisposable(result)) { + // Propagate a (runtime) async disposer so cascade teardown can + // await it serially; statically `dispose()` is typed void. + const out = result.dispose() as unknown as void | Promise; + return out; + } + return undefined; + }, `service:${String(id)}`); + this._instanceEntries.set(result, entry); + this.cascade.observedMaterialization(id); return result; } finally { const popIdx = root._inProgress.lastIndexOf(id); @@ -442,7 +670,9 @@ export class InstantiationService implements IInstantiationService { private _setCreatedServiceInstance(id: ServiceIdentifier, instance: T): void { if (this._services.get(id) instanceof SyncDescriptor) { - this._services.set(id, instance); + // Keeps the recipe on the entry so a cascade teardown can unmaterialize + // back to it (and rebuild later). + this._services.materialize(id, instance); } else if (this._parent) { this._parent._setCreatedServiceInstance(id, instance); } else { diff --git a/packages/agent-core-v2/src/_base/di/lifecycle.ts b/packages/agent-core-v2/src/_base/di/lifecycle.ts index ed294b4d660..0593c3c8290 100644 --- a/packages/agent-core-v2/src/_base/di/lifecycle.ts +++ b/packages/agent-core-v2/src/_base/di/lifecycle.ts @@ -3,6 +3,11 @@ */ import { onUnexpectedError } from '../errors/unexpectedError'; +import { Ledger, type LedgerEntry } from '../lifecycle/ledger'; + +function disposableLabel(d: IDisposable): string { + return `disposable:${d.constructor?.name ?? 'anonymous'}`; +} export interface IDisposableTracker { trackDisposable(disposable: IDisposable): void; @@ -220,7 +225,8 @@ export function combinedDisposable(...disposables: IDisposable[]): IDisposable { } export class DisposableStore implements IDisposable { - private readonly _toDispose = new Set(); + private readonly _ledger = new Ledger('DisposableStore'); + private readonly _entries = new Map(); private _isDisposed = false; constructor() { @@ -236,7 +242,14 @@ export class DisposableStore implements IDisposable { d.dispose(); return d; } - this._toDispose.add(d); + if (!this._entries.has(d)) { + this._entries.set( + d, + this._ledger.register(() => { + d.dispose(); + }, disposableLabel(d)), + ); + } return d; } @@ -245,23 +258,30 @@ export class DisposableStore implements IDisposable { if ((d as unknown as DisposableStore) === this) { throw new Error('Cannot dispose a disposable on itself!'); } - this._toDispose.delete(d); + const entry = this._entries.get(d); + if (entry) { + this._entries.delete(d); + entry.release(); + } d.dispose(); } deleteAndLeak(d: T): void { if (this._isDisposed) return; - if (this._toDispose.delete(d)) { + const entry = this._entries.get(d); + if (entry) { + this._entries.delete(d); + entry.release(); setParentOfDisposable(d, null); } } clear(): void { - if (this._toDispose.size === 0) return; + if (this._entries.size === 0) return; try { - dispose(this._toDispose); + void this._ledger.clear('scope-close'); } finally { - this._toDispose.clear(); + this._entries.clear(); } } @@ -269,7 +289,11 @@ export class DisposableStore implements IDisposable { if (this._isDisposed) return; this._isDisposed = true; markAsDisposed(this); - this.clear(); + try { + void this._ledger.teardown('scope-close'); + } finally { + this._entries.clear(); + } } get isDisposed(): boolean { @@ -312,6 +336,8 @@ export namespace Disposable { } export class MutableDisposable implements IDisposable { + private readonly _ledger = new Ledger('MutableDisposable'); + private _entry: LedgerEntry | undefined; private _value: T | undefined; private _isDisposed = false; @@ -331,20 +357,31 @@ export class MutableDisposable implements IDisposable { return; } if (this._value === value) return; + this._entry?.release(); + this._entry = undefined; this._value?.dispose(); if (value) setParentOfDisposable(value, this); this._value = value; + if (value) { + this._entry = this._ledger.register(() => { + value.dispose(); + }, disposableLabel(value)); + } } dispose(): void { if (this._isDisposed) return; this._isDisposed = true; markAsDisposed(this); + const entry = this._entry; const prev = this._value; + this._entry = undefined; + this._value = undefined; + entry?.release(); + void this._ledger.teardown('scope-close'); if (prev !== undefined) { prev.dispose(); } - this._value = undefined; } clear(): void { @@ -356,6 +393,8 @@ export class MutableDisposable implements IDisposable { if (this._isDisposed) return undefined; const prev = this._value; this._value = undefined; + this._entry?.release(); + this._entry = undefined; if (prev !== undefined) setParentOfDisposable(prev, null); return prev; } diff --git a/packages/agent-core-v2/src/_base/di/scope.ts b/packages/agent-core-v2/src/_base/di/scope.ts index 8d2db5bb138..347293e7e0d 100644 --- a/packages/agent-core-v2/src/_base/di/scope.ts +++ b/packages/agent-core-v2/src/_base/di/scope.ts @@ -9,6 +9,7 @@ import { SyncDescriptor } from './descriptors'; import type { ServiceIdentifier, ServicesAccessor, IInstantiationService } from './instantiation'; import { InstantiationService } from './instantiationService'; import { DisposableStore, type IDisposable } from './lifecycle'; +import { Ledger, type LedgerEntry } from '../lifecycle/ledger'; import { ServiceCollection } from './serviceCollection'; export enum LifecycleScope { @@ -139,6 +140,8 @@ export class Scope implements IDisposable { readonly accessor: ServicesAccessor; private readonly _store = new DisposableStore(); + private readonly _ledger: Ledger; + private _ledgerEntry: LedgerEntry | undefined; private _disposed = false; private constructor( @@ -147,6 +150,15 @@ export class Scope implements IDisposable { readonly instantiation: IInstantiationService, private readonly _parent?: Scope, ) { + // Registration order is reversed at teardown: children (registered later) + // go first, then the store, then the instantiation container. + this._ledger = new Ledger(`scope:${id}`); + this._ledger.register(() => { + this.instantiation.dispose(); + }, 'instantiation'); + this._ledger.register(() => { + this._store.dispose(); + }, 'store'); this.accessor = { get: (serviceId: ServiceIdentifier): T => instantiation.invokeFunction((a) => a.get(serviceId)), @@ -192,6 +204,9 @@ export class Scope implements IDisposable { } const child = new Scope(id, kind, childInstantiation, this); this.children.set(id, child); + child._ledgerEntry = this._ledger.register(() => { + child.dispose(); + }, `scope:${id}`); return child; } @@ -205,17 +220,15 @@ export class Scope implements IDisposable { } this._disposed = true; - const kids = Array.from(this.children.values()); - this.children.clear(); - for (const child of kids) { - child.dispose(); - } - - this._store.dispose(); - this.instantiation.dispose(); - - if (this._parent) { - this._parent.children.delete(this.id); + this._ledgerEntry?.release(); + this._ledgerEntry = undefined; + try { + void this._ledger.teardown('scope-close'); + } finally { + this.children.clear(); + if (this._parent) { + this._parent.children.delete(this.id); + } } } } diff --git a/packages/agent-core-v2/src/_base/di/serviceCollection.ts b/packages/agent-core-v2/src/_base/di/serviceCollection.ts index 13fb2803a0c..2f81b203bd6 100644 --- a/packages/agent-core-v2/src/_base/di/serviceCollection.ts +++ b/packages/agent-core-v2/src/_base/di/serviceCollection.ts @@ -1,30 +1,127 @@ /** - * `di` domain — `ServiceCollection` map of service id → descriptor or instance. + * `di` domain — `ServiceCollection`: the dynamic registry (L1). + * + * Maps a service id to its recipe (`SyncDescriptor`) or materialized instance. + * Every write stamps the entry with a container-monotonic `uid` (a generation + * marker used for introspection and history — it plays no role in change + * detection) and fires the token's availability event with `{ oldUid, newUid }`. */ -import type { SyncDescriptor } from './descriptors'; +import { Emitter } from '../event'; +import { SyncDescriptor } from './descriptors'; import type { ServiceIdentifier } from './instantiation'; +import type { IDisposable } from './lifecycle'; + +export interface ServiceCollectionEntry { + readonly value: T | SyncDescriptor; + readonly uid: number; + readonly pinned: boolean; + /** The recipe a materialized instance was created from (kept for rebuilds). */ + readonly recipe?: SyncDescriptor; +} + +export interface AvailabilityChange { + readonly oldUid: number | undefined; + readonly newUid: number | undefined; +} export class ServiceCollection { // eslint-disable-next-line @typescript-eslint/no-explicit-any - private readonly _entries = new Map, unknown>(); + private readonly _entries = new Map, ServiceCollectionEntry>(); + private readonly _emitters = new Map< + // eslint-disable-next-line @typescript-eslint/no-explicit-any + ServiceIdentifier, + Emitter + >(); + private _nextUid = 0; constructor( // eslint-disable-next-line @typescript-eslint/no-explicit-any ...entries: ReadonlyArray, unknown]> ) { for (const [id, value] of entries) { - this._entries.set(id, value); + this.set(id, value); } } set( id: ServiceIdentifier, instanceOrDescriptor: T | SyncDescriptor, + options?: { readonly pinned?: boolean }, ): T | SyncDescriptor | undefined { const prev = this._entries.get(id); - this._entries.set(id, instanceOrDescriptor); - return prev as T | SyncDescriptor | undefined; + const uid = ++this._nextUid; + this._entries.set(id, { + value: instanceOrDescriptor, + uid, + pinned: options?.pinned ?? prev?.pinned ?? false, + }); + this._emitterFor(id).fire({ oldUid: prev?.uid, newUid: uid }); + return prev?.value as T | SyncDescriptor | undefined; + } + + /** + * Swap a descriptor entry for its materialized instance, keeping the uid, + * pinned flag, and the recipe (so the entry can be unmaterialized back to + * the recipe when the instance is torn down). Not a new generation. + */ + materialize(id: ServiceIdentifier, instance: T): void { + const prev = this._entries.get(id); + if (prev === undefined || !(prev.value instanceof SyncDescriptor)) { + return; + } + this._entries.set(id, { + value: instance, + uid: prev.uid, + pinned: prev.pinned, + recipe: prev.value, + }); + } + + /** Swap a materialized entry back to its recipe (instance torn down). */ + unmaterialize(id: ServiceIdentifier): void { + const prev = this._entries.get(id); + if (prev === undefined || prev.recipe === undefined) { + return; + } + this._entries.set(id, { + value: prev.recipe, + uid: prev.uid, + pinned: prev.pinned, + }); + } + + delete(id: ServiceIdentifier): T | SyncDescriptor | undefined { + const prev = this._entries.get(id); + if (prev === undefined) { + return undefined; + } + this._entries.delete(id); + this._emitterFor(id).fire({ oldUid: prev.uid, newUid: undefined }); + return prev.value as T | SyncDescriptor | undefined; + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + entry(id: ServiceIdentifier): ServiceCollectionEntry | undefined { + return this._entries.get(id); + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + uidOf(id: ServiceIdentifier): number | undefined { + return this._entries.get(id)?.uid; + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + isPinned(id: ServiceIdentifier): boolean { + return this._entries.get(id)?.pinned ?? false; + } + + /** Fired when the token's availability changes (set → new uid, delete → undefined). */ + onDidChange( + id: ServiceIdentifier, + listener: (change: AvailabilityChange) => void, + ): IDisposable { + return this._emitterFor(id).event(listener); } // eslint-disable-next-line @typescript-eslint/no-explicit-any @@ -33,7 +130,7 @@ export class ServiceCollection { } get(id: ServiceIdentifier): T | SyncDescriptor | undefined { - return this._entries.get(id) as T | SyncDescriptor | undefined; + return this._entries.get(id)?.value as T | SyncDescriptor | undefined; } forEach( @@ -43,6 +140,22 @@ export class ServiceCollection { value: unknown, ) => void, ): void { - this._entries.forEach((value, id) => callback(id, value)); + this._entries.forEach((entry, id) => { callback(id, entry.value); }); + } + + dispose(): void { + for (const emitter of this._emitters.values()) { + emitter.dispose(); + } + this._emitters.clear(); + } + + private _emitterFor(id: ServiceIdentifier): Emitter { + let emitter = this._emitters.get(id); + if (emitter === undefined) { + emitter = new Emitter(); + this._emitters.set(id, emitter); + } + return emitter; } } diff --git a/packages/agent-core-v2/src/_base/di/testInstantiationService.ts b/packages/agent-core-v2/src/_base/di/testInstantiationService.ts index f0b51a971b8..258db38e024 100644 --- a/packages/agent-core-v2/src/_base/di/testInstantiationService.ts +++ b/packages/agent-core-v2/src/_base/di/testInstantiationService.ts @@ -53,7 +53,12 @@ export class TestInstantiationService extends InstantiationService implements ID id: ServiceIdentifier, instanceOrDescriptor: T | SyncDescriptor, ): T | SyncDescriptor | undefined { - return this._serviceCollection.set(id, instanceOrDescriptor); + // Routed through provide so test overrides get production semantics: + // a replaced materialized instance is retired, a new generation starts. + // Descriptors stay lazy (constructed at first resolution), as before. + const prev = this._serviceCollection.get(id); + this.provide(id, instanceOrDescriptor, { activation: 'ondemand' }); + return prev; } public mock(id: ServiceIdentifier): T | sinon.SinonMock { @@ -269,14 +274,11 @@ export class TestInstantiationService extends InstantiationService implements ID } public override createChild(services: ServiceCollection): TestInstantiationService { - if (!(services instanceof ServiceCollection)) { - throw new TypeError( - 'createChild requires a ServiceCollection instance (got something else)', - ); - } - const child = new TestInstantiationService(services, false, this); - (this as unknown as { _children: Set })._children.add(child); - return child; + return super.createChild(services) as TestInstantiationService; + } + + protected override _createChildService(services: ServiceCollection): InstantiationService { + return new TestInstantiationService(services, false, this); } public override dispose(): void { diff --git a/packages/agent-core-v2/src/_base/lifecycle/disposer.ts b/packages/agent-core-v2/src/_base/lifecycle/disposer.ts new file mode 100644 index 00000000000..1278dd77d23 --- /dev/null +++ b/packages/agent-core-v2/src/_base/lifecycle/disposer.ts @@ -0,0 +1,56 @@ +/** + * `_base.lifecycle` — disposer types shared by the Ledger. + * + * A `Disposer` undoes one registered side effect. Disposers are dual-track + * (sync / async), mirroring ES explicit resource management: a Ledger whose + * entries are all synchronous tears down within a single tick; any async + * entry suspends the teardown promise until it settles. + */ + +/** Why the ledger is being torn down; threaded through to every disposer. */ +export type TeardownReason = 'scope-close' | 'cascade' | 'unload'; + +export type Disposer = (reason: TeardownReason) => void | Promise; + +/** + * The four return forms accepted from an effect body: + * - `void` — nothing to roll back (the entry still exists for introspection); + * - `Disposer` — a single rollback action; + * - `Promise` — an asynchronously produced rollback action; + * - sync / async iterator of `Disposer`s — each yielded value is one rollback + * action; if iteration throws partway, the already-yielded disposers are + * rolled back in reverse before the error propagates. + */ +export type EffectResult = + | void + | Disposer + | Promise + | Iterable + | AsyncIterable; + +export type EffectBody = () => EffectResult; + +export function isPromiseLike(value: unknown): value is PromiseLike { + return ( + typeof value === 'object' && + value !== null && + typeof (value as { then?: unknown }).then === 'function' + ); +} + +export function isSyncIterable(value: unknown): value is Iterable { + return ( + typeof value === 'object' && + value !== null && + typeof (value as { [Symbol.iterator]?: unknown })[Symbol.iterator] === 'function' + ); +} + +export function isAsyncIterable(value: unknown): value is AsyncIterable { + return ( + typeof value === 'object' && + value !== null && + typeof (value as { [Symbol.asyncIterator]?: unknown })[Symbol.asyncIterator] === + 'function' + ); +} diff --git a/packages/agent-core-v2/src/_base/lifecycle/errors.ts b/packages/agent-core-v2/src/_base/lifecycle/errors.ts new file mode 100644 index 00000000000..4ea32c5d982 --- /dev/null +++ b/packages/agent-core-v2/src/_base/lifecycle/errors.ts @@ -0,0 +1,16 @@ +/** + * `_base.lifecycle` — Ledger errors. + */ + +export class LedgerDisposedError extends Error { + constructor( + readonly ledgerLabel: string, + readonly operation: string, + readonly ledgerState: 'disposing' | 'disposed', + ) { + super( + `Ledger '${ledgerLabel}' is ${ledgerState}: cannot ${operation} on a ${ledgerState} ledger`, + ); + this.name = 'LedgerDisposedError'; + } +} diff --git a/packages/agent-core-v2/src/_base/lifecycle/index.ts b/packages/agent-core-v2/src/_base/lifecycle/index.ts new file mode 100644 index 00000000000..0fcdf1c5205 --- /dev/null +++ b/packages/agent-core-v2/src/_base/lifecycle/index.ts @@ -0,0 +1,4 @@ +export * from './disposer'; +export * from './errors'; +export * from './ledger'; +export * from './lifecycleMachine'; diff --git a/packages/agent-core-v2/src/_base/lifecycle/ledger.ts b/packages/agent-core-v2/src/_base/lifecycle/ledger.ts new file mode 100644 index 00000000000..3dd4846a54d --- /dev/null +++ b/packages/agent-core-v2/src/_base/lifecycle/ledger.ts @@ -0,0 +1,352 @@ +/** + * `_base.lifecycle` — `Ledger`: an ordered book of rollbackable registrations. + * + * A Ledger records entries (disposers, effects, child ledgers) in registration + * order and tears them down in strict reverse order, awaiting each entry + * serially — never in parallel. Rollback is uninterruptible: a failing entry + * is logged (with its label) and teardown continues. Registering into a + * disposing/disposed ledger throws immediately. + * + * The Ledger knows nothing about DI; scopes and containers build on top of it. + */ + +import { onUnexpectedError } from '../errors/unexpectedError'; +import { + isAsyncIterable, + isPromiseLike, + isSyncIterable, + type Disposer, + type EffectBody, + type TeardownReason, +} from './disposer'; +import { LedgerDisposedError } from './errors'; + +export type LedgerState = 'active' | 'disposing' | 'disposed'; + +export interface LedgerEntryInfo { + readonly label: string; + readonly kind: 'disposer' | 'effect' | 'ledger'; + readonly stack?: string; + readonly children?: readonly LedgerEntryInfo[]; +} + +/** Handle to one ledger entry: remove it, or remove-and-run it. */ +export interface LedgerEntry { + readonly label: string; + /** True once the entry has been disposed, released, or torn down. */ + readonly disposed: boolean; + /** Remove the entry and run its disposer (guarded). Idempotent. */ + dispose(reason?: TeardownReason): void | Promise; + /** Remove the entry without running its disposer. Idempotent. */ + release(): void; +} + +interface EntryRecord { + label: string; + kind: 'disposer' | 'effect' | 'ledger'; + stack?: string; + active: boolean; + run: Disposer; + /** Set for child-ledger entries, for introspection. */ + ledger?: Ledger; +} + +export class Ledger { + /** Dev-mode toggle: capture the registration stack on every entry. */ + static captureStacks = false; + + private _state: LedgerState = 'active'; + private readonly _records: EntryRecord[] = []; + private _teardownPromise: Promise | undefined; + private _parentEntry: LedgerEntry | undefined; + + constructor(readonly label: string = 'ledger') {} + + get state(): LedgerState { + return this._state; + } + + get isActive(): boolean { + return this._state === 'active'; + } + + get isDisposed(): boolean { + return this._state === 'disposed'; + } + + /** Number of live entries. */ + get size(): number { + return this._records.reduce((count, record) => count + (record.active ? 1 : 0), 0); + } + + register(disposer: Disposer, label: string = 'disposer'): LedgerEntry { + this._assertActive('register'); + return this._push({ label, kind: 'disposer', active: true, run: disposer }); + } + + effect(body: EffectBody, label: string = 'effect'): LedgerEntry { + this._assertActive('effect'); + const out = body(); + if (typeof out === 'function') { + return this._push({ label, kind: 'effect', active: true, run: out }); + } + if (isPromiseLike(out)) { + const promise = Promise.resolve(out); + return this._push({ + label, + kind: 'effect', + active: true, + run: async (reason) => { + const disposer = await promise; + if (typeof disposer === 'function') await disposer(reason); + }, + }); + } + if (isAsyncIterable(out)) { + const promise = driveAsyncEffect(out); + return this._push({ + label, + kind: 'effect', + active: true, + run: async (reason) => { + const disposer = await promise; + await disposer(reason); + }, + }); + } + if (isSyncIterable(out)) { + // Drives the iterator immediately; a mid-iteration throw rolls back the + // already-yielded disposers before rethrowing (construction failure). + const run = driveSyncEffect(out); + return this._push({ label, kind: 'effect', active: true, run }); + } + return this._push({ label, kind: 'effect', active: true, run: () => {} }); + } + + /** A child ledger is itself one entry of this ledger. */ + createChild(label: string = 'ledger'): Ledger { + this._assertActive('createChild'); + const child = new Ledger(label); + child._parentEntry = this._push({ + label, + kind: 'ledger', + active: true, + run: (reason) => child.teardown(reason), + ledger: child, + }); + return child; + } + + /** + * Tear down every entry in strict reverse registration order, awaiting each + * one serially. Idempotent: a second call while disposing returns the + * in-flight promise; after disposal it is a no-op. Returns `undefined` when + * every entry completed synchronously (state is then synchronously + * `disposed`), otherwise a promise that settles once teardown completes. + */ + teardown(reason: TeardownReason = 'scope-close'): void | Promise { + if (this._state !== 'active') { + return this._teardownPromise; + } + this._state = 'disposing'; + this._detachFromParent(); + const out = drainRecords(this._records, reason); + if (isPromiseLike(out)) { + this._teardownPromise = Promise.resolve(out).then(() => { + this._state = 'disposed'; + }); + return this._teardownPromise; + } + this._state = 'disposed'; + return undefined; + } + + /** Tear down all current entries but keep the ledger active. */ + clear(reason: TeardownReason = 'scope-close'): void | Promise { + this._assertActive('clear'); + return drainRecords(this._records, reason); + } + + /** Introspection snapshot of the live entries (child ledgers recurse). */ + entries(): LedgerEntryInfo[] { + const infos: LedgerEntryInfo[] = []; + for (const record of this._records) { + if (!record.active) continue; + infos.push({ + label: record.label, + kind: record.kind, + stack: record.stack, + children: record.ledger?.entries(), + }); + } + return infos; + } + + private _push(record: EntryRecord): LedgerEntry { + if (Ledger.captureStacks) { + record.stack = new Error('Ledger registration').stack; + } + this._records.push(record); + return { + label: record.label, + get disposed() { + return !record.active; + }, + dispose: (reason: TeardownReason = 'scope-close') => { + if (!record.active) return undefined; + record.active = false; + this._remove(record); + return runGuarded(record, reason); + }, + release: () => { + if (!record.active) return; + record.active = false; + this._remove(record); + }, + }; + } + + private _remove(record: EntryRecord): void { + const index = this._records.indexOf(record); + if (index >= 0) { + this._records.splice(index, 1); + } + } + + /** Called by a child ledger when it tears itself down. */ + private _detachFromParent(): void { + this._parentEntry?.release(); + this._parentEntry = undefined; + } + + private _assertActive(operation: string): void { + if (this._state !== 'active') { + throw new LedgerDisposedError(this.label, operation, this._state); + } + } +} + +/** Run one entry's disposer, logging (with label) instead of throwing. */ +function runGuarded(record: EntryRecord, reason: TeardownReason): void | Promise { + let out: void | Promise; + try { + out = record.run(reason); + } catch (error) { + onUnexpectedError(tagged(error, record.label)); + return undefined; + } + if (isPromiseLike(out)) { + return Promise.resolve(out).catch((error: unknown) => { + onUnexpectedError(tagged(error, record.label)); + }); + } + return undefined; +} + +function tagged(error: unknown, label: string): unknown { + if (error instanceof Error) { + error.message = `[ledger:${label}] ${error.message}`; + return error; + } + return new Error(`[ledger:${label}] ${String(error)}`); +} + +/** + * Tear down a record list from the tail, serially. Sync fast path: when no + * entry returns a promise, the whole drain completes within the tick. + */ +function drainRecords(records: EntryRecord[], reason: TeardownReason): void | Promise { + let index = records.length; + const step = (): void | Promise => { + while (index > 0) { + index -= 1; + const record = records[index]!; + if (!record.active) continue; + record.active = false; + const out = runGuarded(record, reason); + if (isPromiseLike(out)) { + return Promise.resolve(out).then(step); + } + } + records.length = 0; + return undefined; + }; + return step(); +} + +/** Run a fixed disposer list in reverse, serially, guarding each entry. */ +function runDisposersReverse( + disposers: readonly Disposer[], + reason: TeardownReason, +): void | Promise { + let index = disposers.length; + const step = (): void | Promise => { + while (index > 0) { + index -= 1; + const disposer = disposers[index]!; + let out: void | Promise; + try { + out = disposer(reason); + } catch (error) { + onUnexpectedError(tagged(error, 'effect')); + continue; + } + if (isPromiseLike(out)) { + return Promise.resolve(out) + .catch((error: unknown) => { onUnexpectedError(tagged(error, 'effect')); }) + .then(step); + } + } + return undefined; + }; + return step(); +} + +function collect(step: IteratorResult, disposers: Disposer[]): void { + if (typeof step.value === 'function') { + disposers.push(step.value); + } +} + +/** + * Drive a sync effect iterator to completion now. On a mid-iteration throw, + * the already-yielded disposers are rolled back in reverse (sync fast path; + * an async rollback continues in the background) and the error is rethrown. + */ +function driveSyncEffect(iterable: Iterable): Disposer { + const disposers: Disposer[] = []; + const iterator = iterable[Symbol.iterator](); + try { + let step = iterator.next(); + while (!step.done) { + collect(step, disposers); + step = iterator.next(); + } + collect(step, disposers); + } catch (error) { + const rollback = runDisposersReverse(disposers, 'unload'); + if (isPromiseLike(rollback)) { + Promise.resolve(rollback).catch((error: unknown) => { onUnexpectedError(error); }); + } + throw error; + } + return (reason) => runDisposersReverse(disposers, reason); +} + +/** Async counterpart of {@link driveSyncEffect}; resolves to the composite disposer. */ +async function driveAsyncEffect(iterable: AsyncIterable): Promise { + const disposers: Disposer[] = []; + const iterator = iterable[Symbol.asyncIterator](); + try { + let step = await iterator.next(); + while (!step.done) { + collect(step, disposers); + step = await iterator.next(); + } + collect(step, disposers); + } catch (error) { + await runDisposersReverse(disposers, 'unload'); + throw error; + } + return (reason) => runDisposersReverse(disposers, reason); +} diff --git a/packages/agent-core-v2/test/_base/di/cascade.test.ts b/packages/agent-core-v2/test/_base/di/cascade.test.ts new file mode 100644 index 00000000000..a24ea8e9e1f --- /dev/null +++ b/packages/agent-core-v2/test/_base/di/cascade.test.ts @@ -0,0 +1,653 @@ +import { afterEach, describe, expect, it } from 'vitest'; + +import { SyncDescriptor } from '#/_base/di/descriptors'; +import { CascadeConflictError } from '#/_base/di/errors'; +import { createDecorator } from '#/_base/di/instantiation'; +import { InstantiationService } from '#/_base/di/instantiationService'; +import { ServiceCollection } from '#/_base/di/serviceCollection'; +import type { Ledger } from '#/_base/lifecycle/ledger'; + +function deferred(): { + promise: Promise; + resolve: (value?: T | PromiseLike) => void; + reject: (reason?: unknown) => void; +} { + let resolve!: (value?: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res as (value?: T | PromiseLike) => void; + reject = rej; + }); + return { promise, resolve, reject }; +} + +function ledgerOf(ix: InstantiationService): Ledger { + return (ix as unknown as { _ledger: Ledger })._ledger; +} + +/** Flush all pending microtask chains (async teardown hops). */ +function flushMicrotasks(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)); +} + +// ------------------------------------------------------------------ fixtures + +interface IRoot { + label: string; +} +const IRoot = createDecorator('cascade-root'); + +interface IMid { + root: IRoot; +} +const IMid = createDecorator('cascade-mid'); + +interface ILeaf { + mid: IMid; +} +const ILeaf = createDecorator('cascade-leaf'); + +interface IExtra { + label: string; +} +const IExtra = createDecorator('cascade-extra'); + +/** Shared per-test event log; fixtures push construct/dispose events. */ +let events: string[] = []; + +class Root implements IRoot { + label = 'root'; + constructor(public readonly tag = 'root') { + events.push(`+${this.tag}`); + } + dispose(): void { + events.push(`-${this.tag}`); + } +} + +class Mid implements IMid { + constructor(@IRoot public readonly root: IRoot) { + events.push('+mid'); + } + dispose(): void { + events.push('-mid'); + } +} + +class Leaf implements ILeaf { + constructor(@IMid public readonly mid: IMid) { + events.push('+leaf'); + } + dispose(): void { + events.push('-leaf'); + } +} + +class Extra implements IExtra { + label = 'extra'; + constructor() { + events.push('+extra'); + } + dispose(): void { + events.push('-extra'); + } +} + +function makeContainer(strict = true): InstantiationService { + return new InstantiationService(new ServiceCollection(), strict); +} + +function provideChain(ix: InstantiationService): void { + ix.provide(IRoot, new SyncDescriptor(Root)); + ix.provide(IMid, new SyncDescriptor(Mid)); + ix.provide(ILeaf, new SyncDescriptor(Leaf)); +} + +afterEach(() => { + events = []; +}); + +describe('cascade engine — mechanism matrix', () => { + it('1. provide X auto-activates dependents from Pending', () => { + const ix = makeContainer(); + events = []; + // Dependent provided before its dependency: goes to the waiting area. + ix.provide(IMid, new SyncDescriptor(Mid)); + expect(ix.cascade.unitState(IMid)).toBe('Pending'); + expect(events).toEqual([]); + + ix.provide(IRoot, new SyncDescriptor(Root)); + expect(ix.cascade.unitState(IRoot)).toBe('Active'); + expect(ix.cascade.unitState(IMid)).toBe('Active'); + expect(events).toEqual(['+root', '+mid']); + ix.dispose(); + }); + + it('2. unprovide X tears transitive dependents down in reverse topo order, back to Pending', () => { + const ix = makeContainer(); + provideChain(ix); + events = []; + const mid = ix.invokeFunction((a) => a.get(IMid)) as Mid; + const leaf = ix.invokeFunction((a) => a.get(ILeaf)) as Leaf; + + ix.unprovide(IRoot); + + expect(events).toEqual(['-leaf', '-mid', '-root']); + expect(ix.cascade.unitState(IRoot)).toBeUndefined(); // removed, not a state + expect(ix.cascade.unitState(IMid)).toBe('Pending'); + expect(ix.cascade.unitState(ILeaf)).toBe('Pending'); + expect(() => ix.invokeFunction((a) => a.get(IRoot))).toThrow(/unknown service/); + // The waiting area retains recipes with the missing token indexed. + expect(ix.cascade.pendingSnapshot().get('cascade-mid')).toEqual(['cascade-root']); + void mid; + void leaf; + ix.dispose(); + }); + + it('3. re-provide rebuilds the waiting area in topo order with fresh instances', () => { + const ix = makeContainer(); + provideChain(ix); + const firstMid = ix.invokeFunction((a) => a.get(IMid)); + const firstLeaf = ix.invokeFunction((a) => a.get(ILeaf)); + ix.unprovide(IRoot); + events = []; + + ix.provide(IRoot, new SyncDescriptor(Root)); + + expect(events).toEqual(['+root', '+mid', '+leaf']); + const secondMid = ix.invokeFunction((a) => a.get(IMid)); + const secondLeaf = ix.invokeFunction((a) => a.get(ILeaf)); + expect(secondMid).not.toBe(firstMid); + expect(secondLeaf).not.toBe(firstLeaf); + expect(secondMid.root).toBe(ix.invokeFunction((a) => a.get(IRoot))); + ix.dispose(); + }); + + it('4. replace is a single transaction: dependents rebuild against the new generation', () => { + const ix = makeContainer(); + provideChain(ix); + const firstMid = ix.invokeFunction((a) => a.get(IMid)); + events = []; + + // Replace the root recipe in one transaction. + class Root2 implements IRoot { + label = 'root2'; + constructor() { + events.push('+root2'); + } + dispose(): void { + events.push('-root2'); + } + } + const historyBefore = ix.cascade.history().length; + ix.provide(IRoot, new SyncDescriptor(Root2)); + + // One transaction covered teardown + rebuild; nothing lingered in Pending. + expect(ix.cascade.history().length).toBe(historyBefore + 1); + const entry = ix.cascade.history().at(-1)!; + expect(entry.tornDown).toEqual(['cascade-leaf', 'cascade-mid', 'cascade-root']); + expect(entry.rebuilt).toEqual(['cascade-root', 'cascade-mid', 'cascade-leaf']); + expect(events).toEqual(['-leaf', '-mid', '-root', '+root2', '+mid', '+leaf']); + expect(ix.cascade.unitState(IMid)).toBe('Active'); + expect(ix.cascade.unitState(ILeaf)).toBe('Active'); + const newMid = ix.invokeFunction((a) => a.get(IMid)); + expect(newMid).not.toBe(firstMid); + expect(newMid.root).toBeInstanceOf(Root2); + ix.dispose(); + }); + + it('5/6. requests submitted during a cascade queue up and merge their contagion sets', async () => { + const ix = makeContainer(); + ix.provide(IRoot, new SyncDescriptor(Root)); + ix.provide(IExtra, new SyncDescriptor(Extra)); + const gate = deferred(); + const hookCalls: string[][] = []; + let calls = 0; + ix.cascade.configure({ + onWillCascade: (affected) => { + calls += 1; + hookCalls.push(affected.map(String)); + // Park only the first transaction at the abort wait. + return calls === 1 ? gate.promise : undefined; + }, + }); + + const first = ix.cascade.submit({ + action: 'unprovide', + token: IRoot, + reason: 'drop root', + }); + // These two queue behind the in-flight transaction and merge into one. + const second = ix.cascade.submit({ + action: 'unprovide', + token: IExtra, + reason: 'drop extra', + }); + const third = ix.cascade.submit({ + action: 'provide', + token: IMid, + descriptor: new SyncDescriptor(Mid), + reason: 'add mid', + }); + + // Queued changes are not applied while the first transaction is in flight. + expect(ix.cascade.isInFlight(IRoot)).toBe(true); + // A token outside the in-flight contagion set resolves normally. + expect(ix.invokeFunction((a) => a.get(IExtra))).toBeInstanceOf(Extra); + + gate.resolve(); + await Promise.all([first, second, third]); + + // Three requests, two transactions: the queued two merged (one hook call each). + expect(calls).toBe(2); + // (The first two history entries are the initial provides.) + const history = ix.cascade.history().slice(-2); + expect(history[0]!.changes).toEqual([{ token: 'cascade-root', action: 'unprovide' }]); + expect(history[1]!.changes).toEqual([ + { token: 'cascade-extra', action: 'unprovide' }, + { token: 'cascade-mid', action: 'provide' }, + ]); + expect(ix.cascade.unitState(IExtra)).toBeUndefined(); + // Mid's dependency is gone: it waits. + expect(ix.cascade.unitState(IMid)).toBe('Pending'); + ix.dispose(); + }); + + it('7. construction failure is sticky Failed; update() reloads', () => { + const ix = makeContainer(); + let shouldThrow = true; + class Flaky implements IExtra { + label = 'flaky'; + constructor() { + if (shouldThrow) { + throw new Error('ctor boom'); + } + events.push('+flaky'); + } + } + ix.provide(IExtra, new SyncDescriptor(Flaky)); + expect(ix.cascade.unitState(IExtra)).toBe('Failed'); + // Resolving a failed unit rethrows the recorded error. + expect(() => ix.invokeFunction((a) => a.get(IExtra))).toThrow('ctor boom'); + + // A dependent of a Failed unit waits. + class NeedsExtra { + constructor(@IExtra public readonly extra: IExtra) {} + } + const INeedsExtra = createDecorator('cascade-needs-extra'); + ix.provide(INeedsExtra, new SyncDescriptor(NeedsExtra)); + expect(ix.cascade.unitState(INeedsExtra)).toBe('Pending'); + + // Sticky: an unrelated transaction does not retry the failed unit. + ix.provide(IRoot, new SyncDescriptor(Root)); + expect(ix.cascade.unitState(IExtra)).toBe('Failed'); + + // Explicit update() recovers it (and wakes its dependent). + shouldThrow = false; + events = []; + return ix.cascade.update(IExtra).then(() => { + expect(ix.cascade.unitState(IExtra)).toBe('Active'); + expect(ix.cascade.unitState(INeedsExtra)).toBe('Active'); + expect(events).toEqual(['+flaky']); + ix.dispose(); + }); + }); + + it('8. async disposers tear down serially in reverse topo order', async () => { + const ix = makeContainer(); + const gates = { root: deferred(), mid: deferred(), leaf: deferred() }; + const makeAsync = (label: string, gate: Promise) => + class { + dispose(): void { + events.push(`${label}-start`); + return gate.then(() => { + events.push(`${label}-end`); + }) as unknown as void; + } + }; + class AsyncRoot extends makeAsync('root', gates.root.promise) implements IRoot { + label = 'root'; + } + class AsyncMid extends makeAsync('mid', gates.mid.promise) implements IMid { + constructor(@IRoot public readonly root: IRoot) { + super(); + } + } + class AsyncLeaf extends makeAsync('leaf', gates.leaf.promise) implements ILeaf { + constructor(@IMid public readonly mid: IMid) { + super(); + } + } + ix.provide(IRoot, new SyncDescriptor(AsyncRoot)); + ix.provide(IMid, new SyncDescriptor(AsyncMid)); + ix.provide(ILeaf, new SyncDescriptor(AsyncLeaf)); + events = []; + + const done = ix.cascade.submit({ action: 'unprovide', token: IRoot, reason: 'async teardown' }); + // Reverse topo: leaf starts first; mid must not start until leaf finishes. + expect(events).toEqual(['leaf-start']); + gates.root.resolve(); + await flushMicrotasks(); + expect(events).toEqual(['leaf-start']); + gates.leaf.resolve(); + await flushMicrotasks(); + expect(events).toEqual(['leaf-start', 'leaf-end', 'mid-start']); + gates.mid.resolve(); + await done; + expect(events).toEqual(['leaf-start', 'leaf-end', 'mid-start', 'mid-end', 'root-start', 'root-end']); + ix.dispose(); + }); + + it('9. the abort hook cancels in-flight work (bounded wait), then forces through on timeout', async () => { + const ix = makeContainer(); + provideChain(ix); + const seen: { affected: string[]; reason: string }[] = []; + let gate: { promise: Promise; resolve: () => void } | undefined; + ix.cascade.configure({ + abortWaitMs: 30, + onWillCascade: (affected, reason) => { + seen.push({ affected: affected.map((ref) => ref.token.toString()), reason }); + gate = deferred(); + return gate.promise; + }, + }); + + // (a) the cascade waits for the abort to complete. + const first = ix.cascade.submit({ action: 'unprovide', token: IRoot, reason: 'feature "x" unloaded' }); + expect(seen).toHaveLength(1); + expect(seen[0]!.reason).toBe('feature "x" unloaded'); + expect(seen[0]!.affected).toContain('cascade-leaf'); + await Promise.resolve(); + expect(ix.cascade.isInFlight(IRoot)).toBe(true); // still waiting + gate!.resolve(); + await first; + expect(ix.cascade.history().at(-1)!.abortWaited).toBe(true); + expect(ix.cascade.history().at(-1)!.abortTimedOut).toBe(false); + expect(ix.cascade.unitState(IRoot)).toBeUndefined(); + + // (b) an abort that never completes is forced after the bounded wait. + provideChain(ix); + const second = ix.cascade.submit({ action: 'unprovide', token: IRoot, reason: 'forced' }); + await second; // never resolved the gate — the bound fired + const entry = ix.cascade.history().at(-1)!; + expect(entry.abortWaited).toBe(true); + expect(entry.abortTimedOut).toBe(true); + expect(ix.cascade.unitState(IRoot)).toBeUndefined(); + ix.dispose(); + }); + + it('10. a resolution hitting the in-flight subgraph suspends and completes after the transaction', async () => { + const ix = makeContainer(); + provideChain(ix); + const gate = deferred(); + ix.cascade.configure({ onWillCascade: () => gate.promise, resolveTimeoutMs: 50 }); + + const replace = ix.cascade.submit({ + action: 'provide', + token: IRoot, + descriptor: new SyncDescriptor(Root), + reason: 'replace root', + }); + expect(ix.cascade.isInFlight(IRoot)).toBe(true); + + // The sync path cannot suspend: it fails fast. + expect(() => ix.invokeFunction((a) => a.get(IRoot))).toThrow(CascadeConflictError); + + // The async path suspends until the transaction completes. + const suspended = ix.cascade.resolveWhenAvailable(IRoot); + gate.resolve(); + await replace; + const root = await suspended; + expect(root).toBeInstanceOf(Root); + + // Timeout variant: a transaction that parks forever rejects suspended resolutions. + const parked = deferred(); + ix.cascade.configure({ onWillCascade: () => parked.promise }); + void ix.cascade.submit({ action: 'unprovide', token: IRoot, reason: 'parked' }); + await expect(ix.cascade.resolveWhenAvailable(IRoot)).rejects.toThrow(CascadeConflictError); + parked.resolve(); + await ix.cascade.whenIdle(); + ix.dispose(); + }); + + it('11. cycle detection holds under dynamic edge add/remove', () => { + const ix = makeContainer(); + // Cyclic recipes provided dynamically: neither can satisfy its dependency, + // so both wait — no construction, no spurious failure. + const IA = createDecorator<{ a: true }>('cascade-cyc-a'); + const IB = createDecorator<{ b: true }>('cascade-cyc-b'); + class A { + constructor(@IB public readonly b: unknown) {} + } + class B { + constructor(@IA public readonly a: unknown) {} + } + ix.provide(IA, new SyncDescriptor(A)); + ix.provide(IB, new SyncDescriptor(B)); + expect(ix.cascade.unitState(IA)).toBe('Pending'); + expect(ix.cascade.unitState(IB)).toBe('Pending'); + expect(ix.dependencyGraph.findCycle((ref) => ref.token.toString())).toBeNull(); + + // Break the cycle: replace A with an independent recipe; both activate. + class A2 { + readonly a = true; + } + ix.provide(IA, new SyncDescriptor(A2)); + expect(ix.cascade.unitState(IA)).toBe('Active'); + expect(ix.cascade.unitState(IB)).toBe('Active'); + expect(ix.dependencyGraph.findCycle((ref) => ref.token.toString())).toBeNull(); + + // Dynamic chain teardown keeps the graph acyclic and edge-free. + ix.unprovide(IA); + expect(ix.cascade.unitState(IB)).toBe('Pending'); + expect(ix.dependencyGraph.edges()).toHaveLength(0); + expect(ix.dependencyGraph.findCycle((ref) => ref.token.toString())).toBeNull(); + ix.dispose(); + }); + + it('12. ledger balance: arbitrary sequences leave no leaks or dangling edges', async () => { const ix = makeContainer(); + provideChain(ix); + // 3 live instances + 3 provide entries on the book. + expect(ledgerOf(ix).size).toBe(6); + + ix.unprovide(IMid); + // Left on the book: the root instance entry + root/leaf provide entries. + expect(ledgerOf(ix).size).toBe(3); + // Leaf is Pending (waiting on mid); mid is removed; root is Active. + expect(ix.cascade.unitState(ILeaf)).toBe('Pending'); + expect(ix.cascade.unitState(IMid)).toBeUndefined(); + + ix.provide(IMid, new SyncDescriptor(Mid)); + expect(ix.cascade.unitState(ILeaf)).toBe('Active'); + expect(ledgerOf(ix).size).toBe(6); + + await ix.cascade.update(IRoot); + expect(ledgerOf(ix).size).toBe(6); + expect(ix.dependencyGraph.edges()).toHaveLength(2); + + ix.unprovide(ILeaf); + ix.unprovide(IMid); + ix.unprovide(IRoot); + expect(ledgerOf(ix).size).toBe(0); + expect(ix.dependencyGraph.edges()).toHaveLength(0); + expect(ix.dependencyGraph.findCycle((ref) => ref.token.toString())).toBeNull(); + expect(ix.cascade.pendingSnapshot().size).toBe(0); + + ix.dispose(); + expect(ledgerOf(ix).size).toBe(0); + }); + + it('13. replacing with a concrete instance cascades into live dependents (D1)', () => { + const ix = makeContainer(); + provideChain(ix); + const firstMid = ix.invokeFunction((a) => a.get(IMid)); + const replacement = new Root('root2'); + events = []; + + ix.provide(IRoot, replacement); + + // Same transaction: dependents were torn down and rebuilt against the instance. + expect(events).toEqual(['-leaf', '-mid', '-root', '+mid', '+leaf']); + const newMid = ix.invokeFunction((a) => a.get(IMid)); + expect(newMid).not.toBe(firstMid); + expect(newMid.root).toBe(replacement); + expect(ix.invokeFunction((a) => a.get(IRoot))).toBe(replacement); + ix.dispose(); + }); + + it('14. a rejecting abort hook is logged, never a veto (best-effort §4.5)', async () => { + const reported: unknown[] = []; + const { setUnexpectedErrorHandler, resetUnexpectedErrorHandler } = await import( + '#/_base/errors/unexpectedError' + ); + setUnexpectedErrorHandler((err) => { reported.push(err); }); + try { + const ix = makeContainer(); + provideChain(ix); + ix.cascade.configure({ + onWillCascade: () => Promise.reject(new Error('abort hook blew up')), + }); + + await ix.cascade.submit({ action: 'unprovide', token: IRoot, reason: 'forced anyway' }); + + expect(ix.cascade.unitState(IRoot)).toBeUndefined(); + expect(() => ix.invokeFunction((a) => a.get(IRoot))).toThrow(/unknown service/); + expect(ix.cascade.history().at(-1)!.abortWaited).toBe(true); + expect(reported).toHaveLength(1); + expect((reported[0] as Error).message).toContain('abort hook blew up'); + ix.dispose(); + } finally { + resetUnexpectedErrorHandler(); + } + }); +}); + +describe('cascade engine — cross-scope orchestration (D9)', () => { + it('a parent change cascades into child-scope dependents and rebuilds them', () => { + const parent = makeContainer(); + parent.provide(IRoot, new SyncDescriptor(Root)); + const child = parent.createChild(new ServiceCollection()); + child.provide(IMid, new SyncDescriptor(Mid)); + events = []; + + parent.unprovide(IRoot); + // Global reverse topo: the child unit dies before its parent dependency. + expect(events).toEqual(['-mid', '-root']); + expect(child.cascade.unitState(IMid)).toBe('Pending'); + expect(parent.cascade.unitState(IRoot)).toBeUndefined(); + + parent.provide(IRoot, new SyncDescriptor(Root)); + // Global topo rebuild: parent first, then the child dependent. + expect(events).toEqual(['-mid', '-root', '+root', '+mid']); + const mid = child.invokeFunction((a) => a.get(IMid)); + expect(mid.root).toBe(parent.invokeFunction((a) => a.get(IRoot))); + parent.dispose(); + }); + + it('orders a three-level chain globally: deepest first for teardown, reverse for rebuild', () => { + const parent = makeContainer(); + parent.provide(IRoot, new SyncDescriptor(Root)); + const child = parent.createChild(new ServiceCollection()); + child.provide(IMid, new SyncDescriptor(Mid)); + const grandchild = child.createChild(new ServiceCollection()); + grandchild.provide(ILeaf, new SyncDescriptor(Leaf)); + events = []; + + parent.unprovide(IRoot); + expect(events).toEqual(['-leaf', '-mid', '-root']); + + parent.provide(IRoot, new SyncDescriptor(Root)); + expect(events).toEqual(['-leaf', '-mid', '-root', '+root', '+mid', '+leaf']); + parent.dispose(); + }); + + it('shadowing: a child shadow of the changed token is not in the contagion set', () => { + const parent = makeContainer(); + parent.provide(IRoot, new SyncDescriptor(Root)); + const child = parent.createChild(new ServiceCollection()); + // The child registers its own IRoot; the child's Mid binds the shadow. + child.provide(IRoot, new SyncDescriptor(Root, ['shadow'])); + child.provide(IMid, new SyncDescriptor(Mid)); + events = []; + + parent.unprovide(IRoot); + // Only the parent's own root is retired; the child's shadow and its + // dependent are untouched. + expect(events).toEqual(['-root']); + expect(child.cascade.unitState(IMid)).toBe('Active'); + expect(child.cascade.unitState(IRoot)).toBe('Active'); + const mid = child.invokeFunction((a) => a.get(IMid)); + expect(mid.root).toBe(child.invokeFunction((a) => a.get(IRoot))); + parent.dispose(); + }); + + it('siblings are isolated: one child scope\'s change never touches the other', () => { + const parent = makeContainer(); + parent.provide(IRoot, new SyncDescriptor(Root)); + const childA = parent.createChild(new ServiceCollection()); + const childB = parent.createChild(new ServiceCollection()); + childA.provide(IMid, new SyncDescriptor(Mid)); + childB.provide(IMid, new SyncDescriptor(Mid)); + events = []; + + childA.unprovide(IMid); + expect(events).toEqual(['-mid']); + expect(childB.cascade.unitState(IMid)).toBe('Active'); + + // But a parent change reaches both subtrees. + parent.unprovide(IRoot); + expect(events).toEqual(['-mid', '-mid', '-root']); + expect(childB.cascade.unitState(IMid)).toBe('Pending'); + parent.dispose(); + }); + + it('a descendant scope dying mid-transaction is skipped idempotently', () => { + const parent = makeContainer(); + parent.provide(IRoot, new SyncDescriptor(Root)); + const child = parent.createChild(new ServiceCollection()); + child.provide(IMid, new SyncDescriptor(Mid)); + events = []; + parent.cascade.configure({ + onWillCascade: () => { + child.dispose(); + }, + }); + + parent.unprovide(IRoot); + + // The child's own dispose already retired mid; the cascade skips the dead + // scope's units and completes its own teardown. + expect(events).toEqual(['-mid', '-root']); + const entry = parent.cascade.history().at(-1)!; + expect(entry.tornDown).toEqual(['cascade-root']); + expect(parent.cascade.unitState(IRoot)).toBeUndefined(); + parent.dispose(); + }); + + it('the in-flight guard and suspension work across scopes', async () => { + const parent = makeContainer(); + parent.provide(IRoot, new SyncDescriptor(Root)); + const child = parent.createChild(new ServiceCollection()); + child.provide(IMid, new SyncDescriptor(Mid)); + const gate = deferred(); + parent.cascade.configure({ onWillCascade: () => gate.promise }); + + const tx = parent.cascade.submit({ + action: 'provide', + token: IRoot, + descriptor: new SyncDescriptor(Root), + reason: 'replace root', + }); + // The child sees its ancestor's token as in flight. + expect(child.cascade.isInFlight(IRoot)).toBe(true); + expect(() => child.invokeFunction((a) => a.get(IRoot))).toThrow(CascadeConflictError); + + const suspended = child.cascade.resolveWhenAvailable(IRoot); + gate.resolve(); + await tx; + const root = await suspended; + expect(root).toBeInstanceOf(Root); + expect(child.cascade.unitState(IMid)).toBe('Active'); + parent.dispose(); + }); +}); diff --git a/packages/agent-core-v2/test/_base/di/child.test.ts b/packages/agent-core-v2/test/_base/di/child.test.ts index 99d3da62e7b..eebce1e32c8 100644 --- a/packages/agent-core-v2/test/_base/di/child.test.ts +++ b/packages/agent-core-v2/test/_base/di/child.test.ts @@ -1,5 +1,9 @@ -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, afterEach } from 'vitest'; +import { + resetUnexpectedErrorHandler, + setUnexpectedErrorHandler, +} from '#/_base/errors/unexpectedError'; import { SyncDescriptor } from '#/_base/di/descriptors'; import { IInstantiationService, @@ -351,7 +355,7 @@ describe('InstantiationService.createChild', () => { }); describe('Disposable base class', () => { - it('insertion order on dispose', () => { + it('reverse registration order on dispose (ledger teardown)', () => { const events: string[] = []; class Child implements IDisposable { constructor(public readonly label: string) {} @@ -369,7 +373,7 @@ describe('Disposable base class', () => { } const o = new Owner(); o.dispose(); - expect(events).toEqual(['disposed first', 'disposed second', 'disposed third']); + expect(events).toEqual(['disposed third', 'disposed second', 'disposed first']); }); it('idempotent dispose on the base class', () => { @@ -409,8 +413,15 @@ describe('Disposable base class', () => { expect(events).toEqual(['disposed']); }); - it('continues teardown and rethrows if one child throws', () => { + it('continues teardown and reports if one child throws (rollback is uninterruptible)', () => { const events: string[] = []; + const reported: unknown[] = []; + setUnexpectedErrorHandler((err) => { + reported.push(err); + }); + afterEach(() => { + resetUnexpectedErrorHandler(); + }); class GoodChild implements IDisposable { dispose(): void { events.push('good'); @@ -436,7 +447,11 @@ describe('Disposable base class', () => { } } const o = new Owner(); - expect(() => { o.dispose(); }).toThrow('boom'); - expect(events).toEqual(['good', 'bad-attempted', 'tail']); + // Ledger semantics: teardown is uninterruptible — a failing entry is + // reported via onUnexpectedError and teardown continues (reverse order). + expect(() => { o.dispose(); }).not.toThrow(); + expect(events).toEqual(['tail', 'bad-attempted', 'good']); + expect(reported).toHaveLength(1); + expect((reported[0] as Error).message).toContain('boom'); }); }); diff --git a/packages/agent-core-v2/test/_base/di/provide.test.ts b/packages/agent-core-v2/test/_base/di/provide.test.ts new file mode 100644 index 00000000000..83c57efa804 --- /dev/null +++ b/packages/agent-core-v2/test/_base/di/provide.test.ts @@ -0,0 +1,239 @@ +import { describe, expect, it } from 'vitest'; + +import { SyncDescriptor } from '#/_base/di/descriptors'; +import { createDecorator } from '#/_base/di/instantiation'; +import { InstantiationService } from '#/_base/di/instantiationService'; +import type { IDisposable } from '#/_base/di/lifecycle'; +import { ServiceCollection } from '#/_base/di/serviceCollection'; +import type { AvailabilityChange } from '#/_base/di/serviceCollection'; + +interface IFoo { + tag: string; +} +const IFoo = createDecorator('provide-foo'); + +interface IBar { + tag: string; +} +const IBar = createDecorator('provide-bar'); + +class Foo implements IFoo, IDisposable { + tag = 'foo'; + disposed = false; + dispose(): void { + this.disposed = true; + } +} + +class Bar implements IBar { + tag = 'bar'; + constructor(@IFoo public readonly foo: IFoo) {} +} + +describe('InstantiationService.provide/unprovide (L1)', () => { + it('provides a service at runtime and resolves it', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + ix.provide(IFoo, new SyncDescriptor(Foo)); + const foo = ix.invokeFunction((a) => a.get(IFoo)); + expect(foo).toBeInstanceOf(Foo); + ix.dispose(); + }); + + it('unprovide removes the token; strict resolution then throws', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + ix.provide(IFoo, new SyncDescriptor(Foo)); + ix.invokeFunction((a) => a.get(IFoo)); + ix.unprovide(IFoo); + expect(() => ix.invokeFunction((a) => a.get(IFoo))).toThrow(/unknown service/); + ix.dispose(); + }); + + it('unprovide retires the materialized instance', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + ix.provide(IFoo, new SyncDescriptor(Foo)); + const foo = ix.invokeFunction((a) => a.get(IFoo)) as Foo; + expect(foo.disposed).toBe(false); + ix.unprovide(IFoo); + expect(foo.disposed).toBe(true); + ix.dispose(); + }); + + it('reprovide retires the old generation and resolves a fresh instance', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + ix.provide(IFoo, new SyncDescriptor(Foo)); + const first = ix.invokeFunction((a) => a.get(IFoo)) as Foo; + + ix.provide(IFoo, new SyncDescriptor(Foo)); + expect(first.disposed).toBe(true); + + const second = ix.invokeFunction((a) => a.get(IFoo)) as Foo; + expect(second).not.toBe(first); + expect(second.disposed).toBe(false); + ix.dispose(); + expect(second.disposed).toBe(true); + }); + + it('stamps every generation with a container-monotonic uid', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + const h1 = ix.provide(IFoo, new SyncDescriptor(Foo)); + const h2 = ix.provide(IBar, new SyncDescriptor(Bar)); + ix.provide(IFoo, new SyncDescriptor(Foo)); + const uidAfter = (ix as unknown as { _services: ServiceCollection })._services.uidOf(IFoo)!; + expect(h2.uid).toBeGreaterThan(h1.uid); + expect(uidAfter).toBeGreaterThan(h2.uid); + ix.dispose(); + }); + + it('fires availability events with { oldUid, newUid } on provide/unprovide', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + const changes: AvailabilityChange[] = []; + const services = (ix as unknown as { _services: ServiceCollection })._services; + services.onDidChange(IFoo, (change) => changes.push(change)); + + const h1 = ix.provide(IFoo, new SyncDescriptor(Foo)); + ix.provide(IFoo, new SyncDescriptor(Foo)); + const uid2 = services.uidOf(IFoo)!; + ix.unprovide(IFoo); + + expect(changes).toEqual([ + { oldUid: undefined, newUid: h1.uid }, + { oldUid: h1.uid, newUid: uid2 }, + { oldUid: uid2, newUid: undefined }, + ]); + ix.dispose(); + }); + + it('records the pinned flag on the entry', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + ix.provide(IFoo, new SyncDescriptor(Foo), { pinned: true }); + const services = (ix as unknown as { _services: ServiceCollection })._services; + expect(services.isPinned(IFoo)).toBe(true); + // Re-provide without the flag keeps the previous pinned metadata. + ix.provide(IFoo, new SyncDescriptor(Foo)); + expect(services.isPinned(IFoo)).toBe(true); + ix.dispose(); + }); + + it('the provide handle is a ledger entry: disposing it unprovides', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + const handle = ix.provide(IFoo, new SyncDescriptor(Foo)); + handle.dispose(); + expect(() => ix.invokeFunction((a) => a.get(IFoo))).toThrow(/unknown service/); + ix.dispose(); + }); + + it('container teardown retires provided services exactly once', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + ix.provide(IFoo, new SyncDescriptor(Foo)); + const foo = ix.invokeFunction((a) => a.get(IFoo)) as Foo; + let calls = 0; + const origDispose = foo.dispose.bind(foo); + foo.dispose = () => { + calls += 1; + origDispose(); + }; + ix.dispose(); + expect(calls).toBe(1); + }); +}); + +describe('persistent dependency graph (L2 substrate)', () => { + it('records constructor-injection edges for materialized services', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + ix.provide(IFoo, new SyncDescriptor(Foo)); + ix.provide(IBar, new SyncDescriptor(Bar)); + ix.invokeFunction((a) => a.get(IBar)); + + const edges = ix.dependencyGraph.edges(); + expect(edges).toHaveLength(1); + expect(edges[0]).toMatchObject({ + consumer: { scope: ix, token: IBar }, + dependency: { scope: ix, token: IFoo }, + kind: 'instance', + }); + ix.dispose(); + }); + + it('affectedSet computes the transitive dependents of a changed token', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + ix.provide(IFoo, new SyncDescriptor(Foo)); + ix.provide(IBar, new SyncDescriptor(Bar)); + ix.invokeFunction((a) => a.get(IBar)); + + const tokens = (refs: readonly { token: unknown }[]): unknown[] => + refs.map((ref) => ref.token); + expect(tokens(ix.dependencyGraph.affectedSet([{ scope: ix, token: IFoo }]))).toEqual([IFoo, IBar]); + expect(tokens(ix.dependencyGraph.affectedSet([{ scope: ix, token: IBar }]))).toEqual([IBar]); + ix.dispose(); + }); + + it('orders the affected set: dependents first for teardown, dependencies first for rebuild', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + ix.provide(IFoo, new SyncDescriptor(Foo)); + ix.provide(IBar, new SyncDescriptor(Bar)); + ix.invokeFunction((a) => a.get(IBar)); + + const affected = ix.dependencyGraph.affectedSet([{ scope: ix, token: IFoo }]); + const tokens = (refs: readonly { token: unknown }[]): unknown[] => + refs.map((ref) => ref.token); + expect(tokens(ix.dependencyGraph.reverseTopoOrder(affected))).toEqual([IBar, IFoo]); + expect(tokens(ix.dependencyGraph.topoOrder(affected))).toEqual([IFoo, IBar]); + ix.dispose(); + }); + + it('retiring a consumer removes its edges', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + ix.provide(IFoo, new SyncDescriptor(Foo)); + ix.provide(IBar, new SyncDescriptor(Bar)); + ix.invokeFunction((a) => a.get(IBar)); + + ix.unprovide(IBar); + expect(ix.dependencyGraph.edges()).toHaveLength(0); + const remaining = ix.dependencyGraph.affectedSet([{ scope: ix, token: IFoo }]); + expect(remaining.map((ref) => ref.token)).toEqual([IFoo]); + ix.dispose(); + }); + + it('container teardown leaves the graph empty (no dangling edges)', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + ix.provide(IFoo, new SyncDescriptor(Foo)); + ix.provide(IBar, new SyncDescriptor(Bar)); + ix.invokeFunction((a) => a.get(IBar)); + ix.dispose(); + expect(ix.dependencyGraph.edges()).toHaveLength(0); + }); + + it('does not track createInstance products (leaves)', () => { + const ix = new InstantiationService(new ServiceCollection(), true); + ix.provide(IFoo, new SyncDescriptor(Foo)); + class Leaf { + constructor(@IFoo public readonly foo: IFoo) {} + } + ix.createInstance(Leaf); + expect(ix.dependencyGraph.edges()).toHaveLength(0); + ix.dispose(); + }); +}); + +describe('TestInstantiationService.set rerouting', () => { + it('set() on a materialized token retires the previous generation', async () => { + const { TestInstantiationService } = await import('#/_base/di/testInstantiationService'); + const ix = new TestInstantiationService(new ServiceCollection(), true); + ix.set(IFoo, new SyncDescriptor(Foo)); + const first = ix.get(IFoo) as Foo; + ix.set(IFoo, new SyncDescriptor(Foo)); + expect(first.disposed).toBe(true); + const second = ix.get(IFoo) as Foo; + expect(second).not.toBe(first); + ix.dispose(); + }); + + it('set() returns the previous value like before', async () => { + const { TestInstantiationService } = await import('#/_base/di/testInstantiationService'); + const ix = new TestInstantiationService(new ServiceCollection(), true); + const seeded = new Foo(); + expect(ix.set(IFoo, seeded)).toBeUndefined(); + expect(ix.set(IFoo, new Foo())).toBe(seeded); + ix.dispose(); + }); +}); diff --git a/packages/agent-core-v2/test/_base/lifecycle/ledger.test.ts b/packages/agent-core-v2/test/_base/lifecycle/ledger.test.ts new file mode 100644 index 00000000000..5ea721df1a8 --- /dev/null +++ b/packages/agent-core-v2/test/_base/lifecycle/ledger.test.ts @@ -0,0 +1,427 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { + resetUnexpectedErrorHandler, + setUnexpectedErrorHandler, +} from '#/_base/errors/unexpectedError'; +import { LedgerDisposedError } from '#/_base/lifecycle/errors'; +import { Ledger } from '#/_base/lifecycle/ledger'; +import type { Disposer, TeardownReason } from '#/_base/lifecycle/disposer'; + +function deferred(): { + promise: Promise; + resolve: (value: T | PromiseLike) => void; + reject: (reason?: unknown) => void; +} { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +describe('Ledger', () => { + describe('teardown ordering', () => { + it('tears down entries in strict reverse registration order', () => { + const events: string[] = []; + const ledger = new Ledger('test'); + ledger.register(() => { events.push('first'); }, 'first'); + ledger.register(() => { events.push('second'); }, 'second'); + ledger.register(() => { events.push('third'); }, 'third'); + void ledger.teardown(); + expect(events).toEqual(['third', 'second', 'first']); + expect(ledger.state).toBe('disposed'); + }); + + it('completes synchronously when every entry is synchronous', () => { + const ledger = new Ledger('test'); + ledger.register(() => {}, 'a'); + ledger.register(() => {}, 'b'); + const out = ledger.teardown(); + expect(out).toBeUndefined(); + expect(ledger.state).toBe('disposed'); + expect(ledger.isDisposed).toBe(true); + }); + + it('mixes sync and async entries: async entries suspend teardown, order holds', async () => { + const events: string[] = []; + const gate = deferred(); + const ledger = new Ledger('test'); + ledger.register(() => { events.push('sync-1'); }, 'sync-1'); + ledger.register(async () => { + events.push('async-start'); + await gate.promise; + events.push('async-end'); + }, 'async'); + ledger.register(() => { events.push('sync-3'); }, 'sync-3'); + + const out = ledger.teardown(); + expect(out).toBeInstanceOf(Promise); + expect(ledger.state).toBe('disposing'); + // Reverse order: sync-3 runs first (same tick), then the async entry starts. + expect(events).toEqual(['sync-3', 'async-start']); + + gate.resolve(); + await out; + expect(events).toEqual(['sync-3', 'async-start', 'async-end', 'sync-1']); + expect(ledger.state).toBe('disposed'); + }); + + it('awaits each entry serially: the next disposer starts only after the previous resolves', async () => { + const events: string[] = []; + const gates = [deferred(), deferred()]; + const ledger = new Ledger('test'); + ledger.register(async () => { + events.push('a-start'); + await gates[0]!.promise; + events.push('a-end'); + }, 'a'); + ledger.register(async () => { + events.push('b-start'); + await gates[1]!.promise; + events.push('b-end'); + }, 'b'); + + const out = ledger.teardown(); + expect(events).toEqual(['b-start']); + // Resolving the later-registered gate must not unblock the earlier one. + gates[0]!.resolve(); + await Promise.resolve(); + expect(events).toEqual(['b-start']); + gates[1]!.resolve(); + await out; + expect(events).toEqual(['b-start', 'b-end', 'a-start', 'a-end']); + }); + }); + + describe('registration guards', () => { + it('throws when registering into a disposed ledger', () => { + const ledger = new Ledger('test'); + void ledger.teardown(); + expect(() => ledger.register(() => {}, 'late')).toThrow(LedgerDisposedError); + expect(() => ledger.effect(() => () => {}, 'late')).toThrow(LedgerDisposedError); + expect(() => ledger.createChild('late')).toThrow(LedgerDisposedError); + }); + + it('throws when registering while teardown is in flight', async () => { + const gate = deferred(); + const ledger = new Ledger('test'); + let caught: unknown; + ledger.register(async () => { + await gate.promise; + // Registration happens while the ledger is 'disposing': it must throw + // at the call site (the entry itself is guarded, so the error does not + // abort the in-flight teardown). + try { + ledger.register(() => {}, 'late'); + } catch (error) { + caught = error; + } + }, 'slow'); + + const out = ledger.teardown(); + gate.resolve(); + await out; + expect(caught).toBeInstanceOf(LedgerDisposedError); + expect(ledger.state).toBe('disposed'); + }); + }); + + describe('uninterruptible rollback', () => { + const reported: unknown[] = []; + + afterEach(() => { + reported.length = 0; + resetUnexpectedErrorHandler(); + }); + + it('a throwing entry is reported (with label) and teardown continues', () => { + setUnexpectedErrorHandler((err) => { reported.push(err); }); + const events: string[] = []; + const ledger = new Ledger('test'); + ledger.register(() => { events.push('first'); }, 'first'); + ledger.register(() => { + events.push('bad-attempted'); + throw new Error('boom'); + }, 'bad'); + ledger.register(() => { events.push('third'); }, 'third'); + + expect(() => ledger.teardown()).not.toThrow(); + expect(events).toEqual(['third', 'bad-attempted', 'first']); + expect(ledger.state).toBe('disposed'); + expect(reported).toHaveLength(1); + expect((reported[0] as Error).message).toContain('boom'); + expect((reported[0] as Error).message).toContain('bad'); + }); + + it('a rejecting async entry is reported and teardown continues', async () => { + setUnexpectedErrorHandler((err) => { reported.push(err); }); + const events: string[] = []; + const ledger = new Ledger('test'); + ledger.register(() => { events.push('first'); }, 'first'); + ledger.register(async () => { + events.push('bad-attempted'); + throw new Error('async boom'); + }, 'bad'); + ledger.register(() => { events.push('third'); }, 'third'); + + await ledger.teardown(); + expect(events).toEqual(['third', 'bad-attempted', 'first']); + expect(reported).toHaveLength(1); + expect((reported[0] as Error).message).toContain('async boom'); + }); + }); + + describe('construction failure auto-rollback', () => { + it('sync iterator: a mid-iteration throw rolls back already-yielded disposers in reverse', () => { + const events: string[] = []; + const ledger = new Ledger('test'); + expect(() => + ledger.effect(function* () { + yield () => { events.push('undo-1'); }; + yield () => { events.push('undo-2'); }; + throw new Error('construct failed'); + }, 'gen'), + ).toThrow('construct failed'); + expect(events).toEqual(['undo-2', 'undo-1']); + expect(ledger.size).toBe(0); + }); + + it('async iterator: a mid-iteration throw rolls back already-yielded disposers in reverse', async () => { + const events: string[] = []; + const ledger = new Ledger('test'); + // eslint-disable-next-line require-yield + const body = async function* (): AsyncGenerator { + yield () => { events.push('undo-1'); }; + yield () => { events.push('undo-2'); }; + throw new Error('async construct failed'); + }; + const entry = ledger.effect(body, 'gen'); + // The failure surfaces through the entry's disposer chain; teardown reports it. + const reported: unknown[] = []; + setUnexpectedErrorHandler((err) => { reported.push(err); }); + try { + await ledger.teardown(); + expect(events).toEqual(['undo-2', 'undo-1']); + expect(reported).toHaveLength(1); + expect((reported[0] as Error).message).toContain('async construct failed'); + } finally { + resetUnexpectedErrorHandler(); + } + expect(entry.disposed).toBe(true); + }); + }); + + describe('effect return forms', () => { + it('void body: entry exists for introspection, nothing to roll back', () => { + const ledger = new Ledger('test'); + ledger.effect(() => {}, 'noop'); + expect(ledger.size).toBe(1); + expect(ledger.entries()[0]).toMatchObject({ label: 'noop', kind: 'effect' }); + void ledger.teardown(); + expect(ledger.state).toBe('disposed'); + }); + + it('plain disposer', () => { + const events: string[] = []; + const ledger = new Ledger('test'); + ledger.effect(() => () => { events.push('disposed'); }, 'fx'); + void ledger.teardown(); + expect(events).toEqual(['disposed']); + }); + + it('Promise: teardown awaits the promise then runs the disposer', async () => { + const events: string[] = []; + const gate = deferred(); + const ledger = new Ledger('test'); + ledger.effect(() => gate.promise, 'async-fx'); + ledger.register(() => { events.push('first'); }, 'first'); + + const out = ledger.teardown(); + expect(events).toEqual(['first']); + gate.resolve(() => { events.push('async-disposed'); }); + await out; + expect(events).toEqual(['first', 'async-disposed']); + }); + + it('sync iterator: yields are rolled back in reverse at teardown', () => { + const events: string[] = []; + const ledger = new Ledger('test'); + ledger.effect(function* () { + events.push('setup-1'); + yield () => { events.push('undo-1'); }; + events.push('setup-2'); + yield () => { events.push('undo-2'); }; + }, 'gen'); + expect(events).toEqual(['setup-1', 'setup-2']); + void ledger.teardown(); + expect(events).toEqual(['setup-1', 'setup-2', 'undo-2', 'undo-1']); + }); + + it('async iterator: yields are rolled back in reverse at teardown', async () => { + const events: string[] = []; + const ledger = new Ledger('test'); + const body = async function* (): AsyncGenerator { + yield () => { events.push('undo-1'); }; + yield () => { events.push('undo-2'); }; + }; + ledger.effect(body, 'gen'); + await ledger.teardown(); + expect(events).toEqual(['undo-2', 'undo-1']); + }); + }); + + describe('child ledgers', () => { + it('a child ledger is one entry of the parent and tears down with it', () => { + const events: string[] = []; + const parent = new Ledger('parent'); + parent.register(() => { events.push('parent-entry'); }, 'parent-entry'); + const child = parent.createChild('child'); + child.register(() => { events.push('child-entry'); }, 'child-entry'); + + void parent.teardown(); + expect(events).toEqual(['child-entry', 'parent-entry']); + expect(child.state).toBe('disposed'); + expect(parent.state).toBe('disposed'); + }); + + it('a child torn down directly detaches from the parent', () => { + const events: string[] = []; + const parent = new Ledger('parent'); + const child = parent.createChild('child'); + child.register(() => { events.push('child-entry'); }, 'child-entry'); + + void child.teardown(); + expect(events).toEqual(['child-entry']); + expect(parent.entries()).toHaveLength(0); + + void parent.teardown(); + expect(events).toEqual(['child-entry']); + }); + }); + + describe('introspection', () => { + it('entries() renders the book as a tree', () => { + const parent = new Ledger('parent'); + parent.register(() => {}, 'a'); + parent.effect(() => () => {}, 'fx'); + const child = parent.createChild('child-scope'); + child.register(() => {}, 'child-a'); + + expect(parent.entries()).toEqual([ + { label: 'a', kind: 'disposer', stack: undefined, children: undefined }, + { label: 'fx', kind: 'effect', stack: undefined, children: undefined }, + { + label: 'child-scope', + kind: 'ledger', + stack: undefined, + children: [ + { label: 'child-a', kind: 'disposer', stack: undefined, children: undefined }, + ], + }, + ]); + }); + + it('captures the registration stack when enabled', () => { + Ledger.captureStacks = true; + try { + const ledger = new Ledger('test'); + ledger.register(() => {}, 'traced'); + expect(ledger.entries()[0]!.stack).toContain('Ledger registration'); + } finally { + Ledger.captureStacks = false; + } + }); + }); + + describe('idempotency', () => { + it('teardown is idempotent: disposers run exactly once', async () => { + const spy = vi.fn(); + const ledger = new Ledger('test'); + ledger.register(spy, 'spy'); + void ledger.teardown(); + void ledger.teardown(); + await ledger.teardown(); + expect(spy).toHaveBeenCalledTimes(1); + }); + + it('a concurrent teardown joins the in-flight one', async () => { + const gate = deferred(); + const spy = vi.fn(async () => { await gate.promise; }); + const ledger = new Ledger('test'); + ledger.register(spy, 'spy'); + const first = ledger.teardown(); + const second = ledger.teardown(); + expect(second).toBe(first); + gate.resolve(); + await first; + expect(spy).toHaveBeenCalledTimes(1); + }); + + it('entry.dispose is idempotent and removes the entry from the book', () => { + const spy = vi.fn(); + const ledger = new Ledger('test'); + const entry = ledger.register(spy, 'spy'); + void entry.dispose(); + void entry.dispose(); + expect(spy).toHaveBeenCalledTimes(1); + expect(entry.disposed).toBe(true); + expect(ledger.size).toBe(0); + }); + + it('entry.release removes the entry without running its disposer', () => { + const spy = vi.fn(); + const ledger = new Ledger('test'); + const entry = ledger.register(spy, 'spy'); + entry.release(); + entry.release(); + expect(entry.disposed).toBe(true); + void ledger.teardown(); + expect(spy).not.toHaveBeenCalled(); + }); + }); + + describe('reason propagation', () => { + it.each(['scope-close', 'cascade', 'unload'] as TeardownReason[])( + 'teardown(%s) reaches every disposer, including effect forms', + async (reason) => { + const seen: TeardownReason[] = []; + const ledger = new Ledger('test'); + ledger.register((r) => { seen.push(r); }, 'plain'); + ledger.effect(() => (r) => { seen.push(r); }, 'fx'); + ledger.effect(function* (): Generator { + yield (r) => { seen.push(r); }; + }, 'gen'); + const child = ledger.createChild('child'); + child.register((r) => { seen.push(r); }, 'child-plain'); + + await ledger.teardown(reason); + expect(seen).toEqual([reason, reason, reason, reason]); + }, + ); + + it('entry.dispose(reason) propagates the reason', () => { + const seen: TeardownReason[] = []; + const ledger = new Ledger('test'); + const entry = ledger.register((r) => { seen.push(r); }, 'plain'); + void entry.dispose('cascade'); + expect(seen).toEqual(['cascade']); + }); + }); + + describe('clear', () => { + it('tears down current entries but keeps the ledger active', () => { + const events: string[] = []; + const ledger = new Ledger('test'); + ledger.register(() => { events.push('a'); }, 'a'); + void ledger.clear(); + expect(events).toEqual(['a']); + expect(ledger.state).toBe('active'); + ledger.register(() => { events.push('b'); }, 'b'); + void ledger.teardown(); + expect(events).toEqual(['a', 'b']); + }); + }); +});