diff --git a/docs/architecture.md b/docs/architecture.md index bff7b03091..344ea08e26 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -135,8 +135,9 @@ forever: drain steering (late steering from previous step's listeners) session('step/start'); emit agent/step-start assembly = ctx.systemPrompt.assemble() ⟵ waterfall system-prompt/assemble + await ctx.parallel('agent/pre-request') ⟵ surface mutation (compaction) before derive req = {model, system, tools, messages: session.deriveMessages(), signal} - req = waterfall agent/request ⟵ hooks, compaction, model switch + req = waterfall agent/request ⟵ hooks, model switch stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks) session('assistant/chunk'); emit agent/stream-chunk if assembler.finish is error/aborted: throw ⟵ adapter's in-band error path → @@ -192,7 +193,7 @@ Every MVP feature (including the TODO-marked ones), with the mechanism that impl | `/loop` | on `agent/turn-end`, `send()` the next iteration; or force-continue | | Dynamic workflow | orchestrator plugin on `agent/turn-end` / `agent/step-end` driving `send`/`steer` (+ sub-agents later) | | Queued + steering messages | core `Agent.send()` / `Agent.steer()` | -| Context compaction (auto + manual) | the `dsh-compact` seam (`ctx.compact`) + a backend (`dsh-compact-basic`) wrapping `agent/request`: a backend summarizes an older surface range into a single `user/message` `replace` op, bracketed by log-only `compact/*` events; auto = check token pressure before each model call, manual = a (deferred) `/compact` tool invoking the same `ctx.compact` routine. See the [compaction capability-seam RFC](rfc/proposed/feature/2026-06-18-compaction-capability-seam.md) | +| Context compaction (auto + manual) | the `dsh-compact` seam (`ctx.compact`) + a backend (`dsh-compact-basic`) on the awaited `agent/pre-request` seam: a backend summarizes an older surface range into a single `user/message` `replace` op, bracketed by log-only `compact/*` events; auto = check token pressure before each model call (every step — runaway-turn survival), manual = a (deferred) `/compact` tool invoking the same `ctx.compact` routine. See the [compaction capability-seam RFC](rfc/implemented/feature/2026-06-18-compaction-capability-seam.md) | | System prompt configurability | `ctx.systemPrompt.section()` with ordering | | AGENTS.md (root) | a section provider reading the file | | AGENTS.md (subdir, on-touch) + file-change notices | `agent.inject()` from a watcher / tool-result listener | @@ -220,6 +221,6 @@ Code skeletons for the three plugin shapes (tool, hook/permission-gate, UI) and Tracked here deliberately — each is designed-for but not implemented: - **Sub-agent spawn/fork semantics** (seam: `AgentLoop.create()`); inter-agent channels beyond `send`/`steer`/events. -- **Compaction** — the `dsh-compact` seam (`ctx.compact`) and the `dsh-compact-basic` backend exist (auto thresholds, summarization on the `agent/request` seam, `compact/*` session events via declaration merging). The model-facing `/compact` consumer tool is still deferred. See [the compaction capability-seam RFC](rfc/proposed/feature/2026-06-18-compaction-capability-seam.md). +- **Compaction** — the `dsh-compact` seam (`ctx.compact`) and the `dsh-compact-basic` backend exist (auto thresholds, summarization on the awaited `agent/pre-request` seam, `compact/*` session events via declaration merging). The model-facing `/compact` consumer tool is still deferred. See [the compaction capability-seam RFC](rfc/implemented/feature/2026-06-18-compaction-capability-seam.md). - **Parallel tool execution** (concurrency-safety hints on ToolDefinition). - **Session branching/tree** (pi-style entry tree) if needed beyond seed-based forking. diff --git a/docs/cordis-catalog/events-and-services.md b/docs/cordis-catalog/events-and-services.md index b8758bef00..7514a5fe41 100644 --- a/docs/cordis-catalog/events-and-services.md +++ b/docs/cordis-catalog/events-and-services.md @@ -11,7 +11,7 @@ The **harness tier** below (the `@deepseek-ai/dsh-*` packages) is the vocabulary ## Events -Dispatch modes: **emit** (fire-and-forget), **waterfall** (each listener gets `next()` and may transform or veto — see [waterfall semantics](../architecture.md#cordis-waterfall-semantics-important)), **parallel** (awaited fan-out, no veto). The harness declares 24 events across 6 scopes. +Dispatch modes: **emit** (fire-and-forget), **waterfall** (each listener gets `next()` and may transform or veto — see [waterfall semantics](../architecture.md#cordis-waterfall-semantics-important)), **parallel** (awaited fan-out, no veto). The harness declares 25 events across 6 scopes. ### `agent/*` @@ -49,7 +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:220`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:242`](../../packages/core/agent/src/types.ts) + +#### `agent/pre-request` — parallel + +Awaited surface-mutation checkpoint, fired BEFORE the step's message history is derived (and thus before agent/request). The loop awaits `ctx.parallel('agent/pre-request', …)` after assembling the system prompt but before `session.deriveMessages()`, then derives 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), 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. + +Awaited (parallel), 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 deriving. `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). + +```ts cordis-catalog +'agent/pre-request'(agent: Agent, turn: number, step: number, system: string, model: string, signal: AbortSignal): Promise | void +``` + +Types: [Agent](../core-data-structures/core.md) + +Source: [`packages/core/agent/src/types.ts:202`](../../packages/core/agent/src/types.ts) #### `agent/queued` — emit @@ -65,7 +79,7 @@ Source: [`packages/core/agent/src/types.ts:156`](../../packages/core/agent/src/t #### `agent/request` — waterfall -Waterfall: mutate the fully-assembled GenerateOptions before the model call (hooks, compaction, model switching, tool filtering, …). Call `next()` to delegate, or return without it to short-circuit. +Waterfall: mutate the fully-assembled GenerateOptions before the model call (hooks, model switching, tool filtering, …). Call `next()` to delegate, or return without it to short-circuit. For surface mutation that must precede history derivation (compaction), use agent/pre-request instead — by the time this fires, `options.messages` is already derived. ```ts cordis-catalog 'agent/request'(agent: Agent, turn: number, step: number, options: GenerateOptions, next: () => Promise): Promise @@ -73,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:189`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:211`](../../packages/core/agent/src/types.ts) #### `agent/status` — emit @@ -97,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:214`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:236`](../../packages/core/agent/src/types.ts) #### `agent/step-end` — emit @@ -121,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:195`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:217`](../../packages/core/agent/src/types.ts) #### `agent/step-start` — emit @@ -145,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:209`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:231`](../../packages/core/agent/src/types.ts) #### `agent/turn-continuation` — waterfall @@ -157,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:202`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:224`](../../packages/core/agent/src/types.ts) #### `agent/turn-end` — emit @@ -373,7 +387,7 @@ 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, systemPrompt?: string, model?: string, signal?: AbortSignal, ): Promise +abstract compactIfNeeded( session: Session, system: string, model: string, signal: AbortSignal, ): Promise abstract compactRegion( session: Session, start: number, end: number, model: string, signal?: AbortSignal, ): Promise ``` diff --git a/docs/core-data-structures/compaction.md b/docs/core-data-structures/compaction.md index ef22d79c94..e637961784 100644 --- a/docs/core-data-structures/compaction.md +++ b/docs/core-data-structures/compaction.md @@ -1,6 +1,6 @@ # Compaction -The compaction seam — a [capability seam](../rfc/implemented/architecture/2026-06-13-capability-seams.md) split like bash: interface ([dsh-compact](../../packages/compact/compact), `ctx.compact`), implementation (a backend such as [dsh-compact-basic](../../packages/compact/compact-basic)), and consumer (a `/compact` tool, deferred). Compaction is **one optional capability**, not part of the agent-loop spine — so its vocabulary lives here, not in [core.md](core.md). A tokenizer- or template-based backend is a sibling package implementing the same interface. Unlike bash, the interface necessarily depends on `dsh-session` and `dsh-llm`: its verbs are defined over a `Session` and its output is the `ContentBlock` vocabulary (see the [compaction capability-seam RFC](../rfc/proposed/feature/2026-06-18-compaction-capability-seam.md)). +The compaction seam — a [capability seam](../rfc/implemented/architecture/2026-06-13-capability-seams.md) split like bash: interface ([dsh-compact](../../packages/compact/compact), `ctx.compact`), implementation (a backend such as [dsh-compact-basic](../../packages/compact/compact-basic)), and consumer (a `/compact` tool, deferred). Compaction is **one optional capability**, not part of the agent-loop spine — so its vocabulary lives here, not in [core.md](core.md). A tokenizer- or template-based backend is a sibling package implementing the same interface. Unlike bash, the interface necessarily depends on `dsh-session` and `dsh-llm`: its verbs are defined over a `Session` and its output is the `ContentBlock` vocabulary (see the [compaction capability-seam RFC](../rfc/implemented/feature/2026-06-18-compaction-capability-seam.md)). Source: [`packages/compact/compact/src/types.ts`](../../packages/compact/compact/src/types.ts) @@ -50,4 +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, systemPrompt?, 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. Both take an optional `signal: AbortSignal` that a backend summarizing via `ctx.llm.stream()` must forward 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(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-request` 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. + +Auto-compaction runs on the awaited `agent/pre-request` loop seam (fired once per step, BEFORE the request history is derived), not the `agent/request` waterfall: compaction mutates the session surface in place, and the loop derives the request from the already-compacted surface. Retention is turn-agnostic — the only structural guard is step-alignment (a compacted region 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. diff --git a/docs/rfc/README.md b/docs/rfc/README.md index a731aa8ff9..07d3418bcb 100644 --- a/docs/rfc/README.md +++ b/docs/rfc/README.md @@ -44,7 +44,6 @@ Do NOT write one for a mechanical or local choice (a variable name, a one-file r | [Agent Client Protocol (ACP) support for external editors](proposed/feature/2026-06-14-acp-agent-client-protocol.md) | 2026-06-14 | | [Multiplex concurrent ACP sessions over one connection](proposed/feature/2026-06-14-acp-multi-session.md) | 2026-06-14 | | [Optional Code Mode — model writes TypeScript against an SDK of all tools](proposed/feature/2026-06-15-optional-code-mode.md) | 2026-06-15 | -| [Compaction as a capability seam (abstract contract + basic backend)](proposed/feature/2026-06-18-compaction-capability-seam.md) | 2026-06-18 | ### Simplification @@ -83,6 +82,7 @@ Do NOT write one for a mechanical or local choice (a variable name, a one-file r | Title | First proposed | |---|---| | [Rich ACP bash rendering — the terminal card (`_meta`) and command classification](implemented/feature/2026-06-18-acp-terminal-and-tool-rendering.md) | 2026-06-18 | +| [Compaction as a capability seam (abstract contract + basic backend)](implemented/feature/2026-06-18-compaction-capability-seam.md) | 2026-06-18 | | [Subagent capability seam](implemented/feature/2026-06-21-subagent-capability-seam.md) | 2026-06-21 | | [ACP subagent backend (out-of-process delegation)](implemented/feature/2026-06-22-acp-subagent-backend.md) | 2026-06-22 | 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 new file mode 100644 index 0000000000..56139cc750 --- /dev/null +++ b/docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md @@ -0,0 +1,118 @@ +# RFC: Compaction as a capability seam (abstract contract + basic backend) + +Status: implemented (2026-06-18; retention/seam reform 2026-06-26) + +## Context + +A long-running agent conversation grows without bound. As the event log accumulates turns, the derived message history eventually approaches the model's context window — the model then truncates mid-response (`max-tokens`) or degrades. **Compaction** is the mitigation: replace a run of older history with a concise summary, keeping recent context intact. + +The [session surface](../../implemented/architecture/2026-06-18-session-surface.md) was built as the foundation for exactly this — a linked list over the event log with a `surfaceOp: { op: 'replace', start, end }` operation purpose-built to shadow a range of nodes and insert a replacement, with `sourceEventSeqs` recording provenance so the decision replays deterministically. What remained was the plugin that *decides what to compact and produces the summary*. + +Two forces shape the design. First, compaction is **swappable**: token counting can be a char/4 heuristic or a real tokenizer, and summarization can be a model call, a template, or a remote service — these vary independently of *when* and *which range* to compact. Second, a later commit (`ce43c25`) closed `SurfaceEventType` to five event types (`user/message`, `assistant/message`, `tool/result`, `context/message`, `steering/message`); only those may carry `surfaceOp`. A bespoke `compaction/*` event therefore **cannot** itself appear on the surface — the compiler rejects `surfaceOp` on it and the invariants plugin rejects it at runtime. + +## Decision + +### Compaction is a capability seam, split interface / implementation + +Per the [capability-seams RFC](../../implemented/architecture/2026-06-13-capability-seams.md), compaction ships as separate packages so the contract, the algorithm, and (later) the consumer surface evolve independently: + +1. **Interface** — `@deepseek-ai/dsh-compact`: an abstract `CompactService` owning the `ctx.compact` key, the `CompactionResult` vocabulary, and the `compact/*` session events. It declares `compactIfNeeded()` and `compactRegion()` as **abstract** — the contract states *what* compaction does, not *how*. +2. **Implementation** — `@deepseek-ai/dsh-compact-basic`: a concrete `BasicCompactService` that owns the entire algorithm — token estimation (char/4 + per-block overhead), the tail→head retention walk, summarization via `ctx.llm.stream()`, the surface replacement, the lock, and the `agent/pre-request` auto-compaction listener. A tokenizer-based or template-based backend is a sibling package (or a subclass overriding the two protected estimation/summarization hooks). +3. **Consumer** — deferred. A `/compact` tool and slash command will `inject: ['compact']` and call the contract; they are intentionally out of scope here so the seam settles first. + +### The contract depends on `dsh-session` and `dsh-llm` — a deliberate deviation + +The capability-seams RFC states the interface package "depends only on cordis" (true of `dsh-bash`, whose vocabulary is self-contained). Compaction **cannot** honor that: its verbs are defined *over* a `Session` (`compactRegion(session, start, end)`) and its output *is* the content vocabulary (`CompactionResult.summary: ContentBlock[]`). There is no way to express the contract without naming `Session`/`SessionEvent` (from `dsh-session`) and `ContentBlock` (from `dsh-llm`). + +This is not a coupling smell — it is the contract's domain. The "only cordis" guidance was always shorthand for "the interface depends only on what the contract genuinely names, and never on an implementation." `dsh-session` and `dsh-llm` are themselves interface/vocabulary packages, not implementations; `dsh-compact` still imports no backend. The seam's real invariant — *consumers and implementations evolve independently behind an abstract service* — holds intact. + +### Abstract `compactIfNeeded` / `compactRegion`, algorithm in the backend + +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). + +### Auto-compaction runs on `agent/pre-request`, a dedicated surface-mutation seam + +Compaction is a **surface mutation**, not a request transform — and that distinction is the seam it belongs on. The loop's request lifecycle, per step, is: assemble the system prompt → derive the message history from the surface → run the `agent/request` waterfall → call the model. An earlier cut wedged compaction into the `agent/request` waterfall, which forced two problems: (1) the loop had already derived `messages` from the *stale* surface, so the listener had to mutate the surface and then *re-derive* and overwrite `request.messages` — a double-derive whose only purpose was to undo the premature first derive; and (2) `agent/request` also carries downstream-injected context a listener might have added to `request.messages`, which compaction cannot act on (it can only compact the surface), inviting the confusion of measuring tokens compaction can't shed. + +The fix is a new awaited loop seam, **`agent/pre-request`** (`@mode parallel`), fired by the loop *after* system assembly and *before* `deriveMessages()`: + +``` +assembly = ctx.systemPrompt.assemble() +await ctx.parallel('agent/pre-request', agent, turn, step, system, model, signal) ⟵ compaction mutates the surface here +messages = session.deriveMessages() ⟵ single derive, reflects the compaction +request = waterfall agent/request ⟵ pure request transform (hooks, model switch) +``` + +This makes the layering correct *by construction*: compaction mutates the surface, the loop derives **once** from the result (no double-derive), and at `pre-request` the assembled `messages` do not yet exist — so a listener structurally *cannot* see or be expected to act on downstream-injected context. `agent/request` reverts to a pure request transformer. The seam is `parallel` (awaited fan-out, no veto), like `session/flush`: a listener mutates the surface as a side effect; there is nothing to transform or return. + +This **amends** the original RFC's claim of "NO changes to `dsh-agent-loop`; compaction is a pure plugin." That claim was load-bearing for a wrong design — reusing `agent/request` was the mistake. Per the pre-release "foundation over blast radius" stance, adding the correct seam (one event declaration in `dsh-agent`, one awaited emit in the loop) beats preserving a no-change boast that locked in the double-derive. + +### Retention is turn-agnostic; step-alignment is the only structural guard + +Auto-compaction fires before **every** model call (every step), not once per turn. This is **load-bearing for runaway-turn survival**: a tool-heavy ReAct turn appends an `assistant/message` + a `tool/result` per step, so the surface grows *within* a turn. A single turn can grow past the window on its own (a "runaway turn") — and the only moment to rescue it before the next model call overflows is the next step's `pre-request`. Gating compaction to a turn's first step (or, worse, retaining the whole in-flight turn verbatim) re-opens exactly the hole compaction exists to close: the harness would die when compaction is most needed. + +So retention does **not** protect the in-flight turn, and turn boundaries play no role in it. `compactIfNeeded` walks the surface nodes tail→head, summing per-node token estimates, and retains the smallest tail-run of **whole units** whose total reaches `retainTokens`; everything older is compacted (head-anchored — see below). A *unit* is either a whole closed step (its `assistant/message` plus its `tool/result`s) or a single no-step node (a pre-step `user/message`, inter-step `steering/message`, or injection `context/message`). The walk rounds toward retaining *more*: when the raw token cutoff lands inside a step, it extends the retained side head-ward until the boundary is a step-aligned start. The single structural guard is therefore **step-alignment** — the compacted region always ends on a step boundary, so it never splits a step's tool-calls from their `tool/result`s (which would produce a transcript every provider rejects). `compactRegion` enforces step-alignment strictly, throwing on a splitting boundary. + +A runaway turn thus compacts exactly like any other history: its early *closed* steps get summarized while its recent steps stay verbatim. When the only compactable content left is an un-splittable open tail step (its tool-calls have no results yet), compaction declines (`null`) and retries once that step closes. + +**Single-unit overflow is out of scope, by design.** If a single retained unit — one closed step, or a large free node such as a pasted `user/message` — *alone* exceeds the budget, compaction cannot help and the next model call may go out over-budget. Bounding an individual unit's size is a separate concern (output truncation), handled elsewhere; compaction makes no promise about it, and the harness without such a mechanism can still break on a single oversized unit. This is named honestly rather than papered over. + +### Head-anchoring: one auto checkpoint, always at the head + +`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 + +`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. `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. + +### Surface replacement: `compact/*` events are log-only; one `user/message` carries the summary + +Because `SurfaceEventType` is closed, the summary cannot ride on a `compact/*` event. The backend instead appends a **single `user/message`** with `surfaceOp: { op: 'replace', start, end }` whose `content` is the (framed) summary and whose `sourceEventSeqs` covers the shadowed nodes *and* the bookkeeping events. The `compact/*` events are pure log records (lock + provenance). The surface mutation sits **inside** the lock — `compact/end` is the last event appended: + +``` +compact/start → log-only. Acquires the lock. +[summarize older range via the backend] +compact/summary → log-only. Provenance: raw summary, range, shadowed seqs, token count. +user/message → surfaceOp { op:'replace', start, end }. THE surface mutation (framed summary). + deriveMessages() renders it as a user-role message. +compact/end → log-only. Releases the lock (carries `error` on a recoverable failure). +``` + +`deriveMessages()` then yields `[summary_as_user_message, ...retained_nodes]`. Reusing `user/message` is honest rather than a workaround: a summary genuinely *is* user-role context. + +### Checkpoint framing + incremental merge (backend-private) + +The landed `user/message` is not the raw summary: the backend wraps it in a checkpoint preamble (so a resuming model reads it as established background, not a fresh request) and `…` tags. The tags make a prior checkpoint detectable on the next cycle, and the summarization prompt then instructs the model to *merge it in place* (preserve still-true facts, drop stale) rather than re-summarize verbatim — a cheap incremental merge that needs no extra log/event machinery. The raw, unframed summary stays on the `compact/summary` provenance event. This framing is entirely a **backend HOW decision** — the contract only promises "a single replace `user/message` carries the (possibly framed) summary; the raw summary lives on `compact/summary`." A template or remote backend may frame differently or not at all. + +### Blocking via a log-recorded lock, plus a crash/recoverable failure taxonomy + +The `compact/start … compact/end` bracket is justified, in order of what now does the work: + +1. **Crash-detectable orphan + provenance** (primary). Summarization is a slow model call persisted *after* `compact/start`. A crash mid-summarization leaves a `compact/start` with no matching `compact/end` — a detectable orphan. Releasing the lock last (rather than first) converts the crash window from *silent corruption* into that detectable orphan. +2. **Prevents concurrent compaction.** `compactRegion` refuses to start if the current turn holds an unmatched `compact/start`. (The loop is single-threaded across the awaited `pre-request`, so this is also a re-entry tripwire — a thrown "already in progress" signals a real bug.) + +Two failure paths, both documented: + +- **Crash** (the loop dies mid-summarization): a dangling `compact/start`, no closer. Because `compact/*` are **log-only**, the orphan is **inert** — the surface replacement never landed, so the full, uncompacted history derives correctly. Generic turn-repair (`interruptedTurnClosers`) closes the turn with a synthetic `turn/end`; the orphan sits *before* that `turn/end`, so the turn-scoped in-progress check never sees it and a crash can't wedge future compaction. Compaction simply re-attempts at the next `pre-request`. +- **Recoverable** (summarization throws but the loop survives): the backend appends `compact/end` with its **`error`** field set, leaving the surface untouched, and the model call proceeds with full history. + +`compact/end` keeps its `error?` field (mirroring `tool/result`'s self-contained error — one event tells success from failure without correlating a sibling). There is no separate `compact/error` event. + +**Core session repair stays compaction-agnostic — deliberately.** `interruptedTurnClosers` is never taught about `compact/*`. Teaching it would force every future `xxx/start … xxx/end` plugin pair to patch a core module — exactly the coupling the capability-seam architecture exists to avoid. Because the log-only orphan is inert, no special repair is needed: generic turn-repair plus the inertness of an un-landed surface mutation is sufficient. + +## Consequences + +- **New packages**: `packages/compact/compact` (interface) and a sibling `compact-basic` (backend) under `packages/compact/`, wired into the root tsconfigs. The consumer tier is deferred. +- **New loop seam**: `agent/pre-request` (`@mode parallel`) declared in `dsh-agent` and emitted by `dsh-agent-loop` between system assembly and history derivation. This is a documented change to the loop — `docs/architecture.md` records it and the generated cordis catalog carries its signature. +- **`SessionEventMap`** gains `compact/start` / `compact/summary` / `compact/end` by declaration merging (merge-extensible); `SurfaceEventType` is **not** touched. These are session events, not cordis `Events`, so the event-taxonomy gate needs no entry. +- **No changes** to `dsh-session` or `dsh-invariants`: the surface replace op, the surface-metadata runtime guard, and the turn-enclosure invariant all already exist and are reused. +- **Wiring**: `dsh-compact-basic` is loaded in `examples/coding-agent`'s `cordis.yml`, so the seam ships in the real demo (it was previously loaded nowhere). + +## Testing + +- **Unit** (`dsh-compact-basic`): the whole-unit retention walk, the convergence-invariant throw, both failure paths (`compact/end` with/without `error`), head-anchoring producing a non-monotonic `shadowedRange`, decline-on-open-tail, crash-orphan inertness, and the **runaway-turn regression** — a single oversized open turn compacts its early closed steps (proven to fail on the layer-2 protection it replaced). Driven through the real `dsh-invariants` plugin and the real Loader/inject path. +- **Loop** (`dsh-agent-loop`): `agent/pre-request` fires once per step, before derive, awaited; a surface mutation in a `pre-request` listener is reflected in the single derived request. +- **With-key e2e** (`examples/coding-agent`): a real model + real bash session with a lowered `contextWindow`/`retainTokens` triggers compaction mid-session; the test verifies the WORLD (a `compact/start…end` pair landed, the surface shrank, the agent still completed the task after compaction). This is compaction's first real-world exercise and the runaway-survival net. +- **Snapshot (deferred, named gap)**: a full-transcript snapshot of a runaway-turn compaction is NOT yet possible — `dsh-llm-replay` derives one model call per `(turn, step)` from `assistant/chunk` events, but the summarization call records no `assistant/chunk`s and carries no `sessionId` (it binds to the anonymous cursor and claims a non-existent extra script). Covering it needs net-new replay infrastructure (record/replay an interleaved summarization call) and is scheduled as a follow-up rather than discovered mid-build. diff --git a/docs/rfc/proposed/feature/2026-06-18-compaction-capability-seam.md b/docs/rfc/proposed/feature/2026-06-18-compaction-capability-seam.md deleted file mode 100644 index 2d559fa65c..0000000000 --- a/docs/rfc/proposed/feature/2026-06-18-compaction-capability-seam.md +++ /dev/null @@ -1,59 +0,0 @@ -# RFC: Compaction as a capability seam (abstract contract + basic backend) - -Status: proposed (2026-06-18) - -## Context - -A long-running agent conversation grows without bound. As the event log accumulates turns, the derived message history eventually approaches the model's context window — the model then truncates mid-response (`max-tokens`) or degrades. **Compaction** is the mitigation: replace a run of older history with a concise summary, keeping recent context intact. - -The [session surface](../../implemented/architecture/2026-06-18-session-surface.md) was built as the foundation for exactly this — a linked list over the event log with a `surfaceOp: { op: 'replace', start, end }` operation purpose-built to shadow a range of nodes and insert a replacement, with `sourceEventSeqs` recording provenance so the decision replays deterministically. What remained was the plugin that *decides what to compact and produces the summary*. - -Two forces shape the design. First, compaction is **swappable**: token counting can be a char/4 heuristic or a real tokenizer, and summarization can be a model call, a template, or a remote service — these vary independently of *when* and *which range* to compact. Second, a later commit (`ce43c25`) closed `SurfaceEventType` to five event types (`user/message`, `assistant/message`, `tool/result`, `context/message`, `steering/message`); only those may carry `surfaceOp`. A bespoke `compaction/*` event therefore **cannot** itself appear on the surface — the compiler rejects `surfaceOp` on it and the invariants plugin rejects it at runtime. - -## Decision - -### Compaction is a capability seam, split interface / implementation - -Per the [capability-seams RFC](../../implemented/architecture/2026-06-13-capability-seams.md), compaction ships as separate packages so the contract, the algorithm, and (later) the consumer surface evolve independently: - -1. **Interface** — `@deepseek-ai/dsh-compact`: an abstract `CompactService` owning the `ctx.compact` key, the `CompactionResult` vocabulary, and the `compact/*` session events. It declares `compactIfNeeded()` and `compactRegion()` as **abstract** — the contract states *what* compaction does, not *how*. -2. **Implementation** — `@deepseek-ai/dsh-compact-basic`: a concrete `BasicCompactService` that owns the entire algorithm — token estimation (char/4 + per-block overhead), the tail→head retention walk, summarization via `ctx.llm.generate()`, the surface replacement, the lock, and the `agent/request` auto-compaction listener. A tokenizer-based or template-based backend is a sibling package (or a subclass overriding the two protected estimation/summarization hooks). -3. **Consumer** — deferred. A `/compact` tool and slash command will `inject: ['compact']` and call the contract; they are intentionally out of scope here so the seam settles first. - -### The contract depends on `dsh-session` and `dsh-llm` — a deliberate deviation - -The capability-seams RFC states the interface package "depends only on cordis" (true of `dsh-bash`, whose vocabulary is self-contained). Compaction **cannot** honor that: its verbs are defined *over* a `Session` (`compactRegion(session, start, end)`) and its output *is* the content vocabulary (`CompactionResult.summary: ContentBlock[]`). There is no way to express the contract without naming `Session`/`SessionEvent` (from `dsh-session`) and `ContentBlock` (from `dsh-llm`). - -This is not a coupling smell — it is the contract's domain. The "only cordis" guidance was always shorthand for "the interface depends only on what the contract genuinely names, and never on an implementation." `dsh-session` and `dsh-llm` are themselves interface/vocabulary packages, not implementations; `dsh-compact` still imports no backend. The seam's real invariant — *consumers and implementations evolve independently behind an abstract service* — holds intact. We record the deviation here so a future reader doesn't mistake it for an accident or "fix" it by smuggling `Session` behind an opaque handle. - -### Abstract `compactIfNeeded` / `compactRegion`, algorithm in the backend - -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 (e.g. turn-count instead of token-budget) 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. - -### Surface replacement: `compact/*` events are log-only; one `user/message` carries the summary - -Because `SurfaceEventType` is closed, the summary cannot ride on a `compact/*` event. The backend instead appends a **single `user/message`** with `surfaceOp: { op: 'replace', start, end }` whose `content` is the summary `ContentBlock[]` and whose `sourceEventSeqs` covers the shadowed nodes *and* the bookkeeping events. The `compact/*` events are pure log records (lock + provenance), never on the surface. The surface mutation sits **inside** the lock — `compact/end` is the last event appended: - -``` -compact/start → log-only. Acquires the lock. -[summarize older range via the backend] -compact/summary → log-only. Provenance: summary, range, shadowed seqs, token count. -user/message → surfaceOp { op:'replace', start, end }. THE surface mutation. - deriveMessages() renders it as a user-role message. -compact/end → log-only. Releases the lock. -``` - -Ordering the surface mutation **before** `compact/end` is deliberate: `session.append()` commits one event at a time, so there is no multi-event transaction to make the sequence atomic. Releasing the lock last converts the crash window from *silent corruption* (a `compact/end` that claims compaction finished while the surface was never shadowed) into a *detectable orphaned lock* (a `compact/start` with no matching `compact/end`), which a persistence backend already detects on reload. A `session/event` listener on `compact/end` likewise never sees the lock free before the replacement has landed. - -`deriveMessages()` then yields `[summary_as_user_message, ...retained_nodes]`. An alternative — extending `SurfaceEventType` to admit a `compact/*` type — was rejected: the closed union is a deliberate safety boundary (only message-producing events reach the model), and a summary genuinely *is* user-role context, so reusing `user/message` is honest rather than a workaround. - -### Blocking via a log-recorded lock, not a mutex - -Compaction must be serialized: no second compaction starts before the first finishes, and no ordinary events interleave the slow summarization. Rather than an in-memory mutex (invisible to replay, lost on crash), the lock **is** the log: `compactRegion` refuses to start if the last `compact/start` has no matching `compact/end` after it. `compact/start` is appended first (fast, synchronous), the slow model call runs, then the `compact/summary` and `user/message` replacement land, and only then is `compact/end` appended — in a `catch` that records the error, so a failed summarization can never wedge the lock. Because the backend runs compaction synchronously inside the `agent/request` waterfall, the loop is single-threaded for that window; the lock additionally gives observability and lets a persistence backend detect an orphaned `compact/start` on reload. - -## Consequences - -- **New packages**: `packages/compact/compact` (interface) and a sibling `compact-basic` (backend) under `packages/compact/`, wired into the three root tsconfigs. The consumer tier is deferred. -- **`SessionEventMap`** gains `compact/start` / `compact/summary` / `compact/end` by declaration merging (merge-extensible); `SurfaceEventType` is **not** touched. These are session events, not cordis `Events`, so the event-taxonomy gate needs no entry. -- **No changes** to `dsh-session`, `dsh-invariants`, or `dsh-agent-loop`: the surface replace op, the surface-metadata runtime guard, and the `agent/request` waterfall all already exist. Compaction is a pure plugin on documented seams. -- The capability-seams convention gains a second reference beyond bash, and a documented case where "interface depends only on cordis" relaxes to "depends only on interface/vocabulary packages the contract genuinely names." On acceptance, [AGENTS.md](../../../../AGENTS.md) § Conventions and [architecture.md](../../../architecture.md) § "Capability seams" should note this relaxation. diff --git a/examples/coding-agent/cordis.yml b/examples/coding-agent/cordis.yml index 0347115cd5..e3a3016dda 100644 --- a/examples/coding-agent/cordis.yml +++ b/examples/coding-agent/cordis.yml @@ -65,6 +65,16 @@ failures before moving on. Verify your work by running the code or tests. Keep answers brief and factual. +# Automatic context compaction: when the derived history approaches the model's +# context window, summarize an older range into a checkpoint so a long-running +# or tool-heavy session keeps fitting. A leaf entry (needs ctx.llm + the +# agent-loop's `agent/pre-request` seam from the app above). +- id: compact-basic + name: '@deepseek-ai/dsh-compact-basic' + config: + contextWindow: 128000 + retainTokens: 20480 + # The subagent seam + BOTH in-process backends + two model-facing tools, as leaf # entries after the app (which provides ctx.agents/ctx.tools). spawn (a fresh # child) and fork (a child seeded with the parent's completed-turn prefix) are diff --git a/examples/coding-agent/tests/compaction.e2e.ts b/examples/coding-agent/tests/compaction.e2e.ts new file mode 100644 index 0000000000..fefcd579d0 --- /dev/null +++ b/examples/coding-agent/tests/compaction.e2e.ts @@ -0,0 +1,95 @@ +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it } from 'vitest' +import type { Context } from 'cordis' +import { AgentId } from '@deepseek-ai/dsh-agent' +import { codingHarness, finalText, SYSTEM_PROMPT, waitForIdle } from './harness.ts' + +/** + * The compaction smoke test: a real model runs a multi-step bash task with a + * deliberately tiny context window, so the auto-compaction listener fires + * MID-SESSION and summarizes the older history into a checkpoint. This is the + * first end-to-end exercise of the compaction seam (it is wired nowhere else), + * and the runaway-survival regression net — it proves a session that grows past + * the window keeps running rather than overflowing. Key-gated. + * + * Verifies the WORLD, not the agent's self-report: a compact/start…end pair + * landed in the real session log, the surface actually shrank (a replace node + * exists and shadowed older nodes), and the agent still produced a final answer + * after compaction (so the summarized history did not break the conversation). + */ + +let workdir: string | undefined +let ctx: Context | undefined + +afterEach(async () => { + await ctx?.fiber.dispose() + ctx = undefined + if (workdir !== undefined) await rm(workdir, { recursive: true, force: true }) + workdir = undefined +}) + +describe.skipIf(!process.env.DEEPSEEK_API_KEY)('compaction: a long session compacts mid-flight and keeps running', () => { + it('summarizes older history into a checkpoint without breaking the task', async () => { + workdir = await mkdtemp(join(tmpdir(), 'dsh-compaction-')) + // A few files for the model to read, so multiple bash steps accumulate + // surface nodes (tool calls + results) and grow the history. + for (let i = 1; i <= 4; i++) { + await writeFile(join(workdir, `file${i}.txt`), `This is file number ${i}. `.repeat(40)) + } + + // Tiny window so a handful of steps crosses the threshold. The convergence + // invariant requires summarizationMaxTokens + retainTokens <= window * + // ratio = floor(8000 * 0.5) = 4000; 1500 + 2000 = 3500 <= 4000. + ctx = await codingHarness(workdir, { + compact: { + contextWindow: 8000, + thresholdRatio: 0.5, + retainTokens: 2000, + summarizationMaxTokens: 1500, + }, + }) + const agent = ctx.agentLoop.create(AgentId('e2e-compaction'), { + model: 'deepseek-v4-flash', + systemPrompt: SYSTEM_PROMPT, + }) + + agent.send([{ + type: 'text', + text: 'Read file1.txt, file2.txt, file3.txt, and file4.txt one at a time using cat ' + + '(a separate bash command for each). After reading all four, tell me how many ' + + 'files you read and the number mentioned in file1.txt.', + }]) + await waitForIdle(ctx, agent) + + const events = [...agent.session.events] + + // A compaction ran: the start…end bracket landed in the real log. + const starts = events.filter(e => e.type === 'compact/start') + const ends = events.filter(e => e.type === 'compact/end') + expect(starts.length).toBeGreaterThan(0) + expect(ends.length).toBe(starts.length) // every start was released + + // It succeeded at least once: a compact/summary provenance event and a + // replace-op user/message (the surface mutation) both landed. + const summaries = events.filter(e => e.type === 'compact/summary') + expect(summaries.length).toBeGreaterThan(0) + const replaceNode = events.find((e) => { + const se = e as unknown as { type: string; surfaceOp?: unknown } + return se.type === 'user/message' && typeof se.surfaceOp === 'object' && se.surfaceOp !== null + }) + expect(replaceNode).toBeDefined() + + // The summary shadowed real older nodes (the surface shrank vs. the raw + // message-producing event count). + const summaryData = summaries[0]!.data as { shadowedSeqs: number[] } + expect(summaryData.shadowedSeqs.length).toBeGreaterThan(0) + + // The conversation survived compaction: the agent produced a final answer + // that reflects the work (it read four files). + const answer = finalText(events).toLowerCase() + expect(answer.length).toBeGreaterThan(0) + expect(answer).toMatch(/\b(4|four)\b/) + }, 240_000) +}) diff --git a/examples/coding-agent/tests/harness.ts b/examples/coding-agent/tests/harness.ts index fbe9b10db5..207652f69b 100644 --- a/examples/coding-agent/tests/harness.ts +++ b/examples/coding-agent/tests/harness.ts @@ -10,6 +10,8 @@ import { LocalBashExecutor } from '@deepseek-ai/dsh-bash-local' import * as ToolBash from '@deepseek-ai/dsh-tool-bash' import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek' import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl' +import { BasicCompactService } from '@deepseek-ai/dsh-compact-basic' +import type { BasicCompactConfig } from '@deepseek-ai/dsh-compact-basic' /** * Shared harness for the coding-agent e2e suites: the full plugin stack @@ -21,7 +23,19 @@ export const SYSTEM_PROMPT = 'You are a coding agent. Your only tool is bash; ' + 'do file operations with cat/grep/heredocs, check [exit code: N] markers, ' + 'and report results briefly.' -export async function codingHarness(workdir: string, persistenceRoot?: string): Promise { +/** Options for {@link codingHarness}. */ +export interface CodingHarnessOptions { + /** Durable JSONL persistence root (the resume suite needs it; others stay file-free). */ + persistenceRoot?: string + /** + * Load {@link BasicCompactService} with this config so the compaction e2e can + * trigger compaction at a small, controlled history size. Omitted ⇒ no + * compaction plugin (the default suites run without it). + */ + compact?: BasicCompactConfig +} + +export async function codingHarness(workdir: string, options: CodingHarnessOptions = {}): Promise { const ctx = new Context() await ctx.plugin(LlmService) await ctx.plugin(SessionStore) @@ -32,10 +46,13 @@ export async function codingHarness(workdir: string, persistenceRoot?: string): await ctx.plugin(LlmDeepSeek, { models: ['deepseek-v4-flash'] }) await ctx.plugin(LocalBashExecutor, { cwd: workdir, timeoutMs: 30_000 }) await ctx.plugin(ToolBash) + // Compaction is opt-in: only the compaction e2e loads it, with a lowered + // contextWindow/retainTokens so a short real session crosses the threshold. + if (options.compact !== undefined) await ctx.plugin(BasicCompactService, options.compact) // Durable JSONL persistence is opt-in: only the resume e2e needs it, and the // other suites stay file-free. Loaded last so a resume's deferred // `ctx.inject(['sessionPersistence'])` resolves once this is present. - if (persistenceRoot !== undefined) await ctx.plugin(SessionPersistenceJsonl, { root: persistenceRoot }) + if (options.persistenceRoot !== undefined) await ctx.plugin(SessionPersistenceJsonl, { root: options.persistenceRoot }) return ctx } diff --git a/examples/coding-agent/tests/resume.e2e.ts b/examples/coding-agent/tests/resume.e2e.ts index 450938fc6d..4be11ed3ea 100644 --- a/examples/coding-agent/tests/resume.e2e.ts +++ b/examples/coding-agent/tests/resume.e2e.ts @@ -38,7 +38,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('resume: continue a persisted ses // Run 1: a fresh agent on a KNOWN session id learns a secret, then we // dispose the whole context (simulating process exit) so only the JSONL // log on disk survives. - ctx = await codingHarness(process.cwd(), root) + ctx = await codingHarness(process.cwd(), { persistenceRoot: root }) const first = ctx.agents.create({ agentId: AgentId('resume-1'), sessionId: SESSION_ID, @@ -52,7 +52,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('resume: continue a persisted ses // Run 2: a brand-new context over the SAME root resumes the persisted // session. The loaded event log seeds the live session, so the model sees // run 1's exchange as conversation history. - ctx = await codingHarness(process.cwd(), root) + ctx = await codingHarness(process.cwd(), { persistenceRoot: root }) const resumed = (await ctx.agents.resume({ agentId: AgentId('resume-2'), resumeSessionId: SESSION_ID, diff --git a/packages/compact/README.md b/packages/compact/README.md index 384fe98ffe..10eaf1617a 100644 --- a/packages/compact/README.md +++ b/packages/compact/README.md @@ -8,4 +8,4 @@ A three-package capability seam (see [capability seams](../../docs/rfc/implement | `compact-basic/` | A backend: char/4 estimation + token-budget retention + `llm.stream()` summarization | (registers `ctx.compact`) | | `tool-compact/` (deferred) | Model-facing `/compact` tool over `ctx.compact` | (registers on `ctx.tools`) | -The interface lives at `compact/compact/`, the backend at `compact/compact-basic/`. Unlike the bash seam, it depends on `dsh-session` and `dsh-llm` — its verbs are defined over a `Session` and its output is the `ContentBlock` vocabulary, so the contract cannot be expressed without naming them. That deviation from the "interface depends only on cordis" guidance is intentional and recorded in the [compaction capability-seam RFC](../../docs/rfc/proposed/feature/2026-06-18-compaction-capability-seam.md). A tokenizer- or template-based backend would replace `compact-basic` without touching the interface or the tool. +The interface lives at `compact/compact/`, the backend at `compact/compact-basic/`. Unlike the bash seam, it depends on `dsh-session` and `dsh-llm` — its verbs are defined over a `Session` and its output is the `ContentBlock` vocabulary, so the contract cannot be expressed without naming them. That deviation from the "interface depends only on cordis" guidance is intentional and recorded in the [compaction capability-seam RFC](../../docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md). A tokenizer- or template-based backend would replace `compact-basic` without touching the interface or the tool. diff --git a/packages/compact/compact-basic/README.md b/packages/compact/compact-basic/README.md index 8aa64a1111..848c0cd6ae 100644 --- a/packages/compact/compact-basic/README.md +++ b/packages/compact/compact-basic/README.md @@ -2,18 +2,20 @@ 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. -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/proposed/feature/2026-06-18-compaction-capability-seam.md) for the design. +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. ## What it owns The abstract contract states only WHAT compaction does; this backend owns every HOW decision: - **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()` ALWAYS retains the in-flight turn's surface nodes verbatim (its initiating request and any mid-turn tool results — the exact input/observation the model is acting on, even if they exceed the budget), then walks the OLDER (closed-turn) nodes tail→head, summing per-node token estimates, and compacts everything older than the first node that overflows the `retainTokens` budget. The cutoff is snapped to a step boundary so the compacted region never splits a step's `assistant/message` tool-calls from their `tool/result`s (the budget is a soft target): it prefers snapping FORWARD to the next clean boundary, and falls back to snapping BACKWARD when the forward snap would reach the protected in-flight turn. If no step-aligned cutoff exists in the older range (e.g. its only content is an open tail step), it declines (returns `null`) and retries once an older step closes. `compactRegion()` enforces step-alignment strictly, throwing on a boundary that would split a step. Token-based (not turn-count) retention keeps more short turns and compacts tool-heavy turns sooner. +- **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 **step-alignment**: the compacted region always ends on a step boundary, 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 step-alignment strictly, throwing on a boundary that would split a step. +- **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. - **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. - **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/request` waterfall listener delegates to `compactIfNeeded()` before every model call (every step, not just a turn's first — a tool-heavy turn grows the surface mid-turn, so a runaway turn still compacts) and re-derives messages after compacting; the listener owns no threshold logic of its own (the single token-pressure check lives in `compactIfNeeded()`). +- **Auto-compaction** — an `agent/pre-request` listener delegates to `compactIfNeeded()` before every model call (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-request` is an awaited surface-mutation checkpoint that fires BEFORE the loop derives the request history, so compaction mutates the surface 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()`). +- **Failure handling** — the `compact/start … compact/end` bracket is a log-recorded lock: it makes a crash mid-summarization a detectable orphan (a `compact/start` with no `compact/end`), records provenance, and prevents a concurrent compaction. Two failure paths: a **crash** (the loop dies mid-summarization) leaves a dangling `compact/start` that is inert — `compact/*` events are log-only, the surface replacement never landed, so the full history derives fine and generic turn-repair closes the turn; a **recoverable** failure (summarization throws but the loop survives) appends `compact/end` with its `error` field set, leaving the surface untouched so the call proceeds with full history. Core session repair stays compaction-agnostic by design — it never learns about `compact/*`. `estimateContentTokens()` and `summarize()` are overridable hooks: a tokenizer-based or template-based backend can subclass `BasicCompactService` and override just those, reusing the retention walk and surface plumbing. @@ -26,7 +28,7 @@ The abstract contract states only WHAT compaction does; this backend owns every | `retainTokens` | `20480` | Tokens of recent context to keep intact. | | `summarizationModel` | `''` | Model for summarization (empty → use the agent's model). | | `summarizationMaxTokens` | `2048` | Max tokens for the summary response. | -| `auto` | `true` | Register the `agent/request` auto-compaction listener. Set `false` for manual-only. | +| `auto` | `true` | Register the `agent/pre-request` auto-compaction listener. Set `false` for manual-only. | ## Usage diff --git a/packages/compact/compact-basic/src/index.ts b/packages/compact/compact-basic/src/index.ts index 28a0040848..2ee0d82133 100644 --- a/packages/compact/compact-basic/src/index.ts +++ b/packages/compact/compact-basic/src/index.ts @@ -32,7 +32,7 @@ import { CompactService } from '@deepseek-ai/dsh-compact' import type { CompactionResult } from '@deepseek-ai/dsh-compact' import { BlockAssembler } from '@deepseek-ai/dsh-llm' import type { ContentBlock, FinishReason, GenerateOptions, Message } from '@deepseek-ai/dsh-llm' -import type { Session, SessionEvent, SurfaceNode } from '@deepseek-ai/dsh-session' +import type { Session, SessionEvent } from '@deepseek-ai/dsh-session' import { isStepAlignedStart, isStepAlignedEnd } from '@deepseek-ai/dsh-session' import type { Agent } from '@deepseek-ai/dsh-agent' import type { BasicCompactConfig, ResolvedConfig } from './types.ts' @@ -170,40 +170,40 @@ export class BasicCompactService extends CompactService { if (this.config.auto) { // Auto-compaction: delegate to compactIfNeeded before EVERY model call — - // every step, not just the first. A tool-heavy ReAct turn appends an - // assistant/message and a tool/result per step, so the surface (and the - // derived token count) grows within a turn; gating to step 1 would let a - // runaway turn overflow the window before the next turn's check. The - // listener stays agnostic — it owns NO threshold logic; compactIfNeeded is - // the single place that decides whether to compact, and its in-progress - // lock serializes concurrent attempts. - ctx.on('agent/request', async (agent: Agent, _turn, _step, request, next) => { - const before = this.estimateTokens(request.messages, request.system) + // every step, not just the first. This is LOAD-BEARING for runaway-turn + // survival: a tool-heavy ReAct turn appends an assistant/message and a + // tool/result per step, so the surface (and the derived token count) grows + // WITHIN a turn. The only moment to rescue a turn that alone approaches the + // window is the next step's pre-request; gating to a turn's first step + // would let a runaway turn overflow before the next turn's check. The + // listener owns NO threshold logic — compactIfNeeded is the single place + // that decides whether to compact, and its in-progress lock serializes + // concurrent attempts. + // + // It runs on `agent/pre-request` (a parallel surface-mutation checkpoint), + // NOT `agent/request`: compaction mutates the session surface, and the loop + // derives the request `messages` AFTER this fires — so a single derive + // already reflects the compaction, with no double-derive and no need to + // rewrite an already-assembled `messages` array. + ctx.on('agent/pre-request', async (agent: Agent, _turn: number, _step: number, system: string, model: string, signal: AbortSignal) => { try { - const result = await this.compactIfNeeded(agent.session, request.system, request.model, request.signal) + const result = await this.compactIfNeeded(agent.session, system, model, signal) if (result) { - // The surface has been mutated — re-derive messages for the call. - const rederived = agent.session.deriveMessages() - const afterTokens = this.estimateTokens(rederived, request.system) - + const after = this.estimateTokens(agent.session.deriveMessages(), system) ctx.logger.info( `compaction: shadowed ${result.shadowedSeqs.length} surface nodes ` + `(seqs ${result.shadowedRange.start}-${result.shadowedRange.end}, ` + `~${result.shadowedTokenCount} tokens) ` + - `→ ${afterTokens} estimated tokens after compaction ` + - `(pressure was ~${before})`, + `→ ${after} estimated tokens after compaction`, ) - - request.messages = rederived } } catch (error: unknown) { - // A failed compaction must not prevent the model call — proceed - // with the original messages. + // A failed compaction must not prevent the model call — the surface is + // untouched on failure, so the loop derives the full history and the + // call proceeds. const msg = error instanceof Error ? error.message : String(error) ctx.logger.warn(`compaction failed: ${msg}; proceeding with full history`) } - - return next() }) } } @@ -312,89 +312,89 @@ export class BasicCompactService extends CompactService { // ---- Core API (implements the abstract contract) ---- /** - * The sole token-pressure gate: estimate the current history, and if it - * exceeds the threshold (`contextWindow * thresholdRatio`), compact the oldest - * surface nodes outside the `retainTokens` budget. The auto-compaction listener - * delegates here rather than pre-checking, so this is the only place the - * decision lives. + * The sole token-pressure gate: estimate the current surface-derived history, + * and if it exceeds the threshold (`contextWindow * thresholdRatio`), compact + * the oldest surface nodes outside the `retainTokens` budget. The auto- + * compaction listener delegates here rather than pre-checking, so this is the + * only place the decision lives. + * + * Retention is a UNIFORM tail→head walk over the whole surface — turn + * boundaries play NO role. Walking node-by-node from the tail and summing + * token estimates, once the retained total reaches `retainTokens` the cutoff + * is rounded to a step-aligned boundary: if the walk stopped INSIDE a step, + * it continues head-ward past that step's `step/start` so the whole step is + * retained (never splitting a step's tool-calls from their results); if it + * stopped on a free node (a node belonging to no step), that is already a + * clean boundary. This always rounds toward retaining MORE (retained ≥ + * `retainTokens`) and is step-aligned by construction — no separate snap pass. + * + * The compacted range is always anchored at the surface HEAD (`nodes[0]`): + * auto-compaction re-consolidates any prior head checkpoint into one fresh + * checkpoint. Declines (`null`) when nothing is over threshold, when the whole + * surface fits the retain budget, or when no step-aligned cutoff exists in the + * compactable range (its only content is an open tail step — retry once it + * closes). */ override async compactIfNeeded( session: Session, - systemPrompt?: string, - model?: string, - signal?: AbortSignal, + system: string, + model: string, + signal: AbortSignal, ): Promise { const messages = session.deriveMessages() - const totalTokens = this.estimateTokens(messages, systemPrompt) + const totalTokens = this.estimateTokens(messages, system) const threshold = Math.floor(this.config.contextWindow * this.config.thresholdRatio) if (totalTokens < threshold) return null - // Walk surface nodes tail→head, accumulating token estimates. const nodes = session.surface.nodes if (nodes.length === 0) return null + const events = session.events const retainBudget = this.config.retainTokens - // ALWAYS retain the IN-FLIGHT turn's surface nodes verbatim — its initiating - // user request and any mid-turn tool results are the exact input/observation - // the model is acting on right now, even if they exceed the soft retain - // budget. Compacting them would hand the model a lossy summary of its own - // current task. Only nodes in PRIOR (closed) turns are eligible to compact; - // `protectedIdx` is the first surface node of the open turn (or `nodes.length` - // when the open turn has no surface nodes yet, e.g. before step 1). - const protectedIdx = this._openTurnFirstSurfaceIdx(session, nodes) - if (protectedIdx === 0) return null + // Walk tail→head summing per-node token estimates. `keepFromIdx` is the + // index of the OLDEST node we retain verbatim; everything strictly older + // (`[0, keepFromIdx - 1]`) is the compactable range. let accumulated = 0 - let cutoffIdx = -1 - // Seed the accumulator with the protected suffix so the retain budget is - // measured against what actually stays, then look for a cutoff only among - // the older (compactable) nodes. - for (let i = nodes.length - 1; i >= protectedIdx; i--) { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - const event = session.events[nodes[i]!.seq] - if (event) accumulated += this.estimateEventTokens(event) - } - - for (let i = protectedIdx - 1; i >= 0; i--) { - // nodes[i] bounded by i >= 0 and i < nodes.length — never undefined. + let keepFromIdx = nodes.length // nothing retained yet + for (let i = nodes.length - 1; i >= 0; i--) { // eslint-disable-next-line @typescript-eslint/no-non-null-assertion const node = nodes[i]! - const event = session.events[node.seq] + const event = events[node.seq] /* v8 ignore next -- node.seq is a surface-node seq, always a valid log index by construction */ - if (!event) continue - accumulated += this.estimateEventTokens(event) - if (accumulated > retainBudget) { - cutoffIdx = i - break - } + if (event) accumulated += this.estimateEventTokens(event) + keepFromIdx = i + if (accumulated >= retainBudget) break } - // If we walked the entire compactable range without exceeding the budget, - // everything outside the protected in-flight turn fits — no compaction - // needed. - if (cutoffIdx === -1) return null + // The whole surface fits the retain budget — nothing to compact. + if (keepFromIdx === 0) return null - // Snap the cutoff to a step-aligned end so the compacted region never splits - // a step (which would orphan a tool-call or its tool/result). The token - // budget is a soft target. PREFER snapping FORWARD (compact slightly more - // recent context to reach a clean boundary), but never into the protected - // in-flight turn: if the forward snap would reach `protectedIdx`, fall back - // to snapping BACKWARD to the previous step-aligned end (compact slightly - // less), and decline only if no step-aligned end exists in the compactable - // range at all. - const events = session.events - cutoffIdx = this._snapCutoff(events, nodes, cutoffIdx, protectedIdx) - if (cutoffIdx === -1) return null + // Round the cutoff to a step boundary: if `keepFromIdx` sits INSIDE a step, + // extend the retained side head-ward until the boundary is a step-aligned + // start, so the compacted range ends on a clean step edge. A node that + // belongs to no step is already a valid start. Decline if no step-aligned + // start exists at or below `keepFromIdx` (the compactable range is only an + // un-splittable open tail step — retry once it closes). + while (keepFromIdx > 0) { + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + if (isStepAlignedStart(events, nodes[keepFromIdx]!.seq)) break + keepFromIdx -= 1 + } + if (keepFromIdx === 0) return null - // nodes is non-empty (checked above) and cutoffIdx is a valid index. + // The compacted range is [head … keepFromIdx - 1], anchored at the head. + // The cutoff node `nodes[keepFromIdx - 1]` is necessarily a step-aligned END: + // the retained start `nodes[keepFromIdx]` is a step-aligned START (a boundary + // marker sits between them in the log), and that same boundary makes the node + // before it a step-aligned end — so no separate end check is needed. // eslint-disable-next-line @typescript-eslint/no-non-null-assertion const firstSeq = nodes[0]!.seq // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - const cutoffSeq = nodes[cutoffIdx]!.seq - const resolvedModel = model ?? '' + const cutoffSeq = nodes[keepFromIdx - 1]!.seq - return this.compactRegion(session, firstSeq, cutoffSeq, resolvedModel, signal) + return this.compactRegion(session, firstSeq, cutoffSeq, model, signal) } override async compactRegion( @@ -521,74 +521,6 @@ export class BasicCompactService extends CompactService { // ---- Internal helpers ---- /** - * The index of the first surface node that belongs to the currently-open turn - * — the boundary of the protected, never-compacted suffix. Returns - * `nodes.length` when the open turn has contributed no verbatim surface node - * yet (e.g. before step 1 appends anything), so the whole surface is - * compaction-eligible up to the tail. - * - * The in-flight turn's verbatim nodes (its request, mid-turn assistant - * messages, tool results — all `append` ops) form a CONTIGUOUS run at the TAIL - * of the surface. A compaction replacement node, though also appended during - * the open turn (seq > `turn/start`), lands at the position of the older range - * it shadowed — earlier in the surface, NOT in the tail run — so it is itself - * compaction-eligible (a later cycle can merge it). The protected suffix is - * therefore the contiguous tail run of nodes whose seq exceeds the open turn's - * `turn/start`, found by walking from the tail. With no open turn (a closed - * session — only manual `compactRegion`, never the auto path), nothing is - * protected and this returns `nodes.length`. - */ - private _openTurnFirstSurfaceIdx(session: Session, nodes: readonly SurfaceNode[]): number { - const openTurn = this._openTurn(session) - if (openTurn === null) return nodes.length - // Find the open turn's turn/start seq (scanning back from the tail). - let turnStartSeq = -1 - for (let i = session.events.length - 1; i >= 0; i--) { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - const e = session.events[i]! - if (e.type === 'turn/start' && e.data.turn === openTurn) { turnStartSeq = e.seq; break } - } - /* v8 ignore next -- _openTurn returned non-null, so its turn/start exists */ - if (turnStartSeq === -1) return nodes.length - // Walk from the tail while nodes belong to the open turn (seq > turn/start), - // taking only the CONTIGUOUS run — a compaction summary node appended this - // turn but sitting earlier in the surface stops the run and stays eligible. - let idx = nodes.length - while (idx > 0 && nodes[idx - 1]!.seq > turnStartSeq) idx -= 1 // eslint-disable-line @typescript-eslint/no-non-null-assertion - return idx - } - - /** - * Snap a raw token-budget cutoff index to a step-aligned end among the nodes - * BELOW the protected suffix (`protectedIdx`, the first node of the in-flight - * turn). Returns the snapped index, or `-1` if no step-aligned end exists in - * the compactable range (e.g. it is empty, or its only content is an open tail - * step). - * - * Prefers snapping FORWARD to the next step-aligned end (compact slightly more - * recent context for a clean boundary); if the forward scan reaches - * `protectedIdx` without finding one, falls back to scanning BACKWARD from the - * raw cutoff (compact slightly less). The protected suffix is never returned — - * it stays verbatim so the model sees its current task, not a summary. - */ - private _snapCutoff( - events: readonly SessionEvent[], - nodes: readonly SurfaceNode[], - rawCutoffIdx: number, - protectedIdx: number, - ): number { - // Forward: the next step-aligned end strictly below the protected suffix. - for (let i = rawCutoffIdx; i < protectedIdx; i++) { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - if (isStepAlignedEnd(events, nodes[i]!.seq)) return i - } - // Backward: the nearest step-aligned end at or below the raw cutoff. - for (let i = rawCutoffIdx - 1; i >= 0; i--) { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - if (isStepAlignedEnd(events, nodes[i]!.seq)) return i - } - return -1 - } /** * Frame the raw summary blocks into the content that lands on the surface: diff --git a/packages/compact/compact-basic/src/types.ts b/packages/compact/compact-basic/src/types.ts index 120fad0b19..7273150d8e 100644 --- a/packages/compact/compact-basic/src/types.ts +++ b/packages/compact/compact-basic/src/types.ts @@ -38,7 +38,35 @@ export const DEFAULTS: ResolvedConfig = { auto: true, } -/** Apply defaults to a partial config. */ +/** + * Apply defaults to a partial config and enforce the single-pass convergence + * invariant. + * + * `summarizationMaxTokens + retainTokens` must not exceed 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-request check passes and a second compaction cannot fire on the same + * content. Without it, a too-large summary budget or retain budget would leave + * the post-compaction history still 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. + * + * @throws if `summarizationMaxTokens + retainTokens > contextWindow * thresholdRatio`. + */ export function resolveConfig(config: BasicCompactConfig): ResolvedConfig { - return { ...DEFAULTS, ...config } + const resolved = { ...DEFAULTS, ...config } + const threshold = Math.floor(resolved.contextWindow * resolved.thresholdRatio) + const postCompactionFloor = resolved.summarizationMaxTokens + resolved.retainTokens + if (postCompactionFloor > threshold) { + throw new Error( + `BasicCompactConfig: summarizationMaxTokens (${resolved.summarizationMaxTokens}) + ` + + `retainTokens (${resolved.retainTokens}) = ${postCompactionFloor} exceeds the compaction ` + + `threshold contextWindow * thresholdRatio = ${threshold}; post-compaction history would ` + + 'stay over threshold and re-compact endlessly. Lower retainTokens/summarizationMaxTokens ' + + 'or raise contextWindow/thresholdRatio.', + ) + } + return resolved } diff --git a/packages/compact/compact-basic/tests/compact-basic.spec.ts b/packages/compact/compact-basic/tests/compact-basic.spec.ts index 3852b825f7..255a59a5b7 100644 --- a/packages/compact/compact-basic/tests/compact-basic.spec.ts +++ b/packages/compact/compact-basic/tests/compact-basic.spec.ts @@ -9,6 +9,9 @@ import type { SessionEvent, SurfaceEvent } from '@deepseek-ai/dsh-session' import * as Invariants from '@deepseek-ai/dsh-invariants' import type { Agent } from '@deepseek-ai/dsh-agent' +/** A never-aborted signal for the required `compactIfNeeded`/listener arg. */ +const SIGNAL = new AbortController().signal + /** * A BasicCompactService with summarize() stubbed (no real model call) and a * predictable token estimate, for deterministic unit tests of the algorithm. @@ -33,31 +36,14 @@ class TestCompactService extends BasicCompactService { } } -/** Create a test service with a throwaway context (auto disabled — no model). */ -function createTestService(config: BasicCompactConfig = {}): TestCompactService { - return new TestCompactService(new Context(), { auto: false, ...config }) -} - /** - * A test service where specific surface seqs (in `bigSeqs`) weigh 1000 tokens - * and every other message-producing event weighs 10 — for exercising the - * "newest node alone exceeds retainTokens" retention path. summarize() is - * stubbed (no model call). + * Create a test service with a throwaway context (auto disabled — no model). + * A small `summarizationMaxTokens` baseline keeps the convergence invariant + * (`summarizationMaxTokens + retainTokens <= contextWindow * thresholdRatio`) + * satisfied for the tiny windows these tests use; a test may override it. */ -class TestCompactServiceVarTokens extends BasicCompactService { - bigSeqs = new Set() - constructor(config: BasicCompactConfig = {}) { - super(new Context(), { auto: false, ...config }) - } - - override estimateEventTokens(event: SessionEvent): number { - if (this.bigSeqs.has(event.seq)) return 1000 - return super.estimateEventTokens(event) - } - - override async summarize(): Promise { - return [{ type: 'text', text: 'summary' }] - } +function createTestService(config: BasicCompactConfig = {}): TestCompactService { + return new TestCompactService(new Context(), { auto: false, summarizationMaxTokens: 1, ...config }) } /** @@ -188,44 +174,48 @@ function expectNoOrphanToolResults(messages: Message[]): void { } describe('BasicCompactService step-alignment (never split a tool-call/result pair)', () => { - it('compactIfNeeded snaps the cutoff forward past a mid-step boundary (no orphaned tool-result)', async () => { - // 3 turns, each one step = { assistant(tool-call) , tool/result }. Surface + it('compactIfNeeded rounds the retained boundary head-ward to keep a whole step (no orphaned tool-result)', async () => { + // 3 turns, each one step = { assistant(tool-call), tool/result }. Surface // (9 nodes): user1, asst1, res1, user2, asst2, res2, user3, asst3, res3 — - // 10/20/10 tokens. With retainTokens=55 the tail→head walk overflows at - // asst2 (idx4), so the RAW cutoff falls BETWEEN asst2 and its result res2 - // (idx5) — splitting turn 2's step. The fix snaps the cutoff forward to res2 - // so the whole step is compacted and no dangling result survives. + // 10/20/10 tokens. The tail→head walk retains by whole units; the compacted + // region always ends on a step boundary, so no step's tool-call is split + // from its result. retainTokens=55 keeps the recent tail; the older steps + // compact intact. const svc = createTestService({ contextWindow: 200, thresholdRatio: 0.5, retainTokens: 55 }) const session = toolTurnSession(3) - const result = await svc.compactIfNeeded(session) + const result = await svc.compactIfNeeded(session, '', 'm', SIGNAL) expect(result).not.toBeNull() - // res2 (idx5) was pulled into the compacted region by the snap, not stranded. + expect(result!.shadowedSeqs.length).toBeGreaterThan(0) + // No dangling tool-result: every compacted/retained step stayed whole. expectNoOrphanToolResults(session.deriveMessages()) - // Turn 3's step is retained intact (summary + user3 + asst3 + res3 = 4 msgs). - expect(session.deriveMessages().length).toBe(4) + // The most-recent step's result is retained verbatim (still on the surface). + const lastResultSeq = session.events.findLast(e => e.type === 'tool/result')!.seq + expect(result!.shadowedSeqs).not.toContain(lastResultSeq) }) - it('compactIfNeeded returns null when the only cutoff would enter an open tail step', async () => { - // A pre-step user/message then an OPEN step (assistant issued a tool-call, no - // tool/result / step/end yet — mid-flight). The token walk wants to compact - // into that open step, but its tool-call has no result yet; compacting it - // would defer the orphan. With no safe step-aligned cutoff, compactIfNeeded - // declines (returns null) rather than summarizing a pending tool-call away. - const s = new Session(SessionId('open-step')) + it('compactIfNeeded returns null when the only compactable region is an un-splittable single step', async () => { + // The surface is exactly ONE step: [assistant(tool-call), tool/result]. Over + // threshold (by the derived role overhead), the tail→head walk stops with the + // retained boundary at the tool/result — which is NOT a step-aligned start (its + // issuing assistant precedes it in the same step). Rounding head-ward to find a + // clean boundary reaches index 0, so there is no step-aligned cutoff in the + // compactable range: compactIfNeeded declines rather than splitting the step. + const s = new Session(SessionId('one-step')) s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - s.append('user/message', { content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) s.append('step/start', { turn: 1, step: 1 }) s.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'calling' }, { type: 'tool-call', id: CallId('c1'), name: 'bash', arguments: '{}' }], }, { surfaceOp: 'append' }) - // no tool/result, no step/end — the step is open at the tail. + s.append('tool/call', { turn: 1, step: 1, callId: CallId('c1'), name: 'bash', arguments: '{}' }) + s.append('tool/result', { turn: 1, step: 1, callId: CallId('c1'), content: [{ type: 'text', text: 'out' }], isError: false }, { surfaceOp: 'append' }) + s.append('step/end', { turn: 1, step: 1 }) + // Turn stays open. - const svc = createTestService({ contextWindow: 50, thresholdRatio: 0.5, retainTokens: 5 }) - const result = await svc.compactIfNeeded(s) + const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.1, retainTokens: 5 }) + const result = await svc.compactIfNeeded(s, '', 'm', SIGNAL) expect(result).toBeNull() - // The open step's assistant survived — its tool-call is intact for the result. expect(s.events.some(e => e.type === 'compact/start')).toBe(false) }) @@ -516,107 +506,103 @@ 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)).toBeNull() + expect(await svc.compactIfNeeded(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) + const result = await svc.compactIfNeeded(session, '', 'm', SIGNAL) expect(result).not.toBeNull() expect(result!.shadowedSeqs.length).toBeGreaterThan(0) }) it('walks tail→head and retains nodes within token budget', async () => { - const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.1, retainTokens: 15 }) + 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) + const result = await svc.compactIfNeeded(session, '', 'm', SIGNAL) expect(result).not.toBeNull() const nodes = session.surface.nodes expect(result!.shadowedSeqs.length).toBeGreaterThan(0) expect(result!.shadowedSeqs).not.toContain(nodes[nodes.length - 1]!.seq) }) - it('returns null when total tokens fit within budget', async () => { - const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.1, retainTokens: 1000 }) + it('returns null when the whole surface fits the retain budget (over threshold by role/system overhead)', async () => { + // threshold = floor(460*0.1) = 46. The 4 surface nodes weigh 10 each (raw 40 + // for the retention walk), but the derived estimate adds 4 role tokens per + // message → 56 ≥ 46, so the threshold check passes and the walk runs. The + // walk accumulates all 40 < retainTokens (45) without crossing the budget, + // so keepFromIdx reaches 0 and compaction declines. The invariant holds: + // summarizationMaxTokens (1) + retainTokens (45) = 46 ≤ threshold 46. + const svc = createTestService({ contextWindow: 460, thresholdRatio: 0.1, retainTokens: 45 }) const session = multiTurnSession(2, 1) - expect(await svc.compactIfNeeded(session)).toBeNull() + expect(await svc.compactIfNeeded(session, '', 'm', SIGNAL)).toBeNull() }) - it('retains the in-flight turn verbatim even when its newest node exceeds retainTokens', async () => { - // The current turn's first step has CLOSED (so its last node is step-aligned - // and would otherwise be a valid compaction cutoff), and that node — a fresh - // tool result — is larger than the whole retain budget. It must NOT be - // compacted: it is the observation the model needs for the turn's next step. - // Only the older closed turns are eligible. - const svc = new TestCompactServiceVarTokens({ contextWindow: 100, thresholdRatio: 0.1, retainTokens: 5 }) - const s = new Session(SessionId('big-tail')) - // Two closed turns (compactable older context). - for (const t of [1, 2]) { - s.append('turn/start', { turn: t, trigger: { kind: 'message', source: { kind: 'user' } } }) - s.append('step/start', { turn: t, step: 1 }) - s.append('user/message', { content: [{ type: 'text', text: `turn ${t}` }], source: { kind: 'user' } }, { surfaceOp: 'append' }) - s.append('assistant/message', { turn: t, step: 1, content: [{ type: 'text', text: `reply ${t}` }] }, { surfaceOp: 'append' }) - s.append('step/end', { turn: t, step: 1 }) - s.append('turn/end', { turn: t, reason: { kind: 'completed' } }) + it('compacts a runaway turn: its early CLOSED steps summarize while recent steps stay verbatim', async () => { + // The REGRESSION that motivated dropping turn-protection. A single in-flight + // (open) turn has grown past the threshold on its own: several CLOSED steps, + // each [assistant(tool-call), tool/result]. Retention is turn-agnostic, so + // the turn's OWN early closed steps are eligible — they compact while the + // recent tail stays verbatim, and the harness survives. + // + // On the OLD layer-2 code this test FAILS: the entire open turn was retained + // verbatim (protectedIdx = first open-turn node = 0), so compactIfNeeded + // returned null and shadowedSeqs would be empty — the runaway turn could + // never compact and the next model call would overflow the window. + const svc = createTestService({ contextWindow: 300, thresholdRatio: 0.1, retainTokens: 25 }) + const s = new Session(SessionId('runaway')) + // ONE open turn with 5 closed steps; each step is [asst(tool-call), result]. + s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + s.append('user/message', { content: [{ type: 'text', text: 'do a big multi-step task' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) + for (let step = 1; step <= 5; step++) { + s.append('step/start', { turn: 1, step }) + s.append('assistant/message', { + turn: 1, step, + content: [{ type: 'text', text: `step ${step}` }, { type: 'tool-call', id: CallId(`c${step}`), name: 'bash', arguments: '{}' }], + }, { surfaceOp: 'append' }) + s.append('tool/call', { turn: 1, step, callId: CallId(`c${step}`), name: 'bash', arguments: '{}' }) + s.append('tool/result', { turn: 1, step, callId: CallId(`c${step}`), content: [{ type: 'text', text: `out ${step}` }], isError: false }, { surfaceOp: 'append' }) + s.append('step/end', { turn: 1, step }) } - // The in-flight turn 3: a user request, then a CLOSED step 1 whose tool - // result is HUGE (1000 tokens). The step is closed (step/end), so the result - // node is step-aligned — without the in-flight-turn protection the retention - // walk would pick it as the cutoff and compact it away. The turn itself is - // still open (no turn/end): the model is mid-turn, about to run step 2. - s.append('turn/start', { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } }) - s.append('user/message', { content: [{ type: 'text', text: 'current request' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) - s.append('step/start', { turn: 3, step: 1 }) - s.append('assistant/message', { turn: 3, step: 1, content: [{ type: 'text', text: 'calling' }, { type: 'tool-call', id: CallId('huge'), name: 'bash', arguments: '{}' }] }, { surfaceOp: 'append' }) - s.append('tool/call', { turn: 3, step: 1, callId: CallId('huge'), name: 'bash', arguments: '{}' }) - const hugeSeq = s.append('tool/result', { - turn: 3, step: 1, callId: CallId('huge'), - content: [{ type: 'text', text: 'HUGE' }], isError: false, - }, { surfaceOp: 'append' }).seq - s.append('step/end', { turn: 3, step: 1 }) - svc.bigSeqs.add(hugeSeq) // make this node weigh 1000 tokens + // The turn stays OPEN (no turn/end) — the model is mid-turn, about to run + // step 6. Surface: user + 5×[asst, result] = 11 nodes. + const nodesBefore = s.surface.nodes.length + expect(nodesBefore).toBe(11) - const result = await svc.compactIfNeeded(s) + const result = await svc.compactIfNeeded(s, '', 'm', SIGNAL) expect(result).not.toBeNull() - // The in-flight turn's nodes — the request, the assistant, AND the huge - // result — are retained: none shadowed, all survive on the surface verbatim. - expect(result!.shadowedSeqs).not.toContain(hugeSeq) - const survivingSeqs = new Set(s.surface.nodes.map(n => n.seq)) - expect(survivingSeqs.has(hugeSeq)).toBe(true) - const requestSeq = s.events.find(e => e.type === 'user/message' && e.data.content.some(b => b.type === 'text' && b.text === 'current request'))!.seq - expect(survivingSeqs.has(requestSeq)).toBe(true) - // The older closed turns WERE compacted. + // Early steps of the SAME open turn were shadowed (impossible under layer 2). expect(result!.shadowedSeqs.length).toBeGreaterThan(0) + // The most-recent step's tool result is retained verbatim (still on surface). + const lastResultSeq = s.events.findLast(e => e.type === 'tool/result')!.seq + expect(result!.shadowedSeqs).not.toContain(lastResultSeq) + expect(s.surface.nodes.some(n => n.seq === lastResultSeq)).toBe(true) + // No orphaned tool-result survives (whole-step boundaries respected). + expectNoOrphanToolResults(s.deriveMessages()) }) it('returns null for an empty surface', async () => { - const svc = createTestService({ contextWindow: 10, thresholdRatio: 0.1 }) + const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.5, retainTokens: 10 }) const session = new Session(SessionId('empty')) - expect(await svc.compactIfNeeded(session)).toBeNull() + expect(await svc.compactIfNeeded(session, '', 'm', SIGNAL)).toBeNull() }) - it('compacts again within the same open turn (the prior summary node is still eligible)', async () => { - // After the first compaction lands a replacement summary node, that node is - // appended DURING the open turn (seq > turn/start) but sits earlier in the - // surface (at the shadowed range's position), NOT in the verbatim tail run. - // It must stay compaction-eligible: a second step in the SAME turn, still - // over threshold, must be able to compact older context — protectedIdx must - // not collapse to 0 and silently disable per-step auto-compaction. - // retainTokens=25 leaves a couple of retained closed-turn nodes after the - // first compaction (so the surface is [summary, …retained], not [summary]). - const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.1, retainTokens: 25 }) + it('compacts again after a prior summary node heads the surface (the summary stays eligible)', async () => { + // After the first compaction lands a replacement summary node at the head, + // a second compaction (still over threshold) re-consolidates it with newer + // context — head-anchoring means the prior checkpoint is always re-included, + // never stranded. retainTokens=25 leaves a couple of retained nodes after + // the first compaction (so the surface is [summary, …retained], not just + // [summary]). + 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) + const first = await svc.compactIfNeeded(s, '', 'm', SIGNAL) expect(first).not.toBeNull() - // The summary node now heads the surface; the open turn has no verbatim tail - // node yet, so the whole surface (incl. the summary) is eligible — the - // protected suffix is the contiguous tail run of open-turn nodes (none yet). - // The summary node's seq exceeds turn 5's turn/start, yet it sits at the - // head (not the tail), so it must NOT be counted as protected. + // The summary node now heads the surface with a fresh high seq. const summaryHeadSeq = s.surface.nodes[0]!.seq const turn5StartSeq = s.events.filter(e => e.type === 'turn/start').at(-1)!.seq expect(summaryHeadSeq).toBeGreaterThan(turn5StartSeq) @@ -629,7 +615,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) + const second = await svc.compactIfNeeded(s, '', 'm', SIGNAL) expect(second).not.toBeNull() expect(second!.shadowedSeqs.length).toBeGreaterThan(0) // The fresh open-turn nodes were NOT compacted. @@ -751,6 +737,27 @@ describe('BasicCompactService HMR safety', () => { }) }) +describe('BasicCompactService convergence invariant (config)', () => { + it('throws when summarizationMaxTokens + retainTokens exceeds the threshold', () => { + // threshold = floor(1000 * 0.5) = 500; 200 + 400 = 600 > 500 → reject. + expect(() => new BasicCompactService(new Context(), { + auto: false, contextWindow: 1000, thresholdRatio: 0.5, retainTokens: 400, summarizationMaxTokens: 200, + })).toThrow(/exceeds the compaction threshold/) + }) + + it('accepts the boundary case (sum equals the threshold)', () => { + // threshold = floor(1000 * 0.5) = 500; 100 + 400 = 500 ≤ 500 → allowed. + expect(() => new BasicCompactService(new Context(), { + auto: false, contextWindow: 1000, thresholdRatio: 0.5, retainTokens: 400, summarizationMaxTokens: 100, + })).not.toThrow() + }) + + it('the default config satisfies the invariant', () => { + // 2048 + 20480 = 22528 ≤ floor(128000 * 0.8) = 102400. + expect(() => new BasicCompactService(new Context(), { auto: false })).not.toThrow() + }) +}) + /** An adapter that emits a fixed summary text, for exercising the real summarize() path. */ class ScriptedAdapter extends LlmAdapter { lastOptions: GenerateOptions | null = null @@ -876,81 +883,73 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => { }) }) -describe('BasicCompactService auto-compaction (agent/request listener)', () => { - /** Fire the agent/request waterfall as the loop does. */ - function fireRequest(ctx: Context, agent: Agent, step: number, options: GenerateOptions): Promise { - return ctx.waterfall('agent/request', agent, 1, step, options, () => Promise.resolve(options)) +describe('BasicCompactService auto-compaction (agent/pre-request listener)', () => { + /** Fire the agent/pre-request parallel checkpoint as the loop does. */ + function firePreRequest(ctx: Context, agent: Agent, step: number, system: string, model: string): Promise { + return ctx.parallel('agent/pre-request', agent, 1, step, system, model, SIGNAL) } - it('compacts and rewrites request.messages when over threshold', async () => { - // Tiny window so the (large) session is over threshold; char/4 estimate. + it('compacts (mutating the surface) when over threshold', async () => { const { ctx } = await ctxWithModel('SUMMARY') - const svc = new BasicCompactService(ctx, { contextWindow: 200, thresholdRatio: 0.5, retainTokens: 20 }) + void new BasicCompactService(ctx, { contextWindow: 200, thresholdRatio: 0.5, retainTokens: 20, summarizationMaxTokens: 50 }) const session = multiTurnSession(5, 1) // 10 surface nodes const agent = stubAgent(session, 'test-model') + const before = session.surface.nodes.length - const messages = session.deriveMessages() - const before = messages.length - const options: GenerateOptions = { model: 'test-model', messages } + await firePreRequest(ctx, agent, 1, '', 'test-model') - const out = await fireRequest(ctx, agent, 1, options) - // The surface shrank — request.messages was re-derived to fewer entries. - expect(out.messages.length).toBeLessThan(before) + // The surface shrank in place, and a summary checkpoint landed. + expect(session.surface.nodes.length).toBeLessThan(before) expect(session.events.some(e => e.type === 'compact/summary')).toBe(true) - // Re-derived first message is the framed summary checkpoint. - expect(out.messages[0]!.content).toContainEqual({ type: 'text', text: 'SUMMARY' }) - expect(svc).toBeDefined() + // The re-derived head message is the framed summary checkpoint. + expect(session.deriveMessages()[0]!.content).toContainEqual({ type: 'text', text: 'SUMMARY' }) }) it('compacts mid-turn on steps after the first (the surface grows within a turn)', async () => { const { ctx } = await ctxWithModel('SUMMARY') - void new BasicCompactService(ctx, { contextWindow: 100, thresholdRatio: 0.5, retainTokens: 10 }) + void new BasicCompactService(ctx, { contextWindow: 100, thresholdRatio: 0.5, retainTokens: 10, summarizationMaxTokens: 30 }) const session = multiTurnSession(3, 1) // over the 0.5 threshold const agent = stubAgent(session, 'test-model') - const options: GenerateOptions = { model: 'test-model', messages: session.deriveMessages() } - // A step-2 request (a tool-heavy turn's later step) must still compact — the - // surface accumulated assistant/message + tool/result nodes since step 1. - await fireRequest(ctx, agent, 2, options) + // 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 firePreRequest(ctx, agent, 2, '', 'test-model') expect(session.events.some(e => e.type === 'compact/start')).toBe(true) }) - it('passes through unchanged when under threshold', async () => { + it('does nothing when under threshold', async () => { const { ctx } = await ctxWithModel('SUMMARY') void new BasicCompactService(ctx, { contextWindow: 128000, thresholdRatio: 0.8 }) const session = multiTurnSession(1, 1) const agent = stubAgent(session, 'test-model') - const msgs = session.deriveMessages() - const options: GenerateOptions = { model: 'test-model', messages: msgs } - const out = await fireRequest(ctx, agent, 1, options) - expect(out.messages).toBe(msgs) + await firePreRequest(ctx, agent, 1, '', 'test-model') expect(session.events.some(e => e.type === 'compact/start')).toBe(false) }) - it('proceeds with original history when compaction fails', async () => { - // No adapter registered for this model → summarize() rejects → caught, proceeds. + it('leaves the surface intact when compaction fails (summarize rejects)', async () => { + // No adapter registered for this model → summarize() rejects → caught, the + // surface is untouched (the loop derives the full history). const ctx = new Context() await ctx.plugin(LlmService) - void new BasicCompactService(ctx, { contextWindow: 100, thresholdRatio: 0.1, retainTokens: 10 }) + void new BasicCompactService(ctx, { contextWindow: 300, thresholdRatio: 0.1, retainTokens: 10, summarizationMaxTokens: 1 }) const session = multiTurnSession(3, 1) const agent = stubAgent(session, 'missing-model') - const msgs = session.deriveMessages() - const options: GenerateOptions = { model: 'missing-model', messages: msgs } + const before = session.surface.nodes.length - const out = await fireRequest(ctx, agent, 1, options) - // Listener swallowed the failure and left messages intact. - expect(out.messages).toBe(msgs) + await firePreRequest(ctx, agent, 1, '', 'missing-model') + // 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) }) it('does not register the listener when auto is false', async () => { const { ctx } = await ctxWithModel('SUMMARY') - void new BasicCompactService(ctx, { auto: false, contextWindow: 10, thresholdRatio: 0.1, retainTokens: 1 }) + void new BasicCompactService(ctx, { auto: false, contextWindow: 100, thresholdRatio: 0.1, retainTokens: 5, summarizationMaxTokens: 1 }) const session = multiTurnSession(3, 1) const agent = stubAgent(session, 'test-model') - const options: GenerateOptions = { model: 'test-model', messages: session.deriveMessages() } - await fireRequest(ctx, agent, 1, options) + await firePreRequest(ctx, agent, 1, '', 'test-model') expect(session.events.some(e => e.type === 'compact/start')).toBe(false) }) }) @@ -1049,46 +1048,65 @@ describe('BasicCompactService edge cases', () => { expect(svc.estimateContentTokens([unknown])).toBeGreaterThan(0) }) - it('compacts and re-derives without re-checking a post-compaction threshold', async () => { + it('compacts once without re-checking a post-compaction threshold', async () => { const { ctx } = await ctxWithModel('SUMMARY') // Even with a window so tiny the post-compaction history still exceeds the // threshold, the agnostic listener does NOT re-gate or warn — it compacts // once (the single check lives in compactIfNeeded) and proceeds. const warnings: string[] = [] ctx.logger.warn = ((msg: string) => void warnings.push(msg)) as typeof ctx.logger.warn - void new BasicCompactService(ctx, { contextWindow: 10, thresholdRatio: 0.1, retainTokens: 5 }) + void new BasicCompactService(ctx, { contextWindow: 300, thresholdRatio: 0.1, retainTokens: 5, summarizationMaxTokens: 5 }) const session = multiTurnSession(4, 1) const agent = stubAgent(session, 'test-model') - const options: GenerateOptions = { model: 'test-model', messages: session.deriveMessages() } - await ctx.waterfall('agent/request', agent, 1, 1, options, () => Promise.resolve(options)) + await ctx.parallel('agent/pre-request', agent, 1, 1, '', 'test-model', SIGNAL) expect(session.events.some(e => e.type === 'compact/summary')).toBe(true) - // The surface was re-derived into the request; no cascade warning is emitted. - expect(options.messages[0]!.content).toContainEqual({ type: 'text', text: 'SUMMARY' }) + // The surface was mutated; the head message is the framed summary checkpoint. + expect(session.deriveMessages()[0]!.content).toContainEqual({ type: 'text', text: 'SUMMARY' }) + // No cascade warning is emitted. expect(warnings.length).toBe(0) }) it('rejects compaction when no turn is open (compaction events must be turn-enclosed)', async () => { const svc = createTestService() - // A session with surface nodes but NO open turn — compaction's compact/* and - // replacement events would be appended outside any turn, which the session-log - // contract forbids. + // A session whose only turn has CLOSED — scanning back from the tail hits + // turn/end before any turn/start, so there is no open turn to enclose + // compaction's compact/* + replacement events, which the log contract forbids. const s = new Session(SessionId('noturn')) + s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + s.append('step/start', { turn: 1, step: 1 }) s.append('user/message', { content: [{ type: 'text', text: 'orphan' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) + s.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'reply' }] }, { surfaceOp: 'append' }) + s.append('step/end', { turn: 1, step: 1 }) + s.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) const nodes = s.surface.nodes - await expect(svc.compactRegion(s, nodes[0]!.seq, nodes[0]!.seq, 'm')) + await expect(svc.compactRegion(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) }) + it('rejects compaction on a session with no turn boundaries at all', async () => { + const svc = createTestService() + // No turn events whatsoever — the open-turn scan falls through to the end + // of the log and finds none, so compaction is rejected (its events have no + // turn to enclose them). + const s = new Session(SessionId('turnless')) + 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')) + .rejects.toThrow(/no open turn/) + expect(s.events.some(e => e.type === 'compact/start')).toBe(false) + }) + it('compactIfNeeded returns null for empty surface even when over threshold', async () => { - const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.1 }) + const svc = createTestService({ contextWindow: 1000, thresholdRatio: 0.1, retainTokens: 5 }) 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(400) // ceil(400/4) = 100 tokens >> threshold 10 - expect(await svc.compactIfNeeded(session, bigPrompt)).toBeNull() + const bigPrompt = 'x'.repeat(800) // ceil(800/4) = 200 tokens >> threshold 100 + expect(await svc.compactIfNeeded(session, bigPrompt, 'm', SIGNAL)).toBeNull() }) it('compactRegion throws when end is not a surface node (start valid)', async () => { @@ -1116,15 +1134,16 @@ describe('BasicCompactService edge cases', () => { const { ctx } = await ctxWithModel('SUMMARY') const warnings: string[] = [] ctx.logger.warn = ((msg: string) => void warnings.push(msg)) as typeof ctx.logger.warn - const svc = new TestCompactService(ctx, { contextWindow: 100, thresholdRatio: 0.1, retainTokens: 10 }) + const svc = new TestCompactService(ctx, { contextWindow: 300, thresholdRatio: 0.1, retainTokens: 10, summarizationMaxTokens: 10 }) svc.summarizeError = 'boom' as unknown as Error const session = multiTurnSession(3, 1) const agent = stubAgent(session, 'test-model') - const msgs = session.deriveMessages() - const options: GenerateOptions = { model: 'test-model', messages: msgs } + const before = session.surface.nodes.length - const out = await ctx.waterfall('agent/request', agent, 1, 1, options, () => Promise.resolve(options)) - expect(out.messages).toBe(msgs) // proceeded with original history + await ctx.parallel('agent/pre-request', agent, 1, 1, '', 'test-model', 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) expect(warnings.some(w => w.includes('compaction failed: boom'))).toBe(true) }) @@ -1132,16 +1151,14 @@ describe('BasicCompactService edge cases', () => { const { ctx } = await ctxWithModel('SUMMARY') // A large system prompt pushes the listener's estimate over threshold, but // retainTokens is huge so compactIfNeeded walks everything and returns null. - const svc = new TestCompactService(ctx, { contextWindow: 100, thresholdRatio: 0.1, retainTokens: 100000 }) + // threshold = floor(2000*0.1) = 200; invariant: 5 + 150 = 155 ≤ 200. + const svc = new TestCompactService(ctx, { contextWindow: 2000, thresholdRatio: 0.1, retainTokens: 150, summarizationMaxTokens: 5 }) const session = multiTurnSession(2, 1) const agent = stubAgent(session, 'test-model') - const bigSystem = 'x'.repeat(400) - const msgs = session.deriveMessages() - const options: GenerateOptions = { model: 'test-model', messages: msgs, system: bigSystem } + const bigSystem = 'x'.repeat(900) // ceil(900/4)=225 > threshold 200 - const out = await ctx.waterfall('agent/request', agent, 1, 1, options, () => Promise.resolve(options)) + await ctx.parallel('agent/pre-request', agent, 1, 1, bigSystem, 'test-model', SIGNAL) expect(session.events.some(e => e.type === 'compact/start')).toBe(false) - expect(out.messages).toBe(msgs) expect(svc.summarizeCalls.length).toBe(0) }) diff --git a/packages/compact/compact/README.md b/packages/compact/compact/README.md index 43737a4231..d75a5da774 100644 --- a/packages/compact/compact/README.md +++ b/packages/compact/compact/README.md @@ -10,7 +10,7 @@ This package is the interface tier of the compaction capability, split so each c | `@deepseek-ai/dsh-compact-basic` | a backend: char/4 estimation + token-budget retention + `llm.stream()` summarization | | `@deepseek-ai/dsh-tool-compact` (deferred) | the model-facing `/compact` tool over `ctx.compact` | -Unlike the bash seam, this interface depends on `@deepseek-ai/dsh-session` and `@deepseek-ai/dsh-llm` — the contract's verbs are defined over a `Session` and its output is the `ContentBlock` vocabulary, so they cannot be expressed without naming those packages. That deviation from the "interface depends only on cordis" guidance is intentional and recorded in the [compaction capability-seam RFC](../../../docs/rfc/proposed/feature/2026-06-18-compaction-capability-seam.md). +Unlike the bash seam, this interface depends on `@deepseek-ai/dsh-session` and `@deepseek-ai/dsh-llm` — the contract's verbs are defined over a `Session` and its output is the `ContentBlock` vocabulary, so they cannot be expressed without naming those packages. That deviation from the "interface depends only on cordis" guidance is intentional and recorded in the [compaction capability-seam RFC](../../../docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md). ## Service API (`ctx.compact`) @@ -18,10 +18,10 @@ Both methods are **abstract** — the backend owns the entire strategy (token es | Member | Semantics | |---|---| -| `compactIfNeeded(session, systemPrompt?, model?, signal?)` | Estimate the 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. | +| `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-request` 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. | -Both methods take an optional `signal: AbortSignal`. 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 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. ## Surface contract diff --git a/packages/compact/compact/src/index.ts b/packages/compact/compact/src/index.ts index 9ff7898468..5e63169fa1 100644 --- a/packages/compact/compact/src/index.ts +++ b/packages/compact/compact/src/index.ts @@ -16,7 +16,7 @@ * depends on `dsh-session` and `dsh-llm`: the contract's verbs are defined over * a `Session` and its output is the `ContentBlock` vocabulary. That deviation * from the "interface depends only on cordis" guidance is intentional and - * recorded in the [compaction capability-seam RFC](../../../../docs/rfc/proposed/feature/2026-06-18-compaction-capability-seam.md). + * recorded in the [compaction capability-seam RFC](../../../../docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md). * * @module @deepseek-ai/dsh-compact */ @@ -62,14 +62,31 @@ export abstract class CompactService extends Service { /** * Check token pressure and compact if the conversation is too large. * - * Estimates the current history size (optionally including a system prompt), - * and if it exceeds the backend's threshold, compacts an older range via - * {@link compactRegion}, keeping recent context intact. + * Estimates the current surface-derived history size (including the system + * prompt), and if it exceeds the backend's threshold, compacts an older range + * via {@link compactRegion}, keeping recent context intact. Returns `null` + * when no compaction is needed. + * + * Scope and guarantees a backend MUST honor: + * - **Surface-derived history only.** The decision is made against the history + * derived from the session surface — the only thing compaction can act on. + * Non-surface context injected downstream (into the request `messages` by a + * later listener) is out of this accounting by construction. + * - **Head-anchored, best-effort.** Auto-compaction consolidates from the + * surface HEAD up to a step-aligned cutoff, so a prior head checkpoint is + * re-summarized into one fresh checkpoint (the surface holds at most one + * auto-generated checkpoint, always at the head). It is best-effort over + * CLOSED steps: when the only compactable content left is an un-splittable + * open tail step, it declines (`null`) and retries once that step closes. + * - **Single-unit overflow is out of scope.** If a single retained unit (one + * closed step, or a large free node such as a 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. * * @param session - the session whose surface may be compacted. - * @param systemPrompt - optional system prompt, counted toward the estimate. - * @param model - optional summarization model (falls back to backend config). - * @param signal - optional cancellation signal. A backend that summarizes via + * @param system - the assembled system prompt, counted toward the estimate. + * @param model - the summarization model (a backend may override via config). + * @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 * leaving an orphaned model call running past the cancellation. @@ -77,9 +94,9 @@ export abstract class CompactService extends Service { */ abstract compactIfNeeded( session: Session, - systemPrompt?: string, - model?: string, - signal?: AbortSignal, + system: string, + model: string, + signal: AbortSignal, ): Promise /** diff --git a/packages/compact/compact/src/types.ts b/packages/compact/compact/src/types.ts index df001ff41a..ba886685d5 100644 --- a/packages/compact/compact/src/types.ts +++ b/packages/compact/compact/src/types.ts @@ -6,7 +6,7 @@ * events are log-only markers (lock + provenance); only the five * surface-eligible types can carry `surfaceOp`. The actual surface mutation is * performed by a separate `user/message` event carrying the summary (see the - * [compaction capability-seam RFC](../../../../docs/rfc/proposed/feature/2026-06-18-compaction-capability-seam.md)). + * [compaction capability-seam RFC](../../../../docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md)). * * Configuration lives in the backend, not here: the contract states WHAT * compaction produces, while every tunable (context window, thresholds, diff --git a/packages/core/agent-loop/src/loop.ts b/packages/core/agent-loop/src/loop.ts index ceef1bca8e..4aeef18eec 100644 --- a/packages/core/agent-loop/src/loop.ts +++ b/packages/core/agent-loop/src/loop.ts @@ -149,8 +149,9 @@ export interface LoopHandle { * drain steering → session('steering/message') ⟵ catches late steering * session('step/start'); emit agent/step-start ⟵ append before emit (the event-sourcing RFC) * assembly = ctx.systemPrompt.assemble() ⟵ waterfall system-prompt/assemble + * await ctx.parallel('agent/pre-request') ⟵ surface mutation (compaction) BEFORE derive * req = {model, system, tools, messages: session.deriveMessages(), signal} - * req = waterfall agent/request ⟵ hooks/compaction/model-switch + * req = waterfall agent/request ⟵ hooks/model-switch * stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks) * session('assistant/chunk'); emit agent/stream-chunk * msg = waterfall agent/step-result ⟵ BEFORE the log append, so the @@ -565,6 +566,13 @@ async function runStep( .filter(text => text.length > 0) .join('\n\n') + // Surface-mutation checkpoint BEFORE deriving history: compaction shadows an + // older range with a summary node here, and the single derive below reflects + // it. Awaited (no veto) — a listener mutates the surface as a side effect. + // `model` is resolved to '' when unset; a compaction listener that needs a + // model falls back to its own config. + await ctx.parallel('agent/pre-request', agent, turn, step, system, options.model ?? '', signal) + let request: GenerateOptions = { model: options.model ?? '', messages: session.deriveMessages(), diff --git a/packages/core/agent-loop/tests/loop.spec.ts b/packages/core/agent-loop/tests/loop.spec.ts index d018eff7a2..f74e160936 100644 --- a/packages/core/agent-loop/tests/loop.spec.ts +++ b/packages/core/agent-loop/tests/loop.spec.ts @@ -320,6 +320,65 @@ describe('agent loop', () => { expect(adapter.requests[0]!.model).toBe('other-model') }) + it('agent/pre-request fires once per step before the request is derived', async () => { + // Two steps (a tool call, then a final text turn) → two model calls → two + // pre-request fires, each carrying the assembled system + model, BEFORE the + // request messages are derived (the request the adapter sees reflects any + // surface state at fire time). + const adapter = new MockAdapter([ + toolCallResponse('c1', 'echo', {}, 'calling echo'), + textResponse('done'), + ]) + const ctx = await harness(adapter) + ctx.tools.register(defineTool({ + name: 'echo', description: 'echo', parameters: {}, + async execute() { return [{ type: 'text', text: 'echoed' }] }, + })) + const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' }) + + const fires: { turn: number; step: number; model: string }[] = [] + ctx.on('agent/pre-request', (subject, turn, step, _system, model) => { + if (subject === agent) fires.push({ turn, step, model }) + }) + + send(agent, 'go') + await waitForIdle(ctx, agent) + + // One fire per step, in order, each with the agent's model. + expect(fires).toEqual([ + { turn: 1, step: 1, model: 'mock' }, + { turn: 1, step: 2, model: 'mock' }, + ]) + }) + + it('a surface mutation in agent/pre-request is reflected in the derived request (single derive)', async () => { + // pre-request fires BEFORE deriveMessages(), so a listener that appends a + // surface node there sees it land in the SAME step's request — proving the + // loop derives once, after the checkpoint, with no stale pre-derive. + const adapter = new MockAdapter([textResponse('ok')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' }) + + let injected = false + ctx.on('agent/pre-request', (subject, turn) => { + if (subject === agent && !injected) { + injected = true + subject.session.append('context/message', { + content: [{ type: 'text', text: 'INJECTED-IN-PRE-REQUEST' }], + source: { kind: 'plugin', plugin: 'test' }, + }, { surfaceOp: 'append' }) + void turn + } + }) + + send(agent, 'go') + await waitForIdle(ctx, agent) + + // The adapter's request includes the node injected during pre-request. + const text = JSON.stringify(adapter.requests[0]!.messages) + expect(text).toContain('INJECTED-IN-PRE-REQUEST') + }) + it('cancel() mid-stream ends the turn with reason aborted', async () => { const adapter = new MockAdapter(['hang']) const ctx = await harness(adapter) diff --git a/packages/core/agent/src/types.ts b/packages/core/agent/src/types.ts index efe392155c..407201ea5d 100644 --- a/packages/core/agent/src/types.ts +++ b/packages/core/agent/src/types.ts @@ -180,10 +180,32 @@ declare module 'cordis' { 'agent/step-end'(agent: Agent, turn: number, step: number): void // ---- interception seams (waterfall) ---- + /** + * Awaited surface-mutation checkpoint, fired BEFORE the step's message + * history is derived (and thus before {@link agent/request}). The loop + * awaits `ctx.parallel('agent/pre-request', …)` after assembling the system + * prompt but before `session.deriveMessages()`, then derives 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), 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. + * + * Awaited (parallel), 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 deriving. `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). + * @mode parallel + */ + 'agent/pre-request'(agent: Agent, turn: number, step: number, system: string, model: string, signal: AbortSignal): Promise | void /** * Waterfall: mutate the fully-assembled {@link GenerateOptions} before the - * model call (hooks, compaction, model switching, tool filtering, …). Call - * `next()` to delegate, or return without it to short-circuit. + * model call (hooks, model switching, tool filtering, …). Call `next()` to + * delegate, or return without it to short-circuit. For surface mutation that + * must precede history derivation (compaction), use {@link agent/pre-request} + * instead — by the time this fires, `options.messages` is already derived. * @mode waterfall */ 'agent/request'(agent: Agent, turn: number, step: number, options: GenerateOptions, next: () => Promise): Promise diff --git a/packages/core/session/tests/surface.spec.ts b/packages/core/session/tests/surface.spec.ts index a7f4c13dad..cd30142773 100644 --- a/packages/core/session/tests/surface.spec.ts +++ b/packages/core/session/tests/surface.spec.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest' import type { SessionEvent, SurfaceEvent, SurfaceEventType } from '@deepseek-ai/dsh-session' -import { Session, SessionId } from '@deepseek-ai/dsh-session' +import { Session, SessionId, isSurfaceEvent } from '@deepseek-ai/dsh-session' import { CallId } from '@deepseek-ai/dsh-llm' /** Build a minimal session with turn boundaries and a single user message. */ @@ -278,4 +278,21 @@ describe('Session.append surface opts', () => { // The string 'append' is a primitive — identity-preserving is fine. expect(event.surfaceOp).toBe('append') }) + + it('isSurfaceEvent rejects a surface-eligible type missing its surfaceOp marker', () => { + // A raw event (not built via append, which mandates the marker) of a + // surface-eligible type but with no surfaceOp must NOT narrow to a + // SurfaceEvent — it would otherwise be silently dropped from the surface. + const noMarker: SessionEvent = { + type: 'user/message', seq: 0, time: 1, + data: { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, + } + expect(isSurfaceEvent(noMarker)).toBe(false) + // A non-surface type is rejected too (the type gate). + const boundary: SessionEvent = { type: 'turn/start', seq: 1, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } } + expect(isSurfaceEvent(boundary)).toBe(false) + // A properly-marked surface event narrows. + const marked = { ...noMarker, surfaceOp: 'append' } as SurfaceEvent + expect(isSurfaceEvent(marked)).toBe(true) + }) })