diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c97568f85..dd9f76898 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,7 +15,7 @@ jobs: runs-on: ${{ matrix.os }} # Without a limit, a step that hangs leaves the check pending until # GitHub's six-hour limit. Passing runs take under seven minutes. A - # test file stuck under Node 20, or a test stuck under Node 24, fails + # test file stuck under Node 22.12, or a test stuck under Node 24, fails # sooner and by name, through npm test's --test-timeout. timeout-minutes: 20 strategy: @@ -24,11 +24,17 @@ jobs: # node_modules tree records only that machine's native binaries, so a # lock made on macOS breaks Linux installs and vice versa. Each runner # catches the lock the other's platform produced. Tests run on ubuntu. - # Node: the engines floor and current LTS — stdio flushing semantics - # around process exit differ between versions and platforms. + # Node: the engines floor, pinned exactly; the newest Node the range + # admits; and, between them, every maintained major line (24), as + # membrane and context-manager test theirs. stdio flushing semantics + # around process exit differ between versions and platforms. macOS runs + # the two ends only. matrix: os: [ubuntu-latest, macos-latest] - node: [20, 24] + node: ['22.12.0', 24, latest] + exclude: + - os: macos-latest + node: 24 steps: - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6.1.0 @@ -61,7 +67,7 @@ jobs: # leaves running, such as a late arrival after stop() ("Queue is # closed"). These files run unforced so that class fails here, not in a # later PR's qualification. They get npm test's --test-timeout too: - # under Node 20 it fails a file whose leftover work keeps it alive, by + # under Node 22.12 it fails a file whose leftover work keeps it alive, by # name, and under Node 24 the job's limit ends such a file. - name: Test unforced (work in flight at stop) run: node --test --test-timeout=600000 dist/test/stop-with-work-in-flight.test.js dist/test/stop-with-host-operations.test.js dist/test/conversation-routing.test.js @@ -78,9 +84,17 @@ jobs: # hours: above the check's own limits (10 minutes each for npm pack and npm # install, a minute for each load), so the check names what hung. pack-check: - name: Packed release installs on its own + name: Packed release installs on its own (node ${{ matrix.node }}) runs-on: ubuntu-latest timeout-minutes: 40 + # The engines floor and current LTS: the check's load hook goes in through + # register below Node 22.15 and through registerHooks from there, so each + # branch a host takes runs before a merge, as publish.yml runs both before + # a release. + strategy: + fail-fast: false + matrix: + node: ['22.12.0', 24] steps: - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6.1.0 @@ -90,7 +104,7 @@ jobs: - name: Setup Node.js uses: actions/setup-node@249970729cb0ef3589644e2896645e5dc5ba9c38 # v6.5.0 with: - node-version: 24 + node-version: ${{ matrix.node }} # As publish installs, so a build that came to need a dependency's # install script fails here, before a tag, rather than in publish. diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index ae7e227cb..f56ed0abe 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -27,7 +27,7 @@ jobs: - name: Setup Node.js uses: actions/setup-node@249970729cb0ef3589644e2896645e5dc5ba9c38 # v6.5.0 with: - node-version: 20 + node-version: '22.12.0' # The same check `npm version` runs first (see CONTRIBUTING.md): an # open issue labelled release-hold stops the release. Both jobs that @@ -64,11 +64,18 @@ jobs: # 6 hours: above the check's own limits (10 minutes each for npm pack and # npm install, a minute for each load), so the check names what hung. pack-check: - name: Packed release installs on its own + name: Packed release installs on its own (node ${{ matrix.node }}) runs-on: ubuntu-latest timeout-minutes: 40 permissions: contents: read + # The engines floor and current LTS: the check's load hook goes in through + # register below Node 22.15 and through registerHooks from there, so each + # branch a host takes runs before a release. + strategy: + fail-fast: false + matrix: + node: ['22.12.0', 24] steps: - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6.1.0 @@ -78,7 +85,7 @@ jobs: - name: Setup Node.js uses: actions/setup-node@249970729cb0ef3589644e2896645e5dc5ba9c38 # v6.5.0 with: - node-version: 24 + node-version: ${{ matrix.node }} # As publish installs, so a build that came to need a dependency's # install script fails here rather than in publish. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 60640fb41..685b7e688 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -215,10 +215,11 @@ npx tsc --noEmit # typecheck code you no longer have — always build before testing, and after switching branches. The `pretest` hook copies `test/fixtures/*.mjs` into `dist/`. -`--test-timeout=600000` bounds each test file as a whole under Node 20, -and each test under Node 24, at ten minutes; the slowest file, -`mcpl-runtime-lifecycle`, takes about three. Each job in `ci.yml` also has -a `timeout-minutes` limit, which ends a hang the test timeout doesn't reach. +`--test-timeout=600000` bounds each test file as a whole under Node +22.12, as it did under Node 20, and each test under Node 24, at ten +minutes; the slowest file, `mcpl-runtime-lifecycle`, takes about three. +Each job in `ci.yml` also has a `timeout-minutes` limit, which ends a hang +the test timeout doesn't reach. Push-time CI (`ci.yml`) builds, typechecks and tests every push and PR on ubuntu and macos. Lockfiles are deliberately gitignored in this repo, so CI @@ -231,3 +232,8 @@ supply, fails the PR. When working against local sibling checkouts rather than published versions, see the ecosystem workspace setup — `bun install` at the `rust-connectome` root links the four packages together. + + +CI check contexts follow the workflow matrices. When a matrix changes, a +maintainer should check that the repository's required-status list matches it; +workflow edits do not change branch-protection settings themselves. diff --git a/README.md b/README.md index 8f97ae48e..f9fbfb1d4 100644 --- a/README.md +++ b/README.md @@ -687,4 +687,4 @@ npm test # Run tests npm run typecheck # Type-check without emitting ``` -Requires Node.js >= 20. Chronicle requires a Rust toolchain for native module compilation (`npm run build` in the chronicle directory). +Requires Node.js >= 22.12. Chronicle requires a Rust toolchain for native module compilation (`npm run build` in the chronicle directory). diff --git a/changelog.d/549-maintenance-withdrawn.fixed.md b/changelog.d/549-maintenance-withdrawn.fixed.md new file mode 100644 index 000000000..d7fa7f1fc --- /dev/null +++ b/changelog.d/549-maintenance-withdrawn.fixed.md @@ -0,0 +1,9 @@ +- A maintenance call agent-framework withdraws, under a surgery's hold or + once `stop()` has begun, no longer counts against a chunk's compression + (#549). `MaintenanceWithdrawnError` now extends context-manager's + `CompressionWithdrawn`, so context-manager rethrows it unchanged from every + rung of its compression fallback ladder and its merge path, recording no + outcome, attempt or quarantine event. Before, a call withdrawn after a + chunk's canonical call was refused was recorded as that rung's provider + error, and once every rung had failed, the chunk's request family was + quarantined, with its alert. diff --git a/changelog.d/floors-membrane-060-cm-017.breaking.md b/changelog.d/floors-membrane-060-cm-017.breaking.md new file mode 100644 index 000000000..a3e26f31b --- /dev/null +++ b/changelog.d/floors-membrane-060-cm-017.breaking.md @@ -0,0 +1,77 @@ +- **Hosts:** agent-framework now depends on membrane `^0.6.0` and + context-manager `^0.18.0`, where it depended on `^0.5.87` and `^0.14.1`. + context-manager 0.18.0 requires membrane `^0.6.0` itself, so the two + install one membrane between them. What reaches a host through + agent-framework, from those releases' notes: + - **Your own membrane:** a host that builds its own `Membrane` and hands it + to agent-framework raises its own `@animalabs/membrane` range to + `^0.6.0`. context-manager counts a compression attempt's prompt through + that instance, and with an older membrane it counts Gemini's cached input + twice (below 0.5.81, OpenAI's too, and OpenRouter's on routes that report + OpenAI's `cached_tokens`), so a compression fallback that fits can be + rejected at admission and its chunk quarantined (context-manager 0.16.0). + - **Your own context-manager:** a host that constructs strategies from its + own `@animalabs/context-manager` must also raise that range to `^0.18.0`, + matching agent-framework's dependency, and verify that installation has + one compatible copy. Leaving the host on 0.14.x installs two copies and + hands an older strategy to a newer manager; strategy interfaces and + compression-pause/quarantine behavior differ between those versions. + - **The kv-unified hysteresis certificate is on by default** for a + strategy built from a recipe: a `kvUnified` block without + `hysteresisCertificate` takes the certified exit, and + `hysteresisCertificate: false` restores the full solver on every compile + (context-manager 0.15.0). + - **Content context-manager won't store** is refused where agent-framework + writes it, as the entry on malformed content says. + - **Stored history:** content an older context-manager stored that doesn't + match `ContentBlock` is now read mended, or left out, with one line per + message naming each place, where some such content made every compile + throw. A `tool_use` stored with `input: null`, as membrane 0.5.87's + OpenAI-family adapters gave it for the arguments `null`, reads as `{}` + with `null` in `unparseableInput`. A request compiled from such a message + can change there, so a provider's prompt cache misses from that point, + once (context-manager 0.16.0). + - **Tools given `null` arguments:** an OpenAI-family tool call whose + arguments are the text `null` reaches the tool as `{}`, where it reached + it as `null`. membrane keeps the arguments' text on the `tool_use` block + as `unparseableInput` (membrane 0.6.0). + - **Anthropic streams:** after a stream's first event, SSE `ping` + keepalives now count as activity for its idle deadline, so a stream that + keeps sending only pings no longer ends at it (membrane 0.6.0). + - **Media notes:** two of the notes membrane writes in place of media it + doesn't send now say the model isn't seeing it. The first request after + upgrading carries the new text wherever it renders such a note, so a + provider's prompt cache misses once, from the first such note on + (membrane 0.6.0). + - **Caller-supplied blob references:** context-manager 0.16.0 and later refuse + `blob_ref` writes with `MalformedContentError`; these references are + store-internal. Read with blobs resolved (the default) before copying a + body into a new write. An unresolved read is not a write-ready payload. + - **Compression quarantine status:** callers that inspect or compare + `getCompressionQuarantineStatus()` must account for its `spans` field, + including an empty array when no spans are quarantined. + - **kv-unified policy:** context-manager 0.17 adds age bands, joint latent + demand and the subtree engine. Explicit recipe overrides remain important: + `policy.ageBandSteps: 0` requests exact ages, `latentDemandMode: 'individual'` + chooses independent ranking, and `engine: 'dag'` selects the DAG engine. + - **Merges across an unowned message:** the floor is 0.18.0, not 0.17.x, + because before it a merge could group the summaries on either side of a + message no chunk owns (a raw pin, a message an unpin freed, or a head + window `resetHeadWindow` left mid-stream), and under kv-unified without + `treeifyNonContiguousSummaries` or `preserveGapBearingSummaries` every + compile after it threw `CanonicalForestError`. A store already holding + such a summary loads as before (context-manager 0.18.0, #287). + - **`close()` is final:** after a context manager's `close()`, every + operation that writes or calls into its strategy throws, with an + app-owned store too; open a new manager on the store to go on with it + (context-manager 0.18.0). + - **Compression debt:** `getCompressionDebt()` answers `unknown` when it + can't read what decides its state, with `null` for each count it couldn't + compute and an operator-facing `error`. A host that shows the report + treats `unknown` as needing attention and never shows a `null` count as + 0 (context-manager 0.18.0). + - **Strategy subclasses:** `AutobiographicalStrategy`'s protected + `pruneToolEntries` is removed; override `toolPruning` or `prunedForSend` + instead (context-manager 0.18.0). + - The rest is in membrane 0.6.0's notes and context-manager 0.15.0's, + 0.16.0's, 0.17.x's and 0.18.0's. diff --git a/changelog.d/malformed-content-refused.fixed.md b/changelog.d/malformed-content-refused.fixed.md new file mode 100644 index 000000000..0bfe283aa --- /dev/null +++ b/changelog.d/malformed-content-refused.fixed.md @@ -0,0 +1,12 @@ +- Content context-manager won't store is refused where it's written, and a deferred write that context-manager refuses no longer holds its agent's later writes. From context-manager 0.16.0, a write whose content doesn't match `ContentBlock` is refused with a `MalformedContentError` (context-manager #184). Before it, such content was stored as given, and a compile that read it could throw. + - **Writes and edits through the framework:** a module's write, and its edit of a write still held for a turn's end, throw `MalformedContentError`, naming each place, as the store's own write and edit do. They throw whether the write would be stored at once or deferred while a turn runs, where a deferred one used to be refused only at a later flush. Content the store mends, a `tool_use` whose input isn't an object, is written as before. The API's `message.send` refuses content that isn't text with an error, where it answered `pending` and stored the message as given. + - **A process response** with such content, in a message it adds or an edit it makes, is refused whole before any response to the event is applied. None of it is applied, its module state update and inference request included, and it doesn't count as accepting console or API input. A `[process-response-refused]` line and a `process:response-refused` trace name the module, the other modules' responses apply, and the `process:completed` trace and the process log list only the responses that applied. + - **The push and channel lanes** check content with context-manager's `malformedContentPlaces` before anything else. Content from an MCPL server is converted at the boundary into blocks the store accepts, so this catches events that never crossed it. When it names any place, the message stores no row and queues no wake. A stderr line, `[push-event-refused]` or `[channel-incoming-refused]`, names the server, the channel where there is one, the event or message, and the places. A new trace, `mcpl:malformed-content-refused`, carries the same, with the places as `places`. Content that's absent is still the empty-content check's. + - **A deferred write that context-manager refuses as malformed is dropped at the drain** and logged, rather than queued again: it would be refused at every flush, and so would hold every later deferred write for its agent for good. Since writes through the framework's `addMessage` are refused before they're queued, this meets an entry a process before this release persisted, a pair `puppetToolCall` queues itself, or a write the strategy's chunking splits into a shard the store refuses, which the line names. Any other failure keeps the entry queued, as before, unless a read-back shows it landed. A `tool_use` dropped this way lets a `tool_result` queued behind it be stored alone. + + The store's own validation checks all content and shards before append, so + an unavailable readback does not make such a refusal retryable. Other errors + whose landing is unknown remain queued and are reconciled by ID before retry. + A synchronous custom strategy hook throwing this same error after append is + tracked separately in context-manager #288; this is not a general guarantee + for arbitrary host callbacks when readback is also unavailable. diff --git a/changelog.d/node-22-12.breaking.md b/changelog.d/node-22-12.breaking.md new file mode 100644 index 000000000..8a1b1e861 --- /dev/null +++ b/changelog.d/node-22-12.breaking.md @@ -0,0 +1,9 @@ +- **Anyone running agent-framework on a Node before 22.12:** agent-framework + now declares `engines.node` `>=22.12.0`, where it declared `>=20.0.0`, as + membrane 0.6.0 and context-manager 0.18.0, which it now depends on, both + do. CI tests at Node 22.12.0 exactly, at 24 and at the newest Node, where it + tested 20 and 24, and the publish workflow builds and tests at 22.12.0. + npm warns when it installs agent-framework on an older Node, and refuses + only with `engine-strict` set. Nothing in agent-framework's code changes + with this; older Node versions are simply no longer tested. + - **Migration:** run Node 22.12 or later. diff --git a/package.json b/package.json index bd0cc1773..f8ff5d17e 100644 --- a/package.json +++ b/package.json @@ -49,8 +49,8 @@ "license": "MIT", "dependencies": { "@animalabs/chronicle": "^0.4.0", - "@animalabs/context-manager": "^0.14.1", - "@animalabs/membrane": "^0.5.87", + "@animalabs/context-manager": "^0.18.0", + "@animalabs/membrane": "^0.6.0", "@modelcontextprotocol/client": "^2.3.1", "chokidar": "^4.0.3", "discord.js": "^14.25.1", @@ -65,6 +65,6 @@ "typescript": "^7.0.2" }, "engines": { - "node": ">=20.0.0" + "node": ">=22.12.0" } } diff --git a/src/api/server.ts b/src/api/server.ts index 7dc2e5388..d95ad44e0 100644 --- a/src/api/server.ts +++ b/src/api/server.ts @@ -489,6 +489,12 @@ export class ApiServer { if (!params.participant || !params.content) { throw new Error('participant and content are required'); } + // ApiModule stores the message as one text block, which context-manager + // refuses unless its text is a string, so content that isn't text is + // refused here, where the caller hears it, rather than at dispatch. + if (typeof params.content !== 'string') { + throw new Error('content must be text'); + } // Push api:message event for the ApiModule to handle this.framework.pushEvent({ diff --git a/src/framework.ts b/src/framework.ts index 9bea1fd49..585b58a6a 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -10,7 +10,7 @@ import { import { JsStore } from '@animalabs/chronicle'; import type { Membrane, ContentBlock, NormalizedRequest, NormalizedResponse, YieldingStream, DetailedUsage, ToolResult as MembraneToolResult, ToolResultContentBlock } from '@animalabs/membrane'; import { MembraneError } from '@animalabs/membrane'; -import { ContextManager, PassthroughStrategy, WindowedPassthroughStrategy, OverBudgetError, UncoveredDropError, stripUnpairedToolBlocks, continuesBody } from '@animalabs/context-manager'; +import { ContextManager, PassthroughStrategy, WindowedPassthroughStrategy, OverBudgetError, UncoveredDropError, stripUnpairedToolBlocks, continuesBody, MalformedContentError, malformedContentPlaces, CompressionWithdrawn } from '@animalabs/context-manager'; import type { CacheWireReceipt } from './kv-unified-wire.js'; import { SILENT_HEARTBEAT_SEPARATOR, type RequestOnlyPrompt, type RequestOnlyStamps } from './silent-heartbeat.js'; import { SUBCONSCIOUS_TOOLS, SUBCONSCIOUS_TOOL_NAMES, type SubconsciousConfig } from './tune-out/tools.js'; @@ -386,6 +386,24 @@ const isConversationalInjection = (metadata?: MessageMetadata): boolean => { return true; }; +/** + * The first message a module's process response adds, or edit it makes, + * whose content context-manager won't store: `where` names it in the + * response, such as `addMessages[1]`, and `places` is malformedContentPlaces' + * answer for it. Null when the response holds none. + */ +function malformedResponseContent(response: EventResponse): { where: string; places: string[] } | null { + const writes = [ + ...(response.addMessages ?? []).map((msg, i) => [`addMessages[${i}]`, msg.content] as const), + ...(response.editMessages ?? []).map((edit, i) => [`editMessages[${i}]`, edit.content] as const), + ]; + for (const [where, content] of writes) { + const places = malformedContentPlaces(content); + if (places.length > 0) return { where, places }; + } + return null; +} + /** * Rough token estimate for the blocks a continuation round is about to * append to the live wire (issue #92 physical-window projection): the @@ -1401,21 +1419,17 @@ function logSeedOmissions( * 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. + * gate, or woken there from its wait. It extends context-manager's + * CompressionWithdrawn, whose constructor sets the brand context-manager reads + * a withdrawal by, so the call is no outcome of the compression: every rung + * of the fallback ladder and the merge path rethrow it unchanged, recording + * no outcome, attempt, streak or quarantine event. Its tick puts the chunk + * back at the head of the queue, unless a rebuild decided the queue afresh + * while the call was out, and a merge stays at the head of its own, for the + * next tick once calls are let through again (context-manager's + * CompressionWithdrawn doc has the rest). */ -class MaintenanceWithdrawnError extends Error { +class MaintenanceWithdrawnError extends CompressionWithdrawn { constructor(reason: string) { super(`maintenance call withdrawn: ${reason}`); this.name = 'MaintenanceWithdrawnError'; @@ -2211,8 +2225,8 @@ function boundFailureFields(record: Record): Record boolean } | undefined; @@ -6709,14 +6722,16 @@ export class AgentFramework { * acknowledged once: the chronicle is synced, then the persisted queue is * rewritten without the batch (ackDeferredWrites). Until then the batch * stays in the durable queue, so a crash mid-flush replays it at the next - * boot, which skips each entry that already landed, by its deferredWriteId - * — no loss, no duplicates. The batch is stored in queue order, whatever - * its targets (writeDrainedInOrder). A write the store rejects stays - * queued, logged, with what is queued behind it for its target, so the - * flush never stores that target's later entries ahead of it, and other - * targets' entries go on. (A write that isn't deferred can still land - * ahead of it, such as one for an idle agent once the host has resumed.) - * A boot that meets it again logs it and goes on. + * boot, which skips each entry that already landed, by its deferredWriteId — + * no loss, no duplicates. The batch is stored in queue order, whatever its + * targets (writeDrainedInOrder). A write the store rejects stays queued, + * logged, with what is queued behind it for its target, so the flush never + * stores that target's later entries ahead of it, and other targets' entries + * go on. One the store refuses as malformed, which it would at every flush, + * is logged and dropped instead, and what is behind it goes on. (A write + * that isn't deferred can still land ahead of it, such as one for an idle + * agent once the host has resumed.) A boot that meets it again logs it and + * goes on. * * Per-target: an agent whose turn is still alive keeps ITS messages * deferred for its own turn boundary; everyone else's flush now. @@ -7232,19 +7247,21 @@ export class AgentFramework { } } // An entry the queue's readers can't handle has no place in it: one whose - // participant isn't text, whose content isn't an array, that holds a - // block that isn't an object, or whose `forAgent` is present but not a - // name, is dropped here with a line naming it, rather than held. Reading - // a null block's type throws, at the tool boundary and in the store. The - // store keeps a string or number block as given, but the context - // manager's token estimate keys a WeakMap on each block, which such a - // value can't be: once one is stored, every compile for that agent - // throws, and a restart doesn't clear it. What a block holds, such as a - // tool result's content, isn't checked here, as it isn't for a fresh - // write. The framework writes `forAgent` only as a non-empty name, so - // another value comes from damage or a hand edit, and the agent it - // meant can't be known: sent to the primary, as a falsy one would be, it - // could put another agent's message in the primary's window. + // participant isn't text, whose content isn't an array, that holds a block + // that isn't an object, or whose `forAgent` is present but not a name, is + // dropped here with a line naming it, rather than held. Reading a null + // block's type throws, at the tool boundary and in the store. From + // context-manager 0.16.0 the store refuses a string or number block and + // its reads leave one out; an older store kept it as given, and since the + // context manager's token estimate keys a WeakMap on each block, every + // compile for that agent then threw, and a restart didn't clear it. What a + // block holds, such as a tool result's content, isn't checked here: a + // recovered write whose content context-manager refuses is dropped at the + // flush that meets it (writeDrainedInOrder). The framework writes + // `forAgent` only as a non-empty name, so another value comes from damage + // or a hand edit, and the agent it meant can't be known: sent to the + // primary, as a falsy one would be, it could put another agent's message + // in the primary's window. const wellFormed = (m: unknown): m is Omit & { id?: unknown; seq?: unknown; heldAt?: unknown } => { if (!m || typeof m !== 'object') return false; const { participant, content, forAgent } = m as { participant?: unknown; content?: unknown; forAgent?: unknown }; @@ -11724,11 +11741,12 @@ export class AgentFramework { const deferred = this.drainDeferredFor(agent.name, (msg) => !isInjectableMidTurn(msg, agent.name)); // Stored in queue order, cut at the first that can't be stored: it // and what follows wait for the turn's end, and only what was - // stored is injected (the round's results go on either way). A - // sharded body read back from its store is injected as later - // compiles render it, its shards joined, and a write that threw - // after storing, which counts as stored, as what landed of it - // (writeDrainedInOrder). + // stored is injected (the round's results go on either way). One + // the store refuses as malformed is dropped instead, and what + // follows goes on. A sharded body read back from its store is + // injected as later compiles render it, its shards joined, and a + // write that threw after storing, which counts as stored, as what + // landed of it (writeDrainedInOrder). const stored = this.writeDrainedInOrder(deferred, `tool boundary for ${agent.name}`, (msg) => agent.getContextManager().addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id))); for (const [msg, { id, content }] of stored) { @@ -12062,14 +12080,37 @@ export class AgentFramework { } } + // A response holding content context-manager won't store, in a message + // it adds or an edit it makes, is refused whole before any response is + // applied: none of it is applied, its module state update and inference + // request included, so no turn runs for content that was never stored, + // and it doesn't count as accepting a surface's input. A line and a + // trace name the module and where, and the other modules' responses + // apply. process:completed and the process log list the responses that + // applied, so a view built on them doesn't show a refused response's + // messages as the event's. A store error while a response applies, such + // as a shard the store refuses when a strategy's chunking makes one, + // propagates as any store error does. + const applying = responses.filter(({ moduleName, response }) => { + const refusal = malformedResponseContent(response); + if (refusal === null) return true; + const { where, places } = refusal; + console.error( + `[process-response-refused] module=${moduleName} event=${event.type} reason=malformed-content: ` + + logDetail(new MalformedContentError(`${where} refused`, places).message), + ); + this.emitTrace({ type: 'process:response-refused', moduleName, eventType: event.type, where, places }); + return false; + }); + // Apply responses. Console/API input is conversation from a non-channel // surface: one acceptance per event, whose envelope every message the // modules add for it carries. const surface = this.surfaceSourceFor(event); - if (surface && responses.some(({ response }) => (response.addMessages?.length ?? 0) > 0)) { + if (surface && applying.some(({ response }) => (response.addMessages?.length ?? 0) > 0)) { this.noteInboundAccepted(surface); } - for (const { moduleName, response } of responses) { + for (const { moduleName, response } of applying) { await this.applyProcessResponse(response, event, moduleName, surface); } @@ -12081,11 +12122,11 @@ export class AgentFramework { const durationMs = Date.now() - startTime; // Always emit trace for observability (UI needs this) - this.emitTrace({ type: 'process:completed', processEvent: event, responses, durationMs }); + this.emitTrace({ type: 'process:completed', processEvent: event, responses: applying, durationMs }); // Log to Chronicle (if enabled) if (this.processLoggingPersist) { - this.logProcessEvent(event, responses); + this.logProcessEvent(event, applying); } } @@ -12223,6 +12264,23 @@ export class AgentFramework { acceptedAt?: number; inboundSource?: InboundSource; }): Promise { + // context-manager won't store content that doesn't match ContentBlock + // (malformedContentPlaces names where). Content from the wire can't be + // such content, since the MCPL boundary converts every block into one the + // store accepts, so this is the backstop for events that never crossed + // it, as the empty-content check below is. Refused here, such a message + // stores no row, spawns no fork and queues no wake, and the line names + // the server and channel it came from; past this point its write would + // throw. Absent content is the empty-content check's. + const malformed = event.content == null ? [] : malformedContentPlaces(event.content); + if (malformed.length > 0) { + console.error(`[channel-incoming-refused] server=${event.serverId} channel=${logValue(event.channelId)} messageId=${logValue(event.messageId)} reason=malformed-content: ${logDetail(malformed.join('; '))}`); + this.emitTrace({ + type: 'mcpl:malformed-content-refused', lane: 'channel', serverId: event.serverId, places: malformed, + channelId: event.channelId, messageId: event.messageId, ...(event.eventId ? { eventId: event.eventId } : {}), + }); + return undefined; + } // Same backstop as the push lane: no row, no wake, no fork spawned for a // message with nothing visible in it. if (isVisiblyEmptyContent(event.content)) { @@ -13775,6 +13833,19 @@ export class AgentFramework { * Convert an MCPL push event to a context message. */ private handleMcplPushEvent(event: McplPushEvent): CoalescingPlacement | undefined { + // Content context-manager won't store is refused first, as on the + // channel lane (handleMcplChannelIncoming): no row, no wake. Absent + // content is the empty-content backstop's. + const malformed = event.content == null ? [] : malformedContentPlaces(event.content); + if (malformed.length > 0) { + const channelId = this.derivePushEventChannel(event.origin)?.channelId; + console.error(`[push-event-refused] server=${event.serverId}${channelId ? ` channel=${logValue(channelId)}` : ''} eventId=${logValue(event.eventId ?? '')} reason=malformed-content: ${logDetail(malformed.join('; '))}`); + this.emitTrace({ + type: 'mcpl:malformed-content-refused', lane: 'push', serverId: event.serverId, places: malformed, + eventId: event.eventId, ...(channelId ? { channelId } : {}), + }); + return undefined; + } // Backstop for the MCPL boundary's empty-content rejection, covering // events that never crossed it (module-emitted, coalescer deliveries): // content with nothing visible stores no row and queues no wake — the @@ -16647,6 +16718,7 @@ export class AgentFramework { // In queue order and cut at the first that can't be stored, as every // flush is (readdDrained), so a store failure neither lands the // messages behind it out of order nor replaces the start's own error. + // One refused as malformed is dropped, not cut at. // Gated on this agent's own tool cycle, as the turn-end flush is. if (!tokenHandedOff && !this.activeTurnTokens.has(agent.name) && this.deferredMessages.length > 0 && !this.pendingAssistantBlocks.has(agent.name)) { @@ -16790,7 +16862,9 @@ export class AgentFramework { // One poison message, or a transient store-write failure, must not // abort the turn or drop the messages behind it in the queue: the // drain cuts there, and it and what follows stay queued, in order - // (writeDrainedInOrder). + // (writeDrainedInOrder). A message the store refuses as malformed, + // which it would at every flush, is dropped instead, and what follows + // goes on. this.flushForStart(agent, 'turn-start'); } @@ -19649,6 +19723,9 @@ export class AgentFramework { // still be stored — not under that turn: queue it as a unit for the // next flush (drainDeferredFor keeps order; the tool_result entry is // never split from its tool_use because both ride the same drain). + // One exception: a tool_use context-manager refuses as malformed is + // dropped at the drain, and the tool_result behind it is then stored + // alone (writeDrainedInOrder). // The finally below flushes it itself if that turn is already over. ownedToEnd = this.activeTurnTokens.get(agentName) === turnToken; if (ownedToEnd) { @@ -21426,6 +21503,12 @@ export class AgentFramework { const { deferredWriteId: _theFrameworks, ...rest } = metadata as MessageMetadata & { deferredWriteId?: unknown }; metadata = rest as MessageMetadata; } + // Content context-manager won't store is refused here, as the store's own + // write refuses it, whether this write is stored now or deferred: the + // writer learns at once, not from a line at a drain it never sees. Content + // the store mends, such as a tool_use whose input isn't an object, passes. + const malformed = malformedContentPlaces(content); + if (malformed.length > 0) throw new MalformedContentError('Message refused', malformed); const agent = this.deliveryAgent(opts?.forAgent); if (!agent) { if (opts?.forAgent) { @@ -21614,12 +21697,34 @@ export class AgentFramework { * * So none of a target's entries lands ahead of an earlier one, and a queued * tool_use that can't be stored keeps the tool_result queued behind it from - * being stored alone. (A tool_use that is stored, followed by a tool_result - * that can't be, stays stored without it until the tool_result lands at a - * later flush. The only queued pair is puppetToolCall's defensive one, - * which no path reaches today.) An entry that keeps failing is logged by - * id at each flush, and holds its target's later queued writes rather than - * letting them land out of order. + * being stored alone, unless context-manager refused it as malformed + * (below). (A tool_use that is stored, followed by a tool_result that + * can't be, stays stored without it until the tool_result lands at a later + * flush, or for good when the store refuses the tool_result as malformed. + * The only queued pair is puppetToolCall's, queued only when its turn's + * token is no longer its own once the pair is built, which its comment + * calls defensive.) An entry that keeps failing is logged by id at each + * flush, and holds its target's later queued writes rather than letting + * them land out of order. + * + * An entry context-manager refuses as malformed (MalformedContentError) is + * the exception for the store's own validation: context-manager checks + * content and every shard before writing. A custom synchronous strategy + * hook can throw the same class after append (context-manager #288); + * unreadable landing in that callback escape remains a limitation until + * the CM follow-up lands. A copy actually found still counts first. Other + * ambiguous errors remain queued for ID reconciliation before retry. A + * malformed refusal would fail every flush and hold its target's later + * writes for good, so it's logged and dropped instead, and its + * target isn't cut: left in the un-acked set, the next ack drops it from + * the durable queue. A tool_use dropped this way leaves a tool_result + * queued behind it to be stored alone. context-manager mends a tool_use + * whose input isn't an object, so it refuses one only when that input has + * no JSON form or the block itself is malformed. addMessage refuses such + * content before it's queued, so an entry meets this here from a queue an + * earlier process persisted, as puppetToolCall's pair, which it queues + * itself, or when the strategy's chunking splits well-formed content into a + * shard the store refuses, which the store's error names. * * From a fresh turn's start to the end of its coalescing assembly, before * its checkpoint, every write deferred for its agent lands in that @@ -21648,8 +21753,8 @@ export class AgentFramework { * written, at most once: the entry isn't requeued and stored a second * time, and what landed, as later compiles render it (asCompiled), is what * it returns as the entry's content. For a sharded body, that can be its - * first shards alone, since the append loop stops at the throw; the rest - * of the body isn't stored. (Requeued, the + * first shards alone if rollback could not remove them; context-manager + * from 0.15 otherwise removes shards from a failed append. (Requeued, the * landed shards would be stored twice, and only the rest could not be * appended later without splitting the body.) Boot's scan for landed * writes counts a landed shard the same way. When the store can't be read @@ -21661,14 +21766,15 @@ export class AgentFramework { * the store can't be searched, it stays queued, holding what's behind it. * * Returns what was written (stored, or deferred again by addMessage, whose - * id is then empty), in queue order: for one target, a prefix of its - * entries. Each comes with its content: as given, unless the store took it - * as more than one message (a sharded body) and can be read back, when its - * copies are returned joined, as later compiles render them (asCompiled), - * or its write threw after storing it (above). `onLanded` hears an entry - * whose write threw after storing it, with the id it landed under, as soon - * as it's found: before the writes after it, so whatever it records keeps - * their order. + * id is then empty), in queue order: for one target, its entries up to the + * first that couldn't be stored, less entries refused as malformed and dropped. Each comes + * with its content: as given, unless the store took it as more than one + * message (a sharded body) and can be read back, when its copies are + * returned joined, as later compiles render them (asCompiled), or its write + * threw after storing it (above). `onLanded` hears an entry whose write + * threw after storing it, with the id it landed under, as soon as it's + * found: before the writes after it, so whatever it records keeps their + * order. */ private writeDrainedInOrder( entries: DeferredWrite[], @@ -21731,6 +21837,14 @@ export class AgentFramework { onLanded?.(msg, landed[0]!.id); continue; } + // The store's own validation refuses content/shards before append. + // A synchronous custom strategy can escape after append (CM #288); + // this is not a guarantee for that callback when readback also fails. + // Any copy readback did find was already handled above. + if (error instanceof MalformedContentError) { + this.logDeferredWriteError(msg, where, error, 'dropped'); + continue; + } // Requeued as unstored, a write that had landed would be stored a // second time; so one whose landing is unknown is marked, and looked // up by its id before it's written again (above). @@ -21784,7 +21898,9 @@ export class AgentFramework { } /** The line naming a drained write that threw: one that couldn't be stored, - * which is already back in the queue; with a count, one stored before the + * which is already back in the queue; with 'dropped', a malformed refusal + * from the store's pre-write validation (see the CM #288 caveat above); + * with a count, one stored before the * throw (for a sharded body, how many of its shards), or, with * `foundLater`, one an earlier write of which landed and was found by its * id; with `'unknown'`, one whose store couldn't be read to tell @@ -21796,7 +21912,7 @@ export class AgentFramework { msg: DeferredWrite, where: string, error: unknown, - outcome?: { count: number; sharded: boolean; foundLater?: boolean } | 'unknown' | 'still-unknown', + outcome?: { count: number; sharded: boolean; foundLater?: boolean } | 'unknown' | 'still-unknown' | 'dropped', ): void { let reason: string; try { @@ -21809,7 +21925,9 @@ export class AgentFramework { ? `${landed.count} shard${landed.count === 1 ? '' : 's'} of its body ${landed.count === 1 ? 'was' : 'were'} stored` : 'it was stored'; let what: string; - if (outcome === 'unknown') { + if (outcome === 'dropped') { + what = `was refused as malformed, so it's dropped rather than queued again, and what is queued behind it for that agent goes on: ${reason}`; + } else if (outcome === 'unknown') { what = `threw, and its store couldn't be read to tell whether it was stored; ${held}, and is looked up ` + `by its id before it's written again: ${reason}`; } else if (outcome === 'still-unknown') { @@ -21830,14 +21948,15 @@ export class AgentFramework { /** * Write what is queued for `agent` into its window past its deferral, in - * queue order and cut at a write that fails (writeDrainedInOrder), between - * a fresh turn's start and its checkpoint, where what is stored is the - * turn's input: at the start, before each write the coalescing assembly - * makes for the agent, and once the assembly ends, the last two not while - * the host is quiesced, when what is queued waits for resume(). What is - * stored is weighed for the start now beginning and for every other - * resident's open turn. Returns whether anything is still queued for the - * agent. + * queue order and cut at a write that fails, though one the store refuses + * as malformed is dropped and what follows goes on (writeDrainedInOrder), + * between a fresh turn's start and its checkpoint, where what is stored is + * the turn's input: at the start, before each write the coalescing + * assembly makes for the agent, and once the assembly ends, the last two + * not while the host is quiesced, when what is queued waits for resume(). + * What is stored is weighed for the start now beginning and for every + * other resident's open turn. Returns whether anything is still queued for + * the agent. */ private flushForStart(agent: Agent, occasion: 'turn-start' | 'assembly'): boolean { const stored = this.writeDrainedInOrder(this.drainDeferredFor(agent.name), `${occasion} flush for ${agent.name}`, (msg) => @@ -22468,6 +22587,10 @@ export class AgentFramework { if (!agent) { throw new Error('No agents configured'); } + // As the store's own edit does, and before a held write's content is + // replaced: a malformed edit leaves the write as it was. + const malformed = malformedContentPlaces(content); + if (malformed.length > 0) throw new MalformedContentError(`Edit of ${id} refused`, malformed); // The held writes are settled first (heldWritesFor), and by a stored id // the row it names too (settleByCopy): a held write that landed is a // stored row from then on, and the edit reaches it as one. diff --git a/src/held-writes.ts b/src/held-writes.ts index 6972af77d..f51d77032 100644 --- a/src/held-writes.ts +++ b/src/held-writes.ts @@ -25,9 +25,9 @@ export function heldWriteId(id: MessageId): string | undefined { /** * Whether a row matches `filter` as context-manager's queryMessages matches a - * stored one. That's its MessageStore's private matchesFilter at 0.14.1, which - * the package doesn't export, so the rule is copied here, and a test compares - * the two on the same rows. A row matches the external source, one of the + * stored one. That's its MessageStore's private matchesFilter, the same at + * 0.14.1, 0.16.0, 0.17.1 and 0.18.0, which the package doesn't export, so the rule is copied + * here, and a test compares the two on the same rows. A row matches the external source, one of the * external ids (an empty list filters nothing), the participant, and each * metadata path, dot-separated, by strict equality. */ diff --git a/src/mcpl/inbound-source.ts b/src/mcpl/inbound-source.ts index e2bc75645..8db98f9db 100644 --- a/src/mcpl/inbound-source.ts +++ b/src/mcpl/inbound-source.ts @@ -198,9 +198,10 @@ export function canonicalJson(value: unknown): string { * 'base64', data, mediaType } }` and nothing else, its data re-encoded from * the decoded bytes, and an image's media type taken from its bytes' * signature where they have one. Any other block comes back as it was given, - * except a `blob_ref`, which the store resolves to its inline media on read: - * it is digested as given here, so a copy holding one doesn't match its - * digest (the MCPL lanes never store one). + * except a store-internal `blob_ref` from an unresolved read, which resolves + * to inline media by default. It is digested as given here, so an unresolved + * read and its resolved form differ. Context-manager refuses caller-supplied + * blob_ref writes; the MCPL lanes never store one. */ function storedBlock(block: unknown): unknown { const media = block as { type?: unknown; source?: { type?: unknown; data?: unknown; mediaType?: unknown } } | null; diff --git a/src/types/framework.ts b/src/types/framework.ts index 2876bf26b..815daff23 100644 --- a/src/types/framework.ts +++ b/src/types/framework.ts @@ -461,13 +461,16 @@ export interface InferenceLogEntry { } /** - * Entry in the process log - records a processed event with all module responses. + * Entry in the process log - records a processed event with the module + * responses that applied. */ export interface ProcessLogEntry { timestamp: number; /** The process event that was handled */ processEvent: ProcessEvent; - /** Responses from all modules, or blob ID if large */ + /** The modules' responses that applied, or blob ID if large. A response + * refused for content context-manager won't store is left out; its stderr + * line and process:response-refused trace name it. */ responses: ModuleProcessResponse[] | { blobId: string }; } @@ -477,7 +480,7 @@ export interface ProcessLogEntry { export interface ProcessLogQuery { /** Filter by process event type */ eventType?: string; - /** Filter by module name (modules that responded) */ + /** Filter by module name (modules whose responses applied) */ moduleName?: string; /** Max number of logs to return */ limit?: number; diff --git a/src/types/trace.ts b/src/types/trace.ts index c1c68f9b1..8b46876fb 100644 --- a/src/types/trace.ts +++ b/src/types/trace.ts @@ -33,9 +33,23 @@ export type TraceEvent = | (TraceEventBase & { type: 'process:completed'; processEvent: ProcessEvent; + /** The modules' responses that applied: one refused for malformed + * content is left out, and named by process:response-refused. */ responses: ModuleProcessResponse[]; durationMs: number; }) + /** A module's process response held content context-manager won't store, + * in a message it adds or an edit it makes, and was refused whole before + * any response to the event applied. `where` names the first such message + * or edit in the response, such as `addMessages[1]`, and `places` is + * malformedContentPlaces' answer for it. */ + | (TraceEventBase & { + type: 'process:response-refused'; + moduleName: string; + eventType: string; + where: string; + places: string[]; + }) // Inference lifecycle | (TraceEventBase & { @@ -375,6 +389,18 @@ export type TraceEvent = field?: string; detail?: string; }) + /** MCPL content that context-manager won't store, because it doesn't + * match ContentBlock, reached the store/wake site and was refused there: + * no row, no wake. `places` is malformedContentPlaces' answer for it. */ + | (TraceEventBase & { + type: 'mcpl:malformed-content-refused'; + lane: 'push' | 'channel'; + serverId: string; + places: string[]; + eventId?: string; + channelId?: string; + messageId?: string; + }) /** MCPL content with nothing model-visible reached the store/wake site * (module-emitted or coalescer-delivered) and was dropped: no row, no wake. */ | (TraceEventBase & { diff --git a/test/deferred-malformed-unknown-landing.test.ts b/test/deferred-malformed-unknown-landing.test.ts new file mode 100644 index 000000000..af3e70884 --- /dev/null +++ b/test/deferred-malformed-unknown-landing.test.ts @@ -0,0 +1,154 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { ContextManager, PassthroughStrategy, MalformedContentError } from '@animalabs/context-manager'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import type { ContentBlock } from '@animalabs/membrane'; +import { AgentFramework } from '../src/framework.js'; + +// Focused merge regression: the unit harness opens no store/provider/network; the final characterization uses a temporary store. +function harness() { + const f = Object.create(AgentFramework.prototype) as any; + const agent = {}; + let readback: any[] | undefined = []; + let lookup: any[] | undefined = undefined; + let count: number | undefined = 0; + const queued: any[] = []; + const outcomes: any[] = []; + f.primaryAgentName = 'a'; + f.deliveryAgent = () => agent; + f.storedCountOf = () => count; + f.storedCopiesOf = () => readback; + f.landedCopiesById = () => lookup; + f.requeueDrained = (entries: any[]) => queued.push(...entries); + f.logDeferredWriteError = (_m: any, _w: any, _e: any, outcome: any) => outcomes.push(outcome); + const msg = (id: string, seq: number, forAgent = 'a') => ({ + id, seq, forAgent, participant: 'user', content: [{ type: 'text', text: id }], + }); + return { f, queued, outcomes, msg, + setReadback: (v: any[] | undefined) => { readback = v; }, + setLookup: (v: any[] | undefined) => { lookup = v; }, + setCount: (v: number | undefined) => { count = v; }, + }; +} +const malformed = () => new MalformedContentError('fixture', ['content[0]']); + +test('malformed refusal with confirmed empty landing drops and unblocks its successor', () => { + const h = harness(); const a = h.msg('a', 1), b = h.msg('b', 2); const calls: string[] = []; + let landed = 0; + const result = h.f.writeDrainedInOrder([a, b], 'test', (m: any) => { + calls.push(m.id); if (m === a) throw malformed(); return 'b-stored'; + }, () => landed++); + assert.deepEqual(calls, ['a', 'b']); assert.equal(result.has(a), false); + assert.equal(result.get(b).id, 'b-stored'); assert.deepEqual(h.queued, []); + assert.equal(landed, 0); assert.deepEqual(h.outcomes, ['dropped']); +}); + +for (const unknown of ['count', 'readback']) { + test(`malformed refusal drops despite unavailable ${unknown}, unblocking its successor`, () => { + const h = harness(); if (unknown === 'count') h.setCount(undefined); else h.setReadback(undefined); + const a = h.msg('a', 1), b = h.msg('b', 2), c = h.msg('c', 3, 'other'); + const calls: string[] = []; + const written = h.f.writeDrainedInOrder([c, b, a], 'test', (m: any) => { + calls.push(m.id); if (m === a) throw malformed(); return m.id + '-stored'; + }); + assert.deepEqual(calls, ['a', 'b', 'c']); assert.deepEqual(h.queued, []); + assert.equal(written.has(a), false); assert.equal((a as any).landing, undefined); + assert.deepEqual(h.outcomes, ['dropped']); + }); + test(`ordinary failure with unknown ${unknown} retains ordering and reconciliation`, () => { + const h = harness(); if (unknown === 'count') h.setCount(undefined); else h.setReadback(undefined); + const a = h.msg('a', 1), b = h.msg('b', 2), c = h.msg('c', 3, 'other'); + const calls: string[] = []; + h.f.writeDrainedInOrder([c, b, a], 'test', (m: any) => { + calls.push(m.id); if (m === a) throw new Error('uncertain I/O failure'); return 'c-stored'; + }); + assert.deepEqual(calls, ['a', 'c']); assert.deepEqual(h.queued, [a, b]); + assert.equal((a as any).landing, 'unknown'); assert.deepEqual(h.outcomes, ['unknown']); + h.queued.length = 0; + h.f.writeDrainedInOrder([a, b], 'retry', () => { throw new Error('must not rewrite before lookup'); }); + assert.deepEqual(h.queued, [a, b]); assert.equal(h.outcomes.at(-1), 'still-unknown'); + }); +} + +test('unknown landing found by ID counts as written without rewriting, then unblocks successor', () => { + const h = harness(); const a = Object.assign(h.msg('a', 1), { landing: 'unknown' }); + const b = h.msg('b', 2); const stored = { id: 'stored-a', participant: 'user', content: a.content }; + h.setLookup([stored]); const calls: string[] = [], callbacks: string[] = []; + const result = h.f.writeDrainedInOrder([a, b], 'retry', (m: any) => { + calls.push(m.id); return 'stored-b'; + }, (_m: any, id: string) => callbacks.push(id)); + assert.deepEqual(calls, ['b']); assert.deepEqual(callbacks, ['stored-a']); + assert.equal(result.get(a).id, 'stored-a'); assert.deepEqual(result.get(a).content, a.content); + assert.deepEqual(h.queued, []); assert.equal(h.outcomes[0].foundLater, true); +}); + +for (const outcome of ['success', 'malformed-empty', 'malformed-unknown']) { + test(`unknown landing confirmed absent retries once: ${outcome}`, () => { + const h = harness(); const a = Object.assign(h.msg('a', 1), { landing: 'unknown' }); + h.setLookup([]); if (outcome === 'malformed-unknown') h.setReadback(undefined); + let calls = 0; + const result = h.f.writeDrainedInOrder([a], 'retry', () => { + calls++; if (outcome !== 'success') throw malformed(); return 'stored-a'; + }); + assert.equal(calls, 1); + if (outcome === 'success') { assert.equal(result.get(a).id, 'stored-a'); assert.deepEqual(h.queued, []); } + if (outcome === 'malformed-empty') { assert.deepEqual(h.outcomes, ['dropped']); assert.deepEqual(h.queued, []); } + if (outcome === 'malformed-unknown') { assert.equal(a.landing, undefined); assert.deepEqual(h.queued, []); assert.deepEqual(h.outcomes, ['dropped']); } + }); +} + +test('ordinary error with empty landing remains queued with its successor', () => { + const h = harness(); const a = h.msg('a', 1), b = h.msg('b', 2); + let calls = 0; + h.f.writeDrainedInOrder([a, b], 'test', () => { calls++; throw new Error('I/O failure'); }); + assert.equal(calls, 1); assert.deepEqual(h.queued, [a, b]); assert.deepEqual(h.outcomes, [undefined]); +}); + +test('confirmed readback copy takes precedence over a malformed refusal', () => { + const h = harness(); const a = h.msg('a', 1); + h.setReadback([{ id: 'stored-a', participant: 'user', content: a.content }]); + let callbacks = 0; + const result = h.f.writeDrainedInOrder([a], 'test', () => { throw malformed(); }, () => callbacks++); + assert.equal(result.get(a).id, 'stored-a'); assert.equal(callbacks, 1); assert.deepEqual(h.queued, []); + assert.equal(h.outcomes[0].count, 1); +}); + + +test('CM typed malformed refusal is pre-write for whole bodies and all ingress shards', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-malformed-prewrite-')); + let shardMalformed = false; + const strategy = Object.assign(new PassthroughStrategy(), { + chunkIngressMessage: () => shardMalformed ? { + bodyGroupId: 'fixture-group', + shards: [ + { shardIndex: 0, content: [{ type: 'text' as const, text: 'valid first shard' }] }, + { shardIndex: 1, content: [{ type: 'text', text: 42 }] as unknown as ContentBlock[] }, + ], + } : null, + }); + const cm = await ContextManager.open({ path: join(dir, 'store'), strategy }); + try { + const initial = cm.addMessage('user', [{ type: 'text', text: 'keep me' }]); + const count = cm.getMessageCount(); + const bodies = [ + [{ type: 'text', text: 42 }], + [{ type: 'text' }], + [{ type: 'blob_ref', hash: 'not-caller-owned' }], + [{ type: 'text', text: 'valid before malformed' }, { type: 'text', text: false }], + [{ type: 'image', source: 'not-an-object' }], + ]; + for (let i = 0; i < bodies.length; i++) { + const id = `refused-${i}`; + assert.throws(() => cm.addMessage('user', bodies[i] as never, { deferredWriteId: id } as never), MalformedContentError); + assert.equal(cm.getMessageCount(), count); + assert.equal(cm.queryMessages({ metadata: { deferredWriteId: id } } as never).messages.length, 0); + } + shardMalformed = true; + assert.throws(() => cm.addMessage('user', [{ type: 'text', text: 'split this valid input' }], { deferredWriteId: 'bad-shard' } as never), MalformedContentError); + assert.equal(cm.getMessageCount(), count, 'even the valid first shard was never appended'); + assert.equal(cm.queryMessages({ metadata: { deferredWriteId: 'bad-shard' } } as never).messages.length, 0); + assert.deepEqual(cm.getMessage(initial)?.content, [{ type: 'text', text: 'keep me' }]); + } finally { cm.close(); rmSync(dir, { recursive: true, force: true }); } +}); diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index 415d56fed..1e4f5925e 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -31,9 +31,8 @@ import { describe, it, beforeEach, afterEach } from 'node:test'; import assert from 'node:assert/strict'; import { tmpdir } from 'node:os'; -import { existsSync, mkdtempSync, readFileSync, rmSync } from 'node:fs'; -import { dirname, join } from 'node:path'; -import { fileURLToPath } from 'node:url'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; import { isDeepStrictEqual } from 'node:util'; import type { Module, @@ -47,7 +46,7 @@ import type { } from '../src/index.js'; import { AgentFramework } from '../src/index.js'; import { heldHandle } from '../src/held-writes.js'; -import { AutobiographicalStrategy } from '@animalabs/context-manager'; +import { AutobiographicalStrategy, MalformedContentError } from '@animalabs/context-manager'; import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; import type { ContentBlock } from '@animalabs/membrane'; @@ -56,33 +55,6 @@ interface Interjection { content: ContentBlock[]; } -/** The installed context-manager's version. Its package's exports don't - * include package.json, so walk up from its entry point to it. */ -function contextManagerVersion(): string { - let dir = dirname(fileURLToPath(import.meta.resolve('@animalabs/context-manager'))); - for (;;) { - const file = join(dir, 'package.json'); - if (existsSync(file)) { - const pkg = JSON.parse(readFileSync(file, 'utf8')) as { name?: string; version?: string }; - if (pkg.name === '@animalabs/context-manager' && pkg.version) return pkg.version; - } - const up = dirname(dir); - if (up === dir) throw new Error("can't find @animalabs/context-manager's package.json"); - dir = up; - } -} - -/** Set, with the reason, when shards can't land before a throw: from - * context-manager 0.15, a sharded write that fails partway removes what it - * wrote. The tests of carrying landed shards pin 0.14.x. */ -const NO_SHARDS_LAND_BEFORE_A_THROW = (() => { - const version = contextManagerVersion(); - const [major, minor] = version.split('.').map(Number); - return major! > 0 || minor! >= 15 - ? `context-manager ${version} removes what a sharded write that fails partway wrote, so no shards land before a throw` - : false; -})(); - /** Tools that deliver messages while they run, the way a module or a channel * would while the agent is busy: `move` continues the turn, `halt` ends it. */ class InterjectingModule implements Module { @@ -541,11 +513,11 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { /** The strategy's onNewMessage throws once, synchronously, for the record * with this text. The manager's own store listener calls it as each record - * is written, so on context-manager 0.14 the throw escapes the append that - * stored the record, partway through a sharded body. Only one test uses - * it, "a sharded body that throws after its first shard", which pins 0.14 - * (NO_SHARDS_LAND_BEFORE_A_THROW) and relies on that escape, which - * context-manager issue #288 reports as a bug. */ + * is written, so the throw escapes the append partway through a sharded + * body, and context-manager removes the shards it wrote. Only one test + * uses it, "a sharded body whose write throws partway lands nothing", + * which relies on that escape, which context-manager issue #288 reports + * as a bug. */ function throwOnceFromStrategyHook(framework: AgentFramework, textOf: string): void { const strategy = (framework.getAgent('assistant')!.getContextManager() as unknown as { strategy: { onNewMessage?: (message: { content: ContentBlock[] }, ctx: unknown) => Promise }; @@ -795,6 +767,69 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { await framework.stop(); }); + for (const failAt of ['initial-count', 'readback'] as const) { + it(`real malformed refusal with unavailable ${failAt} drops, unblocks and stays dropped after reopen`, async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const queue = internalsOf(framework); + queue.deferredWritesPersisted = true; + const cm = framework.getAgent('assistant')!.getContextManager(); + const first = { ...entry('w-malformed-real', 'bad'), content: [{ type: 'text', text: 42 }] as unknown as ContentBlock[] }; + const second = { ...entry('w-after-malformed', 'after'), seq: 2 }; + const count = cm.getMessageCount.bind(cm); + let failCount = false; + let countPhase = 'drain'; + const failedPhases: string[] = []; + cm.getMessageCount = () => { + if (failCount) { failCount = false; failedPhases.push(countPhase); throw new Error('injected count/readback failure'); } + return count(); + }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + let reopened: AgentFramework | undefined; + const writes: string[] = []; + try { + queue.deferredMessages.push(first, second); + const drained = queue.drainDeferredFor('assistant') as DeferredEntry[]; + assert.deepEqual(failedPhases, [], 'handoff must not consume the writer fault'); + countPhase = 'writer-before-add'; + failCount = failAt === 'initial-count'; + internals.writeDrainedInOrder(drained, 'test-real-store', (m) => { + countPhase = 'write-callback'; + writes.push(m.id); + try { return cm.addMessage(m.participant, m.content, { deferredWriteId: m.id } as never); } + catch (e) { + assert.ok(e instanceof MalformedContentError, 'real CM refusal, not a framework readback stub'); + countPhase = 'writer-readback'; + if (failAt === 'readback') failCount = true; + throw e; + } + }); + assert.deepEqual(failedPhases, [failAt === 'initial-count' ? 'writer-before-add' : 'writer-readback'], + 'the intended writer count boundary, not queue handoff or the write callback, failed exactly once'); + assert.deepEqual(writes, [first.id, second.id], 'successor proceeds in the same flush'); + assert.deepEqual(internals.deferredMessages, [], 'malformed refusal is not retained as unknown'); + assert.equal(cm.queryMessages({ metadata: { deferredWriteId: first.id } } as never).messages.length, 0); + assert.equal(cm.queryMessages({ metadata: { deferredWriteId: second.id } } as never).messages.length, 1); + assert.ok(errors.some(e => e.includes('dropped')), errors.join('\n')); + cm.getMessageCount = count; + queue.ackDeferredWrites(); + await framework.stop(); + reopened = await createFramework(); + const restored = reopened.getAgent('assistant')!.getContextManager(); + assert.equal(restored.queryMessages({ metadata: { deferredWriteId: first.id } } as never).messages.length, 0, 'refused body never stored'); + assert.equal(restored.queryMessages({ metadata: { deferredWriteId: second.id } } as never).messages.length, 1, 'successor not duplicated'); + assert.deepEqual(drainInternals(reopened).deferredMessages, [], 'durable queue does not resurrect refused entry'); + } finally { + console.error = error; + cm.getMessageCount = count; + if (reopened) await reopened.stop(); + else await framework.stop(); + } + }); + } + for (const landed of [true, false]) { it(`a write whose landing is unknown keeps its mark across a restart, so it's stored once after it (${landed ? 'its first write landed' : 'nothing landed'})`, async () => { const framework = await createFramework(); @@ -1225,7 +1260,7 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { : null; } - it("at a tool boundary, a sharded body that throws after its first shard: what landed is carried and counts as written, and the rest isn't stored", { skip: NO_SHARDS_LAND_BEFORE_A_THROW }, async () => { + it('at a tool boundary, a sharded body whose write throws partway lands nothing: it waits for the turn end with what follows, and is stored whole then', async () => { membrane.pushResponse(createMockResponse([ { type: 'text', text: 'Moving.' }, { type: 'tool_use', id: 'c1', name: 'robot--move', input: {} }, @@ -1233,8 +1268,8 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { membrane.pushResponse(createMockResponse(text('Done.'))); const framework = await createFramework(); shardOn(framework, 'part one'); - // The strategy throws on the first shard's add, which stops the append - // loop before the second shard. + // The strategy throws on the first shard's add, and context-manager + // removes the shard it wrote before the write throws. throwOnceFromStrategyHook(framework, 'part one'); module.interjections = [ { participant: 'Antra', content: [...text('part one'), ...text('part two')] }, @@ -1247,25 +1282,28 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { framework.pushEvent({ type: 'external-message', source: 'test', content: 'go' } as unknown as ProcessEvent); await framework.runUntilIdle(); } finally { console.error = error; } - assert.deepEqual(injected(), ['Antra: part one', 'Antra: look right'], 'the request carries what the store holds'); + assert.deepEqual(injected(), [], 'nothing of the body landed, so the boundary carried nothing'); assert.deepEqual(windowOf(framework).slice(2), [ 'user: ', + 'assistant: Done.', 'Antra: part one', + 'Antra: part two', 'Antra: look right', - 'assistant: Done.', - ], 'the first shard alone, stored once'); - assert.equal(internalsOf(framework).deferredMessages.length, 0, 'nothing went back into the queue'); - assert.ok(errors.some((e) => e.includes("threw after 1 shard of its body was stored; what was stored counts as written and isn't stored again") - && e.includes('strategy failed')), errors.join('\n')); + ], 'the body, stored whole at the turn end, then what was queued behind it'); + assert.equal(internalsOf(framework).deferredMessages.length, 0, 'nothing is left in the queue'); + assert.ok(errors.some((e) => e.includes('could not be stored; it stays queued') && e.includes('strategy failed')), errors.join('\n')); await framework.stop(); }); - it('at a tool boundary, the shards that landed before a throw are carried as the next compile renders them', { skip: NO_SHARDS_LAND_BEFORE_A_THROW }, async () => { + it('at a tool boundary, when removing the shards of a body whose write threw fails too, the shards that stayed are carried as the next compile renders them', async () => { // Probe by Ansel-2428: the autobiographical strategy shards a long // message on ingest and joins a body's consecutive shards when it // compiles, so the boundary must carry what landed in that shape. The // body with an image between two texts fails here if the join changes // where it puts a non-text block, so the copy of the join can't drift. + // A write that throws partway removes the shards it wrote, so shards + // stay only when that removal fails as well, which the store's removal + // is made to do here. const long = Array.from({ length: 60 }, (_, i) => `Sentence ${i} of a long message from Antra.`).join(' '); const ofLong = (messages: Array<{ content: unknown }>) => messages.filter((m) => JSON.stringify(m.content).includes('Sentence 0 ')); @@ -1295,11 +1333,19 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { } as never], }); const cm = framework.getAgent('assistant')!.getContextManager(); + const internals = cm as unknown as { + strategy: { onNewMessage?: (message: { bodyGroupId?: string }, ctx: unknown) => Promise }; + messageStore: { removeBodyGroup: (id: string) => void }; + }; + // The strategy throws on the throwAt-th shard's add, as the manager's + // own store listener calls it for each record written. + const onNewMessage = internals.strategy.onNewMessage; let shards = 0; - cm.onMessage((event) => { - if (event.type !== 'add' || !(event.message as { bodyGroupId?: string }).bodyGroupId) return; - if (++shards === throwAt) throw new Error('listener failed'); - }); + internals.strategy.onNewMessage = function (this: unknown, message, ctx) { + if (message.bodyGroupId && ++shards === throwAt) throw new Error('strategy failed'); + return onNewMessage ? onNewMessage.call(this, message, ctx) : Promise.resolve(); + }; + internals.messageStore.removeBodyGroup = () => { throw new Error('removal failed'); }; module.interjections = [{ participant: 'Antra', content }]; const error = console.error; console.error = () => {}; @@ -1307,7 +1353,7 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { framework.pushEvent({ type: 'external-message', source: 'test', content: 'go' } as unknown as ProcessEvent); await framework.runUntilIdle(); const landed = cm.getAllMessages().filter((m) => (m as { bodyGroupId?: string }).bodyGroupId); - assert.equal(landed.length, throwAt, `${name}: ${throwAt} shards landed before the throw stopped the body`); + assert.equal(landed.length, throwAt, `${name}: ${throwAt} shards stayed, the removal having failed`); const injected = ofLong(membrane.lastStream!.receivedToolResultOptions[0]?.injectedMessages ?? []); // The next turn's request is compiled from the store. membrane.pushResponse(createMockResponse(text('Okay.'))); @@ -1324,6 +1370,37 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { } }); + it('at a tool boundary, a queued write context-manager refuses as malformed is dropped, and what is queued behind it lands', async () => { + // addMessage refuses such content before it's queued, so an entry like + // this one reaches a drain only from a queue an earlier process + // persisted: it's placed in the queue as boot's restore would place it, + // ahead of the interjection by its seq. + module.duringTool = () => { + (module.framework as unknown as { deferredMessages: unknown[] }).deferredMessages.push({ + id: 'persisted-before', seq: 0, participant: 'Antra', forAgent: 'assistant', heldAt: Date.now(), + // A text block whose text isn't a string, which the store refuses. + content: [{ type: 'text', text: 5 } as unknown as ContentBlock], + }); + }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + let framework: AgentFramework; + try { + framework = await runTurn('move', [{ participant: 'Antra', content: text('look right') }]); + } finally { console.error = error; } + assert.deepEqual(atBoundary(), ['Antra: look right', 'Antra: 5'], 'both were queued when the boundary drained'); + assert.deepEqual(injected(), ['Antra: look right'], 'what was queued behind it is carried'); + assert.deepEqual(windowOf(framework).slice(2), [ + 'user: ', + 'Antra: look right', + 'assistant: Done.', + ], 'stored where the stream carried it, and the refused write nowhere'); + assert.equal(internalsOf(framework).deferredMessages.length, 0, 'the refused write was not queued again'); + assert.ok(errors.some((e) => e.includes("was refused as malformed, so it's dropped rather than queued again") && e.includes('content[0]')), errors.join('\n')); + await framework.stop(); + }); + it('at a tool boundary, a sharded body stored whole is carried as the next compile renders it', async () => { // A body the strategy shards on ingest is stored as several messages, // which a compile joins. A body of one text block reads the same as diff --git a/test/hide-then-turn.test.ts b/test/hide-then-turn.test.ts index cea75f597..429f7c710 100644 --- a/test/hide-then-turn.test.ts +++ b/test/hide-then-turn.test.ts @@ -5,7 +5,7 @@ * context-manager 0.14.0 mishandled: it left later compiles reading message * positions from before the removal, so with the middle of three messages * removed, the agent's next turn was refused with UncoveredDropError - * (anima-research/context-manager#180, fixed in 0.14.1, this package's floor). + * (anima-research/context-manager#180, fixed in 0.14.1, below this package's floor). */ import { test } from 'node:test'; import assert from 'node:assert/strict'; diff --git a/test/host-quiesce.test.ts b/test/host-quiesce.test.ts index bed21fdd2..96f1bd5c3 100644 --- a/test/host-quiesce.test.ts +++ b/test/host-quiesce.test.ts @@ -1296,10 +1296,9 @@ test("resume stores several agents' writes in the order they arrived, and cuts e add('one', 'agent'); add('two', 'scout'); add('three', 'agent'); - // Nothing checks a fresh write's blocks beyond reading each one's type, - // so a block that isn't an object can be queued; the log line names it - // by its JavaScript type. - internals.addMessage('user', [{ type: 'text', text: 'stuck' }, 'a bare string'], undefined, { forAgent: 'agent' }); + // A write whose content the store would accept, which the store below + // refuses all the same; the log line names it by its blocks' types. + internals.addMessage('user', [{ type: 'text', text: 'stuck' }, { type: 'text', text: 'and more' }], undefined, { forAgent: 'agent' }); add('for scout', 'scout'); add('behind the stuck one', 'agent'); add('also for scout', 'scout'); @@ -1314,7 +1313,7 @@ test("resume stores several agents' writes in the order they arrived, and cuts e return realAdd(participant, content, metadata, causedBy); }) as typeof agentCm.addMessage; const named = () => lines.filter((line) => - line.includes(`deferred write ${stuckId} (user: text+string) for agent could not be stored`) + line.includes(`deferred write ${stuckId} (user: text) for agent could not be stored`) && line.includes('a thrown value that has no text')).length; let status = await framework.resume(); @@ -1364,9 +1363,10 @@ test('boot recovery drops a malformed recovered deferred write, naming it, and r // them in the recovery file: a null block, a string block and a number // block beside a text block, content that isn't blocks, a participant // that isn't text, which the line names by its type, and a target that - // is empty or isn't text. Once stored, a string or number block would - // make every compile for the agent throw, since the context manager's - // token estimate keys a WeakMap on each block. + // is empty or isn't text. A store before context-manager 0.16.0 kept a + // string or number block, and every compile for the agent then threw, + // since the context manager's token estimate keys a WeakMap on each + // block; from 0.16.0 the store refuses one. const file = join(storePath, 'recovery', 'deferred-writes.json'); const doc = JSON.parse(readFileSync(file, 'utf8')) as { pending: Array> }; assert.equal(doc.pending.length, 2); @@ -1403,6 +1403,58 @@ test('boot recovery drops a malformed recovered deferred write, naming it, and r } }); +test('a recovered deferred write context-manager refuses as malformed is dropped at the first flush, and the ack takes it out of the recovery file', { timeout: 60_000 }, async () => { + // A write addMessage refuses now could sit in a queue an earlier process + // persisted: a text block whose text isn't a string is an object block, so + // boot restores it, and the store refuses it at the flush that meets it. + const dir = mkdtempSync(join(tmpdir(), 'host-quiesce-refused-')); + const storePath = join(dir, 'store'); + const config = () => ({ + storePath, + membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test' }], + modules: [], + }); + let framework = await AgentFramework.create(config()); + const lines: string[] = []; + const error = console.error; + try { + await framework.quiesce({ reason: 'surgery' }); + const before = framework.getAgent('agent')!.getContextManager().getAllMessages().length; + const add = (framework as unknown as { addMessage(p: string, c: unknown[]): unknown }).addMessage.bind(framework); + add('user', [{ type: 'text', text: 'first' }]); + add('user', [{ type: 'text', text: 'second' }]); + await framework.stop(); + + const file = join(storePath, 'recovery', 'deferred-writes.json'); + const doc = JSON.parse(readFileSync(file, 'utf8')) as { pending: Array> }; + assert.equal(doc.pending.length, 2); + // Between the two writes, so the second is queued behind it: the second + // moves up one seq, and the malformed entry takes its place. + const second = doc.pending[1]!; + const seq = second.seq as number; + second.seq = seq + 1; + doc.pending.splice(1, 0, { id: 'number-text', seq, participant: 'user', content: [{ type: 'text', text: 5 }] }); + writeFileSync(file, JSON.stringify(doc)); + + framework = await AgentFramework.create(config()); + assert.equal(framework.getHostModeStatus().deferredWrites, 3, 'boot restores it with the others'); + console.error = (...args: unknown[]) => { lines.push(args.map(String).join(' ')); }; + await framework.resume(); + console.error = error; + assert.ok(lines.some((line) => line.includes("was refused as malformed, so it's dropped rather than queued again") && line.includes('content[0]')), lines.join('\n')); + const texts = framework.getAgent('agent')!.getContextManager().getAllMessages().slice(before) + .map((m) => (m.content[0] as { text: string }).text); + assert.deepEqual(texts, ['first', 'second'], 'what was queued behind it landed, in order'); + assert.equal(framework.getHostModeStatus().deferredWrites, 0, 'nothing is queued again'); + assert.ok(!readFileSync(file, 'utf8').includes('number-text'), 'and the recovery file no longer holds it'); + } finally { + console.error = error; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } +}); + test('boot recovery keeps one entry per id, and numbers a seq that isn\'t a positive whole number after the others', { timeout: 60_000 }, async () => { const dir = mkdtempSync(join(tmpdir(), 'host-quiesce-identity-')); const storePath = join(dir, 'store'); diff --git a/test/inbound-source.test.ts b/test/inbound-source.test.ts index f3cc531b0..2bcc6da53 100644 --- a/test/inbound-source.test.ts +++ b/test/inbound-source.test.ts @@ -29,7 +29,7 @@ import { type ToolCall, type ToolResult, } from '../src/index.js'; -import { ContextManager } from '@animalabs/context-manager'; +import { ContextManager, MalformedContentError } from '@animalabs/context-manager'; import type { ContentBlock } from '@animalabs/membrane'; import { MockMembrane } from './helpers/mock-membrane.js'; import { canonicalJson } from '../src/mcpl/inbound-source.js'; @@ -729,7 +729,7 @@ describe('sourceBodyDigest hashes a body as the store keeps it', () => { } }); - it('a block the store refuses is refused at write, and a blob_ref is digested as given, which its copy read back does not match', async () => { + it('refused content and caller-supplied blob_ref cannot be written; unresolved and resolved reads have distinct digests', async () => { const dir = mkdtempSync(join(tmpdir(), 'stored-digest-refused-')); try { const cm = await ContextManager.open({ path: join(dir, 'store.chronicle'), strategy: new PassthroughStrategy() }); @@ -743,17 +743,19 @@ describe('sourceBodyDigest hashes a body as the store keeps it', () => { ] as const) { assert.throws(() => cm.addMessage('user', [block] as never), Error, `${what}: refused at write`); } - // A blob_ref names media the store already holds; it is resolved to - // inline media on read, so the copy read back hashes differently. + // Only the store may write blob_ref. An unresolved read can expose + // one, but callers must resolve it before attempting a new write. const held = cm.addMessage('user', [image(png)] as never); const ref = cm.getMessageWindow(0, cm.getMessageCount(), { resolveBlobs: false }).messages .find((m) => m.id === held)!.content[0] as { type: string }; assert.equal(ref.type, 'blob_ref'); const given = [ref]; - const id = cm.addMessage('user', given as never, { storedBodyDigest: sourceBodyDigest(given) }); - const back = cm.getMessage(id)!; + const before = cm.getMessageCount(); + assert.throws(() => cm.addMessage('user', given as never), MalformedContentError, 'a raw blob_ref is store-only'); + assert.equal(cm.getMessageCount(), before, 'refused write adds no record'); + const back = cm.getMessage(held)!; assert.deepEqual(back.content, [image(png)], 'resolved on read'); - assert.notEqual(sourceBodyDigest(back.content), (back.metadata as Record).storedBodyDigest, 'the documented mismatch'); + assert.notEqual(sourceBodyDigest(given), sourceBodyDigest(back.content), 'unresolved refs are not inline content'); } finally { cm.close(); } diff --git a/test/inference-exhaustion-rewind.test.ts b/test/inference-exhaustion-rewind.test.ts index b21aa683f..2e81f5740 100644 --- a/test/inference-exhaustion-rewind.test.ts +++ b/test/inference-exhaustion-rewind.test.ts @@ -190,8 +190,8 @@ test('nothing left to shed: the breaker degrades to logging, no crash, no loop', assert.ok(errs.some((e) => e.includes('nothing left to shed'))); }); -// Anarchid's review of membrane #56 (M3): from #56 (in membrane 0.5.87, this -// package's floor) one status table serves every adapter, so `invalid_request` also +// Anarchid's review of membrane #56 (M3): from #56 (in membrane 0.5.87, below +// this package's floor) one status table serves every adapter, so `invalid_request` also // reaches agent-framework for an Anthropic 404 (a model that doesn't exist), // which no shed can fix. With 0.5.86 the raw-fetch Responses adapter already // files every status it doesn't list as invalid_request. The breaker runs @@ -305,7 +305,7 @@ test("a membrane error's type and status reach the gate through the exhausted tr }); test('shedMayHelp: an invalid_request at 400, 413 or 422 or with no status, or a context_length at 413, and nothing else', () => { - // Every membrane error type (0.5.87), the framework's own two, and none. + // Every membrane error type (0.6.0), the framework's own two, and none. const types = ['invalid_request', 'rate_limit', 'context_length', 'auth', 'server', 'network', 'timeout', 'abort', 'safety', 'unsupported', 'unknown', 'over_budget', 'context_refusal', undefined]; const statuses: Array = [undefined]; diff --git a/test/maintenance-withdrawal-ladder.test.ts b/test/maintenance-withdrawal-ladder.test.ts new file mode 100644 index 000000000..5e62aa080 --- /dev/null +++ b/test/maintenance-withdrawal-ladder.test.ts @@ -0,0 +1,115 @@ +/** + * A maintenance call withdrawn partway through context-manager's compression + * fallback ladder is a withdrawal, not a failure (agent-framework#549). + * + * context-manager 0.16.0 rethrows an error branded as its CompressionWithdrawn + * from every rung of the ladder, recording nothing, and records anything else + * a rung's call throws as that rung's provider error, quarantining the chunk's + * request family once every rung has failed. The framework withdraws a + * maintenance call at its provider gate while a surgery holds the store. So + * the compression here runs through the framework's own maintenance membrane: + * its canonical call is refused, and a surgery takes the store during that + * call, so the ladder's next rung, the source-only call, is withdrawn at the + * gate. + */ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { AutobiographicalStrategy, ContextManager, PassthroughStrategy } from '@animalabs/context-manager'; +import type { Membrane, NormalizedRequest, NormalizedResponse } from '@animalabs/membrane'; +import { AgentFramework } from '../src/index.js'; + +const MODEL = 'test-model'; + +/** A provider that refuses the first call it gets, the compression's + * canonical call, and runs `during` while answering it. */ +class RefusingMembrane { + calls: NormalizedRequest[] = []; + during: () => void = () => {}; + + async complete(request: NormalizedRequest): Promise { + this.calls.push(request); + if (this.calls.length === 1) this.during(); + return { + content: [], + stopReason: 'refusal', + usage: { inputTokens: 100, outputTokens: 0 }, + raw: { response: { stop_details: { category: 'reasoning_extraction' } } }, + } as unknown as NormalizedResponse; + } + + asMembrane(): Membrane { + return this as unknown as Membrane; + } +} + +describe("a maintenance call withdrawn inside context-manager's fallback ladder", () => { + it('is no failure: nothing is quarantined, and nothing after it is sent', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-withdrawal-ladder-')); + const provider = new RefusingMembrane(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane: provider.asMembrane(), + agents: [{ name: 'agent', model: MODEL, systemPrompt: 'test', strategy: new PassthroughStrategy() }], + modules: [], + syncIntervalMs: 0, + }); + const internals = framework as unknown as { + auxiliaryMembraneFor(agentName: string): Membrane; + reserveStoreForSurgery(verb: string, agentName: string): () => void; + }; + let release: (() => void) | undefined; + provider.during = () => { release = internals.reserveStoreForSurgery('roll back', 'agent'); }; + + const strategy = new AutobiographicalStrategy({ + compressionModel: MODEL, + targetChunkTokens: 100, + recentWindowTokens: 0, + headWindowTokens: 0, + autoTickOnNewMessage: false, + minChunkCharsForLLM: 0, + mergeThreshold: 99, + // After a refused canonical call, the source-only rung needs no recall + // to drop, so it's the rung that runs next here. + compressionRefusalCurveFallbacks: 0, + compressionSourceOnlyFallback: true, + }); + // The framework's maintenance membrane, with each call it's asked for and + // what it threw. + const maintenance = internals.auxiliaryMembraneFor('agent'); + const asked: NormalizedRequest[] = []; + const thrown: unknown[] = []; + const counting = { + complete: async (request: NormalizedRequest) => { + asked.push(request); + try { return await maintenance.complete(request); } catch (error) { thrown.push(error); throw error; } + }, + } as unknown as Membrane; + const manager = await ContextManager.open({ path: join(dir, 'compression'), strategy, membrane: counting }); + const logs: string[] = []; + const error = console.error; + console.error = (...parts: unknown[]) => { logs.push(parts.map(String).join(' ')); }; + try { + for (let i = 0; i < 14; i++) { + manager.addMessage(i % 2 ? 'Claude' : 'User', [{ type: 'text', text: `raw-${i} ` + 'substantive '.repeat(18) }]); + } + const [settled] = await Promise.allSettled([manager.tick()]); + assert.ok(release, 'the canonical call was made, and the surgery took the store during it'); + assert.equal(asked.length, 2, 'the ladder went on to the source-only rung'); + assert.equal(provider.calls.length, 1, 'which was withdrawn at the gate, before its call was sent'); + assert.equal((thrown[0] as Error | undefined)?.name, 'MaintenanceWithdrawnError'); + assert.equal(settled.status, 'rejected', 'the tick ends with the withdrawal'); + assert.equal((settled as PromiseRejectedResult).reason, thrown[0], 'rethrown unchanged'); + assert.deepEqual(strategy.getCompressionQuarantineStatus(), { count: 0, keys: [], spans: [] }, 'nothing is quarantined'); + assert.ok(!logs.some((line) => line.includes('Failed to compress chunk')), logs.join('\n')); + } finally { + console.error = error; + release?.(); + manager.close(); + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); +}); diff --git a/test/malformed-content-refusal.test.ts b/test/malformed-content-refusal.test.ts new file mode 100644 index 000000000..7e144743d --- /dev/null +++ b/test/malformed-content-refusal.test.ts @@ -0,0 +1,173 @@ +/** + * MCPL content that context-manager won't store is refused where it would + * be stored or queued. context-manager refuses a write whose content + * doesn't match ContentBlock (MalformedContentError, its #185), and + * malformedContentPlaces names where without throwing. Content from an MCPL + * server is converted at the boundary into blocks the store accepts, so the + * push and channel lanes' check is the backstop for events that never + * crossed it, such as a module's, which these tests hand the lanes + * directly. The lanes ask it first, so such a message stores no row, spawns + * no fork and queues no wake, and their log line names where it came from. + * Content the store mends, a tool_use whose input isn't an object, is + * stored as before, and absent content is the empty-content check's. + */ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import type { ContentBlock } from '@animalabs/membrane'; +import { AgentFramework } from '../src/index.js'; +import { logValue } from '../src/mcpl/log-value.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; + +const TS = '2026-10-18T00:00:00.000Z'; + +/** Content the store refuses, each with a phrase of the place it names. */ +const MALFORMED: Array<[string, unknown[], string]> = [ + ['a block that is a string', ['hello'], 'content[0]'], + ['a text block whose text is a number', [{ type: 'text', text: 'hi' }, { type: 'text', text: 5 }], 'content[1]'], + ['an image whose source is a string', [{ type: 'image', source: 'data' }], 'content[0]'], +]; + +async function harness(extra: Partial[0]> = {}) { + const dir = mkdtempSync(join(tmpdir(), 'malformed-content-af-')); + const membrane = new MockMembrane(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), membrane: membrane.asMembrane(), + agents: [{ name: 'assistant', model: 'test', systemPrompt: 'test' }], + modules: [], + ...extra, + }); + const added: string[] = []; + const refused: Array> = []; + const traces: Array> = []; + let starts = 0; + framework.onTrace((e: any) => { + traces.push(e); + if (e.type === 'message:added') added.push(e.source); + if (e.type === 'mcpl:malformed-content-refused') refused.push(e); + if (e.type === 'inference:started') starts++; + }); + // What the agent's own store holds, which a refusal mustn't add to. + const stored = () => framework.getAgent('assistant')!.getContextManager().getAllMessages().length; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + return { + framework, membrane, added, refused, traces, errors, stored, starts: () => starts, + close: async () => { + console.error = error; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + }, + }; +} + +for (const [label, content, place] of MALFORMED) { + test(`push: ${label} is refused, storing no row and queueing no wake, and the line names the server and channel`, async (t) => { + const x = await harness(); t.after(x.close); + x.membrane.pushResponse(createMockResponse([{ type: 'text', text: 'invented cause' }] as ContentBlock[])); + const before = x.stored(); + const placement = (x.framework as any).handleMcplPushEvent({ + type: 'mcpl:push-event', serverId: 'module', featureSet: 'fs', eventId: 'e1', + content, timestamp: TS, inferenceId: 'i', triggerInference: true, + origin: { mcplChannelId: 'discord:g:c' }, + }); + await x.framework.runUntilIdle(); + assert.equal(placement, undefined, 'nothing was placed'); + assert.deepEqual(x.added, [], 'no row traced'); + assert.equal(x.stored(), before, 'and none in the store'); + assert.equal(x.starts(), 0, 'no wake'); + assert.equal(x.refused.length, 1); + assert.equal(x.refused[0]!.lane, 'push'); + assert.equal(x.refused[0]!.serverId, 'module'); + assert.equal(x.refused[0]!.eventId, 'e1'); + assert.equal(x.refused[0]!.channelId, 'discord:g:c'); + assert.ok((x.refused[0]!.places as string[]).some((p) => p.includes(place)), JSON.stringify(x.refused[0])); + const line = `[push-event-refused] server=module channel=${logValue('discord:g:c')} eventId=${logValue('e1')} reason=malformed-content: `; + assert.ok(x.errors.some((e) => e.startsWith(line) && e.includes(place)), x.errors.join('\n')); + }); + + test(`channel: ${label} is refused, storing no row and queueing no wake, and the line names the server and channel`, async (t) => { + const x = await harness(); t.after(x.close); + x.membrane.pushResponse(createMockResponse([{ type: 'text', text: 'invented cause' }] as ContentBlock[])); + const before = x.stored(); + const placement = await (x.framework as any).handleMcplChannelIncoming({ + type: 'mcpl:channel-incoming', serverId: 'discord', channelId: 'discord:g:c', messageId: 'm1', eventId: 'ev1', + author: { id: 'u', name: 'User' }, content, timestamp: TS, triggerInference: true, + }); + await x.framework.runUntilIdle(); + assert.equal(placement, undefined, 'nothing was placed'); + assert.deepEqual(x.added, [], 'no row traced'); + assert.equal(x.stored(), before, 'and none in the store'); + assert.equal(x.starts(), 0, 'no wake'); + assert.deepEqual(x.refused.map((r) => [r.lane, r.serverId, r.channelId, r.messageId, r.eventId]), [['channel', 'discord', 'discord:g:c', 'm1', 'ev1']]); + assert.ok((x.refused[0]!.places as string[]).some((p) => p.includes(place)), JSON.stringify(x.refused[0])); + const line = `[channel-incoming-refused] server=discord channel=${logValue('discord:g:c')} messageId=${logValue('m1')} reason=malformed-content: `; + assert.ok(x.errors.some((e) => e.startsWith(line) && e.includes(place)), x.errors.join('\n')); + }); +} + +test('push: a tool_use whose input is not an object is mended by the store, so it is stored, not refused', async (t) => { + const x = await harness(); t.after(x.close); + x.membrane.pushResponse(createMockResponse([{ type: 'text', text: 'heard' }] as ContentBlock[])); + (x.framework as any).handleMcplPushEvent({ + type: 'mcpl:push-event', serverId: 'module', featureSet: 'fs', eventId: 'e2', + content: [{ type: 'text', text: 'a call' }, { type: 'tool_use', id: 't1', name: 'x', input: 'not an object' }], + timestamp: TS, inferenceId: 'i', triggerInference: true, + }); + await x.framework.runUntilIdle(); + assert.deepEqual(x.refused, []); + assert.equal(x.added.length, 1, 'stored'); + const stored = x.framework.getAgent('assistant')!.getContextManager().getAllMessages() + .flatMap((m) => m.content).find((b) => b.type === 'tool_use') as { input?: unknown } | undefined; + assert.deepEqual(stored?.input, {}, 'with its input mended to an object'); +}); + +test("channel: with conversation routing on, a malformed message that would bind a fork spawns none and stores nothing, where the same message well-formed binds one", async (t) => { + const x = await harness({ conversations: { templateAgent: 'assistant', bind: { channel: 'mention' } } } as never); + t.after(x.close); + const agents = () => x.framework.getAllAgents().length; + const before = agents(); + const message = (messageId: string, content: unknown[], triggerInference: boolean) => (x.framework as any).handleMcplChannelIncoming({ + type: 'mcpl:channel-incoming', serverId: 'srv', channelId: 'slack:C1', messageId, + author: { id: 'U1', name: 'alice' }, content, timestamp: TS, triggerInference, + metadata: { mentioned: true }, + }); + // Visible text beside a malformed block, so the empty-content check would + // let it through and only the malformed-content check stops it. + const placement = await message('m2', [{ type: 'text', text: 'bot, help' }, { type: 'image', source: 'not-an-object' }], true); + await x.framework.runUntilIdle(); + assert.equal(placement, undefined, 'nothing was placed'); + assert.equal(agents(), before, 'no fork was spawned'); + assert.deepEqual(x.traces.filter((e) => String(e.type).startsWith('mcpl:conversation-')), [], 'nor bound'); + assert.deepEqual(x.added, [], 'no row traced'); + assert.equal(x.refused.length, 1, 'refused instead'); + assert.equal(x.starts(), 0, 'and no wake'); + await message('m3', [{ type: 'text', text: 'bot, help' }], false); + await x.framework.runUntilIdle(); + assert.equal(agents(), before + 1, 'the same message well-formed spawns a fork'); +}); + +for (const lane of ['push', 'channel'] as const) { + test(`${lane}: absent content, undefined or null, is the empty-content check's, not refused as malformed`, async (t) => { + const x = await harness(); t.after(x.close); + for (const [id, content] of [['e3', undefined], ['e4', null]] as const) { + const placement = lane === 'push' + ? (x.framework as any).handleMcplPushEvent({ + type: 'mcpl:push-event', serverId: 'module', featureSet: 'fs', eventId: id, + content, timestamp: TS, inferenceId: 'i', triggerInference: true, + }) + : await (x.framework as any).handleMcplChannelIncoming({ + type: 'mcpl:channel-incoming', serverId: 'discord', channelId: 'discord:g:c', messageId: `m-${id}`, eventId: id, + author: { id: 'u', name: 'User' }, content, timestamp: TS, triggerInference: true, + }); + await x.framework.runUntilIdle(); + assert.equal(placement, undefined, `${String(content)}: nothing was placed`); + } + assert.deepEqual(x.refused, [], 'no malformed-content refusal'); + assert.deepEqual(x.traces.filter((e) => e.type === 'mcpl:empty-content-dropped').map((e) => e.lane), [lane, lane]); + assert.deepEqual(x.added, []); + }); +} diff --git a/test/malformed-content-writes.test.ts b/test/malformed-content-writes.test.ts new file mode 100644 index 000000000..50f72cf10 --- /dev/null +++ b/test/malformed-content-writes.test.ts @@ -0,0 +1,213 @@ +/** + * Content context-manager won't store is refused where the framework writes + * it, whether the write would be stored at once or deferred while a turn + * runs, as the store's own write and edit refuse it: a module's write, and + * its edit of a write still held for the turn's end, throw + * context-manager's MalformedContentError. A process response with such + * content is refused whole before any response applies, with a line and a + * trace, and throws nothing; the API's message.send refuses content that + * isn't text at its call. Before, a deferred write was refused only at a + * later flush, where the writer never heard of it, and a held write's edit + * could replace valid content with content the drain then dropped. + */ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import type { ContentBlock } from '@animalabs/membrane'; +import { MalformedContentError, malformedContentPlaces } from '@animalabs/context-manager'; +import type { Module, ModuleContext, ProcessEvent, ProcessState, EventResponse, ToolDefinition, ToolCall, ToolResult } from '../src/index.js'; +import { AgentFramework, ApiServer } from '../src/index.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; + +const text = (t: string): ContentBlock[] => [{ type: 'text', text: t } as ContentBlock]; +/** A text block whose text isn't a string, which the store refuses. */ +const MALFORMED = [{ type: 'text', text: 5 }] as unknown as ContentBlock[]; +/** An image whose source is a string, which the store refuses for another place. */ +const MALFORMED_IMAGE = [{ type: 'image', source: 'not-an-object' }] as unknown as ContentBlock[]; + +/** A module with one tool, `act`, that runs `during` while the turn waits + * for it, and that answers an external message with `onExternal`'s response + * and a `ping` with `onPing`'s. */ +class Writer implements Module { + readonly name: string; + ctx!: ModuleContext; + during: (() => void) | null = null; + onExternal: EventResponse = { addMessages: [{ participant: 'Antra', content: text('go') }], requestInference: true }; + onPing: EventResponse = {}; + constructor(name = 'writer') { this.name = name; } + async start(ctx: ModuleContext): Promise { this.ctx = ctx; } + async stop(): Promise {} + getTools(): ToolDefinition[] { + return [{ name: 'act', description: 'Act', inputSchema: { type: 'object', properties: {} } }]; + } + async handleToolCall(_call: ToolCall): Promise { + const during = this.during; + this.during = null; + during?.(); + return { success: true, data: { ok: true } }; + } + async onProcess(event: ProcessEvent, _state: ProcessState): Promise { + if (event.type === 'external-message') return this.onExternal; + if ((event as { type: string }).type === 'ping') return this.onPing; + return {}; + } +} + +async function harness(modules: Writer[]) { + const dir = mkdtempSync(join(tmpdir(), 'malformed-writes-af-')); + const membrane = new MockMembrane(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), membrane: membrane.asMembrane(), + agents: [{ name: 'assistant', model: 'test', systemPrompt: 'test' }], + modules, + processLogging: { persist: true }, + }); + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + const traces: Array> = []; + framework.onTrace((e) => { traces.push(e as unknown as Record); }); + const window = () => framework.getAgent('assistant')!.getContextManager().getAllMessages() + .map((m) => `${m.participant}: ${m.content.map((b) => (b.type === 'text' ? (b as { text: string }).text : `<${b.type}>`)).join(' ')}`); + const queued = () => (framework as unknown as { deferredMessages: unknown[] }).deferredMessages.length; + return { + framework, membrane, errors, traces, window, queued, + close: async () => { + console.error = error; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + }, + }; +} + +/** A turn that calls `writer--act` once, then answers. */ +function scriptTurn(membrane: MockMembrane): void { + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Acting.' }, + { type: 'tool_use', id: 'c1', name: 'writer--act', input: {} }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse(text('Done.'))); +} + +function catchFrom(f: () => unknown): unknown { + try { f(); } catch (error) { return error; } + return undefined; +} + +test("a module's malformed write throws while a turn runs, as it does while the agent is idle, and nothing is queued", async (t) => { + const writer = new Writer(); + const x = await harness([writer]); t.after(x.close); + scriptTurn(x.membrane); + let duringTurn: unknown; + let queuedThen = -1; + writer.during = () => { + duringTurn = catchFrom(() => writer.ctx.addMessage('Antra', MALFORMED)); + queuedThen = x.queued(); + writer.ctx.addMessage('Antra', text('look right')); + }; + x.framework.pushEvent({ type: 'external-message', source: 'test', content: 'go' } as unknown as ProcessEvent); + await x.framework.runUntilIdle(); + assert.ok(duringTurn instanceof MalformedContentError, String(duringTurn)); + assert.match((duringTurn as Error).message, /^Message refused: content\[0\]/); + assert.equal(queuedThen, 0, 'the refused write was never queued'); + assert.ok(x.window().includes('Antra: look right'), 'the next write landed'); + const idle = catchFrom(() => writer.ctx.addMessage('Antra', MALFORMED)); + assert.ok(idle instanceof MalformedContentError, String(idle)); + assert.equal((idle as Error).message, (duringTurn as Error).message, 'the same refusal either way'); +}); + +test("an edit of a held write to malformed content throws and leaves the write as it was, which lands", async (t) => { + const writer = new Writer(); + const x = await harness([writer]); t.after(x.close); + scriptTurn(x.membrane); + const edits: unknown[] = []; + writer.during = () => { + const held = writer.ctx.addMessage('Antra', text('original')); + edits.push(catchFrom(() => writer.ctx.editMessage(held, MALFORMED))); + edits.push(catchFrom(() => writer.ctx.editMessage(held, [null] as unknown as ContentBlock[]))); + }; + x.framework.pushEvent({ type: 'external-message', source: 'test', content: 'go' } as unknown as ProcessEvent); + await x.framework.runUntilIdle(); + assert.equal(edits.length, 2); + for (const edit of edits) { + assert.ok(edit instanceof MalformedContentError, String(edit)); + assert.match((edit as Error).message, /^Edit of .* refused: content\[0\]/); + } + const window = x.window(); + assert.ok(window.includes('Antra: original'), `the held write landed as it was:\n${window.join('\n')}`); + assert.equal(window.at(-1), 'assistant: Done.', 'and the turn went on past its tool boundary'); +}); + +test("a process response with a malformed message or edit is refused whole, queueing no inference, and other modules' responses still apply", async (t) => { + const bad = new Writer('bad'); + const badEdit = new Writer('bad-edit'); + const good = new Writer('good'); + bad.onPing = { addMessages: [{ participant: 'Antra', content: text('first') }, { participant: 'Antra', content: MALFORMED }], requestInference: true }; + badEdit.onPing = { addMessages: [{ participant: 'Edit', content: text('before the edit') }], editMessages: [{ messageId: 'm-any', content: MALFORMED_IMAGE }] }; + good.onPing = { addMessages: [{ participant: 'Good', content: text('from good') }] }; + const x = await harness([bad, badEdit, good]); t.after(x.close); + x.framework.pushEvent({ type: 'ping', source: 'test' } as unknown as ProcessEvent); + await x.framework.runUntilIdle(); + const window = x.window(); + assert.ok(!window.includes('Antra: first'), `none of the refused response's messages was added:\n${window.join('\n')}`); + assert.ok(!window.includes('Edit: before the edit'), 'nor of the one whose edit is refused'); + assert.ok(window.includes('Good: from good'), 'the other module\'s response was applied'); + assert.equal(x.membrane.calls.length, 0, 'no inference ran'); + const lines = (module: string) => x.errors.filter((e) => e.startsWith(`[process-response-refused] module=${module} event=ping reason=malformed-content: `)); + assert.equal(lines('bad').length, 1, x.errors.join('\n')); + assert.ok(lines('bad')[0]!.includes('addMessages[1] refused'), lines('bad')[0]); + assert.equal(lines('bad-edit').length, 1, x.errors.join('\n')); + assert.ok(lines('bad-edit')[0]!.includes('editMessages[0] refused'), lines('bad-edit')[0]); + assert.equal(lines('good').length, 0, 'one line per refused response, and none for the one that applied'); + const completed = x.traces.filter((e) => e.type === 'process:completed' && (e.processEvent as { type: string }).type === 'ping'); + assert.deepEqual(completed.map((e) => (e.responses as Array<{ moduleName: string }>).map((r) => r.moduleName)), [['good']], + 'the trace lists only the response that applied'); + const logged = x.framework.queryProcessLogs({ eventType: 'ping' }).entries + .map(({ entry }) => (entry.responses as Array<{ moduleName: string }>).map((r) => r.moduleName)); + assert.deepEqual(logged, [['good']], 'and so does the process log'); + assert.deepEqual( + x.traces.filter((e) => e.type === 'process:response-refused').map((e) => [e.moduleName, e.eventType, e.where, e.places]), + [ + ['bad', 'ping', 'addMessages[1]', malformedContentPlaces(MALFORMED)], + ['bad-edit', 'ping', 'editMessages[0]', malformedContentPlaces(MALFORMED_IMAGE)], + ], + 'a trace names each refused response, with the places of the message or edit it refused', + ); + assert.notDeepEqual(malformedContentPlaces(MALFORMED), malformedContentPlaces(MALFORMED_IMAGE), 'the two refusals name different places'); +}); + +test("the API's message.send refuses content that isn't text, before any module sees it", async (t) => { + const writer = new Writer(); + const x = await harness([writer]); t.after(x.close); + const server = new ApiServer(x.framework, { port: 0 }); + const send = (content: unknown) => + (server as unknown as { cmdMessageSend(p: unknown): Promise }).cmdMessageSend({ participant: 'user', content }); + await assert.rejects(send(5), /content must be text/); + await assert.rejects(send([{ type: 'text', text: 'blocks' }]), /content must be text/); + await x.framework.runUntilIdle(); + const apiEvents = () => x.traces.filter((e) => e.type === 'process:received' && (e.processEvent as { type: string }).type === 'api:message').length; + assert.equal(apiEvents(), 0, 'no event was queued'); + assert.deepEqual(await send('hello'), { messageId: 'pending' }, 'text is accepted as before'); + await x.framework.runUntilIdle(); + assert.equal(apiEvents(), 1, 'and its event queued'); +}); + +test("console input whose only message is refused isn't counted as accepted", async (t) => { + const writer = new Writer(); + const x = await harness([writer]); t.after(x.close); + const accepted: unknown[] = []; + (x.framework as unknown as { inboundAcceptanceObserver: unknown }).inboundAcceptanceObserver = { + inboundAccepted: (source: unknown) => accepted.push(source), + }; + writer.onExternal = { addMessages: [{ participant: 'user', content: text('hello') }] }; + x.framework.pushEvent({ type: 'external-message', source: 'tui', content: 'hello', metadata: {} } as unknown as ProcessEvent); + await x.framework.runUntilIdle(); + assert.equal(accepted.length, 1, 'input a module turns into a message is accepted'); + writer.onExternal = { addMessages: [{ participant: 'user', content: MALFORMED }] }; + x.framework.pushEvent({ type: 'external-message', source: 'tui', content: 'hello', metadata: {} } as unknown as ProcessEvent); + await x.framework.runUntilIdle(); + assert.equal(accepted.length, 1, 'input whose message is refused is not'); + assert.ok(x.errors.some((e) => e.startsWith('[process-response-refused] module=writer event=external-message')), x.errors.join('\n')); +}); diff --git a/test/npm-test-timeout.test.ts b/test/npm-test-timeout.test.ts index 68b6591c3..80df8dccd 100644 --- a/test/npm-test-timeout.test.ts +++ b/test/npm-test-timeout.test.ts @@ -1,18 +1,19 @@ /** - * `npm test` passes --test-timeout=600000 to Node's runner. Under Node 20 the - * bound applies to each test file as a whole, from the file's own start: a - * file still running at ten minutes is cancelled and fails by its path, - * whether it's stuck in a test or while loading. Under Node 24 it applies to - * each test and each hook: one still running at ten minutes fails, and the - * report names its test or suite. A file stuck outside its tests and hooks, - * such as while loading, runs on until CI's timeout-minutes ends the job. + * `npm test` passes --test-timeout=600000 to Node's runner. Under Node 22.12, + * as under 20, the bound applies to each test file as a whole, from the file's + * own start: a file still running at ten minutes is cancelled and fails by its + * path, whether it's stuck in a test or while loading. Under Node 24 it + * applies to each test and each hook: one still running at ten minutes fails, + * and the report names its test or suite. A file stuck outside its tests and + * hooks, such as while loading, runs on until CI's timeout-minutes ends the + * job. * - * Ten minutes is loose on purpose: under Node 20 the slowest file, - * mcpl-runtime-lifecycle, takes about three, and a bound near that could fail - * a correct change on a loaded runner. + * Ten minutes is loose on purpose: the slowest file, mcpl-runtime-lifecycle, + * took about three under Node 20, and a bound near that could fail a correct + * change on a loaded runner. * * ci.yml's unforced step passes the same bound. Build & Test's time limit - * stays at least a passing job's time above it, so that under Node 20 a file + * stays at least a passing job's time above it, so that under Node 22.12 a file * that sticks within that time fails by name before GitHub cancels the job. */ import { test } from 'node:test'; diff --git a/test/tool-input-null-arguments.test.ts b/test/tool-input-null-arguments.test.ts new file mode 100644 index 000000000..833002547 --- /dev/null +++ b/test/tool-input-null-arguments.test.ts @@ -0,0 +1,113 @@ +/** + * A tool call whose arguments text is JSON `null`, as an OpenAI-family + * provider can send it. From membrane 0.6.0 every adapter hands a call's + * input over as a plain object, so `"null"` arrives as `{}`, with the text + * kept in `unparseableInput`. On membrane 0.5.87 the OpenAI-family adapters' + * `safeParseJson` gave `null`, and a tool that read a field of its input + * answered the model with a TypeError ("Cannot read properties of null"). + * `skip_reply` reads a null input as no wake since agent-framework#501, so + * the framework half below, `skip_reply` running as a call with no + * arguments, is a control: it passes with either input. The input + * assertions are what need membrane 0.6.0. + */ +import { describe, it, beforeEach, afterEach } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { Membrane, NativeFormatter, OpenAICompatibleAdapter } from '@animalabs/membrane'; +import type { ContentBlock } from '@animalabs/membrane'; +import { AgentFramework } from '../src/index.js'; +import type { TraceEvent } from '../src/index.js'; +import { ChannelRegistry } from '../src/mcpl/channel-registry.js'; +import type { McplServerRegistry } from '../src/mcpl/server-registry.js'; +import type { FeatureSetManager } from '../src/mcpl/feature-set-manager.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; + +/** The content blocks membrane's OpenAI-compatible adapter makes of a reply + * that calls `skip_reply` with `args` as its arguments text, with the HTTP + * call stubbed. */ +async function openAICompatibleReply(args: string): Promise { + const body = { + id: 'chatcmpl-1', + object: 'chat.completion', + model: 'test-model', + choices: [{ + index: 0, + finish_reason: 'tool_calls', + message: { + role: 'assistant', + content: null, + tool_calls: [{ id: 'call_1', type: 'function', function: { name: 'skip_reply', arguments: args } }], + }, + }], + usage: { prompt_tokens: 1, completion_tokens: 1 }, + }; + const realFetch = globalThis.fetch; + globalThis.fetch = (async () => new Response(JSON.stringify(body), { + status: 200, + headers: { 'content-type': 'application/json' }, + })) as typeof fetch; + try { + const membrane = new Membrane( + new OpenAICompatibleAdapter({ baseURL: 'http://127.0.0.1:9', apiKey: 'test' }), + { formatter: new NativeFormatter() }, + ); + const response = await membrane.complete({ + messages: [{ participant: 'user', content: [{ type: 'text', text: 'nothing to say' }] }], + config: { model: 'test-model', maxTokens: 100 }, + tools: [{ name: 'skip_reply', description: 'End the turn without replying.', inputSchema: { type: 'object', properties: {} } }], + } as never); + return (response as { content: ContentBlock[] }).content; + } finally { + globalThis.fetch = realFetch; + } +} + +describe('a tool call whose arguments are JSON null', () => { + let dir: string; + beforeEach(() => { dir = mkdtempSync(join(tmpdir(), 'tool-input-null-')); }); + afterEach(() => { rmSync(dir, { recursive: true, force: true }); }); + + it('arrives from the OpenAI-compatible adapter as {}, and skip_reply runs as a call with no arguments', async () => { + const content = await openAICompatibleReply('null'); + const call = content.find((block) => block.type === 'tool_use') as + | { type: 'tool_use'; name: string; input: unknown; unparseableInput?: string } + | undefined; + assert.ok(call, `the reply carries the call: ${JSON.stringify(content)}`); + assert.deepEqual(call.input, {}, 'the input is an empty object, not null'); + assert.equal(call.unparseableInput, 'null', 'the arguments text is kept beside it'); + + const membrane = new MockMembrane(); + membrane.pushResponse(createMockResponse(content, 'tool_use')); + // Requested only if the turn went on past the call. + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'after the call' }] as ContentBlock[])); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane: membrane.asMembrane(), + agents: [{ name: 'agent', model: 'test', systemPrompt: 'test' }], + modules: [], + }); + (framework as unknown as { channelRegistry: unknown }).channelRegistry = new ChannelRegistry( + {} as McplServerRegistry, {} as FeatureSetManager, () => {}, () => {}, { eventGatePresent: false }, + ); + const traces: TraceEvent[] = []; + framework.onTrace((event) => { traces.push(event); }); + try { + await (framework as unknown as { + startAgentStream(agent: unknown, trigger: unknown): Promise; + }).startAgentStream(framework.getAgent('agent')!, { + agentName: 'agent', reason: 'test', source: 'test', timestamp: Date.now(), + }); + await framework.runUntilIdle(); + + const outcome = traces.filter((t) => (t.type === 'tool:completed' || t.type === 'tool:failed') + && (t as { callId?: string }).callId === 'call_1'); + assert.deepEqual(outcome.map((t) => t.type), ['tool:completed'], `skip_reply ran without an error: ${JSON.stringify(outcome)}`); + assert.deepEqual(membrane.lastStream?.receivedToolResults, [], 'the turn ended at the call, so no result went back to the model'); + assert.equal(membrane.calls.length, 1, 'and nothing more was requested'); + } finally { + await framework.stop(); + } + }); +});