Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(compact): address PR 110 review findings

Browse Source 
Honor cancellation and disposal around async pre-step setup before the loop can open a step or call the model.

Route compaction summarization through agent/request so router agents can select the model, and remove the stale model argument from agent/pre-step.

Document serial events and the approximate convergence bound, regenerate the Cordis catalog, and add regression coverage for router compaction, HMR cleanup, and assembly/pre-step interruption.
This commit is contained in:
Hypatia May
2026-06-29 15:59:52 +08:00
parent c13f25586c
commit 1f35a4446d
18 changed files with 551 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/compact/compact-basic/README.md
View File

@@ -1,6 +1,6 @@
# @deepseek-ai/dsh-compact-basic
The **basic compaction backend**: a `BasicCompactService` implementing the `@deepseek-ai/dsh-compact` seam with a char/4 token heuristic, token-budget retention, and `ctx.llm.stream()` summarization.
The **basic compaction backend**: a `BasicCompactService` implementing the `@deepseek-ai/dsh-compact` seam with a char/4 token heuristic, token-budget retention, and summarization routed through the agent request pipeline.
This is the implementation tier of the compaction capability — see the [interface package](../compact/README.md) for the seam and the [capability-seam RFC](../../../docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md) for the design.
@@ -11,7 +11,7 @@ The abstract contract states only WHAT compaction does; this backend owns every
- **Token estimation** — `estimateContentTokens()`: char/4 with per-block structural overhead (`text`/`reasoning` = `ceil(len/4) + 4`, `tool-call` from name + arguments, `tool-result` recursive, `image` = 85, unknown blocks via JSON length).
- **Retention policy** — `compactIfNeeded()` walks the surface nodes tail→head summing per-node token estimates, and retains the smallest tail-run of WHOLE units (a closed step, or a single no-step node such as a pre-step `user/message` or inter-step `steering/message`) whose total reaches `retainTokens`; everything older is compacted. Retention is **turn-agnostic** — turn boundaries play no role, so a single runaway turn that alone exceeds the window compacts its OWN early closed steps rather than being retained verbatim (the failure mode that motivated dropping turn-protection: a tool-heavy turn must stay compactable or the harness dies exactly when compaction is needed). The only structural guard is **tool-pairing balance**: the compacted region's edges are balanced cuts on the surface (no unanswered tool-call crosses either edge), so it never splits a step's `assistant/message` tool-calls from their `tool/result`s. When the only compactable content left is an un-splittable open tail step, it declines (returns `null`) and retries once an older step closes. **Single-unit overflow is out of scope, by design**: if one retained unit (a single closed step, or a large pasted `user/message`) ALONE exceeds the budget, compaction cannot help and the call may go out over-budget — bounding an individual unit's size is a separate concern. `compactRegion()` enforces tool-pairing balance strictly, throwing on a boundary that would split a step. `dsh-session` exports `isToolPairingBalanced` for the check.
- **Single-pass convergence** — `resolveConfig()` rejects (throws) any config where `summarizationMaxTokens + retainTokens >= contextWindow * thresholdRatio`. The invariant guarantees the post-compaction history (the bounded summary plus the retained recent tail) is structurally below the threshold, so a compaction never immediately triggers another: consecutive re-compaction is impossible by construction. The bound is strict (`>=` rejects) because the token-pressure gate declines only when the estimate is `< threshold` — a post-compaction history sitting exactly at the threshold would re-trigger.
- **Summarization** — `summarize()`: a `ctx.llm.stream()` call assembled via `BlockAssembler` (the single model-call surface) with a fixed system prompt that asks for a structured checkpoint (Primary Request and Intent · Key Technical Concepts · Files and Code · Errors and Fixes · Pending Tasks · Current Work · Next Step · Critical Context), every section mandatory, exact paths/commands/identifiers preserved. The compacted region is flattened to a plain-text transcript first: text and reasoning contribute their text, and every non-text block (image, tool-call, tool-result, plugin-added types) contributes a type-tagged placeholder (`[image]`, `[tool-call: name(args)]`, …) so the summarizer is told what existed rather than silently dropping it.
- **Summarization** — `summarize()`: a `GenerateOptions` request assembled via `BlockAssembler` with a fixed system prompt that asks for a structured checkpoint (Primary Request and Intent · Key Technical Concepts · Files and Code · Errors and Fixes · Pending Tasks · Current Work · Next Step · Critical Context), every section mandatory, exact paths/commands/identifiers preserved. The request runs through the `agent/request` waterfall before `ctx.llm.stream()`, so router agents that choose the concrete model there also route compaction summaries. The compacted region is flattened to a plain-text transcript first: text and reasoning contribute their text, and every non-text block (image, tool-call, tool-result, plugin-added types) contributes a type-tagged placeholder (`[image]`, `[tool-call: name(args)]`, …) so the summarizer is told what existed rather than silently dropping it.
- **Checkpoint framing** — the raw summary is not landed directly. `compactRegion()` wraps it in a checkpoint preamble (so a resuming model reads it as a checkpoint, not a fresh user request, and builds on the captured context rather than restating it) plus `<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, no-veto) surface-mutation checkpoint that fires after `turn/start` and BEFORE the step opens (`step/start`) and its request history is derived, so compaction mutates the surface — with its log-only `compact/*` records landing cleanly outside any step — and the loop derives once from the result: no double-derive, and the listener cannot see (or need to rewrite) an already-assembled `messages` array. The listener owns no threshold logic of its own (the single token-pressure check lives in `compactIfNeeded()`).

60
packages/compact/compact-basic/src/index.ts
View File

@@ -10,7 +10,7 @@
* compaction declines and retries once it closes).
* - **Summarization** — `ctx.llm.stream()` assembled via `BlockAssembler`
* (the single model-call surface; same path the loop uses) with a fixed
* condense-the-history system prompt.
* condense-the-history system prompt routed through `agent/request`.
* - **Surface mutation** — a single `user/message` replace node carries the
* summary; `compact/*` events are log-only lock + provenance records.
* - **Auto-compaction** — an `agent/pre-step` listener delegates to
@@ -153,12 +153,6 @@ function finishError(finish: FinishReason): Error | undefined {
* context.
*/
export class BasicCompactService extends CompactService {
/**
* `summarize()` reads `ctx.llm.stream()`. Declaring `llm` here lets the cordis
* context proxy resolve it when this service loads as a sibling of LlmService:
* without the inject, `this.ctx.llm` cannot be resolved from this fiber and
* compaction throws at runtime (see postmortem 0001).
*/
static inject = ['llm']
/** Resolved configuration (defaults applied). */
@@ -188,11 +182,11 @@ export class BasicCompactService extends CompactService {
// log-only `compact/*` records and the replacement node cleanly outside a
// step, so a crash mid-compaction leaves an inert orphan the turn-repair
// closes — never a half-open step.
ctx.on('agent/pre-step', async (agent: Agent, _turn: number, _step: number, system: string, model: string, signal: AbortSignal) => {
ctx.on('agent/pre-step', async (agent: Agent, turn: number, step: number, fullSystemPrompt: string, signal: AbortSignal) => {
try {
const result = await this.compactIfNeeded(agent.session, system, model, signal)
const result = await this.compactIfNeeded(agent, turn, step, fullSystemPrompt, signal)
if (result) {
const after = this.estimateTokens(agent.session.deriveMessages(), system)
const after = this.estimateTokens(agent.session.deriveMessages(), fullSystemPrompt)
ctx.logger.info(
`compaction: shadowed ${result.shadowedSeqs.length} surface nodes ` +
`(seqs ${result.shadowedRange.start}-${result.shadowedRange.end}, ` +
@@ -274,8 +268,9 @@ export class BasicCompactService extends CompactService {
}
/**
* Summarize conversation text into content blocks via `ctx.llm.stream()`
* assembled through a `BlockAssembler` (the single model-call surface).
* Summarize conversation text into content blocks via `agent/request` plus
* `ctx.llm.stream()` assembled through a `BlockAssembler` (the single
* model-call surface).
* Override in a subclass for a template or remote summarizer.
*
* Honors the adapter failure contract: an adapter may report a model failure
@@ -286,12 +281,10 @@ export class BasicCompactService extends CompactService {
* Forwards `signal` into `GenerateOptions.signal` so an abort/dispose tears
* down the in-flight summarization rather than orphaning the model call.
*/
async summarize(text: string, model: string, signal?: AbortSignal): Promise<ContentBlock[]> {
if (!model) throw new Error('no model available for summarization')
async summarize(text: string, agent: Agent, turn: number, step: number, signal?: AbortSignal): Promise<ContentBlock[]> {
const assembler = new BlockAssembler()
const options: GenerateOptions = {
model,
model: this.config.summarizationModel || agent.options.model || '',
messages: [{
role: 'user',
content: [{ type: 'text', text: `Summarize this conversation history:\n\n${text}\n\nSummary:` }],
@@ -302,7 +295,11 @@ export class BasicCompactService extends CompactService {
// exactOptionalPropertyTypes: only set `signal` when present — assigning
// `undefined` to an optional `signal?: AbortSignal` is a type error.
if (signal) options.signal = signal
for await (const chunk of this.ctx.llm.stream(options)) {
const request = await this.ctx.waterfall('agent/request', agent, turn, step, options, () => Promise.resolve(options))
if (!request.model) {
throw new Error('no model available for summarization: set BasicCompactConfig.summarizationModel, AgentOptions.model, or supply one via the agent/request waterfall')
}
for await (const chunk of this.ctx.llm.stream(request)) {
assembler.push(chunk)
}
@@ -341,13 +338,15 @@ export class BasicCompactService extends CompactService {
* closes).
*/
override async compactIfNeeded(
session: Session,
system: string,
model: string,
agent: Agent,
turn: number,
step: number,
fullSystemPrompt: string,
signal: AbortSignal,
): Promise<CompactionResult | null> {
const session = agent.session
const messages = session.deriveMessages()
const totalTokens = this.estimateTokens(messages, system)
const totalTokens = this.estimateTokens(messages, fullSystemPrompt)
const threshold = Math.floor(this.config.contextWindow * this.config.thresholdRatio)
if (totalTokens < threshold) return null
@@ -401,14 +400,16 @@ export class BasicCompactService extends CompactService {
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const cutoffSeq = nodes[keepFromIdx - 1]!.seq
return this.compactRegion(session, firstSeq, cutoffSeq, model, signal)
return this.compactRegion(session, firstSeq, cutoffSeq, agent, turn, step, signal)
}
override async compactRegion(
session: Session,
start: number,
end: number,
model: string,
agent: Agent,
turn: number,
step: number,
signal?: AbortSignal,
): Promise<CompactionResult> {
// Resolve the range by surface POSITION, not numeric seq interval. A prior
@@ -458,8 +459,8 @@ export class BasicCompactService extends CompactService {
// strictly inside the open turn (but outside any step). A manual call on a
// fully-closed session has no turn to enclose the events, so reject rather
// than emit an un-enclosed run.
const turn = this._openTurn(session)
if (turn === null) {
const openTurn = this._openTurn(session)
if (openTurn === null) {
throw new Error('compactRegion: no open turn — compaction events must be enclosed in a turn')
}
// Slice the ordered surface nodes [startIdx, endIdx] inclusive — the
@@ -467,13 +468,12 @@ export class BasicCompactService extends CompactService {
const shadowedSeqs = nodes.slice(startIdx, endIdx + 1).map(n => n.seq)
// --- Acquire lock ---
const startEvent = session.append('compact/start', { turn })
const startEvent = session.append('compact/start', { turn: openTurn })
try {
// --- Extract text and summarize ---
const text = this._extractText(session, shadowedSeqs)
const summaryModel = this.config.summarizationModel || model
const summary = await this.summarize(text, summaryModel, signal)
const summary = await this.summarize(text, agent, turn, step, signal)
// Estimate token count of the shadowed content for provenance.
let shadowedTokenCount = 0
@@ -511,7 +511,7 @@ export class BasicCompactService extends CompactService {
// compact/start and here leaves a detectable orphaned lock (a compact/start
// with no matching compact/end) rather than a compact/end that falsely
// claims compaction finished before the surface replacement landed.
const endEvent = session.append('compact/end', { turn })
const endEvent = session.append('compact/end', { turn: openTurn })
return {
startSeq: startEvent.seq,
@@ -526,7 +526,7 @@ export class BasicCompactService extends CompactService {
// Always release the lock — append compact/end with the error so a
// wedged lock is impossible.
const msg = error instanceof Error ? error.message : String(error)
session.append('compact/end', { turn, error: msg })
session.append('compact/end', { turn: openTurn, error: msg })
throw error
}
}

23
packages/compact/compact-basic/src/types.ts
View File

@@ -39,21 +39,20 @@ export const DEFAULTS: ResolvedConfig = {
}
/**
* Apply defaults to a partial config and enforce the single-pass convergence
* Apply defaults to a partial config and enforce the approximate convergence
* invariant.
*
* `summarizationMaxTokens + retainTokens` must be strictly BELOW the compaction
* threshold (`contextWindow * thresholdRatio`). The invariant guarantees that
* after a compaction the derived history — the (bounded) summary plus the
* retained recent tail — is structurally below the threshold, so the very next
* pre-step check passes and a second compaction cannot fire on the same
* content. The bound is strict (`>=` rejects) because `compactIfNeeded` declines
* only when the estimate is `< threshold`: a post-compaction history sitting
* EXACTLY at the threshold would re-trigger on the next check. Without the
* invariant, a too-large summary or retain budget would leave the
* post-compaction history at/over threshold, triggering compaction again and
* again. Pre-release we reject rather than clamp: a config that cannot guarantee
* convergence is a bug at the call site, not something to silently paper over.
* threshold (`contextWindow * thresholdRatio`). The invariant bounds the two
* variable pieces of post-compaction history — the summary and the retained
* recent tail — but it is intentionally approximate: checkpoint framing,
* per-message role overhead, system-prompt size, and the char/4 estimator's
* error can still leave a narrow accepted config near the threshold. The bound
* is strict (`>=` rejects) because `compactIfNeeded` declines only when the
* estimate is `< threshold`: a post-compaction history sitting EXACTLY at the
* threshold would re-trigger on the next check. Pre-release we reject rather
* than clamp: a config that cannot satisfy even this structural bound is a bug
* at the call site, not something to silently paper over.
*
* @throws if `summarizationMaxTokens + retainTokens >= contextWindow * thresholdRatio`.
*/

192
packages/compact/compact-basic/tests/compact-basic.spec.ts
View File

@@ -29,7 +29,8 @@ class TestCompactService extends BasicCompactService {
return blocks.length * 10
}
override async summarize(text: string, model: string): Promise<ContentBlock[]> {
override async summarize(text: string, agent: Agent): Promise<ContentBlock[]> {
const model = this.config.summarizationModel || agent.options.model || ''
this.summarizeCalls.push({ text, model })
if (this.summarizeError) throw this.summarizeError
return this.mockSummary
@@ -184,7 +185,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai
const svc = createTestService({ contextWindow: 200, thresholdRatio: 0.5, retainTokens: 55 })
const session = toolTurnSession(3)
const result = await svc.compactIfNeeded(session, '', 'm', SIGNAL)
const result = await compactIfNeeded(svc, session, '', 'm', SIGNAL)
expect(result).not.toBeNull()
expect(result!.shadowedSeqs.length).toBeGreaterThan(0)
// No dangling tool-result: every compacted/retained step stayed whole.
@@ -214,7 +215,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai
// Turn stays open.
const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.1, retainTokens: 5 })
const result = await svc.compactIfNeeded(s, '', 'm', SIGNAL)
const result = await compactIfNeeded(svc, s, '', 'm', SIGNAL)
expect(result).toBeNull()
expect(s.events.some(e => e.type === 'compact/start')).toBe(false)
})
@@ -227,7 +228,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai
const resultSeq = nodes[2]!.seq
// start = the tool/result: its issuing assistant precedes it IN THE SAME STEP,
// so starting here would orphan that assistant's tool-call. end is fine (user).
await expect(svc.compactRegion(session, resultSeq, resultSeq, 'm'))
await expect(compactRegion(svc, session, resultSeq, resultSeq, 'm'))
.rejects.toThrow(/start seq .* is not a balanced boundary/)
expect(userSeq).toBeLessThan(resultSeq) // sanity: ordering as expected
})
@@ -240,7 +241,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai
const asstSeq = nodes[1]!.seq
// end = the assistant/message: its tool/result follows IN THE SAME STEP, so
// ending here would strand that result. start is fine (the pre-step user).
await expect(svc.compactRegion(session, userSeq, asstSeq, 'm'))
await expect(compactRegion(svc, session, userSeq, asstSeq, 'm'))
.rejects.toThrow(/end seq .* is not a balanced boundary/)
})
@@ -257,7 +258,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai
const nodes = s.surface.nodes // [user, asst]
const userSeq = nodes[0]!.seq
const asstSeq = nodes[1]!.seq
await expect(svc.compactRegion(s, userSeq, asstSeq, 'm'))
await expect(compactRegion(svc, s, userSeq, asstSeq, 'm'))
.rejects.toThrow(/end seq .* is not a balanced boundary/)
})
@@ -267,7 +268,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai
const nodes = session.surface.nodes // [user1, asst1, res1, user2, asst2, res2]
const startSeq = nodes[0]!.seq // pre-step user1 (free boundary)
const endSeq = nodes[2]!.seq // res1 = last node of turn 1's closed step
const result = await svc.compactRegion(session, startSeq, endSeq, 'm')
const result = await compactRegion(svc, session, startSeq, endSeq, 'm')
expect(result.shadowedRange).toEqual({ start: startSeq, end: endSeq })
expectNoOrphanToolResults(session.deriveMessages())
})
@@ -277,7 +278,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai
const session = toolTurnSession(1)
const nodes = session.surface.nodes
const userSeq = nodes[0]!.seq // pre-step user: free boundary both ways
const result = await svc.compactRegion(session, userSeq, userSeq, 'm')
const result = await compactRegion(svc, session, userSeq, userSeq, 'm')
expect(result.shadowedRange).toEqual({ start: userSeq, end: userSeq })
})
@@ -292,7 +293,7 @@ describe('BasicCompactService step-alignment (never split a tool-call/result pai
s.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
const nodes = s.surface.nodes
const ctxSeq = nodes[0]!.seq
const result = await svc.compactRegion(s, ctxSeq, ctxSeq, 'm')
const result = await compactRegion(svc, s, ctxSeq, ctxSeq, 'm')
expect(result.shadowedRange).toEqual({ start: ctxSeq, end: ctxSeq })
})
})
@@ -352,7 +353,7 @@ describe('BasicCompactService.compactRegion', () => {
const firstSeq = nodes[0]!.seq
const secondSeq = nodes[1]!.seq
const result = await svc.compactRegion(session, firstSeq, secondSeq, 'test-model')
const result = await compactRegion(svc, session, firstSeq, secondSeq, 'test-model')
expect(result.shadowedSeqs).toEqual([firstSeq, secondSeq])
expect(result.shadowedRange.start).toBe(firstSeq)
@@ -405,7 +406,7 @@ describe('BasicCompactService.compactRegion', () => {
it('throws when start or end are not surface nodes', async () => {
const svc = createTestService()
const session = multiTurnSession(1, 1)
await expect(svc.compactRegion(session, 999, 1000, 'm'))
await expect(compactRegion(svc, session, 999, 1000, 'm'))
.rejects.toThrow(/start seq 999 not found in surface/)
})
@@ -413,7 +414,7 @@ describe('BasicCompactService.compactRegion', () => {
const svc = createTestService()
const session = multiTurnSession(2, 1)
const nodes = session.surface.nodes
await expect(svc.compactRegion(session, nodes[1]!.seq, nodes[0]!.seq, 'm'))
await expect(compactRegion(svc, session, nodes[1]!.seq, nodes[0]!.seq, 'm'))
.rejects.toThrow(/is after end seq .* on the surface/)
})
@@ -422,7 +423,7 @@ describe('BasicCompactService.compactRegion', () => {
const session = multiTurnSession(2, 1)
const nodes = session.surface.nodes
session.append('compact/start', { turn: 2 })
await expect(svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm'))
await expect(compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm'))
.rejects.toThrow(/compaction already in progress/)
})
@@ -432,7 +433,7 @@ describe('BasicCompactService.compactRegion', () => {
const session = multiTurnSession(2, 1)
const nodes = session.surface.nodes
await expect(svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm'))
await expect(compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm'))
.rejects.toThrow('model unavailable')
const endEvent = session.events.findLast(e => e.type === 'compact/end')
@@ -455,7 +456,7 @@ describe('BasicCompactService.compactRegion', () => {
const session = multiTurnSession(1, 2)
const nodes = session.surface.nodes
await svc.compactRegion(session, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm')
await compactRegion(svc, session, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm')
expect(svc.summarizeCalls.length).toBe(1)
const { text, model } = svc.summarizeCalls[0]!
@@ -470,7 +471,7 @@ describe('BasicCompactService.compactRegion', () => {
const session = multiTurnSession(3, 1)
const nodes = session.surface.nodes
const result = await svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm')
const result = await compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm')
// Provenance (compact/summary) carries the RAW, unframed summary.
expect(result.summary).toEqual([{ type: 'text', text: 'STRUCTURED SUMMARY' }])
@@ -492,7 +493,7 @@ describe('BasicCompactService.compactRegion', () => {
const firstSeq = nodes[0]!.seq
const lastSeq = nodes[nodes.length - 1]!.seq
await svc.compactRegion(session, firstSeq, lastSeq, 'm')
await compactRegion(svc, session, firstSeq, lastSeq, 'm')
expect(svc.summarizeCalls.length).toBe(1)
const { text } = svc.summarizeCalls[0]!
@@ -506,14 +507,14 @@ describe('BasicCompactService.compactIfNeeded', () => {
it('returns null when tokens are under threshold', async () => {
const svc = createTestService({ contextWindow: 128000, thresholdRatio: 0.8 })
const session = multiTurnSession(1, 1)
expect(await svc.compactIfNeeded(session, '', 'm', SIGNAL)).toBeNull()
expect(await compactIfNeeded(svc, session, '', 'm', SIGNAL)).toBeNull()
})
it('compacts when tokens exceed threshold', async () => {
const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.5, retainTokens: 10 })
const session = multiTurnSession(3, 1) // 6 surface nodes, 10 tokens each = 60
const result = await svc.compactIfNeeded(session, '', 'm', SIGNAL)
const result = await compactIfNeeded(svc, session, '', 'm', SIGNAL)
expect(result).not.toBeNull()
expect(result!.shadowedSeqs.length).toBeGreaterThan(0)
})
@@ -522,7 +523,7 @@ describe('BasicCompactService.compactIfNeeded', () => {
const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.2, retainTokens: 15 })
const session = multiTurnSession(5, 1) // 10 surface nodes = ~100 tokens
const result = await svc.compactIfNeeded(session, '', 'm', SIGNAL)
const result = await compactIfNeeded(svc, session, '', 'm', SIGNAL)
expect(result).not.toBeNull()
const nodes = session.surface.nodes
expect(result!.shadowedSeqs.length).toBeGreaterThan(0)
@@ -538,7 +539,7 @@ describe('BasicCompactService.compactIfNeeded', () => {
// summarizationMaxTokens (1) + retainTokens (45) = 46 < threshold 47.
const svc = createTestService({ contextWindow: 470, thresholdRatio: 0.1, retainTokens: 45 })
const session = multiTurnSession(2, 1)
expect(await svc.compactIfNeeded(session, '', 'm', SIGNAL)).toBeNull()
expect(await compactIfNeeded(svc, session, '', 'm', SIGNAL)).toBeNull()
})
it('compacts a runaway turn: its early CLOSED steps summarize while recent steps stay verbatim', async () => {
@@ -572,7 +573,7 @@ describe('BasicCompactService.compactIfNeeded', () => {
const nodesBefore = s.surface.nodes.length
expect(nodesBefore).toBe(11)
const result = await svc.compactIfNeeded(s, '', 'm', SIGNAL)
const result = await compactIfNeeded(svc, s, '', 'm', SIGNAL)
expect(result).not.toBeNull()
// Early steps of the SAME open turn were shadowed (impossible under layer 2).
expect(result!.shadowedSeqs.length).toBeGreaterThan(0)
@@ -587,7 +588,7 @@ describe('BasicCompactService.compactIfNeeded', () => {
it('returns null for an empty surface', async () => {
const svc = createTestService({ contextWindow: 100, thresholdRatio: 0.5, retainTokens: 10 })
const session = new Session(SessionId('empty'))
expect(await svc.compactIfNeeded(session, '', 'm', SIGNAL)).toBeNull()
expect(await compactIfNeeded(svc, session, '', 'm', SIGNAL)).toBeNull()
})
it('compacts again after a prior summary node heads the surface (the summary stays eligible)', async () => {
@@ -600,7 +601,7 @@ describe('BasicCompactService.compactIfNeeded', () => {
const svc = createTestService({ contextWindow: 300, thresholdRatio: 0.1, retainTokens: 25 })
const s = multiTurnSession(4, 1) // turns 1-4 closed, turn 5 open (no surface yet)
const first = await svc.compactIfNeeded(s, '', 'm', SIGNAL)
const first = await compactIfNeeded(svc, s, '', 'm', SIGNAL)
expect(first).not.toBeNull()
// The summary node now heads the surface with a fresh high seq.
const summaryHeadSeq = s.surface.nodes[0]!.seq
@@ -615,7 +616,7 @@ describe('BasicCompactService.compactIfNeeded', () => {
s.append('assistant/message', { turn: 5, step: 1, content: [{ type: 'text', text: 'reply 5' }] }, { surfaceOp: 'append' })
s.append('step/end', { turn: 5, step: 1 })
const second = await svc.compactIfNeeded(s, '', 'm', SIGNAL)
const second = await compactIfNeeded(svc, s, '', 'm', SIGNAL)
expect(second).not.toBeNull()
expect(second!.shadowedSeqs.length).toBeGreaterThan(0)
// The fresh open-turn nodes were NOT compacted.
@@ -630,7 +631,7 @@ describe('BasicCompactService replay equivalence', () => {
const session = multiTurnSession(3, 1)
const nodes = session.surface.nodes
await svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm')
await compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm')
const derived = session.deriveMessages()
const replayed = new Session(SessionId('replay'), [...session.events])
@@ -646,7 +647,7 @@ describe('BasicCompactService blocking (compaction in progress)', () => {
const nodes = session.surface.nodes
// Whole step (user → assistant) is a step-aligned region, so the call reaches
// the in-progress check rather than being rejected for splitting a step.
await expect(svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm'))
await expect(compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm'))
.rejects.toThrow(/compaction already in progress/)
})
@@ -656,7 +657,7 @@ describe('BasicCompactService blocking (compaction in progress)', () => {
const nodes = session.surface.nodes
session.append('compact/start', { turn: 1 })
session.append('compact/end', { turn: 1 })
const result = await svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm')
const result = await compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm')
expect(result).toBeDefined()
})
@@ -679,7 +680,7 @@ describe('BasicCompactService blocking (compaction in progress)', () => {
const nodes = s.surface.nodes
// The stale start is before the turn/end, so it is NOT seen as in-progress.
const result = await svc.compactRegion(s, nodes[0]!.seq, nodes[1]!.seq, 'm')
const result = await compactRegion(svc, s, nodes[0]!.seq, nodes[1]!.seq, 'm')
expect(result).toBeDefined()
})
})
@@ -829,12 +830,37 @@ function stubAgent(session: Session, model?: string): Agent {
return { session, options: { model } } as unknown as Agent
}
function compactIfNeeded(
svc: BasicCompactService,
session: Session,
fullSystemPrompt: string,
model: string,
signal: AbortSignal,
) {
return svc.compactIfNeeded(stubAgent(session, model), 1, 1, fullSystemPrompt, signal)
}
function compactRegion(
svc: BasicCompactService,
session: Session,
start: number,
end: number,
model: string,
signal?: AbortSignal,
) {
return svc.compactRegion(session, start, end, stubAgent(session, model), 1, 1, signal)
}
function summarize(svc: BasicCompactService, text: string, model: string) {
return svc.summarize(text, stubAgent(new Session(SessionId('summary')), model), 1, 1)
}
describe('BasicCompactService.summarize (real ctx.llm.stream)', () => {
it('summarizes via the registered adapter and returns its content', async () => {
const { ctx, adapter } = await ctxWithModel('SUMMARY TEXT')
const svc = new BasicCompactService(ctx, { auto: false, summarizationMaxTokens: 512 })
const summary = await svc.summarize('User: hi\n\nAssistant: hello', 'test-model')
const summary = await summarize(svc, 'User: hi\n\nAssistant: hello', 'test-model')
expect(summary).toEqual([{ type: 'text', text: 'SUMMARY TEXT' }])
// The fixed system prompt and maxTokens flow through.
expect(adapter.lastOptions!.system).toContain('compaction engine')
@@ -846,19 +872,19 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => {
it('throws when no model is provided', async () => {
const { ctx } = await ctxWithModel('x')
const svc = new BasicCompactService(ctx, { auto: false })
await expect(svc.summarize('text', '')).rejects.toThrow(/no model available/)
await expect(summarize(svc, 'text', '')).rejects.toThrow(/no model available/)
})
it('rethrows when the stream ends with a finish-error chunk', async () => {
const ctx = await ctxWithFinish({ kind: 'error', message: 'provider 401', code: 'UNAUTHORIZED' })
const svc = new BasicCompactService(ctx, { auto: false })
await expect(svc.summarize('text', 'test-model')).rejects.toMatchObject({ message: 'provider 401', code: 'UNAUTHORIZED' })
await expect(summarize(svc, 'text', 'test-model')).rejects.toMatchObject({ message: 'provider 401', code: 'UNAUTHORIZED' })
})
it('rethrows a finish-error chunk without a code (code stays undefined)', async () => {
const ctx = await ctxWithFinish({ kind: 'error', message: 'opaque failure' })
const svc = new BasicCompactService(ctx, { auto: false })
const error = await svc.summarize('text', 'test-model').then(() => null, (e: unknown) => e as Error & { code?: string })
const error = await summarize(svc, 'text', 'test-model').then(() => null, (e: unknown) => e as Error & { code?: string })
expect(error?.message).toBe('opaque failure')
expect(error?.code).toBeUndefined()
})
@@ -866,13 +892,13 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => {
it('rethrows when the stream ends with a finish-aborted chunk', async () => {
const ctx = await ctxWithFinish({ kind: 'aborted' })
const svc = new BasicCompactService(ctx, { auto: false })
await expect(svc.summarize('text', 'test-model')).rejects.toMatchObject({ message: 'summarization stream aborted', code: 'ABORTED' })
await expect(summarize(svc, 'text', 'test-model')).rejects.toMatchObject({ message: 'summarization stream aborted', code: 'ABORTED' })
})
it('fails closed on a max-tokens finish (an incomplete checkpoint must not commit)', async () => {
const ctx = await ctxWithFinish({ kind: 'max-tokens' })
const svc = new BasicCompactService(ctx, { auto: false })
await expect(svc.summarize('text', 'test-model')).rejects.toMatchObject({ code: 'MAX_TOKENS' })
await expect(summarize(svc, 'text', 'test-model')).rejects.toMatchObject({ code: 'MAX_TOKENS' })
})
it('compactRegion leaves the surface intact when summarization hits max-tokens', async () => {
@@ -882,7 +908,7 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => {
const before = [...session.surface.nodes]
const nodes = session.surface.nodes
await expect(svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'test-model'))
await expect(compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'test-model'))
.rejects.toMatchObject({ code: 'MAX_TOKENS' })
// No replacement landed — the surface is byte-identical, and the lock was
@@ -899,7 +925,7 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => {
const session = multiTurnSession(2, 1)
const nodes = session.surface.nodes
const result = await svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'test-model')
const result = await compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'test-model')
expect(result.summary).toEqual([{ type: 'text', text: 'CONDENSED' }])
// The raw summary is wrapped in the checkpoint framing on the surface.
expect(session.deriveMessages()[0]!.content).toContainEqual({ type: 'text', text: 'CONDENSED' })
@@ -908,8 +934,8 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => {
describe('BasicCompactService auto-compaction (agent/pre-step listener)', () => {
/** Fire the agent/pre-step serial checkpoint as the loop does. */
function firePreStep(ctx: Context, agent: Agent, step: number, system: string, model: string): Promise<unknown> {
return ctx.serial('agent/pre-step', agent, 1, step, system, model, SIGNAL)
function firePreStep(ctx: Context, agent: Agent, step: number, fullSystemPrompt: string): Promise<unknown> {
return ctx.serial('agent/pre-step', agent, 1, step, fullSystemPrompt, SIGNAL)
}
it('compacts (mutating the surface) when over threshold', async () => {
@@ -919,7 +945,7 @@ describe('BasicCompactService auto-compaction (agent/pre-step listener)', () =>
const agent = stubAgent(session, 'test-model')
const before = session.surface.nodes.length
await firePreStep(ctx, agent, 1, '', 'test-model')
await firePreStep(ctx, agent, 1, '')
// The surface shrank in place, and a summary checkpoint landed.
expect(session.surface.nodes.length).toBeLessThan(before)
@@ -936,7 +962,7 @@ describe('BasicCompactService auto-compaction (agent/pre-step listener)', () =>
// A step-2 checkpoint (a tool-heavy turn's later step) must still compact —
// the surface accumulated assistant/message + tool/result nodes since step 1.
await firePreStep(ctx, agent, 2, '', 'test-model')
await firePreStep(ctx, agent, 2, '')
expect(session.events.some(e => e.type === 'compact/start')).toBe(true)
})
@@ -946,7 +972,7 @@ describe('BasicCompactService auto-compaction (agent/pre-step listener)', () =>
const session = multiTurnSession(1, 1)
const agent = stubAgent(session, 'test-model')
await firePreStep(ctx, agent, 1, '', 'test-model')
await firePreStep(ctx, agent, 1, '')
expect(session.events.some(e => e.type === 'compact/start')).toBe(false)
})
@@ -960,7 +986,7 @@ describe('BasicCompactService auto-compaction (agent/pre-step listener)', () =>
const agent = stubAgent(session, 'missing-model')
const before = session.surface.nodes.length
await firePreStep(ctx, agent, 1, '', 'missing-model')
await firePreStep(ctx, agent, 1, '')
// No summary landed; the surface is unchanged.
expect(session.events.some(e => e.type === 'compact/summary')).toBe(false)
expect(session.surface.nodes.length).toBe(before)
@@ -972,9 +998,44 @@ describe('BasicCompactService auto-compaction (agent/pre-step listener)', () =>
const session = multiTurnSession(3, 1)
const agent = stubAgent(session, 'test-model')
await firePreStep(ctx, agent, 1, '', 'test-model')
await firePreStep(ctx, agent, 1, '')
expect(session.events.some(e => e.type === 'compact/start')).toBe(false)
})
it('routes summarization through agent/request so router agents can choose the model', async () => {
const { ctx, adapter } = await ctxWithModel('ROUTED SUMMARY', 'routed-model')
ctx.on('agent/request', async (_agent, _turn, _step, options, next) => {
options.model = 'routed-model'
return next()
})
void new BasicCompactService(ctx, { contextWindow: 200, thresholdRatio: 0.5, retainTokens: 20, summarizationMaxTokens: 50 })
const session = multiTurnSession(5, 1)
const agent = stubAgent(session)
await ctx.serial('agent/pre-step', agent, 1, 1, '', SIGNAL)
expect(adapter.lastOptions?.model).toBe('routed-model')
expect(session.events.some(e => e.type === 'compact/summary')).toBe(true)
expect(session.deriveMessages()[0]!.content).toContainEqual({ type: 'text', text: 'ROUTED SUMMARY' })
})
it('removes the auto pre-step listener when the plugin fiber is disposed', async () => {
const { ctx } = await ctxWithModel('SUMMARY')
const fiber = await ctx.plugin(BasicCompactService, {
contextWindow: 200,
thresholdRatio: 0.5,
retainTokens: 20,
summarizationMaxTokens: 50,
})
const session = multiTurnSession(5, 1)
const agent = stubAgent(session, 'test-model')
await fiber.dispose()
await firePreStep(ctx, agent, 1, '')
expect(session.events.some(e => e.type === 'compact/start')).toBe(false)
expect(ctx.get('compact')).toBeUndefined()
})
})
describe('BasicCompactService._extractText branches', () => {
@@ -1001,7 +1062,7 @@ describe('BasicCompactService._extractText branches', () => {
s.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
const nodes = s.surface.nodes
await svc.compactRegion(s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm')
await compactRegion(svc, s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm')
const { text } = svc.summarizeCalls[0]!
expect(text).toContain('[Context: project context here]')
@@ -1030,7 +1091,7 @@ describe('BasicCompactService._extractText branches', () => {
s.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
const nodes = s.surface.nodes
await svc.compactRegion(s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm')
await compactRegion(svc, s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm')
expect(svc.summarizeCalls[0]!.text).toContain('Tool error (call c9): boom failure')
})
})
@@ -1064,7 +1125,7 @@ describe('BasicCompactService edge cases', () => {
s.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
const nodes = s.surface.nodes
await svc.compactRegion(s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm')
await compactRegion(svc, s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm')
const { text } = svc.summarizeCalls[0]!
expect(text).toContain('[tool-result: [image]]') // nested tool-result with content
expect(text).toContain('[custom-widget]') // unknown block placeholder
@@ -1089,7 +1150,7 @@ describe('BasicCompactService edge cases', () => {
const session = multiTurnSession(4, 1)
const agent = stubAgent(session, 'test-model')
await ctx.serial('agent/pre-step', agent, 1, 1, '', 'test-model', SIGNAL)
await ctx.serial('agent/pre-step', agent, 1, 1, '', SIGNAL)
expect(session.events.some(e => e.type === 'compact/summary')).toBe(true)
// The surface was mutated; the head message is the framed summary checkpoint.
expect(session.deriveMessages()[0]!.content).toContainEqual({ type: 'text', text: 'SUMMARY' })
@@ -1111,7 +1172,7 @@ describe('BasicCompactService edge cases', () => {
s.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
const nodes = s.surface.nodes
await expect(svc.compactRegion(s, nodes[0]!.seq, nodes[1]!.seq, 'm'))
await expect(compactRegion(svc, s, nodes[0]!.seq, nodes[1]!.seq, 'm'))
.rejects.toThrow(/no open turn/)
// The lock was never acquired — no compact/start landed.
expect(s.events.some(e => e.type === 'compact/start')).toBe(false)
@@ -1126,7 +1187,7 @@ describe('BasicCompactService edge cases', () => {
s.append('user/message', { content: [{ type: 'text', text: 'orphan' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
const nodes = s.surface.nodes
await expect(svc.compactRegion(s, nodes[0]!.seq, nodes[0]!.seq, 'm'))
await expect(compactRegion(svc, s, nodes[0]!.seq, nodes[0]!.seq, 'm'))
.rejects.toThrow(/no open turn/)
expect(s.events.some(e => e.type === 'compact/start')).toBe(false)
})
@@ -1136,14 +1197,14 @@ describe('BasicCompactService edge cases', () => {
const session = new Session(SessionId('empty-but-pressured'))
// No surface nodes, but a large system prompt pushes the estimate over threshold.
const bigPrompt = 'x'.repeat(800) // ceil(800/4) = 200 tokens >> threshold 100
expect(await svc.compactIfNeeded(session, bigPrompt, 'm', SIGNAL)).toBeNull()
expect(await compactIfNeeded(svc, session, bigPrompt, 'm', SIGNAL)).toBeNull()
})
it('compactRegion throws when end is not a surface node (start valid)', async () => {
const svc = createTestService()
const session = multiTurnSession(1, 1)
const nodes = session.surface.nodes
await expect(svc.compactRegion(session, nodes[0]!.seq, 9999, 'm'))
await expect(compactRegion(svc, session, nodes[0]!.seq, 9999, 'm'))
.rejects.toThrow(/end seq 9999 not found in surface/)
})
@@ -1155,7 +1216,7 @@ describe('BasicCompactService edge cases', () => {
const nodes = session.surface.nodes
// Whole step (user → assistant): a step-aligned region that reaches summarize.
await expect(svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'm')).rejects.toBe('plain string failure')
await expect(compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm')).rejects.toBe('plain string failure')
const endEvent = session.events.findLast(e => e.type === 'compact/end')!
expect(endEvent.data).toMatchObject({ error: 'plain string failure' })
})
@@ -1170,7 +1231,7 @@ describe('BasicCompactService edge cases', () => {
const agent = stubAgent(session, 'test-model')
const before = session.surface.nodes.length
await ctx.serial('agent/pre-step', agent, 1, 1, '', 'test-model', SIGNAL)
await ctx.serial('agent/pre-step', agent, 1, 1, '', SIGNAL)
// The failure was swallowed; the surface is untouched and a warning logged.
expect(session.surface.nodes.length).toBe(before)
expect(session.events.some(e => e.type === 'compact/summary')).toBe(false)
@@ -1187,7 +1248,7 @@ describe('BasicCompactService edge cases', () => {
const agent = stubAgent(session, 'test-model')
const bigSystem = 'x'.repeat(900) // ceil(900/4)=225 > threshold 200
await ctx.serial('agent/pre-step', agent, 1, 1, bigSystem, 'test-model', SIGNAL)
await ctx.serial('agent/pre-step', agent, 1, 1, bigSystem, SIGNAL)
expect(session.events.some(e => e.type === 'compact/start')).toBe(false)
expect(svc.summarizeCalls.length).toBe(0)
})
@@ -1221,7 +1282,7 @@ describe('BasicCompactService edge cases', () => {
s.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
const nodes = s.surface.nodes
await svc.compactRegion(s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm')
await compactRegion(svc, s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm')
// Every empty-content message (user text, empty reasoning, empty-content
// tool/result, empty context, empty steering) extracted to nothing and was
// skipped — the only surviving line is the assistant's tool-call (which a
@@ -1256,7 +1317,7 @@ describe('BasicCompactService edge cases', () => {
s.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
const nodes = s.surface.nodes
await svc.compactRegion(s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm')
await compactRegion(svc, s, nodes[0]!.seq, nodes[nodes.length - 1]!.seq, 'm')
const { text } = svc.summarizeCalls[0]!
// Every non-text block surfaces as a placeholder rather than being dropped.
expect(text).toContain('User: [image]')
@@ -1280,7 +1341,7 @@ describe('BasicCompactService positional range (surface seqs are not monotonic a
// First compaction: shadow the two oldest surface nodes.
const nodes0 = session.surface.nodes
const first = await svc.compactRegion(session, nodes0[0]!.seq, nodes0[1]!.seq, 'm')
const first = await compactRegion(svc, session, nodes0[0]!.seq, nodes0[1]!.seq, 'm')
// The summary node now sits at the head with a seq HIGHER than the
// retained older nodes that follow it — the non-monotonic surface. (The
@@ -1298,7 +1359,7 @@ describe('BasicCompactService positional range (surface seqs are not monotonic a
const startSeq = nodes1[0]!.seq
const endSeq = nodes1[2]!.seq
expect(startSeq).toBeGreaterThan(endSeq)
const second = await svc.compactRegion(session, startSeq, endSeq, 'm')
const second = await compactRegion(svc, session, startSeq, endSeq, 'm')
// Exactly the three nodes at surface positions [0..2] are shadowed, in
// surface order — the positional slice, regardless of their seq values.
@@ -1316,14 +1377,14 @@ describe('BasicCompactService positional range (surface seqs are not monotonic a
// First compaction shadows the oldest two surface nodes, landing a high-seq
// summary node at the head.
const n0 = session.surface.nodes
await svc.compactRegion(session, n0[0]!.seq, n0[1]!.seq, 'm')
await compactRegion(svc, session, n0[0]!.seq, n0[1]!.seq, 'm')
// Second compaction spans [head summary … turn-2's step end]. The head's seq
// is higher than the older retained nodes' seqs, so a log-seq-order walk
// would emit the older messages BEFORE the checkpoint.
const n1 = session.surface.nodes
svc.summarizeCalls = []
await svc.compactRegion(session, n1[0]!.seq, n1[2]!.seq, 'm')
await compactRegion(svc, session, n1[0]!.seq, n1[2]!.seq, 'm')
// The extracted transcript follows surface order: the checkpoint (head)
// first, then the older retained messages — matching deriveMessages().
@@ -1355,7 +1416,7 @@ describe('BasicCompactService llm inject (real plugin-load path)', () => {
const svc = ctx.compact as BasicCompactService
const session = multiTurnSession(2, 1)
const nodes = session.surface.nodes
const result = await svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'test-model')
const result = await compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'test-model')
expect(result.summary).toEqual([{ type: 'text', text: 'CONDENSED' }])
// Tear the fiber down so this test owns no leaked registration; the
@@ -1403,7 +1464,7 @@ describe('BasicCompactService under the real invariants plugin', () => {
const nodes = session.surface.nodes
// No invariant throws here: compact/* + the replacement are all in turn 3.
const result = await svc.compactRegion(session, nodes[0]!.seq, nodes[1]!.seq, 'test-model')
const result = await compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'test-model')
expect(result.shadowedSeqs.length).toBe(2)
expect(session.surface.nodes[0]!.seq).toBeGreaterThan(session.surface.nodes[1]!.seq)
})
@@ -1416,15 +1477,14 @@ describe('BasicCompactService under the real invariants plugin', () => {
session.append('turn/start', { turn: 4, trigger: { kind: 'message', source: { kind: 'user' } } })
const n0 = session.surface.nodes
await svc.compactRegion(session, n0[0]!.seq, n0[1]!.seq, 'test-model')
await compactRegion(svc, session, n0[0]!.seq, n0[1]!.seq, 'test-model')
// Surface head now carries a higher seq than the older retained nodes. A
// second compaction spanning [head … a later closed-step end] must pass the
// invariants' positional replace check even though startSeq > endSeq.
const n1 = session.surface.nodes
expect(n1[0]!.seq).toBeGreaterThan(n1[2]!.seq)
const second = await svc.compactRegion(session, n1[0]!.seq, n1[2]!.seq, 'test-model')
const second = await compactRegion(svc, session, n1[0]!.seq, n1[2]!.seq, 'test-model')
expect(second.shadowedSeqs).toEqual([n1[0]!.seq, n1[1]!.seq, n1[2]!.seq])
})
})

6
packages/compact/compact/README.md
View File

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

28
packages/compact/compact/src/index.ts
View File

@@ -27,6 +27,12 @@ import type { CompactionResult } from './types.ts'
export type { CompactionResult } from './types.ts'
/** Minimal agent context compaction needs without depending on the agent package. */
export interface CompactAgentContext {
session: Session
options: { model?: string }
}
declare module 'cordis' {
interface Context {
compact: CompactService
@@ -84,9 +90,10 @@ export abstract class CompactService extends Service {
* exceeds the budget, compaction cannot help and the call may go out
* over-budget. Bounding an individual unit's size is a separate concern.
*
* @param session - the session whose surface may be compacted.
* @param system - the assembled system prompt, counted toward the estimate.
* @param model - the summarization model (a backend may override via config).
* @param agent - agent context owning the session surface and model options.
* @param turn - turn number of the pre-step checkpoint.
* @param step - step number about to start.
* @param fullSystemPrompt - assembled system prompt, counted toward the estimate.
* @param signal - cancellation signal. A backend summarizing via
* `ctx.llm.stream()` MUST forward this into the call's `GenerateOptions.signal`
* so an abort/dispose tears down the in-flight summarization rather than
@@ -94,9 +101,10 @@ export abstract class CompactService extends Service {
* @returns the compaction result, or `null` if no compaction was needed.
*/
abstract compactIfNeeded(
session: Session,
system: string,
model: string,
agent: CompactAgentContext,
turn: number,
step: number,
fullSystemPrompt: string,
signal: AbortSignal,
): Promise<CompactionResult | null>
@@ -120,7 +128,9 @@ export abstract class CompactService extends Service {
* @param session - the session whose surface is mutated.
* @param start - inclusive seq of the first surface node to compact.
* @param end - inclusive seq of the last surface node to compact.
* @param model - summarization model.
* @param agent - agent context used by router-aware summarizers.
* @param turn - lifecycle turn forwarded to request-routing seams.
* @param step - lifecycle step forwarded to request-routing seams.
* @param signal - optional cancellation signal. A backend that summarizes via
* `ctx.llm.stream()` MUST forward this into the call's `GenerateOptions.signal`
* so an abort/dispose tears down the in-flight summarization rather than
@@ -136,7 +146,9 @@ export abstract class CompactService extends Service {
session: Session,
start: number,
end: number,
model: string,
agent: CompactAgentContext,
turn: number,
step: number,
signal?: AbortSignal,
): Promise<CompactionResult>
}

27
packages/compact/compact/tests/compact.spec.ts
View File

@@ -3,6 +3,7 @@ import { Context } from 'cordis'
import { CompactService } from '@deepseek-ai/dsh-compact'
import type { CompactionResult } from '@deepseek-ai/dsh-compact'
import { Session, SessionId } from '@deepseek-ai/dsh-session'
import type { CompactAgentContext } from '@deepseek-ai/dsh-compact'
/**
* A trivial concrete CompactService implementing the abstract contract. The
@@ -15,10 +16,11 @@ class StubCompactService extends CompactService {
lastSignal: AbortSignal | undefined
override async compactIfNeeded(
_session: Session,
_systemPrompt?: string,
_model?: string,
signal?: AbortSignal,
_agent: CompactAgentContext,
_turn: number,
_step: number,
_fullSystemPrompt: string,
signal: AbortSignal,
): Promise<CompactionResult | null> {
this.lastSignal = signal
return null
@@ -28,7 +30,9 @@ class StubCompactService extends CompactService {
session: Session,
start: number,
end: number,
_model: string,
_agent: CompactAgentContext,
_turn: number,
_step: number,
signal?: AbortSignal,
): Promise<CompactionResult> {
this.lastSignal = signal
@@ -54,6 +58,10 @@ class StubCompactService extends CompactService {
}
describe('CompactService seam', () => {
function stubAgent(session: Session, model?: string): CompactAgentContext {
return { session, options: model === undefined ? {} : { model } }
}
it('registers as ctx.compact', () => {
const ctx = new Context()
void new StubCompactService(ctx)
@@ -72,7 +80,8 @@ describe('CompactService seam', () => {
it('exposes the abstract contract methods', async () => {
const ctx = new Context()
const svc = new StubCompactService(ctx)
expect(await svc.compactIfNeeded(new Session(SessionId('s')))).toBeNull()
const session = new Session(SessionId('s'))
expect(await svc.compactIfNeeded(stubAgent(session), 1, 1, '', new AbortController().signal)).toBeNull()
})
it('compact/* events merge into SessionEventMap and are log-only', async () => {
@@ -80,7 +89,7 @@ describe('CompactService seam', () => {
const svc = new StubCompactService(ctx)
const session = new Session(SessionId('s'))
const result = await svc.compactRegion(session, 0, 0, 'm')
const result = await svc.compactRegion(session, 0, 0, stubAgent(session, 'm'), 1, 1)
const startEvent = session.events.find(e => e.type === 'compact/start')
expect(startEvent).toBeDefined()
@@ -98,10 +107,10 @@ describe('CompactService seam', () => {
const session = new Session(SessionId('s'))
const controller = new AbortController()
await svc.compactRegion(session, 0, 0, 'm', controller.signal)
await svc.compactRegion(session, 0, 0, stubAgent(session, 'm'), 1, 1, controller.signal)
expect(svc.lastSignal).toBe(controller.signal)
await svc.compactIfNeeded(session, undefined, undefined, controller.signal)
await svc.compactIfNeeded(stubAgent(session), 1, 1, '', controller.signal)
expect(svc.lastSignal).toBe(controller.signal)
})
})

58
packages/core/agent-loop/src/loop.ts
View File

@@ -388,29 +388,34 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
// (or turn-start listeners on the first step) joins before the request.
drainSteering(ctx, agent, turn)
// Assemble the system prompt for this step. Done HERE (before step/start)
// because the pre-step seam needs it: compaction measures token pressure
// against the system prompt (it counts toward the budget) and a listener
// also receives the model to summarize with. runStep reuses this same
// assembly for the request, so the prompt is assembled once per step.
const assembly = await ctx.systemPrompt.assemble()
const system = [renderPrompt(assembly), agent.options.systemPrompt ?? '']
.filter(text => text.length > 0)
.join('\n\n')
// The step's AbortController exists BEFORE the pre-step seam so a cancel()
// during the seam aborts any in-flight work a listener started (e.g. a
// compaction summarization call). Cleared on every exit path below.
// The step's AbortController exists BEFORE any async pre-step work so a
// dispose() or cancel() — in a synchronous turn-start listener or an
// async listener whose effect fires before we block — always has an armed
// abort to cancel against. isDisposed below covers disposal, which does
// NOT set the cancel marker. Cleared on every exit path below.
const abort = new AbortController()
handle.setAbort(abort)
// Cancel landing before the seam: a synchronous `agent/turn-start` listener
// (or the previous step's continuation listeners) can have called
// `cancel()`. Drop the about-to-start step WITHOUT running the seam — no
// step is open yet, so end the turn `aborted` directly.
if (handle.isCancelled()) {
// Assemble the system prompt for this step. Done HERE (before step/start)
// because the pre-step seam needs it: compaction measures token pressure
// against the system prompt (it counts toward the budget). runStep reuses
// this same assembly for the request, so the prompt is assembled once per
// step.
const assembly = await ctx.systemPrompt.assemble()
const fullSystemPrompt = [renderPrompt(assembly), agent.options.systemPrompt ?? '']
.filter(text => text.length > 0)
.join('\n\n')
// Interruption landing after assembly: dispose() or cancel() in a
// turn-start listener (or a listener whose promise resolved before the
// await above) arms either handle.isDisposed() or handle.isCancelled().
// The Abort was created first, so any concurrent abort also lands on it.
// Drop the about-to-start step WITHOUT running the seam — no step is open
// yet, so end the turn accordingly (disposed wins for an unambiguous
// reason).
if (handle.isCancelled() || handle.isDisposed()) {
handle.setAbort(undefined)
reason = { kind: 'aborted', reason: handle.cancelReason() }
reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
break
}
@@ -425,7 +430,7 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
// throwing listener escapes to the outer catch, which closes the (not-yet-
// open) step as a no-op and ends the turn via failTurn — a broken
// pre-step plugin ends the turn, not the loop.
await ctx.serial('agent/pre-step', agent, turn, step, system, agent.options.model ?? '', abort.signal)
await ctx.serial('agent/pre-step', agent, turn, step, fullSystemPrompt, abort.signal)
session.append('step/start', { turn, step })
stepOpen = true
@@ -433,19 +438,20 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
// Cancel landing in the seam / step-start window: a `cancel()` during the
// pre-step seam (it aborted `abort.signal` above) OR a synchronous
// `agent/step-start` listener that cancels. Check AFTER setAbort/step-start
// and before `runStep`: drop the step, end the turn `aborted`. closeStep
// balances the already-appended step/start.
if (handle.isCancelled()) {
// `agent/step-start` listener that cancels. And disposal, which the earlier
// assembly check may have missed if it only checked isCancelled. Check
// AFTER step/start append + emit and before `runStep`: drop the step, end
// the turn accordingly. closeStep balances the already-appended step/start.
if (handle.isCancelled() || handle.isDisposed()) {
handle.setAbort(undefined)
reason = { kind: 'aborted', reason: handle.cancelReason() }
reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
closeStep()
break
}
let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
try {
stepOutcome = await runStep(ctx, agent, turn, step, assembly, system, abort.signal)
stepOutcome = await runStep(ctx, agent, turn, step, assembly, fullSystemPrompt, abort.signal)
} catch (error: unknown) {
stepOutcome = { error: toError(error) }
} finally {

18
packages/core/agent-loop/tests/loop.spec.ts
View File

@@ -322,9 +322,9 @@ describe('agent loop', () => {
it('agent/pre-step fires once per step before the step is opened', async () => {
// Two steps (a tool call, then a final text turn) → two model calls → two
// pre-step fires, each carrying the assembled system + model, BEFORE the
// step is opened and its request is derived (the request the adapter sees
// reflects any surface state at fire time).
// pre-step fires, each carrying the assembled full system prompt, BEFORE
// the step is opened and its request is derived (the request the adapter
// sees reflects any surface state at fire time).
const adapter = new MockAdapter([
toolCallResponse('c1', 'echo', {}, 'calling echo'),
textResponse('done'),
@@ -336,18 +336,18 @@ describe('agent loop', () => {
}))
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
const fires: { turn: number; step: number; model: string }[] = []
ctx.on('agent/pre-step', (subject, turn, step, _system, model) => {
if (subject === agent) fires.push({ turn, step, model })
const fires: { turn: number; step: number; fullSystemPrompt: string }[] = []
ctx.on('agent/pre-step', (subject, turn, step, fullSystemPrompt) => {
if (subject === agent) fires.push({ turn, step, fullSystemPrompt })
})
send(agent, 'go')
await waitForIdle(ctx, agent)
// One fire per step, in order, each with the agent's model.
// One fire per step, in order, each with the assembled system prompt.
expect(fires).toEqual([
{ turn: 1, step: 1, model: 'mock' },
{ turn: 1, step: 2, model: 'mock' },
{ turn: 1, step: 1, fullSystemPrompt: '' },
{ turn: 1, step: 2, fullSystemPrompt: '' },
])
})

270
packages/core/agent-loop/tests/review-fixes.spec.ts
View File

@@ -1047,3 +1047,273 @@ describe('surface: assistant/message omits sourceEventSeqs when no chunks stream
expect(JSON.stringify(agent.session.deriveMessages())).toContain('injected')
})
})
describe('disposal/cancel honored during pre-step assembly (P1-1)', () => {
it('disposal during system-prompt assembly drops the about-to-start step as disposed', { timeout: 30000 }, async () => {
// Block `system-prompt/assemble` on a promise. Start disposal (which
// calls stop() synchronously, setting status=disposed), then release the
// block. The loop must check isDisposed() after assembly and end the turn
// `disposed` — no LLM call. Don't await fiber.dispose() before releasing
// the blocker: the dispose chain awaits agent.done, which hangs until the
// loop unblocks.
const adapter = new MockAdapter(['hang'])
let releaseAssemble!: () => void
const blocked = new Promise<void>(r => void (releaseAssemble = r))
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(Invariants, { freeze: false })
ctx.llm.registerAdapter(['mock'], adapter)
// Blocking listener on the parent context (survives fiber disposal).
const unlisten = ctx.on('system-prompt/assemble', async function (_assembly, next) {
await blocked
return next()
})
let agent!: ReactLoopAgent
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create(AgentId('a-dispose-assemble'), { model: 'mock' })
}, { inject: ['agentLoop'] }))
const reasons: TurnEndReason[] = []
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
send(agent, 'go')
// Give the loop time to enter the step and reach assemble().
await new Promise(r => setTimeout(r, 50))
// Start disposal — stop() sets status=disposed synchronously, then the
// disposer's await agent.done hangs because the loop is blocked in the
// waterfall. Do NOT await yet; release the blocker first.
const disposalDone = fiber.dispose()
// Now release the blocked waterfall — the loop unblocks, checks
// isDisposed(), and exits, which resolves agent.done and disposalDone.
releaseAssemble()
await disposalDone
await agent.done
unlisten()
const e = [...agent.session.events]
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' })
// No step was opened, no LLM call was made.
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
// agent/turn-end may not fire when disposal happens during assembly: the
// fiber's disposer (stop→status=disposed) runs before closeTurn(true)'s
// emit, and the LIFO chain disposes effects in reverse registration order.
// The turn/end durable record is the one that matters.
})
it('cancel during system-prompt assembly drops the about-to-start step as aborted', { timeout: 30000 }, async () => {
const adapter = new MockAdapter([textResponse('should not appear')])
let releaseAssemble!: () => void
const blocker = new Promise<void>(r => void (releaseAssemble = r))
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(Invariants, { freeze: false })
ctx.llm.registerAdapter(['mock'], adapter)
const unlisten = ctx.on('system-prompt/assemble', async function (_assembly, next) {
await blocker
return next()
})
let agent!: ReactLoopAgent
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create(AgentId('a-cancel-assemble'), { model: 'mock' })
}, { inject: ['agentLoop'] }))
const reasons: TurnEndReason[] = []
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
send(agent, 'go')
await new Promise(r => setTimeout(r, 50))
agent.cancel('user cancelled during assembly')
releaseAssemble()
await waitForIdle(ctx, agent)
await fiber.dispose()
await agent.done
unlisten()
const e = [...agent.session.events]
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({
kind: 'aborted',
reason: 'user cancelled during assembly',
})
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
expect(e.some(x => x.type === 'assistant/message')).toBe(false)
expect(adapter.requests).toHaveLength(0)
expect(reasons).toEqual([{ kind: 'aborted', reason: 'user cancelled during assembly' }])
})
it('disposal during agent/pre-step seam ends the turn disposed', { timeout: 15000 }, async () => {
// Block the `agent/pre-step` serial seam on a promise we control, then
// dispose the agent's fiber. When the block releases, the loop must see
// isDisposed() at the post-seam check and end the turn disposed.
const adapter = new MockAdapter(['hang'])
let releasePreStep!: () => void
const blocker = new Promise<void>(r => void (releasePreStep = r))
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(Invariants, { freeze: false })
ctx.llm.registerAdapter(['mock'], adapter)
ctx.on('agent/pre-step', async () => {
await blocker
})
let agent!: ReactLoopAgent
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create(AgentId('a-dispose-prestep'), { model: 'mock' })
}, { inject: ['agentLoop'] }))
const reasons: TurnEndReason[] = []
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
send(agent, 'go')
await new Promise(r => setTimeout(r, 50))
// Start disposal, then release the block, then await disposal.
const disposalDone = fiber.dispose()
releasePreStep()
await disposalDone
await agent.done
// After the pre-step seam finishes, the post-seam cancel/dispose check
// catches disposal. The step was never opened, no LLM call was made.
const e = [...agent.session.events]
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
// Disposal wins the post-seam check — reason is `disposed`.
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' })
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
// agent/turn-end may not fire when disposal happens during pre-step: the
// fiber's disposer runs before closeTurn(true)'s emit. The durable turn/end
// is the authoritative record.
})
it('cancel during agent/pre-step seam ends the turn aborted', { timeout: 15000 }, async () => {
// Block `agent/pre-step`, then cancel() the agent. When the block releases,
// the post-seam check catches cancellation and ends the turn aborted.
const adapter = new MockAdapter(['hang'])
let releasePreStep!: () => void
const blocker = new Promise<void>(r => void (releasePreStep = r))
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(Invariants, { freeze: false })
ctx.llm.registerAdapter(['mock'], adapter)
ctx.on('agent/pre-step', async () => {
await blocker
})
let agent!: ReactLoopAgent
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create(AgentId('a-cancel-prestep'), { model: 'mock' })
}, { inject: ['agentLoop'] }))
const reasons: TurnEndReason[] = []
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
send(agent, 'go')
await new Promise(r => setTimeout(r, 30))
agent.cancel('user cancelled')
releasePreStep()
await waitForIdle(ctx, agent)
await fiber.dispose()
await agent.done
const e = [...agent.session.events]
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: 'user cancelled' })
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
expect(reasons).toEqual([{ kind: 'aborted', reason: 'user cancelled' }])
})
it('disposal during assembly does not leak an LLM call or append assistant/chunk', { timeout: 15000 }, async () => {
// The key assertion from the original bug report: after disposal, no
// assistant/chunk or assistant/message appears — the turn ends disposed
// before any model interaction.
const adapter = new MockAdapter([textResponse('should not appear')])
let releaseAssemble!: () => void
const blocker = new Promise<void>(r => void (releaseAssemble = r))
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(Invariants, { freeze: false })
ctx.llm.registerAdapter(['mock'], adapter)
ctx.on('system-prompt/assemble', async function (_assembly, next) {
await blocker
return next()
})
let agent!: ReactLoopAgent
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create(AgentId('a-dispose-no-leak'), { model: 'mock' })
}, { inject: ['agentLoop'] }))
send(agent, 'go')
await new Promise(r => setTimeout(r, 50))
const disposalDone = fiber.dispose()
releaseAssemble()
await disposalDone
await agent.done
const e = [...agent.session.events]
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
// The critical assertions: after disposal, the turn has no assistant
// artifacts — the turn ended disposed before the model was invoked.
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
expect(e.some(x => x.type === 'assistant/message')).toBe(false)
expect(adapter.requests).toHaveLength(0)
// The durable turn/end reason is the authoritative record; agent/turn-end
// may not fire when disposal interleaves with closeTurn(true)'s emit.
})
})

9
packages/core/agent/src/types.ts
View File

@@ -200,13 +200,12 @@ declare module 'cordis' {
* transform or veto, but the loop must wait for the mutation to complete
* before opening the step and deriving, and serial isolates listeners from
* each other (one finishes its surface append before the next runs).
* `system`/`model` are the assembled values a listener needs to measure
* pressure (system counts toward the budget) and to summarize (the model).
* `signal` cancels any in-flight work a listener starts (e.g. a summarization
* model call).
* `fullSystemPrompt` is the assembled prompt a listener needs to measure
* pressure (the system prompt counts toward the budget). `signal` cancels any
* in-flight work a listener starts (e.g. a summarization model call).
* @mode serial
*/
'agent/pre-step'(agent: Agent, turn: number, step: number, system: string, model: string, signal: AbortSignal): Promise<void> | void
'agent/pre-step'(agent: Agent, turn: number, step: number, fullSystemPrompt: string, signal: AbortSignal): Promise<void> | void
/**
* Waterfall: mutate the fully-assembled {@link GenerateOptions} before the
* model call (hooks, model switching, tool filtering, …). Call `next()` to