From 1f35a4446d28dbd67030ade4bd26d65443956a47 Mon Sep 17 00:00:00 2001 From: Hypatia May Date: Mon, 29 Jun 2026 15:59:52 +0800 Subject: [PATCH] fix(compact): address PR 110 review findings Honor cancellation and disposal around async pre-step setup before the loop can open a step or call the model. Route compaction summarization through agent/request so router agents can select the model, and remove the stale model argument from agent/pre-step. Document serial events and the approximate convergence bound, regenerate the Cordis catalog, and add regression coverage for router compaction, HMR cleanup, and assembly/pre-step interruption. --- AGENTS.md | 2 +- docs/cordis-catalog/events-and-services.md | 24 +- docs/core-data-structures/compaction.md | 4 +- docs/core-data-structures/core.md | 2 +- .../2026-06-18-compaction-capability-seam.md | 8 +- .../2026-06-20-generated-cordis-catalog.md | 2 +- examples/coding-agent/tests/compaction.e2e.ts | 7 +- packages/compact/compact-basic/README.md | 4 +- packages/compact/compact-basic/src/index.ts | 60 ++-- packages/compact/compact-basic/src/types.ts | 23 +- .../compact-basic/tests/compact-basic.spec.ts | 192 ++++++++----- packages/compact/compact/README.md | 6 +- packages/compact/compact/src/index.ts | 28 +- .../compact/compact/tests/compact.spec.ts | 27 +- packages/core/agent-loop/src/loop.ts | 58 ++-- packages/core/agent-loop/tests/loop.spec.ts | 18 +- .../agent-loop/tests/review-fixes.spec.ts | 270 ++++++++++++++++++ packages/core/agent/src/types.ts | 9 +- 18 files changed, 551 insertions(+), 193 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index e08dd02e5f..f02940b5ae 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -236,7 +236,7 @@ In the **core** packages (`packages/llm/llm`, `packages/core/tools`, `packages/c Verbose documentation is fine **as long as docs and code stay strictly in sync**. Out-of-sync docs are worse than no docs. **When you change code, update its docs in the SAME change** — grep the package README and the module/JSDoc comments for the old behavior (config keys, defaults, error codes, wire field names, event names) and fix every hit. CI runs `pnpm run doc-sync` (`doc-typecheck` + `verify-cordis-catalog` + `verify-md-wrap` + `verify-md-links` + `verify-doc-refs` + `verify-package-paths` + `verify-rfc-classification` + `verify-type-equiv`), which typechecks every fenced `ts` block in `README.md`, `docs/**/*.md`, and `packages/*/*.md`, regenerates the cordis events/services catalog from source and fails if the committed copy is stale, asserts no hard-wrapped prose paragraphs, checks that every relative Markdown cross-link resolves, checks that every `docs/*.md` path cited in a source comment resolves, checks that every `packages/` reference naming a real package resolves, checks that every RFC is filed under a valid class folder and listed in its index, and checks that every ` ```ts type-equiv ` doc block still matches its source type — across those files plus `AGENTS.md` / `packages/AGENTS.md` — but that scope does NOT catch prose drift in `AGENTS.md` / `packages/AGENTS.md` / `packages/README.md` (config keys, defaults, error codes), so keeping those in sync remains on the author. Every module has a module-level doc comment explaining its role. Every exported class, interface, type, function, and non-obvious method has a JSDoc that explains semantics (not just the name) — contracts (what events fire when), disposal behavior, error behavior, and extension intent. Internal helpers get docs only where non-obvious. Prefer one-liners when one line suffices. -**Tag every new event with `@mode`.** The cordis events/services catalog ([docs/cordis-catalog/events-and-services.md](docs/cordis-catalog/events-and-services.md)) is GENERATED from source by `scripts/gen-cordis-catalog.ts` — never hand-edit it; run `pnpm run gen-cordis-catalog` and commit the result. When you add an event to an `interface Events` block, its JSDoc MUST carry a `@mode emit|waterfall|parallel` tag (the generator hard-errors without it): use `waterfall` when the signature ends with a `next: () => …` parameter (the listener transforms or vetoes via `next()`), `parallel` when the loop awaits a fan-out with no veto (e.g. an awaited `Promise | void` checkpoint like `session/flush`), and `emit` for plain fire-and-forget notifications. The generator also cross-checks the tag against the signature where the shape is conclusive (a trailing `next` ⇒ waterfall) and hard-errors on a contradiction. Write the rest of the event's JSDoc to stand alone — it is the catalog entry's prose. +**Tag every new event with `@mode`.** The cordis events/services catalog ([docs/cordis-catalog/events-and-services.md](docs/cordis-catalog/events-and-services.md)) is GENERATED from source by `scripts/gen-cordis-catalog.ts` — never hand-edit it; run `pnpm run gen-cordis-catalog` and commit the result. When you add an event to an `interface Events` block, its JSDoc MUST carry a `@mode emit|waterfall|parallel|serial` tag (the generator hard-errors without it): use `waterfall` when the signature ends with a `next: () => …` parameter (the listener transforms or vetoes via `next()`), `parallel` when the loop awaits a fan-out with no veto (e.g. an awaited `Promise | void` checkpoint like `session/flush`), `serial` when the loop awaits listeners in registration order with no veto (e.g. an ordered surface-mutation checkpoint like `agent/pre-step`), and `emit` for plain fire-and-forget notifications. The generator also cross-checks the tag against the signature where the shape is conclusive (a trailing `next` ⇒ waterfall) and hard-errors on a contradiction. Write the rest of the event's JSDoc to stand alone — it is the catalog entry's prose. **The core-data-structures catalog is a maintained surface, not a write-once artifact.** [docs/core-data-structures/](docs/core-data-structures/core.md) catalogs the spine vocabulary (core.md) and the per-seam types (sub-pages). When a change adds, removes, or reshapes a type the catalog documents — a new `…Map` variant, a new content-block or session-event type, a field on `GenerateOptions`/`Agent`/`ToolDefinition`/a bash type, or a whole new core/seam type — update the catalog in the SAME change: edit the prose, and for a pasted ` ```ts type-equiv ` block, re-copy it verbatim and keep `scripts/type-equiv.manifest.json` 1:1 with the blocks. The `verify-type-equiv` gate catches a *drifted paste* of an already-documented type, but it canNOT tell you a brand-new core type was never documented — that judgment is on the author and the reviewer. The definition of "core" (the spine-vs-seam line) is in [core.md § What counts as "core"](docs/core-data-structures/core.md#what-counts-as-core); a genuinely spine-level new type belongs in core.md, a new capability's vocabulary on a sub-page. See [development.md](docs/development.md#documenting-types-verbatim-ts-type-equiv) for the `ts type-equiv` mechanics. diff --git a/docs/cordis-catalog/events-and-services.md b/docs/cordis-catalog/events-and-services.md index 5ace95c782..3823cf8144 100644 --- a/docs/cordis-catalog/events-and-services.md +++ b/docs/cordis-catalog/events-and-services.md @@ -49,21 +49,21 @@ A step or turn errored. The loop reports a failure here (plus the logger) even w Types: [Agent](../core-data-structures/core.md) -Source: [`packages/core/agent/src/types.ts:249`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:248`](../../packages/core/agent/src/types.ts) #### `agent/pre-step` — serial Awaited pre-step surface-mutation checkpoint, fired once per step AFTER `turn/start` (and after the prior step closed) but BEFORE this step's `step/start` — so anything a listener appends lands OUTSIDE the step, between `turn/start`/`step/end` and the upcoming `step/start`. `step` is the number of the step about to start. The loop awaits `ctx.serial('agent/pre-step', …)` after assembling the system prompt, then opens the step and derives the request history ONCE from whatever the surface now holds. This is where compaction belongs: it mutates the session surface in place (shadowing an older range with a summary node) with its log-only `compact/*` records cleanly outside any step, and the single subsequent derive reflects the mutation — so there is no double-derive and no listener can see (or be expected to act on) an assembled `messages` array that does not exist yet. -Serial (awaited, in registration order, no veto), not a waterfall: a listener mutates the surface as a side effect; there is nothing to transform or veto, but the loop must wait for the mutation to complete before opening the step and deriving, and serial isolates listeners from each other (one finishes its surface append before the next runs). `system`/`model` are the assembled values a listener needs to measure pressure (system counts toward the budget) and to summarize (the model). `signal` cancels any in-flight work a listener starts (e.g. a summarization model call). +Serial (awaited, in registration order, no veto), not a waterfall: a listener mutates the surface as a side effect; there is nothing to transform or veto, but the loop must wait for the mutation to complete before opening the step and deriving, and serial isolates listeners from each other (one finishes its surface append before the next runs). `fullSystemPrompt` is the assembled prompt a listener needs to measure pressure (the system prompt counts toward the budget). `signal` cancels any in-flight work a listener starts (e.g. a summarization model call). ```ts cordis-catalog -'agent/pre-step'(agent: Agent, turn: number, step: number, system: string, model: string, signal: AbortSignal): Promise | void +'agent/pre-step'(agent: Agent, turn: number, step: number, fullSystemPrompt: string, signal: AbortSignal): Promise | void ``` Types: [Agent](../core-data-structures/core.md) -Source: [`packages/core/agent/src/types.ts:209`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:208`](../../packages/core/agent/src/types.ts) #### `agent/queued` — emit @@ -87,7 +87,7 @@ Waterfall: mutate the fully-assembled GenerateOptions before the model call (hoo Types: [Agent](../core-data-structures/core.md) · [GenerateOptions](../core-data-structures/core.md) -Source: [`packages/core/agent/src/types.ts:218`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:217`](../../packages/core/agent/src/types.ts) #### `agent/status` — emit @@ -111,7 +111,7 @@ Steering content was injected into a running turn. Types: [Agent](../core-data-structures/core.md) · [ContentBlock](../core-data-structures/core.md) · [MessageSource](../core-data-structures/core.md) -Source: [`packages/core/agent/src/types.ts:243`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:242`](../../packages/core/agent/src/types.ts) #### `agent/step-end` — emit @@ -135,7 +135,7 @@ Waterfall: post-process the assembled assistant Message before tool dispatch (va Types: [Agent](../core-data-structures/core.md) · [Message](../core-data-structures/core.md) -Source: [`packages/core/agent/src/types.ts:224`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:223`](../../packages/core/agent/src/types.ts) #### `agent/step-start` — emit @@ -159,7 +159,7 @@ A raw StreamChunk arrived from the model (token-level UI/log feed). Types: [Agent](../core-data-structures/core.md) · [StreamChunk](../core-data-structures/llm-streaming.md) -Source: [`packages/core/agent/src/types.ts:238`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:237`](../../packages/core/agent/src/types.ts) #### `agent/turn-continuation` — waterfall @@ -171,7 +171,7 @@ Waterfall: override the turn-continuation decision. The default (computed by the Types: [Agent](../core-data-structures/core.md) -Source: [`packages/core/agent/src/types.ts:231`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:230`](../../packages/core/agent/src/types.ts) #### `agent/turn-end` — emit @@ -387,11 +387,11 @@ Implementations MUST honor: - **Blocking**: no compaction begins while another is in progress for the same session. The recommended mechanism is the log-recorded lock — append `compact/start` before the slow work and `compact/end` after (even on failure) — so the lock is visible to replay and crash recovery. ```ts cordis-catalog -abstract compactIfNeeded( session: Session, system: string, model: string, signal: AbortSignal, ): Promise -abstract compactRegion( session: Session, start: number, end: number, model: string, signal?: AbortSignal, ): Promise +abstract compactIfNeeded( agent: CompactAgentContext, turn: number, step: number, fullSystemPrompt: string, signal: AbortSignal, ): Promise +abstract compactRegion( session: Session, start: number, end: number, agent: CompactAgentContext, turn: number, step: number, signal?: AbortSignal, ): Promise ``` -Source: [`packages/compact/compact/src/index.ts:57`](../../packages/compact/compact/src/index.ts) +Source: [`packages/compact/compact/src/index.ts:63`](../../packages/compact/compact/src/index.ts) ### `ctx.llm` — `LlmService` diff --git a/docs/core-data-structures/compaction.md b/docs/core-data-structures/compaction.md index d8a05dc8cf..a1ca8978a6 100644 --- a/docs/core-data-structures/compaction.md +++ b/docs/core-data-structures/compaction.md @@ -50,6 +50,6 @@ interface CompactionResult { ## The service -`CompactService` (`ctx.compact`, abstract — defined in [`packages/compact/compact/src/index.ts`](../../packages/compact/compact/src/index.ts)) declares two abstract methods: `compactIfNeeded(session, system, model, signal)` checks token pressure and compacts an older range if the history is too large (returning `null` when nothing needs it), and `compactRegion(session, start, end, model, signal?)` forcibly summarizes surface nodes `[start, end]` into a single replacement node. `compactIfNeeded`'s parameters are all required — the loop's `agent/pre-step` checkpoint always supplies the assembled `system`, the `model`, and the turn `signal`. A backend summarizing via `ctx.llm.stream()` must forward `signal` into the call's `GenerateOptions.signal`, so an abort or dispose tears down the in-flight summarization. The entire strategy — token estimation, retention policy, event sequencing, summarization — is a HOW decision owned by the implementation. +`CompactService` (`ctx.compact`, abstract — defined in [`packages/compact/compact/src/index.ts`](../../packages/compact/compact/src/index.ts)) declares two abstract methods: `compactIfNeeded(agent, turn, step, fullSystemPrompt, signal)` checks token pressure and compacts an older range if the history is too large (returning `null` when nothing needs it), and `compactRegion(session, start, end, agent, turn, step, signal?)` forcibly summarizes surface nodes `[start, end]` into a single replacement node. `compactIfNeeded`'s parameters are all required — the loop's `agent/pre-step` checkpoint supplies the agent, lifecycle context, assembled `fullSystemPrompt`, and turn `signal`. A backend summarizing via `ctx.llm.stream()` must forward `signal` into the call's `GenerateOptions.signal`, so an abort or dispose tears down the in-flight summarization. The entire strategy — token estimation, retention policy, event sequencing, summarization — is a HOW decision owned by the implementation. -Auto-compaction runs on the serial `agent/pre-step` loop seam (fired once per step, after `turn/start` and BEFORE the step opens and its request history is derived), not the `agent/request` waterfall: compaction mutates the session surface in place — with its log-only `compact/*` records landing cleanly outside any step — and the loop derives the request from the already-compacted surface. Retention is turn-agnostic — the only structural guard is tool-pairing balance (a compacted region's edges are balanced cuts on the surface, so it never splits a step's tool-calls from their results), so a single runaway turn that alone exceeds the window compacts its own early closed steps rather than being retained verbatim. The backend that ships this (`dsh-compact-basic`) documents the retention walk, the single-pass convergence invariant, and the crash/recoverable failure taxonomy. +Auto-compaction runs on the serial `agent/pre-step` loop seam (fired once per step, after `turn/start` and BEFORE the step opens and its request history is derived), not the `agent/request` waterfall: compaction mutates the session surface in place — with its log-only `compact/*` records landing cleanly outside any step — and the loop derives the request from the already-compacted surface. Retention is turn-agnostic — the only structural guard is tool-pairing balance (a compacted region's edges are balanced cuts on the surface, so it never splits a step's tool-calls from their results), so a single runaway turn that alone exceeds the window compacts its own early closed steps rather than being retained verbatim. The backend that ships this (`dsh-compact-basic`) documents the retention walk, the approximate convergence invariant, and the crash/recoverable failure taxonomy. diff --git a/docs/core-data-structures/core.md b/docs/core-data-structures/core.md index ee4f06e14a..0b9c8452e3 100644 --- a/docs/core-data-structures/core.md +++ b/docs/core-data-structures/core.md @@ -306,7 +306,7 @@ interface Agent { } ``` -`AgentStatus` is `'idle' | 'running' | 'disposed'`. `AgentId` is a branded string. `AgentOptions` (`model?`, `systemPrompt?`) is merge-extensible — plugins add creation options by declaration merging. The `agent/*` event taxonomy (lifecycle, turn/step boundaries, the `agent/request`/`agent/step-result`/`agent/turn-continuation` waterfalls) is in [architecture.md § Event taxonomy](../architecture.md#event-taxonomy). +`AgentStatus` is `'idle' | 'running' | 'disposed'`. `AgentId` is a branded string. `AgentOptions` (`model?`, `systemPrompt?`) is merge-extensible — plugins add creation options by declaration merging. The `agent/*` event taxonomy (lifecycle, turn/step boundaries, the serial `agent/pre-step` surface-mutation seam, and the `agent/request`/`agent/step-result`/`agent/turn-continuation` waterfalls) is in [architecture.md § Event taxonomy](../architecture.md#event-taxonomy). ## `ToolDefinition` diff --git a/docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md b/docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md index f1ca9dad2f..ba6f6c6ae6 100644 --- a/docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md +++ b/docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md @@ -30,7 +30,7 @@ This is not a coupling smell — it is the contract's domain. The "only cordis" An earlier draft put the full algorithm (the retention walk, token-summing, text extraction) as concrete methods on the interface, with only `estimateContentTokens()` and `summarize()` abstract. That recouples the contract to one strategy: a backend that wants a different retention policy or a different event-sequencing would have to fight inherited concrete code. Making both core methods abstract puts every *how* decision in the backend, where it belongs, and keeps the interface a pure statement of *what*. The backend remains internally factored — `estimateContentTokens()` and `summarize()` are `protected` hooks a sub-backend can override without reimplementing the walk — but that factoring is the backend's private concern, not the contract's. -`compactIfNeeded(session, system, model, signal)` takes **required** parameters (not the original all-optional shape). The auto-compaction seam (below) always supplies all four — the assembled system prompt (counted toward the estimate), the model (summarization fallback), and the turn's abort signal — so optionality would only invite a hidden default at the seam. `compactRegion(session, start, end, model, signal?)` keeps an optional signal (a manual caller may omit it). +`compactIfNeeded(agent, turn, step, fullSystemPrompt, signal)` takes **required** parameters (not the original all-optional shape). The auto-compaction seam (below) always supplies the agent, lifecycle context, assembled system prompt (counted toward the estimate), and the turn's abort signal, so optionality would only invite a hidden default at the seam. The session being compacted comes from the agent context. `compactRegion(session, start, end, agent, turn, step, signal?)` keeps an optional signal (a manual caller may omit it). Passing lifecycle context rather than a concrete model keeps router agents honest: the backend's summarization request can run through `agent/request`, where model-routing plugins already choose the actual model. ### Auto-compaction runs on `agent/pre-step`, a dedicated surface-mutation seam @@ -40,7 +40,7 @@ The fix is a dedicated loop seam, **`agent/pre-step`** (`@mode serial`), fired b ``` assembly = ctx.systemPrompt.assemble() -await ctx.serial('agent/pre-step', agent, turn, step, system, model, signal) ⟵ compaction mutates the surface here +await ctx.serial('agent/pre-step', agent, turn, step, system, signal) ⟵ compaction mutates the surface here session('step/start') ⟵ the step opens AFTER the seam messages = session.deriveMessages() ⟵ single derive, reflects the compaction request = waterfall agent/request ⟵ pure request transform (hooks, model switch) @@ -64,9 +64,9 @@ A runaway turn thus compacts exactly like any other history: its early *closed* `compactIfNeeded` always anchors the compacted range at the surface **head** (`nodes[0]`). After a first compaction lands a summary node at the head, the *second* compaction's range starts at that summary node and re-summarizes it together with the steps accumulated since — so the surface holds **at most one** auto-generated checkpoint, always at the head, re-consolidated each cycle (the backend's checkpoint-merge prompt makes this a cheap incremental merge — see below). This is *why* `CompactionResult.shadowedRange` is a **surface-position span, not a numeric seq interval**: after a replace lands a fresh high-seq summary node at an older range's position, `start` can be numerically **greater** than `end`. The range is resolved positionally (index into the ordered node list and slice), and `shadowedSeqs` is the authoritative set in surface order. (Manual `compactRegion` may target any aligned mid-range and so *can* leave several checkpoints; the checkpoint framing does not claim everything after it is recent.) -### Single-pass convergence invariant +### Approximate convergence invariant -`resolveConfig` **rejects** (throws at construction) any config where `summarizationMaxTokens + retainTokens >= contextWindow * thresholdRatio`. The invariant guarantees the post-compaction history — the bounded summary plus the retained recent tail — is structurally below the threshold, so a compaction never immediately triggers another: consecutive re-compaction is impossible by construction, with no thrash throttle needed. The bound is **strict** (`>=` rejects, not `>`): the token-pressure gate declines only when the estimate is `< threshold`, so a post-compaction history sitting *exactly* at the threshold would re-trigger on the very next check — equality is a leak, not a safe boundary. `summarizationMaxTokens` stays an explicit *quality* knob (terse summaries); the invariant only forbids setting it so high it breaks convergence. The sole residual is the single-unit-overflow case above (a backward-rounded oversized step can push the retained tail over budget) — which is exactly the out-of-scope concern, not a thrash bug. Per the pre-release reject-don't-migrate stance, a config that cannot guarantee convergence is a bug at the call site, not something to silently clamp. +`resolveConfig` **rejects** (throws at construction) any config where `summarizationMaxTokens + retainTokens >= contextWindow * thresholdRatio`. The invariant bounds the two variable parts of post-compaction history — the bounded summary plus the retained recent tail — but it is intentionally approximate: checkpoint framing, per-message role overhead, system-prompt size, and the char/4 estimator's error can still leave a narrow accepted config near the threshold. The bound is **strict** (`>=` rejects, not `>`): the token-pressure gate declines only when the estimate is `< threshold`, so a post-compaction history sitting *exactly* at the threshold would re-trigger on the very next check — equality is a leak, not a safe boundary. `summarizationMaxTokens` stays an explicit *quality* knob (terse summaries); the invariant only forbids setting it so high it breaks the structural budget. The sole residual is the single-unit-overflow case above (a backward-rounded oversized step can push the retained tail over budget) — which is exactly the out-of-scope concern, not a thrash bug. Per the pre-release reject-don't-migrate stance, a config that cannot satisfy the structural bound is a bug at the call site, not something to silently clamp. ### Surface replacement: `compact/*` events are log-only; one `user/message` carries the summary diff --git a/docs/rfc/implemented/process/2026-06-20-generated-cordis-catalog.md b/docs/rfc/implemented/process/2026-06-20-generated-cordis-catalog.md index 2801ae209c..ae07f4b14c 100644 --- a/docs/rfc/implemented/process/2026-06-20-generated-cordis-catalog.md +++ b/docs/rfc/implemented/process/2026-06-20-generated-cordis-catalog.md @@ -20,7 +20,7 @@ Pure generation is correct here because the codebase is disciplined enough that Specific choices: -- **`@mode` tag, cross-checked.** Each harness event's JSDoc carries an explicit `@mode emit|waterfall|parallel` tag; the generator hard-errors on a missing tag. Where the signature shape is conclusive — a trailing `next: () => …` parameter is structurally a waterfall — it asserts the tag agrees and hard-errors on a contradiction. The emit-vs-parallel distinction is not structurally visible (`session/flush` returns `Promise | void` with no `next`), so it is trusted from the tag. The authoring rule lives in [AGENTS.md](../../../../AGENTS.md). +- **`@mode` tag, cross-checked.** Each harness event's JSDoc carries an explicit `@mode emit|waterfall|parallel|serial` tag; the generator hard-errors on a missing tag. Where the signature shape is conclusive — a trailing `next: () => …` parameter is structurally a waterfall — it asserts the tag agrees and hard-errors on a contradiction. The emit/parallel/serial distinction is not structurally visible (`session/flush` returns `Promise | void` with no `next`, as does the ordered `agent/pre-step` checkpoint), so it is trusted from the tag. The authoring rule lives in [AGENTS.md](../../../../AGENTS.md). - **Tiered scope.** The harness tier (the 8 `@deepseek-ai/dsh-*` services + their events) is rendered in full from source. The inherited tier (cordis-core `ctx.on/emit/effect/provide/…` + the `internal/*` events + loader/hmr/timer) is pinned vendor source a plugin also sees; it is rendered tersely (name + one-line + source pointer) from a curated table in the generator, NOT walked from the vendor AST — the cordis-core `Context` mixes true ctx members with non-service fields (`root`, `baseUrl`, `logger`), and the vendor surface changes only on a deliberate vendor sync. - **Cross-links to the data-structure catalog.** A type name in a signature (`GenerateOptions`, `StreamChunk`, `ToolDefinition`, …) links to the core-data-structures page that documents it. The map is a small hand-curated const in the generator — NOT `type-equiv.manifest.json`, which documents the `…Map` symbols while signatures reference the derived union names, and lists a few symbols on two pages. - **A dedicated fence.** Signature blocks use a ` ```ts cordis-catalog ` info string that `doc-typecheck` recognizes and skips (a bare signature fragment is not standalone-compilable), excluded from the opt-out ratio — the same treatment `type-equiv` blocks get. diff --git a/examples/coding-agent/tests/compaction.e2e.ts b/examples/coding-agent/tests/compaction.e2e.ts index 2c9814c79d..17186055b1 100644 --- a/examples/coding-agent/tests/compaction.e2e.ts +++ b/examples/coding-agent/tests/compaction.e2e.ts @@ -43,14 +43,17 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('compaction: a long session compa // Tiny window so a couple of steps crosses the threshold. The convergence // invariant requires summarizationMaxTokens + retainTokens to be strictly // BELOW the threshold = floor(contextWindow * thresholdRatio) = - // floor(2400 * 0.5) = 1200; 300 + 500 = 800 < 1200. + // floor(2400 * 0.5) = 1200; 600 + 500 = 1100 < 1200. The summary cap + // stays high enough for the live model to emit the required checkpoint + // sections; a truncated checkpoint fails closed and leaves no summary. ctx = await codingHarness(workdir, { compact: { contextWindow: 2400, thresholdRatio: 0.5, retainTokens: 500, - summarizationMaxTokens: 300, + summarizationMaxTokens: 600, }, + persistenceRoot: './.sessions', }) const agent = ctx.agentLoop.create(AgentId('e2e-compaction'), { model: 'deepseek-v4-flash', diff --git a/packages/compact/compact-basic/README.md b/packages/compact/compact-basic/README.md index cc171f62b4..2b13666033 100644 --- a/packages/compact/compact-basic/README.md +++ b/packages/compact/compact-basic/README.md @@ -1,6 +1,6 @@ # @deepseek-ai/dsh-compact-basic -The **basic compaction backend**: a `BasicCompactService` implementing the `@deepseek-ai/dsh-compact` seam with a char/4 token heuristic, token-budget retention, and `ctx.llm.stream()` summarization. +The **basic compaction backend**: a `BasicCompactService` implementing the `@deepseek-ai/dsh-compact` seam with a char/4 token heuristic, token-budget retention, and summarization routed through the agent request pipeline. This is the implementation tier of the compaction capability — see the [interface package](../compact/README.md) for the seam and the [capability-seam RFC](../../../docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md) for the design. @@ -11,7 +11,7 @@ The abstract contract states only WHAT compaction does; this backend owns every - **Token estimation** — `estimateContentTokens()`: char/4 with per-block structural overhead (`text`/`reasoning` = `ceil(len/4) + 4`, `tool-call` from name + arguments, `tool-result` recursive, `image` = 85, unknown blocks via JSON length). - **Retention policy** — `compactIfNeeded()` walks the surface nodes tail→head summing per-node token estimates, and retains the smallest tail-run of WHOLE units (a closed step, or a single no-step node such as a pre-step `user/message` or inter-step `steering/message`) whose total reaches `retainTokens`; everything older is compacted. Retention is **turn-agnostic** — turn boundaries play no role, so a single runaway turn that alone exceeds the window compacts its OWN early closed steps rather than being retained verbatim (the failure mode that motivated dropping turn-protection: a tool-heavy turn must stay compactable or the harness dies exactly when compaction is needed). The only structural guard is **tool-pairing balance**: the compacted region's edges are balanced cuts on the surface (no unanswered tool-call crosses either edge), so it never splits a step's `assistant/message` tool-calls from their `tool/result`s. When the only compactable content left is an un-splittable open tail step, it declines (returns `null`) and retries once an older step closes. **Single-unit overflow is out of scope, by design**: if one retained unit (a single closed step, or a large pasted `user/message`) ALONE exceeds the budget, compaction cannot help and the call may go out over-budget — bounding an individual unit's size is a separate concern. `compactRegion()` enforces tool-pairing balance strictly, throwing on a boundary that would split a step. `dsh-session` exports `isToolPairingBalanced` for the check. - **Single-pass convergence** — `resolveConfig()` rejects (throws) any config where `summarizationMaxTokens + retainTokens >= contextWindow * thresholdRatio`. The invariant guarantees the post-compaction history (the bounded summary plus the retained recent tail) is structurally below the threshold, so a compaction never immediately triggers another: consecutive re-compaction is impossible by construction. The bound is strict (`>=` rejects) because the token-pressure gate declines only when the estimate is `< threshold` — a post-compaction history sitting exactly at the threshold would re-trigger. -- **Summarization** — `summarize()`: a `ctx.llm.stream()` call assembled via `BlockAssembler` (the single model-call surface) with a fixed system prompt that asks for a structured checkpoint (Primary Request and Intent · Key Technical Concepts · Files and Code · Errors and Fixes · Pending Tasks · Current Work · Next Step · Critical Context), every section mandatory, exact paths/commands/identifiers preserved. The compacted region is flattened to a plain-text transcript first: text and reasoning contribute their text, and every non-text block (image, tool-call, tool-result, plugin-added types) contributes a type-tagged placeholder (`[image]`, `[tool-call: name(args)]`, …) so the summarizer is told what existed rather than silently dropping it. +- **Summarization** — `summarize()`: a `GenerateOptions` request assembled via `BlockAssembler` with a fixed system prompt that asks for a structured checkpoint (Primary Request and Intent · Key Technical Concepts · Files and Code · Errors and Fixes · Pending Tasks · Current Work · Next Step · Critical Context), every section mandatory, exact paths/commands/identifiers preserved. The request runs through the `agent/request` waterfall before `ctx.llm.stream()`, so router agents that choose the concrete model there also route compaction summaries. The compacted region is flattened to a plain-text transcript first: text and reasoning contribute their text, and every non-text block (image, tool-call, tool-result, plugin-added types) contributes a type-tagged placeholder (`[image]`, `[tool-call: name(args)]`, …) so the summarizer is told what existed rather than silently dropping it. - **Checkpoint framing** — the raw summary is not landed directly. `compactRegion()` wraps it in a checkpoint preamble (so a resuming model reads it as a checkpoint, not a fresh user request, and builds on the captured context rather than restating it) plus `…` tags. Because region compaction can be invoked manually, a surface may hold several checkpoints, so the framing does not claim everything after it is recent or verbatim. The tags make a prior checkpoint detectable in the transcript on the next compaction cycle: the summarization prompt then instructs the model to merge it in place (preserve still-true facts, drop stale ones) rather than re-summarize it verbatim — a cheap incremental merge that needs no extra log/event machinery. The unframed summary stays on the `compact/summary` provenance event. - **Surface mutation** — `compactRegion()` appends the `compact/start` → `compact/summary` → `compact/end` log records and the single `user/message` replace node carrying the framed summary (see the interface README). - **Auto-compaction** — an `agent/pre-step` listener delegates to `compactIfNeeded()` before every step (not just a turn's first — a tool-heavy turn grows the surface mid-turn, so a runaway turn still compacts, and per-step firing is the only moment to rescue it before overflow). `agent/pre-step` is a serial (awaited, in-order, no-veto) surface-mutation checkpoint that fires after `turn/start` and BEFORE the step opens (`step/start`) and its request history is derived, so compaction mutates the surface — with its log-only `compact/*` records landing cleanly outside any step — and the loop derives once from the result: no double-derive, and the listener cannot see (or need to rewrite) an already-assembled `messages` array. The listener owns no threshold logic of its own (the single token-pressure check lives in `compactIfNeeded()`). diff --git a/packages/compact/compact-basic/src/index.ts b/packages/compact/compact-basic/src/index.ts index 479553a17c..e474be7cd9 100644 --- a/packages/compact/compact-basic/src/index.ts +++ b/packages/compact/compact-basic/src/index.ts @@ -10,7 +10,7 @@ * compaction declines and retries once it closes). * - **Summarization** — `ctx.llm.stream()` assembled via `BlockAssembler` * (the single model-call surface; same path the loop uses) with a fixed - * condense-the-history system prompt. + * condense-the-history system prompt routed through `agent/request`. * - **Surface mutation** — a single `user/message` replace node carries the * summary; `compact/*` events are log-only lock + provenance records. * - **Auto-compaction** — an `agent/pre-step` listener delegates to @@ -153,12 +153,6 @@ function finishError(finish: FinishReason): Error | undefined { * context. */ export class BasicCompactService extends CompactService { - /** - * `summarize()` reads `ctx.llm.stream()`. Declaring `llm` here lets the cordis - * context proxy resolve it when this service loads as a sibling of LlmService: - * without the inject, `this.ctx.llm` cannot be resolved from this fiber and - * compaction throws at runtime (see postmortem 0001). - */ static inject = ['llm'] /** Resolved configuration (defaults applied). */ @@ -188,11 +182,11 @@ export class BasicCompactService extends CompactService { // log-only `compact/*` records and the replacement node cleanly outside a // step, so a crash mid-compaction leaves an inert orphan the turn-repair // closes — never a half-open step. - ctx.on('agent/pre-step', async (agent: Agent, _turn: number, _step: number, system: string, model: string, signal: AbortSignal) => { + ctx.on('agent/pre-step', async (agent: Agent, turn: number, step: number, fullSystemPrompt: string, signal: AbortSignal) => { try { - const result = await this.compactIfNeeded(agent.session, system, model, signal) + const result = await this.compactIfNeeded(agent, turn, step, fullSystemPrompt, signal) if (result) { - const after = this.estimateTokens(agent.session.deriveMessages(), system) + const after = this.estimateTokens(agent.session.deriveMessages(), fullSystemPrompt) ctx.logger.info( `compaction: shadowed ${result.shadowedSeqs.length} surface nodes ` + `(seqs ${result.shadowedRange.start}-${result.shadowedRange.end}, ` + @@ -274,8 +268,9 @@ export class BasicCompactService extends CompactService { } /** - * Summarize conversation text into content blocks via `ctx.llm.stream()` - * assembled through a `BlockAssembler` (the single model-call surface). + * Summarize conversation text into content blocks via `agent/request` plus + * `ctx.llm.stream()` assembled through a `BlockAssembler` (the single + * model-call surface). * Override in a subclass for a template or remote summarizer. * * Honors the adapter failure contract: an adapter may report a model failure @@ -286,12 +281,10 @@ export class BasicCompactService extends CompactService { * Forwards `signal` into `GenerateOptions.signal` so an abort/dispose tears * down the in-flight summarization rather than orphaning the model call. */ - async summarize(text: string, model: string, signal?: AbortSignal): Promise { - if (!model) throw new Error('no model available for summarization') - + async summarize(text: string, agent: Agent, turn: number, step: number, signal?: AbortSignal): Promise { const assembler = new BlockAssembler() const options: GenerateOptions = { - model, + model: this.config.summarizationModel || agent.options.model || '', messages: [{ role: 'user', content: [{ type: 'text', text: `Summarize this conversation history:\n\n${text}\n\nSummary:` }], @@ -302,7 +295,11 @@ export class BasicCompactService extends CompactService { // exactOptionalPropertyTypes: only set `signal` when present — assigning // `undefined` to an optional `signal?: AbortSignal` is a type error. if (signal) options.signal = signal - for await (const chunk of this.ctx.llm.stream(options)) { + const request = await this.ctx.waterfall('agent/request', agent, turn, step, options, () => Promise.resolve(options)) + if (!request.model) { + throw new Error('no model available for summarization: set BasicCompactConfig.summarizationModel, AgentOptions.model, or supply one via the agent/request waterfall') + } + for await (const chunk of this.ctx.llm.stream(request)) { assembler.push(chunk) } @@ -341,13 +338,15 @@ export class BasicCompactService extends CompactService { * closes). */ override async compactIfNeeded( - session: Session, - system: string, - model: string, + agent: Agent, + turn: number, + step: number, + fullSystemPrompt: string, signal: AbortSignal, ): Promise { + const session = agent.session const messages = session.deriveMessages() - const totalTokens = this.estimateTokens(messages, system) + const totalTokens = this.estimateTokens(messages, fullSystemPrompt) const threshold = Math.floor(this.config.contextWindow * this.config.thresholdRatio) if (totalTokens < threshold) return null @@ -401,14 +400,16 @@ export class BasicCompactService extends CompactService { // eslint-disable-next-line @typescript-eslint/no-non-null-assertion const cutoffSeq = nodes[keepFromIdx - 1]!.seq - return this.compactRegion(session, firstSeq, cutoffSeq, model, signal) + return this.compactRegion(session, firstSeq, cutoffSeq, agent, turn, step, signal) } override async compactRegion( session: Session, start: number, end: number, - model: string, + agent: Agent, + turn: number, + step: number, signal?: AbortSignal, ): Promise { // Resolve the range by surface POSITION, not numeric seq interval. A prior @@ -458,8 +459,8 @@ export class BasicCompactService extends CompactService { // strictly inside the open turn (but outside any step). A manual call on a // fully-closed session has no turn to enclose the events, so reject rather // than emit an un-enclosed run. - const turn = this._openTurn(session) - if (turn === null) { + const openTurn = this._openTurn(session) + if (openTurn === null) { throw new Error('compactRegion: no open turn — compaction events must be enclosed in a turn') } // Slice the ordered surface nodes [startIdx, endIdx] inclusive — the @@ -467,13 +468,12 @@ export class BasicCompactService extends CompactService { const shadowedSeqs = nodes.slice(startIdx, endIdx + 1).map(n => n.seq) // --- Acquire lock --- - const startEvent = session.append('compact/start', { turn }) + const startEvent = session.append('compact/start', { turn: openTurn }) try { // --- Extract text and summarize --- const text = this._extractText(session, shadowedSeqs) - const summaryModel = this.config.summarizationModel || model - const summary = await this.summarize(text, summaryModel, signal) + const summary = await this.summarize(text, agent, turn, step, signal) // Estimate token count of the shadowed content for provenance. let shadowedTokenCount = 0 @@ -511,7 +511,7 @@ export class BasicCompactService extends CompactService { // compact/start and here leaves a detectable orphaned lock (a compact/start // with no matching compact/end) rather than a compact/end that falsely // claims compaction finished before the surface replacement landed. - const endEvent = session.append('compact/end', { turn }) + const endEvent = session.append('compact/end', { turn: openTurn }) return { startSeq: startEvent.seq, @@ -526,7 +526,7 @@ export class BasicCompactService extends CompactService { // Always release the lock — append compact/end with the error so a // wedged lock is impossible. const msg = error instanceof Error ? error.message : String(error) - session.append('compact/end', { turn, error: msg }) + session.append('compact/end', { turn: openTurn, error: msg }) throw error } } diff --git a/packages/compact/compact-basic/src/types.ts b/packages/compact/compact-basic/src/types.ts index b7261eb093..13365b7ed1 100644 --- a/packages/compact/compact-basic/src/types.ts +++ b/packages/compact/compact-basic/src/types.ts @@ -39,21 +39,20 @@ export const DEFAULTS: ResolvedConfig = { } /** - * Apply defaults to a partial config and enforce the single-pass convergence + * Apply defaults to a partial config and enforce the approximate convergence * invariant. * * `summarizationMaxTokens + retainTokens` must be strictly BELOW the compaction - * threshold (`contextWindow * thresholdRatio`). The invariant guarantees that - * after a compaction the derived history — the (bounded) summary plus the - * retained recent tail — is structurally below the threshold, so the very next - * pre-step check passes and a second compaction cannot fire on the same - * content. The bound is strict (`>=` rejects) because `compactIfNeeded` declines - * only when the estimate is `< threshold`: a post-compaction history sitting - * EXACTLY at the threshold would re-trigger on the next check. Without the - * invariant, a too-large summary or retain budget would leave the - * post-compaction history at/over threshold, triggering compaction again and - * again. Pre-release we reject rather than clamp: a config that cannot guarantee - * convergence is a bug at the call site, not something to silently paper over. + * threshold (`contextWindow * thresholdRatio`). The invariant bounds the two + * variable pieces of post-compaction history — the summary and the retained + * recent tail — but it is intentionally approximate: checkpoint framing, + * per-message role overhead, system-prompt size, and the char/4 estimator's + * error can still leave a narrow accepted config near the threshold. The bound + * is strict (`>=` rejects) because `compactIfNeeded` declines only when the + * estimate is `< threshold`: a post-compaction history sitting EXACTLY at the + * threshold would re-trigger on the next check. Pre-release we reject rather + * than clamp: a config that cannot satisfy even this structural bound is a bug + * at the call site, not something to silently paper over. * * @throws if `summarizationMaxTokens + retainTokens >= contextWindow * thresholdRatio`. */ diff --git a/packages/compact/compact-basic/tests/compact-basic.spec.ts b/packages/compact/compact-basic/tests/compact-basic.spec.ts index 971aadb9dd..fbbb8258c5 100644 --- a/packages/compact/compact-basic/tests/compact-basic.spec.ts +++ b/packages/compact/compact-basic/tests/compact-basic.spec.ts @@ -29,7 +29,8 @@ class TestCompactService extends BasicCompactService { return blocks.length * 10 } - override async summarize(text: string, model: string): Promise { + override async summarize(text: string, agent: Agent): Promise { + const model = this.config.summarizationModel || agent.options.model || '' this.summarizeCalls.push({ text, model }) if (this.summarizeError) throw this.summarizeError return this.mockSummary @@ -184,7 +185,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai const svc = createTestService({ contextWindow: 200, thresholdRatio: 0.5, retainTokens: 55 }) const session = toolTurnSession(3) - const result = await svc.compactIfNeeded(session, '', 'm', SIGNAL) + const result = await compactIfNeeded(svc, session, '', 'm', SIGNAL) expect(result).not.toBeNull() expect(result!.shadowedSeqs.length).toBeGreaterThan(0) // No dangling tool-result: every compacted/retained step stayed whole. @@ -214,7 +215,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai // Turn stays open. const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.1, retainTokens: 5 }) - const result = await svc.compactIfNeeded(s, '', 'm', SIGNAL) + const result = await compactIfNeeded(svc, s, '', 'm', SIGNAL) expect(result).toBeNull() expect(s.events.some(e => e.type === 'compact/start')).toBe(false) }) @@ -227,7 +228,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai const resultSeq = nodes[2]!.seq // start = the tool/result: its issuing assistant precedes it IN THE SAME STEP, // so starting here would orphan that assistant's tool-call. end is fine (user). - await expect(svc.compactRegion(session, resultSeq, resultSeq, 'm')) + await expect(compactRegion(svc, session, resultSeq, resultSeq, 'm')) .rejects.toThrow(/start seq .* is not a balanced boundary/) expect(userSeq).toBeLessThan(resultSeq) // sanity: ordering as expected }) @@ -240,7 +241,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai const asstSeq = nodes[1]!.seq // end = the assistant/message: its tool/result follows IN THE SAME STEP, so // ending here would strand that result. start is fine (the pre-step user). - await expect(svc.compactRegion(session, userSeq, asstSeq, 'm')) + await expect(compactRegion(svc, session, userSeq, asstSeq, 'm')) .rejects.toThrow(/end seq .* is not a balanced boundary/) }) @@ -257,7 +258,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai const nodes = s.surface.nodes // [user, asst] const userSeq = nodes[0]!.seq const asstSeq = nodes[1]!.seq - await expect(svc.compactRegion(s, userSeq, asstSeq, 'm')) + await expect(compactRegion(svc, s, userSeq, asstSeq, 'm')) .rejects.toThrow(/end seq .* is not a balanced boundary/) }) @@ -267,7 +268,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai const nodes = session.surface.nodes // [user1, asst1, res1, user2, asst2, res2] const startSeq = nodes[0]!.seq // pre-step user1 (free boundary) const endSeq = nodes[2]!.seq // res1 = last node of turn 1's closed step - const result = await svc.compactRegion(session, startSeq, endSeq, 'm') + const result = await compactRegion(svc, session, startSeq, endSeq, 'm') expect(result.shadowedRange).toEqual({ start: startSeq, end: endSeq }) expectNoOrphanToolResults(session.deriveMessages()) }) @@ -277,7 +278,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai const session = toolTurnSession(1) const nodes = session.surface.nodes const userSeq = nodes[0]!.seq // pre-step user: free boundary both ways - const result = await svc.compactRegion(session, userSeq, userSeq, 'm') + const result = await compactRegion(svc, session, userSeq, userSeq, 'm') expect(result.shadowedRange).toEqual({ start: userSeq, end: userSeq }) }) @@ -292,7 +293,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai s.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) const nodes = s.surface.nodes const ctxSeq = nodes[0]!.seq - const result = await svc.compactRegion(s, ctxSeq, ctxSeq, 'm') + const result = await compactRegion(svc, s, ctxSeq, ctxSeq, 'm') expect(result.shadowedRange).toEqual({ start: ctxSeq, end: ctxSeq }) }) }) @@ -352,7 +353,7 @@ describe('BasicCompactService.compactRegion', () => { const firstSeq = nodes[0]!.seq const secondSeq = nodes[1]!.seq - const result = await svc.compactRegion(session, firstSeq, secondSeq, 'test-model') + const result = await compactRegion(svc, session, firstSeq, secondSeq, 'test-model') expect(result.shadowedSeqs).toEqual([firstSeq, secondSeq]) expect(result.shadowedRange.start).toBe(firstSeq) @@ -405,7 +406,7 @@ describe('BasicCompactService.compactRegion', () => { it('throws when start or end are not surface nodes', async () => { const svc = createTestService() const session = multiTurnSession(1, 1) - await expect(svc.compactRegion(session, 999, 1000, 'm')) + await expect(compactRegion(svc, session, 999, 1000, 'm')) .rejects.toThrow(/start seq 999 not found in surface/) }) @@ -413,7 +414,7 @@ describe('BasicCompactService.compactRegion', () => { const svc = createTestService() const session = multiTurnSession(2, 1) const nodes = session.surface.nodes - await expect(svc.compactRegion(session, nodes[1]!.seq, nodes[0]!.seq, 'm')) + await expect(compactRegion(svc, session, nodes[1]!.seq, nodes[0]!.seq, 'm')) .rejects.toThrow(/is after end seq .* on the surface/) }) @@ -422,7 +423,7 @@ describe('BasicCompactService.compactRegion', () => { const session = multiTurnSession(2, 1) const nodes = session.surface.nodes session.append('compact/start', { turn: 2 }) - await expect(svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm')) + await expect(compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm')) .rejects.toThrow(/compaction already in progress/) }) @@ -432,7 +433,7 @@ describe('BasicCompactService.compactRegion', () => { const session = multiTurnSession(2, 1) const nodes = session.surface.nodes - await expect(svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm')) + await expect(compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm')) .rejects.toThrow('model unavailable') const endEvent = session.events.findLast(e => e.type === 'compact/end') @@ -455,7 +456,7 @@ describe('BasicCompactService.compactRegion', () => { const session = multiTurnSession(1, 2) const nodes = session.surface.nodes - await svc.compactRegion(session, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm') + await compactRegion(svc, session, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm') expect(svc.summarizeCalls.length).toBe(1) const { text, model } = svc.summarizeCalls[0]! @@ -470,7 +471,7 @@ describe('BasicCompactService.compactRegion', () => { const session = multiTurnSession(3, 1) const nodes = session.surface.nodes - const result = await svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm') + const result = await compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm') // Provenance (compact/summary) carries the RAW, unframed summary. expect(result.summary).toEqual([{ type: 'text', text: 'STRUCTURED SUMMARY' }]) @@ -492,7 +493,7 @@ describe('BasicCompactService.compactRegion', () => { const firstSeq = nodes[0]!.seq const lastSeq = nodes[nodes.length - 1]!.seq - await svc.compactRegion(session, firstSeq, lastSeq, 'm') + await compactRegion(svc, session, firstSeq, lastSeq, 'm') expect(svc.summarizeCalls.length).toBe(1) const { text } = svc.summarizeCalls[0]! @@ -506,14 +507,14 @@ describe('BasicCompactService.compactIfNeeded', () => { it('returns null when tokens are under threshold', async () => { const svc = createTestService({ contextWindow: 128000, thresholdRatio: 0.8 }) const session = multiTurnSession(1, 1) - expect(await svc.compactIfNeeded(session, '', 'm', SIGNAL)).toBeNull() + expect(await compactIfNeeded(svc, session, '', 'm', SIGNAL)).toBeNull() }) it('compacts when tokens exceed threshold', async () => { const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.5, retainTokens: 10 }) const session = multiTurnSession(3, 1) // 6 surface nodes, 10 tokens each = 60 - const result = await svc.compactIfNeeded(session, '', 'm', SIGNAL) + const result = await compactIfNeeded(svc, session, '', 'm', SIGNAL) expect(result).not.toBeNull() expect(result!.shadowedSeqs.length).toBeGreaterThan(0) }) @@ -522,7 +523,7 @@ describe('BasicCompactService.compactIfNeeded', () => { const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.2, retainTokens: 15 }) const session = multiTurnSession(5, 1) // 10 surface nodes = ~100 tokens - const result = await svc.compactIfNeeded(session, '', 'm', SIGNAL) + const result = await compactIfNeeded(svc, session, '', 'm', SIGNAL) expect(result).not.toBeNull() const nodes = session.surface.nodes expect(result!.shadowedSeqs.length).toBeGreaterThan(0) @@ -538,7 +539,7 @@ describe('BasicCompactService.compactIfNeeded', () => { // summarizationMaxTokens (1) + retainTokens (45) = 46 < threshold 47. const svc = createTestService({ contextWindow: 470, thresholdRatio: 0.1, retainTokens: 45 }) const session = multiTurnSession(2, 1) - expect(await svc.compactIfNeeded(session, '', 'm', SIGNAL)).toBeNull() + expect(await compactIfNeeded(svc, session, '', 'm', SIGNAL)).toBeNull() }) it('compacts a runaway turn: its early CLOSED steps summarize while recent steps stay verbatim', async () => { @@ -572,7 +573,7 @@ describe('BasicCompactService.compactIfNeeded', () => { const nodesBefore = s.surface.nodes.length expect(nodesBefore).toBe(11) - const result = await svc.compactIfNeeded(s, '', 'm', SIGNAL) + const result = await compactIfNeeded(svc, s, '', 'm', SIGNAL) expect(result).not.toBeNull() // Early steps of the SAME open turn were shadowed (impossible under layer 2). expect(result!.shadowedSeqs.length).toBeGreaterThan(0) @@ -587,7 +588,7 @@ describe('BasicCompactService.compactIfNeeded', () => { it('returns null for an empty surface', async () => { const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.5, retainTokens: 10 }) const session = new Session(SessionId('empty')) - expect(await svc.compactIfNeeded(session, '', 'm', SIGNAL)).toBeNull() + expect(await compactIfNeeded(svc, session, '', 'm', SIGNAL)).toBeNull() }) it('compacts again after a prior summary node heads the surface (the summary stays eligible)', async () => { @@ -600,7 +601,7 @@ describe('BasicCompactService.compactIfNeeded', () => { const svc = createTestService({ contextWindow: 300, thresholdRatio: 0.1, retainTokens: 25 }) const s = multiTurnSession(4, 1) // turns 1-4 closed, turn 5 open (no surface yet) - const first = await svc.compactIfNeeded(s, '', 'm', SIGNAL) + const first = await compactIfNeeded(svc, s, '', 'm', SIGNAL) expect(first).not.toBeNull() // The summary node now heads the surface with a fresh high seq. const summaryHeadSeq = s.surface.nodes[0]!.seq @@ -615,7 +616,7 @@ describe('BasicCompactService.compactIfNeeded', () => { s.append('assistant/message', { turn: 5, step: 1, content: [{ type: 'text', text: 'reply 5' }] }, { surfaceOp: 'append' }) s.append('step/end', { turn: 5, step: 1 }) - const second = await svc.compactIfNeeded(s, '', 'm', SIGNAL) + const second = await compactIfNeeded(svc, s, '', 'm', SIGNAL) expect(second).not.toBeNull() expect(second!.shadowedSeqs.length).toBeGreaterThan(0) // The fresh open-turn nodes were NOT compacted. @@ -630,7 +631,7 @@ describe('BasicCompactService replay equivalence', () => { const session = multiTurnSession(3, 1) const nodes = session.surface.nodes - await svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm') + await compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm') const derived = session.deriveMessages() const replayed = new Session(SessionId('replay'), [...session.events]) @@ -646,7 +647,7 @@ describe('BasicCompactService blocking (compaction in progress)', () => { const nodes = session.surface.nodes // Whole step (user → assistant) is a step-aligned region, so the call reaches // the in-progress check rather than being rejected for splitting a step. - await expect(svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm')) + await expect(compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm')) .rejects.toThrow(/compaction already in progress/) }) @@ -656,7 +657,7 @@ describe('BasicCompactService blocking (compaction in progress)', () => { const nodes = session.surface.nodes session.append('compact/start', { turn: 1 }) session.append('compact/end', { turn: 1 }) - const result = await svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm') + const result = await compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm') expect(result).toBeDefined() }) @@ -679,7 +680,7 @@ describe('BasicCompactService blocking (compaction in progress)', () => { const nodes = s.surface.nodes // The stale start is before the turn/end, so it is NOT seen as in-progress. - const result = await svc.compactRegion(s, nodes[0]!.seq, nodes[1]!.seq, 'm') + const result = await compactRegion(svc, s, nodes[0]!.seq, nodes[1]!.seq, 'm') expect(result).toBeDefined() }) }) @@ -829,12 +830,37 @@ function stubAgent(session: Session, model?: string): Agent { return { session, options: { model } } as unknown as Agent } +function compactIfNeeded( + svc: BasicCompactService, + session: Session, + fullSystemPrompt: string, + model: string, + signal: AbortSignal, +) { + return svc.compactIfNeeded(stubAgent(session, model), 1, 1, fullSystemPrompt, signal) +} + +function compactRegion( + svc: BasicCompactService, + session: Session, + start: number, + end: number, + model: string, + signal?: AbortSignal, +) { + return svc.compactRegion(session, start, end, stubAgent(session, model), 1, 1, signal) +} + +function summarize(svc: BasicCompactService, text: string, model: string) { + return svc.summarize(text, stubAgent(new Session(SessionId('summary')), model), 1, 1) +} + describe('BasicCompactService.summarize (real ctx.llm.stream)', () => { it('summarizes via the registered adapter and returns its content', async () => { const { ctx, adapter } = await ctxWithModel('SUMMARY TEXT') const svc = new BasicCompactService(ctx, { auto: false, summarizationMaxTokens: 512 }) - const summary = await svc.summarize('User: hi\n\nAssistant: hello', 'test-model') + const summary = await summarize(svc, 'User: hi\n\nAssistant: hello', 'test-model') expect(summary).toEqual([{ type: 'text', text: 'SUMMARY TEXT' }]) // The fixed system prompt and maxTokens flow through. expect(adapter.lastOptions!.system).toContain('compaction engine') @@ -846,19 +872,19 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => { it('throws when no model is provided', async () => { const { ctx } = await ctxWithModel('x') const svc = new BasicCompactService(ctx, { auto: false }) - await expect(svc.summarize('text', '')).rejects.toThrow(/no model available/) + await expect(summarize(svc, 'text', '')).rejects.toThrow(/no model available/) }) it('rethrows when the stream ends with a finish-error chunk', async () => { const ctx = await ctxWithFinish({ kind: 'error', message: 'provider 401', code: 'UNAUTHORIZED' }) const svc = new BasicCompactService(ctx, { auto: false }) - await expect(svc.summarize('text', 'test-model')).rejects.toMatchObject({ message: 'provider 401', code: 'UNAUTHORIZED' }) + await expect(summarize(svc, 'text', 'test-model')).rejects.toMatchObject({ message: 'provider 401', code: 'UNAUTHORIZED' }) }) it('rethrows a finish-error chunk without a code (code stays undefined)', async () => { const ctx = await ctxWithFinish({ kind: 'error', message: 'opaque failure' }) const svc = new BasicCompactService(ctx, { auto: false }) - const error = await svc.summarize('text', 'test-model').then(() => null, (e: unknown) => e as Error & { code?: string }) + const error = await summarize(svc, 'text', 'test-model').then(() => null, (e: unknown) => e as Error & { code?: string }) expect(error?.message).toBe('opaque failure') expect(error?.code).toBeUndefined() }) @@ -866,13 +892,13 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => { it('rethrows when the stream ends with a finish-aborted chunk', async () => { const ctx = await ctxWithFinish({ kind: 'aborted' }) const svc = new BasicCompactService(ctx, { auto: false }) - await expect(svc.summarize('text', 'test-model')).rejects.toMatchObject({ message: 'summarization stream aborted', code: 'ABORTED' }) + await expect(summarize(svc, 'text', 'test-model')).rejects.toMatchObject({ message: 'summarization stream aborted', code: 'ABORTED' }) }) it('fails closed on a max-tokens finish (an incomplete checkpoint must not commit)', async () => { const ctx = await ctxWithFinish({ kind: 'max-tokens' }) const svc = new BasicCompactService(ctx, { auto: false }) - await expect(svc.summarize('text', 'test-model')).rejects.toMatchObject({ code: 'MAX_TOKENS' }) + await expect(summarize(svc, 'text', 'test-model')).rejects.toMatchObject({ code: 'MAX_TOKENS' }) }) it('compactRegion leaves the surface intact when summarization hits max-tokens', async () => { @@ -882,7 +908,7 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => { const before = [...session.surface.nodes] const nodes = session.surface.nodes - await expect(svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'test-model')) + await expect(compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'test-model')) .rejects.toMatchObject({ code: 'MAX_TOKENS' }) // No replacement landed — the surface is byte-identical, and the lock was @@ -899,7 +925,7 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => { const session = multiTurnSession(2, 1) const nodes = session.surface.nodes - const result = await svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'test-model') + const result = await compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'test-model') expect(result.summary).toEqual([{ type: 'text', text: 'CONDENSED' }]) // The raw summary is wrapped in the checkpoint framing on the surface. expect(session.deriveMessages()[0]!.content).toContainEqual({ type: 'text', text: 'CONDENSED' }) @@ -908,8 +934,8 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => { describe('BasicCompactService auto-compaction (agent/pre-step listener)', () => { /** Fire the agent/pre-step serial checkpoint as the loop does. */ - function firePreStep(ctx: Context, agent: Agent, step: number, system: string, model: string): Promise { - return ctx.serial('agent/pre-step', agent, 1, step, system, model, SIGNAL) + function firePreStep(ctx: Context, agent: Agent, step: number, fullSystemPrompt: string): Promise { + return ctx.serial('agent/pre-step', agent, 1, step, fullSystemPrompt, SIGNAL) } it('compacts (mutating the surface) when over threshold', async () => { @@ -919,7 +945,7 @@ describe('BasicCompactService auto-compaction (agent/pre-step listener)', () => const agent = stubAgent(session, 'test-model') const before = session.surface.nodes.length - await firePreStep(ctx, agent, 1, '', 'test-model') + await firePreStep(ctx, agent, 1, '') // The surface shrank in place, and a summary checkpoint landed. expect(session.surface.nodes.length).toBeLessThan(before) @@ -936,7 +962,7 @@ describe('BasicCompactService auto-compaction (agent/pre-step listener)', () => // A step-2 checkpoint (a tool-heavy turn's later step) must still compact — // the surface accumulated assistant/message + tool/result nodes since step 1. - await firePreStep(ctx, agent, 2, '', 'test-model') + await firePreStep(ctx, agent, 2, '') expect(session.events.some(e => e.type === 'compact/start')).toBe(true) }) @@ -946,7 +972,7 @@ describe('BasicCompactService auto-compaction (agent/pre-step listener)', () => const session = multiTurnSession(1, 1) const agent = stubAgent(session, 'test-model') - await firePreStep(ctx, agent, 1, '', 'test-model') + await firePreStep(ctx, agent, 1, '') expect(session.events.some(e => e.type === 'compact/start')).toBe(false) }) @@ -960,7 +986,7 @@ describe('BasicCompactService auto-compaction (agent/pre-step listener)', () => const agent = stubAgent(session, 'missing-model') const before = session.surface.nodes.length - await firePreStep(ctx, agent, 1, '', 'missing-model') + await firePreStep(ctx, agent, 1, '') // No summary landed; the surface is unchanged. expect(session.events.some(e => e.type === 'compact/summary')).toBe(false) expect(session.surface.nodes.length).toBe(before) @@ -972,9 +998,44 @@ describe('BasicCompactService auto-compaction (agent/pre-step listener)', () => const session = multiTurnSession(3, 1) const agent = stubAgent(session, 'test-model') - await firePreStep(ctx, agent, 1, '', 'test-model') + await firePreStep(ctx, agent, 1, '') expect(session.events.some(e => e.type === 'compact/start')).toBe(false) }) + + it('routes summarization through agent/request so router agents can choose the model', async () => { + const { ctx, adapter } = await ctxWithModel('ROUTED SUMMARY', 'routed-model') + ctx.on('agent/request', async (_agent, _turn, _step, options, next) => { + options.model = 'routed-model' + return next() + }) + void new BasicCompactService(ctx, { contextWindow: 200, thresholdRatio: 0.5, retainTokens: 20, summarizationMaxTokens: 50 }) + const session = multiTurnSession(5, 1) + const agent = stubAgent(session) + + await ctx.serial('agent/pre-step', agent, 1, 1, '', SIGNAL) + + expect(adapter.lastOptions?.model).toBe('routed-model') + expect(session.events.some(e => e.type === 'compact/summary')).toBe(true) + expect(session.deriveMessages()[0]!.content).toContainEqual({ type: 'text', text: 'ROUTED SUMMARY' }) + }) + + it('removes the auto pre-step listener when the plugin fiber is disposed', async () => { + const { ctx } = await ctxWithModel('SUMMARY') + const fiber = await ctx.plugin(BasicCompactService, { + contextWindow: 200, + thresholdRatio: 0.5, + retainTokens: 20, + summarizationMaxTokens: 50, + }) + const session = multiTurnSession(5, 1) + const agent = stubAgent(session, 'test-model') + + await fiber.dispose() + await firePreStep(ctx, agent, 1, '') + + expect(session.events.some(e => e.type === 'compact/start')).toBe(false) + expect(ctx.get('compact')).toBeUndefined() + }) }) describe('BasicCompactService._extractText branches', () => { @@ -1001,7 +1062,7 @@ describe('BasicCompactService._extractText branches', () => { s.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) const nodes = s.surface.nodes - await svc.compactRegion(s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm') + await compactRegion(svc, s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm') const { text } = svc.summarizeCalls[0]! expect(text).toContain('[Context: project context here]') @@ -1030,7 +1091,7 @@ describe('BasicCompactService._extractText branches', () => { s.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) const nodes = s.surface.nodes - await svc.compactRegion(s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm') + await compactRegion(svc, s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm') expect(svc.summarizeCalls[0]!.text).toContain('Tool error (call c9): boom failure') }) }) @@ -1064,7 +1125,7 @@ describe('BasicCompactService edge cases', () => { s.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) const nodes = s.surface.nodes - await svc.compactRegion(s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm') + await compactRegion(svc, s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm') const { text } = svc.summarizeCalls[0]! expect(text).toContain('[tool-result: [image]]') // nested tool-result with content expect(text).toContain('[custom-widget]') // unknown block placeholder @@ -1089,7 +1150,7 @@ describe('BasicCompactService edge cases', () => { const session = multiTurnSession(4, 1) const agent = stubAgent(session, 'test-model') - await ctx.serial('agent/pre-step', agent, 1, 1, '', 'test-model', SIGNAL) + await ctx.serial('agent/pre-step', agent, 1, 1, '', SIGNAL) expect(session.events.some(e => e.type === 'compact/summary')).toBe(true) // The surface was mutated; the head message is the framed summary checkpoint. expect(session.deriveMessages()[0]!.content).toContainEqual({ type: 'text', text: 'SUMMARY' }) @@ -1111,7 +1172,7 @@ describe('BasicCompactService edge cases', () => { s.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) const nodes = s.surface.nodes - await expect(svc.compactRegion(s, nodes[0]!.seq, nodes[1]!.seq, 'm')) + await expect(compactRegion(svc, s, nodes[0]!.seq, nodes[1]!.seq, 'm')) .rejects.toThrow(/no open turn/) // The lock was never acquired — no compact/start landed. expect(s.events.some(e => e.type === 'compact/start')).toBe(false) @@ -1126,7 +1187,7 @@ describe('BasicCompactService edge cases', () => { s.append('user/message', { content: [{ type: 'text', text: 'orphan' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) const nodes = s.surface.nodes - await expect(svc.compactRegion(s, nodes[0]!.seq, nodes[0]!.seq, 'm')) + await expect(compactRegion(svc, s, nodes[0]!.seq, nodes[0]!.seq, 'm')) .rejects.toThrow(/no open turn/) expect(s.events.some(e => e.type === 'compact/start')).toBe(false) }) @@ -1136,14 +1197,14 @@ describe('BasicCompactService edge cases', () => { const session = new Session(SessionId('empty-but-pressured')) // No surface nodes, but a large system prompt pushes the estimate over threshold. const bigPrompt = 'x'.repeat(800) // ceil(800/4) = 200 tokens >> threshold 100 - expect(await svc.compactIfNeeded(session, bigPrompt, 'm', SIGNAL)).toBeNull() + expect(await compactIfNeeded(svc, session, bigPrompt, 'm', SIGNAL)).toBeNull() }) it('compactRegion throws when end is not a surface node (start valid)', async () => { const svc = createTestService() const session = multiTurnSession(1, 1) const nodes = session.surface.nodes - await expect(svc.compactRegion(session, nodes[0]!.seq, 9999, 'm')) + await expect(compactRegion(svc, session, nodes[0]!.seq, 9999, 'm')) .rejects.toThrow(/end seq 9999 not found in surface/) }) @@ -1155,7 +1216,7 @@ describe('BasicCompactService edge cases', () => { const nodes = session.surface.nodes // Whole step (user → assistant): a step-aligned region that reaches summarize. - await expect(svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm')).rejects.toBe('plain string failure') + await expect(compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm')).rejects.toBe('plain string failure') const endEvent = session.events.findLast(e => e.type === 'compact/end')! expect(endEvent.data).toMatchObject({ error: 'plain string failure' }) }) @@ -1170,7 +1231,7 @@ describe('BasicCompactService edge cases', () => { const agent = stubAgent(session, 'test-model') const before = session.surface.nodes.length - await ctx.serial('agent/pre-step', agent, 1, 1, '', 'test-model', SIGNAL) + await ctx.serial('agent/pre-step', agent, 1, 1, '', SIGNAL) // The failure was swallowed; the surface is untouched and a warning logged. expect(session.surface.nodes.length).toBe(before) expect(session.events.some(e => e.type === 'compact/summary')).toBe(false) @@ -1187,7 +1248,7 @@ describe('BasicCompactService edge cases', () => { const agent = stubAgent(session, 'test-model') const bigSystem = 'x'.repeat(900) // ceil(900/4)=225 > threshold 200 - await ctx.serial('agent/pre-step', agent, 1, 1, bigSystem, 'test-model', SIGNAL) + await ctx.serial('agent/pre-step', agent, 1, 1, bigSystem, SIGNAL) expect(session.events.some(e => e.type === 'compact/start')).toBe(false) expect(svc.summarizeCalls.length).toBe(0) }) @@ -1221,7 +1282,7 @@ describe('BasicCompactService edge cases', () => { s.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) const nodes = s.surface.nodes - await svc.compactRegion(s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm') + await compactRegion(svc, s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm') // Every empty-content message (user text, empty reasoning, empty-content // tool/result, empty context, empty steering) extracted to nothing and was // skipped — the only surviving line is the assistant's tool-call (which a @@ -1256,7 +1317,7 @@ describe('BasicCompactService edge cases', () => { s.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) const nodes = s.surface.nodes - await svc.compactRegion(s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm') + await compactRegion(svc, s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm') const { text } = svc.summarizeCalls[0]! // Every non-text block surfaces as a placeholder rather than being dropped. expect(text).toContain('User: [image]') @@ -1280,7 +1341,7 @@ describe('BasicCompactService positional range (surface seqs are not monotonic a // First compaction: shadow the two oldest surface nodes. const nodes0 = session.surface.nodes - const first = await svc.compactRegion(session, nodes0[0]!.seq, nodes0[1]!.seq, 'm') + const first = await compactRegion(svc, session, nodes0[0]!.seq, nodes0[1]!.seq, 'm') // The summary node now sits at the head with a seq HIGHER than the // retained older nodes that follow it — the non-monotonic surface. (The @@ -1298,7 +1359,7 @@ describe('BasicCompactService positional range (surface seqs are not monotonic a const startSeq = nodes1[0]!.seq const endSeq = nodes1[2]!.seq expect(startSeq).toBeGreaterThan(endSeq) - const second = await svc.compactRegion(session, startSeq, endSeq, 'm') + const second = await compactRegion(svc, session, startSeq, endSeq, 'm') // Exactly the three nodes at surface positions [0..2] are shadowed, in // surface order — the positional slice, regardless of their seq values. @@ -1316,14 +1377,14 @@ describe('BasicCompactService positional range (surface seqs are not monotonic a // First compaction shadows the oldest two surface nodes, landing a high-seq // summary node at the head. const n0 = session.surface.nodes - await svc.compactRegion(session, n0[0]!.seq, n0[1]!.seq, 'm') + await compactRegion(svc, session, n0[0]!.seq, n0[1]!.seq, 'm') // Second compaction spans [head summary … turn-2's step end]. The head's seq // is higher than the older retained nodes' seqs, so a log-seq-order walk // would emit the older messages BEFORE the checkpoint. const n1 = session.surface.nodes svc.summarizeCalls = [] - await svc.compactRegion(session, n1[0]!.seq, n1[2]!.seq, 'm') + await compactRegion(svc, session, n1[0]!.seq, n1[2]!.seq, 'm') // The extracted transcript follows surface order: the checkpoint (head) // first, then the older retained messages — matching deriveMessages(). @@ -1355,7 +1416,7 @@ describe('BasicCompactService llm inject (real plugin-load path)', () => { const svc = ctx.compact as BasicCompactService const session = multiTurnSession(2, 1) const nodes = session.surface.nodes - const result = await svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'test-model') + const result = await compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'test-model') expect(result.summary).toEqual([{ type: 'text', text: 'CONDENSED' }]) // Tear the fiber down so this test owns no leaked registration; the @@ -1403,7 +1464,7 @@ describe('BasicCompactService under the real invariants plugin', () => { const nodes = session.surface.nodes // No invariant throws here: compact/* + the replacement are all in turn 3. - const result = await svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'test-model') + const result = await compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'test-model') expect(result.shadowedSeqs.length).toBe(2) expect(session.surface.nodes[0]!.seq).toBeGreaterThan(session.surface.nodes[1]!.seq) }) @@ -1416,15 +1477,14 @@ describe('BasicCompactService under the real invariants plugin', () => { session.append('turn/start', { turn: 4, trigger: { kind: 'message', source: { kind: 'user' } } }) const n0 = session.surface.nodes - await svc.compactRegion(session, n0[0]!.seq, n0[1]!.seq, 'test-model') + await compactRegion(svc, session, n0[0]!.seq, n0[1]!.seq, 'test-model') // Surface head now carries a higher seq than the older retained nodes. A // second compaction spanning [head … a later closed-step end] must pass the // invariants' positional replace check even though startSeq > endSeq. const n1 = session.surface.nodes expect(n1[0]!.seq).toBeGreaterThan(n1[2]!.seq) - const second = await svc.compactRegion(session, n1[0]!.seq, n1[2]!.seq, 'test-model') + const second = await compactRegion(svc, session, n1[0]!.seq, n1[2]!.seq, 'test-model') expect(second.shadowedSeqs).toEqual([n1[0]!.seq, n1[1]!.seq, n1[2]!.seq]) }) }) - diff --git a/packages/compact/compact/README.md b/packages/compact/compact/README.md index 8a95277c17..3cc3ba0333 100644 --- a/packages/compact/compact/README.md +++ b/packages/compact/compact/README.md @@ -18,10 +18,10 @@ Both methods are **abstract** — the backend owns the entire strategy (token es | Member | Semantics | |---|---| -| `compactIfNeeded(session, system, model, signal)` | Estimate the surface-derived history size; if over the backend's threshold, compact an older range via `compactRegion`, keeping recent context intact. Returns the `CompactionResult`, or `null` if nothing needed compacting. All parameters required — the loop's `agent/pre-step` checkpoint always supplies the assembled `system`, the `model`, and the turn `signal`. | -| `compactRegion(session, start, end, model, signal?)` | Forcibly summarize surface nodes `[start, end]` (inclusive seqs) into a single replacement node. **Throws** if a compaction is already in progress, if `start`/`end` aren't surface nodes, or if `start` is positioned after `end` on the surface. The range is a SURFACE-POSITION span, not a numeric seq interval — after a prior replace lands a fresh high-seq summary node at the shadowed range's position, surface order no longer tracks seq order. | +| `compactIfNeeded(agent, turn, step, fullSystemPrompt, signal)` | Estimate the surface-derived history size; if over the backend's threshold, compact an older range via `compactRegion`, keeping recent context intact. Returns the `CompactionResult`, or `null` if nothing needed compacting. All parameters required — the loop's `agent/pre-step` checkpoint supplies the agent, lifecycle context, assembled `fullSystemPrompt`, and turn `signal`; router-aware summarizers can use the agent lifecycle context to route their own model call through `agent/request`. | +| `compactRegion(session, start, end, agent, turn, step, signal?)` | Forcibly summarize surface nodes `[start, end]` (inclusive seqs) into a single replacement node. **Throws** if a compaction is already in progress, if `start`/`end` aren't surface nodes, or if `start` is positioned after `end` on the surface. The range is a SURFACE-POSITION span, not a numeric seq interval — after a prior replace lands a fresh high-seq summary node at the shadowed range's position, surface order no longer tracks seq order. | -`compactIfNeeded` takes a required `signal`; `compactRegion`'s is optional. A backend that summarizes via `ctx.llm.stream()` **must** forward it into the call's `GenerateOptions.signal`, so an abort or fiber dispose tears down the in-flight summarization instead of leaving an orphaned model call running past the cancellation. The turn that the `compact/*` events belong to is not a parameter — it is recoverable from the log (the currently-open turn), so the backend stamps it without the caller supplying it. +`compactIfNeeded` takes a required `signal`; `compactRegion`'s is optional. A backend that summarizes via `ctx.llm.stream()` **must** forward it into the call's `GenerateOptions.signal`, so an abort or fiber dispose tears down the in-flight summarization instead of leaving an orphaned model call running past the cancellation. The session being compacted comes from the agent context; the turn that the `compact/*` events belong to is recoverable from the log (the currently-open turn), so the backend stamps it from the log rather than trusting a caller-supplied value. ## Surface contract diff --git a/packages/compact/compact/src/index.ts b/packages/compact/compact/src/index.ts index c84c147ca7..3001783d7d 100644 --- a/packages/compact/compact/src/index.ts +++ b/packages/compact/compact/src/index.ts @@ -27,6 +27,12 @@ import type { CompactionResult } from './types.ts' export type { CompactionResult } from './types.ts' +/** Minimal agent context compaction needs without depending on the agent package. */ +export interface CompactAgentContext { + session: Session + options: { model?: string } +} + declare module 'cordis' { interface Context { compact: CompactService @@ -84,9 +90,10 @@ export abstract class CompactService extends Service { * exceeds the budget, compaction cannot help and the call may go out * over-budget. Bounding an individual unit's size is a separate concern. * - * @param session - the session whose surface may be compacted. - * @param system - the assembled system prompt, counted toward the estimate. - * @param model - the summarization model (a backend may override via config). + * @param agent - agent context owning the session surface and model options. + * @param turn - turn number of the pre-step checkpoint. + * @param step - step number about to start. + * @param fullSystemPrompt - assembled system prompt, counted toward the estimate. * @param signal - cancellation signal. A backend summarizing via * `ctx.llm.stream()` MUST forward this into the call's `GenerateOptions.signal` * so an abort/dispose tears down the in-flight summarization rather than @@ -94,9 +101,10 @@ export abstract class CompactService extends Service { * @returns the compaction result, or `null` if no compaction was needed. */ abstract compactIfNeeded( - session: Session, - system: string, - model: string, + agent: CompactAgentContext, + turn: number, + step: number, + fullSystemPrompt: string, signal: AbortSignal, ): Promise @@ -120,7 +128,9 @@ export abstract class CompactService extends Service { * @param session - the session whose surface is mutated. * @param start - inclusive seq of the first surface node to compact. * @param end - inclusive seq of the last surface node to compact. - * @param model - summarization model. + * @param agent - agent context used by router-aware summarizers. + * @param turn - lifecycle turn forwarded to request-routing seams. + * @param step - lifecycle step forwarded to request-routing seams. * @param signal - optional cancellation signal. A backend that summarizes via * `ctx.llm.stream()` MUST forward this into the call's `GenerateOptions.signal` * so an abort/dispose tears down the in-flight summarization rather than @@ -136,7 +146,9 @@ export abstract class CompactService extends Service { session: Session, start: number, end: number, - model: string, + agent: CompactAgentContext, + turn: number, + step: number, signal?: AbortSignal, ): Promise } diff --git a/packages/compact/compact/tests/compact.spec.ts b/packages/compact/compact/tests/compact.spec.ts index b3ad9d1501..5b9e033fcc 100644 --- a/packages/compact/compact/tests/compact.spec.ts +++ b/packages/compact/compact/tests/compact.spec.ts @@ -3,6 +3,7 @@ import { Context } from 'cordis' import { CompactService } from '@deepseek-ai/dsh-compact' import type { CompactionResult } from '@deepseek-ai/dsh-compact' import { Session, SessionId } from '@deepseek-ai/dsh-session' +import type { CompactAgentContext } from '@deepseek-ai/dsh-compact' /** * A trivial concrete CompactService implementing the abstract contract. The @@ -15,10 +16,11 @@ class StubCompactService extends CompactService { lastSignal: AbortSignal | undefined override async compactIfNeeded( - _session: Session, - _systemPrompt?: string, - _model?: string, - signal?: AbortSignal, + _agent: CompactAgentContext, + _turn: number, + _step: number, + _fullSystemPrompt: string, + signal: AbortSignal, ): Promise { this.lastSignal = signal return null @@ -28,7 +30,9 @@ class StubCompactService extends CompactService { session: Session, start: number, end: number, - _model: string, + _agent: CompactAgentContext, + _turn: number, + _step: number, signal?: AbortSignal, ): Promise { this.lastSignal = signal @@ -54,6 +58,10 @@ class StubCompactService extends CompactService { } describe('CompactService seam', () => { + function stubAgent(session: Session, model?: string): CompactAgentContext { + return { session, options: model === undefined ? {} : { model } } + } + it('registers as ctx.compact', () => { const ctx = new Context() void new StubCompactService(ctx) @@ -72,7 +80,8 @@ describe('CompactService seam', () => { it('exposes the abstract contract methods', async () => { const ctx = new Context() const svc = new StubCompactService(ctx) - expect(await svc.compactIfNeeded(new Session(SessionId('s')))).toBeNull() + const session = new Session(SessionId('s')) + expect(await svc.compactIfNeeded(stubAgent(session), 1, 1, '', new AbortController().signal)).toBeNull() }) it('compact/* events merge into SessionEventMap and are log-only', async () => { @@ -80,7 +89,7 @@ describe('CompactService seam', () => { const svc = new StubCompactService(ctx) const session = new Session(SessionId('s')) - const result = await svc.compactRegion(session, 0, 0, 'm') + const result = await svc.compactRegion(session, 0, 0, stubAgent(session, 'm'), 1, 1) const startEvent = session.events.find(e => e.type === 'compact/start') expect(startEvent).toBeDefined() @@ -98,10 +107,10 @@ describe('CompactService seam', () => { const session = new Session(SessionId('s')) const controller = new AbortController() - await svc.compactRegion(session, 0, 0, 'm', controller.signal) + await svc.compactRegion(session, 0, 0, stubAgent(session, 'm'), 1, 1, controller.signal) expect(svc.lastSignal).toBe(controller.signal) - await svc.compactIfNeeded(session, undefined, undefined, controller.signal) + await svc.compactIfNeeded(stubAgent(session), 1, 1, '', controller.signal) expect(svc.lastSignal).toBe(controller.signal) }) }) diff --git a/packages/core/agent-loop/src/loop.ts b/packages/core/agent-loop/src/loop.ts index f6b286cb28..2982a60fc8 100644 --- a/packages/core/agent-loop/src/loop.ts +++ b/packages/core/agent-loop/src/loop.ts @@ -388,29 +388,34 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle, // (or turn-start listeners on the first step) joins before the request. drainSteering(ctx, agent, turn) - // Assemble the system prompt for this step. Done HERE (before step/start) - // because the pre-step seam needs it: compaction measures token pressure - // against the system prompt (it counts toward the budget) and a listener - // also receives the model to summarize with. runStep reuses this same - // assembly for the request, so the prompt is assembled once per step. - const assembly = await ctx.systemPrompt.assemble() - const system = [renderPrompt(assembly), agent.options.systemPrompt ?? ''] - .filter(text => text.length > 0) - .join('\n\n') - - // The step's AbortController exists BEFORE the pre-step seam so a cancel() - // during the seam aborts any in-flight work a listener started (e.g. a - // compaction summarization call). Cleared on every exit path below. + // The step's AbortController exists BEFORE any async pre-step work so a + // dispose() or cancel() — in a synchronous turn-start listener or an + // async listener whose effect fires before we block — always has an armed + // abort to cancel against. isDisposed below covers disposal, which does + // NOT set the cancel marker. Cleared on every exit path below. const abort = new AbortController() handle.setAbort(abort) - // Cancel landing before the seam: a synchronous `agent/turn-start` listener - // (or the previous step's continuation listeners) can have called - // `cancel()`. Drop the about-to-start step WITHOUT running the seam — no - // step is open yet, so end the turn `aborted` directly. - if (handle.isCancelled()) { + // Assemble the system prompt for this step. Done HERE (before step/start) + // because the pre-step seam needs it: compaction measures token pressure + // against the system prompt (it counts toward the budget). runStep reuses + // this same assembly for the request, so the prompt is assembled once per + // step. + const assembly = await ctx.systemPrompt.assemble() + const fullSystemPrompt = [renderPrompt(assembly), agent.options.systemPrompt ?? ''] + .filter(text => text.length > 0) + .join('\n\n') + + // Interruption landing after assembly: dispose() or cancel() in a + // turn-start listener (or a listener whose promise resolved before the + // await above) arms either handle.isDisposed() or handle.isCancelled(). + // The Abort was created first, so any concurrent abort also lands on it. + // Drop the about-to-start step WITHOUT running the seam — no step is open + // yet, so end the turn accordingly (disposed wins for an unambiguous + // reason). + if (handle.isCancelled() || handle.isDisposed()) { handle.setAbort(undefined) - reason = { kind: 'aborted', reason: handle.cancelReason() } + reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() } break } @@ -425,7 +430,7 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle, // throwing listener escapes to the outer catch, which closes the (not-yet- // open) step as a no-op and ends the turn via failTurn — a broken // pre-step plugin ends the turn, not the loop. - await ctx.serial('agent/pre-step', agent, turn, step, system, agent.options.model ?? '', abort.signal) + await ctx.serial('agent/pre-step', agent, turn, step, fullSystemPrompt, abort.signal) session.append('step/start', { turn, step }) stepOpen = true @@ -433,19 +438,20 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle, // Cancel landing in the seam / step-start window: a `cancel()` during the // pre-step seam (it aborted `abort.signal` above) OR a synchronous - // `agent/step-start` listener that cancels. Check AFTER setAbort/step-start - // and before `runStep`: drop the step, end the turn `aborted`. closeStep - // balances the already-appended step/start. - if (handle.isCancelled()) { + // `agent/step-start` listener that cancels. And disposal, which the earlier + // assembly check may have missed if it only checked isCancelled. Check + // AFTER step/start append + emit and before `runStep`: drop the step, end + // the turn accordingly. closeStep balances the already-appended step/start. + if (handle.isCancelled() || handle.isDisposed()) { handle.setAbort(undefined) - reason = { kind: 'aborted', reason: handle.cancelReason() } + reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() } closeStep() break } let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error } try { - stepOutcome = await runStep(ctx, agent, turn, step, assembly, system, abort.signal) + stepOutcome = await runStep(ctx, agent, turn, step, assembly, fullSystemPrompt, abort.signal) } catch (error: unknown) { stepOutcome = { error: toError(error) } } finally { diff --git a/packages/core/agent-loop/tests/loop.spec.ts b/packages/core/agent-loop/tests/loop.spec.ts index 2c6f9e06e8..aa565f15b5 100644 --- a/packages/core/agent-loop/tests/loop.spec.ts +++ b/packages/core/agent-loop/tests/loop.spec.ts @@ -322,9 +322,9 @@ describe('agent loop', () => { it('agent/pre-step fires once per step before the step is opened', async () => { // Two steps (a tool call, then a final text turn) → two model calls → two - // pre-step fires, each carrying the assembled system + model, BEFORE the - // step is opened and its request is derived (the request the adapter sees - // reflects any surface state at fire time). + // pre-step fires, each carrying the assembled full system prompt, BEFORE + // the step is opened and its request is derived (the request the adapter + // sees reflects any surface state at fire time). const adapter = new MockAdapter([ toolCallResponse('c1', 'echo', {}, 'calling echo'), textResponse('done'), @@ -336,18 +336,18 @@ describe('agent loop', () => { })) const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' }) - const fires: { turn: number; step: number; model: string }[] = [] - ctx.on('agent/pre-step', (subject, turn, step, _system, model) => { - if (subject === agent) fires.push({ turn, step, model }) + const fires: { turn: number; step: number; fullSystemPrompt: string }[] = [] + ctx.on('agent/pre-step', (subject, turn, step, fullSystemPrompt) => { + if (subject === agent) fires.push({ turn, step, fullSystemPrompt }) }) send(agent, 'go') await waitForIdle(ctx, agent) - // One fire per step, in order, each with the agent's model. + // One fire per step, in order, each with the assembled system prompt. expect(fires).toEqual([ - { turn: 1, step: 1, model: 'mock' }, - { turn: 1, step: 2, model: 'mock' }, + { turn: 1, step: 1, fullSystemPrompt: '' }, + { turn: 1, step: 2, fullSystemPrompt: '' }, ]) }) diff --git a/packages/core/agent-loop/tests/review-fixes.spec.ts b/packages/core/agent-loop/tests/review-fixes.spec.ts index a092bd8419..5b02ca60c1 100644 --- a/packages/core/agent-loop/tests/review-fixes.spec.ts +++ b/packages/core/agent-loop/tests/review-fixes.spec.ts @@ -1047,3 +1047,273 @@ describe('surface: assistant/message omits sourceEventSeqs when no chunks stream expect(JSON.stringify(agent.session.deriveMessages())).toContain('injected') }) }) + + + +describe('disposal/cancel honored during pre-step assembly (P1-1)', () => { + it('disposal during system-prompt assembly drops the about-to-start step as disposed', { timeout: 30000 }, async () => { + // Block `system-prompt/assemble` on a promise. Start disposal (which + // calls stop() synchronously, setting status=disposed), then release the + // block. The loop must check isDisposed() after assembly and end the turn + // `disposed` — no LLM call. Don't await fiber.dispose() before releasing + // the blocker: the dispose chain awaits agent.done, which hangs until the + // loop unblocks. + const adapter = new MockAdapter(['hang']) + let releaseAssemble!: () => void + const blocked = new Promise(r => void (releaseAssemble = r)) + + const ctx = new Context() + await ctx.plugin(LlmService) + await ctx.plugin(SessionStore) + await ctx.plugin(SystemPrompt) + await ctx.plugin(ToolRegistry) + await ctx.plugin(AgentRegistry) + await ctx.plugin(AgentLoop, { agents: [] }) + await ctx.plugin(Invariants, { freeze: false }) + ctx.llm.registerAdapter(['mock'], adapter) + + // Blocking listener on the parent context (survives fiber disposal). + const unlisten = ctx.on('system-prompt/assemble', async function (_assembly, next) { + await blocked + return next() + }) + + let agent!: ReactLoopAgent + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + agent = inner.agentLoop.create(AgentId('a-dispose-assemble'), { model: 'mock' }) + }, { inject: ['agentLoop'] })) + + const reasons: TurnEndReason[] = [] + ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason)) + + send(agent, 'go') + // Give the loop time to enter the step and reach assemble(). + await new Promise(r => setTimeout(r, 50)) + + // Start disposal — stop() sets status=disposed synchronously, then the + // disposer's await agent.done hangs because the loop is blocked in the + // waterfall. Do NOT await yet; release the blocker first. + const disposalDone = fiber.dispose() + + // Now release the blocked waterfall — the loop unblocks, checks + // isDisposed(), and exits, which resolves agent.done and disposalDone. + releaseAssemble() + await disposalDone + await agent.done + unlisten() + + const e = [...agent.session.events] + expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1) + expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1) + const turnEnd = e.findLast(x => x.type === 'turn/end') + expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' }) + // No step was opened, no LLM call was made. + expect(e.some(x => x.type === 'step/start')).toBe(false) + expect(e.some(x => x.type === 'assistant/chunk')).toBe(false) + // agent/turn-end may not fire when disposal happens during assembly: the + // fiber's disposer (stop→status=disposed) runs before closeTurn(true)'s + // emit, and the LIFO chain disposes effects in reverse registration order. + // The turn/end durable record is the one that matters. + }) + + it('cancel during system-prompt assembly drops the about-to-start step as aborted', { timeout: 30000 }, async () => { + const adapter = new MockAdapter([textResponse('should not appear')]) + let releaseAssemble!: () => void + const blocker = new Promise(r => void (releaseAssemble = r)) + + const ctx = new Context() + await ctx.plugin(LlmService) + await ctx.plugin(SessionStore) + await ctx.plugin(SystemPrompt) + await ctx.plugin(ToolRegistry) + await ctx.plugin(AgentRegistry) + await ctx.plugin(AgentLoop, { agents: [] }) + await ctx.plugin(Invariants, { freeze: false }) + ctx.llm.registerAdapter(['mock'], adapter) + + const unlisten = ctx.on('system-prompt/assemble', async function (_assembly, next) { + await blocker + return next() + }) + + let agent!: ReactLoopAgent + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + agent = inner.agentLoop.create(AgentId('a-cancel-assemble'), { model: 'mock' }) + }, { inject: ['agentLoop'] })) + + const reasons: TurnEndReason[] = [] + ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason)) + + send(agent, 'go') + await new Promise(r => setTimeout(r, 50)) + agent.cancel('user cancelled during assembly') + + releaseAssemble() + await waitForIdle(ctx, agent) + await fiber.dispose() + await agent.done + unlisten() + + const e = [...agent.session.events] + expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1) + expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1) + const turnEnd = e.findLast(x => x.type === 'turn/end') + expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ + kind: 'aborted', + reason: 'user cancelled during assembly', + }) + expect(e.some(x => x.type === 'step/start')).toBe(false) + expect(e.some(x => x.type === 'assistant/chunk')).toBe(false) + expect(e.some(x => x.type === 'assistant/message')).toBe(false) + expect(adapter.requests).toHaveLength(0) + expect(reasons).toEqual([{ kind: 'aborted', reason: 'user cancelled during assembly' }]) + }) + + it('disposal during agent/pre-step seam ends the turn disposed', { timeout: 15000 }, async () => { + // Block the `agent/pre-step` serial seam on a promise we control, then + // dispose the agent's fiber. When the block releases, the loop must see + // isDisposed() at the post-seam check and end the turn disposed. + const adapter = new MockAdapter(['hang']) + let releasePreStep!: () => void + const blocker = new Promise(r => void (releasePreStep = r)) + + const ctx = new Context() + await ctx.plugin(LlmService) + await ctx.plugin(SessionStore) + await ctx.plugin(SystemPrompt) + await ctx.plugin(ToolRegistry) + await ctx.plugin(AgentRegistry) + await ctx.plugin(AgentLoop, { agents: [] }) + await ctx.plugin(Invariants, { freeze: false }) + ctx.llm.registerAdapter(['mock'], adapter) + + ctx.on('agent/pre-step', async () => { + await blocker + }) + + let agent!: ReactLoopAgent + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + agent = inner.agentLoop.create(AgentId('a-dispose-prestep'), { model: 'mock' }) + }, { inject: ['agentLoop'] })) + + const reasons: TurnEndReason[] = [] + ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason)) + + send(agent, 'go') + await new Promise(r => setTimeout(r, 50)) + + // Start disposal, then release the block, then await disposal. + const disposalDone = fiber.dispose() + releasePreStep() + await disposalDone + await agent.done + + // After the pre-step seam finishes, the post-seam cancel/dispose check + // catches disposal. The step was never opened, no LLM call was made. + const e = [...agent.session.events] + expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1) + expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1) + const turnEnd = e.findLast(x => x.type === 'turn/end') + // Disposal wins the post-seam check — reason is `disposed`. + expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' }) + expect(e.some(x => x.type === 'assistant/chunk')).toBe(false) + // agent/turn-end may not fire when disposal happens during pre-step: the + // fiber's disposer runs before closeTurn(true)'s emit. The durable turn/end + // is the authoritative record. + }) + + it('cancel during agent/pre-step seam ends the turn aborted', { timeout: 15000 }, async () => { + // Block `agent/pre-step`, then cancel() the agent. When the block releases, + // the post-seam check catches cancellation and ends the turn aborted. + const adapter = new MockAdapter(['hang']) + let releasePreStep!: () => void + const blocker = new Promise(r => void (releasePreStep = r)) + + const ctx = new Context() + await ctx.plugin(LlmService) + await ctx.plugin(SessionStore) + await ctx.plugin(SystemPrompt) + await ctx.plugin(ToolRegistry) + await ctx.plugin(AgentRegistry) + await ctx.plugin(AgentLoop, { agents: [] }) + await ctx.plugin(Invariants, { freeze: false }) + ctx.llm.registerAdapter(['mock'], adapter) + + ctx.on('agent/pre-step', async () => { + await blocker + }) + + let agent!: ReactLoopAgent + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + agent = inner.agentLoop.create(AgentId('a-cancel-prestep'), { model: 'mock' }) + }, { inject: ['agentLoop'] })) + + const reasons: TurnEndReason[] = [] + ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason)) + + send(agent, 'go') + await new Promise(r => setTimeout(r, 30)) + agent.cancel('user cancelled') + + releasePreStep() + await waitForIdle(ctx, agent) + await fiber.dispose() + await agent.done + + const e = [...agent.session.events] + expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1) + expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1) + const turnEnd = e.findLast(x => x.type === 'turn/end') + expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: 'user cancelled' }) + expect(e.some(x => x.type === 'assistant/chunk')).toBe(false) + expect(reasons).toEqual([{ kind: 'aborted', reason: 'user cancelled' }]) + }) + + it('disposal during assembly does not leak an LLM call or append assistant/chunk', { timeout: 15000 }, async () => { + // The key assertion from the original bug report: after disposal, no + // assistant/chunk or assistant/message appears — the turn ends disposed + // before any model interaction. + const adapter = new MockAdapter([textResponse('should not appear')]) + let releaseAssemble!: () => void + const blocker = new Promise(r => void (releaseAssemble = r)) + + const ctx = new Context() + await ctx.plugin(LlmService) + await ctx.plugin(SessionStore) + await ctx.plugin(SystemPrompt) + await ctx.plugin(ToolRegistry) + await ctx.plugin(AgentRegistry) + await ctx.plugin(AgentLoop, { agents: [] }) + await ctx.plugin(Invariants, { freeze: false }) + ctx.llm.registerAdapter(['mock'], adapter) + + ctx.on('system-prompt/assemble', async function (_assembly, next) { + await blocker + return next() + }) + + let agent!: ReactLoopAgent + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + agent = inner.agentLoop.create(AgentId('a-dispose-no-leak'), { model: 'mock' }) + }, { inject: ['agentLoop'] })) + + send(agent, 'go') + await new Promise(r => setTimeout(r, 50)) + + const disposalDone = fiber.dispose() + releaseAssemble() + await disposalDone + await agent.done + + const e = [...agent.session.events] + expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1) + expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1) + // The critical assertions: after disposal, the turn has no assistant + // artifacts — the turn ended disposed before the model was invoked. + expect(e.some(x => x.type === 'assistant/chunk')).toBe(false) + expect(e.some(x => x.type === 'assistant/message')).toBe(false) + expect(adapter.requests).toHaveLength(0) + // The durable turn/end reason is the authoritative record; agent/turn-end + // may not fire when disposal interleaves with closeTurn(true)'s emit. + }) +}) diff --git a/packages/core/agent/src/types.ts b/packages/core/agent/src/types.ts index 5bb603f1f6..83b3dfa4e7 100644 --- a/packages/core/agent/src/types.ts +++ b/packages/core/agent/src/types.ts @@ -200,13 +200,12 @@ declare module 'cordis' { * transform or veto, but the loop must wait for the mutation to complete * before opening the step and deriving, and serial isolates listeners from * each other (one finishes its surface append before the next runs). - * `system`/`model` are the assembled values a listener needs to measure - * pressure (system counts toward the budget) and to summarize (the model). - * `signal` cancels any in-flight work a listener starts (e.g. a summarization - * model call). + * `fullSystemPrompt` is the assembled prompt a listener needs to measure + * pressure (the system prompt counts toward the budget). `signal` cancels any + * in-flight work a listener starts (e.g. a summarization model call). * @mode serial */ - 'agent/pre-step'(agent: Agent, turn: number, step: number, system: string, model: string, signal: AbortSignal): Promise | void + 'agent/pre-step'(agent: Agent, turn: number, step: number, fullSystemPrompt: string, signal: AbortSignal): Promise | void /** * Waterfall: mutate the fully-assembled {@link GenerateOptions} before the * model call (hooks, model switching, tool filtering, …). Call `next()` to