diff --git a/CHANGELOG.md b/CHANGELOG.md index a7cf31a15..ffa091bd5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -711,7 +711,10 @@ framework. 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. + 0.14.0's changelog describes. The reset is for a store that already + opened on 0.14.0 and ran a committing compile, which re-stamped its + multiplier: `^0.14.0` now installs 0.14.1, which keeps a still-unstamped + multiplier provisional and replaces or drops it on its own. - **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 diff --git a/changelog.d/body-removals.fixed.md b/changelog.d/body-removals.fixed.md index d71c3d50a..1df396d8e 100644 --- a/changelog.d/body-removals.fixed.md +++ b/changelog.d/body-removals.fixed.md @@ -1 +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. +- 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). Its range form takes what was stored between its two messages, from the earlier one's first copy through the later one's first, and every copy of both, each whole, so a copy stored outside that span is taken on its own, and nothing else stored after the later message began is taken. Offline recovery's suppressed ranges (`suppressRanges`) take the same: before, a range ran through the later message's last copy, taking what was stored between that message's copies, and left out a copy of the earlier message stored after it. Either form's 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 index 8e997f2ea..ea46f82f3 100644 --- a/changelog.d/cm-0.14.changed.md +++ b/changelog.d/cm-0.14.changed.md @@ -1,2 +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. +- 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 logged, alerted and recorded in the maintenance history when it's first met, 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. A pass that met only failures already reported, and took no tick, isn't added to the history unless an operator's `maintain` (`maintenanceTick()`) asked for it, so a lasting failure no longer pushes out the runs that did work. `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. `stop()` waits for them the same way, a drain kick included, before it closes the store, so the tick in flight ends before the store closes, and once `stop()` has begun, no drain kick starts and none starts a new tick. Maintenance makes no provider call meanwhile, a conversation agent's included: one made under the hold, or once `stop()` has begun, is withdrawn without being sent, and so is a resident's call waiting at its provider gate when the hold is taken or `stop()` begins. That's no maintenance failure, and for a call outside context-manager 0.14's compression fallback ladder, a merge's or a chunk's canonical call with its retries after a `tool_use` stop, context-manager counts no attempt. One inside the ladder, after a chunk's canonical call was refused or cut short, is recorded there as a provider error, and once every rung has failed, the chunk's request family is quarantined, with its alert. Before, a resident's call made once `stop()` had begun was refused as a failure, so a restart that caught a maintenance tick before one of its calls logged a maintenance failure and raised a `context-maintenance-failed` alert; now the agent's maintenance logs that it stopped. An over-budget compile's `context-refusal` alert says whether the drain breaker kicked: `kicking`, `already draining`, or `not kicked` with the reason, where it said `kicking` whenever the agent was found, and `unavailable (no agent handle)` for one it can't find, as before. A kick joins what `stop()`, an operator surgery and `switchBranch` wait for before the alert is raised, so a listener on the alert that begins any of them waits for the kick, which then starts no tick. A merge keeps its place in the queue. When a chunk's compression is withdrawn outside the ladder, context-manager logs its "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 sent 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 index a4e121c59..dc016c453 100644 --- a/changelog.d/cm-0.14.fixed.md +++ b/changelog.d/cm-0.14.fixed.md @@ -1,2 +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. +- `previewAgentRuntimeSettings()` no longer throws when an agent's context manager (0.14, required since 0.21.1) refuses its queries, as after its strategy failed to initialize: it answers `{available: false, reason}`, with the agent's live budget and the patch's path, as it does for any preview it can't compute. So `resume()` no longer throws for such an agent either. It warns, naming the agent and the reason, 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/agent.ts b/src/agent.ts index 9780d879d..7e12d3636 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -496,6 +496,44 @@ export class Agent { return result; } + /** + * The budget compiles plan at, the path a patch's budget takes from it, and + * whether a descent is in flight, from the agent's own state alone: what a + * runtime-settings preview can still say while the context manager refuses + * its queries. + * @internal For the framework's runtime-settings preview; not a host API. + */ + runtimeBudgetPath(patch?: AgentRuntimeSettingsPatch): { + liveBudgetTokens: number; + path: 'immediate' | 'paced' | 'none'; + transition: 'stable' | 'converging'; + } { + const live = this.contextBudgetTokens ?? DEFAULT_CONTEXT_BUDGET_TOKENS; + const budget = patch?.contextBudgetTokens; + return { + liveBudgetTokens: live, + path: budget === undefined ? 'none' : budget >= live || patch?.immediate ? 'immediate' : 'paced', + transition: this.contextBudgetTargetTokens === undefined ? 'stable' : 'converging', + }; + } + + /** + * Whether the context manager answers its queries, as its readiness says. + * It refuses them while its strategy's initialization is pending or has + * failed, until a switchBranch or setStrategy succeeds, and a strategy can + * refuse its readiness on its own, as the autobiographical one does for a + * branch it hasn't reloaded. + * @internal For the framework's runtime-settings preview; not a host API. + */ + contextManagerAnswers(): boolean { + try { + this.contextManager.isReady(); + return true; + } catch { + return false; + } + } + /** * Read-only mirror of `updateRuntimeSettings`' budget semantics: derive what * the next compile would actually plan at if `patch` were applied, WITHOUT @@ -538,9 +576,8 @@ export class Agent { this.validateRuntimeSettingsPatch(patch); } const hot = this.getHotContextSettings(); - const live = this.contextBudgetTokens ?? DEFAULT_CONTEXT_BUDGET_TOKENS; + const { liveBudgetTokens: live, path } = this.runtimeBudgetPath(patch); const budget = patch?.contextBudgetTokens; - const path = budget === undefined ? 'none' : budget >= live || patch?.immediate ? 'immediate' : 'paced'; const overrides: Record = {}; if (patch?.tailTokens !== undefined) { @@ -757,6 +794,18 @@ export class Agent { } private validateRuntimeSettingsPatch(patch: AgentRuntimeSettingsPatch): void { + this.checkRuntimeSettingsPatch(patch); + if ((patch.tailTokens !== undefined || patch.transitionPaceTokens !== undefined) && + !this.getHotContextSettings()) { + throw new Error('The active context strategy does not support live tail/transition settings'); + } + } + + /** + * The checks of a runtime-settings patch that need nothing from the context manager. + * @internal For the framework's runtime-settings preview; not a host API. + */ + checkRuntimeSettingsPatch(patch: AgentRuntimeSettingsPatch): void { if (Object.keys(patch).length === 0) throw new Error('At least one setting is required'); if (patch.contextBudgetTokens !== undefined) { if (!Number.isSafeInteger(patch.contextBudgetTokens) || patch.contextBudgetTokens <= this.maxTokens) { @@ -776,10 +825,6 @@ export class Agent { patch.sameRoundThinkTextPolicy !== 'private') { throw new Error("sameRoundThinkTextPolicy must be 'public' or 'private'"); } - if ((patch.tailTokens !== undefined || patch.transitionPaceTokens !== undefined) && - !this.getHotContextSettings()) { - throw new Error('The active context strategy does not support live tail/transition settings'); - } } /** Settle a converging budget once the window is prepared; true if it settled. */ diff --git a/src/framework.ts b/src/framework.ts index 919f722fe..d4bce939c 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -721,22 +721,44 @@ 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). + * A maintenance provider call withdrawn because a surgery holds the store, or + * because stop() has closed provider admission: either is the framework's own + * doing, and no failure. It sent nothing: it was refused at the provider + * gate, or woken there from its wait. For a call outside context-manager + * 0.14's compression fallback ladder, a merge's or a chunk's canonical call + * with its retries after a tool_use stop, context-manager counts no attempt. + * A merge keeps its queue entry, so the next tick after the hold, or after + * the next start, 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). A call inside the ladder, after a chunk's canonical call + * was refused or cut short, is caught by the rung that made it and recorded + * as a provider error, since 0.14 rethrows only its own + * CompressionBranchDiscard there; once every rung has failed, the chunk's + * request family is quarantined, with its alert, and the tick returns. */ class MaintenanceWithdrawnError extends Error { - constructor(hold: { verb: string; agentName: string }) { - super(`maintenance call withdrawn: a live ${hold.verb} for ${hold.agentName} holds the store`); + constructor(reason: string) { + super(`maintenance call withdrawn: ${reason}`); this.name = 'MaintenanceWithdrawnError'; } } +/** + * A runtime-settings preview's reason, from what stopped it: the strategy's + * known throws as the codes `RuntimeSettingsPreview.reason` documents, and + * anything else verbatim. Message matching is fragile by nature, so an + * unknown message surfaces as the reason rather than being swallowed. + */ +function previewStoppedReason(error: unknown): string { + const message = error instanceof Error ? error.message : String(error); + if (message.includes('requires reinitialization for the current branch generation')) return 'branch_generation_changed'; + if (message.includes('already running; previews must not overlap')) return 'preview_in_flight'; + if (message.includes('requires adaptiveResolution')) return 'no_adaptive_resolution'; + return message; +} + class DiscordAwarenessAccountingError extends Error { constructor(operation: string, cause: unknown) { const detail = cause instanceof Error ? cause.message : String(cause); @@ -1558,6 +1580,10 @@ export class AgentFramework { * it was alerted, until its maintenance goes without one * (reportMaintenanceFailure). */ private maintenanceFailureReported = new Map(); + /** Run records whose failure repeated one already reported, and whose report was suppressed. */ + private maintenanceRepeats = new WeakSet(); + /** An operator's maintain (maintenanceTick) asked for the pass running or about to run: the run it records is kept in the history whatever it met. A pass with nothing to run records none. */ + private maintenancePassRequested = false; private providerGates: Map = new Map(); private providerAccelerationCooldowns: Map = new Map(); private providerAccelerationRecoveries: Map = new Map(); @@ -2541,7 +2567,8 @@ export class AgentFramework { * @param teardownTimeoutMs Optional budget for each wait on cleanup: an * operator's operations already running, then modules and MCPL together. * Omit to wait until cleanup settles, as required by existing module/child - * contracts. This does not bound earlier stream or event-loop quiescence, or + * contracts. This does not bound earlier stream, event-loop or maintenance + * quiescence (a maintenance pass or over-budget drain kick in flight), or * cancel cleanup. */ stop(teardownTimeoutMs = DEFAULT_SHUTDOWN_TIMEOUT_MS): Promise { @@ -2693,9 +2720,9 @@ export class AgentFramework { clearInterval(this.maintenanceTimer); this.maintenanceTimer = null; } - if (this.maintenancePass) { - await this.maintenancePass; - } + // The pass and any over-budget drain kick: a compression in flight lands + // before the store closes. + await this.settleMaintenancePass(); if (this.loopPromise) { await this.loopPromise; @@ -2835,8 +2862,11 @@ export class AgentFramework { } private async withAuxiliaryAdmission(agentName: string, run: () => Promise, model?: string): Promise { const gate = this.providerGate(agentName); - this.refuseHeldProviderModel(agentName, model); + // A surgery's hold or a stop withdraws the call before a provider wait + // can refuse it: either is the framework's own doing, and no failure. this.withdrawMaintenanceUnderSurgery(); + this.withdrawMaintenanceWhenStopping(agentName); + this.refuseHeldProviderModel(agentName, model); // 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. @@ -2851,9 +2881,7 @@ export class AgentFramework { // 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})`); - } + this.withdrawMaintenanceWhenStopping(agentName); // A wait recorded while this call was parked binds it too. this.refuseHeldProviderModel(agentName, model); gate.auxiliaryInFlight++; @@ -2883,7 +2911,19 @@ export class AgentFramework { * cooldown queues, which the hold itself keeps from starting. */ private withdrawMaintenanceUnderSurgery(): void { - if (this.surgeryHold) throw new MaintenanceWithdrawnError(this.surgeryHold); + if (this.surgeryHold) { + throw new MaintenanceWithdrawnError(`a live ${this.surgeryHold.verb} for ${this.surgeryHold.agentName} holds the store`); + } + } + /** + * Nor once stop() has closed provider admission: a stop that catches a + * tick is the framework stopping on purpose, so the call is withdrawn, as + * under a surgery's hold, and no failure is reported for it. + */ + private withdrawMaintenanceWhenStopping(agentName: string): void { + if (this.providerAdmissionClosed) { + throw new MaintenanceWithdrawnError(`provider admission closed while stopping (${agentName})`); + } } private auxiliaryMembraneFor(agentName: string): Membrane { return this.membraneAdmitting((run, model) => this.withAuxiliaryAdmission(agentName, run, model)); @@ -2896,10 +2936,16 @@ export class AgentFramework { * as a resident's are, so they bind its auxiliary calls too: a held model * is refused without a call, and an auxiliary call's own stated wait is * recorded for it. (An ephemeral run records nothing, its name being - * single-use, so its context manager keeps the plain membrane.) + * single-use, so its context manager keeps the plain membrane.) Its + * compression is the background work a resident's is, whether a + * maintenance tick or a drain after a message starts it, so it is withdrawn + * as a resident's is: a call made while a surgery holds the store, or once + * stop() has closed provider admission, is withdrawn without being sent. */ private conversationAuxiliaryMembraneFor(agentName: string): Membrane { return this.membraneAdmitting(async (run, model) => { + this.withdrawMaintenanceUnderSurgery(); + this.withdrawMaintenanceWhenStopping(agentName); this.refuseHeldProviderModel(agentName, model); try { return await run(); } catch (error) { @@ -3483,9 +3529,10 @@ export class AgentFramework { 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. + // A tick whose provider call a surgery or stop() withdrew ends this + // agent's work for the pass, as the hold or the stop 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 ( @@ -3532,7 +3579,13 @@ export class AgentFramework { run.finishedAt = Date.now(); run.agents.sort((a, b) => a.agentName.localeCompare(b.agentName)); this.currentMaintenanceRun = null; - this.maintenanceHistory.unshift(run); + // A pass that only met failures already reported, and took no tick, + // adds nothing the history doesn't hold, and a lasting failure would + // otherwise fill it in minutes, pushing out the runs that did work. One + // an operator asked for is kept: it's the pass their maintain reports. + const requested = this.maintenancePassRequested; + this.maintenancePassRequested = false; + if (requested || !run.agents.every((record) => record.ticks === 0 && this.maintenanceRepeats.has(record))) this.maintenanceHistory.unshift(run); if (this.maintenanceHistory.length > 50) this.maintenanceHistory.length = 50; } } @@ -3541,12 +3594,14 @@ 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). + * strategy failed to initialize, meets every pass, a few seconds apart. 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). A record whose + * failure isn't reported again is marked as a repeat, and a pass whose + * every record is a repeat that took no tick stays out of the history. */ private reportMaintenanceFailure(agentName: string, record: ContextMaintenanceAgentRun, wholeError: string): void { record.error = boundFailureText(wholeError); @@ -3558,7 +3613,10 @@ export class AgentFramework { // 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; + if (now - since < this.opsAlertCooldownMs) { + this.maintenanceRepeats.add(record); + return; + } } this.maintenanceFailureReported.set(agentName, { text: wholeError, at: now }); console.error(`[context-maintenance] agent=${agentName} failed:`, record.error); @@ -4355,7 +4413,8 @@ export class AgentFramework { `advance merges to go lower). Applying; the live compile budget is unchanged.`, ); } - } else if (preview.reason) { + } else if (preview.reason && agent.contextManagerAnswers()) { + // A manager that refuses its queries refuses the update below too. console.warn( `[budget-preflight] preview unavailable for ${agentName} (${preview.reason}); ` + `applying without preflight`, @@ -4674,26 +4733,16 @@ export class AgentFramework { if (this.maintenanceRunning()) await this.settleMaintenancePass(); const failing: Array<{ agentName: string; preview: RuntimeSettingsPreview }> = []; for (const agentName of this.agents.keys()) { - let preview: RuntimeSettingsPreview; - try { + // A preview that isn't available gives no verdict, and resume goes on + // for the others. That includes an agent whose context manager refuses + // its queries, as when its strategy failed to initialize: that manager + // refuses its compiles and writes until a switchBranch or setStrategy + // succeeds, so the agent can't run over budget, and maintenance reports it. + 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(() => {}); 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 }); @@ -5097,8 +5146,11 @@ export class AgentFramework { ); return { ...this.getContextMaintenanceSnapshot(), ran: false }; } + this.maintenancePassRequested = true; this.startQueuedMaintenance(); await this.maintenancePass; + // A pass with nothing queued records no run, and the next pass wasn't asked for. + this.maintenancePassRequested = false; return { ...this.getContextMaintenanceSnapshot(), ran: true }; } @@ -5295,9 +5347,14 @@ export class AgentFramework { * the compile budget and rejected safe-by-construction paced descents. * * Never throws for preview-layer reasons: strategy without previewContext, - * preview overlap, stale branch generation, and the hierarchical (non- - * adaptive) path all come back as `{available: false, reason}`. Unknown - * agent still throws, matching the sibling accessors. + * preview overlap, stale branch generation, the hierarchical (non- + * adaptive) path, and a context manager that refuses its queries (its + * strategy's initialization pending or failed) all come back as + * `{available: false, reason}`. Unknown agent still throws, matching the + * sibling accessors, and so does a patch whose values the agent refuses, + * whatever its context manager's state. Whether the strategy takes live + * tail and pace settings is the manager's to say, so while it refuses, a + * patch that sets them gets no verdict either. */ previewAgentRuntimeSettings( agentName: string, @@ -5313,8 +5370,22 @@ export class AgentFramework { agent: Agent, patch?: AgentRuntimeSettingsPatch, ): RuntimeSettingsPreview { - const plan = agent.planRuntimeSettings(patch); - const settings = agent.getRuntimeSettings(); + // A patch whose values the agent refuses throws, whatever its context + // manager's state. + if (patch && Object.keys(patch).length > 0) agent.checkRuntimeSettingsPatch(patch); + let plan: ReturnType; + let settings: ReturnType; + try { + plan = agent.planRuntimeSettings(patch); + settings = agent.getRuntimeSettings(); + } catch (error) { + // A manager that refuses its queries gives no verdict, as a preview + // that can't be computed gives none: the agent's own state still says + // its budget and the patch's path, and whatever stopped the preview is + // its reason. When the manager answers, anything else throws as before. + if (agent.contextManagerAnswers()) throw error; + return { available: false, reason: previewStoppedReason(error), ...agent.runtimeBudgetPath(patch) }; + } const base: RuntimeSettingsPreview = { available: false, path: plan.path, @@ -5338,20 +5409,7 @@ export class AgentFramework { } return { result: result as Record }; } catch (error) { - const message = error instanceof Error ? error.message : String(error); - // Structured mapping of the strategy's known throw paths. Message - // matching is fragile by nature, so unknown messages surface verbatim - // as the reason rather than being swallowed or rethrown. - if (message.includes('requires reinitialization for the current branch generation')) { - return { reason: 'branch_generation_changed' }; - } - if (message.includes('already running; previews must not overlap')) { - return { reason: 'preview_in_flight' }; - } - if (message.includes('requires adaptiveResolution')) { - return { reason: 'no_adaptive_resolution' }; - } - return { reason: message }; + return { reason: previewStoppedReason(error) }; } }; @@ -6983,8 +7041,26 @@ export class AgentFramework { return id === undefined || id === null || id === '' ? undefined : String(id); }; const byDiscordId = (did: string) => all.findIndex((m) => discordIdOf(m) === String(did)); + /** Runs in stored order, those that overlap or touch made one. */ + const merged = (runs: Array<{ from: number; to: number }>) => { + const out: Array<{ from: number; to: number }> = []; + for (const { from, to } of [...runs].sort((a, b) => a.from - b.from)) { + const last = out[out.length - 1]; + if (last && from <= last.to + 1) last.to = Math.max(last.to, to); + else out.push({ from, to }); + } + return out; + }; + // A message stored as a body (a long one, in shards) goes whole, and so + // does every copy of it: a write cut short leaves a partial body before + // its retry, both carrying the Discord id. + const copiesOf = (did: string) => { + const runs: Array<{ from: number; to: number }> = []; + all.forEach((m, i) => { if (discordIdOf(m) === String(did)) runs.push(bodyRangeIn(all, i)); }); + return runs; + }; - // Collect Discord refs of every message in [lo, hi] that carries one, + // Collect Discord refs of every message in the runs 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. @@ -7028,14 +7104,22 @@ export class AgentFramework { if (toIdx < 0) { return { ok: false, error: `Message ${params.toMessageId} is not an addressable message in context.` }; } - 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)); - removeStoredMessages(cm, all[lo], all[hi]); + // What's between the two messages, after the earlier one began and + // before the later one began: from the earlier one's first stored + // copy through the later one's first, each whole. Each end is every + // copy of it besides, as the single form takes, so a copy stored + // outside that span (a write cut short and retried later) is taken + // on its own, and nothing else stored after the later one began is + // taken. Offline recovery's range is the same. + const span = { + from: bodyRangeIn(all, Math.min(fromIdx, toIdx)).from, + to: bodyRangeIn(all, Math.max(fromIdx, toIdx)).to, + }; + const runs = merged([span, ...copiesOf(params.fromMessageId), ...copiesOf(params.toMessageId)]); + const refs = refsIn(runs); + const hidden = messagesIn(runs); + const rangeIds = 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} range removed=${hidden} ` + `(${params.fromMessageId}..${params.toMessageId}) by=${params.requesterName ?? params.requesterId ?? 'unknown'} (server=${serverId})`, @@ -7060,17 +7144,7 @@ export class AgentFramework { lastVisible: await this.lastVisiblePreview(agentName), }; } - // 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 runs = merged(copiesOf(params.fromMessageId)); const refs = refsIn(runs); const hidden = messagesIn(runs); const hiddenIds = runs.flatMap(({ from, to }) => all.slice(from, to + 1).map((m) => String(m.id))); @@ -15714,33 +15788,36 @@ export class AgentFramework { // this turn — surface it on the ops-alert pipeline (ops:alert trace + // webhook → fleet-watch) immediately, not only at the hard-down streak. // The per-(agent,kind) cooldown throttles repeats; failures.log already - // has the per-exhaustion record from logFailure above. - if (overBudget) { - this.opsAlert( - 'context-refusal', - agentName, - `compile refused (${errorType ?? 'unclassified'}): ${boundFailureText(wholeReason, FAILURE_SUMMARY_MAX_CHARS)} — ` + - `drain breaker ${agent ? 'kicking' : 'unavailable (no agent handle)'}; ` + - `if this persists, raise contextBudgetTokens or overBudgetGraceRatio.`, - { skipLog: true }, - ); - } - if (agent && overBudget && !this.overBudgetDrainInFlight.has(agentName) && !this.surgeryHold) { + // has the per-exhaustion record from logFailure above. It says whether + // the drain breaker kicks, decided where the kick is. What ends drains is + // stop(), not a framework that was never started: one run with + // runUntilIdle() has no maintenance timer, so a drain is its recovery. + const drainState = !overBudget ? undefined + : !agent ? 'unavailable (no agent handle)' + : this.providerAdmissionClosed ? 'not kicked: the framework is stopping' + : this.surgeryHold ? 'not kicked: a surgery holds the store' + : this.overBudgetDrainInFlight.has(agentName) ? 'already draining' + : 'kicking'; + if (agent && drainState === 'kicking') { this.overBudgetDrainInFlight.add(agentName); - // 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 () => { + // Like a maintenance pass: none starts once stop() has begun, it starts + // no tick while a surgery holds the store, and a surgery, a + // switchBranch or stop() waits for the tick in flight. It joins the + // drains they wait for before the alert is raised and before its body + // runs, so the alert's listeners find it: one that begins stop() or a + // surgery waits for it, and its body then ticks none and says why. So + // does a stop() made from inside its first tick, before that tick's + // first await. + const drain = Promise.resolve().then(async () => { let ticks = 0; try { const cm = agent.getContextManager(); - while (ticks < 8 && !this.surgeryHold) { + while (ticks < 8 && !this.providerAdmissionClosed && !this.surgeryHold) { await cm.tick(); ticks++; } - console.error( - `[inference-failed] drain kicked for ${agentName} (OverBudget breaker, ${ticks} ticks` + - `${ticks < 8 ? ', stopped while a surgery holds the store' : ''})`, - ); + const stopped = ticks >= 8 ? '' : this.providerAdmissionClosed ? ', stopped as the framework stops' : ', stopped while a surgery holds the store'; + console.error(`[inference-failed] drain kicked for ${agentName} (OverBudget breaker, ${ticks} ticks${stopped})`); } catch (err) { if (err instanceof MaintenanceWithdrawnError) { console.error(`[inference-failed] drain kicked for ${agentName} (OverBudget breaker, ${ticks} ticks, stopped: ${err.message})`); @@ -15750,10 +15827,20 @@ export class AgentFramework { } finally { this.overBudgetDrainInFlight.delete(agentName); } - })(); + }); this.overBudgetDrains.add(drain); void drain.finally(() => { this.overBudgetDrains.delete(drain); }); } + if (overBudget) { + this.opsAlert( + 'context-refusal', + agentName, + `compile refused (${errorType ?? 'unclassified'}): ${boundFailureText(wholeReason, FAILURE_SUMMARY_MAX_CHARS)} — ` + + `drain breaker ${drainState}; ` + + `if this persists, raise contextBudgetTokens or overBudgetGraceRatio.`, + { skipLog: true }, + ); + } // (3) Hard-down escalation on repeated identical failure. if (streak >= this.inferenceFailureEscalationThreshold) { diff --git a/src/recovery/offline-branch.ts b/src/recovery/offline-branch.ts index 49137f1af..0b34243f7 100644 --- a/src/recovery/offline-branch.ts +++ b/src/recovery/offline-branch.ts @@ -240,9 +240,7 @@ interface SuppressionInterval { interface MessageLocation { first: number; - last: number; firstId: StoredWindowMessage['id']; - lastId: StoredWindowMessage['id']; /** Usually one run; multiple runs tolerate duplicate external metadata. */ runs: SuppressionInterval[]; } @@ -405,9 +403,7 @@ function buildSuppressionPlan( }; locations.set(externalId, { first: index, - last: index, firstId: messages[i].id, - lastId: messages[i].id, runs: [run], }); } else { @@ -423,8 +419,6 @@ function buildSuppressionPlan( toId: messages[i].id, }); } - location.last = index; - location.lastId = messages[i].id; } } } @@ -447,12 +441,19 @@ function buildSuppressionPlan( const to = locations.get(range.toMessageId)!; const earlier = from.first <= to.first ? from : to; const later = from.first <= to.first ? to : from; + // What's between the two messages, after the earlier one began and + // before the later one began: from the earlier one's first stored copy + // through the later one's first. Each end is every copy of it besides, + // as a single suppression is, so a copy stored outside that span (a + // write cut short and retried later) is taken on its own, and nothing + // else stored after the later one began is taken. intervals.push({ start: earlier.first, - end: later.last, + end: later.first, fromId: earlier.firstId, - toId: later.lastId, + toId: later.firstId, }); + for (const end of [from, to]) intervals.push(...end.runs.map((run) => ({ ...run }))); } intervals.sort((a, b) => a.start - b.start || a.end - b.end); diff --git a/test/host-quiesce.test.ts b/test/host-quiesce.test.ts index 55213e976..aa6a1978b 100644 --- a/test/host-quiesce.test.ts +++ b/test/host-quiesce.test.ts @@ -329,6 +329,42 @@ test('resume proceeds without a verdict for an agent whose context manager refus await assert.rejects(cm.setStrategy(failing), /strategy initialization failed for the test/); assert.throws(() => cm.getHotContextSettings(), /requires successful strategy initialization/); + // The preview answers rather than throws: no verdict, with what the + // agent's own state still says. + // The agent runs at the default budget of 100,000 tokens, so 50,000 is a paced descent. + const preview = framework.previewAgentRuntimeSettings('agent', { contextBudgetTokens: 50_000 }); + assert.equal(preview.available, false); + assert.match(String(preview.reason), /requires successful strategy initialization/); + assert.equal(preview.path, 'paced'); + assert.equal(preview.liveBudgetTokens, 100_000); + assert.equal(preview.transition, 'stable'); + assert.equal(preview.effective, undefined); + // A patch whose values the agent refuses still throws. + assert.throws(() => framework.previewAgentRuntimeSettings('agent', { contextBudgetTokens: 5 }), /contextBudgetTokens must be/); + assert.throws(() => framework.previewAgentRuntimeSettings('agent', { tailTokens: -1 }), /tailTokens must be/); + // Whether the strategy takes a live tail or pace is the manager's to say, + // so while it refuses, a patch that sets one gets no verdict either. + for (const patch of [{ tailTokens: 2_000 }, { transitionPaceTokens: 1_000 }]) { + const unanswered = framework.previewAgentRuntimeSettings('agent', patch); + assert.equal(unanswered.available, false, JSON.stringify(patch)); + assert.match(String(unanswered.reason), /requires successful strategy initialization/); + assert.equal(unanswered.liveBudgetTokens, 100_000); + } + + // An update of the budget is refused, by the manager, and the preflight + // doesn't say it's applying one. + const preflight = t.mock.method(console, 'warn', () => {}); + try { + assert.throws( + () => framework.updateAgentRuntimeSettings('agent', { contextBudgetTokens: 50_000 }), + /requires successful strategy initialization/, + ); + assert.ok(!preflight.mock.calls.some((call) => String(call.arguments[0]).includes('applying without preflight'))); + } finally { + preflight.mock.restore(); + } + assert.equal(framework.getAgent('agent')!.runtimeBudgetPath().liveBudgetTokens, 100_000, 'the budget is as it was'); + const warn = t.mock.method(console, 'warn', () => {}); let status: Awaited>; try { @@ -339,10 +375,35 @@ test('resume proceeds without a verdict for an agent whose context manager refus 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") + return line.includes('feasibility preview unavailable for agent') && line.includes('requires successful strategy initialization') - && line.includes('proceeding without a verdict for it'); - }), 'resume warns, naming the agent and the error'); + && line.includes('proceeding without a verdict'); + }), 'resume warns, naming the agent and the reason'); + }); +}); + +test('a preview stopped by a stale branch generation says branch_generation_changed, as the reason documents', async () => { + const membrane = new MockMembrane(); + await withFramework(membrane, async (framework) => { + await framework.quiesce(); + // As the autobiographical strategy refuses a branch it hasn't reloaded: + // its readiness and its settings both say so. + const stale = 'TestStrategy.getHotContextSettings requires reinitialization for the current branch generation'; + const strategy = { + name: 'stale-generation-test', + async initialize() {}, + checkReadiness: () => { throw new Error(stale); }, + getHotContextSettings: () => { throw new Error(stale); }, + updateHotContextSettings: () => { throw new Error(stale); }, + async tick() {}, + select: () => [], + } as unknown as ContextStrategy; + const cm = framework.getAgent('agent')!.getContextManager(); + await cm.setStrategy(strategy); + const preview = framework.previewAgentRuntimeSettings('agent', { contextBudgetTokens: 50_000 }); + assert.equal(preview.available, false); + assert.equal(preview.reason, 'branch_generation_changed'); + assert.equal(preview.liveBudgetTokens, 100_000); }); }); diff --git a/test/maintenance-timer.test.ts b/test/maintenance-timer.test.ts index 8b5e6612d..138a9b0ba 100644 --- a/test/maintenance-timer.test.ts +++ b/test/maintenance-timer.test.ts @@ -25,6 +25,7 @@ import type { import { AgentFramework } from '../src/index.js'; import { OperatorActionError } from '../src/operator-log.js'; import type { NormalizedRequest, NormalizedResponse } from '@animalabs/membrane'; +import { MembraneError } from '@animalabs/membrane'; class QueuedStrategy implements ContextStrategy { readonly name = 'queued-test'; @@ -101,6 +102,29 @@ class FailingStrategy implements ContextStrategy { } } +/** A strategy whose every pass takes one tick, then meets the same failure. */ +class WorkThenFailStrategy implements ContextStrategy { + readonly name = 'work-then-fail-test'; + ticks = 0; + + checkReadiness(): ReadinessState { + return { ready: false, description: 'compression queued' }; + } + + async tick(_ctx: StrategyContext): Promise { + this.ticks++; + if (this.ticks % 2 === 0) throw new Error('provider rejected maintenance request'); + } + + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } +} + /** A strategy whose initialization always fails. */ class RefusingInitStrategy implements ContextStrategy { readonly name = 'refusing-init-test'; @@ -257,6 +281,7 @@ class GatedTickStrategy implements ContextStrategy { ticks = 0; ticksUnderHold = 0; branchAtFirstTickEnd?: string; + storeOpenAtFirstTickEnd?: boolean; framework?: AgentFramework; private open?: () => void; private readonly gate = new Promise((resolve) => { this.open = resolve; }); @@ -274,6 +299,7 @@ class GatedTickStrategy implements ContextStrategy { if ((this.framework as unknown as { surgeryHold: unknown }).surgeryHold) this.ticksUnderHold++; if (this.ticks === 1) { await this.gate; + this.storeOpenAtFirstTickEnd = !(this.framework as unknown as { store: { isClosed(): boolean } }).store.isClosed(); this.branchAtFirstTickEnd = this.framework!.getAgent('agent')!.getContextManager().currentBranch().name; } } @@ -329,6 +355,124 @@ class ParkedCallTickStrategy implements ContextStrategy { } } +/** A provider's stated wait of `ms`, as membrane reports a 429 with retry-after. */ +function heldFor(ms: number): MembraneError { + return new MembraneError({ + type: 'rate_limit', retryable: true, httpStatus: 429, retryAfterMs: ms, + message: 'rate limited for the test', rawError: { status: 429 }, + }); +} + +/** A strategy whose first tick calls stop() before its first await, then notes whether the store was still open when it ended. */ +class StopInTickStrategy implements ContextStrategy { + readonly name = 'stop-in-tick-test'; + ticks = 0; + framework?: AgentFramework; + stopping?: Promise; + stoppedBeforeTickEnd?: boolean; + storeOpenAtTickEnd?: boolean; + + checkReadiness(): ReadinessState { + return { ready: false, description: 'compression queued' }; + } + + async tick(_ctx: StrategyContext): Promise { + this.ticks++; + if (this.ticks > 1) return; + this.stopping = this.framework!.stop(); + // A stop() that doesn't wait for this tick settles first; one that does + // can't settle before it ends, so the second second is this tick's. + const first = await Promise.race([ + this.stopping.then(() => 'stopped' as const, () => 'stopped' as const), + new Promise<'waited'>((resolve) => setTimeout(() => resolve('waited'), 1000)), + ]); + this.stoppedBeforeTickEnd = first === 'stopped'; + this.storeOpenAtTickEnd = !(this.framework as unknown as { store: { isClosed(): boolean } }).store.isClosed(); + } + + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } +} + +/** A strategy whose every tick meets the same failure, its second only once its gate opens. */ +class GatedRepeatFailureStrategy implements ContextStrategy { + readonly name = 'gated-repeat-failure-test'; + ticks = 0; + private open?: () => void; + private readonly gate = new Promise((resolve) => { this.open = resolve; }); + + release(): void { + this.open?.(); + } + + checkReadiness(): ReadinessState { + return { ready: false, description: 'compression blocked' }; + } + + async tick(_ctx: StrategyContext): Promise { + this.ticks++; + if (this.ticks === 2) await this.gate; + throw new Error('provider rejected maintenance request'); + } + + select( + _store: MessageStoreView, + _log: ContextLogView, + _budget: TokenBudget, + ): ContextEntry[] { + return []; + } +} + +/** A strategy whose first tick waits on a gate, then calls its manager's membrane, as a compression would. */ +class CallAfterGateStrategy implements ContextStrategy { + readonly name = 'call-after-gate-test'; + ticks = 0; + entered = false; + outcome?: string; + private open?: () => void; + private readonly gate = new Promise((resolve) => { this.open = resolve; }); + + release(): void { + this.open?.(); + } + + checkReadiness(): ReadinessState { + return { ready: false, description: 'a compression queued' }; + } + + async tick(ctx: StrategyContext): Promise { + this.ticks++; + if (this.ticks > 1) return; + this.entered = true; + await this.gate; + try { + await ctx.membrane?.complete({ + model: 'test-model', + config: { model: 'test-model' }, + messages: [{ role: 'user', content: [{ type: 'text', text: 'compress' }] }], + } as unknown as NormalizedRequest); + this.outcome = 'sent'; + } 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'; @@ -588,11 +732,12 @@ describe('queued context maintenance timer', () => { pending: 'compression blocked', }, }); - // 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'); - } + // Every pass meets the same failure. It is alerted once, and the + // passes that only repeat it stay out of the history. + await waitFor(() => (framework as unknown as { maintenanceRunId: number }).maintenanceRunId >= 3); + const history = framework.getContextMaintenanceSnapshot().history; + assert.equal(history.length, 1); + assert.equal(history[0]!.agents[0]!.error, 'provider rejected maintenance request'); assert.equal(alerts.length, 1); } finally { await framework.stop(); @@ -801,9 +946,10 @@ describe('queued context maintenance timer', () => { await fence(); try { framework.start(); - await waitFor(() => runsWithError() >= 3); + await waitFor(() => passes >= 3); assert.deepEqual(alerts, ['context-maintenance-failed:fenced']); assert.equal(failedLines(), 1); + assert.equal(runsWithError(), 1, 'the passes that only repeat the failure stay out of the history'); // Recovered with nothing to do: passes find it ready and skip it. await cm.setStrategy(new ReadyStrategy()); @@ -1179,6 +1325,516 @@ describe('queued context maintenance timer', () => { } }); + for (const kind of ['resident', 'conversation agent'] as const) { + for (const held of [false, true]) { + it(`a ${kind}'s tick that reaches its provider under a surgery's hold is withdrawn${held ? ', though a provider wait holds its model' : ''}`, { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-surgery-withdrawn-')); + const provider = new CapturingMembrane(); + const tick = new CallAfterGateStrategy(); + const conversational = kind === 'conversation agent'; + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane: provider.asMembrane(), + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy: conversational ? new ReadyStrategy() : tick }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + ...(conversational ? { conversations: { templateAgent: 'agent', strategyFactory: () => tick } } : {}), + }); + const internals = framework as unknown as { + createConversationAgent(name: string, channelId: string): Promise; + reserveStoreForSurgery(verb: string, agentName: string): () => void; + recordProviderWait(agentName: string, model: string, error: Error): unknown; + }; + const name = conversational ? 'agent-conv-1' : 'agent'; + const alerts: string[] = []; + (framework as any).opsAlert = (kind: string, agentName: string) => alerts.push(`${kind}:${agentName}`); + const quiet = console.error; + console.error = () => {}; + try { + if (conversational) await internals.createConversationAgent(name, 'chan-1'); + if (held) internals.recordProviderWait(name, 'test-model', heldFor(60_000)); + framework.start(); + await waitFor(() => tick.entered); + const release = internals.reserveStoreForSurgery('roll back', 'agent'); + try { + tick.release(); + await waitFor(() => tick.outcome !== undefined); + assert.equal(tick.outcome, 'MaintenanceWithdrawnError'); + assert.equal(provider.calls.length, 0, 'the withdrawn call sent nothing'); + } finally { + release(); + } + assert.deepEqual(alerts.filter((alert) => alert.startsWith('context-maintenance-failed')), []); + } finally { + console.error = quiet; + tick.release(); + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + } + } + + it('a pass that only meets a failure already reported stays out of the history', { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-repeats-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy: new FailingStrategy() }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + (framework as any).opsAlert = () => {}; + const internals = framework as unknown as { maintenanceRunId: number }; + const quiet = console.error; + console.error = () => {}; + try { + framework.start(); + await waitFor(() => internals.maintenanceRunId >= 4); + const history = framework.getContextMaintenanceSnapshot().history; + assert.equal(history.length, 1, 'the first report only, though four passes met the failure'); + assert.equal(history[0]!.id, 1); + assert.ok(history[0]!.agents[0]!.error); + } finally { + console.error = quiet; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("an operator's maintain keeps the pass it ran in the history, though the pass only met a failure already reported", { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-requested-pass-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy: new FailingStrategy() }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 0, + }); + (framework as any).opsAlert = () => {}; + const quiet = console.error; + console.error = () => {}; + try { + framework.start(); + const first = await framework.maintenanceTick(); + assert.equal(first.ran, true); + const reported = first.history[0]!.id; + assert.ok(first.history[0]!.agents[0]!.error, 'the first pass reports the failure'); + const second = await framework.maintenanceTick(); + assert.equal(second.ran, true); + assert.ok(second.history[0]!.id > reported, 'the newest run is the pass the second maintain ran'); + assert.ok(second.history[0]!.agents[0]!.error); + } finally { + console.error = quiet; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + + it("a maintain whose pass had nothing to run leaves no request for the next pass", { timeout: 10_000 }, async () => { + // Gordon-2414's probe. While the agent's own call holds its provider gate, + // a maintain's pass queues nothing and records no run, so the next pass, + // which nobody asked for and which meets only the failure already + // reported, stays out of the history. + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-request-reset-')); + const strategy = new FailingStrategy(); + let ticks = 0; + const failingTick = strategy.tick.bind(strategy); + strategy.tick = async (ctx: StrategyContext) => { ticks++; return failingTick(ctx); }; + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 0, + }); + const internals = framework as unknown as { + acquirePrimaryProviderGate(agentName: string): void; + releasePrimaryProviderGate(agentName: string): void; + startQueuedMaintenance(): void; + maintenancePass: Promise | null; + }; + (framework as any).opsAlert = () => {}; + const quiet = console.error; + const quietWarn = console.warn; + console.error = () => {}; + console.warn = () => {}; + try { + framework.start(); + const first = await framework.maintenanceTick(); + assert.equal(first.history.length, 1, 'the first maintain reports the failure'); + internals.acquirePrimaryProviderGate('agent'); + const second = await framework.maintenanceTick(); + assert.equal(second.ran, true); + assert.equal(second.history.length, 1, 'a pass with nothing queued records no run'); + assert.equal(ticks, 1, 'the second maintain ran no tick'); + internals.releasePrimaryProviderGate('agent'); + internals.startQueuedMaintenance(); + await internals.maintenancePass; + assert.equal(ticks, 2, 'the pass nobody asked for ran its tick'); + assert.equal(framework.getContextMaintenanceSnapshot().history.length, 1, 'the repeat nobody asked for stays out of the history'); + } finally { + console.error = quiet; + console.warn = quietWarn; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("a maintain that joins a pass already running keeps that pass in the history", { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-request-join-')); + const strategy = new GatedRepeatFailureStrategy(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 0, + }); + const internals = framework as unknown as { startQueuedMaintenance(): void }; + (framework as any).opsAlert = () => {}; + const quiet = console.error; + const quietWarn = console.warn; + console.error = () => {}; + console.warn = () => {}; + try { + framework.start(); + const first = await framework.maintenanceTick(); + assert.equal(first.history.length, 1, 'the first maintain reports the failure'); + // A pass nobody asked for, whose tick waits on the gate. + internals.startQueuedMaintenance(); + await waitFor(() => strategy.ticks === 2); + // The maintain joins it, and the pass then meets the failure already reported. + const joining = framework.maintenanceTick(); + strategy.release(); + const joined = await joining; + assert.equal(joined.ran, true); + assert.equal(strategy.ticks, 2, 'the maintain started no pass of its own'); + assert.equal(joined.history.length, 2, 'the pass the maintain joined is kept'); + assert.ok(joined.history[0]!.id > first.history[0]!.id); + } finally { + console.error = quiet; + console.warn = quietWarn; + strategy.release(); + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + it('a pass that took a tick before meeting a failure already reported stays in the history', { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-repeats-after-work-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy: new WorkThenFailStrategy() }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + }); + (framework as any).opsAlert = () => {}; + const internals = framework as unknown as { maintenanceRunId: number }; + const quiet = console.error; + console.error = () => {}; + try { + framework.start(); + await waitFor(() => internals.maintenanceRunId >= 4); + const history = framework.getContextMaintenanceSnapshot().history; + assert.ok(history.length >= 3, `each finished pass did work, so each is kept (${history.length})`); + for (const run of history) { + assert.equal(run.agents[0]!.ticks, 1); + assert.ok(run.agents[0]!.error); + } + } finally { + console.error = quiet; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + for (const kind of ['resident', 'conversation agent'] as const) for (const held of [false, true]) { + it(`a ${kind}'s maintenance call made once stop() has begun is withdrawn${held ? ', though a provider wait holds its model,' : ''} and no failure is reported`, { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-stopping-')); + const provider = new CapturingMembrane(); + const tick = new CallAfterGateStrategy(); + const conversational = kind === 'conversation agent'; + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane: provider.asMembrane(), + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy: conversational ? new ReadyStrategy() : tick }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 10, + ...(conversational ? { conversations: { templateAgent: 'agent', strategyFactory: () => tick } } : {}), + }); + const internals = framework as unknown as { + createConversationAgent(name: string, channelId: string): Promise; + recordProviderWait(agentName: string, model: string, error: Error): unknown; + }; + const name = conversational ? 'agent-conv-1' : 'agent'; + const alerts: string[] = []; + (framework as any).opsAlert = (...args: unknown[]) => { alerts.push(args.map(String).join(' ')); }; + const quiet = console.error; + const logged: string[] = []; + console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; + let stopping: Promise | undefined; + try { + if (conversational) await internals.createConversationAgent(name, 'chan-1'); + if (held) internals.recordProviderWait(name, 'test-model', heldFor(60_000)); + framework.start(); + await waitFor(() => tick.entered); + stopping = framework.stop(); + tick.release(); + await waitFor(() => tick.outcome !== undefined); + await stopping; + assert.equal(tick.outcome, 'MaintenanceWithdrawnError'); + assert.equal(provider.calls.length, 0, 'the withdrawn call sent nothing'); + assert.ok(logged.some((line) => line.includes( + `[context-maintenance] agent=${name} stopped: maintenance call withdrawn: provider admission closed while stopping (${name})`, + )), logged.join('\n')); + assert.deepEqual(alerts, [], 'no alert'); + for (const run of framework.getContextMaintenanceSnapshot().history) { + for (const record of run.agents) assert.equal(record.error, undefined, `no failure recorded for ${record.agentName}`); + } + } finally { + console.error = quiet; + tick.release(); + await (stopping ?? framework.stop()); + rmSync(dir, { recursive: true, force: true }); + } + }); + } + + it('stop() withdraws a maintenance call parked at the provider gate, and that is no failure', { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-parked-stop-')); + 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 = (...args: unknown[]) => { alerts.push(args.map(String).join(' ')); }; + const internals = framework as unknown as { providerGate(name: string): { auxiliaryWaiters: unknown[] } }; + const quiet = console.error; + console.error = () => {}; + try { + framework.start(); + await waitFor(() => internals.providerGate('agent').auxiliaryWaiters.length === 1); + await framework.stop(); + assert.equal(strategy.outcome, 'MaintenanceWithdrawnError'); + assert.equal(strategy.calls, 0, 'the withdrawn call sent nothing'); + assert.deepEqual(alerts, []); + } finally { + console.error = quiet; + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('an over-budget drain kick starts in a framework that was never started, as one run with runUntilIdle() is', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-drain-unstarted-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy: new ReadyStrategy() }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 0, + }); + const internals = framework as unknown as { + noteInferenceExhausted(agentName: string, reason: string, retryable?: boolean, errorType?: string): void; + overBudgetDrains: Set>; + }; + const alerts: string[] = []; + (framework as any).opsAlert = (...args: unknown[]) => { alerts.push(args.map(String).join(' ')); }; + const quiet = console.error; + const logged: string[] = []; + console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; + try { + internals.noteInferenceExhausted('agent', 'compile refused', false, 'over_budget'); + assert.equal(internals.overBudgetDrains.size, 1, 'a kick was started'); + assert.ok(alerts.some((alert) => alert.includes('drain breaker kicking;')), alerts.join('\n')); + await Promise.all([...internals.overBudgetDrains]); + assert.ok(logged.some((line) => line.includes('drain kicked for agent (OverBudget breaker, 8 ticks)')), logged.join('\n')); + } finally { + console.error = quiet; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("a stop() begun from inside the refusal's alert waits for the kick the alert reports, which ticks none", { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-drain-stop-in-alert-')); + 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 internals = framework as unknown as { + noteInferenceExhausted(agentName: string, reason: string, retryable?: boolean, errorType?: string): void; + overBudgetDrains: Set>; + }; + const alerts: string[] = []; + let drainsAtAlert = -1; + let stopping: Promise | undefined; + // An ops:alert listener that stops the framework, as the alert is raised. + (framework as any).opsAlert = (kind: string, _agentName: string, message: string) => { + if (kind !== 'context-refusal') return; + alerts.push(message); + drainsAtAlert = internals.overBudgetDrains.size; + stopping ??= framework.stop(); + }; + const logged: string[] = []; + const quiet = console.error; + console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; + try { + internals.noteInferenceExhausted('agent', 'compile refused', false, 'over_budget'); + assert.ok(stopping, 'the alert began stop()'); + assert.match(alerts[0] ?? '', /drain breaker kicking;/); + assert.equal(drainsAtAlert, 1, 'the kick joined the drains stop() waits for before the alert'); + await stopping; + assert.equal(strategy.ticks, 0, 'the kick ticked none'); + assert.ok( + logged.some((line) => line.includes('drain kicked for agent (OverBudget breaker, 0 ticks, stopped as the framework stops)')), + 'the kick says it stopped, and why', + ); + } finally { + console.error = quiet; + strategy.release(); + await (stopping ?? framework.stop()); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("a surgery begun from inside the refusal's alert waits for the kick the alert reports, which ticks none", { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-drain-surgery-in-alert-')); + 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 { + noteInferenceExhausted(agentName: string, reason: string, retryable?: boolean, errorType?: string): void; + overBudgetDrains: Set>; + }; + const alerts: string[] = []; + let drainsAtAlert = -1; + let moving: Promise | undefined; + // An ops:alert listener that rolls the agent back, as the alert is raised. + (framework as any).opsAlert = (kind: string, _agentName: string, message: string) => { + if (kind !== 'context-refusal') return; + alerts.push(message); + drainsAtAlert = internals.overBudgetDrains.size; + moving ??= framework.rollbackToMessage('agent', { messageId: first }); + }; + const logged: string[] = []; + const quiet = console.error; + console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; + try { + internals.noteInferenceExhausted('agent', 'compile refused', false, 'over_budget'); + assert.ok(moving, 'the alert began a rollback'); + assert.match(alerts[0] ?? '', /drain breaker kicking;/); + assert.equal(drainsAtAlert, 1, 'the kick joined the drains the rollback waits for before the alert'); + await moving; + assert.equal(strategy.ticks, 0, 'the kick ticked none'); + assert.ok( + logged.some((line) => line.includes('drain kicked for agent (OverBudget breaker, 0 ticks, stopped while a surgery holds the store)')), + 'the kick says it stopped, and why', + ); + assert.notEqual(cm.currentBranch().name, 'main', 'the rollback went on once the kick ended'); + } finally { + console.error = quiet; + strategy.release(); + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + it("a stop() made from a drain's first tick, before that tick's first await, waits for the tick", { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-drain-stop-in-tick-')); + const strategy = new StopInTickStrategy(); + 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 internals = framework as unknown as { + noteInferenceExhausted(agentName: string, reason: string, retryable?: boolean, errorType?: string): void; + }; + (framework as any).opsAlert = () => {}; + const quiet = console.error; + console.error = () => {}; + try { + internals.noteInferenceExhausted('agent', 'compile refused', false, 'over_budget'); + await waitFor(() => strategy.stopping !== undefined); + await strategy.stopping; + assert.equal(strategy.stoppedBeforeTickEnd, false, 'stop() waited for the tick'); + assert.equal(strategy.storeOpenAtTickEnd, true, 'the tick ended before the store closed'); + } finally { + console.error = quiet; + await (strategy.stopping ?? framework.stop()); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('no over-budget drain kick starts once stop() has begun', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-drain-after-stop-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane, + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test', strategy: new ReadyStrategy() }], + modules: [], + syncIntervalMs: 0, + maintenanceIntervalMs: 0, + }); + const internals = framework as unknown as { + noteInferenceExhausted(agentName: string, reason: string, retryable?: boolean, errorType?: string): void; + overBudgetDrains: Set>; + }; + const alerts: string[] = []; + (framework as any).opsAlert = (...args: unknown[]) => { alerts.push(args.map(String).join(' ')); }; + const quiet = console.error; + const logged: string[] = []; + console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; + try { + framework.start(); + const stopping = framework.stop(); + internals.noteInferenceExhausted('agent', 'compile refused', false, 'over_budget'); + assert.equal(internals.overBudgetDrains.size, 0, 'no kick was started'); + assert.ok(alerts.some((alert) => alert.startsWith('context-refusal agent ') + && alert.includes('drain breaker not kicked: the framework is stopping;')), alerts.join('\n')); + await stopping; + assert.ok(!logged.some((line) => line.includes('drain kicked')), logged.join('\n')); + } finally { + console.error = quiet; + 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(); @@ -1261,6 +1917,46 @@ describe('queued context maintenance timer', () => { } }); + it("stop() waits for an over-budget drain kick's tick in flight, and the kick ticks no more once stop() has begun", { timeout: 10_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-maintenance-drain-stop-')); + 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 internals = framework as unknown as { + noteInferenceExhausted(agentName: string, reason: string, retryable?: boolean, errorType?: string): void; + }; + const quiet = console.error; + console.error = () => {}; + let stopped = false; + try { + framework.start(); + // 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 stopping = framework.stop().then(() => { stopped = true; }); + // A stop() that didn't wait would close the store with the tick held: + // it gets half a second to. + await waitFor(() => stopped, 500).catch(() => {}); + assert.equal(stopped, false, 'stop() waits for the kick\'s tick in flight'); + strategy.release(); + await stopping; + assert.equal(strategy.storeOpenAtFirstTickEnd, true, 'the tick in flight ended before the store closed'); + assert.equal(strategy.ticks, 1, 'the kick starts no tick once stop() has begun'); + } finally { + console.error = quiet; + strategy.release(); + if (!stopped) 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(); diff --git a/test/offline-recovery-branch.test.ts b/test/offline-recovery-branch.test.ts index 640cc0f59..24e7bf453 100644 --- a/test/offline-recovery-branch.test.ts +++ b/test/offline-recovery-branch.test.ts @@ -190,6 +190,98 @@ test('offline recovery suppresses inclusive ranges in context order', async () = } }); +test('a suppressed range takes every copy of both its messages, and leaves what was stored after the later one began', async () => { + const dir = mkdtempSync(join(tmpdir(), 'offline-recovery-range-copies-')); + const storePath = join(dir, 'agent.chronicle'); + try { + const store = JsStore.openOrCreate({ path: storePath }); + const cm = await ContextManager.open({ store, namespace: 'agents/cairn' }); + const add = (text: string, messageId: string) => cm.addMessage('user', [{ type: 'text', text }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId, + }); + add('keep', 'm-keep'); + // B was stored twice, a write cut short and retried, with Z between: + // Z came after B began, so it isn't between A and B. + add('A', 'm-a'); + add('B, cut short', 'm-b'); + add('Z', 'm-z'); + add('B, retried', 'm-b'); + // C's retry was stored after D and E: it goes too, on its own. + add('C, cut short', 'm-c'); + add('D', 'm-d'); + add('E', 'm-e'); + add('C, retried', 'm-c'); + add('current anchor', 'm-current'); + cm.close(); + store.close(); + + const result = await createOfflineRecoveryBranch({ + storePath, + agentName: 'cairn', + messageId: 'm-current', + suppressRanges: [ + { fromMessageId: 'm-a', toMessageId: 'm-b' }, + { fromMessageId: 'm-d', toMessageId: 'm-c' }, + ], + branchName: 'recovery/cairn/range-copies', + }); + + assert.equal(result.messagesSuppressed, 6); + assert.deepEqual(result.refs.map((ref) => ref.messageId), ['m-a', 'm-b', 'm-c', 'm-d']); + 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-keep', 'm-z', 'm-e', 'm-current'], + ); + recoveredCm.close(); + recoveredStore.close(); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('a suppressed range whose two messages are stored in shards takes each whole', async () => { + const dir = mkdtempSync(join(tmpdir(), 'offline-recovery-range-shards-')); + const storePath = join(dir, 'agent.chronicle'); + try { + const store = JsStore.openOrCreate({ path: storePath }); + const cm = await ContextManager.open({ store, namespace: 'agents/cairn' }); + const add = (text: string, messageId: string) => cm.addMessage('user', [{ type: 'text', text }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId, + }); + add('keep', 'm-keep'); + // Each shard of a long message carries its Discord message id. + appendShards(cm, 'body-a', ['m-a', 'm-a']); + add('between', 'm-between'); + appendShards(cm, 'body-b', ['m-b', 'm-b', 'm-b']); + add('after', 'm-after'); + add('current anchor', 'm-current'); + cm.close(); + store.close(); + + const result = await createOfflineRecoveryBranch({ + storePath, + agentName: 'cairn', + messageId: 'm-current', + suppressRanges: [{ fromMessageId: 'm-b', toMessageId: 'm-a' }], + branchName: 'recovery/cairn/range-shards', + }); + + assert.equal(result.messagesSuppressed, 6, 'both bodies, every shard, and what lies between'); + 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-keep', 'm-after', 'm-current'], + ); + recoveredCm.close(); + recoveredStore.close(); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + test('offline recovery accepts an exact internal context ID anchor', async () => { const dir = mkdtempSync(join(tmpdir(), 'offline-recovery-context-id-')); const storePath = join(dir, 'agent.chronicle'); diff --git a/test/operator-surgery.test.ts b/test/operator-surgery.test.ts index 46bacc245..f56ba01ad 100644 --- a/test/operator-surgery.test.ts +++ b/test/operator-surgery.test.ts @@ -429,6 +429,57 @@ describe('live operator surgery', () => { 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); + + // The range form takes what was stored between its two messages, from the + // earlier one's first copy through the later one's first, in either + // order, and every copy of both: a range ending at a message stored twice + // takes its retry too. + const hideRange = (fromMessageId: string, toMessageId: string) => internals.handleHostCommand('discord', { + command: 'hide', agentName: 'scout', fromMessageId, toMessageId, requesterName: 'antra', + }); + const plain = (text: string, discordId: string) => cm().addMessage('user', [{ type: 'text', text }], { + serverId: 'discord', channelId: 'discord:g1:c1', messageId: discordId, + }); + appendBody('d-cut2', 'g-cut2', ['p0', 'p1']); + appendBody('d-cut2', 'g-cut2', ['r0', 'r1', 'r2']); + plain('after', 'd-after'); + const forward = await hideRange('d4', 'd-cut2'); + assert.equal(forward.ok, true, String(forward.error)); + assert.equal(forward.hidden, 4, 'm4, m5, and both copies of the cut message'); + assert.deepEqual(refsOf(forward), ['d4', 'd5', 'd-cut2']); + assert.deepEqual(remainingTexts(), ['m1', 'm2', 'm3', 'after']); + + appendBody('d-cut3', 'g-cut3', ['q0']); + appendBody('d-cut3', 'g-cut3', ['s0', 's1']); + plain('last', 'd-last'); + const backward = await hideRange('d-cut3', 'd2'); + assert.equal(backward.ok, true, String(backward.error)); + assert.equal(backward.hidden, 5, 'm2, m3, after, and both copies of the cut message'); + assert.deepEqual(remainingTexts(), ['m1', 'last']); + + // What was stored between a message's first copy and its retry came after + // that message began, so a range ending at it leaves that in. + plain('A', 'd-a'); + appendBody('d-b', 'g-b', ['bp0']); + plain('Z', 'd-z'); + appendBody('d-b', 'g-b', ['br0', 'br1']); + const between = await hideRange('d-a', 'd-b'); + assert.equal(between.ok, true, String(between.error)); + assert.equal(between.hidden, 3, 'A, and both copies of B'); + assert.deepEqual(refsOf(between), ['d-a', 'd-b']); + assert.deepEqual(remainingTexts(), ['m1', 'last', 'Z']); + + // A copy of the earlier message stored after the later one goes too, on + // its own: what was stored between them stays. + appendBody('d-c', 'g-c', ['cp0']); + plain('D', 'd-d'); + plain('E', 'd-e'); + appendBody('d-c', 'g-c', ['cr0', 'cr1']); + const mirror = await hideRange('d-d', 'd-c'); + assert.equal(mirror.ok, true, String(mirror.error)); + assert.equal(mirror.hidden, 3, 'both copies of C, and D'); + assert.deepEqual(refsOf(mirror), ['d-c', 'd-d']); + assert.deepEqual(remainingTexts(), ['m1', 'last', 'Z', 'E']); }); it('hide refuses while a surgery holds the store, and a missing Discord id matches nothing', async () => {