diff --git a/CHANGELOG.md b/CHANGELOG.md index f27913fd3..a7cf31a15 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -695,6 +695,32 @@ framework. grows quadratically with session length (context-manager#148). Stores written under 0.13.x still load. Patch release from the v0.21.0 tag with this dependency bump only; the work on `main` since 0.21.0 is not included. + - **Operators of stores that learned an estimator calibration under + context-manager 0.13 or earlier** (added after the release): 0.14.0 stamps + its persisted calibration multiplier with the pricing it was learned under. + An unstamped multiplier is kept and re-stamped, unless the store holds + XML-history carriers, in which case it restarts from 1.0 and the value it + discarded is kept as `reset.discardedMultiplier` in the calibration state + (shown in `RenderStats.calibration`). Two kinds of store can then refuse + every compile with `OverBudgetError` before inference. One holds no + carriers and kept a multiplier last learned before its host moved to + context-manager 0.11.0. Its record can't show that; the host's own + history can, in its lockfile's or its deployments' history: a host that + went from a context-manager older than 0.11.0 straight to 0.14.0 can + hold one. The other holds carriers and had learned a multiplier + below 1, which its `reset.discardedMultiplier` shows. With the host + stopped, reset the first kind's multiplier once, and re-stamp the second + kind's checked value, as the **Operators** entry of context-manager + 0.14.0's changelog describes. + - **What else hosts meet with context-manager 0.14.0** (added after the + release): a preview, or any compile made with `dryRun`, leaves the context + manager's estimate, its calibration and a pending kv-unified draft's marker + indices as it found them. The context manager's state queries (`isReady`, + `getPendingWork`, `getHotContextSettings`, `previewContext` and others) + throw while its strategy's initialization is pending or has failed, where + they used to answer. A reposted sharded document renders as two bodies, so + a stored conversation holding one misses the prompt cache once after the + upgrade. ## 0.21.0 — 2026-10-03 diff --git a/changelog.d/body-removals.fixed.md b/changelog.d/body-removals.fixed.md new file mode 100644 index 000000000..d71c3d50a --- /dev/null +++ b/changelog.d/body-removals.fixed.md @@ -0,0 +1 @@ +- Removing a single message that belongs to a body, a one-shard body's included, no longer fails: context-manager's `removeMessage` refuses any message in a body, so agent-framework now removes one as a range in offline recovery's suppressions, and in a Discord suppression resumed after a crash. The host `hide` command now hides a message stored as a body (a long one, in shards) whole, with every stored copy of it (a write cut short leaves a partial copy before its retry), and its operator-log entry lists at most 50 of the ids it removed, with `removedIdsTruncated` when there were more. It's refused as busy while a surgery holds the store, and a message with no Discord id never matches one. A module's removal of a message it added, and the coalescer's, take its whole body too, and an automatic rewind whose newest message is a shard of a body takes the whole body. diff --git a/changelog.d/cm-0.14.changed.md b/changelog.d/cm-0.14.changed.md new file mode 100644 index 000000000..8e997f2ea --- /dev/null +++ b/changelog.d/cm-0.14.changed.md @@ -0,0 +1,2 @@ +- A context-maintenance pass no longer fails as a whole when one agent's readiness can't be read. context-manager 0.14, which agent-framework has required since 0.21.1, refuses its readiness queries while a strategy's initialization is pending (inside `switchBranch`, `fork`, `setStrategy` or `ContextManager.open`) or has failed, and a strategy can refuse on its own, as the autobiographical one does for a branch it hasn't reloaded. That agent's run now records the error, the agent gets its own `context-maintenance-failed` alert (`scope: 'agent'`), and the other agents run. Before, such a throw ended the pass for every agent, with one alert for the whole pass (`scope: 'pass'`). Anything else that throws while one agent's maintenance is prepared is that agent's failure in the same way. A failure that lasts, met by every pass, is recorded on each run, and logged and alerted once, then again when its text changes, every 15 minutes while it lasts (the alert webhook's cooldown for it, counted from the webhook's last post of it), or after that agent's maintenance has gone without a failure. `getContextMaintenanceSnapshot()` reports such an agent as `ready: false` with an `error`, and a strategy that can't give a progress snapshot reads as having none. +- While an operator surgery (`rollbackToMessage` or `suppressMessages`) or `switchBranch` holds the store, no maintenance pass starts and no over-budget drain kick is started, a running pass or drain kick starts no new tick, and the operator call waits for them to finish before it switches or forks the branch, so no maintenance work overlaps the move. Maintenance makes no provider call meanwhile: one made then, or one waiting at the agent's provider gate when the hold is taken, is withdrawn without being sent. That's no maintenance failure, and context-manager counts no attempt for it. A merge keeps its place in the queue. A chunk's compression logs context-manager's "Failed to compress chunk (hierarchical)" line with the withdrawal, and the chunk is queued again at that manager's next rebuild, such as its next message or compile. The operator call never waits on a provider cooldown, or on the retry turn a cooldown queues, which the hold keeps from starting. `switchBranch` can therefore move the store after the call returns, and the API server's `branch.switch` announces the move it made once it has, while `branch.current`, the branch list, `/health` and `/status` read the store's branch whenever they report one. `maintenanceTick()` answers `ran: false` meanwhile, and `resume()` refuses with `OperatorActionError` (`agent-busy`), as the other operator calls do, including when the surgery took the store while `resume()` waited. Resume once the surgery has finished. diff --git a/changelog.d/cm-0.14.fixed.md b/changelog.d/cm-0.14.fixed.md new file mode 100644 index 000000000..a4e121c59 --- /dev/null +++ b/changelog.d/cm-0.14.fixed.md @@ -0,0 +1,2 @@ +- `resume()` no longer throws when one agent's runtime settings can't be read, as when its context manager (0.14, required since 0.21.1) refuses after its strategy failed to initialize. It warns, naming the agent and the error, and proceeds without a feasibility verdict for that agent: the verdict guards against an agent that would run over budget, and an agent whose manager refuses its compiles can't run. That manager refuses the agent's turns and writes until a `switchBranch` or `setStrategy` succeeds, and maintenance reports it. +- Offline recovery (`createOfflineRecoveryBranch`) reads a sharded message's body as context-manager 0.14.0 does: a message continues the body before it only when it carries the same group id at a higher shard index. A document posted twice stores two copies under one group id, so an anchor at the end of one copy, or a suppression of one copy alone, is no longer refused. A selection inside one copy is still refused. An anchor at the end of a write that was cut short keeps that partial body and leaves out its retry, which context-manager 0.14.0 reads as a body of its own. diff --git a/src/api/server.ts b/src/api/server.ts index 0c7502229..0c6b0fed6 100644 --- a/src/api/server.ts +++ b/src/api/server.ts @@ -78,7 +78,20 @@ export class ApiServer { private wss: WebSocketServer | null = null; private httpServer: ReturnType | null = null; private clients: Set = new Set(); - private currentBranch = 'main'; + /** The store's branch as last read: what the API reports once the store has closed. */ + private lastKnownBranch = 'main'; + + /** + * The store's current branch, read from the store while it's open. A cached + * name would go stale behind any move the API server didn't make itself (a + * rollback, a host's switch), and behind a switch whose move comes after + * its call returns, as one does once running maintenance has finished. + */ + private get currentBranch(): string { + const store = this.framework.getStore(); + if (!store.isClosed()) this.lastKnownBranch = store.currentBranch().name; + return this.lastKnownBranch; + } /** Map from client to their subscription IDs */ private clientSubscriptions: Map> = new Map(); @@ -572,43 +585,31 @@ export class ApiServer { throw new Error('name is required'); } - const store = this.framework.getStore(); - const from = store.isClosed() ? this.currentBranch : store.currentBranch().name; + const from = this.currentBranch; // The framework switches the store, starts the delivery of its Discord // awareness marks behind the data-plane gate, and re-materializes the // config mount. It refuses during a live turn or once stop() has begun, - // and stop() waits for one already running. - // The store has moved when the call returns, before its promise settles - // (a refusal moves nothing). The branch read then is this switch's - // destination: another switch can move the store before this one's - // awaited steps end, so the event names what was read here. - const switching = this.framework.switchBranch(params.name); - void switching.catch(() => {}); // awaited below; handled even if the next line throws - this.followStoreBranch(); - const to = this.currentBranch; - const moved = to !== from; + // and stop() waits for one already running. It moves the store once any + // maintenance already running has finished, so the move can come after + // this call returns. The event names this switch's own move, from the + // branch read here to the one it was asked for: another switch can move + // the store again before this one continues. let switched = false; try { - await switching; + await this.framework.switchBranch(params.name); switched = true; } finally { // A later step can fail after the store has moved: clients still hear // where it went. A refusal moved nothing, so there's nothing to tell. - if (switched || moved) { - this.broadcast('branch:switched', { from, to }); + if (switched || (this.currentBranch === params.name && params.name !== from)) { + this.broadcast('branch:switched', { from, to: params.name }); } } return { switched: true }; } - /** Take the current branch from the store, while it's open. */ - private followStoreBranch(): void { - const store = this.framework.getStore(); - if (!store.isClosed()) this.currentBranch = store.currentBranch().name; - } - private async cmdBranchCurrent(): Promise<{ name: string }> { return { name: this.currentBranch }; } @@ -646,7 +647,6 @@ export class ApiServer { const result = this.framework.undoLastTurn(params.agentName); if (result.undone) { - this.currentBranch = result.toBranch!; // Marks follow behind the data-plane gate; the undo doesn't wait for Discord. this.framework.startDiscordAwarenessDelivery('undo', params.agentName); this.broadcast('branch:switched', { @@ -667,7 +667,6 @@ export class ApiServer { const result = this.framework.redo(params.agentName); if (result.redone) { - this.currentBranch = result.toBranch!; // Marks follow behind the data-plane gate; the redo doesn't wait for Discord. this.framework.startDiscordAwarenessDelivery('redo', params.agentName); this.broadcast('branch:switched', { diff --git a/src/framework.ts b/src/framework.ts index 9877d24c3..919f722fe 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -19,6 +19,14 @@ import type { StoredMessage, TokenBudget, } from '@animalabs/context-manager'; +import { + bodyGroupRun, + bodyRangeIn, + locateMessageIndex, + locateMessageIndices, + removeStoredMessage, + removeStoredMessages, +} from './stored-removal.js'; import type { FrameworkConfig, InferencePolicy, @@ -712,6 +720,23 @@ function logSeedOmissions( ); } +/** + * A maintenance provider call withdrawn because a surgery holds the store. It + * sent nothing: it was refused at the provider gate, or woken there from its + * wait. context-manager counts no attempt for it. A merge keeps its queue + * entry, so the next tick after the hold does it again. A chunk's + * compression has already left the queue: it logs "Failed to compress chunk + * (hierarchical)" with this error and records the call, and the chunk, still + * uncompressed, is queued again at that manager's next rebuild (its next + * initialization, message, sent compile or released hold). + */ +class MaintenanceWithdrawnError extends Error { + constructor(hold: { verb: string; agentName: string }) { + super(`maintenance call withdrawn: a live ${hold.verb} for ${hold.agentName} holds the store`); + this.name = 'MaintenanceWithdrawnError'; + } +} + class DiscordAwarenessAccountingError extends Error { constructor(operation: string, cause: unknown) { const detail = cause instanceof Error ? cause.message : String(cause); @@ -842,72 +867,6 @@ class DefaultInferencePolicy implements InferencePolicy { } } -/** Minimal read surface the live-surgery helpers need from a ContextManager. */ -interface SurgeryReadableCm { - getMessageCount(): number; - getMessageWindow( - offset: number, - limit: number, - opts?: { resolveBlobs?: boolean; alignToBodyGroups?: boolean }, - ): { messages: Array<{ id: unknown; bodyGroupId?: string }>; startIndex: number }; -} - -const SURGERY_SCAN_WINDOW = 500; - -/** - * Slot index of `messageId` on the active branch, or -1. Scans from the tail - * in blob-free windows: operator targets are almost always recent, and a - * full `getAllMessages()` re-inflates every attachment on the branch. - */ -function locateMessageIndex(cm: SurgeryReadableCm, messageId: string): number { - const hit = locateMessageIndices(cm, new Set([messageId])).get(messageId); - return hit ? hit.index : -1; -} - -/** Tail-first scan resolving many ids at once; stops as soon as all are found. */ -function locateMessageIndices( - cm: SurgeryReadableCm, - ids: Set, -): Map { - const found = new Map(); - const total = cm.getMessageCount(); - for (let end = total; end > 0 && found.size < ids.size; end -= SURGERY_SCAN_WINDOW) { - const start = Math.max(0, end - SURGERY_SCAN_WINDOW); - const { messages, startIndex } = cm.getMessageWindow(start, end - start, { resolveBlobs: false }); - for (let i = messages.length - 1; i >= 0; i--) { - const id = String(messages[i].id); - if (ids.has(id) && !found.has(id)) { - found.set(id, { - index: startIndex + i, - ...(messages[i].bodyGroupId ? { bodyGroupId: messages[i].bodyGroupId } : {}), - }); - } - } - } - return found; -} - -/** - * Inclusive slot range of the body-group run containing `index` (the slot - * itself when it is not sharded). Delegates to the store's own - * `alignToBodyGroups` edge walk — one contiguity assumption, not two. - */ -function bodyGroupRun( - cm: SurgeryReadableCm, - index: number, -): { from: number; to: number; fromId: string; toId: string; group: boolean } { - const win = cm.getMessageWindow(index, 1, { alignToBodyGroups: true, resolveBlobs: false }); - const from = win.startIndex; - const to = win.startIndex + win.messages.length - 1; - return { - from, - to, - fromId: String(win.messages[0].id), - toId: String(win.messages[win.messages.length - 1].id), - group: win.messages[0].bodyGroupId !== undefined, - }; -} - function describeMarkers(m: SurgeryMarkerReceipt): string { if (m.status === 'queued') return `queued:${m.queued}`; if (m.status === 'not-scheduled') return 'not-scheduled(batch retired)'; @@ -1595,6 +1554,10 @@ export class AgentFramework { private maintenanceRunId = 0; private currentMaintenanceRun: ContextMaintenanceRun | null = null; private maintenanceHistory: ContextMaintenanceRun[] = []; + /** Each agent's last reported maintenance failure, its whole text and when + * it was alerted, until its maintenance goes without one + * (reportMaintenanceFailure). */ + private maintenanceFailureReported = new Map(); private providerGates: Map = new Map(); private providerAccelerationCooldowns: Map = new Map(); private providerAccelerationRecoveries: Map = new Map(); @@ -1754,6 +1717,8 @@ export class AgentFramework { * tick counts in the success log stop meaning anything. One kick per agent * at a time; cleared when the kick settles (success or failure). */ private overBudgetDrainInFlight: Set = new Set(); + /** The over-budget drain kicks running, which a branch move waits for as it waits for a maintenance pass. */ + private overBudgetDrains = new Set>(); /** Per-agent current rewind episode: the single consolidated marker's id and * how many turns have been shed so far. One marker per episode, updated in * place (see updateRewindMarker); cleared when the episode ends. */ @@ -2871,13 +2836,20 @@ export class AgentFramework { private async withAuxiliaryAdmission(agentName: string, run: () => Promise, model?: string): Promise { const gate = this.providerGate(agentName); this.refuseHeldProviderModel(agentName, model); - if (this.providerGateBlocked(agentName)) { + this.withdrawMaintenanceUnderSurgery(); + // A parked call is woken when the gate opens, when stop() closes + // admission, or when a surgery takes the store; one woken while the gate + // is still closed parks again. + while (this.providerGateBlocked(agentName)) { gate.deferredAuxiliary++; try { await new Promise((resolve) => gate.auxiliaryWaiters.push(resolve)); } finally { gate.deferredAuxiliary = Math.max(0, gate.deferredAuxiliary - 1); } + if (this.providerAdmissionClosed) break; + // A surgery that took the store while this call was parked withdraws it. + this.withdrawMaintenanceUnderSurgery(); } if (this.providerAdmissionClosed) { throw new Error(`Provider admission closed while stopping (${agentName})`); @@ -2903,6 +2875,16 @@ export class AgentFramework { if (gate.auxiliaryInFlight === 0) for (const resolve of gate.idleWaiters.splice(0)) resolve(); } } + /** + * Maintenance makes no provider call while a surgery holds the store. The + * surgery waits for the maintenance work that is running before it moves + * the branch, so a call that parked at the gate would make it wait on + * whatever reopens the gate: the end of a cooldown, or the retry turn a + * cooldown queues, which the hold itself keeps from starting. + */ + private withdrawMaintenanceUnderSurgery(): void { + if (this.surgeryHold) throw new MaintenanceWithdrawnError(this.surgeryHold); + } private auxiliaryMembraneFor(agentName: string): Membrane { return this.membraneAdmitting((run, model) => this.withAuxiliaryAdmission(agentName, run, model)); } @@ -3390,7 +3372,32 @@ export class AgentFramework { }); } + /** resume()'s refusal while a surgery holds the store. */ + private refuseResumeUnderSurgery(): void { + if (!this.surgeryHold) return; + throw new OperatorActionError( + 'agent-busy', + `Cannot resume: a live ${this.surgeryHold.verb} for ${this.surgeryHold.agentName} holds the store — retry when it completes`, + ); + } + + /** Whether maintenance work is running: a pass, or an over-budget drain kick. */ + private maintenanceRunning(): boolean { + return this.maintenancePass !== null || this.overBudgetDrains.size > 0; + } + + /** Wait for the maintenance work that is running, whatever its outcome. */ + private async settleMaintenancePass(): Promise { + await Promise.allSettled([...(this.maintenancePass ? [this.maintenancePass] : []), ...this.overBudgetDrains]); + } + private async runQueuedMaintenance(): Promise { + // A surgery holds the store while it switches or forks a branch. No pass + // starts while it's held (the timer's next pass after it runs as usual), + // and a running pass starts no new tick. The surgery waits for that pass + // to finish before it moves the branch, so no maintenance work overlaps + // the move, and a failure a pass meets is never the surgery's doing. + if (this.surgeryHold) return; // A guard whose agent has left the map gets no activation to retry its // queue; retry it here. It leaves busyGuards once nothing of it is left. const live = new Set([...this.agents.values()].map((agent) => agent.toolResultGuard)); @@ -3402,32 +3409,66 @@ export class AgentFramework { console.error(`[tool-result-guard] agent=${name} storage retry failed during maintenance:`, error); } } - const queued = [...this.agents.values()].flatMap((agent) => { + for (const name of this.maintenanceFailureReported.keys()) { + if (!this.agents.has(name)) this.maintenanceFailureReported.delete(name); + } + type Queued = { agent: Agent; record: ContextMaintenanceAgentRun } + & ({ failed: true } | { failed: false; cm: ContextManager }); + const queued = [...this.agents.values()].flatMap((agent): Queued[] => { try { agent.toolResultGuard.flushUnrecorded(); } catch (error) { console.error(`[tool-result-guard] agent=${agent.name} storage retry failed during maintenance:`, error); } - const cm = agent.getContextManager(); - const tools = this.compressionToolsForAgent(agent.name); - cm.setToolDefinitions(tools); - if (this.providerGateBlocked(agent.name)) return []; - if (cm.isReady()) return []; - const pending = cm.getPendingWork()?.description; + // Whatever throws while one agent's maintenance is prepared is that + // agent's failure, reported for it alone, while the others run. + const failedAgent = (error: unknown): Queued[] => { + const record: ContextMaintenanceAgentRun = { + agentName: agent.name, + startedAt: Date.now(), + ticks: 0, + readyBefore: false, + }; + record.finishedAt = record.startedAt; + this.reportMaintenanceFailure( + agent.name, + record, + error instanceof Error ? String(error.message) : failureDescription(error), + ); + return [{ agent, record, failed: true }]; + }; + let cm: ContextManager; + let state: ReturnType; + try { + cm = agent.getContextManager(); + const tools = this.compressionToolsForAgent(agent.name); + cm.setToolDefinitions(tools); + if (this.providerGateBlocked(agent.name)) return []; + state = this.contextReadiness(cm); + } catch (error) { + return failedAgent(error); + } + if ('ready' in state && state.ready) { + this.maintenanceFailureReported.delete(agent.name); + return []; + } const progress = this.contextProgress(cm); const record: ContextMaintenanceAgentRun = { agentName: agent.name, startedAt: Date.now(), ticks: 0, readyBefore: false, - ...(pending ? { pendingBefore: pending } : {}), + ...('pending' in state && state.pending ? { pendingBefore: state.pending } : {}), ...(progress ? { progressBefore: progress } : {}), }; - return [{ - agent, - cm, - record, - }]; + if ('error' in state) { + // Its readiness can't be read: that agent's failure, reported for it + // alone, while the others run. + record.finishedAt = Date.now(); + this.reportMaintenanceFailure(agent.name, record, state.error); + return [{ agent, record, failed: true }]; + } + return [{ agent, cm, record, failed: false }]; }); if (queued.length === 0) return; @@ -3439,41 +3480,54 @@ export class AgentFramework { this.currentMaintenanceRun = run; try { - await Promise.all(queued.map(async ({ agent, cm, record }) => { + const running = queued.filter((item): item is Extract => !item.failed); + // Settled, not all: the pass ends only when every agent's work has. + const outcomes = await Promise.allSettled(running.map(async ({ agent, cm, record }) => { + // A tick whose provider call a surgery withdrew ends this agent's + // work for the pass, as the hold would have before its next tick: it + // is no failure, and what the failure reports remember stays as it is. + let withdrawn = false; try { for ( let i = 0; - i < MAINTENANCE_TICKS_PER_PASS && this.running && !cm.isReady(); + i < MAINTENANCE_TICKS_PER_PASS && this.running && !this.surgeryHold && !cm.isReady(); i++ ) { await cm.tick(); record.ticks++; } } catch (error) { - const wholeError = error instanceof Error ? String(error.message) : failureDescription(error); - record.error = boundFailureText(wholeError); - console.error(`[context-maintenance] agent=${agent.name} failed:`, record.error); - this.opsAlert( - 'context-maintenance-failed', - agent.name, - wholeError, - { - data: { - scope: 'agent', - ...(record.pendingBefore ? { pending: record.pendingBefore } : {}), - ...(record.progressBefore ? { progress: record.progressBefore } : {}), - }, - }, - ); + if (error instanceof MaintenanceWithdrawnError) { + withdrawn = true; + console.error(`[context-maintenance] agent=${agent.name} stopped: ${error.message}`); + } else { + this.reportMaintenanceFailure( + agent.name, + record, + error instanceof Error ? String(error.message) : failureDescription(error), + ); + } } finally { - const pending = cm.getPendingWork()?.description; record.finishedAt = Date.now(); - record.readyAfter = cm.isReady(); - if (pending) record.pendingAfter = pending; + const state = this.contextReadiness(cm); + if ('error' in state) { + // A run records an agent's first failure: a readiness refusal + // after a tick that failed is read again by the next pass. + if (record.error === undefined && !withdrawn) this.reportMaintenanceFailure(agent.name, record, state.error); + } else { + record.readyAfter = state.ready; + if (state.pending) record.pendingAfter = state.pending; + if (record.error === undefined && !withdrawn) this.maintenanceFailureReported.delete(agent.name); + } const progress = this.contextProgress(cm); if (progress) record.progressAfter = progress; } })); + for (const outcome of outcomes) { + if (outcome.status === 'rejected') { + console.error('[context-maintenance] reporting an agent\'s failure failed:', describeFailureWithFrames(outcome.reason)); + } + } } finally { run.finishedAt = Date.now(); run.agents.sort((a, b) => a.agentName.localeCompare(b.agentName)); @@ -3483,15 +3537,78 @@ export class AgentFramework { } } + /** + * One agent's maintenance failure: recorded on its run, bounded as every + * failure sink is (`record.error`), and logged and alerted for that agent + * with the whole text. A failure that lasts, such as a manager whose + * strategy failed to initialize, meets every pass, a few seconds apart: + * each run records it, and it is logged and alerted again only when its + * whole text changes, once opsAlert's cooldown for it has passed since the + * webhook last posted it (so the webhook, which posts a kind at most once + * per cooldown, hears a failure that lasts), or after a pass in which that + * agent's maintenance went without a failure (maintenanceFailureReported). + */ + private reportMaintenanceFailure(agentName: string, record: ContextMaintenanceAgentRun, wholeError: string): void { + record.error = boundFailureText(wholeError); + const reported = this.maintenanceFailureReported.get(agentName); + const now = Date.now(); + if (reported && reported.text === wholeError) { + // The same failure again is alerted once the webhook's cooldown for it + // has passed, counted from the webhook's last post of it: a report of + // changed text that the webhook didn't post doesn't push the reminder + // back. With no webhook, it's counted from this failure's last report. + const since = this.opsAlertLastSent.get(`${agentName}:context-maintenance-failed`) ?? reported.at; + if (now - since < this.opsAlertCooldownMs) return; + } + this.maintenanceFailureReported.set(agentName, { text: wholeError, at: now }); + console.error(`[context-maintenance] agent=${agentName} failed:`, record.error); + this.opsAlert( + 'context-maintenance-failed', + agentName, + wholeError, + { + data: { + scope: 'agent', + ...(record.pendingBefore ? { pending: record.pendingBefore } : {}), + ...(record.progressBefore ? { progress: record.progressBefore } : {}), + }, + }, + ); + } + + /** + * An agent's context readiness, or the error that reading it threw. The + * context manager refuses while its strategy's initialization is pending + * or has failed (context-manager 0.14), and a strategy can refuse on its + * own, as the autobiographical one does after a store-level branch switch + * it hasn't reloaded for. Every such throw is that agent's failure. + */ + private contextReadiness( + cm: ContextManager, + ): { ready: boolean; pending?: string } | { error: string } { + try { + const ready = cm.isReady(); + const pending = ready ? undefined : cm.getPendingWork()?.description; + return pending ? { ready, pending } : { ready }; + } catch (error) { + return { error: error instanceof Error ? String(error.message) : failureDescription(error) }; + } + } + + /** The strategy's progress snapshot, for diagnostics: one it can't give reads as none. */ private contextProgress(cm: ContextManager): Record | undefined { - const strategy = cm.getStrategy() as { - getProgressSnapshot?: () => unknown; - }; - if (typeof strategy.getProgressSnapshot !== 'function') return undefined; - const progress = strategy.getProgressSnapshot(); - return progress && typeof progress === 'object' - ? progress as Record - : undefined; + try { + const strategy = cm.getStrategy() as { + getProgressSnapshot?: () => unknown; + }; + if (typeof strategy.getProgressSnapshot !== 'function') return undefined; + const progress = strategy.getProgressSnapshot(); + return progress && typeof progress === 'object' + ? progress as Record + : undefined; + } catch { + return undefined; + } } /** @@ -4542,22 +4659,41 @@ export class AgentFramework { private async runResume(opts?: { force?: boolean }): Promise { if (!this.quiesced) return this.getHostModeStatus(); + // A surgery switches or forks the branch while it holds the store, and a + // manager whose initialization that left pending refuses its settings + // read: resume once it's done, as the other operator calls do. Checked + // again after the waits below, since a surgery can take the store while + // resume waits for a pass or a preview. + this.refuseResumeUnderSurgery(); // The canonical operator flow is `maintain; resume`, and a timer-driven // pass may still be mid-flight: previewContext refuses to overlap one // ("already running"), which used to degrade the gate to "no opinion". // Wait for the pass first, then retry a few times for the case where a // fresh pass started in between. - if (this.maintenancePass) { - await this.maintenancePass.catch(() => {}); - } + if (this.maintenanceRunning()) await this.settleMaintenancePass(); const failing: Array<{ agentName: string; preview: RuntimeSettingsPreview }> = []; for (const agentName of this.agents.keys()) { - let preview = this.previewAgentRuntimeSettings(agentName); - for (let attempt = 0; attempt < 8 && !preview.available && preview.reason === 'preview_in_flight'; attempt++) { - await new Promise((resolve) => setTimeout(resolve, 250)); - if (this.maintenancePass) await this.maintenancePass.catch(() => {}); + let preview: RuntimeSettingsPreview; + try { preview = this.previewAgentRuntimeSettings(agentName); + for (let attempt = 0; attempt < 8 && !preview.available && preview.reason === 'preview_in_flight'; attempt++) { + await new Promise((resolve) => setTimeout(resolve, 250)); + if (this.maintenancePass) await this.maintenancePass.catch(() => {}); + preview = this.previewAgentRuntimeSettings(agentName); + } + } catch (error) { + // The agent's settings can't be read, as when its context manager's + // strategy failed to initialize: that manager refuses its compiles and + // writes until a switchBranch or setStrategy succeeds, and maintenance + // reports it. The gate guards against an agent that would run over + // budget, and this one can't run, so it gets no verdict, as a preview + // that isn't available gets none, and resume goes on for the others. + console.warn( + `[host-mode] resume: ${agentName}'s settings can't be read ` + + `(${error instanceof Error ? error.message : String(error)}) — proceeding without a verdict for it`, + ); + continue; } if (preview.available && preview.effective && !preview.effective.fits) { failing.push({ agentName, preview }); @@ -4568,6 +4704,7 @@ export class AgentFramework { ); } } + this.refuseResumeUnderSurgery(); // stop() closes the queue first. If it has begun, before this call or // during the waits above, leave the host quiesced, as it will boot, // rather than reopen what stop() is closing. @@ -4953,6 +5090,13 @@ export class AgentFramework { console.warn('[host-mode] maintenanceTick: framework is not running — no pass executed'); return { ...this.getContextMaintenanceSnapshot(), ran: false }; } + if (this.surgeryHold) { + // No pass starts while a surgery holds the store (runQueuedMaintenance). + console.warn( + `[host-mode] maintenanceTick: a ${this.surgeryHold.verb} for ${this.surgeryHold.agentName} holds the store — no pass executed`, + ); + return { ...this.getContextMaintenanceSnapshot(), ran: false }; + } this.startQueuedMaintenance(); await this.maintenancePass; return { ...this.getContextMaintenanceSnapshot(), ran: true }; @@ -4962,12 +5106,13 @@ export class AgentFramework { getContextMaintenanceSnapshot(): ContextMaintenanceSnapshot { const agents = [...this.agents.values()].map((agent) => { const cm = agent.getContextManager(); - const pending = cm.getPendingWork()?.description; + const state = this.contextReadiness(cm); const progress = this.contextProgress(cm); return { agentName: agent.name, - ready: cm.isReady(), - ...(pending ? { pending } : {}), + ...('error' in state + ? { ready: false, error: boundFailureText(state.error) } + : { ready: state.ready, ...(state.pending ? { pending: state.pending } : {}) }), ...(progress ? { progress } : {}), }; }); @@ -6388,27 +6533,41 @@ export class AgentFramework { idx = i; break; } if (idx < 0) return null; + // The newest message can be the last shard of a body (a document stored + // in shards). context-manager removes a shard only with the rest of its + // body, so the rewind takes the whole body. + const first = bodyRangeIn(all, idx).from; + const members = all.slice(first, idx + 1); const msg = all[idx]; const md = (msg.metadata ?? {}) as { messageId?: unknown; channelId?: unknown; system?: unknown; }; - const content = Array.isArray(msg.content) ? msg.content : []; + const content = members.flatMap((m) => (Array.isArray(m.content) ? m.content : [])); const toolBlocks = content.filter((b) => typeOf(b) === 'tool_result'); const images = content.filter((b) => typeOf(b) === 'image').length; const textLen = content .filter((b) => typeOf(b) === 'text') .reduce((n, b) => n + String((b as { text?: string }).text ?? '').length, 0); - // Determine the complete exchange to remove. - const removedIds: MessageId[] = [msg.id]; + // Determine the complete exchange to remove, newest first. + const removedIds: MessageId[] = members.map((m) => m.id).reverse(); + // Where the exchange starts: this body, or the tool_use turn paired with it. + let from = first; let kind: RewindRecord['kind']; let descriptor: string; if (toolBlocks.length > 0) { kind = 'tool'; - // Pair the tool_result with the tool_use assistant turn right before it. - if (idx - 1 >= 0 && hasBlock(all[idx - 1], 'tool_use')) { - removedIds.push(all[idx - 1].id); + // Pair the tool_result with the tool_use assistant turn right before + // it, the whole of it: a turn stored as a body carries its tool_use on + // its first shard, not its last. + if (first - 1 >= 0) { + const turn = bodyRangeIn(all, first - 1); + const turnMembers = all.slice(turn.from, turn.to + 1); + if (turnMembers.some((m) => hasBlock(m, 'tool_use'))) { + from = turn.from; + removedIds.push(...turnMembers.map((m) => m.id).reverse()); + } } const sz = toolBlocks.reduce( (n, b) => n + String((b as { content?: unknown }).content ?? '').length, 0); @@ -6427,7 +6586,8 @@ export class AgentFramework { ? { channelId: String(md.channelId), messageId: String(md.messageId) } : undefined; - for (const id of removedIds) cm.removeMessage(id); + // One removal for the whole exchange, so it is never half applied. + removeStoredMessages(cm, all[from], msg); return { kind, descriptor, removedIds, discordRef }; } @@ -6802,28 +6962,57 @@ export class AgentFramework { if (agent.state.status !== 'idle') { return { ok: false, error: `Cannot hide while agent is ${agent.state.status}` }; } + // A surgery switches or forks the branch while it holds the store, so + // a removal now would land on the branch it's leaving, or meet a + // manager mid-switch: refused as busy, as the other operator calls are. + if (this.surgeryHold) { + return { + ok: false, + error: `Cannot hide: a live ${this.surgeryHold.verb} for ${this.surgeryHold.agentName} holds the store — retry when it completes`, + }; + } if (!params.fromMessageId) { return { ok: false, error: 'hide: fromMessageId is required' }; } const cm = agent.getContextManager(); const all = cm.getAllMessages(); - const byDiscordId = (did: string) => - all.findIndex((m) => String((m.metadata as { messageId?: unknown } | undefined)?.messageId) === String(did)); + // Only a message that carries a Discord id matches: a missing one + // would otherwise read as the text "undefined". + const discordIdOf = (m: StoredMessage) => { + const id = (m.metadata as { messageId?: unknown } | undefined)?.messageId; + return id === undefined || id === null || id === '' ? undefined : String(id); + }; + const byDiscordId = (did: string) => all.findIndex((m) => discordIdOf(m) === String(did)); // Collect Discord refs of every message in [lo, hi] that carries one, // so the surface can mark them. channelId here is whatever the ingest // path stored (raw id or "discord:guild:channel"); the surface // normalizes it. - const refsIn = (lo: number, hi: number) => { + // Each Discord message once, though every shard of a body carries it. + const refsIn = (runs: Array<{ from: number; to: number }>) => { const refs: Array<{ channelId: string; messageId: string }> = []; - for (let i = lo; i <= hi; i++) { - const md = all[i].metadata as { messageId?: unknown; channelId?: unknown } | undefined; - if (md?.messageId && md?.channelId) { + const seen = new Set(); + for (const { from, to } of runs) { + for (let i = from; i <= to; i++) { + const md = all[i].metadata as { messageId?: unknown; channelId?: unknown } | undefined; + if (!md?.messageId || !md?.channelId) continue; + const key = `${String(md.channelId)}\u0000${String(md.messageId)}`; + if (seen.has(key)) continue; + seen.add(key); refs.push({ channelId: String(md.channelId), messageId: String(md.messageId) }); } } return refs; }; + // How many stored messages the runs hold, a body (a long message, in + // shards) counting once. + const messagesIn = (runs: Array<{ from: number; to: number }>) => { + let count = 0; + for (const { from, to } of runs) { + for (let i = from; i <= to; i = bodyRangeIn(all, i).to + 1) count++; + } + return count; + }; const fromIdx = byDiscordId(params.fromMessageId); if (fromIdx < 0) { @@ -6839,12 +7028,16 @@ export class AgentFramework { if (toIdx < 0) { return { ok: false, error: `Message ${params.toMessageId} is not an addressable message in context.` }; } - const [lo, hi] = fromIdx <= toIdx ? [fromIdx, toIdx] : [toIdx, fromIdx]; - const refs = refsIn(lo, hi); + const [first, last] = fromIdx <= toIdx ? [fromIdx, toIdx] : [toIdx, fromIdx]; + // A message stored as a body (a long one, in shards) is hidden whole. + const lo = bodyRangeIn(all, first).from; + const hi = bodyRangeIn(all, last).to; + const refs = refsIn([{ from: lo, to: hi }]); + const hidden = messagesIn([{ from: lo, to: hi }]); const rangeIds = all.slice(lo, hi + 1).map((m) => String(m.id)); - cm.removeMessages(all[lo].id, all[hi].id); + removeStoredMessages(cm, all[lo], all[hi]); console.error( - `[host-command] hide agent=${agentName} range removed=${hi - lo + 1} ` + + `[host-command] hide agent=${agentName} range removed=${hidden} ` + `(${params.fromMessageId}..${params.toMessageId}) by=${params.requesterName ?? params.requesterId ?? 'unknown'} (server=${serverId})`, ); const loggedRange = capIds(rangeIds); @@ -6855,35 +7048,55 @@ export class AgentFramework { params: { fromMessageId: params.fromMessageId, toMessageId: params.toMessageId }, result: { branch: this.store.currentBranch().name, - hidden: hi - lo + 1, + hidden, removedIds: loggedRange.ids, ...(loggedRange.truncated ? { removedIdsTruncated: true } : {}), }, }); return { ok: true, - hidden: hi - lo + 1, + hidden, hiddenRefs: refs, lastVisible: await this.lastVisiblePreview(agentName), }; } - const refs = refsIn(fromIdx, fromIdx); - const hiddenId = String(all[fromIdx].id); - cm.removeMessage(all[fromIdx].id); + // A message stored as a body (a long one, in shards) is hidden whole, + // and so is every copy of it: a write cut short leaves a partial body + // before its retry, both carrying the Discord id. + const runs: Array<{ from: number; to: number }> = []; + all.forEach((m, i) => { + if (discordIdOf(m) !== String(params.fromMessageId)) return; + const { from, to } = bodyRangeIn(all, i); + const last = runs[runs.length - 1]; + if (last && from <= last.to + 1) last.to = Math.max(last.to, to); + else runs.push({ from, to }); + }); + const refs = refsIn(runs); + const hidden = messagesIn(runs); + const hiddenIds = runs.flatMap(({ from, to }) => all.slice(from, to + 1).map((m) => String(m.id))); + for (const { from, to } of runs) removeStoredMessages(cm, all[from], all[to]); console.error( - `[host-command] hide agent=${agentName} removed=1 (${params.fromMessageId}) ` + + `[host-command] hide agent=${agentName} removed=${hidden} (${params.fromMessageId}) ` + `by=${params.requesterName ?? params.requesterId ?? 'unknown'} (server=${serverId})`, ); + // Every shard of every copy goes, so the ids are bounded as the + // range form's are. + const loggedIds = capIds(hiddenIds); this.recordOperatorAction({ kind: 'hide', agent: agentName, requester: hostCommandRequester(serverId, params), params: { fromMessageId: params.fromMessageId }, - result: { branch: this.store.currentBranch().name, hidden: 1, removedIds: [hiddenId] }, + result: { + branch: this.store.currentBranch().name, + hidden, + removedIds: loggedIds.ids, + ...(loggedIds.truncated ? { removedIdsTruncated: true } : {}), + }, }); return { ok: true, - hidden: 1, + hidden, hiddenRefs: refs, lastVisible: await this.lastVisiblePreview(agentName), }; @@ -7286,6 +7499,8 @@ export class AgentFramework { this.refuseOperatorAfterStop('switch branch'); const release = this.reserveStoreForSurgery('switch branch', 'every agent'); try { + // A maintenance tick already running finishes before the switch. + if (this.maintenanceRunning()) await this.settleMaintenancePass(); this.store.switchBranch(name); // Reversible Discord awareness markers are part of the branch // projection. Their delivery starts behind the data-plane gate, which @@ -7544,6 +7759,9 @@ export class AgentFramework { // arrivals during the awaited switch (see surgeryHold). const hold = { verb, agentName, since: Date.now() }; this.surgeryHold = hold; + // Maintenance calls parked at a provider gate wake, find the hold and are + // withdrawn, so the wait for running maintenance can't wait on them. + for (const gate of this.providerGates.values()) for (const resolve of gate.auxiliaryWaiters.splice(0)) resolve(); // stop() waits for the release (see stopOwned): a surgery admitted before // stop() began finishes, flush included, before the store closes. let settle!: () => void; @@ -7683,6 +7901,8 @@ export class AgentFramework { let branchName: string; let markers: SurgeryMarkerReceipt; try { + // A maintenance tick already running finishes before the switch. + if (this.maintenanceRunning()) await this.settleMaintenancePass(); // Prepare the external side effect before switching Chronicle. If the // process dies after the switch but before activate(), startup // promotes this batch by matching targetBranch to the active branch. @@ -7851,6 +8071,8 @@ export class AgentFramework { let createdBranch: string; let markers: SurgeryMarkerReceipt; try { + // A maintenance tick already running finishes before the fork. + if (this.maintenanceRunning()) await this.settleMaintenancePass(); const markerBatch = this.discordAwarenessOutbox?.prepare({ agentName, sourceBranch, @@ -9986,7 +10208,7 @@ export class AgentFramework { const agent = this.agents.get(p.agent); if (!agent) return false; try { - agent.getContextManager().removeMessage(p.messageId); + removeStoredMessage(agent.getContextManager(), p.messageId as MessageId); this.emitTrace({ type: 'message:removed', messageId: p.messageId, source: 'mcpl:coalescing' }); return true; } catch (err) { @@ -15018,7 +15240,7 @@ export class AgentFramework { if (!agent) { throw new Error('No agents configured'); } - agent.getContextManager().removeMessage(id); + removeStoredMessage(agent.getContextManager(), id); } private getMessage(id: MessageId): StoredMessage | null { @@ -15503,23 +15725,34 @@ export class AgentFramework { { skipLog: true }, ); } - if (agent && overBudget && !this.overBudgetDrainInFlight.has(agentName)) { + if (agent && overBudget && !this.overBudgetDrainInFlight.has(agentName) && !this.surgeryHold) { this.overBudgetDrainInFlight.add(agentName); - void (async () => { + // Like a maintenance pass: it starts no tick while a surgery holds the + // store, and a branch move waits for the tick in flight. + const drain = (async () => { let ticks = 0; try { const cm = agent.getContextManager(); - while (ticks < 8) { + while (ticks < 8 && !this.surgeryHold) { await cm.tick(); ticks++; } - console.error(`[inference-failed] drain kicked for ${agentName} (OverBudget breaker, ${ticks} ticks)`); + console.error( + `[inference-failed] drain kicked for ${agentName} (OverBudget breaker, ${ticks} ticks` + + `${ticks < 8 ? ', stopped while a surgery holds the store' : ''})`, + ); } catch (err) { - console.error(`[inference-failed] drain kick failed for ${agentName} after ${ticks} ticks: ${describeFailureWithFrames(err)}`); + if (err instanceof MaintenanceWithdrawnError) { + console.error(`[inference-failed] drain kicked for ${agentName} (OverBudget breaker, ${ticks} ticks, stopped: ${err.message})`); + } else { + console.error(`[inference-failed] drain kick failed for ${agentName} after ${ticks} ticks: ${describeFailureWithFrames(err)}`); + } } finally { this.overBudgetDrainInFlight.delete(agentName); } })(); + this.overBudgetDrains.add(drain); + void drain.finally(() => { this.overBudgetDrains.delete(drain); }); } // (3) Hard-down escalation on repeated identical failure. @@ -15893,8 +16126,7 @@ export class AgentFramework { `${interval.fromId}..${interval.toId} is only partially present`, ); } - if (interval.fromId === interval.toId) cm.removeMessage(from.id); - else cm.removeMessages(from.id, to.id); + removeStoredMessages(cm, from, to); } this.discordAwarenessOutbox.activate(batch.id); console.error(`[discord-awareness] resumed suppression batch ${batch.id}`); diff --git a/src/recovery/offline-branch.ts b/src/recovery/offline-branch.ts index 6c2a202c8..49137f1af 100644 --- a/src/recovery/offline-branch.ts +++ b/src/recovery/offline-branch.ts @@ -1,5 +1,6 @@ import { JsStore } from '@animalabs/chronicle'; -import { ContextManager } from '@animalabs/context-manager'; +import { ContextManager, continuesBody, type BodyShard } from '@animalabs/context-manager'; +import { removeStoredInterval } from '../stored-removal.js'; import { DEFAULT_DISCORD_AWARENESS_EMOJI, DiscordAwarenessOutbox, @@ -202,11 +203,12 @@ export async function createOfflineRecoveryBranch( const createdBranch = contextManager.branchAt(target.id, targetBranch); try { await contextManager.switchBranch(createdBranch); - // Work from newest to oldest so earlier redactions never perturb the - // live positions used by later range endpoints. + // Each interval is removed by its messages' ids, so the order doesn't + // change the result. removeStoredInterval takes a single message outside + // any body through removeMessage, and a shard or a longer interval as a + // range, which refuses one that splits a body. for (const interval of [...suppression.intervals].reverse()) { - if (interval.fromId === interval.toId) contextManager.removeMessage(interval.fromId); - else contextManager.removeMessages(interval.fromId, interval.toId); + removeStoredInterval(contextManager, interval.fromId, interval.toId); } if (batch) outbox.activate(batch.id); } catch (error) { @@ -299,14 +301,15 @@ function validateAnchorBoundary( total: number, target: StoredWindowMessage, ): void { - const groupId = (target as StoredWindowMessage & { bodyGroupId?: string }).bodyGroupId; - if (!groupId || targetIndex + 1 >= total) return; + if (targetIndex + 1 >= total) return; const next = contextManager.getMessageWindow(targetIndex + 1, 1, { resolveBlobs: false, - }).messages[0] as (StoredWindowMessage & { bodyGroupId?: string }) | undefined; - if (next?.bodyGroupId === groupId) { + }).messages[0]; + // A body continues only into a later shard of the same ingestion: a + // reposted document shares its group id but starts a body of its own. + if (continuesBody(target as BodyShard, next as BodyShard | undefined)) { throw new Error( - `Recovery anchor ${String(target.id)} splits body group ${groupId}; choose its final shard.`, + `Recovery anchor ${String(target.id)} splits body group ${(target as BodyShard).bodyGroupId}; choose its final shard.`, ); } } @@ -321,7 +324,9 @@ function validateRemovalIntegrity( index >= interval.start && index <= interval.end); const toolUses = new Map(); const toolResults = new Map(); - const groupMembership = new Map(); + // The message before, in store order: a selection splits a body when a + // message continuing the one before it lands on the other side. + let previous: { message: BodyShard; removed: boolean } | null = null; for (let offset = 0; offset < total; offset += RECOVERY_SCAN_WINDOW) { const messages = contextManager.getMessageWindow( @@ -332,16 +337,14 @@ function validateRemovalIntegrity( for (let i = 0; i < messages.length; i++) { const index = offset + i; const isRemoved = removed(index); - const message = messages[i] as StoredWindowMessage & { bodyGroupId?: string }; - if (message.bodyGroupId) { - const prior = groupMembership.get(message.bodyGroupId); - if (prior !== undefined && prior !== isRemoved) { - throw new Error( - `Recovery selection splits body group ${message.bodyGroupId}; adjust the anchor or suppression boundary.`, - ); - } - groupMembership.set(message.bodyGroupId, isRemoved); + const message = messages[i]; + const shard = message as BodyShard; + if (previous && previous.removed !== isRemoved && continuesBody(previous.message, shard)) { + throw new Error( + `Recovery selection splits body group ${shard.bodyGroupId}; adjust the anchor or suppression boundary.`, + ); } + previous = { message: shard, removed: isRemoved }; for (const rawBlock of message.content) { const block = rawBlock as { type?: unknown; id?: unknown; toolUseId?: unknown }; if (block.type === 'tool_use' && typeof block.id === 'string') { diff --git a/src/stored-removal.ts b/src/stored-removal.ts new file mode 100644 index 000000000..c0b73d8d0 --- /dev/null +++ b/src/stored-removal.ts @@ -0,0 +1,126 @@ +import { bodyBoundsIn, type BodyShard, type ContextManager, type MessageId } from '@animalabs/context-manager'; + +type StoredLike = BodyShard & { id: MessageId }; +type Removals = Pick; + +/** Minimal read surface the live-surgery helpers need from a ContextManager. */ +export interface SurgeryReadableCm { + getMessageCount(): number; + getMessageWindow( + offset: number, + limit: number, + opts?: { resolveBlobs?: boolean; alignToBodyGroups?: boolean }, + ): { messages: Array<{ id: unknown; bodyGroupId?: string }>; startIndex: number }; +} + +const SURGERY_SCAN_WINDOW = 500; + +/** + * Slot index of `messageId` on the active branch, or -1. Scans from the tail + * in blob-free windows: operator targets are almost always recent, and a + * full `getAllMessages()` re-inflates every attachment on the branch. + */ +export function locateMessageIndex(cm: SurgeryReadableCm, messageId: string): number { + const hit = locateMessageIndices(cm, new Set([messageId])).get(messageId); + return hit ? hit.index : -1; +} + +/** Tail-first scan resolving many ids at once; stops as soon as all are found. */ +export function locateMessageIndices( + cm: SurgeryReadableCm, + ids: Set, +): Map { + const found = new Map(); + const total = cm.getMessageCount(); + for (let end = total; end > 0 && found.size < ids.size; end -= SURGERY_SCAN_WINDOW) { + const start = Math.max(0, end - SURGERY_SCAN_WINDOW); + const { messages, startIndex } = cm.getMessageWindow(start, end - start, { resolveBlobs: false }); + for (let i = messages.length - 1; i >= 0; i--) { + const id = String(messages[i].id); + if (ids.has(id) && !found.has(id)) { + found.set(id, { + index: startIndex + i, + ...(messages[i].bodyGroupId ? { bodyGroupId: messages[i].bodyGroupId } : {}), + }); + } + } + } + return found; +} + +/** + * Inclusive slot range of the body-group run containing `index` (the slot + * itself when it is not sharded). Delegates to the store's own + * `alignToBodyGroups` edge walk — one contiguity assumption, not two. + */ +export function bodyGroupRun( + cm: SurgeryReadableCm, + index: number, +): { from: number; to: number; fromId: string; toId: string; group: boolean } { + const win = cm.getMessageWindow(index, 1, { alignToBodyGroups: true, resolveBlobs: false }); + const from = win.startIndex; + const to = win.startIndex + win.messages.length - 1; + return { + from, + to, + fromId: String(win.messages[0].id), + toId: String(win.messages[win.messages.length - 1].id), + group: win.messages[0].bodyGroupId !== undefined, + }; +} + +/** + * Remove the stored messages `first`..`last` (inclusive, in store order). + * + * A single message outside any body goes through `removeMessage`. + * context-manager's `removeMessage` refuses any message that carries a body + * group id, a one-shard body's included, so a shard, and any longer interval, + * goes through `removeMessages`, which refuses an interval that splits a + * body. + */ +export function removeStoredMessages(cm: Removals, first: StoredLike, last: StoredLike): void { + if (first.id === last.id && !first.bodyGroupId) { + cm.removeMessage(first.id); + return; + } + cm.removeMessages(first.id, last.id); +} + +/** + * Remove one stored message by its id, with the rest of its body when it's + * in one: a module or a coalescer removes a message it added by the id + * `addMessage` returned, which for a long message stored in shards is its + * first shard's. + */ +export function removeStoredMessage(cm: ContextManager, id: MessageId): void { + const message = cm.getMessage(id) as StoredLike | null; + if (!message?.bodyGroupId) { + cm.removeMessage(id); + return; + } + // Found in blob-free windows: a full read of the branch would inflate every + // attachment on it, once for each message a sweep removes. + const index = locateMessageIndex(cm, String(id)); + if (index < 0) { + cm.removeMessage(id); + return; + } + const run = bodyGroupRun(cm, index); + cm.removeMessages(run.fromId as MessageId, run.toId as MessageId); +} + +/** removeStoredMessages for an interval known by its ids. */ +export function removeStoredInterval(cm: ContextManager, fromId: MessageId, toId: MessageId): void { + const first = cm.getMessage(fromId) as StoredLike | null; + removeStoredMessages(cm, first ?? { id: fromId }, { id: toId }); +} + +/** + * The first and last positions (inclusive) of the body that holds + * `messages[index]`, read the way context-manager 0.14.0 reads bodies + * (`continuesBody`, pairwise in store order). A message outside any body is a + * body of its own. + */ +export function bodyRangeIn(messages: readonly unknown[], index: number): { from: number; to: number } { + return bodyBoundsIn(messages as readonly (BodyShard | null | undefined)[], index); +} diff --git a/src/types/framework.ts b/src/types/framework.ts index 9f5995c68..5725aee5d 100644 --- a/src/types/framework.ts +++ b/src/types/framework.ts @@ -302,6 +302,10 @@ export interface ContextMaintenanceAgentRun { pendingAfter?: string; progressBefore?: Record; progressAfter?: Record; + /** + * The agent's failure in this pass: a tick that threw, or its readiness + * that couldn't be read, before the ticks or after them. + */ error?: string; } @@ -323,6 +327,12 @@ export interface ContextMaintenanceSnapshot { agentName: string; ready: boolean; pending?: string; + /** + * Why the agent's readiness can't be read: the context manager refuses + * while its strategy's initialization is pending or has failed, and a + * strategy can refuse on its own. `ready` is false then. + */ + error?: string; progress?: Record; }>; } diff --git a/test/discord-awareness-outbox.test.ts b/test/discord-awareness-outbox.test.ts index 3a4f55bd3..c22e0a9ca 100644 --- a/test/discord-awareness-outbox.test.ts +++ b/test/discord-awareness-outbox.test.ts @@ -143,6 +143,104 @@ test('framework resumes an interrupted suppression before activating its markers } }); +test('a resumed suppression removes a one-shard body as a range and a message outside any body alone', async () => { + const dir = mkdtempSync(join(tmpdir(), 'discord-awareness-resume-shard-')); + const path = join(dir, 'outbox.json'); + try { + const outbox = new DiscordAwarenessOutbox(path); + const batch = outbox.prepare({ + agentName: 'cairn', + sourceBranch: 'main', + targetBranch: 'recovery/cairn/suppressed', + activationPolicy: 'explicit', + suppressionIntervals: [ + { fromId: 'plain', toId: 'plain' }, + { fromId: 'shard', toId: 'shard' }, + ], + refs: [ + { serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm1' }, + ], + })!; + const present = new Set(['plain', 'shard']); + const removals: string[] = []; + // As context-manager 0.14 does, removeMessage refuses a shard, a + // one-shard body's included. + const cm = { + getMessage: (id: string) => present.has(id) + ? { id, ...(id === 'shard' ? { bodyGroupId: 'doc', shardIndex: 0 } : {}) } + : null, + removeMessage: (id: string) => { + if (id === 'shard') throw new Error(`Cannot remove shard ${id} in isolation: it is part of bodyGroup doc.`); + present.delete(id); + removals.push(id); + }, + removeMessages: (from: string, to: string) => { + present.delete(from); present.delete(to); removals.push(`${from}..${to}`); + }, + }; + const framework = Object.create(AgentFramework.prototype) as any; + framework.discordAwarenessOutbox = outbox; + framework.store = { + currentBranch: () => ({ name: 'recovery/cairn/suppressed' }), + listBranches: () => [], + }; + framework.agents = new Map([['cairn', { getContextManager: () => cm }]]); + + await framework.resumePreparedDiscordSuppressions(); + + assert.deepEqual(removals, ['shard..shard', 'plain']); + assert.equal(outbox.batches().find((candidate) => candidate.id === batch.id)?.status, 'active'); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('a resumed suppression skips an interval that committed before the crash, and refuses one only partly present', async () => { + const dir = mkdtempSync(join(tmpdir(), 'discord-awareness-resume-states-')); + try { + const run = async (present: Set) => { + const outbox = new DiscordAwarenessOutbox(join(dir, `outbox-${[...present].join('-') || 'none'}.json`)); + const batch = outbox.prepare({ + agentName: 'cairn', + sourceBranch: 'main', + targetBranch: 'recovery/cairn/suppressed', + activationPolicy: 'explicit', + suppressionIntervals: [ + { fromId: 'a0', toId: 'a1' }, + { fromId: 'b0', toId: 'b1' }, + ], + refs: [{ serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm1' }], + })!; + const removals: string[] = []; + const cm = { + getMessage: (id: string) => (present.has(id) ? { id } : null), + removeMessage: (id: string) => { present.delete(id); removals.push(id); }, + removeMessages: (from: string, to: string) => { present.delete(from); present.delete(to); removals.push(`${from}..${to}`); }, + }; + const framework = Object.create(AgentFramework.prototype) as any; + framework.discordAwarenessOutbox = outbox; + framework.store = { currentBranch: () => ({ name: 'recovery/cairn/suppressed' }), listBranches: () => [] }; + framework.agents = new Map([['cairn', { getContextManager: () => cm }]]); + const outcome = await framework.resumePreparedDiscordSuppressions().then(() => 'resumed', (error: Error) => error.message); + return { outcome, removals, status: outbox.batches().find((candidate) => candidate.id === batch.id)?.status }; + }; + + // The later interval committed before the crash: only the earlier is removed. + const committed = await run(new Set(['a0', 'a1'])); + assert.equal(committed.outcome, 'resumed'); + assert.deepEqual(committed.removals, ['a0..a1']); + assert.equal(committed.status, 'active'); + + // An interval with one end gone isn't safe to finish. + const partial = await run(new Set(['a0', 'a1', 'b0'])); + assert.match(partial.outcome, /interval b0\.\.b1 is only partially present/); + assert.deepEqual(partial.removals, []); + assert.equal(partial.status, 'prepared'); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + test('framework drain records failures per ref and continues past permanent failures', async () => { const dir = mkdtempSync(join(tmpdir(), 'discord-awareness-drain-')); const path = join(dir, 'outbox.json'); @@ -265,6 +363,11 @@ test('message-granular undo prepares markers before switching branches', async ( framework.pendingAssistantBlocks = new Map(); // A host/command and a store surgery are operations stop() waits for. framework.operatorOperations = new Set(); + // A surgery waits for maintenance work that is running: here, none. + framework.maintenancePass = null; + framework.overBudgetDrains = new Set(); + // Its hold wakes maintenance calls parked at the provider gates: here, none. + framework.providerGates = new Map(); framework.operatorLog = new OperatorLog(undefined); const result = await framework.handleHostCommand('discord', { diff --git a/test/host-quiesce.test.ts b/test/host-quiesce.test.ts index cae60190e..55213e976 100644 --- a/test/host-quiesce.test.ts +++ b/test/host-quiesce.test.ts @@ -312,6 +312,40 @@ test('resume proceeds with a warn when the strategy cannot preview', async () => }); }); +test('resume proceeds without a verdict for an agent whose context manager refuses, and warns naming it', async (t) => { + const membrane = new MockMembrane(); + await withFramework(membrane, async (framework) => { + await framework.quiesce(); + // A failed strategy initialization fences the context manager: its + // settings queries throw until a successful switchBranch or setStrategy. + const cm = framework.getAgent('agent')!.getContextManager(); + const failing = { + name: 'refusing-init-test', + async initialize() { throw new Error('strategy initialization failed for the test'); }, + checkReadiness: () => ({ ready: true }), + async tick() {}, + select: () => [], + } as unknown as ContextStrategy; + await assert.rejects(cm.setStrategy(failing), /strategy initialization failed for the test/); + assert.throws(() => cm.getHotContextSettings(), /requires successful strategy initialization/); + + const warn = t.mock.method(console, 'warn', () => {}); + let status: Awaited>; + try { + status = await framework.resume(); + } finally { + warn.mock.restore(); + } + assert.equal(status.quiesced, false); + assert.ok(warn.mock.calls.some((call) => { + const line = String(call.arguments[0]); + return line.includes("agent's settings can't be read") + && line.includes('requires successful strategy initialization') + && line.includes('proceeding without a verdict for it'); + }), 'resume warns, naming the agent and the error'); + }); +}); + test('maintenanceTick runs the real maintenance pass while quiesced', async () => { class QueuedStrategy implements ContextStrategy { readonly name = 'queued-test'; @@ -341,6 +375,47 @@ test('maintenanceTick runs the real maintenance pass while quiesced', async () = }, { strategy }); }); +test('maintenanceTick runs no pass while a surgery holds the store, and says so', async (t) => { + class QueuedStrategy implements ContextStrategy { + readonly name = 'queued-test'; + ticks = 0; + checkReadiness(): ReadinessState { + return { ready: false, description: 'test maintenance queued' }; + } + async tick(_ctx: StrategyContext): Promise { + this.ticks++; + } + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } + } + const strategy = new QueuedStrategy(); + const membrane = new MockMembrane(); + await withFramework(membrane, async (framework) => { + // Held before start(), whose first pass would otherwise tick at once. + const internals = framework as unknown as { surgeryHold: { verb: string; agentName: string; since: number } | null }; + internals.surgeryHold = { verb: 'rollback', agentName: 'agent', since: Date.now() }; + framework.start(); + await framework.quiesce({ reason: 'compression window' }); + const warn = t.mock.method(console, 'warn', () => {}); + try { + const snapshot = await framework.maintenanceTick(); + assert.equal(snapshot.ran, false); + assert.equal(strategy.ticks, 0); + assert.equal(snapshot.history.length, 0); + assert.ok(warn.mock.calls.some((call) => + String(call.arguments[0]).includes('a rollback for agent holds the store — no pass executed'))); + } finally { + internals.surgeryHold = null; + warn.mock.restore(); + } + }, { strategy }); +}); + // --------------------------------------------------------------------------- // End-to-end MCPL leg: data plane held while quiesced, resume deliverable // over host/command (control plane), funnel reopens and flushes the backlog. diff --git a/test/inference-exhaustion-rewind.test.ts b/test/inference-exhaustion-rewind.test.ts index 0be5ad778..61f54b87d 100644 --- a/test/inference-exhaustion-rewind.test.ts +++ b/test/inference-exhaustion-rewind.test.ts @@ -32,6 +32,13 @@ function makeHarness(opts?: { maxRewinds?: number; messages?: any[] }) { const i = messages.findIndex((m) => m.id === id); if (i >= 0) messages.splice(i, 1); }, + // A range by its two ends, inclusive, as context-manager's removeMessages. + removeMessages: (fromId: string, toId: string) => { + const from = messages.findIndex((m) => m.id === fromId); + const to = messages.findIndex((m) => m.id === toId); + if (from < 0 || to < from) throw new Error(`no range ${fromId}..${toId}`); + for (const m of messages.splice(from, to - from + 1)) removed.push(m.id); + }, addMessage: (_p: string, content: any[], meta: any) => { const id = `marker-${added.length}`; added.push({ content, meta }); diff --git a/test/inference-failure-bounds.test.ts b/test/inference-failure-bounds.test.ts index 025d25256..6fe63cdbe 100644 --- a/test/inference-failure-bounds.test.ts +++ b/test/inference-failure-bounds.test.ts @@ -54,6 +54,7 @@ function makeHarness(opts?: { tick?: () => Promise; addMessage?: (p: strin fw.lastInferenceAt = new Map(); fw.pendingRequests = []; fw.overBudgetDrainInFlight = new Set(); + fw.overBudgetDrains = new Set(); fw.opsAlertLastSent = new Map(); fw.opsAlertCooldownMs = 15 * 60_000; const traces: any[] = []; diff --git a/test/maintenance-timer.test.ts b/test/maintenance-timer.test.ts index 915abce97..8b5e6612d 100644 --- a/test/maintenance-timer.test.ts +++ b/test/maintenance-timer.test.ts @@ -23,6 +23,7 @@ import type { ToolResult, } from '../src/index.js'; import { AgentFramework } from '../src/index.js'; +import { OperatorActionError } from '../src/operator-log.js'; import type { NormalizedRequest, NormalizedResponse } from '@animalabs/membrane'; class QueuedStrategy implements ContextStrategy { @@ -100,6 +101,259 @@ class FailingStrategy implements ContextStrategy { } } +/** A strategy whose initialization always fails. */ +class RefusingInitStrategy implements ContextStrategy { + readonly name = 'refusing-init-test'; + + async initialize(_ctx: StrategyContext): Promise { + throw new Error('strategy initialization failed for the test'); + } + + checkReadiness(): ReadinessState { + return { ready: false, description: 'not expected' }; + } + + async tick(_ctx: StrategyContext): Promise {} + + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } +} + +/** A strategy that refuses on its own, as the autobiographical one does for a branch it hasn't reloaded. */ +class StaleStrategy implements ContextStrategy { + readonly name = 'stale-test'; + + checkReadiness(): ReadinessState { + throw new Error('StaleStrategy.checkReadiness requires reinitialization for the current branch generation'); + } + + getProgressSnapshot(): Record { + throw new Error('StaleStrategy.getProgressSnapshot requires reinitialization for the current branch generation'); + } + + async tick(_ctx: StrategyContext): Promise {} + + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } +} + +/** A strategy whose readiness query refuses with whatever text it's given. */ +class SayingStrategy implements ContextStrategy { + readonly name = 'saying-test'; + says = 'the first refusal'; + + checkReadiness(): ReadinessState { + throw new Error(this.says); + } + + async tick(_ctx: StrategyContext): Promise {} + + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } +} + +/** A strategy with work pending that no tick finishes. */ +class NeverReadyStrategy implements ContextStrategy { + readonly name = 'never-ready-test'; + + checkReadiness(): ReadinessState { + return { ready: false, description: 'never ready' }; + } + + async tick(_ctx: StrategyContext): Promise {} + + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } +} + +/** A strategy whose tick fences its own manager with a failed initialization. */ +class FencingTickStrategy implements ContextStrategy { + readonly name = 'fencing-tick-test'; + manager?: { setStrategy(strategy: ContextStrategy): Promise }; + + checkReadiness(): ReadinessState { + return { ready: false, description: 'fencing tick queued' }; + } + + async tick(_ctx: StrategyContext): Promise { + await this.manager?.setStrategy(new RefusingInitStrategy()).catch(() => undefined); + } + + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } +} + +/** A strategy with work pending whose tick runs a hook: a surgery can begin inside a pass. */ +class HookedTickStrategy implements ContextStrategy { + readonly name = 'hooked-tick-test'; + ticks = 0; + onTick?: () => void; + + checkReadiness(): ReadinessState { + return { ready: false, description: 'hooked tick queued' }; + } + + async tick(_ctx: StrategyContext): Promise { + this.ticks++; + this.onTick?.(); + } + + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } +} + +/** A strategy with no maintenance to do. */ +class ReadyStrategy implements ContextStrategy { + readonly name = 'ready-test'; + + checkReadiness(): ReadinessState { + return { ready: true }; + } + + async tick(_ctx: StrategyContext): Promise {} + + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } +} + +/** A strategy whose first tick waits on a gate, and that notes what it sees. */ +class GatedTickStrategy implements ContextStrategy { + readonly name = 'gated-tick-test'; + ticks = 0; + ticksUnderHold = 0; + branchAtFirstTickEnd?: string; + framework?: AgentFramework; + private open?: () => void; + private readonly gate = new Promise((resolve) => { this.open = resolve; }); + + release(): void { + this.open?.(); + } + + checkReadiness(): ReadinessState { + return { ready: false, description: 'gated tick queued' }; + } + + async tick(_ctx: StrategyContext): Promise { + this.ticks++; + if ((this.framework as unknown as { surgeryHold: unknown }).surgeryHold) this.ticksUnderHold++; + if (this.ticks === 1) { + await this.gate; + this.branchAtFirstTickEnd = this.framework!.getAgent('agent')!.getContextManager().currentBranch().name; + } + } + + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } +} + +/** + * A strategy whose first tick makes a provider call that parks at the agent's + * provider gate, as one does once a cooldown a primary call started has ended + * and the retry turn it queued hasn't run yet. + */ +class ParkedCallTickStrategy implements ContextStrategy { + readonly name = 'parked-call-tick-test'; + ticks = 0; + calls = 0; + outcome?: string; + framework?: AgentFramework; + + checkReadiness(): ReadinessState { + return { ready: false, description: 'parked call queued' }; + } + + async tick(_ctx: StrategyContext): Promise { + this.ticks++; + if (this.ticks > 1) return; + const internals = this.framework as unknown as { + providerGate(name: string): { primaryPending: boolean }; + withAuxiliaryAdmission(name: string, run: () => Promise): Promise; + }; + internals.providerGate('agent').primaryPending = true; + try { + await internals.withAuxiliaryAdmission('agent', async () => { this.calls++; }); + this.outcome = 'ran'; + } catch (error) { + this.outcome = (error as Error).name; + throw error; + } + } + + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } +} + +/** A strategy whose ticks succeed and whose readiness can't be read once all of a pass's ticks have run. */ +class RefusesAfterTicksStrategy implements ContextStrategy { + readonly name = 'refuses-after-ticks-test'; + ticks = 0; + failTick = false; + + checkReadiness(): ReadinessState { + if (this.ticks >= 8 || (this.failTick && this.ticks > 0)) throw new Error('readiness refused after the ticks'); + return { ready: false, description: 'work queued' }; + } + + async tick(_ctx: StrategyContext): Promise { + this.ticks++; + if (this.failTick) throw new Error('tick failed'); + } + + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } +} + class ToolModule implements Module { readonly name = 'maintenance-tools'; @@ -150,6 +404,13 @@ class CapturingMembrane { } } +/** The first pass the timer recorded. The history lists the newest first, and + * under load more passes can end before a test reads it. */ +function firstPass(framework: AgentFramework) { + const history = framework.getContextMaintenanceSnapshot().history; + return history[history.length - 1]!; +} + async function waitFor(predicate: () => boolean, timeoutMs = 1000): Promise { const deadline = Date.now() + timeoutMs; while (!predicate()) { @@ -327,11 +588,807 @@ describe('queued context maintenance timer', () => { pending: 'compression blocked', }, }); - await waitFor(() => framework.getContextMaintenanceSnapshot().history.length > 0); - assert.equal( - framework.getContextMaintenanceSnapshot().history[0].agents[0].error, - 'provider rejected maintenance request', + // Every pass meets the same failure and records it; it is alerted once. + await waitFor(() => framework.getContextMaintenanceSnapshot().history.length >= 3); + for (const run of framework.getContextMaintenanceSnapshot().history) { + assert.equal(run.agents[0].error, 'provider rejected maintenance request'); + } + assert.equal(alerts.length, 1); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('an agent whose readiness can\'t be read fails alone: its run records the error and it gets its own alert', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-refused-')); + const healthy = new QueuedStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [ + { name: 'fenced', model: 'test-model', systemPrompt: 'test', strategy: new QueuedStrategy() }, + { name: 'healthy', model: 'test-model', systemPrompt: 'test', strategy: healthy }, + ], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + const alerts: Array<{ kind: string; agentName: string; data?: Record }> = []; + (framework as any).opsAlert = ( + kind: string, + agentName: string, + _message: string, + opts?: { data?: Record }, + ) => alerts.push({ kind, agentName, data: opts?.data }); + + // A failed strategy initialization fences the context manager: its state + // queries throw until a successful switchBranch or setStrategy. + const fenced = framework.getAgent('fenced')!.getContextManager(); + await assert.rejects(fenced.setStrategy(new RefusingInitStrategy()), /strategy initialization failed for the test/); + assert.throws(() => fenced.isReady(), /requires successful strategy initialization/); + + try { + framework.start(); + await waitFor(() => healthy.ticks >= 2); + await waitFor(() => framework.getContextMaintenanceSnapshot().history.some( + (run) => run.agents.some((agent) => agent.agentName === 'healthy' && agent.readyAfter === true), + )); + const snapshot = framework.getContextMaintenanceSnapshot(); + const run = snapshot.history.find((r) => r.agents.some((agent) => agent.agentName === 'healthy'))!; + const fencedRun = run.agents.find((agent) => agent.agentName === 'fenced'); + assert.equal(fencedRun?.ticks, 0); + assert.match(fencedRun?.error ?? '', /requires successful strategy initialization/); + const entry = snapshot.agents.find((agent) => agent.agentName === 'fenced'); + assert.equal(entry?.ready, false); + assert.match(entry?.error ?? '', /requires successful strategy initialization/); + assert.ok(alerts.some((alert) => alert.agentName === 'fenced' && alert.data?.scope === 'agent')); + assert.equal(alerts.some((alert) => alert.data?.scope === 'pass'), false); + assert.equal(alerts.some((alert) => alert.agentName === 'healthy'), false); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a failure that begins during the pass is that agent\'s error and alert, not the pass\'s', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-refused-after-')); + const strategy = new FencingTickStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + strategy.manager = framework.getAgent('agent')!.getContextManager(); + const alerts: Array<{ kind: string; agentName: string; data?: Record }> = []; + (framework as any).opsAlert = ( + kind: string, + agentName: string, + _message: string, + opts?: { data?: Record }, + ) => alerts.push({ kind, agentName, data: opts?.data }); + + try { + framework.start(); + await waitFor(() => framework.getContextMaintenanceSnapshot().history.length >= 1); + const run = firstPass(framework).agents[0]; + assert.equal(run.agentName, 'agent'); + assert.equal(run.ticks, 1); + assert.match(run.error ?? '', /requires successful strategy initialization/); + assert.equal(run.readyAfter, undefined); + assert.ok(alerts.some((alert) => alert.agentName === 'agent' && alert.kind === 'context-maintenance-failed' && alert.data?.scope === 'agent')); + assert.equal(alerts.some((alert) => alert.data?.scope === 'pass'), false); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a strategy that can\'t report its readiness or its progress fails that agent alone, and the snapshot still answers', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-stale-')); + const healthy = new QueuedStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [ + { name: 'stale', model: 'test-model', systemPrompt: 'test', strategy: new StaleStrategy() }, + { name: 'healthy', model: 'test-model', systemPrompt: 'test', strategy: healthy }, + ], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + const alerts: Array<{ kind: string; agentName: string; data?: Record }> = []; + (framework as any).opsAlert = ( + kind: string, + agentName: string, + _message: string, + opts?: { data?: Record }, + ) => alerts.push({ kind, agentName, data: opts?.data }); + + try { + framework.start(); + await waitFor(() => healthy.ticks >= 2); + await waitFor(() => framework.getContextMaintenanceSnapshot().history.some( + (run) => run.agents.some((agent) => agent.agentName === 'healthy' && agent.readyAfter === true), + )); + const snapshot = framework.getContextMaintenanceSnapshot(); + const entry = snapshot.agents.find((agent) => agent.agentName === 'stale'); + assert.equal(entry?.ready, false); + assert.match(entry?.error ?? '', /requires reinitialization for the current branch generation/); + assert.equal(entry?.progress, undefined); + assert.ok(alerts.some((alert) => alert.agentName === 'stale' && alert.data?.scope === 'agent')); + assert.equal(alerts.some((alert) => alert.data?.scope === 'pass'), false); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('whatever throws while one agent\'s maintenance is prepared is that agent\'s failure alone', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-prepare-throws-')); + const healthy = new QueuedStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [ + { name: 'broken', model: 'test-model', systemPrompt: 'test', strategy: new QueuedStrategy() }, + { name: 'healthy', model: 'test-model', systemPrompt: 'test', strategy: healthy }, + ], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + const alerts: Array<{ kind: string; agentName: string; data?: Record }> = []; + (framework as any).opsAlert = ( + kind: string, + agentName: string, + _message: string, + opts?: { data?: Record }, + ) => alerts.push({ kind, agentName, data: opts?.data }); + const internals = framework as unknown as { providerGateBlocked: (name: string) => boolean }; + const gateBlocked = internals.providerGateBlocked.bind(framework); + internals.providerGateBlocked = (name: string) => { + if (name === 'broken') throw new Error('gate read failed'); + return gateBlocked(name); + }; + try { + framework.start(); + await waitFor(() => healthy.ticks >= 2); + await waitFor(() => framework.getContextMaintenanceSnapshot().history.some( + (run) => run.agents.some((agent) => agent.agentName === 'broken' && agent.error !== undefined), + )); + const run = framework.getContextMaintenanceSnapshot().history.find( + (candidate) => candidate.agents.some((agent) => agent.agentName === 'broken'), + )!; + const broken = run.agents.find((agent) => agent.agentName === 'broken')!; + assert.equal(broken.error, 'gate read failed'); + assert.equal(broken.ticks, 0); + assert.ok(alerts.some((alert) => alert.agentName === 'broken' && alert.data?.scope === 'agent')); + assert.equal(alerts.some((alert) => alert.data?.scope === 'pass'), false); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a failure that lasts is logged and alerted once, and again once the agent has recovered', async (t) => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-lasting-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'fenced', model: 'test-model', systemPrompt: 'test', strategy: new QueuedStrategy() }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + const alerts: string[] = []; + (framework as any).opsAlert = (kind: string, agentName: string) => alerts.push(`${kind}:${agentName}`); + const internals = framework as unknown as { runQueuedMaintenance: () => Promise }; + let passes = 0; + const runPass = internals.runQueuedMaintenance.bind(framework); + internals.runQueuedMaintenance = () => { passes++; return runPass(); }; + const logged = t.mock.method(console, 'error', () => {}); + const failedLines = () => logged.mock.calls.filter((call) => + String(call.arguments[0]).includes('[context-maintenance] agent=fenced failed:')).length; + const runsWithError = () => framework.getContextMaintenanceSnapshot().history.filter((run) => + run.agents.some((agent) => agent.agentName === 'fenced' && agent.error !== undefined)).length; + const cm = framework.getAgent('fenced')!.getContextManager(); + const fence = () => assert.rejects(cm.setStrategy(new RefusingInitStrategy()), /strategy initialization failed for the test/); + await fence(); + try { + framework.start(); + await waitFor(() => runsWithError() >= 3); + assert.deepEqual(alerts, ['context-maintenance-failed:fenced']); + assert.equal(failedLines(), 1); + + // Recovered with nothing to do: passes find it ready and skip it. + await cm.setStrategy(new ReadyStrategy()); + const after = passes; + await waitFor(() => passes >= after + 3); + await fence(); + await waitFor(() => alerts.length === 2); + assert.equal(failedLines(), 2); + + // Recovered with work no tick finishes: a pass runs it without a failure. + await cm.setStrategy(new NeverReadyStrategy()); + await waitFor(() => framework.getContextMaintenanceSnapshot().history.some((run) => + run.agents.some((agent) => agent.agentName === 'fenced' && agent.ticks > 0 && agent.error === undefined))); + await fence(); + await waitFor(() => alerts.length === 3); + assert.equal(failedLines(), 3); + assert.deepEqual(new Set(alerts), new Set(['context-maintenance-failed:fenced'])); + } finally { + logged.mock.restore(); + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a failure that lasts is alerted again when its whole text changes, and again once the alert cooldown has passed', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-lasting-text-')); + const strategy = new SayingStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'saying', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + const alerts: string[] = []; + (framework as any).opsAlert = (_kind: string, _agentName: string, text: string) => alerts.push(text); + const internals = framework as unknown as { runQueuedMaintenance: () => Promise; opsAlertCooldownMs: number }; + let passes = 0; + const runPass = internals.runQueuedMaintenance.bind(framework); + internals.runQueuedMaintenance = () => { passes++; return runPass(); }; + const passed = async (n: number) => { const after = passes; await waitFor(() => passes >= after + n); }; + const quiet = console.error; + console.error = () => {}; + try { + framework.start(); + await waitFor(() => alerts.length === 1); + await passed(3); + assert.deepEqual(alerts, ['the first refusal']); + + // Two long texts that differ only in their middle, which the bounded + // text a run records leaves out: the alert follows the whole text. + const long = (middle: string) => `${'a'.repeat(3_000)} ${middle} ${'z'.repeat(3_000)}`; + strategy.says = long('one'); + await waitFor(() => alerts.length === 2); + strategy.says = long('two'); + await waitFor(() => alerts.length === 3); + await passed(3); + assert.equal(alerts.length, 3); + + // Unchanged, it is alerted again once the cooldown has passed. + internals.opsAlertCooldownMs = 0; + await waitFor(() => alerts.length >= 5); + assert.ok(alerts.slice(3).every((text) => text === long('two'))); + } finally { + console.error = quiet; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a failure that lasts is posted to the webhook again a cooldown after its last post, whatever changed-text report came between', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-webhook-reminder-')); + const strategy = new SayingStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'saying', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + const internals = framework as unknown as { runQueuedMaintenance: () => Promise }; + let passes = 0; + const runPass = internals.runQueuedMaintenance.bind(framework); + internals.runQueuedMaintenance = () => { passes++; return runPass(); }; + const passed = async (n: number) => { const after = passes; await waitFor(() => passes >= after + n); }; + // The real opsAlert, with a webhook whose posts are recorded. + const posts: string[] = []; + const realFetch = globalThis.fetch; + globalThis.fetch = (async (_url: unknown, init?: { body?: string }) => { + posts.push(String(JSON.parse(String(init?.body)).content)); + return new Response('{}'); + }) as typeof fetch; + const hook = process.env.CONNECTOME_OPS_WEBHOOK; + process.env.CONNECTOME_OPS_WEBHOOK = 'http://webhook.invalid/hook'; + const realNow = Date.now; + const start = realNow(); + let offset = 0; + Date.now = () => realNow() + offset; + const minute = 60_000; + const maintenancePosts = () => posts.filter((post) => post.includes('context-maintenance-failed')); + const quiet = console.error; + console.error = () => {}; + try { + framework.start(); + await waitFor(() => maintenancePosts().length === 1); + assert.match(maintenancePosts()[0]!, /the first refusal/); + // 14 minutes on, the text changes: it's reported, and the webhook, + // still within its cooldown, doesn't post it. + offset = 14 * minute; + strategy.says = 'the second refusal'; + await passed(3); + assert.equal(maintenancePosts().length, 1); + // At 15 minutes the webhook's cooldown since its post has passed. + offset = 15 * minute + 1_000; + await waitFor(() => maintenancePosts().length === 2); + assert.match(maintenancePosts()[1]!, /the second refusal/); + assert.ok(Date.now() - start < 16 * minute); + } finally { + console.error = quiet; + Date.now = realNow; + globalThis.fetch = realFetch; + if (hook === undefined) delete process.env.CONNECTOME_OPS_WEBHOOK; else process.env.CONNECTOME_OPS_WEBHOOK = hook; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a readiness refusal read after an agent\'s last tick is that agent\'s failure, alerted once', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-refused-after-ticks-')); + const strategy = new RefusesAfterTicksStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + // start() runs one pass at once; none follows within the test. + maintenanceIntervalMs: 60_000, + }); + const alerts: string[] = []; + (framework as any).opsAlert = (kind: string, agentName: string, message: string) => alerts.push(`${kind}:${agentName}:${message}`); + try { + framework.start(); + await waitFor(() => framework.getContextMaintenanceSnapshot().history.length >= 1); + const run = firstPass(framework).agents[0]; + assert.equal(run.ticks, 8); + assert.equal(run.error, 'readiness refused after the ticks'); + assert.equal(run.readyAfter, undefined); + assert.deepEqual(alerts, ['context-maintenance-failed:agent:readiness refused after the ticks']); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a run records an agent\'s first failure: a readiness refusal after a failed tick isn\'t reported again', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-first-failure-')); + const strategy = new RefusesAfterTicksStrategy(); + strategy.failTick = true; + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 60_000, + }); + const alerts: string[] = []; + (framework as any).opsAlert = (kind: string, agentName: string, message: string) => alerts.push(`${kind}:${agentName}:${message}`); + try { + framework.start(); + await waitFor(() => framework.getContextMaintenanceSnapshot().history.length >= 1); + const run = firstPass(framework).agents[0]; + assert.equal(run.ticks, 0); + assert.equal(run.error, 'tick failed'); + assert.deepEqual(alerts, ['context-maintenance-failed:agent:tick failed']); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + for (const surgery of [ + { verb: 'roll back', run: (framework: AgentFramework, first: string) => framework.rollbackToMessage('agent', { messageId: first }) }, + { verb: 'suppress', run: (framework: AgentFramework, first: string) => framework.suppressMessages('agent', { messageIds: [first] }) }, + { + verb: 'switch branch', + who: 'every agent', + run: (framework: AgentFramework, first: string) => { + framework.getAgent('agent')!.getContextManager().branchAt(first as never, 'elsewhere'); + return framework.switchBranch('elsewhere'); + }, + }, + ] as Array<{ verb: string; who?: string; run: (framework: AgentFramework, first: string) => Promise }>) { + it(`a ${surgery.verb} that begins during a tick waits for it before it moves the branch, and resume refuses while the ${surgery.verb} holds the store`, { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-surgery-waits-')); + const strategy = new GatedTickStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + strategy.framework = framework; + const cm = framework.getAgent('agent')!.getContextManager(); + const first = String(cm.addMessage('user', [{ type: 'text', text: 'm1' }])); + cm.addMessage('user', [{ type: 'text', text: 'm2' }]); + const internals = framework as unknown as { surgeryHold: unknown }; + try { + await framework.quiesce(); + framework.start(); + await waitFor(() => strategy.ticks === 1); + const moved = surgery.run(framework, first); + await waitFor(() => internals.surgeryHold !== null); + await assert.rejects( + framework.resume(), + (error: unknown) => error instanceof OperatorActionError && error.code === 'agent-busy' + && error.message.includes(`a live ${surgery.verb} for ${surgery.who ?? 'agent'} holds the store`), + ); + assert.equal(cm.currentBranch().name, 'main', `the ${surgery.verb} waits for the running tick`); + strategy.release(); + await moved; + assert.equal(strategy.branchAtFirstTickEnd, 'main'); + assert.equal(strategy.ticksUnderHold, 0, `the pass starts no tick while the ${surgery.verb} holds the store`); + assert.notEqual(cm.currentBranch().name, 'main'); + const status = await framework.resume(); + assert.equal(status.quiesced, false); + } finally { + strategy.release(); + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + } + + for (const surgery of [ + { verb: 'roll back', run: (framework: AgentFramework, first: string) => framework.rollbackToMessage('agent', { messageId: first }) }, + { verb: 'suppress', run: (framework: AgentFramework, first: string) => framework.suppressMessages('agent', { messageIds: [first] }) }, + { + verb: 'switch branch', + run: (framework: AgentFramework, first: string) => { + framework.getAgent('agent')!.getContextManager().branchAt(first as never, 'elsewhere'); + return framework.switchBranch('elsewhere'); + }, + }, + ] as Array<{ verb: string; run: (framework: AgentFramework, first: string) => Promise }>) { + it(`a ${surgery.verb} withdraws a maintenance call parked at the provider gate rather than waiting on it, and that is no failure`, { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-parked-call-')); + const strategy = new ParkedCallTickStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + strategy.framework = framework; + const alerts: string[] = []; + (framework as any).opsAlert = (kind: string, agentName: string, message: string) => alerts.push(`${kind}:${agentName}:${message}`); + const cm = framework.getAgent('agent')!.getContextManager(); + const first = String(cm.addMessage('user', [{ type: 'text', text: 'm1' }])); + cm.addMessage('user', [{ type: 'text', text: 'm2' }]); + const internals = framework as unknown as { providerGate(name: string): { auxiliaryWaiters: unknown[] } }; + const quiet = console.error; + console.error = () => {}; + try { + await framework.quiesce(); + framework.start(); + await waitFor(() => internals.providerGate('agent').auxiliaryWaiters.length === 1); + const moved = surgery.run(framework, first); + const outcome = await Promise.race([ + moved.then(() => 'moved'), + new Promise((resolve) => setTimeout(() => resolve(`the ${surgery.verb} still waits`), 3_000)), + ]); + assert.equal(outcome, 'moved'); + assert.equal(strategy.outcome, 'MaintenanceWithdrawnError'); + assert.equal(strategy.calls, 0, 'the withdrawn call sent nothing'); + assert.notEqual(cm.currentBranch().name, 'main'); + assert.deepEqual(alerts.filter((alert) => alert.startsWith('context-maintenance-failed')), []); + const run = firstPass(framework).agents[0]!; + assert.equal(run.error, undefined); + } finally { + console.error = quiet; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + } + + it('a maintenance provider call made while a surgery holds the store is withdrawn before it is sent, with the gate closed too', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-withdrawn-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy: new GatedTickStrategy() }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 0, + }); + const internals = framework as unknown as { + reserveStoreForSurgery(verb: string, agentName: string): () => void; + withAuxiliaryAdmission(name: string, run: () => Promise): Promise; + providerGate(name: string): { primaryPending: boolean }; + }; + try { + for (const closed of [false, true]) { + let ran = false; + const release = internals.reserveStoreForSurgery('roll back', 'agent'); + // Closed: a cooldown's retry turn is queued, which the hold keeps from + // starting, so a call that parked would never be woken. + internals.providerGate('agent').primaryPending = closed; + try { + const admitted = internals.withAuxiliaryAdmission('agent', async () => { ran = true; }); + void admitted.catch(() => {}); + const outcome = await Promise.race([ + admitted.then(() => 'admitted', (error: unknown) => error), + new Promise((resolve) => setTimeout(() => resolve('parked'), 1_000)), + ]); + assert.ok( + outcome instanceof Error && outcome.name === 'MaintenanceWithdrawnError' + && outcome.message === 'maintenance call withdrawn: a live roll back for agent holds the store', + `${closed ? 'gate closed' : 'gate open'}: ${outcome instanceof Error ? outcome.message : String(outcome)}`, + ); + assert.equal(ran, false); + } finally { + internals.providerGate('agent').primaryPending = false; + release(); + } + } + assert.equal(await internals.withAuxiliaryAdmission('agent', async () => 7), 7, 'admitted again once the hold is released'); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a parked maintenance call woken while its gate is still closed parks again, and goes once the gate opens', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-parks-again-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy: new GatedTickStrategy() }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 0, + }); + const internals = framework as unknown as { + withAuxiliaryAdmission(name: string, run: () => Promise): Promise; + providerGate(name: string): { primaryPending: boolean; auxiliaryWaiters: Array<() => void> }; + flushAuxiliaryAdmission(name: string): void; + }; + try { + const gate = internals.providerGate('agent'); + gate.primaryPending = true; + let ran = false; + const call = internals.withAuxiliaryAdmission('agent', async () => { ran = true; return 7; }); + await waitFor(() => gate.auxiliaryWaiters.length === 1); + for (const resolve of gate.auxiliaryWaiters.splice(0)) resolve(); + await waitFor(() => gate.auxiliaryWaiters.length === 1); + assert.equal(ran, false, 'woken with the gate still closed, it parked again'); + gate.primaryPending = false; + internals.flushAuxiliaryAdmission('agent'); + assert.equal(await call, 7); + assert.equal(ran, true); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a surgery that takes the store while resume waits for a tick is refused by resume after the wait', { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-resume-rechecks-')); + const strategy = new GatedTickStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + strategy.framework = framework; + const cm = framework.getAgent('agent')!.getContextManager(); + const first = String(cm.addMessage('user', [{ type: 'text', text: 'm1' }])); + cm.addMessage('user', [{ type: 'text', text: 'm2' }]); + const internals = framework as unknown as { surgeryHold: unknown }; + try { + await framework.quiesce(); + framework.start(); + await waitFor(() => strategy.ticks === 1); + // resume() finds no surgery and waits for the running tick; the + // rollback takes the store meanwhile. + const resuming = framework.resume(); + void resuming.catch(() => {}); + const moved = framework.rollbackToMessage('agent', { messageId: first }); + await waitFor(() => internals.surgeryHold !== null); + strategy.release(); + await assert.rejects( + resuming, + (error: unknown) => error instanceof OperatorActionError && error.code === 'agent-busy' + && error.message.includes('a live roll back for agent holds the store'), ); + assert.equal(framework.getHostModeStatus().quiesced, true, 'the host stays quiesced'); + await moved; + } finally { + strategy.release(); + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a rollback waits for an over-budget drain kick\'s tick in flight, and the kick ticks no more under the hold', { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-drain-hold-')); + const strategy = new GatedTickStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 0, + }); + strategy.framework = framework; + const cm = framework.getAgent('agent')!.getContextManager(); + const first = String(cm.addMessage('user', [{ type: 'text', text: 'm1' }])); + cm.addMessage('user', [{ type: 'text', text: 'm2' }]); + const internals = framework as unknown as { + surgeryHold: unknown; + noteInferenceExhausted(agentName: string, reason: string, retryable?: boolean, errorType?: string): void; + }; + const quiet = console.error; + console.error = () => {}; + try { + // A refused compile kicks the drain, whose first tick waits on the gate. + internals.noteInferenceExhausted('agent', 'compile refused', false, 'over_budget'); + await waitFor(() => strategy.ticks === 1); + const moved = framework.rollbackToMessage('agent', { messageId: first }); + await waitFor(() => internals.surgeryHold !== null); + assert.equal(cm.currentBranch().name, 'main', 'the rollback waits for the drain kick\'s tick'); + strategy.release(); + await moved; + assert.equal(strategy.branchAtFirstTickEnd, 'main'); + assert.equal(strategy.ticksUnderHold, 0, 'the kick starts no tick while the rollback holds the store'); + assert.notEqual(cm.currentBranch().name, 'main'); + } finally { + console.error = quiet; + strategy.release(); + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('no pass starts while a surgery holds the store', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-surgery-hold-')); + const strategy = new QueuedStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + const internals = framework as unknown as { + surgeryHold: { verb: string; agentName: string; since: number } | null; + runQueuedMaintenance: () => Promise; + }; + internals.surgeryHold = { verb: 'rollback', agentName: 'agent', since: Date.now() }; + // Count the passes the timer starts while the store is held. + let passes = 0; + const runPass = internals.runQueuedMaintenance.bind(framework); + internals.runQueuedMaintenance = () => { passes++; return runPass(); }; + try { + framework.start(); + await waitFor(() => passes >= 3); + assert.equal(strategy.ticks, 0); + assert.equal(framework.getContextMaintenanceSnapshot().history.length, 0); + internals.surgeryHold = null; + await waitFor(() => strategy.ticks >= 2); + } finally { + internals.surgeryHold = null; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a pass starts no new tick once a surgery holds the store', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-surgery-mid-pass-')); + const strategy = new HookedTickStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + const internals = framework as unknown as { surgeryHold: { verb: string; agentName: string; since: number } | null }; + const alerts: string[] = []; + (framework as any).opsAlert = (kind: string) => alerts.push(kind); + strategy.onTick = () => { + internals.surgeryHold = { verb: 'rollback', agentName: 'agent', since: Date.now() }; + }; + try { + framework.start(); + await waitFor(() => framework.getContextMaintenanceSnapshot().history.length >= 1); + const run = firstPass(framework).agents[0]; + assert.equal(strategy.ticks, 1); + assert.equal(run.ticks, 1); + assert.equal(run.error, undefined); + // The surgery waits for the pass before it switches, so reading the + // agent's readiness after its tick is safe. + assert.equal(run.readyAfter, false); + assert.equal(run.pendingAfter, 'hooked tick queued'); + assert.deepEqual(alerts, []); + } finally { + internals.surgeryHold = null; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a failure a pass meets while a surgery waits for it is reported as any failure', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-surgery-failure-')); + const strategy = new HookedTickStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + const internals = framework as unknown as { surgeryHold: { verb: string; agentName: string; since: number } | null }; + const alerts: string[] = []; + (framework as any).opsAlert = (kind: string) => alerts.push(kind); + strategy.onTick = () => { + internals.surgeryHold = { verb: 'rollback', agentName: 'agent', since: Date.now() }; + throw new Error('the tick failed while a surgery waited'); + }; + const logged: string[] = []; + const consoleError = console.error; + console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; + try { + framework.start(); + await waitFor(() => framework.getContextMaintenanceSnapshot().history.length >= 1); + const run = firstPass(framework).agents[0]; + assert.equal(run.ticks, 0); + assert.equal(run.error, 'the tick failed while a surgery waited'); + assert.deepEqual(alerts, ['context-maintenance-failed']); + assert.ok(logged.some((line) => + line.includes('[context-maintenance] agent=agent failed:') + && line.includes('the tick failed while a surgery waited'))); + } finally { + console.error = consoleError; + internals.surgeryHold = null; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a pass that ends before its agent is ready records what is still pending', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-pending-after-')); + const strategy = new NeverReadyStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + try { + framework.start(); + await waitFor(() => framework.getContextMaintenanceSnapshot().history.length >= 1); + const run = framework.getContextMaintenanceSnapshot().history.at(-1)!.agents[0]; + assert.equal(run.ticks, 8); + assert.equal(run.readyAfter, false); + assert.equal(run.pendingBefore, 'never ready'); + assert.equal(run.pendingAfter, 'never ready'); } finally { await framework.stop(); rmSync(dir, { recursive: true, force: true }); diff --git a/test/offline-recovery-branch.test.ts b/test/offline-recovery-branch.test.ts index c514bc3d5..640cc0f59 100644 --- a/test/offline-recovery-branch.test.ts +++ b/test/offline-recovery-branch.test.ts @@ -268,3 +268,333 @@ test('suppression rejects a range that splits a tool exchange', async () => { rmSync(dir, { recursive: true, force: true }); } }); + +// A document posted twice stores two runs of shards under one bodyGroupId +// (the chunker names a group by its text). Each run is a body of its own. +type ShardAppender = { + messageStore: { + append: ( + participant: string, + content: unknown[], + metadata?: unknown, + causedBy?: unknown, + extra?: { bodyGroupId: string; shardIndex: number }, + ) => { id: unknown }; + }; +}; + +/** One body of `texts.length` shards, each stored with its own Discord message id. */ +function appendShards(cm: ContextManager, group: string, discordIds: string[]): string[] { + return discordIds.map((discordMessageId, shardIndex) => String( + (cm as unknown as ShardAppender).messageStore.append( + 'user', + [{ type: 'text', text: `shard ${shardIndex}` }], + { serverId: 'discord', channelId: 'discord:g1:c1', messageId: discordMessageId }, + undefined, + { bodyGroupId: group, shardIndex }, + ).id, + )); +} + +function appendDocumentCopy(cm: ContextManager, discordMessageId: string): string[] { + return ['part one', 'part two'].map((text, shardIndex) => String( + (cm as unknown as ShardAppender).messageStore.append( + 'user', + [{ type: 'text', text }], + { serverId: 'discord', channelId: 'discord:g1:c1', messageId: discordMessageId }, + undefined, + { bodyGroupId: 'doc-group', shardIndex }, + ).id, + )); +} + +test('an anchor at the end of one copy of a reposted document does not split it', async () => { + const dir = mkdtempSync(join(tmpdir(), 'offline-recovery-repost-anchor-')); + const storePath = join(dir, 'agent.chronicle'); + try { + const store = JsStore.openOrCreate({ path: storePath }); + const cm = await ContextManager.open({ store, namespace: 'agents/cairn' }); + cm.addMessage('user', [{ type: 'text', text: 'before' }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm-before', + }); + const first = appendDocumentCopy(cm, 'm-doc-1'); + appendDocumentCopy(cm, 'm-doc-2'); + cm.close(); + store.close(); + + const result = await createOfflineRecoveryBranch({ + storePath, + agentName: 'cairn', + contextId: first[1], + branchName: 'recovery/cairn/repost-anchor', + }); + + assert.equal(result.messagesRemoved, 2); + assert.deepEqual(result.refs.map((ref) => ref.messageId), ['m-doc-2']); + const recoveredStore = JsStore.openOrCreate({ path: storePath }); + const recoveredCm = await ContextManager.open({ store: recoveredStore, namespace: 'agents/cairn' }); + assert.deepEqual( + recoveredCm.getAllMessages().map((message) => message.metadata?.messageId), + ['m-before', 'm-doc-1', 'm-doc-1'], + ); + recoveredCm.close(); + recoveredStore.close(); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('an anchor inside one copy of a document still splits its body', async () => { + const dir = mkdtempSync(join(tmpdir(), 'offline-recovery-split-anchor-')); + const storePath = join(dir, 'agent.chronicle'); + try { + const store = JsStore.openOrCreate({ path: storePath }); + const cm = await ContextManager.open({ store, namespace: 'agents/cairn' }); + cm.addMessage('user', [{ type: 'text', text: 'before' }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm-before', + }); + const first = appendDocumentCopy(cm, 'm-doc-1'); + appendDocumentCopy(cm, 'm-doc-2'); + cm.close(); + store.close(); + + await assert.rejects( + createOfflineRecoveryBranch({ + storePath, + agentName: 'cairn', + contextId: first[0], + branchName: 'recovery/cairn/split-anchor', + }), + /splits body group doc-group; choose its final shard/, + ); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('suppressing one copy of a reposted document keeps the other', async () => { + const dir = mkdtempSync(join(tmpdir(), 'offline-recovery-repost-suppress-')); + const storePath = join(dir, 'agent.chronicle'); + try { + const store = JsStore.openOrCreate({ path: storePath }); + const cm = await ContextManager.open({ store, namespace: 'agents/cairn' }); + cm.addMessage('user', [{ type: 'text', text: 'before' }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm-before', + }); + appendDocumentCopy(cm, 'm-doc-1'); + appendDocumentCopy(cm, 'm-doc-2'); + cm.addMessage('user', [{ type: 'text', text: 'current' }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm-current', + }); + cm.close(); + store.close(); + + const result = await createOfflineRecoveryBranch({ + storePath, + agentName: 'cairn', + messageId: 'm-current', + suppressMessageIds: ['m-doc-2'], + }); + + assert.equal(result.messagesRemoved, 0); + assert.equal(result.messagesSuppressed, 2); + assert.deepEqual(result.refs.map((ref) => ref.messageId), ['m-doc-2']); + const recoveredStore = JsStore.openOrCreate({ path: storePath }); + const recoveredCm = await ContextManager.open({ store: recoveredStore, namespace: 'agents/cairn' }); + assert.deepEqual( + recoveredCm.getAllMessages().map((message) => message.metadata?.messageId), + ['m-before', 'm-doc-1', 'm-doc-1', 'm-current'], + ); + recoveredCm.close(); + recoveredStore.close(); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('a suppression that splits one copy of a document is refused', async () => { + const dir = mkdtempSync(join(tmpdir(), 'offline-recovery-split-suppress-')); + const storePath = join(dir, 'agent.chronicle'); + try { + const store = JsStore.openOrCreate({ path: storePath }); + const cm = await ContextManager.open({ store, namespace: 'agents/cairn' }); + cm.addMessage('user', [{ type: 'text', text: 'before' }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm-before', + }); + appendShards(cm, 'doc-group', ['m-s0', 'm-s1']); + appendShards(cm, 'doc-group', ['m-t0', 'm-t1']); + cm.addMessage('user', [{ type: 'text', text: 'current' }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm-current', + }); + cm.close(); + store.close(); + + await assert.rejects( + createOfflineRecoveryBranch({ + storePath, + agentName: 'cairn', + messageId: 'm-current', + suppressMessageIds: ['m-s1'], + }), + /Recovery selection splits body group doc-group/, + ); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('a split across a scan window is refused too', async () => { + const dir = mkdtempSync(join(tmpdir(), 'offline-recovery-split-seam-')); + const storePath = join(dir, 'agent.chronicle'); + try { + const store = JsStore.openOrCreate({ path: storePath }); + const cm = await ContextManager.open({ store, namespace: 'agents/cairn' }); + // 999 plain messages put the body's shards at positions 999 and 1000, + // on either side of the validator's 1,000-message scan window. + for (let i = 0; i < 999; i++) cm.addMessage('user', [{ type: 'text', text: `filler ${i}` }]); + appendShards(cm, 'seam-group', ['m-s0', 'm-s1']); + cm.addMessage('user', [{ type: 'text', text: 'current' }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm-current', + }); + assert.equal(cm.getMessageCount(), 1002); + cm.close(); + store.close(); + + await assert.rejects( + createOfflineRecoveryBranch({ + storePath, + agentName: 'cairn', + messageId: 'm-current', + suppressMessageIds: ['m-s1'], + }), + /Recovery selection splits body group seam-group/, + ); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('suppressing a one-shard copy of a reposted document keeps the other', async () => { + const dir = mkdtempSync(join(tmpdir(), 'offline-recovery-one-shard-')); + const storePath = join(dir, 'agent.chronicle'); + try { + const store = JsStore.openOrCreate({ path: storePath }); + const cm = await ContextManager.open({ store, namespace: 'agents/cairn' }); + cm.addMessage('user', [{ type: 'text', text: 'before' }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm-before', + }); + appendShards(cm, 'one-shard-group', ['m-doc-1']); + appendShards(cm, 'one-shard-group', ['m-doc-2']); + cm.addMessage('user', [{ type: 'text', text: 'current' }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm-current', + }); + cm.close(); + store.close(); + + const result = await createOfflineRecoveryBranch({ + storePath, + agentName: 'cairn', + messageId: 'm-current', + suppressMessageIds: ['m-doc-2'], + }); + + assert.equal(result.messagesSuppressed, 1); + const recoveredStore = JsStore.openOrCreate({ path: storePath }); + const recoveredCm = await ContextManager.open({ store: recoveredStore, namespace: 'agents/cairn' }); + assert.deepEqual( + recoveredCm.getAllMessages().map((message) => message.metadata?.messageId), + ['m-before', 'm-doc-1', 'm-current'], + ); + recoveredCm.close(); + recoveredStore.close(); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('a suppressed message outside any body is removed alone, and a one-shard copy as a range', async (t) => { + const dir = mkdtempSync(join(tmpdir(), 'offline-recovery-removal-kind-')); + const storePath = join(dir, 'agent.chronicle'); + try { + const store = JsStore.openOrCreate({ path: storePath }); + const cm = await ContextManager.open({ store, namespace: 'agents/cairn' }); + cm.addMessage('user', [{ type: 'text', text: 'before' }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm-before', + }); + const plain = String(cm.addMessage('user', [{ type: 'text', text: 'plain' }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm-plain', + })); + appendShards(cm, 'one-shard-group', ['m-doc-1']); + const [copy] = appendShards(cm, 'one-shard-group', ['m-doc-2']); + cm.addMessage('user', [{ type: 'text', text: 'current' }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm-current', + }); + cm.close(); + store.close(); + + // removeMessage's remove event also drops the context-log entries copied + // from the message; context-manager refuses it for any shard. + const removeMessage = t.mock.method(ContextManager.prototype, 'removeMessage'); + const removeMessages = t.mock.method(ContextManager.prototype, 'removeMessages'); + const result = await createOfflineRecoveryBranch({ + storePath, + agentName: 'cairn', + messageId: 'm-current', + suppressMessageIds: ['m-plain', 'm-doc-2'], + }); + + assert.equal(result.messagesSuppressed, 2); + assert.deepEqual(removeMessage.mock.calls.map((call) => String(call.arguments[0])), [plain]); + assert.deepEqual(removeMessages.mock.calls.map((call) => call.arguments.map(String)), [[copy, copy]]); + removeMessage.mock.restore(); + removeMessages.mock.restore(); + const recoveredStore = JsStore.openOrCreate({ path: storePath }); + const recoveredCm = await ContextManager.open({ store: recoveredStore, namespace: 'agents/cairn' }); + assert.deepEqual( + recoveredCm.getAllMessages().map((message) => message.metadata?.messageId), + ['m-before', 'm-doc-1', 'm-current'], + ); + recoveredCm.close(); + recoveredStore.close(); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('an anchor at the end of a write that was cut short keeps that partial body and leaves out its retry', async () => { + const dir = mkdtempSync(join(tmpdir(), 'offline-recovery-cut-short-')); + const storePath = join(dir, 'agent.chronicle'); + try { + const store = JsStore.openOrCreate({ path: storePath }); + const cm = await ContextManager.open({ store, namespace: 'agents/cairn' }); + cm.addMessage('user', [{ type: 'text', text: 'before' }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'm-before', + }); + // Two of a body's shards, written before an interruption, then its retry + // from shard 0: the retry is a body of its own. + const partial = appendShards(cm, 'cut-group', ['m-cut-0', 'm-cut-1']); + appendShards(cm, 'cut-group', ['m-retry-0', 'm-retry-1', 'm-retry-2']); + cm.close(); + store.close(); + + const result = await createOfflineRecoveryBranch({ + storePath, + agentName: 'cairn', + contextId: partial[1], + branchName: 'recovery/cairn/cut-short', + }); + + assert.equal(result.messagesRemoved, 3); + assert.deepEqual(result.refs.map((ref) => ref.messageId), ['m-retry-0', 'm-retry-1', 'm-retry-2']); + const recoveredStore = JsStore.openOrCreate({ path: storePath }); + const recoveredCm = await ContextManager.open({ store: recoveredStore, namespace: 'agents/cairn' }); + assert.deepEqual( + recoveredCm.getAllMessages().map((message) => message.metadata?.messageId), + ['m-before', 'm-cut-0', 'm-cut-1'], + ); + recoveredCm.close(); + recoveredStore.close(); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); diff --git a/test/operator-surgery.test.ts b/test/operator-surgery.test.ts index aeb162b59..46bacc245 100644 --- a/test/operator-surgery.test.ts +++ b/test/operator-surgery.test.ts @@ -372,6 +372,189 @@ describe('live operator surgery', () => { void s0; }); + it('hide removes a message stored as a body whole, a one-shard body and every copy included', async () => { + // Each shard of a long message carries its Discord message id. + const appendBody = (discordId: string, group: string, texts: string[], from = 0): string[] => + texts.map((text, i) => + (cm() as unknown as ShardStore).messageStore.append( + 'user', + [{ type: 'text', text }], + { serverId: 'discord', channelId: 'discord:g1:c1', messageId: discordId }, + undefined, + { bodyGroupId: group, shardIndex: from + i }, + ).id, + ); + const [lone] = appendBody('d-lone', 'g-lone', ['lone']); + const long = appendBody('d-long', 'g-long', ['l0', 'l1', 'l2']); + // A write cut short after two shards, then its retry from shard 0. + const partial = appendBody('d-cut', 'g-cut', ['c0', 'c1']); + const retry = appendBody('d-cut', 'g-cut', ['c0', 'c1', 'c2']); + const internals = framework as unknown as { + handleHostCommand(serverId: string, params: Record): Promise>; + }; + const hide = (fromMessageId: string) => internals.handleHostCommand('discord', { + command: 'hide', + agentName: 'scout', + fromMessageId, + requesterName: 'antra', + }); + const refsOf = (result: Record) => + (result.hiddenRefs as Array<{ messageId: string }>).map((ref) => ref.messageId); + + const one = await hide('d-lone'); + assert.equal(one.ok, true, String(one.error)); + assert.equal(one.hidden, 1); + assert.equal(cm().getMessage(lone as never), null); + + const whole = await hide('d-long'); + assert.equal(whole.ok, true, String(whole.error)); + assert.equal(whole.hidden, 1, 'one message, though stored in three shards'); + assert.deepEqual(refsOf(whole), ['d-long']); + for (const id of long) assert.equal(cm().getMessage(id as never), null); + + const copies = await hide('d-cut'); + assert.equal(copies.ok, true, String(copies.error)); + assert.equal(copies.hidden, 2, 'the partial copy and its retry'); + assert.deepEqual(refsOf(copies), ['d-cut']); + for (const id of [...partial, ...retry]) assert.equal(cm().getMessage(id as never), null); + assert.deepEqual(remainingTexts(), ['m1', 'm2', 'm3', 'm4', 'm5']); + + // A body of more shards than the operator log keeps ids for: its entry + // keeps the first 50 and says the list is cut. + const huge = appendBody('d-huge', 'g-huge', Array.from({ length: 60 }, (_, i) => `h${i}`)); + const big = await hide('d-huge'); + assert.equal(big.ok, true, String(big.error)); + assert.equal(big.hidden, 1); + const entry = readLog(framework.getOperatorLogPath()!).filter((e) => e.kind === 'hide').at(-1)!; + assert.deepEqual(entry.result?.removedIds, huge.slice(0, 50)); + assert.equal(entry.result?.removedIdsTruncated, true); + for (const id of huge) assert.equal(cm().getMessage(id as never), null); + }); + + it('hide refuses while a surgery holds the store, and a missing Discord id matches nothing', async () => { + cm().addMessage('user', [{ type: 'text', text: 'held' }], { serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'd-held' }); + const internals = framework as unknown as { + handleHostCommand(serverId: string, params: Record): Promise>; + reserveStoreForSurgery(verb: string, agentName: string): () => void; + }; + const hide = (fromMessageId: string) => internals.handleHostCommand('discord', { + command: 'hide', agentName: 'scout', fromMessageId, requesterName: 'antra', + }); + const release = internals.reserveStoreForSurgery('roll back', 'scout'); + try { + const refused = await hide('d-held'); + assert.equal(refused.ok, false); + assert.match(String(refused.error), /a live roll back for scout holds the store/); + } finally { + release(); + } + // Two messages with no Discord id, such as the agent's replies: the + // text "undefined" names neither of them. + cm().addMessage('scout', [{ type: 'text', text: 'reply-1' }]); + cm().addMessage('scout', [{ type: 'text', text: 'reply-2' }]); + const none = await hide('undefined'); + assert.equal(none.ok, false); + assert.match(String(none.error), /is not an addressable message in context/); + assert.deepEqual(remainingTexts(), ['m1', 'm2', 'm3', 'm4', 'm5', 'held', 'reply-1', 'reply-2']); + const hidden = await hide('d-held'); + assert.equal(hidden.ok, true, String(hidden.error)); + assert.deepEqual(remainingTexts(), ['m1', 'm2', 'm3', 'm4', 'm5', 'reply-1', 'reply-2']); + }); + + it('hide over a range widens both ends to whole bodies, and counts each message once', async () => { + const appendBody = (discordId: string, group: string, texts: string[]): string[] => + texts.map((text, shardIndex) => + (cm() as unknown as ShardStore).messageStore.append( + 'user', + [{ type: 'text', text }], + { serverId: 'discord', channelId: 'discord:g1:c1', messageId: discordId }, + undefined, + { bodyGroupId: group, shardIndex }, + ).id, + ); + appendBody('d-a', 'g-a', ['a0', 'a1']); + cm().addMessage('user', [{ type: 'text', text: 'between' }], { serverId: 'discord', channelId: 'discord:g1:c1', messageId: 'd-mid' }); + appendBody('d-b', 'g-b', ['b0', 'b1', 'b2']); + const internals = framework as unknown as { + handleHostCommand(serverId: string, params: Record): Promise>; + }; + // From the second body's first shard back to the first body's: both + // widen to their whole bodies. + const result = await internals.handleHostCommand('discord', { + command: 'hide', + agentName: 'scout', + fromMessageId: 'd-b', + toMessageId: 'd-a', + requesterName: 'antra', + }); + assert.equal(result.ok, true, String(result.error)); + assert.equal(result.hidden, 3); + assert.deepEqual((result.hiddenRefs as Array<{ messageId: string }>).map((ref) => ref.messageId), ['d-a', 'd-mid', 'd-b']); + assert.deepEqual(remainingTexts(), ['m1', 'm2', 'm3', 'm4', 'm5']); + }); + + it("a module's removal of a message in a body takes the whole body, and a plain message alone, with no full read of the branch", async () => { + const [b0, b1, b2] = appendShards('g-mod', ['b0', 'b1', 'b2']); + const plain = String(cm().addMessage('user', [{ type: 'text', text: 'plain' }])); + const internals = framework as unknown as { removeMessage(id: string): void }; + // A full read inflates every attachment on the branch: the body is found + // in blob-free windows instead. + const reads = cm() as unknown as { getAllMessages(): unknown }; + const getAllMessages = reads.getAllMessages; + reads.getAllMessages = () => { throw new Error('read the whole branch'); }; + try { + internals.removeMessage(b1); + for (const id of [b0, b1, b2]) assert.equal(cm().getMessage(id as never), null); + internals.removeMessage(plain); + assert.equal(cm().getMessage(plain as never), null); + } finally { + reads.getAllMessages = getAllMessages; + } + assert.deepEqual(remainingTexts(), ['m1', 'm2', 'm3', 'm4', 'm5']); + }); + + it('a rewind whose newest message is the last shard of a body takes the whole body', async () => { + const [s0, s1, s2] = appendShards('g-doc', ['d0', 'd1', 'd2']); + const internals = framework as unknown as { + shedNewestTurn(agent: unknown): { kind: string; removedIds: string[] } | null; + }; + const record = internals.shedNewestTurn(framework.getAgent('scout')); + assert.ok(record); + assert.deepEqual(record.removedIds, [s2, s1, s0]); + assert.deepEqual(remainingTexts(), ['m1', 'm2', 'm3', 'm4', 'm5']); + }); + + it('a rewind of a tool result takes its tool_use turn whole, a turn stored as a body, in one removal', async () => { + // As context-manager's chunker stores them: the tool_use block on the + // turn's first shard, text after it; then the result, also a body. + const store = (cm() as unknown as ShardStore).messageStore; + const turn = [ + store.append('assistant', [{ type: 'tool_use', id: 't9', name: 'shell', input: {} }], {}, undefined, { bodyGroupId: 'g-turn', shardIndex: 0 }).id, + store.append('assistant', [{ type: 'text', text: 'narration' }], {}, undefined, { bodyGroupId: 'g-turn', shardIndex: 1 }).id, + ]; + const result = [ + store.append('user', [{ type: 'tool_result', toolUseId: 't9', content: 'r0' }], {}, undefined, { bodyGroupId: 'g-result', shardIndex: 0 }).id, + store.append('user', [{ type: 'text', text: 'r1' }], {}, undefined, { bodyGroupId: 'g-result', shardIndex: 1 }).id, + ]; + const c = cm() as unknown as { removeMessages: (a: string, b: string) => void }; + const ranges: Array<[string, string]> = []; + const original = c.removeMessages.bind(c); + c.removeMessages = (a: string, b: string) => { ranges.push([a, b]); original(a, b); }; + try { + const internals = framework as unknown as { + shedNewestTurn(agent: unknown): { kind: string; removedIds: string[] } | null; + }; + const record = internals.shedNewestTurn(framework.getAgent('scout')); + assert.ok(record); + assert.equal(record.kind, 'tool'); + assert.deepEqual(record.removedIds, [result[1], result[0], turn[1], turn[0]]); + assert.deepEqual(ranges, [[turn[0], result[1]]], "one removal, from the turn's first shard to the result's last"); + } finally { + c.removeMessages = original; + } + assert.deepEqual(remainingTexts(), ['m1', 'm2', 'm3', 'm4', 'm5']); + }); + // --- failure path ---------------------------------------------------------- it('a failed suppression restores the source branch, retires its outbox batch, and leaves the store bootable', async () => { diff --git a/test/overbudget-drain-breaker.test.ts b/test/overbudget-drain-breaker.test.ts index b21de6904..3defc651e 100644 --- a/test/overbudget-drain-breaker.test.ts +++ b/test/overbudget-drain-breaker.test.ts @@ -32,6 +32,7 @@ function makeHarness(tick: () => Promise) { fw.rewindEpisode = new Map(); fw.lastInferenceAt = new Map(); fw.overBudgetDrainInFlight = new Set(); + fw.overBudgetDrains = new Set(); fw.opsAlertLastSent = new Map(); fw.pendingRequests = []; fw.traceListeners = []; diff --git a/test/refusal-rewind.test.ts b/test/refusal-rewind.test.ts index a46a90f9f..0a8188e3a 100644 --- a/test/refusal-rewind.test.ts +++ b/test/refusal-rewind.test.ts @@ -21,6 +21,13 @@ function makeAgentHarness(messages: any[]) { const i = messages.findIndex((m) => m.id === id); if (i >= 0) messages.splice(i, 1); }, + // A range by its two ends, inclusive, as context-manager's removeMessages. + removeMessages: (fromId: string, toId: string) => { + const from = messages.findIndex((m) => m.id === fromId); + const to = messages.findIndex((m) => m.id === toId); + if (from < 0 || to < from) throw new Error(`no range ${fromId}..${toId}`); + for (const m of messages.splice(from, to - from + 1)) removed.push(m.id); + }, addMessage: (participant: string, content: any[], meta: any) => { const id = `marker-${added.length}`; added.push({ participant, content, meta }); diff --git a/test/stop-with-host-operations.test.ts b/test/stop-with-host-operations.test.ts index cfd9feba5..1685b13de 100644 --- a/test/stop-with-host-operations.test.ts +++ b/test/stop-with-host-operations.test.ts @@ -724,12 +724,41 @@ describe('stop() with an operator operation in flight', () => { const broadcasts: Array<[string, unknown]> = []; api.broadcast = (event, data) => { broadcasts.push([event, data]); }; // As after a rollback made outside the API server. - (api as unknown as { currentBranch: string }).currentBranch = 'stale'; + (api as unknown as { lastKnownBranch: string }).lastKnownBranch = 'stale'; await api.cmdBranchSwitch({ name: 'side' }); assert.deepEqual(broadcasts, [['branch:switched', { from: 'main', to: 'side' }]]); } finally { await framework.stop(); } }); + it('a switch from the API server that waits for running maintenance names the branch it moved to, and the current branch is the store\'s throughout', async () => { + const framework = await create(); + try { + const api = new ApiServer(framework) as unknown as { + cmdBranchCreate(params: { name: string }): Promise; + cmdBranchSwitch(params: { name: string }): Promise; + cmdBranchCurrent(): Promise<{ name: string }>; + broadcast(event: string, data: unknown): void; + }; + await api.cmdBranchCreate({ name: 'side' }); + const broadcasts: Array<[string, unknown]> = []; + api.broadcast = (event, data) => { broadcasts.push([event, data]); }; + // A maintenance pass still running: the switch moves the store once it ends. + let finish!: () => void; + internals(framework).maintenancePass = new Promise((resolve) => { finish = resolve; }); + const switching = api.cmdBranchSwitch({ name: 'side' }); + await new Promise((resolve) => setTimeout(resolve, 20)); + assert.equal(framework.getStore().currentBranch().name, 'main', 'the switch waits for the pass'); + assert.equal((await api.cmdBranchCurrent()).name, 'main'); + assert.deepEqual(broadcasts, []); + finish(); + internals(framework).maintenancePass = null; + await switching; + assert.equal(framework.getStore().currentBranch().name, 'side'); + assert.equal((await api.cmdBranchCurrent()).name, 'side'); + assert.deepEqual(broadcasts, [['branch:switched', { from: 'main', to: 'side' }]]); + } finally { await framework.stop(); } + }); + it('two API switches handled together each announce their own move', async () => { // With no workspace module, a switch holds the store for no await, so a // second one can move the store before the first's continuation runs. @@ -849,7 +878,7 @@ describe('stop() with an operator operation in flight', () => { const broadcasts: unknown[] = []; (api as unknown as { broadcast(event: string, data: unknown): void }).broadcast = (event, data) => { broadcasts.push([event, data]); }; // A stale cache, as after a rollback made outside the API server. - (api as unknown as { currentBranch: string }).currentBranch = 'stale'; + (api as unknown as { lastKnownBranch: string }).lastKnownBranch = 'stale'; // A turn holds its token from dequeue, before its stream registers. const tokens = internals(framework).activeTurnTokens; tokens.set('agent', -1);