diff --git a/changelog.d/boot-state-read-errors.fixed.md b/changelog.d/boot-state-read-errors.fixed.md new file mode 100644 index 000000000..baaa8a9ee --- /dev/null +++ b/changelog.d/boot-state-read-errors.fixed.md @@ -0,0 +1,13 @@ +- Restored state is no longer taken as empty at boot. `AgentFramework.create()` now fails when a state it restores can't be read, or reads as something this version can't take for that state. The error names the state, or its recovery file, and says what to do; nothing it names is changed. Before, each was taken as empty, the framework served without it, and the next write replaced what was stored. Nothing stored still starts empty, and a store any release wrote boots as before unless one of these reads fails. + - **Host mode** (`recovery/host-mode.json`, or `framework/state` without one): a quiesced host booted serving, and landed the writes it had deferred during the maintenance. + - **Deferred writes** (`recovery/deferred-writes.json`, or `framework/deferred-writes` without one): the queue restored nothing, and the next deferral replaced it. An agent's context that can't be checked for writes that landed before their ack now fails `create()` too, where such a write landed a second time. + - **A recovery file that can't be reached** (an error other than its absence, such as `EACCES` on its directory) fails `create()` too, where it read as absent. + - **Usage totals** (`framework/state`): they started at zero, and the next inference stored zeros over them. A `framework/state` that isn't an object now fails `create()`, where every reader took it as empty. + - **An agent's persisted runtime settings** (`framework/state`): the agent booted at its recipe's settings, a `private` think-text policy as `public`, and its next settings change stored over the record. 0.22.2's notes said such a read is logged and the agent boots at the recipe's; it now fails `create()`. + - **Tool-result guards** (`framework/state`): a `toolResultGuards` that isn't an object read as no setting, and the next guard change stored over it. It's now read by each agent's own name, so an agent named `constructor` no longer fails `create()` once any guard is stored. + - **MCPL coalescing** (`mcpl/coalescing`, `mcpl/coalescing-recent`): an acknowledged deferred acceptance or a pending batch was never delivered, and the next receipt or flush replaced the stored ones. A snapshot of a version this one doesn't read was ignored and replaced; it now fails `create()` and is kept for the version that wrote it. An agent's context that can't be checked for an occurrence already delivered fails `create()` too, where the occurrence was delivered again. + - **MCPL checkpoint trees** (`mcpl/checkpoints`): a stored value without `trees` read as no checkpoints, and the next feature set registered stored over it. + - On a framework started with no MCPL servers, the coalescing state and the checkpoint trees are read at the first `connectMcplServer()`, which fails instead, with nothing of MCPL built, and the next connect reads them again. +- A host-managed MCPL feature set's state (`mcpl/state//`) is now read after a restart: when MCPL starts, for each configured server, and before `connectMcplServer()` adds a server. It wasn't, so the state read as empty, and the feature set's next delta checkpoint applied to nothing and stored over it. A failed read of it fails `create()`, or that `connectMcplServer()`, naming it, before the server connects. A feature set with no stored tree, such as one registering for the first time or after its server was disconnected, has its state read when it registers, as before. A failed read there fails that server's connect, as before, and `create()` logs that and goes on, as for any server's failed connect. The state is now read again the next time the feature set registers; before, its tree was held after the failed read, so the state was taken as empty. +- A usage count stored as `null`, which a provider's usage without a count left as `NaN`, restores as 0, and a usage without a count no longer makes the totals `NaN`. +- A failed read of the persisted `tool_result_inline_max_chars` caps at boot is now logged. As before, they don't apply until a restart, and setting one keeps the others stored. diff --git a/src/framework.ts b/src/framework.ts index 8351c2a97..60a6f9aa0 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -415,7 +415,7 @@ function validateUtilityArgs( import { CheckpointManager } from './mcpl/checkpoint-manager.js'; import { isToolAllowed } from './mcpl/tool-policy.js'; import { EventGate, formatShadowWarning } from './gate/event-gate.js'; -import { UsageTracker, type PersistedUsageState } from './usage/usage-tracker.js'; +import { UsageTracker, persistedUsageProblem, type PersistedUsageState } from './usage/usage-tracker.js'; import type { SessionUsageSnapshot, UsageUpdatedEvent } from './usage/types.js'; import { stderrTailOf, withStderr, type McplConnectionCloseInfo, type McplServerConnection } from './mcpl/server-connection.js'; import { assertServerConfigBooleans, isWebSocketTransport } from './mcpl/transport.js'; @@ -1646,6 +1646,20 @@ function thrownText(error: unknown): string { return text.length <= THROWN_TEXT_LIMIT ? text : `${text.slice(0, THROWN_TEXT_LIMIT)}… (${text.length} characters)`; } +/** A stored value that read but isn't in a form this code takes, as against + * a read that failed: starting again changes nothing until it's repaired, + * so the refusal that quotes it says so instead of "once it can be read". */ +class StoredFormError extends Error {} + +/** A stored JSON value's kind, for a refusal that names what it found + * without its content. */ +function jsonKind(value: unknown): string { + return value === null ? 'null' + : Array.isArray(value) ? 'an array' + : typeof value === 'object' ? 'an object' + : `a ${typeof value}`; +} + /** * A failed handler's reason, never throwing: an Error's message, else "the * handler failed" and what it threw (nothing, for `undefined`). Telling an @@ -4498,11 +4512,34 @@ export class AgentFramework { /** A new agent's guard: its persisted setting, the tool-results spill its * withheld originals are written to, and the set stop() waits on - * (agent-framework #277). */ + * (agent-framework #277). The setting is read from framework/state's + * toolResultGuards by the agent's own name. A read that fails, or a + * toolResultGuards that isn't an object, throws, naming the state: taken + * as no setting, the agent would take its recipe's guard, and the next + * guard change would store over a toolResultGuards that isn't an object. */ private adoptToolResultGuard(agent: Agent): void { - const enabled = (this.store.getStateJson(FRAMEWORK_STATE_ID) as { - toolResultGuards?: Record; - } | null)?.toolResultGuards?.[agent.name]; + const refuse = (problem: string, cause?: unknown): Error => new Error( + `the tool-result guards (${FRAMEWORK_STATE_ID}, toolResultGuards) ${problem}. Agent "${agent.name}" isn't ` + + `created, since without them it would take its recipe's guard` + + (cause === undefined + ? ` and the next guard change would store over them. They're unchanged, and creating it again won't help until the host's operator repairs them.` + : `. They're unchanged, and it's created once they can be read.`), + cause === undefined ? undefined : { cause }, + ); + let data: unknown; + try { + data = this.store.getStateJson(FRAMEWORK_STATE_ID); + } catch (error) { + throw refuse(`could not be read: ${thrownText(error)}`, error); + } + const guards = data && typeof data === 'object' ? (data as Record).toolResultGuards : undefined; + if (guards !== undefined && (!guards || typeof guards !== 'object' || Array.isArray(guards))) { + throw refuse(`are ${jsonKind(guards)}, not an object of agents' guards`); + } + // Own keys only: an agent named "constructor" has no inherited override. + const enabled = guards !== undefined && Object.prototype.hasOwnProperty.call(guards, agent.name) + ? (guards as Record)[agent.name] + : undefined; if (enabled !== undefined && typeof enabled !== 'boolean') { throw new Error(`Invalid persisted tool_result_guard for ${agent.name}: expected a boolean`); } @@ -5316,7 +5353,9 @@ export class AgentFramework { * Which of `ids` already exist in some agent's store as a landed deferred * write (`metadata.deferredWriteId`). Scans each context manager's recent * tail blob-free — an interrupted flush is always within the last few - * hundred slots — and tolerates facades without windowed reads. + * hundred slots — and tolerates facades without windowed reads. A context + * that can't be read throws, naming its agent: counted as holding none, a + * write that landed there before its ack would land a second time. */ private landedDeferredWriteIds(ids: Set, scanFrom: Map): Set { const landed = new Set(); @@ -5346,8 +5385,8 @@ export class AgentFramework { } else { for (const m of cm.getAllMessages()) consider(m); } - } catch (err) { - console.error(`[host-mode] could not scan ${agent.name} for landed deferred writes:`, err); + } catch (error) { + throw new Error(`${agent.name}'s context could not be read: ${thrownText(error)}`, { cause: error }); } } return landed; @@ -5362,9 +5401,24 @@ export class AgentFramework { renameSync(temporary, path); } + /** A recovery file's document, or undefined when there's no file. Only a + * missing file means none was written: any other failure to read it + * (EACCES, ENOTDIR, EIO, which existsSync would answer false for) + * throws, and text that isn't JSON throws a StoredFormError carrying + * the parser's reason, since reading it again won't change it. */ private readRecoveryFile(path: string): unknown { - if (!existsSync(path)) return undefined; - return JSON.parse(readFileSync(path, 'utf8')); + let text: string; + try { + text = readFileSync(path, 'utf8'); + } catch (error) { + if ((error as { code?: unknown } | null)?.code === 'ENOENT') return undefined; + throw error; + } + try { + return JSON.parse(text); + } catch (error) { + throw new StoredFormError(thrownText(error), { cause: error }); + } } private warnRecoveryFallbackOnce(): void { @@ -5433,137 +5487,181 @@ export class AgentFramework { } } + /** + * Restore the deferred-write queue a crash or a stop left durable. Nothing + * stored is an empty queue. A queue that can't be read, or that reads as + * anything but a queue, throws, since taken as empty the next deferral + * would replace the stored one; so does a context that can't be checked + * for writes that already landed (see landedDeferredWriteIds). + */ private restorePersistedDeferredWrites(): number { + // The recovery file wins whenever it exists. Stores that predate it (or + // store-only configs) keep the queue in the store's slot. + let file = this.deferredWritesPath; + const refuse = (problem: string, cause?: unknown): Error => new Error( + `the deferred context writes (${file ?? DEFERRED_WRITES_ID}) ${problem}. The framework doesn't start without them, ` + + `since the next deferral would replace the stored queue. ` + + (file + ? `The file is unchanged, and the framework starts once ${cause === undefined + ? 'it holds a queue of writes, such as {"version":1,"pending":[]} for none. Removing it reads the queue ' + + `from ${DEFERRED_WRITES_ID} instead, which the file's writes don't keep, so it restores whatever older ` + + `queue is there, or none` + : 'it can be read'}.` + : cause === undefined + ? `They're unchanged, and starting again won't help until they're repaired.` + : `They're unchanged, and the framework starts once they can be read.`), + cause === undefined ? undefined : { cause }, + ); + let raw: unknown; try { - let raw: unknown; - if (this.deferredWritesPath && existsSync(this.deferredWritesPath)) { - raw = this.readRecoveryFile(this.deferredWritesPath); - } else { - // Stores that predate the recovery file (or store-only configs). + raw = file ? this.readRecoveryFile(file) : undefined; + if (raw === undefined) { + file = undefined; raw = this.store.getStateJson(DEFERRED_WRITES_ID); } - // v1 slot payloads were a bare array; v1/v2 files and v2 slots are - // `{ pending, scanFrom? }`. - const doc = Array.isArray(raw) ? { pending: raw } : (raw as { pending?: unknown; scanFrom?: unknown } | undefined); - const data = doc?.pending; - if (!Array.isArray(data) || data.length === 0) return 0; - const scanFrom = new Map(); - if (doc?.scanFrom && typeof doc.scanFrom === 'object') { - for (const [k, v] of Object.entries(doc.scanFrom as Record)) { - if (typeof v === 'number' && Number.isFinite(v)) scanFrom.set(k, Math.max(0, Math.floor(v))); - } - } - // An entry the queue's readers can't handle has no place in it: one whose - // participant isn't text, whose content isn't an array, that holds a - // block that isn't an object, or whose `forAgent` is present but not a - // name, is dropped here with a line naming it, rather than held. Reading - // a null block's type throws, at the tool boundary and in the store. The - // store keeps a string or number block as given, but the context - // manager's token estimate keys a WeakMap on each block, which such a - // value can't be: once one is stored, every compile for that agent - // throws, and a restart doesn't clear it. What a block holds, such as a - // tool result's content, isn't checked here, as it isn't for a fresh - // write. The framework writes `forAgent` only as a non-empty name, so - // another value comes from damage or a hand edit, and the agent it - // meant can't be known: sent to the primary, as a falsy one would be, it - // could put another agent's message in the primary's window. - const wellFormed = (m: unknown): m is Omit & { id?: unknown; seq?: unknown } => { - if (!m || typeof m !== 'object') return false; - const { participant, content, forAgent } = m as { participant?: unknown; content?: unknown; forAgent?: unknown }; - return typeof participant === 'string' - && Array.isArray(content) && content.every((b) => !!b && typeof b === 'object') - && (forAgent === undefined || (typeof forAgent === 'string' && forAgent !== '')); + } catch (error) { + throw error instanceof StoredFormError + ? refuse(`hold text that isn't JSON: ${error.message}`) + : refuse(`could not be read: ${thrownText(error)}`, error); + } + // Nothing stored is an empty queue: no file, and nothing in the slot. A + // file that holds null isn't a queue (no release writes one). + if (!file && (raw === null || raw === undefined)) return 0; + // v1 slot payloads were a bare array; v1/v2 files and v2 slots are + // `{ pending, scanFrom? }`. + const doc = (Array.isArray(raw) ? { pending: raw } : raw !== null && typeof raw === 'object' ? raw : undefined) as + { pending?: unknown; scanFrom?: unknown } | undefined; + if (!doc || !Array.isArray(doc.pending)) { + throw refuse(doc + ? `hold an object whose pending is ${doc.pending === undefined ? 'missing' : jsonKind(doc.pending)}, not a list of writes` + : `hold ${jsonKind(raw)}, not a queue of writes`); + } + const data = doc.pending; + if (data.length === 0) return 0; + const scanFrom = new Map(); + if (doc.scanFrom && typeof doc.scanFrom === 'object') { + for (const [k, v] of Object.entries(doc.scanFrom as Record)) { + if (typeof v === 'number' && Number.isFinite(v)) scanFrom.set(k, Math.max(0, Math.floor(v))); + } + } + // An entry the queue's readers can't handle has no place in it: one whose + // participant isn't text, whose content isn't an array, that holds a + // block that isn't an object, or whose `forAgent` is present but not a + // name, is dropped here with a line naming it, rather than held. Reading + // a null block's type throws, at the tool boundary and in the store. The + // store keeps a string or number block as given, but the context + // manager's token estimate keys a WeakMap on each block, which such a + // value can't be: once one is stored, every compile for that agent + // throws, and a restart doesn't clear it. What a block holds, such as a + // tool result's content, isn't checked here, as it isn't for a fresh + // write. The framework writes `forAgent` only as a non-empty name, so + // another value comes from damage or a hand edit, and the agent it + // meant can't be known: sent to the primary, as a falsy one would be, it + // could put another agent's message in the primary's window. + const wellFormed = (m: unknown): m is Omit & { id?: unknown; seq?: unknown } => { + if (!m || typeof m !== 'object') return false; + const { participant, content, forAgent } = m as { participant?: unknown; content?: unknown; forAgent?: unknown }; + return typeof participant === 'string' + && Array.isArray(content) && content.every((b) => !!b && typeof b === 'object') + && (forAgent === undefined || (typeof forAgent === 'string' && forAgent !== '')); + }; + const malformed = data.filter((m) => !wellFormed(m)); + for (const m of malformed) { + // What the entry holds there: a value that isn't text by its type. + const field = (key: string): string => { + const value = m && typeof m === 'object' ? (m as Record)[key] : undefined; + if (typeof value === 'string') return value; + if (value === undefined) return `no ${key}`; + return `${key} ${value === null ? 'null' : `of type ${typeof value}`}`; }; - const malformed = data.filter((m) => !wellFormed(m)); - for (const m of malformed) { - // What the entry holds there: a value that isn't text by its type. - const field = (key: string): string => { - const value = m && typeof m === 'object' ? (m as Record)[key] : undefined; - if (typeof value === 'string') return value; - if (value === undefined) return `no ${key}`; - return `${key} ${value === null ? 'null' : `of type ${typeof value}`}`; - }; + console.error( + `[host-mode] a recovered deferred context write is malformed and was dropped: ` + + `${field('id')} (${field('participant')})`, + ); + } + // One entry per id: the boot dedup and the ack each take an id as one + // write, so an id that comes again is dropped with a line, and the first + // entry with it is kept. An entry with no id is given a fresh one, with + // a line: a copy of it that already landed carries the id it lost, so + // the boot dedup can't find it, and it may be stored again. + const seen = new Set(); + const kept: Array & { id: string; seq?: unknown }> = []; + let duplicates = 0; + for (const m of data.filter(wellFormed)) { + const hasId = typeof m.id === 'string' && m.id !== ''; + const id = hasId ? (m.id as string) : randomUUID(); + if (!hasId) { + console.error(`[host-mode] a recovered deferred context write (${m.participant}) has no id; it is given ${id}`); + } + if (seen.has(id)) { + duplicates++; console.error( - `[host-mode] a recovered deferred context write is malformed and was dropped: ` + - `${field('id')} (${field('participant')})`, + `[host-mode] a recovered deferred context write repeats id ${id} (${m.participant}) and was dropped; ` + + `the first entry with that id is kept`, ); + continue; } - // One entry per id: the boot dedup and the ack each take an id as one - // write, so an id that comes again is dropped with a line, and the first - // entry with it is kept. An entry with no id is given a fresh one, with - // a line: a copy of it that already landed carries the id it lost, so - // the boot dedup can't find it, and it may be stored again. - const seen = new Set(); - const kept: Array & { id: string; seq?: unknown }> = []; - let duplicates = 0; - for (const m of data.filter(wellFormed)) { - const hasId = typeof m.id === 'string' && m.id !== ''; - const id = hasId ? (m.id as string) : randomUUID(); - if (!hasId) { - console.error(`[host-mode] a recovered deferred context write (${m.participant}) has no id; it is given ${id}`); - } - if (seen.has(id)) { - duplicates++; + seen.add(id); + kept.push({ ...m, id }); + } + // A seq is a positive whole number no larger than + // Number.MAX_SAFE_INTEGER, the entry's place in arrival order. + // An entry without one (a file that predates seq, which is in write + // order already) or with any other value (damage or a hand edit: + // JSON's 1e400 reads as Infinity) is numbered after every valid seq, in + // file order, so the counter stays finite and the rest keep their order. + const validSeq = (seq: unknown): seq is number => Number.isSafeInteger(seq) && (seq as number) > 0; + let lastSeq = 0; + for (const m of kept) if (validSeq(m.seq)) lastSeq = Math.max(lastSeq, m.seq); + let renumbered = 0; + const candidates: DeferredWrite[] = kept + .map((m) => { + if (validSeq(m.seq)) return { ...m, seq: m.seq }; + renumbered++; + if (m.seq !== undefined) { + // A value that isn't a number is named by its type: it could be + // any size. + const seq = typeof m.seq === 'number' ? String(m.seq) : m.seq === null ? 'null' : `of type ${typeof m.seq}`; console.error( - `[host-mode] a recovered deferred context write repeats id ${id} (${m.participant}) and was dropped; ` + - `the first entry with that id is kept`, + `[host-mode] a recovered deferred context write (${m.id}) has seq ${seq}, not a positive ` + + `whole number up to Number.MAX_SAFE_INTEGER; it is queued after the others`, ); - continue; } - seen.add(id); - kept.push({ ...m, id }); - } - // A seq is a positive whole number no larger than - // Number.MAX_SAFE_INTEGER, the entry's place in arrival order. - // An entry without one (a file that predates seq, which is in write - // order already) or with any other value (damage or a hand edit: - // JSON's 1e400 reads as Infinity) is numbered after every valid seq, in - // file order, so the counter stays finite and the rest keep their order. - const validSeq = (seq: unknown): seq is number => Number.isSafeInteger(seq) && (seq as number) > 0; - let lastSeq = 0; - for (const m of kept) if (validSeq(m.seq)) lastSeq = Math.max(lastSeq, m.seq); - let renumbered = 0; - const candidates: DeferredWrite[] = kept - .map((m) => { - if (validSeq(m.seq)) return { ...m, seq: m.seq }; - renumbered++; - if (m.seq !== undefined) { - // A value that isn't a number is named by its type: it could be - // any size. - const seq = typeof m.seq === 'number' ? String(m.seq) : m.seq === null ? 'null' : `of type ${typeof m.seq}`; - console.error( - `[host-mode] a recovered deferred context write (${m.id}) has seq ${seq}, not a positive ` + - `whole number up to Number.MAX_SAFE_INTEGER; it is queued after the others`, - ); - } - return { ...m, seq: ++lastSeq }; - }) - .sort(bySeq); - this.deferredSeq = Math.max(this.deferredSeq, lastSeq); - // Exactly-once: a message that landed before its ack was written is - // already in the store under its deferredWriteId — do not replay it. - // The scan covers every slot the interrupted batch could occupy (from - // the receipt's per-agent position), or the whole store when the - // receipt predates the position field. - const landed = this.landedDeferredWriteIds(new Set(candidates.map((m) => m.id)), scanFrom); - const restored = candidates.filter((m) => !landed.has(m.id)); - if (landed.size > 0) { - console.error( - `[host-mode] ${landed.size} deferred context write(s) had already landed before their ack ` + - `(interrupted flush) — skipped, not replayed`, - ); - } - this.deferredMessages.push(...restored); - this.deferredWritesPersisted = true; - // Rewrite the durable queue without the already-landed entries so a - // second interrupted boot does not re-scan them, without the malformed - // and repeated ones, and with the seqs given here. - if (landed.size > 0 || malformed.length > 0 || duplicates > 0 || renumbered > 0) this.persistDeferredWrites(); - return restored.length; - } catch (err) { - console.error('[host-mode] failed to restore persisted deferred context writes:', err); - return 0; + return { ...m, seq: ++lastSeq }; + }) + .sort(bySeq); + this.deferredSeq = Math.max(this.deferredSeq, lastSeq); + // Exactly-once: a message that landed before its ack was written is + // already in the store under its deferredWriteId — do not replay it. + // The scan covers every slot the interrupted batch could occupy (from + // the receipt's per-agent position), or the whole store when the + // receipt predates the position field. + let landed: Set; + try { + landed = this.landedDeferredWriteIds(new Set(candidates.map((m) => m.id)), scanFrom); + } catch (error) { + throw new Error( + `the deferred context writes (${file ?? DEFERRED_WRITES_ID}) can't be checked for writes that already landed: ` + + `${thrownText(error)}. The framework doesn't start without that check, since a write that landed ` + + `before its ack would land a second time. The queue is unchanged, and the framework starts once ` + + `that context can be read.`, + { cause: error }, + ); + } + const restored = candidates.filter((m) => !landed.has(m.id)); + if (landed.size > 0) { + console.error( + `[host-mode] ${landed.size} deferred context write(s) had already landed before their ack ` + + `(interrupted flush) — skipped, not replayed`, + ); } + this.deferredMessages.push(...restored); + this.deferredWritesPersisted = true; + // Rewrite the durable queue without the already-landed entries so a + // second interrupted boot does not re-scan them, without the malformed + // and repeated ones, and with the seqs given here. + if (landed.size > 0 || malformed.length > 0 || duplicates > 0 || renumbered > 0) this.persistDeferredWrites(); + return restored.length; } /** @@ -5962,19 +6060,47 @@ export class AgentFramework { return this.usageTracker.getSnapshot(); } + /** Restore the usage totals a prior session stored. None stored starts + * fresh (the constructor's tracker). Totals that can't be read, or that + * read as anything but totals, throw: taken as none, the next inference's + * persistUsageState would store fresh totals over them. + * + * This is create()'s first read of framework/state, so it also refuses a + * framework/state that isn't an object: every reader of that state would + * take it as holding nothing, and its first write would replace it. */ private restoreUsageState(): void { + const refuse = (problem: string, cause?: unknown): Error => new Error( + `the usage totals (${FRAMEWORK_STATE_ID}, usage) ${problem}. The framework doesn't start without them, ` + + `since the next inference would store fresh totals over them. ` + + (cause === undefined + ? `They're unchanged, and starting again won't help until they're repaired.` + : `They're unchanged, and the framework starts once they can be read.`), + cause === undefined ? undefined : { cause }, + ); + let data: unknown; try { - const data = this.store.getStateJson(FRAMEWORK_STATE_ID); - if (data && typeof data === 'object' && (data as any).usage) { - const restored = (data as any).usage as PersistedUsageState; - this.usageTracker = new UsageTracker({ - emitTrace: (e: UsageUpdatedEvent) => this.emitTrace({ ...e }), - restored, - }); - } - } catch { - // No prior state or corrupt — start fresh (already initialized in constructor) + data = this.store.getStateJson(FRAMEWORK_STATE_ID); + } catch (error) { + throw refuse(`could not be read: ${thrownText(error)}`, error); + } + if (data === null || data === undefined) return; + if (typeof data !== 'object' || Array.isArray(data)) { + throw new Error( + `${FRAMEWORK_STATE_ID} holds ${jsonKind(data)}, not an object. The framework doesn't start on it, since its ` + + `first write would replace it and everything it keeps, such as the usage totals and each agent's settings. ` + + `It's unchanged, and starting again won't help until it's repaired.`, + ); } + const usage = (data as Record).usage; + if (usage === undefined) return; + const problem = persistedUsageProblem(usage); + if (problem === 'not a record') throw refuse(`are ${jsonKind(usage)}, not a usage record`); + if (problem === 'totals') throw refuse(`are a record whose totals aren't token counts`); + if (problem === 'byAgent') throw refuse(`are a record whose byAgent isn't a list`); + this.usageTracker = new UsageTracker({ + emitTrace: (e: UsageUpdatedEvent) => this.emitTrace({ ...e }), + restored: usage as PersistedUsageState, + }); } private persistUsageState(): void { @@ -5988,13 +6114,25 @@ export class AgentFramework { } } - /** An agent's persisted record, or null when it has none; throws when the store can't be read. */ + /** An agent's persisted record, or null when it has none. Throws the + * store's error when it can't be read, and a StoredFormError when what's + * stored for the agent, or the map that holds every agent's, isn't a record. */ private readAgentRuntimeSettings(agentName: string): AgentRuntimeSettingsOverrides | null { - const data = this.store.getStateJson(FRAMEWORK_STATE_ID) as { - agentRuntimeSettings?: Record; - } | null; - const stored = data?.agentRuntimeSettings?.[agentName]; - return stored ? { ...stored } : null; + const data = this.store.getStateJson(FRAMEWORK_STATE_ID); + const map = data && typeof data === 'object' ? (data as Record).agentRuntimeSettings : undefined; + if (map === undefined) return null; + if (!map || typeof map !== 'object' || Array.isArray(map)) { + throw new StoredFormError(`the map that holds every agent's settings is ${jsonKind(map)}, not an object`); + } + // Own keys only: an agent named "constructor" has no inherited record. + const stored = Object.prototype.hasOwnProperty.call(map, agentName) + ? (map as Record)[agentName] + : undefined; + if (stored === undefined) return null; + if (!stored || typeof stored !== 'object' || Array.isArray(stored)) { + throw new StoredFormError(`what's stored for the agent is ${jsonKind(stored)}, not a settings record`); + } + return { ...(stored as AgentRuntimeSettingsOverrides) }; } private validatePersistedAgentRuntimeSettings( @@ -6036,27 +6174,38 @@ export class AgentFramework { } /** Load resident-set inline caps from framework state, dropping (loudly) - * anything that fails the same validation the live update enforces. */ + * anything that fails the same validation the live update enforces. A + * failed read is logged and boot goes on without them: nothing is lost, + * since persistToolResultInlineCap reads the stored caps again before it + * writes one. */ private readPersistedToolResultInlineCaps(): Record { + let data: { toolResultInlineCaps?: Record } | null; try { - const data = this.store.getStateJson(FRAMEWORK_STATE_ID) as { - toolResultInlineCaps?: Record; - } | null; - const out: Record = {}; - for (const [name, value] of Object.entries(data?.toolResultInlineCaps ?? {})) { - const n = Number(value); - if (Number.isFinite(n) && n >= 1000) { - out[name] = Math.floor(n); - } else { - console.error( - `[agent-settings] dropping invalid persisted tool_result_inline_max_chars for '${name}': ${JSON.stringify(value)}`, - ); - } - } - return out; - } catch { + data = this.store.getStateJson(FRAMEWORK_STATE_ID) as typeof data; + } catch (error) { + console.error( + `[agent-settings] the persisted tool_result_inline_max_chars caps (${FRAMEWORK_STATE_ID}) could not be read: ` + + `${thrownText(error)}; they don't apply until a restart, and setting one keeps the others stored`, + ); return {}; } + const out: Record = {}; + for (const [name, value] of Object.entries(data?.toolResultInlineCaps ?? {})) { + // Number() throws for an object it can't make a primitive of, such as + // {"toString":null}: that entry is invalid like any other, and dropped. + let n = Number.NaN; + try { + n = Number(value); + } catch { /* not a number: dropped below */ } + if (Number.isFinite(n) && n >= 1000) { + out[name] = Math.floor(n); + } else { + console.error( + `[agent-settings] dropping invalid persisted tool_result_inline_max_chars for '${name}': ${JSON.stringify(value)}`, + ); + } + } + return out; } /** Write (cap) or clear (null) one agent's resident-set inline cap. */ @@ -6274,29 +6423,52 @@ export class AgentFramework { } /** The recovery file wins whenever it exists; the branch-local slot is - * read only for stores that predate it (or store-only configs). */ + * read only for stores that predate it (or store-only configs). No record + * is a host that was never quiesced. A record that can't be read, or that + * reads as anything but a host-mode record, throws: taken as "not + * quiesced", a host that stopped mid-maintenance would boot serving. */ private readHostMode(): { quiesced: boolean; reason?: string; since?: number } | null { + let file = this.hostModePath; + const refuse = (problem: string, cause?: unknown): Error => new Error( + `the host mode (${file ?? `${FRAMEWORK_STATE_ID}, hostMode`}) ${problem}. The framework doesn't start without it, ` + + `since a host quiesced for maintenance would boot serving. ` + + (file + ? `The file is unchanged, and the framework starts once ${cause === undefined + ? 'it holds a host-mode record, such as {"version":1,"quiesced":true} to boot quiesced or ' + + `{"version":1,"quiesced":false} to serve. Removing it doesn't help: without the file, host mode is read ` + + `from ${FRAMEWORK_STATE_ID}, which a quiesce doesn't write, so a quiesced host would boot serving` + : 'it can be read'}.` + : cause === undefined + ? `It's unchanged, and starting again won't help until it's repaired.` + : `It's unchanged, and the framework starts once it can be read.`), + cause === undefined ? undefined : { cause }, + ); + let record: unknown; try { - let record: Record | null = null; - if (this.hostModePath && existsSync(this.hostModePath)) { - const doc = this.readRecoveryFile(this.hostModePath); - if (doc && typeof doc === 'object') record = doc as Record; - } else { + record = file ? this.readRecoveryFile(file) : undefined; + if (record === undefined) { + file = undefined; const data = this.store.getStateJson(FRAMEWORK_STATE_ID); - if (!data || typeof data !== 'object') return null; - const mode = (data as Record).hostMode; - if (!mode || typeof mode !== 'object') return null; - record = mode as Record; + record = data && typeof data === 'object' ? (data as Record).hostMode : undefined; } - if (!record) return null; - return { - quiesced: record.quiesced === true, - ...(typeof record.reason === 'string' ? { reason: record.reason } : {}), - ...(typeof record.since === 'number' ? { since: record.since } : {}), - }; - } catch { - return null; + } catch (error) { + throw error instanceof StoredFormError + ? refuse(`holds text that isn't JSON: ${error.message}`) + : refuse(`could not be read: ${thrownText(error)}`, error); + } + if (record === undefined) return null; + const isRecord = !!record && typeof record === 'object' && !Array.isArray(record); + const { quiesced, reason, since } = (isRecord ? record : {}) as Record; + if (typeof quiesced !== 'boolean') { + throw refuse(isRecord + ? `is a record whose quiesced is ${quiesced === undefined ? 'missing' : jsonKind(quiesced)}, not true or false` + : `holds ${jsonKind(record)}, not a host-mode record`); } + return { + quiesced, + ...(typeof reason === 'string' ? { reason } : {}), + ...(typeof since === 'number' ? { since } : {}), + }; } /** @@ -9095,13 +9267,21 @@ export class AgentFramework { this.adoptToolResultGuard(agent); agent.markContextConsumed(); // RFC-006: nothing stored before now is replaceable this.sharedSlotAgents.add(agent); - let restoredSettings: AgentRuntimeSettingsOverrides | null = null; + let restoredSettings: AgentRuntimeSettingsOverrides | null; try { restoredSettings = this.readAgentRuntimeSettings(config.name); } catch (error) { - console.error( - `[agent-settings] ${config.name}: its persisted settings can't be read ` + - `(${error instanceof Error ? error.message : String(error)}); it boots at the recipe's`, + // Booted at the recipe's, the agent would serve settings it didn't + // choose (a private think-text policy as public), and its next + // persisting change would store over the record this read skipped. + throw new Error( + `the persisted settings of agent "${config.name}" (${FRAMEWORK_STATE_ID}, agentRuntimeSettings) can't be ` + + `restored: ${thrownText(error)}. The framework doesn't start without them, since the agent would boot at ` + + `its recipe's settings and the next settings change would store over them. ` + + (error instanceof StoredFormError + ? `They're unchanged, and starting again won't help until they're repaired.` + : `They're unchanged, and the framework starts once they can be read.`), + { cause: error }, ); } if (restoredSettings) { @@ -10454,7 +10634,60 @@ export class AgentFramework { return createHash('sha256').update(JSON.stringify([serverId, config?.url ?? null, config?.command ?? null, config?.args ?? null])).digest('hex').slice(0, 16); } - private initializePushCoalescer(): void { + /** + * The push coalescer, built and restored from its stored receipts and + * batches. initializeMcpl calls this before it builds anything else, so a + * refusal here leaves the MCPL subsystem as it was. Nothing stored is a + * coalescer with no history. What can't be read, reads as anything but + * this version's snapshot and receipts, or can't be restored, throws, and + * so does an agent's context that can't be checked for an occurrence + * already delivered (wasPublished): taken as no history, the next + * acceptance's receipt and the next flush would replace what's stored, and + * taken as undelivered, an occurrence would be delivered again. + */ + private restorePushCoalescer(): { coalescer: PushCoalescer; recent?: CoalescingReceiptRecord[] } { + const what = (stateId: string): string => + `the coalescing ${stateId === COALESCING_STATE_ID ? 'receipts and batches' : 'receipts since the last snapshot'} (${stateId})`; + const stops = `MCPL doesn't start without them, and neither does create() with MCPL servers configured, nor a ` + + `first connectMcplServer(), since the next coalesced push or message would replace them.`; + const unchanged = `They're unchanged, and starting again won't help until they're repaired.`; + const refuse = (stateId: string, problem: string, remedy: string, cause?: unknown): Error => + new Error(`${what(stateId)} ${problem}. ${stops} ${remedy}`, cause === undefined ? undefined : { cause }); + const read = (stateId: string): unknown => { + try { + return this.store.getStateJson(stateId); + } catch (error) { + throw refuse(stateId, `could not be read: ${thrownText(error)}`, `They're unchanged, and MCPL starts once they can be read.`, error); + } + }; + const snapshot = read(COALESCING_STATE_ID); + const recent = read(COALESCING_RECENT_ID); + if (snapshot !== null && snapshot !== undefined) { + const version = typeof snapshot === 'object' && !Array.isArray(snapshot) + ? (snapshot as { version?: unknown }).version + : undefined; + if (version !== 1) { + throw typeof version === 'number' + ? refuse( + COALESCING_STATE_ID, + `are a version ${version} snapshot, which this version of agent-framework doesn't read`, + `They're unchanged, and the version that wrote them reads them.`, + ) + : refuse( + COALESCING_STATE_ID, + version === undefined + ? `are ${jsonKind(snapshot)} with no snapshot version` + : `are a snapshot whose version is ${jsonKind(version)}, not 1`, + unchanged, + ); + } + } + if (recent !== null && recent !== undefined && !Array.isArray(recent)) { + throw refuse(COALESCING_RECENT_ID, `are ${jsonKind(recent)}, not a list of receipts`, unchanged); + } + // Set when wasPublished can't read an agent's context, so the restore that + // asked refuses with what couldn't be checked rather than as malformed. + let unreadable: { agent: string; error: unknown } | undefined; const coalescer = new PushCoalescer({ isUnread: (p) => this.isUnreadPlacement(p), remove: (p) => this.removePlacement(p), @@ -10492,13 +10725,18 @@ export class AgentFramework { this.store.setStateJson(COALESCING_RECENT_ID, this.coalescingRecentReceipts); }, wasPublished: (subject, eventId) => { - // Boot-time only: the occurrence's durable delivery identity - // (subject + eventId in message metadata) in any agent's context. + // Restore-time only: the occurrence's durable delivery identity + // (subject + eventId in message metadata) in any agent's context. A + // context that can't be read can't say, so the restore refuses. for (const agent of this.agents.values()) { + let messages: ReturnType; try { - if (agent.getContextManager().getAllMessages().some((m) => - m.metadata?.coalescingSubject === subject && m.metadata?.eventId === eventId)) return true; - } catch { /* a context that cannot be read cannot prove publication */ } + messages = agent.getContextManager().getAllMessages(); + } catch (error) { + unreadable = { agent: agent.name, error }; + throw error; + } + if (messages.some((m) => m.metadata?.coalescingSubject === subject && m.metadata?.eventId === eventId)) return true; } return false; }, @@ -10511,22 +10749,38 @@ export class AgentFramework { this.store.setStateJson(COALESCING_RECENT_ID, []); }, }); - try { - coalescer.restore(this.store.getStateJson(COALESCING_STATE_ID) as CoalescingSnapshot | null); - const recent = this.store.getStateJson(COALESCING_RECENT_ID) as CoalescingReceiptRecord[] | null; - if (Array.isArray(recent)) { - coalescer.restoreReceipts(recent); - // Until the next snapshot flush checkpoints the recovered state, these - // records are its only durable form: keep them in the writable buffer - // so the next acceptance appends to them instead of replacing them. - this.coalescingRecentReceipts = recent; + const restore = (stateId: string, run: () => void): void => { + try { + run(); + } catch (error) { + if (unreadable) { + throw new Error( + `${what(stateId)} can't be checked for occurrences already delivered: ${unreadable.agent}'s context ` + + `could not be read: ${thrownText(unreadable.error)}. MCPL doesn't start without that check, and neither ` + + `does create() with MCPL servers configured, nor a first connectMcplServer(), since an occurrence ` + + `delivered before the restart would be delivered again. They're unchanged, and MCPL starts once that ` + + `context can be read.`, + { cause: unreadable.error }, + ); + } + throw refuse(stateId, `could not be restored: ${thrownText(error)}`, unchanged, error); } - } catch (err) { - console.error('[coalescing] could not restore coalescing state; starting with unknown history:', err); - } - this.pushCoalescer = coalescer; + }; + restore(COALESCING_STATE_ID, () => coalescer.restore(snapshot as CoalescingSnapshot | null | undefined)); + if (!Array.isArray(recent)) return { coalescer }; + restore(COALESCING_RECENT_ID, () => coalescer.restoreReceipts(recent as CoalescingReceiptRecord[])); + return { coalescer, recent: recent as CoalescingReceiptRecord[] }; + } + + /** Install a coalescer restorePushCoalescer built: none of this can fail. */ + private installPushCoalescer(restored: { coalescer: PushCoalescer; recent?: CoalescingReceiptRecord[] }): void { + // Until the next snapshot flush checkpoints the recovered state, these + // records are its only durable form: keep them in the writable buffer + // so the next acceptance appends to them instead of replacing them. + if (restored.recent) this.coalescingRecentReceipts = restored.recent; + this.pushCoalescer = restored.coalescer; // Batches restored from a snapshot still owe the agent a wake (§14 v41). - for (const occ of coalescer.pendingBatchOccurrences()) void this.wakeForCoalescedBatch(occ); + for (const occ of restored.coalescer.pendingBatchOccurrences()) void this.wakeForCoalescedBatch(occ); } /** @@ -17099,6 +17353,19 @@ export class AgentFramework { serverConfigs: McplServerConfig[], inferenceRouting?: import('./mcpl/types.js').InferenceRoutingPolicy, ): Promise { + // First, before anything is built: a coalescing state or checkpoint trees + // that can't be taken refuse here and leave the subsystem as it was, so a + // first connectMcplServer() refused for them leaves nothing for the guards + // that test mcplServerRegistry. Each configured server's restored + // host-managed state is read here too, so a boot refused for one connects + // no server; a runtime connect reads its server's in + // runConnectMcplServer, once MCPL is up. + const coalescing = this.restorePushCoalescer(); + const checkpoints = new CheckpointManager( + this.store, + (event) => this.emitTrace(event as { type: TraceEvent['type']; [key: string]: unknown }), + ); + for (const config of serverConfigs) checkpoints.loadHostStates(config.id); this.mcplServerRegistry = new McplServerRegistry(); this.featureSetManager = new FeatureSetManager( (event) => this.emitTrace(event as { type: TraceEvent['type']; [key: string]: unknown }), @@ -17123,7 +17390,7 @@ export class AgentFramework { ?? (this.eventGate ? this.eventGate.asShouldTriggerCallback() : undefined); // Push events handler (Step 6) - this.initializePushCoalescer(); + this.installPushCoalescer(coalescing); this.pushHandler = new PushHandler( this.featureSetManager, (event) => this.pushEvent(event as unknown as ProcessEvent), @@ -17145,11 +17412,8 @@ export class AgentFramework { }, ); - // Checkpoint manager (Step 8) - this.checkpointManager = new CheckpointManager( - this.store, - (event) => this.emitTrace(event as { type: TraceEvent['type']; [key: string]: unknown }), - ); + // Checkpoint manager (Step 8), built and loaded at the top + this.checkpointManager = checkpoints; // Channel registry (Step 7) this.channelRegistry = new ChannelRegistry( @@ -18273,6 +18537,13 @@ export class AgentFramework { ); } + // The server's restored host-managed trees are read before the server is + // added, so a read that fails leaves it unconnected; MCPL itself stays + // up, even when this connect brought it up. A feature set with no + // restored tree, as after a disconnect removed it, is read when it + // registers. A server already connected is refused by addServer below, + // so its trees are left as they are. + if (!this.mcplServerRegistry?.getServer(config.id)) this.checkpointManager?.loadHostStates(config.id); const oldToolNames = new Set(this.mcplTools.map(t => t.name)); await this.connectMcplServerInternal(config); await this.refreshMcplTools(); diff --git a/src/mcpl/checkpoint-manager.ts b/src/mcpl/checkpoint-manager.ts index 4c7a26b42..111063bf4 100644 --- a/src/mcpl/checkpoint-manager.ts +++ b/src/mcpl/checkpoint-manager.ts @@ -26,6 +26,34 @@ const CHECKPOINTS_STATE_ID = 'mcpl/checkpoints'; /** Prefix for per-feature-set host-managed state. */ const STATE_PREFIX = 'mcpl/state'; +/** What the refusals below say about MCPL when stored trees or state can't be taken. */ +const MCPL_STOPS = `MCPL doesn't start without them, and neither does create() with MCPL servers configured, ` + + `nor a first connectMcplServer()`; + +/** A thrown value's text, bounded and never throwing, as framework.ts's thrownText. */ +function thrownText(error: unknown): string { + let text: string; + try { + text = String(error instanceof Error ? error.message : error); + } catch { + text = '(a value that could not be read)'; + } + return text.length <= 500 ? text : `${text.slice(0, 500)}… (${text.length} characters)`; +} + +/** A stored JSON value's kind, for a refusal that names what it found without its content. */ +function storedKind(value: unknown): string { + return value === undefined ? 'missing' + : value === null ? 'null' + : Array.isArray(value) ? 'an array' + : typeof value === 'object' ? 'an object' + : `a ${typeof value}`; +} + +function isObject(value: unknown): value is Record { + return !!value && typeof value === 'object' && !Array.isArray(value); +} + // ============================================================================ // Internal types // ============================================================================ @@ -214,31 +242,21 @@ export class CheckpointManager { opts: { hostState: boolean; rollback: boolean }, ): void { const key = this.key(serverId, featureSet); - if (this.trees.has(key)) return; // Already registered (e.g. from loaded state) + // Already registered, or restored from the store by loadFromStore, whose + // host-managed state loadHostStates reads before its server connects. + if (this.trees.has(key)) return; - this.trees.set(key, { + const tree: FeatureSetState = { hostState: opts.hostState, rollback: opts.rollback, currentCheckpoint: null, currentState: undefined, nodes: new Map(), - }); - - // Register per-feature-set state slot for host-managed state - if (opts.hostState) { - const stateId = this.stateId(serverId, featureSet); - try { - this.store.registerState({ id: stateId, strategy: 'snapshot' }); - } catch { - // Already registered - } - // Load existing state if any - const existing = this.store.getStateJson(stateId); - if (existing !== null && existing !== undefined) { - const tree = this.trees.get(key)!; - tree.currentState = existing; - } - } + }; + // Read before the tree is held, so a registration that fails here is + // made whole by the next one. + if (opts.hostState) this.loadHostState(serverId, featureSet, tree, 'registration'); + this.trees.set(key, tree); this.emitTraceFn({ type: 'mcpl:state_registered', @@ -540,12 +558,84 @@ export class CheckpointManager { this.store.setStateJson(CHECKPOINTS_STATE_ID, serialized); } - /** Load checkpoint tree from Chronicle on startup. */ + /** A host-managed feature set's stored state, read into its tree; nothing + * stored leaves it unset. A read that fails throws, naming the state: + * taken as nothing stored, the next patch would apply to nothing and + * store over it. `when` says what the refusal stops: a restored tree's + * read comes before its server connects, and a feature set with no tree + * is read when it registers, once its server has connected. */ + private loadHostState( + serverId: string, + featureSet: string, + tree: FeatureSetState, + when: 'before-connect' | 'registration', + ): void { + const stateId = this.stateId(serverId, featureSet); + try { + this.store.registerState({ id: stateId, strategy: 'snapshot' }); + } catch { + // Already registered + } + let existing: unknown; + try { + existing = this.store.getStateJson(stateId); + } catch (error) { + throw new Error( + `the host-managed state of feature set "${featureSet}" on MCPL server "${serverId}" (${stateId}) could not ` + + `be read: ${thrownText(error)}. ` + + (when === 'before-connect' + ? `Neither create() with the server configured nor a connectMcplServer() of it goes on without it, since ` + + `the feature set's next checkpoint would store over it. It's unchanged, and they go on once it can be read.` + : `The feature set isn't registered without it, since its next checkpoint would store over it, so the ` + + `server's connect fails. It's unchanged, and it's read again when the feature set next registers.`), + { cause: error }, + ); + } + if (existing !== null && existing !== undefined) tree.currentState = existing; + } + + /** Read the stored state of every host-managed tree held for `serverId` + * with none in memory, whatever its feature sets declare now. The + * framework calls it at MCPL's start for each configured server and + * before a runtime connect adds its server, so no server connects with a + * restored tree whose state its next checkpoint would store over unread. + * A tree whose state is in memory, as a connected server's is, keeps it. + * A server's trees are the keys `${serverId}:`, as removeServer finds + * them. */ + loadHostStates(serverId: string): void { + const prefix = `${serverId}:`; + for (const [key, tree] of this.trees) { + if (!tree.hostState || tree.currentState !== undefined || !key.startsWith(prefix)) continue; + this.loadHostState(serverId, key.slice(prefix.length), tree, 'before-connect'); + } + } + + /** Load checkpoint trees from Chronicle on startup. Nothing stored is no + * trees. Trees that can't be read, or a stored value that isn't an object + * whose trees is an object, throw: taken as no trees, the next feature set + * registered would store over them. */ private loadFromStore(): void { - const data = this.store.getStateJson(CHECKPOINTS_STATE_ID) as SerializedTrees | null; - if (!data?.trees) return; + const what = `the MCPL checkpoint trees (${CHECKPOINTS_STATE_ID})`; + const stops = `${MCPL_STOPS}, since the next feature set registered would store over them.`; + let data: unknown; + try { + data = this.store.getStateJson(CHECKPOINTS_STATE_ID); + } catch (error) { + throw new Error( + `${what} could not be read: ${thrownText(error)}. ${stops} They're unchanged, and MCPL starts once they can be read.`, + { cause: error }, + ); + } + if (data === null || data === undefined) return; + const trees = isObject(data) ? data.trees : undefined; + if (!isObject(trees)) { + throw new Error( + `${what} are ${isObject(data) ? `an object whose trees is ${storedKind(trees)}` : storedKind(data)}, not ` + + `checkpoint trees. ${stops} They're unchanged, and starting again won't help until they're repaired.`, + ); + } - for (const [key, entry] of Object.entries(data.trees)) { + for (const [key, entry] of Object.entries(trees as SerializedTrees['trees'])) { const nodes = new Map(); for (const [id, nodeData] of Object.entries(entry.nodes)) { nodes.set(id, { @@ -561,7 +651,7 @@ export class CheckpointManager { hostState: entry.hostState, rollback: entry.rollback, currentCheckpoint: entry.current, - currentState: undefined, // Will be loaded per-feature-set during registerFeatureSet + currentState: undefined, // a host-managed tree's state is read by loadHostStates, before its server connects nodes, }); } diff --git a/src/usage/usage-tracker.ts b/src/usage/usage-tracker.ts index d5abf09be..dae6c69da 100644 --- a/src/usage/usage-tracker.ts +++ b/src/usage/usage-tracker.ts @@ -12,10 +12,14 @@ function accumulateUsage( tokens: { inputTokens: number; outputTokens: number; cacheCreationTokens: number; cacheReadTokens: number }, estimatedCost?: { total: number; currency: string } ): void { - target.inputTokens += tokens.inputTokens; - target.outputTokens += tokens.outputTokens; - target.cacheCreationTokens += tokens.cacheCreationTokens; - target.cacheReadTokens += tokens.cacheReadTokens; + // A provider's usage can lack a count (membrane 0.5.87's Bedrock stream + // takes a message_delta's output count as reported); adding it would make + // the total NaN, which is stored as null. + const count = (n: number): number => (Number.isFinite(n) ? n : 0); + target.inputTokens += count(tokens.inputTokens); + target.outputTokens += count(tokens.outputTokens); + target.cacheCreationTokens += count(tokens.cacheCreationTokens); + target.cacheReadTokens += count(tokens.cacheReadTokens); if (estimatedCost) { if (!target.estimatedCost) { target.estimatedCost = { total: 0, currency: estimatedCost.currency }; @@ -31,10 +35,16 @@ function accumulateUsage( function parseUsage(raw: unknown): SessionUsage | null { if (!raw || typeof raw !== 'object') return null; const o = raw as Record; - if (typeof o.inputTokens !== 'number' || typeof o.outputTokens !== 'number') return null; + // A count stored as null is a NaN total, as JSON stores one: an earlier + // release could write it when a provider's usage lacked a count. The sum + // before it was lost when it became NaN, so it reads as 0. + const count = (v: unknown): number | undefined => (v === null ? 0 : typeof v === 'number' ? v : undefined); + const inputTokens = count(o.inputTokens); + const outputTokens = count(o.outputTokens); + if (inputTokens === undefined || outputTokens === undefined) return null; return { - inputTokens: o.inputTokens, - outputTokens: o.outputTokens as number, + inputTokens, + outputTokens, cacheCreationTokens: typeof o.cacheCreationTokens === 'number' ? o.cacheCreationTokens : 0, cacheReadTokens: typeof o.cacheReadTokens === 'number' ? o.cacheReadTokens : 0, estimatedCost: o.estimatedCost && typeof o.estimatedCost === 'object' @@ -49,6 +59,18 @@ export interface PersistedUsageState { inferenceCount: number; } +/** What keeps a stored usage record from restoring, or undefined when it + * restores: it must be an object whose totals are token counts and whose + * byAgent, if it has one, is a list. Its per-agent entries are checked one + * by one as they restore, and one that isn't an agent's usage is skipped. */ +export function persistedUsageProblem(value: unknown): 'not a record' | 'totals' | 'byAgent' | undefined { + if (!value || typeof value !== 'object' || Array.isArray(value)) return 'not a record'; + const { totals, byAgent } = value as { totals?: unknown; byAgent?: unknown }; + if (parseUsage(totals) === null) return 'totals'; + if (byAgent !== undefined && !Array.isArray(byAgent)) return 'byAgent'; + return undefined; +} + export class UsageTracker { private totals: SessionUsage = { inputTokens: 0, outputTokens: 0, cacheCreationTokens: 0, cacheReadTokens: 0 }; private byAgent: Map = new Map(); @@ -71,7 +93,9 @@ export class UsageTracker { this.totals = totals; this.inferenceCount = typeof state.inferenceCount === 'number' ? state.inferenceCount : 0; if (Array.isArray(state.byAgent)) { - for (const entry of state.byAgent) { + for (const item of state.byAgent as unknown[]) { + if (!item || typeof item !== 'object') continue; + const entry = item as { agentName?: unknown; usage?: unknown; inferenceCount?: unknown }; const usage = parseUsage(entry.usage); if (usage && typeof entry.agentName === 'string') { this.byAgent.set(entry.agentName, { diff --git a/test/agent-runtime-settings.test.ts b/test/agent-runtime-settings.test.ts index 4eed3e766..1da7c4f6f 100644 --- a/test/agent-runtime-settings.test.ts +++ b/test/agent-runtime-settings.test.ts @@ -7,6 +7,7 @@ import { JsStore } from '@animalabs/chronicle'; import { AutobiographicalStrategy, ContextManager } from '@animalabs/context-manager'; import { Agent, AgentFramework, BudgetPreflightError, type AgentRuntimeSettingsPatch } from '../src/index.js'; +import { rejectsBuild } from './helpers/failed-create.js'; const membrane = {} as any; @@ -853,7 +854,7 @@ describe('a reached budget across a restart', () => { }); }); - it("a record the boot can't read is logged, and the agent boots at the recipe's", async () => { + it("a record the boot can't read fails create(), naming it, and the next boot restores it", async () => { await run(async ({ framework, restart }) => { framework().updateAgentRuntimeSettings('agent', { contextBudgetTokens: 60_000, immediate: true }); // Only the boot's read of the record fails. @@ -863,19 +864,37 @@ describe('a reached budget across a restart', () => { if (new Error().stack?.includes('readAgentRuntimeSettings')) throw new Error('store unreadable'); return original.apply(this, args); }; - const log = quietErrors(); - let restarted: AgentFramework; try { - restarted = await restart(); + // Booted at the recipe's, its next settings change would store over the record. + await rejectsBuild(restart, /^the persisted settings of agent "agent" \(framework\/state, agentRuntimeSettings\) can't be restored: store unreadable\. The framework doesn't start without them, since the agent would boot at its recipe's settings and the next settings change would store over them\. They're unchanged, and the framework starts once they can be read\.$/); } finally { proto.getStateJson = original; - log.restore(); } - assert.deepEqual(budgets(restarted), { target: 100_000, effective: 100_000, transition: 'stable' }); - assert.ok( - log.lines.some((l) => l.includes("agent: its persisted settings can't be read (store unreadable); it boots at the recipe's")), - log.lines.join('\n'), - ); + const restarted = await restart(); + assert.equal(budgets(restarted).target, 60_000); + assert.deepEqual(persisted(restarted), { contextBudgetTokens: 60_000 }); + }); + }); + + it("what's stored for an agent that isn't a settings record fails create(), naming what it is", async () => { + await run(async ({ framework, restart, config }) => { + writeRecord(framework(), ['contextBudgetTokens', 60_000] as never); + await rejectsBuild(restart, /^the persisted settings of agent "agent" \(framework\/state, agentRuntimeSettings\) can't be restored: what's stored for the agent is an array, not a settings record\. The framework doesn't start without them, since the agent would boot at its recipe's settings and the next settings change would store over them\. They're unchanged, and starting again won't help until they're repaired\.$/); + // The refused boots closed their stores, so write the next form through one of the test's own. + const store = JsStore.openOrCreate({ path: config().storePath }); + try { + const state = store.getStateJson('framework/state') as { agentRuntimeSettings?: object }; + store.setStateJson('framework/state', { ...state, agentRuntimeSettings: { ...state.agentRuntimeSettings, agent: null } }); + } finally { store.close(); } + await rejectsBuild(restart, /can't be restored: what's stored for the agent is null, not a settings record\./); + }); + }); + + it("a map of agents' settings that isn't an object fails create(), naming what it is", async () => { + await run(async ({ framework, restart }) => { + const store = framework().getStore(); + store.setStateJson('framework/state', { ...(store.getStateJson('framework/state') as object), agentRuntimeSettings: 'agent' }); + await rejectsBuild(restart, /^the persisted settings of agent "agent" \(framework\/state, agentRuntimeSettings\) can't be restored: the map that holds every agent's settings is a string, not an object\. /); }); }); diff --git a/test/boot-state-read-errors.test.ts b/test/boot-state-read-errors.test.ts new file mode 100644 index 000000000..c532cdd23 --- /dev/null +++ b/test/boot-state-read-errors.test.ts @@ -0,0 +1,880 @@ +/** + * At boot, a stored state that can't be read, or that reads as something this + * code doesn't read as that state, is a refusal that names it, not an empty + * state. Taken as empty, each was lost or ignored: a quiesced host booted + * serving and landed its deferred writes mid-surgery; a deferred-write queue + * restored nothing and the next deferral replaced it; a deferred write that + * had landed before its ack landed again when an agent's context couldn't be + * scanned; usage totals started at zero and the next inference stored zeros + * over them; an agent's settings booted at its recipe's and the next change + * stored over the record; coalesced work acknowledged before a crash was never + * delivered, and the next receipt or flush replaced what was stored. Each + * refusal leaves the state as it was, so the next boot that can read it + * restores it. The inline caps' failed read loses nothing, since setting a cap + * reads them again, so it is logged and boot goes on. Nothing stored still + * reads as empty. + */ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdirSync, mkdtempSync, readFileSync, renameSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { JsStore } from '@animalabs/chronicle'; +import { ContextManager } from '@animalabs/context-manager'; +import { AgentFramework } from '../src/index.js'; +import type { FrameworkConfig } from '../src/types/framework.js'; +import type { McplServerConfig } from '../src/mcpl/types.js'; +import { MockMembrane } from './helpers/mock-membrane.js'; +import { fixture, crash } from './helpers/coalescing-fixture.js'; +import { rejectsBuild } from './helpers/failed-create.js'; +import { isStateExistsError } from '../src/module-registry.js'; +import { CheckpointManager } from '../src/mcpl/checkpoint-manager.js'; + +/** An I/O error from a store read of `stateId` by the framework's `reader`, + * `times` times, as a failing disk would give. */ +function failStoreRead(stateId: string, reader: string, times = 1): () => void { + const proto = JsStore.prototype as unknown as { getStateJson(this: JsStore, id: string): unknown }; + const original = proto.getStateJson; + let left = times; + proto.getStateJson = function (this: JsStore, id: string) { + if (id === stateId && left > 0 && new Error().stack?.includes(reader)) { + left--; + throw Object.assign(new Error(`injected EIO reading ${id}`), { code: 'EIO' }); + } + return original.call(this, id); + }; + return () => { proto.getStateJson = original; }; +} + +/** An I/O error from the next read of the recovery file whose path ends in `name`. */ +function failRecoveryRead(name: string): () => void { + const proto = AgentFramework.prototype as unknown as { readRecoveryFile(path: string): unknown }; + const original = proto.readRecoveryFile; + let left = 1; + proto.readRecoveryFile = function (this: unknown, path: string) { + if (path.endsWith(name) && left > 0) { + left--; + throw Object.assign(new Error(`injected EIO reading ${name}`), { code: 'EIO' }); + } + return original.call(this, path); + }; + return () => { proto.readRecoveryFile = original; }; +} + +function quietErrors(): { lines: string[]; restore: () => void } { + const lines: string[] = []; + const original = console.error; + console.error = (...args: unknown[]) => { lines.push(args.map(String).join(' ')); }; + return { lines, restore: () => { console.error = original; } }; +} + +const agents = [{ name: 'agent', model: 'test-model', systemPrompt: 'test' }]; +const config = (storePath: string): FrameworkConfig => + ({ storePath, membrane: new MockMembrane().asMembrane(), agents, modules: [] }) as unknown as FrameworkConfig; +const texts = (f: AgentFramework): string[] => + f.getAgent('agent')!.getContextManager().getAllMessages().map((m) => (m.content[0] as { text?: string } | undefined)?.text ?? ''); + +/** A store directory removed after `body`, which receives its storePath. */ +function inStoreDir(prefix: string, body: (storePath: string) => Promise): () => Promise { + return async () => { + const dir = mkdtempSync(join(tmpdir(), prefix)); + try { await body(join(dir, 'store')); } finally { rmSync(dir, { recursive: true, force: true }); } + }; +} + +/** Write `framework/state` while no framework holds the store. */ +function seedFrameworkState(storePath: string, state: Record): void { + const store = JsStore.openOrCreate({ path: storePath }); + try { + try { + store.registerState({ id: 'framework/state', strategy: 'snapshot' }); + } catch (error) { + if (!isStateExistsError(error)) throw error; + } + store.setStateJson('framework/state', state); + } finally { store.close(); } +} +function readState(storePath: string, stateId: string): unknown { + const store = JsStore.openOrCreate({ path: storePath }); + try { return store.getStateJson(stateId); } finally { store.close(); } +} + +/** Quiesce with one write deferred, and stop. */ +async function quiescedWithAWrite(storePath: string, text = 'said during surgery'): Promise { + const f = await AgentFramework.create(config(storePath)); + try { + await f.quiesce({ reason: 'surgery' }); + (f as unknown as { addMessage(p: string, c: unknown[]): unknown }).addMessage('user', [{ type: 'text', text }]); + } finally { await f.stop(); } +} + +// --------------------------------------------------------------------------- +// Host mode +// --------------------------------------------------------------------------- + +test("a host-mode file that can't be read fails create(), and the next boot is quiesced with its writes", inStoreDir('boot-read-hostmode-', async (storePath) => { + await quiescedWithAWrite(storePath); + const file = join(storePath, 'recovery', 'host-mode.json'); + const before = readFileSync(file, 'utf8'); + const restore = failRecoveryRead('host-mode.json'); + try { + await rejectsBuild(() => AgentFramework.create(config(storePath)), new RegExp( + `^the host mode \\(${file.replace(/[.*+?^${}()|[\]\\/]/g, '\\$&')}\\) could not be read: injected EIO reading host-mode\\.json\\. ` + + `The framework doesn't start without it, since a host quiesced for maintenance would boot serving\\. ` + + `The file is unchanged, and the framework starts once it can be read\\.$`, + )); + } finally { restore(); } + assert.equal(readFileSync(file, 'utf8'), before, 'the file is as it was'); + const f = await AgentFramework.create(config(storePath)); + try { + assert.equal(f.getHostModeStatus().quiesced, true); + assert.equal(f.getHostModeStatus().deferredWrites, 1); + assert.ok(!texts(f).includes('said during surgery'), 'the deferred write waits for resume'); + } finally { await f.stop(); } +})); + +test('a host-mode file that holds no record, text that isn\'t JSON, or a record whose quiesced is not a boolean, fails create(), naming what it holds', inStoreDir('boot-read-hostmode-form-', async (storePath) => { + await quiescedWithAWrite(storePath); + const file = join(storePath, 'recovery', 'host-mode.json'); + const literal = (s: string) => s.replace(/[.*+?^${}()|[\]\\/]/g, '\\$&'); + for (const [content, holds] of [ + ['["quiesced"]', literal('holds an array, not a host-mode record')], + ['null', literal('holds null, not a host-mode record')], + ['{"version":1,"quiesced":"yes"}', literal('is a record whose quiesced is a string, not true or false')], + ['{"version":1}', literal('is a record whose quiesced is missing, not true or false')], + // The parser's own words differ between Node versions. + ['{"version":1,"quiesced":tr', "holds text that isn't JSON: .+"], + ]) { + writeFileSync(file, content); + await rejectsBuild(() => AgentFramework.create(config(storePath)), new RegExp( + `^the host mode \\(.*host-mode\\.json\\) ${holds}\\. The framework doesn't start without it, ` + + `since a host quiesced for maintenance would boot serving\\. The file is unchanged, and the framework starts once it holds a ` + + `host-mode record, such as \\{"version":1,"quiesced":true\\} to boot quiesced or \\{"version":1,"quiesced":false\\} to serve\\. ` + + `Removing it doesn't help: without the file, host mode is read from framework/state, which a quiesce doesn't write, so a ` + + `quiesced host would boot serving\\.$`, + ), content); + assert.equal(readFileSync(file, 'utf8'), content, `${content}: the file is as it was`); + } +})); + +test("a recovery file that can't be reached, not only one that won't read, fails create(), and doesn't read as absent", inStoreDir('boot-read-hostmode-enotdir-', async (storePath) => { + // A file where the host-mode file's directory should be makes its read fail + // with ENOTDIR, an error existsSync answers false for, as it does EACCES. + const dir = join(storePath, '..', 'host-mode-dir'); + const withPath = (): FrameworkConfig => ({ ...config(storePath), hostModePath: join(dir, 'host-mode.json') }) as FrameworkConfig; + const first = await AgentFramework.create(withPath()); + try { await first.quiesce({ reason: 'surgery' }); } finally { await first.stop(); } + renameSync(dir, `${dir}.aside`); + writeFileSync(dir, 'not a directory'); + try { + await rejectsBuild(() => AgentFramework.create(withPath()), /^the host mode \(.*host-mode-dir\/host-mode\.json\) could not be read: ENOTDIR\b.*The file is unchanged, and the framework starts once it can be read\.$/); + } finally { + rmSync(dir); + renameSync(`${dir}.aside`, dir); + } + const f = await AgentFramework.create(withPath()); + try { assert.equal(f.getHostModeStatus().quiesced, true); } finally { await f.stop(); } +})); + +test("a deferred-writes file that can't be reached fails create(), and doesn't read as absent", inStoreDir('boot-read-deferred-enotdir-', async (storePath) => { + // As for the host-mode file: ENOTDIR from a file where the queue file's directory should be. + const dir = join(storePath, '..', 'deferred-dir'); + const withPath = (): FrameworkConfig => ({ ...config(storePath), deferredWritesPath: join(dir, 'deferred-writes.json') }) as FrameworkConfig; + const first = await AgentFramework.create(withPath()); + try { + await first.quiesce({ reason: 'surgery' }); + (first as unknown as { addMessage(p: string, c: unknown[]): unknown }).addMessage('user', [{ type: 'text', text: 'said during surgery' }]); + } finally { await first.stop(); } + renameSync(dir, `${dir}.aside`); + writeFileSync(dir, 'not a directory'); + try { + await rejectsBuild(() => AgentFramework.create(withPath()), /^the deferred context writes \(.*deferred-dir\/deferred-writes\.json\) could not be read: ENOTDIR\b.*The file is unchanged, and the framework starts once it can be read\.$/); + } finally { + rmSync(dir); + renameSync(`${dir}.aside`, dir); + } + const f = await AgentFramework.create(withPath()); + try { assert.equal(f.getHostModeStatus().deferredWrites, 1, 'the next reachable boot restores the queue'); } finally { await f.stop(); } +})); + +test("with no recovery file, a host mode that can't be read from framework/state fails create()", async () => { + const dir = mkdtempSync(join(tmpdir(), 'boot-read-hostmode-slot-')); + const store = JsStore.openOrCreate({ path: join(dir, 'store') }); + // A store passed in, with no recovery paths, keeps host mode in framework/state. + const slotConfig = () => ({ store, membrane: new MockMembrane().asMembrane(), agents, modules: [] }) as unknown as FrameworkConfig; + try { + const first = await AgentFramework.create(slotConfig()); + await first.quiesce({ reason: 'surgery' }); + await first.stop(); + const restore = failStoreRead('framework/state', 'readHostMode'); + try { + await rejectsBuild(() => AgentFramework.create(slotConfig()), /^the host mode \(framework\/state, hostMode\) could not be read: injected EIO reading framework\/state\. The framework doesn't start without it, since a host quiesced for maintenance would boot serving\. It's unchanged, and the framework starts once it can be read\.$/); + } finally { restore(); } + const f = await AgentFramework.create(slotConfig()); + try { assert.equal(f.getHostModeStatus().quiesced, true); } finally { await f.stop(); } + for (const [hostMode, holds] of [['quiesced', 'holds a string'], [null, 'holds null']] as const) { + store.setStateJson('framework/state', { ...(store.getStateJson('framework/state') as object), hostMode }); + await rejectsBuild(() => AgentFramework.create(slotConfig()), new RegExp( + `^the host mode \\(framework/state, hostMode\\) ${holds}, not a host-mode record\\. The framework doesn't start without it, ` + + `since a host quiesced for maintenance would boot serving\\. It's unchanged, and starting again won't help until it's repaired\\.$`, + ), String(hostMode)); + } + } finally { + store.close(); + rmSync(dir, { recursive: true, force: true }); + } +}); + +// --------------------------------------------------------------------------- +// Deferred writes +// --------------------------------------------------------------------------- + +test("a deferred-write queue that can't be read fails create(), and the next boot lands its writes at resume", inStoreDir('boot-read-deferred-', async (storePath) => { + await quiescedWithAWrite(storePath); + const file = join(storePath, 'recovery', 'deferred-writes.json'); + const before = readFileSync(file, 'utf8'); + const restore = failRecoveryRead('deferred-writes.json'); + try { + await rejectsBuild(() => AgentFramework.create(config(storePath)), /^the deferred context writes \(.*deferred-writes\.json\) could not be read: injected EIO reading deferred-writes\.json\. The framework doesn't start without them, since the next deferral would replace the stored queue\. The file is unchanged, and the framework starts once it can be read\.$/); + } finally { restore(); } + assert.equal(readFileSync(file, 'utf8'), before, 'the queue is as it was'); + const f = await AgentFramework.create(config(storePath)); + try { + assert.equal(f.getHostModeStatus().deferredWrites, 1); + await f.resume(); + assert.deepEqual(texts(f).filter((t) => t === 'said during surgery').length, 1); + } finally { await f.stop(); } +})); + +test("a deferred-write queue that isn't a queue, or isn't JSON, fails create(), naming what it holds", inStoreDir('boot-read-deferred-form-', async (storePath) => { + await quiescedWithAWrite(storePath); + const file = join(storePath, 'recovery', 'deferred-writes.json'); + for (const [content, holds] of [ + ['"pending"', 'hold a string, not a queue of writes'], + ['null', 'hold null, not a queue of writes'], + ['{"version":2,"pending":{"0":{}}}', 'hold an object whose pending is an object, not a list of writes'], + ['{"version":2}', 'hold an object whose pending is missing, not a list of writes'], + // The parser's own words differ between Node versions. + ['{"version":2,', "hold text that isn't JSON: .+"], + ] as const) { + writeFileSync(file, content); + await rejectsBuild(() => AgentFramework.create(config(storePath)), new RegExp( + `^the deferred context writes \\(.*deferred-writes\\.json\\) ${holds}\\. The framework doesn't start without them, since the next ` + + `deferral would replace the stored queue\\. The file is unchanged, and the framework starts once it holds a queue of writes, ` + + `such as \\{"version":1,"pending":\\[\\]\\} for none\\. Removing it reads the queue from framework/deferred-writes instead, ` + + `which the file's writes don't keep, so it restores whatever older queue is there, or none\\.$`, + ), content); + assert.equal(readFileSync(file, 'utf8'), content, `${content}: the file is as it was`); + } +})); + +test("with no recovery file, a deferred-write queue that can't be read, or isn't a queue, fails create()", async () => { + const dir = mkdtempSync(join(tmpdir(), 'boot-read-deferred-slot-')); + const store = JsStore.openOrCreate({ path: join(dir, 'store') }); + // A store passed in, with no recovery paths, keeps the queue in framework/deferred-writes. + const slotConfig = () => ({ store, membrane: new MockMembrane().asMembrane(), agents, modules: [] }) as unknown as FrameworkConfig; + try { + const first = await AgentFramework.create(slotConfig()); + await first.quiesce({ reason: 'surgery' }); + (first as unknown as { addMessage(p: string, c: unknown[]): unknown }).addMessage('user', [{ type: 'text', text: 'said during surgery' }]); + await first.stop(); + const queue = store.getStateJson('framework/deferred-writes'); + const restore = failStoreRead('framework/deferred-writes', 'restorePersistedDeferredWrites'); + try { + await rejectsBuild(() => AgentFramework.create(slotConfig()), /^the deferred context writes \(framework\/deferred-writes\) could not be read: injected EIO reading framework\/deferred-writes\. The framework doesn't start without them, since the next deferral would replace the stored queue\. They're unchanged, and the framework starts once they can be read\.$/); + } finally { restore(); } + assert.deepEqual(store.getStateJson('framework/deferred-writes'), queue, 'the queue is as it was'); + const f = await AgentFramework.create(slotConfig()); + try { assert.equal(f.getHostModeStatus().deferredWrites, 1); } finally { await f.stop(); } + store.setStateJson('framework/deferred-writes', 'pending'); + await rejectsBuild(() => AgentFramework.create(slotConfig()), /^the deferred context writes \(framework\/deferred-writes\) hold a string, not a queue of writes\. The framework doesn't start without them, since the next deferral would replace the stored queue\. They're unchanged, and starting again won't help until they're repaired\.$/); + } finally { + store.close(); + rmSync(dir, { recursive: true, force: true }); + } +}); + +test("a deferred write whose agent's context can't be scanned fails create(), and the next boot doesn't land it twice", inStoreDir('boot-read-deferred-landed-', async (storePath) => { + // A write that landed before its ack: in the queue, and in the context with its id. + await quiescedWithAWrite(storePath); + const [queued] = JSON.parse(readFileSync(join(storePath, 'recovery', 'deferred-writes.json'), 'utf8')).pending as Array<{ id: string }>; + const staging = await AgentFramework.create(config(storePath)); + try { + staging.getAgent('agent')!.getContextManager() + .addMessage('user', [{ type: 'text', text: 'said during surgery' }], { deferredWriteId: queued!.id } as never); + } finally { await staging.stop(); } + // A crash between a write landing and its ack leaves the receipt its + // hand-off recorded, from before the copy landed. Staging's stop handed the + // queue off again from after the copy, so the receipt goes back to where + // the agent's context began. + const queueFile = join(storePath, 'recovery', 'deferred-writes.json'); + writeFileSync(queueFile, JSON.stringify({ ...JSON.parse(readFileSync(queueFile, 'utf8')), scanFrom: { agent: 0 } })); + const proto = ContextManager.prototype as unknown as { getMessageCount(this: ContextManager): number }; + const original = proto.getMessageCount; + proto.getMessageCount = function (this: ContextManager) { + if (new Error().stack?.includes('landedDeferredWriteIds')) throw new Error('injected EIO scanning the context'); + return original.call(this); + }; + try { + await rejectsBuild(() => AgentFramework.create(config(storePath)), /^the deferred context writes \(.*deferred-writes\.json\) can't be checked for writes that already landed: agent's context could not be read: injected EIO scanning the context\. The framework doesn't start without that check, since a write that landed before its ack would land a second time\. The queue is unchanged, and the framework starts once that context can be read\.$/); + } finally { proto.getMessageCount = original; } + const f = await AgentFramework.create(config(storePath)); + try { + assert.equal(f.getHostModeStatus().deferredWrites, 0, 'the landed write is skipped'); + await f.resume(); + assert.equal(texts(f).filter((t) => t === 'said during surgery').length, 1, 'it is in the context once'); + } finally { await f.stop(); } +})); + +// --------------------------------------------------------------------------- +// Usage totals +// --------------------------------------------------------------------------- + +const storedUsage = { + totals: { inputTokens: 1000, outputTokens: 500 }, + inferenceCount: 7, + byAgent: [{ agentName: 'agent', usage: { inputTokens: 1000, outputTokens: 500 }, inferenceCount: 7 }], +}; + +test("usage totals that can't be read fail create(), and the next boot restores them", inStoreDir('boot-read-usage-', async (storePath) => { + seedFrameworkState(storePath, { usage: storedUsage }); + const restore = failStoreRead('framework/state', 'restoreUsageState'); + try { + await rejectsBuild(() => AgentFramework.create(config(storePath)), /^the usage totals \(framework\/state, usage\) could not be read: injected EIO reading framework\/state\. The framework doesn't start without them, since the next inference would store fresh totals over them\. They're unchanged, and the framework starts once they can be read\.$/); + } finally { restore(); } + assert.deepEqual((readState(storePath, 'framework/state') as { usage: unknown }).usage, storedUsage); + const f = await AgentFramework.create(config(storePath)); + try { + assert.equal(f.getSessionUsage().totals.inputTokens, 1000); + assert.equal(f.getSessionUsage().totals.outputTokens, 500); + } finally { await f.stop(); } +})); + +test("usage totals that aren't a usage record fail create(), naming what they are", inStoreDir('boot-read-usage-form-', async (storePath) => { + for (const [usage, are] of [ + [{ totals: 'lots' }, "are a record whose totals aren't token counts"], + [[storedUsage], 'are an array, not a usage record'], + [null, 'are null, not a usage record'], + [{ ...storedUsage, byAgent: 'agent' }, "are a record whose byAgent isn't a list"], + ] as const) { + seedFrameworkState(storePath, { usage }); + await rejectsBuild(() => AgentFramework.create(config(storePath)), new RegExp( + `^the usage totals \\(framework/state, usage\\) ${are}\\. The framework doesn't start without them, since the next inference ` + + `would store fresh totals over them\\. They're unchanged, and starting again won't help until they're repaired\\.$`, + ), JSON.stringify(usage)); + assert.deepEqual((readState(storePath, 'framework/state') as { usage: unknown }).usage, usage); + } +})); + +test('usage totals an earlier release wrote with a null count, or with a byAgent entry that is not an agent, restore', inStoreDir('boot-read-usage-null-', async (storePath) => { + // A NaN total, which a provider's usage without a count made, is stored as null. + seedFrameworkState(storePath, { usage: { + totals: { inputTokens: null, outputTokens: 500 }, + inferenceCount: 7, + byAgent: [null, { agentName: 'agent', usage: { inputTokens: 1000, outputTokens: null }, inferenceCount: 7 }], + } }); + const f = await AgentFramework.create(config(storePath)); + try { + const usage = f.getSessionUsage(); + assert.equal(usage.totals.inputTokens, 0); + assert.equal(usage.totals.outputTokens, 500); + assert.deepEqual(usage.byAgent.map((a) => [a.agentName, a.usage.inputTokens, a.usage.outputTokens]), [['agent', 1000, 0]]); + } finally { await f.stop(); } +})); + +test('a framework/state that is not an object fails create(), naming it, and is left as it was', inStoreDir('boot-read-state-root-', async (storePath) => { + const seed = JsStore.openOrCreate({ path: storePath }); + try { + seed.registerState({ id: 'framework/state', strategy: 'snapshot' }); + seed.setStateJson('framework/state', 'usage'); + } finally { seed.close(); } + await rejectsBuild(() => AgentFramework.create(config(storePath)), /^framework\/state holds a string, not an object\. The framework doesn't start on it, since its first write would replace it and everything it keeps, such as the usage totals and each agent's settings\. It's unchanged, and starting again won't help until it's repaired\.$/); + assert.equal(readState(storePath, 'framework/state'), 'usage'); +})); + +// --------------------------------------------------------------------------- +// Inline caps +// --------------------------------------------------------------------------- + +test("inline caps that can't be read are logged, boot goes on, and setting one keeps the others stored", inStoreDir('boot-read-caps-', async (storePath) => { + seedFrameworkState(storePath, { toolResultInlineCaps: { agent: 5000 } }); + const twoAgents = { ...config(storePath), agents: [...agents, { name: 'other', model: 'test-model', systemPrompt: 'test' }] } as FrameworkConfig; + const restore = failStoreRead('framework/state', 'readPersistedToolResultInlineCaps'); + const log = quietErrors(); + let f: AgentFramework; + try { + f = await AgentFramework.create(twoAgents); + } finally { restore(); log.restore(); } + try { + assert.ok( + log.lines.some((l) => l.includes( + "[agent-settings] the persisted tool_result_inline_max_chars caps (framework/state) could not be read: injected EIO reading framework/state; they don't apply until a restart, and setting one keeps the others stored", + )), + log.lines.join('\n'), + ); + const settings = (f as unknown as { + collectAgentSettingsExtensions(): Map; + update(agentName: string, patch: Record): unknown; + }>; + }).collectAgentSettingsExtensions().get('_framework')!; + assert.equal(settings.get('agent').tool_result_inline_max_chars, null, "the stored cap doesn't apply"); + settings.update('other', { tool_result_inline_max_chars: 2000 }); + } finally { await f.stop(); } + assert.deepEqual((readState(storePath, 'framework/state') as { toolResultInlineCaps: unknown }).toolResultInlineCaps, { agent: 5000, other: 2000 }); +})); + +test("a stored inline cap that can't become a number is dropped, loudly, and the others apply", inStoreDir('boot-read-caps-entry-', async (storePath) => { + // Number() throws for an object it can't make a primitive of. + seedFrameworkState(storePath, { toolResultInlineCaps: { agent: { toString: null }, other: 5000 } }); + const twoAgents = { ...config(storePath), agents: [...agents, { name: 'other', model: 'test-model', systemPrompt: 'test' }] } as FrameworkConfig; + const log = quietErrors(); + let f: AgentFramework; + try { + f = await AgentFramework.create(twoAgents); + } finally { log.restore(); } + try { + assert.ok( + log.lines.some((l) => l.includes(`[agent-settings] dropping invalid persisted tool_result_inline_max_chars for 'agent': {"toString":null}`)), + log.lines.join('\n'), + ); + const settings = (f as unknown as { + collectAgentSettingsExtensions(): Map }>; + }).collectAgentSettingsExtensions().get('_framework')!; + assert.equal(settings.get('agent').tool_result_inline_max_chars, null); + assert.equal(settings.get('other').tool_result_inline_max_chars, 5000); + } finally { await f.stop(); } +})); + +// --------------------------------------------------------------------------- +// Tool-result guards +// --------------------------------------------------------------------------- + +const guardOf = (f: AgentFramework, name: string): boolean | undefined => + (f.getAgent(name) as unknown as { toolResultGuard: { settingOverride?: boolean } }).toolResultGuard.settingOverride; + +test("tool-result guards that can't be read, or that aren't an object, fail create(), naming them, and are left as they were", inStoreDir('boot-read-guards-', async (storePath) => { + const first = await AgentFramework.create(config(storePath)); + await first.stop(); + seedFrameworkState(storePath, { toolResultGuards: { agent: true } }); + const restore = failStoreRead('framework/state', 'adoptToolResultGuard'); + try { + await rejectsBuild(() => AgentFramework.create(config(storePath)), /^the tool-result guards \(framework\/state, toolResultGuards\) could not be read: injected EIO reading framework\/state\. Agent "agent" isn't created, since without them it would take its recipe's guard\. They're unchanged, and it's created once they can be read\.$/); + } finally { restore(); } + const f = await AgentFramework.create(config(storePath)); + try { assert.equal(guardOf(f, 'agent'), true, 'the next boot restores the guard'); } finally { await f.stop(); } + for (const [guards, are] of [['agent', 'a string'], [null, 'null'], [[true], 'an array']] as const) { + seedFrameworkState(storePath, { toolResultGuards: guards }); + await rejectsBuild(() => AgentFramework.create(config(storePath)), new RegExp( + `^the tool-result guards \\(framework/state, toolResultGuards\\) are ${are}, not an object of agents' guards\\. Agent "agent" isn't ` + + `created, since without them it would take its recipe's guard and the next guard change would store over them\\. They're ` + + `unchanged, and creating it again won't help until the host's operator repairs them\\.$`, + ), JSON.stringify(guards)); + assert.deepEqual((readState(storePath, 'framework/state') as { toolResultGuards: unknown }).toolResultGuards, guards); + } +})); + +// --------------------------------------------------------------------------- +// Coalescing +// --------------------------------------------------------------------------- + +test("coalescing receipts that can't be read fail create(), and the next boot delivers the acknowledged work", async (t) => { + const f = await fixture(); t.after(f.close); + const params = f.params('1', 'accepted_before_crash', { deferred: true, data: { k: 'v' } }); + const original = await f.send('push/event', params); + await crash(f); + const restore = failStoreRead('mcpl/coalescing-recent', 'restorePushCoalescer'); + try { + await rejectsBuild(() => f.create(), /^the coalescing receipts since the last snapshot \(mcpl\/coalescing-recent\) could not be read: injected EIO reading mcpl\/coalescing-recent\. MCPL doesn't start without them, and neither does create\(\) with MCPL servers configured, nor a first connectMcplServer\(\), since the next coalesced push or message would replace them\. They're unchanged, and MCPL starts once they can be read\.$/); + } finally { restore(); } + await f.create(); + const retry = await f.send('push/event', params); + assert.deepEqual(retry.result, original.result, 'the receipt survived'); + await f.framework.runUntilIdle(); + assert.equal(f.membrane.calls.length, 1, 'the accepted work was delivered'); + assert.ok(f.lastRequest().includes('accepted_before_crash')); +}); + +test("a coalescing snapshot that can't be read fails create(), and the next boot renders its batch", async (t) => { + const f = await fixture(); t.after(f.close); + await f.send('push/event', f.params('1', 'pending_before_crash', { deferred: true })); + (f.framework as unknown as { flushCoalescingSnapshot(): void }).flushCoalescingSnapshot(); + await crash(f); + const restore = failStoreRead('mcpl/coalescing', 'restorePushCoalescer'); + try { + await rejectsBuild(() => f.create(), /^the coalescing receipts and batches \(mcpl\/coalescing\) could not be read: injected EIO reading mcpl\/coalescing\. MCPL doesn't start without them/); + } finally { restore(); } + await f.create(); + await f.framework.runUntilIdle(); + assert.equal(f.renders.length, 1, 'the restored batch was rendered'); + assert.ok(f.lastRequest().includes('document_diff')); +}); + +test("coalescing receipts whose delivery an agent's context can't confirm fail create(), and the next boot delivers once", async (t) => { + const f = await fixture(); t.after(f.close); + const params = f.params('1', 'accepted_before_crash', { deferred: true, data: { k: 'v' } }); + const original = await f.send('push/event', params); + await crash(f); + // Restoring a receipt asks every agent's context whether its occurrence was delivered. + const proto = ContextManager.prototype as unknown as { getAllMessages(this: ContextManager): unknown[] }; + const real = proto.getAllMessages; + proto.getAllMessages = function (this: ContextManager) { + if (new Error().stack?.includes('wasPublished')) throw new Error('injected EIO reading the context'); + return real.call(this); + }; + try { + await rejectsBuild(() => f.create(), /^the coalescing receipts since the last snapshot \(mcpl\/coalescing-recent\) can't be checked for occurrences already delivered: agent's context could not be read: injected EIO reading the context\. MCPL doesn't start without that check, and neither does create\(\) with MCPL servers configured, nor a first connectMcplServer\(\), since an occurrence delivered before the restart would be delivered again\. They're unchanged, and MCPL starts once that context can be read\.$/); + } finally { proto.getAllMessages = real; } + await f.create(); + const retry = await f.send('push/event', params); + assert.deepEqual(retry.result, original.result, 'the receipt survived'); + await f.framework.runUntilIdle(); + assert.equal(f.membrane.calls.length, 1, 'the accepted work was delivered once'); +}); + +test("coalescing state this code doesn't read fails create(), naming what it is, and is left as it was", async (t) => { + const f = await fixture(); t.after(f.close); + await f.send('push/event', f.params('1', 'pending', { deferred: true })); + (f.framework as unknown as { flushCoalescingSnapshot(): void }).flushCoalescingSnapshot(); + await crash(f); + const storePath = join(f.dir, 'store'); + const snapshot = readState(storePath, 'mcpl/coalescing') as Record; + const cases: Array<[string, string, unknown, string]> = [ + ['a version 2 snapshot', 'mcpl/coalescing', { ...snapshot, version: 2 }, "receipts and batches \\(mcpl/coalescing\\) are a version 2 snapshot, which this version of agent-framework doesn't read\\. MCPL doesn't start without them, and neither does create\\(\\) with MCPL servers configured, nor a first connectMcplServer\\(\\), since the next coalesced push or message would replace them\\. They're unchanged, and the version that wrote them reads them\\.$"], + ['a snapshot whose version is a string', 'mcpl/coalescing', { ...snapshot, version: '1' }, "receipts and batches \\(mcpl/coalescing\\) are a snapshot whose version is a string, not 1\\. .* They're unchanged, and starting again won't help until they're repaired\\.$"], + ['a snapshot with no version', 'mcpl/coalescing', { receipts: [], subjects: [] }, 'receipts and batches \\(mcpl/coalescing\\) are an object with no snapshot version\\.'], + ['a snapshot that is an array', 'mcpl/coalescing', [snapshot], 'receipts and batches \\(mcpl/coalescing\\) are an array with no snapshot version\\.'], + ['a snapshot restore throws on', 'mcpl/coalescing', { ...snapshot, subjects: 5 }, 'receipts and batches \\(mcpl/coalescing\\) could not be restored: .*\\. MCPL doesn\'t start'], + ['receipts that are an object', 'mcpl/coalescing-recent', { 0: 'receipt' }, 'receipts since the last snapshot \\(mcpl/coalescing-recent\\) are an object, not a list of receipts\\.'], + ['receipts restoreReceipts throws on', 'mcpl/coalescing-recent', [null], 'receipts since the last snapshot \\(mcpl/coalescing-recent\\) could not be restored: .*\\. MCPL doesn\'t start'], + ]; + for (const [name, stateId, value, says] of cases) { + const store = JsStore.openOrCreate({ path: storePath }); + const previous = store.getStateJson(stateId); + store.setStateJson(stateId, value); + store.close(); + await rejectsBuild(() => f.create(), new RegExp(`^the coalescing ${says}`), name); + assert.deepEqual(readState(storePath, stateId), value, `${name}: left as it was`); + const reset = JsStore.openOrCreate({ path: storePath }); + reset.setStateJson(stateId, previous); + reset.close(); + } +}); + +// --------------------------------------------------------------------------- +// MCPL checkpoints +// --------------------------------------------------------------------------- + +test("MCPL checkpoint trees that can't be read, or that aren't checkpoint trees, fail create(), and are left as they were", async (t) => { + const f = await fixture(); t.after(f.close); + await crash(f); + const storePath = join(f.dir, 'store'); + const restore = failStoreRead('mcpl/checkpoints', 'loadFromStore'); + try { + await rejectsBuild(() => f.create(), /^the MCPL checkpoint trees \(mcpl\/checkpoints\) could not be read: injected EIO reading mcpl\/checkpoints\. MCPL doesn't start without them, and neither does create\(\) with MCPL servers configured, nor a first connectMcplServer\(\), since the next feature set registered would store over them\. They're unchanged, and MCPL starts once they can be read\.$/); + } finally { restore(); } + const before = readState(storePath, 'mcpl/checkpoints'); + for (const [name, value, are] of [ + ['an object without trees', { tree: {} }, 'an object whose trees is missing'], + ['an array', [{ trees: {} }], 'an array'], + ['trees that are a string', { trees: 'srv:doc' }, 'an object whose trees is a string'], + ] as const) { + const store = JsStore.openOrCreate({ path: storePath }); + store.setStateJson('mcpl/checkpoints', value); + store.close(); + await rejectsBuild(() => f.create(), new RegExp( + `^the MCPL checkpoint trees \\(mcpl/checkpoints\\) are ${are}, not checkpoint trees\\. MCPL doesn't start without them, .*` + + `They're unchanged, and starting again won't help until they're repaired\\.$`, + ), name); + assert.deepEqual(readState(storePath, 'mcpl/checkpoints'), value, `${name}: left as it was`); + } + const reset = JsStore.openOrCreate({ path: storePath }); + reset.setStateJson('mcpl/checkpoints', before); + reset.close(); + await f.create(); +}); + +/** A stored host-managed checkpoint tree for `server`'s `featureSet`, holding `state`, as the checkpoint manager stores them. */ +function seedHostManaged(storePath: string, server: string, featureSet: string, state: unknown): void { + const store = JsStore.openOrCreate({ path: storePath }); + try { + for (const id of ['mcpl/checkpoints', `mcpl/state/${server}/${featureSet}`]) { + try { store.registerState({ id, strategy: 'snapshot' }); } catch (error) { if (!isStateExistsError(error)) throw error; } + } + store.setStateJson('mcpl/checkpoints', { trees: { [`${server}:${featureSet}`]: { + hostState: true, rollback: false, current: 'c1', nodes: { c1: { parent: null, children: [], data: state } }, + } } }); + store.setStateJson(`mcpl/state/${server}/${featureSet}`, state); + } finally { store.close(); } +} + +/** The refusal of a host-managed state read before its server connects, or, with `registration`, when its feature set registers. */ +const hostStateUnreadable = (server: string, featureSet: string, when: 'before-connect' | 'registration' = 'before-connect'): RegExp => new RegExp( + `^the host-managed state of feature set "${featureSet}" on MCPL server "${server}" \\(mcpl/state/${server}/${featureSet}\\) could not ` + + `be read: injected EIO reading mcpl/state/${server}/${featureSet}\\. ` + + (when === 'before-connect' + ? `Neither create\\(\\) with the server configured nor a connectMcplServer\\(\\) of it goes on without it, since the feature set's ` + + `next checkpoint would store over it\\. It's unchanged, and they go on once it can be read\\.$` + : `The feature set isn't registered without it, since its next checkpoint would store over it, so the server's connect ` + + `fails\\. It's unchanged, and it's read again when the feature set next registers\\.$`), +); + +test("at boot, a configured server's restored host-managed state that can't be read fails create() before any server connects, and the next boot reads it", async (t) => { + // The fixture declares `doc` with neither hostState nor rollback, so it never + // registers with the checkpoint manager: its stored tree is read at MCPL's start. + const f = await fixture(); t.after(f.close); + await crash(f); + const storePath = join(f.dir, 'store'); + seedHostManaged(storePath, 'editor', 'doc', { count: 1 }); + const initializes = f.hostCaps.length; + const restore = failStoreRead('mcpl/state/editor/doc', 'loadHostState'); + try { + await rejectsBuild(() => f.create(), hostStateUnreadable('editor', 'doc')); + } finally { restore(); } + assert.equal(f.hostCaps.length, initializes, 'no server connected: the fixture saw no initialize'); + assert.deepEqual(readState(storePath, 'mcpl/state/editor/doc'), { count: 1 }, 'the state is as it was'); + await f.create(); + const checkpoints = (f.framework as unknown as { checkpointManager: CheckpointManager }).checkpointManager; + assert.deepEqual(checkpoints.getCurrentState('editor', 'doc'), { count: 1 }, 'the next boot reads it'); +}); + +test("at a runtime connect, a restored host-managed state that can't be read fails the connect before the server is added, and the next connect reads it", async (t) => { + const f = await fixture(); t.after(f.close); + await crash(f); + const storePath = join(f.dir, 'store'); + seedHostManaged(storePath, 'editor', 'doc', { count: 1 }); + // With no MCPL servers configured, the subsystem comes up at the first connect. + const framework = await AgentFramework.create({ + storePath, membrane: f.membrane.asMembrane(), + agents: [{ name: 'agent', model: 'test', systemPrompt: 'test' }], modules: [], + } as unknown as FrameworkConfig); + try { + const server = { id: 'editor', url: `ws://127.0.0.1:${f.port}`, enabledFeatureSets: ['doc'] } as McplServerConfig; + const internals = framework as unknown as { + mcplServerRegistry: { getServer(id: string): unknown } | null; + checkpointManager: CheckpointManager | null; + }; + const restore = failStoreRead('mcpl/state/editor/doc', 'loadHostState'); + try { + await assert.rejects(framework.connectMcplServer(server), { message: hostStateUnreadable('editor', 'doc') }); + } finally { restore(); } + // The first connect brought MCPL up, and the read failed before the server was added. + assert.equal(internals.mcplServerRegistry?.getServer('editor'), null, 'the server was not added'); + await framework.connectMcplServer(server); + assert.ok(internals.mcplServerRegistry?.getServer('editor'), 'the next connect adds it'); + assert.deepEqual(internals.checkpointManager?.getCurrentState('editor', 'doc'), { count: 1 }, 'and reads its state'); + } finally { await framework.stop(); } +}); + +test("a connect of a server already connected is refused as before, without reading its trees again", async (t) => { + const f = await fixture(); t.after(f.close); + await crash(f); + // A host-managed tree whose stored state holds nothing, so it holds no state in memory once read. + seedHostManaged(join(f.dir, 'store'), 'editor', 'notes', null); + await f.create(); + const server = { id: 'editor', url: `ws://127.0.0.1:${f.port}`, enabledFeatureSets: ['doc'] } as McplServerConfig; + const restore = failStoreRead('mcpl/state/editor/notes', 'loadHostState'); + try { + await assert.rejects(f.framework.connectMcplServer(server), { message: 'MCPL server "editor" is already registered' }); + } finally { restore(); } +}); + +test('a host-managed feature set restored from the store has its state read by loadHostStates, so its next patch applies to it', inStoreDir('boot-read-host-state-', async (storePath) => { + const noop = () => {}; + const first = JsStore.openOrCreate({ path: storePath }); + try { + const checkpoints = new CheckpointManager(first, noop); + checkpoints.registerFeatureSet('srv', 'doc', { hostState: true, rollback: false }); + checkpoints.recordCheckpoint('srv', 'doc', { checkpoint: 'c1', data: { count: 1 } } as never); + } finally { first.close(); } + // A restart: the store opened again, its trees restored, their states read before the server connects, and the feature set registered again. + const store = JsStore.openOrCreate({ path: storePath }); + try { + const checkpoints = new CheckpointManager(store, noop); + assert.equal(checkpoints.getCurrentState('srv', 'doc'), undefined, 'restored, the tree holds no state until it is read'); + checkpoints.loadHostStates('srv'); + checkpoints.registerFeatureSet('srv', 'doc', { hostState: true, rollback: false }); + assert.deepEqual(checkpoints.getCurrentState('srv', 'doc'), { count: 1 }); + checkpoints.recordCheckpoint('srv', 'doc', { checkpoint: 'c2', parent: 'c1', patch: [{ op: 'add', path: '/other', value: 2 }] } as never); + assert.deepEqual(store.getStateJson('mcpl/state/srv/doc'), { count: 1, other: 2 }); + } finally { store.close(); } +})); + +test("loadHostStates reads only its own server's host-managed trees, and only those with no state in memory", inStoreDir('boot-read-host-state-scope-', async (storePath) => { + const noop = () => {}; + const first = JsStore.openOrCreate({ path: storePath }); + try { + const checkpoints = new CheckpointManager(first, noop); + checkpoints.registerFeatureSet('srv', 'doc', { hostState: true, rollback: false }); + checkpoints.recordCheckpoint('srv', 'doc', { checkpoint: 'c1', data: { count: 1 } } as never); + checkpoints.registerFeatureSet('other', 'doc', { hostState: true, rollback: false }); + checkpoints.recordCheckpoint('other', 'doc', { checkpoint: 'c1', data: { count: 2 } } as never); + // A rollback-only tree keeps no host-managed state. + checkpoints.registerFeatureSet('srv', 'undo', { hostState: false, rollback: true }); + } finally { first.close(); } + const store = JsStore.openOrCreate({ path: storePath }); + // The states loadHostStates reads. + const proto = JsStore.prototype as unknown as { getStateJson(this: JsStore, id: string): unknown }; + const original = proto.getStateJson; + const read: string[] = []; + proto.getStateJson = function (this: JsStore, id: string) { + if (new Error().stack?.includes('loadHostState')) read.push(id); + return original.call(this, id); + }; + try { + const checkpoints = new CheckpointManager(store, noop); + checkpoints.loadHostStates('srv'); + assert.deepEqual(read, ['mcpl/state/srv/doc'], "only srv's host-managed tree is read"); + assert.deepEqual(checkpoints.getCurrentState('srv', 'doc'), { count: 1 }); + assert.equal(checkpoints.getCurrentState('other', 'doc'), undefined, "another server's tree waits for its own connect"); + // A state already in memory, as a connected server's is, isn't read again. + checkpoints.loadHostStates('srv'); + assert.deepEqual(read, ['mcpl/state/srv/doc'], 'nothing is read a second time'); + } finally { + proto.getStateJson = original; + store.close(); + } +})); + +test("a host-managed feature set's state that can't be read fails loadHostStates for a restored tree, or the registration for a first-time one, naming it, and the next one reads it", inStoreDir('boot-read-host-state-read-', async (storePath) => { + const noop = () => {}; + const first = JsStore.openOrCreate({ path: storePath }); + try { + const checkpoints = new CheckpointManager(first, noop); + checkpoints.registerFeatureSet('srv', 'doc', { hostState: true, rollback: false }); + checkpoints.recordCheckpoint('srv', 'doc', { checkpoint: 'c1', data: { count: 1 } } as never); + // A second feature set's state, stored with no tree: its first registration reads it. + first.registerState({ id: 'mcpl/state/srv/notes', strategy: 'snapshot' }); + first.setStateJson('mcpl/state/srv/notes', { count: 3 }); + } finally { first.close(); } + const store = JsStore.openOrCreate({ path: storePath }); + try { + const checkpoints = new CheckpointManager(store, noop); + // `doc` is a tree restored from the store, read by loadHostStates; `notes` + // is registered for the first time, and read by its registration. + const reads = { + doc: () => checkpoints.loadHostStates('srv'), + notes: () => checkpoints.registerFeatureSet('srv', 'notes', { hostState: true, rollback: false }), + }; + for (const [featureSet, state] of [['doc', { count: 1 }], ['notes', { count: 3 }]] as const) { + const stateId = `mcpl/state/srv/${featureSet}`; + const restore = failStoreRead(stateId, 'loadHostState'); + try { + assert.throws(reads[featureSet], { + message: hostStateUnreadable('srv', featureSet, featureSet === 'doc' ? 'before-connect' : 'registration'), + }); + } finally { restore(); } + reads[featureSet](); + assert.deepEqual(checkpoints.getCurrentState('srv', featureSet), state, `${featureSet}: the next one reads it`); + } + // The registration that failed held no tree, so the next one registered it whole, stored tree included. + const trees = (store.getStateJson('mcpl/checkpoints') as { trees: Record }).trees; + assert.ok(trees['srv:notes'], 'the first-time feature set is in the stored trees'); + } finally { store.close(); } +})); + +test("at a runtime connect, coalescing state or checkpoint trees that can't be read fail the connect with nothing of MCPL built, and the next connect restores them", async (t) => { + const f = await fixture(); t.after(f.close); + const params = f.params('1', 'accepted_before_crash', { deferred: true, data: { k: 'v' } }); + const original = await f.send('push/event', params); + await crash(f); + // With no MCPL servers configured, the subsystem comes up at the first connect. + const framework = await AgentFramework.create({ + storePath: join(f.dir, 'store'), membrane: f.membrane.asMembrane(), + agents: [{ name: 'agent', model: 'test', systemPrompt: 'test' }], modules: [], + } as unknown as FrameworkConfig); + try { + const server = { id: 'editor', url: `ws://127.0.0.1:${f.port}`, enabledFeatureSets: ['doc'] } as McplServerConfig; + const store = framework.getStore() as unknown as JsStore; + const stored = () => ['mcpl/coalescing', 'mcpl/coalescing-recent', 'mcpl/checkpoints'].map((id) => store.getStateJson(id)); + const before = stored(); + const internals = framework as unknown as { mcplServerRegistry: unknown; pushCoalescer: unknown; checkpointManager: unknown }; + for (const [stateId, reader, says] of [ + ['mcpl/coalescing-recent', 'restorePushCoalescer', 'the coalescing receipts since the last snapshot'], + ['mcpl/coalescing', 'restorePushCoalescer', 'the coalescing receipts and batches'], + ['mcpl/checkpoints', 'loadFromStore', 'the MCPL checkpoint trees'], + ] as const) { + const restore = failStoreRead(stateId, reader); + try { + await assert.rejects(framework.connectMcplServer(server), { + message: new RegExp(`^${says} \\(${stateId}\\) could not be read: injected EIO reading ${stateId}\\. MCPL doesn't start without them`), + }); + } finally { restore(); } + assert.equal(internals.mcplServerRegistry, null, `${stateId}: nothing of the MCPL subsystem is built`); + assert.equal(internals.pushCoalescer, null, `${stateId}: no coalescer is installed`); + assert.equal(internals.checkpointManager, null, `${stateId}: no checkpoint manager is installed`); + assert.deepEqual(stored(), before, `${stateId}: the stored state is as it was`); + } + await framework.connectMcplServer(server); + const retry = await f.send('push/event', params); + assert.deepEqual(retry.result, original.result, 'the receipt survived the refused connects'); + } finally { await framework.stop(); } +}); + +test('the forms a release wrote boot as before', inStoreDir('boot-read-release-forms-', async (storePath) => { + // As every release from 0.16.0 writes them, with the head's readers: a + // resumed host's record, and the queue's cleared and pending forms. + const recovery = join(storePath, 'recovery'); + const first = await AgentFramework.create(config(storePath)); + await first.stop(); + mkdirSync(recovery, { recursive: true }); + const write = (name: string, doc: unknown) => writeFileSync(join(recovery, name), `${JSON.stringify(doc, null, 2)}\n`); + const boot = async (check: (f: AgentFramework) => void) => { + const f = await AgentFramework.create(config(storePath)); + try { check(f); } finally { await f.stop(); } + }; + write('host-mode.json', { version: 1, quiesced: false, clearedAt: 1 }); + write('deferred-writes.json', { version: 1, pending: [] }); + await boot((f) => { + assert.equal(f.getHostModeStatus().quiesced, false); + assert.equal(f.getHostModeStatus().deferredWrites, 0); + }); + write('host-mode.json', { version: 1, quiesced: true, reason: 'surgery', since: 1 }); + write('deferred-writes.json', { + version: 2, pending: [{ participant: 'user', content: [{ type: 'text', text: 'queued' }], id: 'w1', seq: 1 }], scanFrom: { agent: 0 }, + }); + await boot((f) => { + assert.equal(f.getHostModeStatus().quiesced, true); + assert.equal(f.getHostModeStatus().reason, 'surgery'); + assert.equal(f.getHostModeStatus().deferredWrites, 1); + }); +})); + +test('a store from before the recovery files boots from its slots, as before', inStoreDir('boot-read-pre-file-', async (storePath) => { + // A store from before host-mode.json and deferred-writes.json existed keeps + // both in the store: framework/state's hostMode, and a bare list of writes. + const first = await AgentFramework.create(config(storePath)); + await first.stop(); + const store = JsStore.openOrCreate({ path: storePath }); + try { + store.setStateJson('framework/state', { hostMode: { quiesced: true, reason: 'surgery', since: 1 } }); + store.setStateJson('framework/deferred-writes', [{ participant: 'user', content: [{ type: 'text', text: 'queued' }] }]); + } finally { store.close(); } + const f = await AgentFramework.create(config(storePath)); + try { + assert.equal(f.getHostModeStatus().quiesced, true); + assert.equal(f.getHostModeStatus().deferredWrites, 1); + } finally { await f.stop(); } +})); + +test('an agent named like an object property has no settings record or guard unless one is stored for it', inStoreDir('boot-read-settings-own-', async (storePath) => { + seedFrameworkState(storePath, { + agentRuntimeSettings: { other: { sameRoundThinkTextPolicy: 'private' } }, + toolResultGuards: { other: true }, + }); + const named = { ...config(storePath), agents: [{ name: 'constructor', model: 'test-model', systemPrompt: 'test' }] } as FrameworkConfig; + const f = await AgentFramework.create(named); + try { + assert.ok(f.getAgent('constructor')); + assert.equal(guardOf(f, 'constructor'), undefined, 'no inherited guard'); + } finally { await f.stop(); } +})); + +test('nothing stored reads as empty: a first boot starts serving, with no usage, deferred writes or coalescing history', async (t) => { + const f = await fixture(); t.after(f.close); + assert.equal(f.framework.getHostModeStatus().quiesced, false); + assert.equal(f.framework.getHostModeStatus().deferredWrites, 0); + assert.equal(f.framework.getSessionUsage().totals.inputTokens, 0); + const first = await f.send('push/event', f.params('1', 'first', { deferred: true })); + assert.ok(first.result, 'a push is accepted'); +}); diff --git a/test/helpers/failed-create.ts b/test/helpers/failed-create.ts new file mode 100644 index 000000000..f5283b4e8 --- /dev/null +++ b/test/helpers/failed-create.ts @@ -0,0 +1,23 @@ +/** + * A build of the framework that must fail: `AgentFramework.create()`, or a + * fixture's restart through it. A failed create() releases what it acquired + * (agent-framework #389), so a test can boot again on the same store. A build + * that succeeds is stopped before the assertion fails, so its store is + * released too. + */ +import assert from 'node:assert/strict'; +import { AgentFramework } from '../../src/index.js'; + +export async function rejectsBuild(build: () => Promise, message: RegExp, what?: string): Promise { + let caught: unknown; + let made: unknown; + try { + made = await build(); + } catch (error) { + caught = error; + } + if (made instanceof AgentFramework) await made.stop(); + assert.ok(caught instanceof Error, `the build should have failed${what ? `: ${what}` : ''}`); + assert.match(caught.message, message, what); + return caught; +} diff --git a/test/stop-with-host-operations.test.ts b/test/stop-with-host-operations.test.ts index 1685b13de..1132961b2 100644 --- a/test/stop-with-host-operations.test.ts +++ b/test/stop-with-host-operations.test.ts @@ -944,6 +944,9 @@ describe('stop() with an operator operation in flight', () => { fw.emitMcplToolDiff = () => {}; fw.mcplServerRegistry = { getAllServers: () => [], + // A connect asks whether the server is connected already, as the + // registry answers for an unknown server. + getServer: () => null, removeServer: async () => { entered = true; await held; diff --git a/test/usage-tracker.test.ts b/test/usage-tracker.test.ts index a5a8db880..6406087ef 100644 --- a/test/usage-tracker.test.ts +++ b/test/usage-tracker.test.ts @@ -154,4 +154,23 @@ describe('UsageTracker', () => { assert.ok(snap.totals.estimatedCost); assert.ok(Math.abs(snap.totals.estimatedCost.total - 0.03) < 1e-10); }); + + it("a usage without a count adds nothing for it, so the totals stay numbers and are stored as numbers", () => { + // membrane passes a Bedrock stream's output count through as reported, so one can be missing. + const { tracker } = createTracker(); + tracker.onInferenceCompleted('agent', { inputTokens: 100, outputTokens: 50, cacheCreationTokens: 10, cacheReadTokens: 20 }); + tracker.onInferenceCompleted('agent', { + inputTokens: 25, + outputTokens: undefined as unknown as number, + cacheCreationTokens: NaN, + cacheReadTokens: NaN, + }); + const stored = JSON.parse(JSON.stringify(tracker.toJSON())); + assert.equal(stored.totals.inputTokens, 125); + assert.equal(stored.totals.outputTokens, 50); + assert.equal(stored.totals.cacheCreationTokens, 10); + assert.equal(stored.totals.cacheReadTokens, 20); + assert.equal(stored.byAgent[0].usage.inputTokens, 125); + assert.equal(stored.byAgent[0].usage.outputTokens, 50); + }); });