From cc546c4580721a78c0276ffa8723523a74679e6a Mon Sep 17 00:00:00 2001 From: Hypatia May Date: Wed, 15 Jul 2026 13:22:44 +0800 Subject: [PATCH 1/2] feat(compact): move pairing helpers (PR1 round 1) --- docs/cordis-catalog/events.md | 8 +- docs/cordis-catalog/services.md | 4 +- docs/core-data-structures/compaction.md | 2 + docs/core-data-structures/session.md | 2 + docs/event-producer-consumer.md | 8 +- .../2026-06-18-compaction-capability-seam.md | 4 +- packages/compact/compact-basic/README.md | 2 +- packages/compact/compact-basic/src/index.ts | 16 +- .../tests/compact-loop-repro.spec.ts | 6 +- packages/compact/compact/README.md | 8 +- packages/compact/compact/src/index.ts | 3 + packages/compact/compact/src/tool-pairing.ts | 160 +++++++++ .../compact/tests/tool-pairing.spec.ts | 327 ++++++++++++++++++ packages/core/session/README.md | 2 +- packages/core/session/src/index.ts | 1 - packages/core/session/src/tool-pairing.ts | 56 --- .../core/session/tests/tool-pairing.spec.ts | 292 ---------------- 17 files changed, 525 insertions(+), 376 deletions(-) create mode 100644 packages/compact/compact/src/tool-pairing.ts create mode 100644 packages/compact/compact/tests/tool-pairing.spec.ts delete mode 100644 packages/core/session/src/tool-pairing.ts delete mode 100644 packages/core/session/tests/tool-pairing.spec.ts diff --git a/docs/cordis-catalog/events.md b/docs/cordis-catalog/events.md index 4af431f8c0..d2102e5e3f 100644 --- a/docs/cordis-catalog/events.md +++ b/docs/cordis-catalog/events.md @@ -245,7 +245,7 @@ Creation announcement during session publication. A synchronous throw vetoes and 'session/created'(this: Scoped, session: Session): void ``` -Source: [`packages/core/session/src/index.ts:47`](../../packages/core/session/src/index.ts) +Source: [`packages/core/session/src/index.ts:46`](../../packages/core/session/src/index.ts) ### `session/disposed` — emit @@ -255,7 +255,7 @@ Emitted once when an announced session leaves the store, including publication r 'session/disposed'(this: Scoped, session: Session): void ``` -Source: [`packages/core/session/src/index.ts:57`](../../packages/core/session/src/index.ts) +Source: [`packages/core/session/src/index.ts:56`](../../packages/core/session/src/index.ts) ### `session/event` — emit @@ -267,7 +267,7 @@ Post-commit, fire-and-forget append feed. The listener snapshot resolves before Types: [SessionEvent](../core-data-structures/core.md) -Source: [`packages/core/session/src/index.ts:69`](../../packages/core/session/src/index.ts) +Source: [`packages/core/session/src/index.ts:68`](../../packages/core/session/src/index.ts) ### `session/flush` — parallel @@ -277,7 +277,7 @@ Awaited parallel durability checkpoint: every listener runs and the caller await 'session/flush'(this: Scoped, session: Session): Promise | void ``` -Source: [`packages/core/session/src/index.ts:79`](../../packages/core/session/src/index.ts) +Source: [`packages/core/session/src/index.ts:78`](../../packages/core/session/src/index.ts) ## `skill/*` diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index 1509afc76a..1197570d23 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -95,7 +95,7 @@ abstract compactRegion( session: Session, start: number, end: number, agent: Com Types: [Message](../core-data-structures/core.md) -Source: [`packages/compact/compact/src/index.ts:36`](../../packages/compact/compact/src/index.ts) +Source: [`packages/compact/compact/src/index.ts:37`](../../packages/compact/compact/src/index.ts) ## `ctx.fs` — `FileSystem` (abstract seam) @@ -200,7 +200,7 @@ list(): Session[] fork(source: SessionForkSource, boundary?: number, childSessionId?: SessionId): Session ``` -Source: [`packages/core/session/src/index.ts:564`](../../packages/core/session/src/index.ts) +Source: [`packages/core/session/src/index.ts:563`](../../packages/core/session/src/index.ts) ## `ctx.skills` — `SkillService` diff --git a/docs/core-data-structures/compaction.md b/docs/core-data-structures/compaction.md index 82bf512eeb..05022f9937 100644 --- a/docs/core-data-structures/compaction.md +++ b/docs/core-data-structures/compaction.md @@ -53,3 +53,5 @@ interface CompactionResult { `CompactService` exposes `compactIfNeeded(...)` for pressure-triggered compaction, returning `null` when no compaction is needed, and `compactRegion(...)` for an explicit inclusive surface range. The pre-step caller supplies the agent, full prompt, session prefix, and abort signal; implementations must forward that signal to summarization. Estimation, retention, event sequencing, and summarization remain backend policy. Auto-compaction runs at serial `agent/pre-step`, before the step and request derivation, so it can replace surface nodes while keeping trace events outside the step. Region boundaries preserve tool-call/result pairing but do not preserve whole turns, allowing early closed steps of one oversized turn to compact. `dsh-compact-basic` owns the retention and failure details. + +The seam exports `toolPairingBalancedBefore(session, node)` and `toolPairingBalancedAfter(session, node)` for those edge checks. Both validate current surface membership, reject stale or missing seqs and orphan results, and ignore a caller-retained `node.next`; the [package contract](../../packages/compact/compact/README.md#tool-pairing-boundaries) owns their cache semantics. diff --git a/docs/core-data-structures/session.md b/docs/core-data-structures/session.md index d8292c49b3..174f06602e 100644 --- a/docs/core-data-structures/session.md +++ b/docs/core-data-structures/session.md @@ -196,6 +196,8 @@ export interface SurfaceNode { } ``` +`SurfaceNode` is positional state, not durable identity. A replacement can remove a caller-retained node or make a copied `next` stale; consumers that cross a surface mutation validate membership and resolve successors from `Session.surface.nodes`. `SurfaceManager.replaceGeneration` increments for each replacement so incremental consumers can distinguish pure tail growth from a rewrite. + ### `SurfaceFoldReplacement` and `SurfaceFoldResult` — a complete surface replay `foldSurface(events)` returns detached current nodes together with the actual node seqs shadowed by each declared replacement range. `SurfaceManager` uses the same transition functions for its incremental cache. diff --git a/docs/event-producer-consumer.md b/docs/event-producer-consumer.md index bdf1c6e320..445078cb37 100644 --- a/docs/event-producer-consumer.md +++ b/docs/event-producer-consumer.md @@ -25,10 +25,10 @@ This matrix shows which packages dispatch each harness-owned event and which pac | `fs/observed` | `emit` | [`packages/fs/fs/src/index.ts:68`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`emit`) | [`fs-policy`](../packages/fs/fs-policy) | | `fs/write-intent` | `waterfall` | [`packages/fs/fs/src/index.ts:51`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`waterfall`) | [`fs-policy`](../packages/fs/fs-policy) | | `llm/stream` | `waterfall` | [`packages/llm/llm/src/index.ts:39`](../packages/llm/llm/src/index.ts) | [`llm`](../packages/llm/llm) (`waterfall`) | [`invariants`](../packages/support/invariants), [`llm-replay`](../packages/support/llm-replay) | -| `session/created` | `emit` | [`packages/core/session/src/index.ts:47`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`invariants`](../packages/support/invariants), [`jsonrpc`](../packages/ui/jsonrpc), [`session-persistence`](../packages/session-persistence/session-persistence) | -| `session/disposed` | `emit` | [`packages/core/session/src/index.ts:57`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | - | -| `session/event` | `emit` | [`packages/core/session/src/index.ts:69`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`acp`](../packages/ui/acp), [`invariants`](../packages/support/invariants), [`jsonrpc`](../packages/ui/jsonrpc), [`session-persistence`](../packages/session-persistence/session-persistence), [`stdio-agent`](../packages/ui/stdio-agent) | -| `session/flush` | `parallel` | [`packages/core/session/src/index.ts:79`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`session-persistence`](../packages/session-persistence/session-persistence) | +| `session/created` | `emit` | [`packages/core/session/src/index.ts:46`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`invariants`](../packages/support/invariants), [`jsonrpc`](../packages/ui/jsonrpc), [`session-persistence`](../packages/session-persistence/session-persistence) | +| `session/disposed` | `emit` | [`packages/core/session/src/index.ts:56`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | - | +| `session/event` | `emit` | [`packages/core/session/src/index.ts:68`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`acp`](../packages/ui/acp), [`invariants`](../packages/support/invariants), [`jsonrpc`](../packages/ui/jsonrpc), [`session-persistence`](../packages/session-persistence/session-persistence), [`stdio-agent`](../packages/ui/stdio-agent) | +| `session/flush` | `parallel` | [`packages/core/session/src/index.ts:78`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`session-persistence`](../packages/session-persistence/session-persistence) | | `skill/provider-added` | `emit` | [`packages/skill/skill/src/index.ts:131`](../packages/skill/skill/src/index.ts) | [`skill`](../packages/skill/skill) (`emit`) | - | | `skill/provider-removed` | `emit` | [`packages/skill/skill/src/index.ts:137`](../packages/skill/skill/src/index.ts) | [`skill`](../packages/skill/skill) (`emit`) | - | | `subagent/end` | `emit` | [`packages/subagent/subagent/src/index.ts:92`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`jsonrpc`](../packages/ui/jsonrpc) | diff --git a/docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md b/docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md index 4a6586b159..29e3347759 100644 --- a/docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md +++ b/docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md @@ -52,7 +52,7 @@ The loop derives messages once after `agent/pre-step`. Running before `step/star Auto-compaction fires before **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-step` checkpoint. 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. -`compactIfNeeded` retains the smallest tail of whole surface units whose estimated size reaches `retainTokens` and compacts older nodes. A unit is a complete closed step or one no-step message. If the token cutoff lands inside a step, retention expands until the cut is tool-pairing balanced. Balance is checked on surface order, not log sequence, because replacement summaries have new sequence numbers at old surface positions. `compactRegion` rejects boundaries that split a tool call from its result. The in-flight turn receives no special retention. +`compactIfNeeded` retains the smallest tail of whole surface units whose estimated size reaches `retainTokens` and compacts older nodes. A unit is a complete closed step or one no-step message. If the token cutoff lands inside a step, retention expands until the cut is tool-pairing balanced. Balance is checked on surface order, not log sequence, because replacement summaries have new sequence numbers at old surface positions. `dsh-compact` exports the before/after edge helpers; their per-session cache folds only appended surface-tail nodes while `replaceGeneration` is unchanged, does no event reads for log-only growth, and rebuilds current membership, positional successors, and balances after replacement. `compactRegion` rejects boundaries that split a tool call from its result. The in-flight turn receives no special retention. 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. @@ -113,7 +113,7 @@ Two failure paths, both documented: - **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-step` (`@mode serial`) declared in `dsh-agent` and emitted by `dsh-agent-loop` after system assembly and before `step/start`. 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. -- **`dsh-session`** gains the tool-pairing balance predicate (`isToolPairingBalanced`, in `tool-pairing.ts`, exported from the package index) that `compactRegion`/`compactIfNeeded` use to keep a collapsed region from splitting a step's tool-call/result pair. The surface `replace` op and the surface-metadata runtime guard already existed and are reused. +- **`dsh-compact`** owns `toolPairingBalancedBefore(session, node)` and `toolPairingBalancedAfter(session, node)`, the cached surface-edge checks that `compactRegion` and `compactIfNeeded` use to avoid splitting a tool-call/result pair. The cache validates current membership by seq and resolves after-edges from its positional successor map instead of trusting a caller-retained `node.next`; stale or missing seqs and orphan results reject. `dsh-session` continues to own the surface `replace` operation, positional nodes, and rewrite generation. - **`dsh-invariants`** drops its `surface replace: start must be <= end` assertion: a head-anchored compaction lands a high-seq replacement node at an older range's *position*, so `start > end` numerically is normal and valid (the range is positional, validated by the surface's `indexOf` checks that remain). The turn-enclosure invariant is reused unchanged. - **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). diff --git a/packages/compact/compact-basic/README.md b/packages/compact/compact-basic/README.md index d9c0f472a8..d05ce47356 100644 --- a/packages/compact/compact-basic/README.md +++ b/packages/compact/compact-basic/README.md @@ -9,7 +9,7 @@ This is the implementation tier of the compaction capability — see the [interf This backend owns the compaction policy: - **Estimation** — a configurable characters-per-token heuristic counts the current session prefix supplied to pre-step, derived history, and system prompt, matching the next request rather than stale logged prefix state. -- **Retention** — compact the oldest whole surface units while preserving a recent tail and balanced tool-call/result cuts. Turn boundaries do not protect old steps inside a runaway turn. An open indivisible tail declines until it closes; a single unit larger than the budget remains out of scope. +- **Retention** — compact the oldest whole surface units while preserving a recent tail and balanced tool-call/result cuts through the [`dsh-compact` boundary helpers](../compact/README.md#tool-pairing-boundaries). Turn boundaries do not protect old steps inside a runaway turn. An open indivisible tail declines until it closes; a single unit larger than the budget remains out of scope. - **Convergence** — retry head-checkpoint compaction up to `compactionRetries`; reject a summary that does not shrink its source, and throw if retries cannot return below threshold. - **Summarization** — a direct `llm/stream` call uses the configured model and cap without running the loop-only `agent/request` seam. The input transcript preserves non-text blocks as tagged placeholders; only returned text enters the checkpoint, excluding reasoning and tool calls that would leak private reasoning or create an orphaned call. - **Framing** — the replacement user message marks established checkpoint context with `` tags. The raw summary remains on the provenance event, and later automatic cycles merge the prior checkpoint. diff --git a/packages/compact/compact-basic/src/index.ts b/packages/compact/compact-basic/src/index.ts index bf349f7f73..fffc8e9a63 100644 --- a/packages/compact/compact-basic/src/index.ts +++ b/packages/compact/compact-basic/src/index.ts @@ -7,12 +7,11 @@ */ import { Context } from 'cordis' -import { CompactService, renderTranscript } from '@deepseek-ai/dsh-compact' +import { CompactService, renderTranscript, toolPairingBalancedAfter, toolPairingBalancedBefore } 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 } from '@deepseek-ai/dsh-session' -import { isToolPairingBalanced } from '@deepseek-ai/dsh-session' import type { Agent } from '@deepseek-ai/dsh-agent' import type { BasicCompactConfig, ResolvedConfig } from './types.ts' import { resolveConfig } from './types.ts' @@ -348,15 +347,14 @@ export class BasicCompactService extends CompactService { } // Both range edges must preserve assistant tool-call/result pairing. - const events = session.events - if (!isToolPairingBalanced(nodes, events, start)) { + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + const startNode = nodes[startIdx]! + if (!toolPairingBalancedBefore(session, startNode)) { throw new Error(`compactRegion: start seq ${start} is not a balanced boundary (would split a step's tool-call/result pair)`) } - // The cut after `end` is named by `end`'s surface successor, or `null` when - // `end` is the tail. // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - const afterEnd: number | null = nodes[endIdx]!.next - if (!isToolPairingBalanced(nodes, events, afterEnd)) { + const endNode = nodes[endIdx]! + if (!toolPairingBalancedAfter(session, endNode)) { throw new Error(`compactRegion: end seq ${end} is not a balanced boundary (would split a step, or the step is still open)`) } @@ -511,7 +509,7 @@ export class BasicCompactService extends CompactService { // splitting an assistant↔result pair. while (keepFromIdx > 0) { // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - if (isToolPairingBalanced(nodes, events, nodes[keepFromIdx]!.seq)) break + if (toolPairingBalancedBefore(session, nodes[keepFromIdx]!)) break keepFromIdx -= 1 } if (keepFromIdx === 0) return null diff --git a/packages/compact/compact-basic/tests/compact-loop-repro.spec.ts b/packages/compact/compact-basic/tests/compact-loop-repro.spec.ts index dbf8a3b737..49fb99acec 100644 --- a/packages/compact/compact-basic/tests/compact-loop-repro.spec.ts +++ b/packages/compact/compact-basic/tests/compact-loop-repro.spec.ts @@ -1,10 +1,10 @@ import { describe, expect, it } from 'vitest' import { Context } from 'cordis' +import { toolPairingBalancedAfter, toolPairingBalancedBefore } from '@deepseek-ai/dsh-compact' import LlmService from '@deepseek-ai/dsh-llm' import type { ContentBlock, GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm' import { CallId, LlmAdapter } from '@deepseek-ai/dsh-llm' import SessionStore from '@deepseek-ai/dsh-session' -import { isToolPairingBalanced } from '@deepseek-ai/dsh-session' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools' import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent' @@ -124,9 +124,9 @@ describe('CBR-001: a real-loop checkpoint is a valid boundary on both sides', () for (const cp of checkpoints) { const node = nodes.find(n => n.seq === cp.seq) if (!node) continue // shadowed by a later checkpoint — no longer an edge. - expect(isToolPairingBalanced(nodes, events, node.seq), + expect(toolPairingBalancedBefore(agent.session, node), `checkpoint seq ${node.seq} must be a balanced region START`).toBe(true) - expect(isToolPairingBalanced(nodes, events, node.next), + expect(toolPairingBalancedAfter(agent.session, node), `checkpoint seq ${node.seq} must be a balanced region END`).toBe(true) } } finally { diff --git a/packages/compact/compact/README.md b/packages/compact/compact/README.md index 9141012281..5b59893561 100644 --- a/packages/compact/compact/README.md +++ b/packages/compact/compact/README.md @@ -6,7 +6,7 @@ This package is the interface tier of the compaction capability, split so each c | Package | Role | |---|---| -| `@deepseek-ai/dsh-compact` (this) | the interface: abstract service + `compact/*` events + `CompactionResult` + the shared transcript renderer (`renderTranscript`/`renderContentBlocks`) | +| `@deepseek-ai/dsh-compact` (this) | the interface: abstract service + `compact/*` events + `CompactionResult` + tool-pairing boundary helpers + the shared transcript renderer (`renderTranscript`/`renderContentBlocks`) | | `@deepseek-ai/dsh-compact-basic` | a backend: chars-per-token estimation (`charsPerToken`, default 4) + token-budget retention + `llm.stream()` summarization | | `@deepseek-ai/dsh-tool-compact` (deferred) | the model-facing `/compact` tool over `ctx.compact` | @@ -23,6 +23,12 @@ Both methods are **abstract** — the backend owns the entire strategy (token es `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. +## Tool-pairing boundaries + +The interface exports `toolPairingBalancedBefore(session, node)` and `toolPairingBalancedAfter(session, node)` for snapping and validating compaction edges. A safe edge has no unanswered assistant tool call crossing it. Each helper validates the node's seq against current surface membership and resolves the trailing edge from its cached positional successor, so a stale caller-held `node.next` cannot choose the cut. + +The private per-session cache is keyed by `session.surface.replaceGeneration` and the processed surface-node count. An unchanged generation extends the fold with unseen tail nodes only; a log-only append with no new surface node does no event reads, while a replacement generation rebuilds current membership, successors, and balances. Missing event seqs and a `tool/result` without a preceding open call reject as corrupt surface state. + ## Surface contract `SurfaceEventType` is a closed union — only `user/message`, `assistant/message`, `tool/result`, `context/message`, and `steering/message` may carry `surfaceOp`. A `compact/*` event therefore **cannot** appear on the surface. A successful compaction instead: diff --git a/packages/compact/compact/src/index.ts b/packages/compact/compact/src/index.ts index f59e0dc328..12eb77b362 100644 --- a/packages/compact/compact/src/index.ts +++ b/packages/compact/compact/src/index.ts @@ -14,6 +14,7 @@ import type { CompactionResult } from './types.ts' export type { CompactionResult } from './types.ts' export { renderContentBlocks, renderTranscript } from './render.ts' +export { toolPairingBalancedAfter, toolPairingBalancedBefore } from './tool-pairing.ts' /** Minimal agent context compaction needs without depending on the agent package. */ export interface CompactAgentContext { @@ -67,6 +68,8 @@ export abstract class CompactService extends Service { * balanced so assistant tool calls remain paired with their results. A model- * backed implementation forwards cancellation and rejects active, missing, * reversed, or unbalanced ranges. + * Use {@link toolPairingBalancedBefore} and {@link toolPairingBalancedAfter} + * for the edge checks. * * @param session - session to mutate. * @param start - first surface seq, inclusive. diff --git a/packages/compact/compact/src/tool-pairing.ts b/packages/compact/compact/src/tool-pairing.ts new file mode 100644 index 0000000000..a9fca01bf6 --- /dev/null +++ b/packages/compact/compact/src/tool-pairing.ts @@ -0,0 +1,160 @@ +/** + * Tool-pairing balance over a session surface. Compaction changes surface + * positions, so safe cuts are derived from tool-call/result content in current + * surface order rather than step markers or linked-list fields supplied by a + * caller. + * @module @deepseek-ai/dsh-compact/tool-pairing + */ + +import type { Session, SessionEvent, SurfaceNode } from '@deepseek-ai/dsh-session' + +/** Incremental balance state for one session surface generation. */ +interface BalanceCache { + /** Surface rewrite generation this state describes. */ + generation: number + /** Number of surface nodes already folded into the state. */ + processedNodes: number + /** Balance of the cut immediately before each current surface node. */ + beforeSeq: Map + /** Current positional successor of each surface node. */ + successorBySeq: Map + /** Unanswered tool-call count after the processed surface tail. */ + depth: number +} + +const balanceCacheBySession = new WeakMap() + +/** Return how one surface event changes the unanswered tool-call count. */ +function nodeDelta(event: SessionEvent): number { + switch (event.type) { + case 'assistant/message': + return event.data.content.filter(block => block.type === 'tool-call').length + case 'tool/result': + return -1 + default: + return 0 + } +} + +/** Read and validate the event named by a surface node. */ +function eventForNode(events: readonly SessionEvent[], node: SurfaceNode): SessionEvent { + const event = events[node.seq] + if (event === undefined || event.seq !== node.seq) { + throw new Error(`tool-pairing balance: surface seq ${node.seq} has no matching session event (corrupt surface)`) + } + return event +} + +/** Build balance state for a complete current surface. */ +function rebuildCache( + session: Session, + nodes: readonly SurfaceNode[], + generation: number, +): BalanceCache { + const beforeSeq = new Map() + const successorBySeq = new Map() + const events = session.events + let depth = 0 + let previousSeq: number | undefined + + for (const node of nodes) { + beforeSeq.set(node.seq, depth === 0) + successorBySeq.set(node.seq, null) + if (previousSeq !== undefined) successorBySeq.set(previousSeq, node.seq) + depth += nodeDelta(eventForNode(events, node)) + if (depth < 0) { + throw new Error(`tool-pairing balance: tool/result at surface seq ${node.seq} has no matching tool-call (corrupt surface)`) + } + previousSeq = node.seq + } + + return { generation, processedNodes: nodes.length, beforeSeq, successorBySeq, depth } +} + +/** Fold a pure surface tail append into existing balance state. */ +function extendCache( + session: Session, + cache: BalanceCache, + nodes: readonly SurfaceNode[], +): BalanceCache { + const tail = nodes.slice(cache.processedNodes) + // Validate the unseen tail before mutating the live cache, so a corrupt + // append cannot leave a partially advanced state behind. + const events = session.events + const pending: Array<{ seq: number; before: boolean }> = [] + let depth = cache.depth + for (const node of tail) { + pending.push({ seq: node.seq, before: depth === 0 }) + depth += nodeDelta(eventForNode(events, node)) + if (depth < 0) { + throw new Error(`tool-pairing balance: tool/result at surface seq ${node.seq} has no matching tool-call (corrupt surface)`) + } + } + + let previousSeq = nodes[cache.processedNodes - 1]?.seq + for (const entry of pending) { + if (previousSeq !== undefined) cache.successorBySeq.set(previousSeq, entry.seq) + cache.beforeSeq.set(entry.seq, entry.before) + cache.successorBySeq.set(entry.seq, null) + previousSeq = entry.seq + } + cache.processedNodes = nodes.length + cache.depth = depth + return cache +} + +/** Return balance state synchronized with the current session surface. */ +function balanceCache(session: Session): BalanceCache { + const surface = session.surface + const nodes = surface.nodes + const generation = surface.replaceGeneration + const cached = balanceCacheBySession.get(session) + + if (cached === undefined || cached.generation !== generation || cached.processedNodes > nodes.length) { + const rebuilt = rebuildCache(session, nodes, generation) + balanceCacheBySession.set(session, rebuilt) + return rebuilt + } + if (cached.processedNodes < nodes.length) return extendCache(session, cached, nodes) + return cached +} + +/** + * Whether the cut immediately before a current surface node is tool-pairing balanced. + * @param session - session whose surface is checked. + * @param node - surface node whose leading cut is checked; only its seq identifies it. + * @returns true when no unanswered tool call crosses the cut. + * @throws when the seq is absent from the current surface, a surface node has no + * matching log event, or a tool result has no preceding open call. + */ +export function toolPairingBalancedBefore(session: Session, node: SurfaceNode): boolean { + const cache = balanceCache(session) + const balanced = cache.beforeSeq.get(node.seq) + if (balanced === undefined) { + throw new Error(`tool-pairing balance: surface seq ${node.seq} not found`) + } + return balanced +} + +/** + * Whether the cut immediately after a current surface node is tool-pairing balanced. + * @param session - session whose surface is checked. + * @param node - surface node whose trailing cut is checked; only its seq identifies it. + * @returns true when no unanswered tool call crosses the cut. + * @throws when the seq is absent from the current surface, a surface node has no + * matching log event, or a tool result has no preceding open call. + */ +export function toolPairingBalancedAfter(session: Session, node: SurfaceNode): boolean { + const cache = balanceCache(session) + const successor = cache.successorBySeq.get(node.seq) + if (successor === undefined) { + throw new Error(`tool-pairing balance: surface seq ${node.seq} not found`) + } + if (successor === null) return cache.depth === 0 + // Current membership and positional successors are cache-owned. A caller may + // retain a node across surface changes, so its mutable-looking `next` field is + // never authoritative for this query. + // The successor map and balance map are committed together. + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + return cache.beforeSeq.get(successor)! +} diff --git a/packages/compact/compact/tests/tool-pairing.spec.ts b/packages/compact/compact/tests/tool-pairing.spec.ts new file mode 100644 index 0000000000..dfde2bd2ba --- /dev/null +++ b/packages/compact/compact/tests/tool-pairing.spec.ts @@ -0,0 +1,327 @@ +import { describe, expect, it } from 'vitest' +import { CallId } from '@deepseek-ai/dsh-llm' +import { toolPairingBalancedAfter, toolPairingBalancedBefore } from '@deepseek-ai/dsh-compact' +import { Session, SessionId } from '@deepseek-ai/dsh-session' +import type { SessionEvent, SurfaceNode } from '@deepseek-ai/dsh-session' + +const SURFACE = { surfaceOp: 'append' as const } + +function seqOf(session: Session, type: SessionEvent['type'], nth = 0): number { + return session.events.filter(event => event.type === type)[nth]!.seq +} + +function nodeAt(session: Session, seq: number): SurfaceNode { + const node = session.surface.nodes.find(candidate => candidate.seq === seq) + if (node === undefined) throw new Error(`seq ${seq} is not a surface node`) + return node +} + +function before(session: Session, type: SessionEvent['type'], nth = 0): boolean { + return toolPairingBalancedBefore(session, nodeAt(session, seqOf(session, type, nth))) +} + +function after(session: Session, type: SessionEvent['type'], nth = 0): boolean { + return toolPairingBalancedAfter(session, nodeAt(session, seqOf(session, type, nth))) +} + +function closedToolStep(): Session { + const session = new Session(SessionId('closed-tool-step')) + session.append('user/message', { + content: [{ type: 'text', text: 'go' }], + source: { kind: 'user' }, + }, SURFACE) + session.append('assistant/message', { + turn: 1, + step: 1, + content: [{ type: 'tool-call', id: CallId('c1'), name: 'bash', arguments: '{}' }], + }, SURFACE) + session.append('tool/result', { + turn: 1, + step: 1, + callId: CallId('c1'), + content: [{ type: 'text', text: 'done' }], + isError: false, + }, SURFACE) + return session +} + +describe('tool-pairing boundaries', () => { + it('classifies closed and open single-call steps', () => { + const closed = closedToolStep() + expect(before(closed, 'user/message')).toBe(true) + expect(after(closed, 'user/message')).toBe(true) + expect(before(closed, 'assistant/message')).toBe(true) + expect(after(closed, 'assistant/message')).toBe(false) + expect(before(closed, 'tool/result')).toBe(false) + expect(after(closed, 'tool/result')).toBe(true) + + const open = new Session(SessionId('open-tool-step')) + open.append('assistant/message', { + turn: 1, + step: 1, + content: [{ type: 'tool-call', id: CallId('open'), name: 'bash', arguments: '{}' }], + }, SURFACE) + expect(toolPairingBalancedAfter(open, open.surface.nodes[0]!)).toBe(false) + }) + + it('requires every result from a multiple-call assistant message', () => { + const session = new Session(SessionId('multiple-calls')) + session.append('assistant/message', { + turn: 1, + step: 1, + content: [ + { type: 'tool-call', id: CallId('c1'), name: 'one', arguments: '{}' }, + { type: 'tool-call', id: CallId('c2'), name: 'two', arguments: '{}' }, + ], + }, SURFACE) + session.append('tool/result', { + turn: 1, step: 1, callId: CallId('c1'), content: [], isError: false, + }, SURFACE) + session.append('tool/result', { + turn: 1, step: 1, callId: CallId('c2'), content: [], isError: false, + }, SURFACE) + + expect(after(session, 'tool/result', 0)).toBe(false) + expect(after(session, 'tool/result', 1)).toBe(true) + }) + + it('keeps neutral nodes inside an open pair unbalanced and free nodes balanced', () => { + const midStep = new Session(SessionId('neutral-mid-step')) + midStep.append('assistant/message', { + turn: 1, + step: 1, + content: [{ type: 'tool-call', id: CallId('c1'), name: 'bash', arguments: '{}' }], + }, SURFACE) + midStep.append('context/message', { + content: [{ type: 'text', text: 'background update' }], + source: { kind: 'plugin', plugin: 'test' }, + }, SURFACE) + midStep.append('tool/result', { + turn: 1, step: 1, callId: CallId('c1'), content: [], isError: false, + }, SURFACE) + expect(before(midStep, 'context/message')).toBe(false) + expect(after(midStep, 'context/message')).toBe(false) + + const free = new Session(SessionId('neutral-free')) + free.append('context/message', { + content: [{ type: 'text', text: 'idle injection' }], + source: { kind: 'user' }, + }, SURFACE) + expect(before(free, 'context/message')).toBe(true) + expect(after(free, 'context/message')).toBe(true) + }) +}) + +describe('tool-pairing surface identity', () => { + it('rebuilds after replace and rejects nodes removed from current membership', () => { + const session = closedToolStep() + const staleTail = nodeAt(session, seqOf(session, 'tool/result')) + expect(toolPairingBalancedAfter(session, staleTail)).toBe(true) + + const nodes = session.surface.nodes + session.append('user/message', { + content: [{ type: 'text', text: 'checkpoint' }], + source: { kind: 'plugin', plugin: 'compact' }, + }, { surfaceOp: { op: 'replace', start: nodes[0]!.seq, end: nodes.at(-1)!.seq } }) + + const checkpoint = session.surface.nodes[0]! + expect(toolPairingBalancedBefore(session, checkpoint)).toBe(true) + expect(toolPairingBalancedAfter(session, checkpoint)).toBe(true) + expect(() => toolPairingBalancedBefore(session, staleTail)).toThrow(/surface seq .* not found/) + expect(() => toolPairingBalancedAfter(session, staleTail)).toThrow(/surface seq .* not found/) + }) + + it('uses the cached positional successor instead of a caller node next field', () => { + const session = closedToolStep() + const assistant = nodeAt(session, seqOf(session, 'assistant/message')) + expect(toolPairingBalancedAfter(session, { ...assistant, next: null })).toBe(false) + expect(toolPairingBalancedAfter(session, { ...assistant, next: 999 })).toBe(false) + }) + + it('rejects missing seqs before and after, including an empty surface', () => { + const session = new Session(SessionId('missing-membership')) + const missing: SurfaceNode = { seq: 999, prev: null, next: null } + expect(() => toolPairingBalancedBefore(session, missing)).toThrow(/surface seq 999 not found/) + expect(() => toolPairingBalancedAfter(session, missing)).toThrow(/surface seq 999 not found/) + + session.append('user/message', { + content: [{ type: 'text', text: 'first node after empty cache' }], + source: { kind: 'user' }, + }, SURFACE) + expect(toolPairingBalancedAfter(session, session.surface.nodes[0]!)).toBe(true) + }) +}) + +describe('tool-pairing cache refresh', () => { + it('does no event reads for unchanged or log-only growth, folds only appended nodes, and rebuilds on replace', () => { + const events: SessionEvent[] = [ + { + type: 'user/message', seq: 0, time: 0, + data: { content: [{ type: 'text', text: 'user' }], source: { kind: 'user' } }, + surfaceOp: 'append', + }, + { + type: 'assistant/message', seq: 1, time: 1, + data: { turn: 1, step: 1, content: [{ type: 'tool-call', id: CallId('c1'), name: 'one', arguments: '{}' }] }, + surfaceOp: 'append', + }, + { + type: 'tool/result', seq: 2, time: 2, + data: { turn: 1, step: 1, callId: CallId('c1'), content: [], isError: false }, + surfaceOp: 'append', + }, + ] + const nodes: SurfaceNode[] = [ + { seq: 0, prev: null, next: 1 }, + { seq: 1, prev: 0, next: 2 }, + { seq: 2, prev: 1, next: null }, + ] + let generation = 0 + let eventCollectionReads = 0 + let eventIndexReads = 0 + const trackedEvents = new Proxy(events, { + get(target, property, receiver) { + if (typeof property === 'string' && /^\d+$/.test(property)) eventIndexReads += 1 + return Reflect.get(target, property, receiver) as unknown + }, + }) + const surface = { + get nodes() { return nodes }, + get replaceGeneration() { return generation }, + } + const session = { + surface, + get events() { + eventCollectionReads += 1 + return trackedEvents + }, + } as unknown as Session + + expect(toolPairingBalancedAfter(session, nodes[2]!)).toBe(true) + expect(eventCollectionReads).toBe(1) + expect(eventIndexReads).toBe(3) + + expect(toolPairingBalancedBefore(session, nodes[0]!)).toBe(true) + expect(toolPairingBalancedAfter(session, nodes[1]!)).toBe(false) + expect(eventCollectionReads).toBe(1) + expect(eventIndexReads).toBe(3) + + events.push({ + type: 'turn/end', seq: 3, time: 3, + data: { turn: 1, reason: { kind: 'completed' } }, + }) + expect(toolPairingBalancedAfter(session, nodes[2]!)).toBe(true) + expect(eventCollectionReads).toBe(1) + expect(eventIndexReads).toBe(3) + + events.push({ + type: 'user/message', seq: 4, time: 4, + data: { content: [{ type: 'text', text: 'tail' }], source: { kind: 'user' } }, + surfaceOp: 'append', + }) + nodes.push({ seq: 4, prev: 2, next: null }) + expect(toolPairingBalancedAfter(session, nodes[3]!)).toBe(true) + expect(eventCollectionReads).toBe(2) + expect(eventIndexReads).toBe(4) + + events.push( + { + type: 'assistant/message', seq: 5, time: 5, + data: { turn: 2, step: 1, content: [{ type: 'tool-call', id: CallId('c2'), name: 'two', arguments: '{}' }] }, + surfaceOp: 'append', + }, + { + type: 'tool/result', seq: 6, time: 6, + data: { turn: 2, step: 1, callId: CallId('c2'), content: [], isError: false }, + surfaceOp: 'append', + }, + ) + nodes.push( + { seq: 5, prev: 4, next: 6 }, + { seq: 6, prev: 5, next: null }, + ) + expect(toolPairingBalancedAfter(session, nodes[5]!)).toBe(true) + expect(eventCollectionReads).toBe(3) + expect(eventIndexReads).toBe(6) + + events.push({ + type: 'user/message', seq: 7, time: 7, + data: { content: [{ type: 'text', text: 'replacement' }], source: { kind: 'user' } }, + surfaceOp: { op: 'replace', start: 0, end: 6 }, + }) + nodes.splice(0, nodes.length, { seq: 7, prev: null, next: null }) + generation += 1 + expect(toolPairingBalancedAfter(session, nodes[0]!)).toBe(true) + expect(eventCollectionReads).toBe(4) + expect(eventIndexReads).toBe(7) + }) + + it('rebuilds defensively when a same-generation surface node count regresses', () => { + const events: SessionEvent[] = [ + { + type: 'user/message', seq: 0, time: 0, + data: { content: [], source: { kind: 'user' } }, surfaceOp: 'append', + }, + { + type: 'user/message', seq: 1, time: 1, + data: { content: [], source: { kind: 'user' } }, surfaceOp: 'append', + }, + ] + const nodes: SurfaceNode[] = [ + { seq: 0, prev: null, next: 1 }, + { seq: 1, prev: 0, next: null }, + ] + const session = { + events, + surface: { nodes, replaceGeneration: 0 }, + } as unknown as Session + expect(toolPairingBalancedAfter(session, nodes[1]!)).toBe(true) + nodes.pop() + expect(toolPairingBalancedAfter(session, nodes[0]!)).toBe(true) + }) +}) + +describe('tool-pairing corrupt surfaces', () => { + it('throws for an orphan result during a rebuild', () => { + const session = new Session(SessionId('orphan-rebuild')) + session.append('tool/result', { + turn: 1, step: 1, callId: CallId('orphan'), content: [], isError: false, + }, SURFACE) + expect(() => toolPairingBalancedAfter(session, session.surface.nodes[0]!)).toThrow(/no matching tool-call/) + }) + + it('retries an orphan result in an appended tail without committing partial cache state', () => { + const session = new Session(SessionId('orphan-tail')) + session.append('user/message', { + content: [{ type: 'text', text: 'safe head' }], source: { kind: 'user' }, + }, SURFACE) + expect(toolPairingBalancedAfter(session, session.surface.nodes[0]!)).toBe(true) + session.append('tool/result', { + turn: 1, step: 1, callId: CallId('orphan'), content: [], isError: false, + }, SURFACE) + expect(() => toolPairingBalancedAfter(session, session.surface.nodes[1]!)).toThrow(/no matching tool-call/) + expect(() => toolPairingBalancedAfter(session, session.surface.nodes[1]!)).toThrow(/no matching tool-call/) + }) + + it('throws when a current surface seq has no matching event or indexes the wrong event', () => { + const missingNode: SurfaceNode = { seq: 1, prev: null, next: null } + const missing = { + events: [{ + type: 'user/message', seq: 0, time: 0, + data: { content: [], source: { kind: 'user' } }, surfaceOp: 'append', + } satisfies SessionEvent], + surface: { nodes: [missingNode], replaceGeneration: 0 }, + } as unknown as Session + expect(() => toolPairingBalancedBefore(missing, missingNode)).toThrow(/no matching session event/) + + const mismatchedNode: SurfaceNode = { seq: 0, prev: null, next: null } + const mismatched = { + events: [{ + type: 'user/message', seq: 99, time: 0, + data: { content: [], source: { kind: 'user' } }, surfaceOp: 'append', + } satisfies SessionEvent], + surface: { nodes: [mismatchedNode], replaceGeneration: 0 }, + } as unknown as Session + expect(() => toolPairingBalancedBefore(mismatched, mismatchedNode)).toThrow(/no matching session event/) + }) +}) diff --git a/packages/core/session/README.md b/packages/core/session/README.md index 7e201ab647..2540f988c7 100644 --- a/packages/core/session/README.md +++ b/packages/core/session/README.md @@ -77,7 +77,7 @@ Every `SessionEvent` carries two optional top-level fields (structural metadata) - Persistence plugins: subscribe to `session/event` (write-behind) and drain on `session/flush` (awaited) and fiber dispose. A durable backend reads the log and reloads it into a live session; the metadata seam (`SessionHeader`, `session.header`) is what such a backend stores beside the log. - Replay/fork: `create(id, { seed })` validates and freezes a contiguous log and rebuilds its surface. `fork(source, boundary?, childSessionId?)` selects a completed-turn prefix and records lineage. -- Compaction: the `dsh-compact-basic` plugin appends a `user/message` with `surfaceOp: { op: 'replace', start, end }` to shadow old surface nodes behind a summary checkpoint. +- Compaction: the `dsh-compact-basic` plugin appends a `user/message` with `surfaceOp: { op: 'replace', start, end }` to shadow old surface nodes behind a summary checkpoint. Tool-pairing boundary policy and its cache belong to the [`dsh-compact` seam](../../compact/compact/README.md), while this package owns surface membership, positional links, and `replaceGeneration`. ## Model Experience diff --git a/packages/core/session/src/index.ts b/packages/core/session/src/index.ts index 89f1627445..c3c99ae885 100644 --- a/packages/core/session/src/index.ts +++ b/packages/core/session/src/index.ts @@ -24,7 +24,6 @@ export type { JsonValue } from './json.ts' export { interruptedTurnClosers } from './repair.ts' export type { SurfaceFoldReplacement, SurfaceFoldResult, SurfaceNode } from './surface.ts' export { foldSurface, isSurfaceEvent, isSurfaceEligibleType } from './surface.ts' -export { isToolPairingBalanced } from './tool-pairing.ts' export { applyHeaderDelta, canonicalHeader, diffHeader, foldRequestHeader, headerEquals } from './request-header.ts' declare module 'cordis' { diff --git a/packages/core/session/src/tool-pairing.ts b/packages/core/session/src/tool-pairing.ts deleted file mode 100644 index ce9dc639c7..0000000000 --- a/packages/core/session/src/tool-pairing.ts +++ /dev/null @@ -1,56 +0,0 @@ -/** - * Tool-pairing balance over a session surface. Compaction changes surface - * positions, so safe cuts are derived from tool-call/result content on the - * surface rather than step markers in the append-only log. - * @module @deepseek-ai/dsh-session/tool-pairing - */ - -import type { SessionEvent } from './types.ts' -import type { SurfaceNode } from './surface.ts' - -/** - * The tool-pairing delta of a surface node: how it shifts the count of - * unanswered tool calls. An `assistant/message` opens one bracket per - * `tool-call` block; a `tool/result` closes one; every other surface node - * (`user/message`, `context/message`, `steering/message`, a usage-only - * `assistant/message` with no tool-call blocks) is pairing-neutral. - */ -function nodeDelta(event: SessionEvent): number { - switch (event.type) { - case 'assistant/message': - return event.data.content.filter(block => block.type === 'tool-call').length - case 'tool/result': - return -1 - // Non-pairing surface nodes and every non-surface event contribute nothing. - default: - return 0 - } -} - -/** - * Check that a surface cut does not split a tool call from its result. A region - * is safe to collapse only when the cuts before its first node and after its - * last node both return `true`. - * @param nodes - the surface linked list in head→tail order. - * @param events - the session log each node's `seq` indexes into. - * @param beforeSeq - node immediately after the cut; `null` or a seq absent from the surface means after-tail. - * @returns whether every call before the cut has its result before the cut. - * @throws if a result appears without a preceding open call. - */ -export function isToolPairingBalanced( - nodes: readonly SurfaceNode[], - events: readonly SessionEvent[], - beforeSeq: number | null, -): boolean { - let depth = 0 - for (const node of nodes) { - if (node.seq === beforeSeq) return depth === 0 - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - depth += nodeDelta(events[node.seq]!) - if (depth < 0) { - throw new Error(`tool-pairing balance: tool/result at surface seq ${node.seq} has no matching tool-call (corrupt surface)`) - } - } - // A missing cut node means the after-tail boundary. - return depth === 0 -} diff --git a/packages/core/session/tests/tool-pairing.spec.ts b/packages/core/session/tests/tool-pairing.spec.ts deleted file mode 100644 index eb7b1a6203..0000000000 --- a/packages/core/session/tests/tool-pairing.spec.ts +++ /dev/null @@ -1,292 +0,0 @@ -import { describe, expect, it } from 'vitest' -import { CallId } from '@deepseek-ai/dsh-llm' -import { Session, SessionId, isToolPairingBalanced } from '../src/index.ts' -import type { SessionEvent, SurfaceNode } from '../src/index.ts' - -/** - * Unit coverage for compaction-cut safety: a cut is balanced only when it - * separates no assistant tool call from its result. Non-step nodes are neutral, - * and replace operations prove surface order—not raw log order—is authoritative. - */ - -const SURFACE = { surfaceOp: 'append' as const } - -/** Surface nodes + log for a session, the two args the balance check takes. */ -function surfaceOf(session: Session): { nodes: readonly SurfaceNode[]; events: readonly SessionEvent[] } { - return { nodes: session.surface.nodes, events: session.events } -} - -/** The cut BEFORE the surface node at `seq` is balanced (safe region start). */ -function startBalanced(session: Session, seq: number): boolean { - const { nodes, events } = surfaceOf(session) - return isToolPairingBalanced(nodes, events, seq) -} - -/** The cut AFTER the surface node at `seq` is balanced (safe region end). */ -function endBalanced(session: Session, seq: number): boolean { - const { nodes, events } = surfaceOf(session) - const node = nodes.find(n => n.seq === seq) - if (!node) throw new Error(`seq ${seq} is not a surface node`) - return isToolPairingBalanced(nodes, events, node.next) -} - -/** Surface seq of the nth (0-based) event of a given type. */ -function seqOf(s: Session, type: SessionEvent['type'], nth = 0): number { - return s.events.filter(e => e.type === type)[nth]!.seq -} - -/** A closed turn with one closed step holding an assistant + its tool result. */ -function toolStepSession(): Session { - const s = new Session(SessionId('tool-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' } }, SURFACE) - 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: '{}' }, - ], - }, SURFACE) - 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 }, SURFACE) - s.append('step/end', { turn: 1, step: 1 }) - s.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) - return s -} - -describe('isToolPairingBalanced — region START (cut before a node)', () => { - it('is true for a pre-step user/message (belongs to no step)', () => { - const s = toolStepSession() - expect(startBalanced(s, seqOf(s, 'user/message'))).toBe(true) - }) - - it('is true for the first surface node of a step (the assistant/message)', () => { - // The cut before the assistant is balanced — nothing unanswered precedes it. - const s = toolStepSession() - expect(startBalanced(s, seqOf(s, 'assistant/message'))).toBe(true) - }) - - it('is false for a tool/result whose assistant/message precedes it in the same step', () => { - // The cut before the tool/result has one unanswered tool-call (the - // assistant's) → starting the region here would orphan that call. - const s = toolStepSession() - expect(startBalanced(s, seqOf(s, 'tool/result'))).toBe(false) - }) - - it('is true at the surface head (nothing precedes)', () => { - const s = new Session(SessionId('lone')) - s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - s.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, SURFACE) - expect(startBalanced(s, seqOf(s, 'user/message'))).toBe(true) - }) -}) - -describe('isToolPairingBalanced — region END (cut after a node)', () => { - it('is true for the last surface node of a closed step (the tool/result)', () => { - // After the tool/result the assistant's single call is answered → balanced. - const s = toolStepSession() - expect(endBalanced(s, seqOf(s, 'tool/result'))).toBe(true) - }) - - it('is false for an assistant/message with a later tool/result in the same step', () => { - // After the assistant its tool-call is still unanswered → ending here strands - // the result. - const s = toolStepSession() - expect(endBalanced(s, seqOf(s, 'assistant/message'))).toBe(false) - }) - - it('is true for a pre-step user/message', () => { - const s = toolStepSession() - expect(endBalanced(s, seqOf(s, 'user/message'))).toBe(true) - }) - - it('is false at the tail when the node is inside an open (unclosed) step', () => { - // step/start then an assistant tool-call, but no tool/result yet (mid-flight). - // The after-tail cut still has one unanswered call → not balanced. - const s = new Session(SessionId('open-step')) - s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - s.append('step/start', { turn: 1, step: 1 }) - s.append('assistant/message', { - turn: 1, step: 1, - content: [{ type: 'tool-call', id: CallId('c1'), name: 'bash', arguments: '{}' }], - }, SURFACE) - expect(endBalanced(s, seqOf(s, 'assistant/message'))).toBe(false) - }) - - it('is true at the tail when the node is a trailing inter-step node (step already closed)', () => { - // A steering message appended after step/end, at the tail. The prior step's - // pair is balanced and steering is neutral → the after-tail cut is balanced. - const s = new Session(SessionId('trailing-steer')) - s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - s.append('step/start', { turn: 1, step: 1 }) - s.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'a' }] }, SURFACE) - s.append('step/end', { turn: 1, step: 1 }) - s.append('steering/message', { turn: 1, content: [{ type: 'text', text: 's' }], source: { kind: 'user' } }, SURFACE) - expect(endBalanced(s, seqOf(s, 'steering/message'))).toBe(true) - }) - - it('is true at the tail when no step ever opened', () => { - const s = new Session(SessionId('no-step')) - s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - s.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, SURFACE) - expect(endBalanced(s, seqOf(s, 'user/message'))).toBe(true) - }) -}) - -describe('isToolPairingBalanced — multiple tool calls in one assistant message', () => { - // An assistant message with two tool-calls needs BOTH results before the cut - // after it is balanced — depth +2, then -1, -1. - function twoCallStep(): Session { - const s = new Session(SessionId('two-call')) - s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - s.append('step/start', { turn: 1, step: 1 }) - s.append('assistant/message', { - turn: 1, step: 1, - content: [ - { type: 'tool-call', id: CallId('c1'), name: 'a', arguments: '{}' }, - { type: 'tool-call', id: CallId('c2'), name: 'b', arguments: '{}' }, - ], - }, SURFACE) - s.append('tool/result', { turn: 1, step: 1, callId: CallId('c1'), content: [{ type: 'text', text: '1' }], isError: false }, SURFACE) - s.append('tool/result', { turn: 1, step: 1, callId: CallId('c2'), content: [{ type: 'text', text: '2' }], isError: false }, SURFACE) - s.append('step/end', { turn: 1, step: 1 }) - s.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) - return s - } - - it('is unbalanced after the first of two results (one call still open)', () => { - const s = twoCallStep() - expect(endBalanced(s, seqOf(s, 'tool/result', 0))).toBe(false) - }) - - it('is balanced after the second result (both calls answered)', () => { - const s = twoCallStep() - expect(endBalanced(s, seqOf(s, 'tool/result', 1))).toBe(true) - }) -}) - -describe('isToolPairingBalanced — a mid-step injection context/message', () => { - // The injected context is pairing-neutral, but both adjacent cuts remain - // unbalanced because the tool call is still open across them. - function midStepInjection(): Session { - const s = new Session(SessionId('mid-inject')) - s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - s.append('step/start', { turn: 1, step: 1 }) - s.append('assistant/message', { - turn: 1, step: 1, - content: [{ type: 'tool-call', id: CallId('c1'), name: 'bash', arguments: '{}' }], - }, SURFACE) - s.append('context/message', { content: [{ type: 'text', text: 'bg task done' }], source: { kind: 'plugin', plugin: 'tool-bash' } }, SURFACE) - s.append('tool/result', { turn: 1, step: 1, callId: CallId('c1'), content: [{ type: 'text', text: 'out' }], isError: false }, SURFACE) - s.append('step/end', { turn: 1, step: 1 }) - s.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) - return s - } - - it('start cut before the mid-step context/message is unbalanced (call still open)', () => { - const s = midStepInjection() - expect(startBalanced(s, seqOf(s, 'context/message'))).toBe(false) - }) - - it('end cut after the mid-step context/message is unbalanced (call still open)', () => { - const s = midStepInjection() - expect(endBalanced(s, seqOf(s, 'context/message'))).toBe(false) - }) -}) - -describe('isToolPairingBalanced on an injection turn (no step)', () => { - // An idle inject() wraps a context/message in a bare turn/start → - // context/message → turn/end with NO step. The context node is a free boundary - // both ways (pairing-neutral, nothing open around it). - function injectionSession(): Session { - const s = new Session(SessionId('injection')) - s.append('turn/start', { turn: 1, trigger: { kind: 'injection', source: { kind: 'user' } } }) - s.append('context/message', { content: [{ type: 'text', text: 'ctx' }], source: { kind: 'user' } }, SURFACE) - s.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) - return s - } - - it('start: balanced', () => { - const s = injectionSession() - expect(startBalanced(s, seqOf(s, 'context/message'))).toBe(true) - }) - - it('end: balanced', () => { - const s = injectionSession() - expect(endBalanced(s, seqOf(s, 'context/message'))).toBe(true) - }) -}) - -describe('isToolPairingBalanced — CBR-001: a head checkpoint left by a replace op', () => { - // A replacement checkpoint has a high log seq but sits at the surface head; - // its cuts are balanced regardless of later raw-log neighbors. - function checkpointHeadedSession(): Session { - const s = new Session(SessionId('checkpoint')) - // A closed turn with a tool step → surface [u1, asst(call), result]. - 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: 'u1' }], source: { kind: 'user' } }, SURFACE) - s.append('assistant/message', { - turn: 1, step: 1, - content: [{ type: 'tool-call', id: CallId('c1'), name: 'bash', arguments: '{}' }], - }, SURFACE) - s.append('tool/result', { turn: 1, step: 1, callId: CallId('c1'), content: [{ type: 'text', text: 'out' }], isError: false }, SURFACE) - s.append('step/end', { turn: 1, step: 1 }) - s.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) - // An OPEN turn whose step is in progress (loop fires compaction here). - s.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) - s.append('step/start', { turn: 2, step: 1 }) - // Compaction replaces the whole turn-1 surface ([u1, asst, result]) with one - // summary user/message — appended now, so it carries a high log seq. - const u1 = seqOf(s, 'user/message') - const result = s.events.find(e => e.type === 'tool/result')!.seq - s.append('user/message', { - content: [{ type: 'text', text: 'CHECKPOINT' }], - source: { kind: 'plugin', plugin: 'compact' }, - }, { surfaceOp: { op: 'replace', start: u1, end: result } }) - // The step's own assistant/message lands AFTER the checkpoint in the log, - // still inside the open step. - s.append('assistant/message', { turn: 2, step: 1, content: [{ type: 'text', text: 'a2' }] }, SURFACE) - return s - } - - it('the head checkpoint sits at the surface head while a later surface node follows it in the log', () => { - const s = checkpointHeadedSession() - const nodes = s.surface.nodes - const checkpointSeq = nodes[0]!.seq - // The checkpoint heads the surface, yet a surface node (the open step's - // assistant) follows it in LOG order — the exact split between surface - // position and log position that the log-position scan tripped on. - const laterSurfaceInLog = s.events.find( - e => e.seq > checkpointSeq && nodes.some(n => n.seq === e.seq), - ) - expect(laterSurfaceInLog).toBeDefined() - expect(nodes[0]!.seq).toBe(checkpointSeq) - }) - - it('start cut before the head checkpoint is balanced (it is the head)', () => { - const s = checkpointHeadedSession() - expect(startBalanced(s, s.surface.nodes[0]!.seq)).toBe(true) - }) - - it('end cut after the head checkpoint is balanced (it carries no tool pair)', () => { - // This is the exact assertion the log-position scan failed: the forward log scan from the - // checkpoint reached the open step's assistant/message and wrongly reported mid-step. - const s = checkpointHeadedSession() - expect(endBalanced(s, s.surface.nodes[0]!.seq)).toBe(true) - }) -}) - -describe('isToolPairingBalanced — corrupt surface guard', () => { - it('throws when a tool/result has no preceding tool-call (depth goes negative)', () => { - // A surface that opens with a tool/result (no assistant call before it) is - // structurally corrupt — surfaced loudly rather than mis-classified. - const s = new Session(SessionId('corrupt')) - s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - s.append('step/start', { turn: 1, step: 1 }) - s.append('tool/result', { turn: 1, step: 1, callId: CallId('c1'), content: [{ type: 'text', text: 'x' }], isError: false }, SURFACE) - const { nodes, events } = surfaceOf(s) - expect(() => isToolPairingBalanced(nodes, events, null)).toThrow(/no matching tool-call/) - }) -}) From 2793325df0b0bc0354068b1dbedb4751edce04da Mon Sep 17 00:00:00 2001 From: Hypatia May Date: Thu, 16 Jul 2026 11:40:00 +0800 Subject: [PATCH 2/2] refactor(compact): store tool-pairing balance once per surface cut toolPairingBalancedAfter previously answered by resolving a cached positional successor and reading its before-balance, with a null-successor depth fallback. Both queries are the same prefix property sampled at adjacent cuts, so the cache now holds one per-cut balance sequence (N nodes -> N+1 cuts) plus a seq->position index; before/after differ only by a cut offset. The successor map, the duplicate rebuild/extend fold loops, and the non-null assertion are gone, and the running counter is named inProgressToolCalls. Docs describing the successor mechanism are updated in place. --- docs/core-data-structures/session.md | 2 +- .../2026-06-18-compaction-capability-seam.md | 4 +- ...-12-simplify-session-log-representation.md | 2 +- packages/compact/compact/README.md | 4 +- packages/compact/compact/src/tool-pairing.ts | 116 +++++++----------- .../compact/tests/tool-pairing.spec.ts | 2 +- 6 files changed, 51 insertions(+), 79 deletions(-) diff --git a/docs/core-data-structures/session.md b/docs/core-data-structures/session.md index 174f06602e..616370fb84 100644 --- a/docs/core-data-structures/session.md +++ b/docs/core-data-structures/session.md @@ -196,7 +196,7 @@ export interface SurfaceNode { } ``` -`SurfaceNode` is positional state, not durable identity. A replacement can remove a caller-retained node or make a copied `next` stale; consumers that cross a surface mutation validate membership and resolve successors from `Session.surface.nodes`. `SurfaceManager.replaceGeneration` increments for each replacement so incremental consumers can distinguish pure tail growth from a rewrite. +`SurfaceNode` is positional state, not durable identity. A replacement can remove a caller-retained node or make a copied `next` stale; consumers that cross a surface mutation validate membership and answer positional queries from `Session.surface.nodes`. `SurfaceManager.replaceGeneration` increments for each replacement so incremental consumers can distinguish pure tail growth from a rewrite. ### `SurfaceFoldReplacement` and `SurfaceFoldResult` — a complete surface replay diff --git a/docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md b/docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md index 29e3347759..75eaf3207f 100644 --- a/docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md +++ b/docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md @@ -52,7 +52,7 @@ The loop derives messages once after `agent/pre-step`. Running before `step/star Auto-compaction fires before **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-step` checkpoint. 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. -`compactIfNeeded` retains the smallest tail of whole surface units whose estimated size reaches `retainTokens` and compacts older nodes. A unit is a complete closed step or one no-step message. If the token cutoff lands inside a step, retention expands until the cut is tool-pairing balanced. Balance is checked on surface order, not log sequence, because replacement summaries have new sequence numbers at old surface positions. `dsh-compact` exports the before/after edge helpers; their per-session cache folds only appended surface-tail nodes while `replaceGeneration` is unchanged, does no event reads for log-only growth, and rebuilds current membership, positional successors, and balances after replacement. `compactRegion` rejects boundaries that split a tool call from its result. The in-flight turn receives no special retention. +`compactIfNeeded` retains the smallest tail of whole surface units whose estimated size reaches `retainTokens` and compacts older nodes. A unit is a complete closed step or one no-step message. If the token cutoff lands inside a step, retention expands until the cut is tool-pairing balanced. Balance is checked on surface order, not log sequence, because replacement summaries have new sequence numbers at old surface positions. `dsh-compact` exports the before/after edge helpers; their per-session cache folds only appended surface-tail nodes while `replaceGeneration` is unchanged, does no event reads for log-only growth, and rebuilds current membership and balances after replacement. `compactRegion` rejects boundaries that split a tool call from its result. The in-flight turn receives no special retention. 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. @@ -113,7 +113,7 @@ Two failure paths, both documented: - **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-step` (`@mode serial`) declared in `dsh-agent` and emitted by `dsh-agent-loop` after system assembly and before `step/start`. 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. -- **`dsh-compact`** owns `toolPairingBalancedBefore(session, node)` and `toolPairingBalancedAfter(session, node)`, the cached surface-edge checks that `compactRegion` and `compactIfNeeded` use to avoid splitting a tool-call/result pair. The cache validates current membership by seq and resolves after-edges from its positional successor map instead of trusting a caller-retained `node.next`; stale or missing seqs and orphan results reject. `dsh-session` continues to own the surface `replace` operation, positional nodes, and rewrite generation. +- **`dsh-compact`** owns `toolPairingBalancedBefore(session, node)` and `toolPairingBalancedAfter(session, node)`, the cached surface-edge checks that `compactRegion` and `compactIfNeeded` use to avoid splitting a tool-call/result pair. The cache validates current membership by seq and answers both edges from one per-cut balance sequence instead of trusting a caller-retained `node.next`; stale or missing seqs and orphan results reject. `dsh-session` continues to own the surface `replace` operation, positional nodes, and rewrite generation. - **`dsh-invariants`** drops its `surface replace: start must be <= end` assertion: a head-anchored compaction lands a high-seq replacement node at an older range's *position*, so `start > end` numerically is normal and valid (the range is positional, validated by the surface's `indexOf` checks that remain). The turn-enclosure invariant is reused unchanged. - **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). diff --git a/docs/rfc/proposed/simplification/2026-07-12-simplify-session-log-representation.md b/docs/rfc/proposed/simplification/2026-07-12-simplify-session-log-representation.md index 715ce93924..1cb8d5708a 100644 --- a/docs/rfc/proposed/simplification/2026-07-12-simplify-session-log-representation.md +++ b/docs/rfc/proposed/simplification/2026-07-12-simplify-session-log-representation.md @@ -6,7 +6,7 @@ Status: proposed The session log maintains two representations that cost more machinery than their consumers require: a pseudo-linked surface and custom request-header deltas. -`SurfaceManager` stores the same order in an array, a seq map, and mutable `prev`/`next` links. Production never reads `prev`; compact's sole `next` read is the successor of an array position. Replacement already uses `indexOf`, so the links do not make its dominant operation constant-time. A seq array with linear replacement lookup has the same asymptotic replacement cost and one representation to validate. +`SurfaceManager` stores the same order in an array, a seq map, and mutable `prev`/`next` links. Production never reads either link: compact's tool-pairing balance answers from per-cut balances cached in surface order. Replacement already uses `indexOf`, so the links do not make its dominant operation constant-time. A seq array with linear replacement lookup has the same asymptotic replacement cost and one representation to validate. The request-header subsystem implements a custom system/tool delta codec and transmission-decision layer even though its contract says deltas are an encoding optimization, not a reconstructability requirement. Retaining the initial/resume full snapshot at each loop-instance boundary, then writing a canonical full `request/header` whenever that instance's assembled header changes, preserves replay while deleting `SystemDelta`, `ToolsDelta`, round-trip fallback, and the durable `request/header-delta` variant. Codec-only vocabulary disappears with the codec, not because its individual arms were invalid. diff --git a/packages/compact/compact/README.md b/packages/compact/compact/README.md index 5b59893561..ef8c300856 100644 --- a/packages/compact/compact/README.md +++ b/packages/compact/compact/README.md @@ -25,9 +25,9 @@ Both methods are **abstract** — the backend owns the entire strategy (token es ## Tool-pairing boundaries -The interface exports `toolPairingBalancedBefore(session, node)` and `toolPairingBalancedAfter(session, node)` for snapping and validating compaction edges. A safe edge has no unanswered assistant tool call crossing it. Each helper validates the node's seq against current surface membership and resolves the trailing edge from its cached positional successor, so a stale caller-held `node.next` cannot choose the cut. +The interface exports `toolPairingBalancedBefore(session, node)` and `toolPairingBalancedAfter(session, node)` for snapping and validating compaction edges. A safe edge has no unanswered assistant tool call crossing it. Each helper identifies the node by seq alone and answers from balances cached per cut in current surface order, so a stale caller-held `node.next` cannot choose the cut. -The private per-session cache is keyed by `session.surface.replaceGeneration` and the processed surface-node count. An unchanged generation extends the fold with unseen tail nodes only; a log-only append with no new surface node does no event reads, while a replacement generation rebuilds current membership, successors, and balances. Missing event seqs and a `tool/result` without a preceding open call reject as corrupt surface state. +The private per-session cache is keyed by `session.surface.replaceGeneration` and the processed surface-node count. An unchanged generation extends the fold with unseen tail nodes only; a log-only append with no new surface node does no event reads, while a replacement generation rebuilds current membership and balances. Missing event seqs and a `tool/result` without a preceding open call reject as corrupt surface state. ## Surface contract diff --git a/packages/compact/compact/src/tool-pairing.ts b/packages/compact/compact/src/tool-pairing.ts index a9fca01bf6..0fc0f68dc8 100644 --- a/packages/compact/compact/src/tool-pairing.ts +++ b/packages/compact/compact/src/tool-pairing.ts @@ -12,19 +12,21 @@ import type { Session, SessionEvent, SurfaceNode } from '@deepseek-ai/dsh-sessio interface BalanceCache { /** Surface rewrite generation this state describes. */ generation: number - /** Number of surface nodes already folded into the state. */ - processedNodes: number - /** Balance of the cut immediately before each current surface node. */ - beforeSeq: Map - /** Current positional successor of each surface node. */ - successorBySeq: Map - /** Unanswered tool-call count after the processed surface tail. */ - depth: number + /** + * Balance of every surface cut in current order: a surface of N nodes has + * N + 1 cuts, entry `i` being the cut before node `i` and the final entry + * the cut after the surface tail. + */ + cutBalanced: readonly boolean[] + /** Current surface position of each node seq, indexing {@link cutBalanced}. */ + indexBySeq: Map + /** In-progress tool-call count after the processed surface tail. */ + inProgressToolCalls: number } const balanceCacheBySession = new WeakMap() -/** Return how one surface event changes the unanswered tool-call count. */ +/** Return how one surface event changes the in-progress tool-call count. */ function nodeDelta(event: SessionEvent): number { switch (event.type) { case 'assistant/message': @@ -45,61 +47,30 @@ function eventForNode(events: readonly SessionEvent[], node: SurfaceNode): Sessi return event } -/** Build balance state for a complete current surface. */ -function rebuildCache( - session: Session, - nodes: readonly SurfaceNode[], - generation: number, -): BalanceCache { - const beforeSeq = new Map() - const successorBySeq = new Map() - const events = session.events - let depth = 0 - let previousSeq: number | undefined - - for (const node of nodes) { - beforeSeq.set(node.seq, depth === 0) - successorBySeq.set(node.seq, null) - if (previousSeq !== undefined) successorBySeq.set(previousSeq, node.seq) - depth += nodeDelta(eventForNode(events, node)) - if (depth < 0) { - throw new Error(`tool-pairing balance: tool/result at surface seq ${node.seq} has no matching tool-call (corrupt surface)`) - } - previousSeq = node.seq - } - - return { generation, processedNodes: nodes.length, beforeSeq, successorBySeq, depth } -} - -/** Fold a pure surface tail append into existing balance state. */ +/** Fold surface nodes not yet in the cache into its balance state. */ function extendCache( session: Session, cache: BalanceCache, nodes: readonly SurfaceNode[], ): BalanceCache { - const tail = nodes.slice(cache.processedNodes) + const processed = cache.cutBalanced.length - 1 + const tail = nodes.slice(processed) // Validate the unseen tail before mutating the live cache, so a corrupt // append cannot leave a partially advanced state behind. const events = session.events - const pending: Array<{ seq: number; before: boolean }> = [] - let depth = cache.depth + const pendingCuts: boolean[] = [] + let inProgressToolCalls = cache.inProgressToolCalls for (const node of tail) { - pending.push({ seq: node.seq, before: depth === 0 }) - depth += nodeDelta(eventForNode(events, node)) - if (depth < 0) { + inProgressToolCalls += nodeDelta(eventForNode(events, node)) + if (inProgressToolCalls < 0) { throw new Error(`tool-pairing balance: tool/result at surface seq ${node.seq} has no matching tool-call (corrupt surface)`) } + pendingCuts.push(inProgressToolCalls === 0) } - let previousSeq = nodes[cache.processedNodes - 1]?.seq - for (const entry of pending) { - if (previousSeq !== undefined) cache.successorBySeq.set(previousSeq, entry.seq) - cache.beforeSeq.set(entry.seq, entry.before) - cache.successorBySeq.set(entry.seq, null) - previousSeq = entry.seq - } - cache.processedNodes = nodes.length - cache.depth = depth + tail.forEach((node, offset) => cache.indexBySeq.set(node.seq, processed + offset)) + cache.cutBalanced = cache.cutBalanced.concat(pendingCuts) + cache.inProgressToolCalls = inProgressToolCalls return cache } @@ -110,15 +81,32 @@ function balanceCache(session: Session): BalanceCache { const generation = surface.replaceGeneration const cached = balanceCacheBySession.get(session) - if (cached === undefined || cached.generation !== generation || cached.processedNodes > nodes.length) { - const rebuilt = rebuildCache(session, nodes, generation) + if (cached === undefined || cached.generation !== generation || cached.cutBalanced.length - 1 > nodes.length) { + // A rebuild is the same fold started from the empty-surface state, whose + // single leading cut is trivially balanced. + const rebuilt = extendCache(session, { + generation, + cutBalanced: [true], + indexBySeq: new Map(), + inProgressToolCalls: 0, + }, nodes) balanceCacheBySession.set(session, rebuilt) return rebuilt } - if (cached.processedNodes < nodes.length) return extendCache(session, cached, nodes) + if (cached.cutBalanced.length - 1 < nodes.length) return extendCache(session, cached, nodes) return cached } +/** Balance of the cut at a node's position plus offset, rejecting seqs outside current membership. */ +function cutBalance(cache: BalanceCache, seq: number, offset: 0 | 1): boolean { + const index = cache.indexBySeq.get(seq) + const balanced = index === undefined ? undefined : cache.cutBalanced[index + offset] + if (balanced === undefined) { + throw new Error(`tool-pairing balance: surface seq ${seq} not found`) + } + return balanced +} + /** * Whether the cut immediately before a current surface node is tool-pairing balanced. * @param session - session whose surface is checked. @@ -128,12 +116,7 @@ function balanceCache(session: Session): BalanceCache { * matching log event, or a tool result has no preceding open call. */ export function toolPairingBalancedBefore(session: Session, node: SurfaceNode): boolean { - const cache = balanceCache(session) - const balanced = cache.beforeSeq.get(node.seq) - if (balanced === undefined) { - throw new Error(`tool-pairing balance: surface seq ${node.seq} not found`) - } - return balanced + return cutBalance(balanceCache(session), node.seq, 0) } /** @@ -145,16 +128,5 @@ export function toolPairingBalancedBefore(session: Session, node: SurfaceNode): * matching log event, or a tool result has no preceding open call. */ export function toolPairingBalancedAfter(session: Session, node: SurfaceNode): boolean { - const cache = balanceCache(session) - const successor = cache.successorBySeq.get(node.seq) - if (successor === undefined) { - throw new Error(`tool-pairing balance: surface seq ${node.seq} not found`) - } - if (successor === null) return cache.depth === 0 - // Current membership and positional successors are cache-owned. A caller may - // retain a node across surface changes, so its mutable-looking `next` field is - // never authoritative for this query. - // The successor map and balance map are committed together. - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - return cache.beforeSeq.get(successor)! + return cutBalance(balanceCache(session), node.seq, 1) } diff --git a/packages/compact/compact/tests/tool-pairing.spec.ts b/packages/compact/compact/tests/tool-pairing.spec.ts index dfde2bd2ba..bd74faeec5 100644 --- a/packages/compact/compact/tests/tool-pairing.spec.ts +++ b/packages/compact/compact/tests/tool-pairing.spec.ts @@ -131,7 +131,7 @@ describe('tool-pairing surface identity', () => { expect(() => toolPairingBalancedAfter(session, staleTail)).toThrow(/surface seq .* not found/) }) - it('uses the cached positional successor instead of a caller node next field', () => { + it('ignores a caller-held node next field and answers from cached balances', () => { const session = closedToolStep() const assistant = nodeAt(session, seqOf(session, 'assistant/message')) expect(toolPairingBalancedAfter(session, { ...assistant, next: null })).toBe(false)