compact: summarize is a direct one-shot llm/stream call, not a loop step

The summarization request no longer dispatches agent/request — that seam
shapes the loop's conversation requests; a hand-built one-shot's
interception surface is llm/stream, like every other direct call. The
model comes from summarizationModel falling back to the agent's own. The
turn/step parameters existed on the compact seam only to feed the
agent/request dispatch and leave compactIfNeeded/compactRegion.

Groundwork for making every conversation request a pure function of the
session log (reconstructability RFC, this branch): the seam split keeps
the loop's request path as the single thing the log must explain.

Ported from worktree-prompt-prefix-stability (PR #162) where it was
review-converged; catalog and producer/consumer graphs regenerated.

packages/compact/compact-basic/README.md

@@ -11,7 +11,7 @@ The abstract contract states only WHAT compaction does; this backend owns every
 - **Token estimation** — `estimateContentTokens()`: chars divided by the `charsPerToken` config (default 4) with per-block structural overhead (`text`/`reasoning` = `ceil(len/charsPerToken) + 4`, `tool-call` from name + arguments, `tool-result` recursive, unknown blocks via JSON length).
 - **Retention policy** — `compactIfNeeded()` walks the surface nodes tail→head summing per-node token estimates, and retains the smallest tail-run of WHOLE units (a closed step, or a single no-step node such as a pre-step `user/message` or inter-step `steering/message`) whose total reaches `retainTokens`; everything older is compacted. Retention is **turn-agnostic** — turn boundaries play no role, so a single runaway turn that alone exceeds the window compacts its OWN early closed steps rather than being retained verbatim (the failure mode that motivated dropping turn-protection: a tool-heavy turn must stay compactable or the harness dies exactly when compaction is needed). The only structural guard is **tool-pairing balance**: the compacted region's edges are balanced cuts on the surface (no unanswered tool-call crosses either edge), so it never splits a step's `assistant/message` tool-calls from their `tool/result`s. When the only compactable content left is an un-splittable open tail step, it declines (returns `null`) and retries once an older step closes. **Single-unit overflow is out of scope, by design**: if one retained unit (a single closed step, or a large pasted `user/message`) ALONE exceeds the budget, compaction cannot help and the call may go out over-budget — bounding an individual unit's size is a separate concern. `compactRegion()` enforces tool-pairing balance strictly, throwing on a boundary that would split a step. `dsh-session` exports `isToolPairingBalanced` for the check.
 - **Dynamic convergence** — no static summary-length config pretends to bound what the model will write. If framing/estimator/system overhead leaves the compacted surface above threshold, `compactIfNeeded()` re-compacts the head checkpoint up to `compactionRetries` extra times; if it still cannot get below threshold, it throws. A summary whose estimated stored size is not smaller than the shadowed content fails closed before it mutates the surface.
-- **Summarization** — `summarize()`: a `GenerateOptions` request assembled via `BlockAssembler` with a fixed system prompt that asks for a structured checkpoint (Primary Request and Intent · Key Technical Concepts · Files and Code · Errors and Fixes · Pending Tasks · Current Work · Next Step · Critical Context), every section mandatory, exact paths/commands/identifiers preserved. The request runs through the `agent/request` waterfall before `ctx.llm.stream()`, so router agents that choose the concrete model there also route compaction summaries. `maxTokens` is the provider-side generation cap; only text blocks from the model's reply are kept before the checkpoint is stored (reasoning is dropped so private chain-of-thought never leaks into the durable summary, and a stray `tool-call` is dropped so the synthesized `user/message` summary cannot land an orphaned call with no matching `tool-result`). The compacted region is flattened to a plain-text transcript first: text and reasoning contribute their text, and every non-text block (tool-call, tool-result, plugin-added types) contributes a type-tagged placeholder (`[tool-call: name(args)]`, `[tool-result: …]`, …) so the summarizer is told what existed rather than silently dropping it.
+- **Summarization** — `summarize()`: a `GenerateOptions` request assembled via `BlockAssembler` with a fixed system prompt that asks for a structured checkpoint (Primary Request and Intent · Key Technical Concepts · Files and Code · Errors and Fixes · Pending Tasks · Current Work · Next Step · Critical Context), every section mandatory, exact paths/commands/identifiers preserved. The request is a direct one-shot `ctx.llm.stream()` call — NOT a loop step, so it does not run `agent/request` (that seam shapes the loop's conversation requests); the model comes from `summarizationModel` falling back to the agent's own, and per-call routing happens at `llm/stream` like any other direct call. `maxTokens` is the provider-side generation cap; only text blocks from the model's reply are kept before the checkpoint is stored (reasoning is dropped so private chain-of-thought never leaks into the durable summary, and a stray `tool-call` is dropped so the synthesized `user/message` summary cannot land an orphaned call with no matching `tool-result`). The compacted region is flattened to a plain-text transcript first: text and reasoning contribute their text, and every non-text block (tool-call, tool-result, plugin-added types) contributes a type-tagged placeholder (`[tool-call: name(args)]`, `[tool-result: …]`, …) 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 `<compacted-summary>…</compacted-summary>` tags. Because region compaction can be invoked manually, a surface may hold several checkpoints, so the framing does not claim everything after it is recent or verbatim. The tags make a prior checkpoint detectable in the transcript on the next compaction cycle: the summarization prompt then instructs the model to merge it in place (preserve still-true facts, drop stale ones) rather than re-summarize it verbatim — a cheap incremental merge that needs no extra log/event machinery. The unframed summary stays on the `compact/summary` provenance event.
 - **Surface mutation** — `compactRegion()` appends the `compact/start` → `compact/summary` → `compact/end` log records and the single `user/message` replace node carrying the framed summary (see the interface README).
 - **Auto-compaction** — an `agent/pre-step` listener delegates to `compactIfNeeded()` before every step (not just a turn's first — a tool-heavy turn grows the surface mid-turn, so a runaway turn still compacts, and per-step firing is the only moment to rescue it before overflow). `agent/pre-step` is a serial (awaited, in-order) surface-mutation checkpoint that fires after `turn/start` and BEFORE the step opens (`step/start`) and its request history is derived, so compaction mutates the surface — with its log-only `compact/*` records landing cleanly outside any step — and the loop derives once from the result: no double-derive, and the listener cannot see (or need to rewrite) an already-assembled `messages` array. The listener owns no threshold logic of its own (the single token-pressure check lives in `compactIfNeeded()`); because Cordis `serial` bails early on non-void return values, the listener returns `void` and does not use the dispatcher's bail channel as a veto surface.

packages/compact/compact-basic/src/index.ts

@@ -182,9 +182,9 @@ export class BasicCompactService extends CompactService {
       // log-only `compact/*` records and the replacement node cleanly outside a
       // step, so a crash mid-compaction leaves an inert orphan the turn-repair
       // closes — never a half-open step.
-      ctx.on('agent/pre-step', async (agent: Agent, turn: number, step: number, fullSystemPrompt: string, signal: AbortSignal) => {
+      ctx.on('agent/pre-step', async (agent: Agent, _turn: number, _step: number, fullSystemPrompt: string, signal: AbortSignal) => {
         try {
-          const result = await this.compactIfNeeded(agent, turn, step, fullSystemPrompt, signal)
+          const result = await this.compactIfNeeded(agent, fullSystemPrompt, signal)
           if (result) {
             const after = this.estimateTokens(agent.session.deriveMessages(), fullSystemPrompt)
             ctx.logger.info(
@@ -271,9 +271,13 @@ export class BasicCompactService extends CompactService {
   }
 
   /**
-   * Summarize conversation text into content blocks via `agent/request` plus
-   * `ctx.llm.stream()` assembled through a `BlockAssembler` (the single
-   * model-call surface).
+   * Summarize conversation text into content blocks via `ctx.llm.stream()`
+   * assembled through a `BlockAssembler`. A direct one-shot model call, NOT a
+   * loop step: it does not run the `agent/request` waterfall (that seam shapes
+   * the loop's conversation requests); per-call
+   * interception happens at `llm/stream` like any other direct call. The model
+   * comes from `BasicCompactConfig.summarizationModel`, falling back to the
+   * agent's own model.
    * Override in a subclass for a template or remote summarizer.
    *
    * Honors the adapter failure contract: an adapter may report a model failure
@@ -284,7 +288,7 @@ export class BasicCompactService extends CompactService {
    * Forwards `signal` into `GenerateOptions.signal` so an abort/dispose tears
    * down the in-flight summarization rather than orphaning the model call.
    */
-  async summarize(text: string, agent: Agent, turn: number, step: number, signal?: AbortSignal): Promise<ContentBlock[]> {
+  async summarize(text: string, agent: Agent, signal?: AbortSignal): Promise<ContentBlock[]> {
     const assembler = new BlockAssembler()
     const options: GenerateOptions = {
       model: this.config.summarizationModel || agent.options.model || '',
@@ -299,11 +303,10 @@ export class BasicCompactService extends CompactService {
     // exactOptionalPropertyTypes: only set `signal` when present — assigning
     // `undefined` to an optional `signal?: AbortSignal` is a type error.
     if (signal) options.signal = signal
-    const request = await this.ctx.waterfall('agent/request', agent, turn, step, options, () => Promise.resolve(options))
-    if (!request.model) {
-      throw new Error('no model available for summarization: set BasicCompactConfig.summarizationModel, AgentOptions.model, or supply one via the agent/request waterfall')
+    if (!options.model) {
+      throw new Error('no model available for summarization: set BasicCompactConfig.summarizationModel or AgentOptions.model')
     }
-    for await (const chunk of this.ctx.llm.stream(request)) {
+    for await (const chunk of this.ctx.llm.stream(options)) {
       assembler.push(chunk)
     }
 
@@ -348,8 +351,6 @@ export class BasicCompactService extends CompactService {
    */
   override async compactIfNeeded(
     agent: Agent,
-    turn: number,
-    step: number,
     fullSystemPrompt: string,
     signal: AbortSignal,
   ): Promise<CompactionResult | null> {
@@ -368,7 +369,7 @@ export class BasicCompactService extends CompactService {
         break
       }
 
-      result = await this.compactRegion(session, range.start, range.end, agent, turn, step, signal)
+      result = await this.compactRegion(session, range.start, range.end, agent, signal)
     }
 
     const totalTokens = this.estimateTokens(session.deriveMessages(), fullSystemPrompt)
@@ -385,8 +386,6 @@ export class BasicCompactService extends CompactService {
     start: number,
     end: number,
     agent: Agent,
-    turn: number,
-    step: number,
     signal?: AbortSignal,
   ): Promise<CompactionResult> {
     // Resolve the range by surface POSITION, not numeric seq interval. A prior
@@ -450,7 +449,7 @@ export class BasicCompactService extends CompactService {
     try {
       // --- Extract text and summarize ---
       const text = this._extractText(session, shadowedSeqs)
-      const summary = await this.summarize(text, agent, turn, step, signal)
+      const summary = await this.summarize(text, agent, signal)
 
       // Estimate token count of the shadowed content for provenance.
       let shadowedTokenCount = 0

packages/compact/compact-basic/tests/compact-basic.spec.ts

@@ -980,7 +980,7 @@ function compactIfNeeded(
   model: string,
   signal: AbortSignal,
 ) {
-  return svc.compactIfNeeded(stubAgent(session, model), 1, 1, fullSystemPrompt, signal)
+  return svc.compactIfNeeded(stubAgent(session, model), fullSystemPrompt, signal)
 }
 
 function compactRegion(
@@ -991,11 +991,11 @@ function compactRegion(
   model: string,
   signal?: AbortSignal,
 ) {
-  return svc.compactRegion(session, start, end, stubAgent(session, model), 1, 1, signal)
+  return svc.compactRegion(session, start, end, stubAgent(session, model), signal)
 }
 
 function summarize(svc: BasicCompactService, text: string, model: string) {
-  return svc.summarize(text, stubAgent(new Session(SessionId('summary')), model), 1, 1)
+  return svc.summarize(text, stubAgent(new Session(SessionId('summary')), model))
 }
 
 describe('BasicCompactService.summarize (real ctx.llm.stream)', () => {
@@ -1231,15 +1231,20 @@ describe('BasicCompactService auto-compaction (agent/pre-step listener)', () =>
     expect(session.events.some(e => e.type === 'compact/start')).toBe(false)
   })
 
-  it('routes summarization through agent/request so router agents can choose the model', async () => {
+  it('summarization is interceptable at llm/stream (model routing for direct calls)', async () => {
     const { ctx, adapter } = await ctxWithModel('ROUTED SUMMARY', 'routed-model')
-    ctx.on('agent/request', async (_agent, _turn, _step, options, next) => {
+    // The summarize call is a direct one-shot model call, not a loop step: it
+    // does not run agent/request (that seam shapes the loop's conversation
+    // requests). llm/stream is its interception surface, and a hand-built
+    // request is not frozen, so mutate-then-next model routing works — the
+    // adapter resolves AFTER the waterfall, so the rewrite picks the adapter.
+    ctx.on('llm/stream', (options, next) => {
       options.model = 'routed-model'
       return next()
     })
     void new BasicCompactService(ctx, cfg({ contextWindow: 200, thresholdRatio: 0.5, retainTokens: 20 }))
     const session = multiTurnSession(5, 1)
-    const agent = stubAgent(session)
+    const agent = stubAgent(session, 'agent-model')
 
     await ctx.serial('agent/pre-step', agent, 1, 1, '', SIGNAL)
 

packages/compact/compact/README.md

@@ -18,8 +18,8 @@ Both methods are **abstract** — the backend owns the entire strategy (token es
 
 | Member | Semantics |
 |---|---|
-| `compactIfNeeded(agent, turn, step, fullSystemPrompt, signal)` | Estimate the surface-derived history size; if over the backend's threshold, compact an older range via `compactRegion`, keeping recent context intact. Returns the `CompactionResult`, or `null` if nothing needed compacting. All parameters required — the loop's `agent/pre-step` checkpoint supplies the agent, lifecycle context, assembled `fullSystemPrompt`, and turn `signal`; router-aware summarizers can use the agent lifecycle context to route their own model call through `agent/request`. |
-| `compactRegion(session, start, end, agent, turn, step, signal?)` | Forcibly summarize surface nodes `[start, end]` (inclusive seqs) into a single replacement node. **Throws** if a compaction is already in progress, if `start`/`end` aren't surface nodes, or if `start` is positioned after `end` on the surface. The range is a SURFACE-POSITION span, not a numeric seq interval — after a prior replace lands a fresh high-seq summary node at the shadowed range's position, surface order no longer tracks seq order. |
+| `compactIfNeeded(agent, fullSystemPrompt, signal)` | Estimate the surface-derived history size; if over the backend's threshold, compact an older range via `compactRegion`, keeping recent context intact. Returns the `CompactionResult`, or `null` if nothing needed compacting. All parameters required — the loop's `agent/pre-step` checkpoint supplies the agent, assembled `fullSystemPrompt`, and turn `signal`. A backend's summarization request is a direct `ctx.llm.stream()` call (not a loop step), so per-call interception happens at `llm/stream`. |
+| `compactRegion(session, start, end, agent, signal?)` | Forcibly summarize surface nodes `[start, end]` (inclusive seqs) into a single replacement node. **Throws** if a compaction is already in progress, if `start`/`end` aren't surface nodes, or if `start` is positioned after `end` on the surface. The range is a SURFACE-POSITION span, not a numeric seq interval — after a prior replace lands a fresh high-seq summary node at the shadowed range's position, surface order no longer tracks seq order. |
 
 `compactIfNeeded` takes a required `signal`; `compactRegion`'s is optional. A backend that summarizes via `ctx.llm.stream()` **must** forward it into the call's `GenerateOptions.signal`, so an abort or fiber dispose tears down the in-flight summarization instead of leaving an orphaned model call running past the cancellation. The session being compacted comes from the agent context; the turn that the `compact/*` events belong to is recoverable from the log (the currently-open turn), so the backend stamps it from the log rather than trusting a caller-supplied value.
 

packages/compact/compact/src/index.ts

@@ -91,8 +91,6 @@ export abstract class CompactService extends Service {
    *   over-budget. Bounding an individual unit's size is a separate concern.
    *
    * @param agent - agent context owning the session surface and model options.
-   * @param turn - turn number of the pre-step checkpoint.
-   * @param step - step number about to start.
    * @param fullSystemPrompt - assembled system prompt, counted toward the estimate.
    * @param signal - cancellation signal. A backend summarizing via
    *   `ctx.llm.stream()` MUST forward this into the call's `GenerateOptions.signal`
@@ -102,8 +100,6 @@ export abstract class CompactService extends Service {
    */
   abstract compactIfNeeded(
     agent: CompactAgentContext,
-    turn: number,
-    step: number,
     fullSystemPrompt: string,
     signal: AbortSignal,
   ): Promise<CompactionResult | null>
@@ -129,8 +125,6 @@ export abstract class CompactService extends Service {
    * @param start - inclusive seq of the first surface node to compact.
    * @param end - inclusive seq of the last surface node to compact.
    * @param agent - agent context used by router-aware summarizers.
-   * @param turn - lifecycle turn forwarded to request-routing seams.
-   * @param step - lifecycle step forwarded to request-routing seams.
    * @param signal - optional cancellation signal. A backend that summarizes via
    *   `ctx.llm.stream()` MUST forward this into the call's `GenerateOptions.signal`
    *   so an abort/dispose tears down the in-flight summarization rather than
@@ -148,8 +142,6 @@ export abstract class CompactService extends Service {
     start: number,
     end: number,
     agent: CompactAgentContext,
-    turn: number,
-    step: number,
     signal?: AbortSignal,
   ): Promise<CompactionResult>
 }

packages/compact/compact/tests/compact.spec.ts

@@ -17,8 +17,6 @@ class StubCompactService extends CompactService {
 
   override async compactIfNeeded(
     _agent: CompactAgentContext,
-    _turn: number,
-    _step: number,
     _fullSystemPrompt: string,
     signal: AbortSignal,
   ): Promise<CompactionResult | null> {
@@ -31,8 +29,6 @@ class StubCompactService extends CompactService {
     start: number,
     end: number,
     _agent: CompactAgentContext,
-    _turn: number,
-    _step: number,
     signal?: AbortSignal,
   ): Promise<CompactionResult> {
     this.lastSignal = signal
@@ -81,7 +77,7 @@ describe('CompactService seam', () => {
     const ctx = new Context()
     const svc = new StubCompactService(ctx)
     const session = new Session(SessionId('s'))
-    expect(await svc.compactIfNeeded(stubAgent(session), 1, 1, '', new AbortController().signal)).toBeNull()
+    expect(await svc.compactIfNeeded(stubAgent(session), '', new AbortController().signal)).toBeNull()
   })
 
   it('compact/* events merge into SessionEventMap and are log-only', async () => {
@@ -89,7 +85,7 @@ describe('CompactService seam', () => {
     const svc = new StubCompactService(ctx)
     const session = new Session(SessionId('s'))
 
-    const result = await svc.compactRegion(session, 0, 0, stubAgent(session, 'm'), 1, 1)
+    const result = await svc.compactRegion(session, 0, 0, stubAgent(session, 'm'))
 
     const startEvent = session.events.find(e => e.type === 'compact/start')
     expect(startEvent).toBeDefined()
@@ -107,10 +103,10 @@ describe('CompactService seam', () => {
     const session = new Session(SessionId('s'))
     const controller = new AbortController()
 
-    await svc.compactRegion(session, 0, 0, stubAgent(session, 'm'), 1, 1, controller.signal)
+    await svc.compactRegion(session, 0, 0, stubAgent(session, 'm'), controller.signal)
     expect(svc.lastSignal).toBe(controller.signal)
 
-    await svc.compactIfNeeded(stubAgent(session), 1, 1, '', controller.signal)
+    await svc.compactIfNeeded(stubAgent(session), '', controller.signal)
     expect(svc.lastSignal).toBe(controller.signal)
   })
 })