From 959a3c3a8b2198a78ad47f7a0d023ed7ac92d84d Mon Sep 17 00:00:00 2001 From: Yichen Jiang Date: Wed, 8 Jul 2026 21:46:03 +0800 Subject: [PATCH] fix(agent-loop): discard an interrupted prefix composition instead of caching it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A cancel/dispose landing inside the first agent/session-prefix waterfall used to commit the listener chain's return value to the instance cache before the interruption check dropped the turn; an abort-aware listener's degraded fallback would then ship on every later request of the instance. The commit now happens only after the composition survives the interruption check — the cache only ever holds a fully composed prefix, and the next turn recomposes under a live signal. --- docs/cordis-catalog/events.md | 10 +++--- docs/event-producer-consumer.md | 8 ++--- packages/core/agent-loop/src/loop.ts | 27 +++++++++----- packages/core/agent-loop/tests/cancel.spec.ts | 36 ++++++++++++++++++- packages/core/agent/src/types.ts | 6 +++- 5 files changed, 67 insertions(+), 20 deletions(-) diff --git a/docs/cordis-catalog/events.md b/docs/cordis-catalog/events.md index 8b820168b3..cd7bb30e29 100644 --- a/docs/cordis-catalog/events.md +++ b/docs/cordis-catalog/events.md @@ -47,7 +47,7 @@ A step or turn errored. The loop reports a failure here (plus the logger) even w Types: [Agent](../core-data-structures/core.md) -Source: [`packages/core/agent/src/types.ts:472`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:476`](../../packages/core/agent/src/types.ts) ### `agent/pre-step` — serial @@ -101,7 +101,7 @@ Source: [`packages/core/agent/src/types.ts:394`](../../packages/core/agent/src/t ### `agent/session-prefix` — waterfall -Waterfall: compose the SESSION PREFIX — request-only messages placed in front of the ENTIRE derived history (directly after the provider's system slot) on every request this loop instance sends. Fired ONCE per loop instance, lazily before its first step's agent/pre-step seam — BEFORE the pre-step so a token-pressure gate (compaction) counts the prefix this instance will actually send, never a previous instance's logged one. The composed result is deep-frozen, recorded as `EpochHeader.messagePrefix` on the instance's anchoring `'initial'`/`'resume'` header snapshot, and reused verbatim for every subsequent request — never recomputed mid-session, so the provider prefix cache holds by construction (a process restart or `ctx.agents.resume()` is a new instance: it recomposes, and any drift lands attributably on the `'resume'` snapshot). Composition runs outside the step, before the boundary snapshot: a composing listener's session append joins the CURRENT request's derived history. +Waterfall: compose the SESSION PREFIX — request-only messages placed in front of the ENTIRE derived history (directly after the provider's system slot) on every request this loop instance sends. Fired ONCE per loop instance, lazily before its first step's agent/pre-step seam — BEFORE the pre-step so a token-pressure gate (compaction) counts the prefix this instance will actually send, never a previous instance's logged one. The composed result is deep-frozen, recorded as `EpochHeader.messagePrefix` on the instance's anchoring `'initial'`/`'resume'` header snapshot, and reused verbatim for every subsequent request — never recomputed mid-session, so the provider prefix cache holds by construction (a process restart or `ctx.agents.resume()` is a new instance: it recomposes, and any drift lands attributably on the `'resume'` snapshot). Composition runs outside the step, before the boundary snapshot: a composing listener's session append joins the CURRENT request's derived history. A composition interrupted by a cancel/dispose landing inside the waterfall is discarded — never cached, logged, or sent — and the next turn recomposes under a live signal, so an abort-aware listener's degraded fallback cannot leak into later requests. This is the home for session-stable openers the model must always see but that must NOT become durable history — a skills catalog, an AGENTS.md digest, a workspace baseline: `Session.deriveMessages()` never returns the prefix, and the header events are its only durable record, so the request stays reconstructable from the log. Content that CHANGES mid-session belongs in the append-only history channels instead — `agent.inject()`, a `tools/post-execute` decision's `additionalContext`, prompt-submit `additionalContext` — each a durable `context/message` paid once and prefix-cached thereafter. @@ -113,7 +113,7 @@ The seed is a frozen empty list; a contributing listener returns a NEW array — Types: [Agent](../core-data-structures/core.md) · [Message](../core-data-structures/core.md) -Source: [`packages/core/agent/src/types.ts:437`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:441`](../../packages/core/agent/src/types.ts) ### `agent/session-start` — emit @@ -149,7 +149,7 @@ Waterfall: post-process the assembled assistant Message before tool dispatch (va Types: [Agent](../core-data-structures/core.md) · [Message](../core-data-structures/core.md) -Source: [`packages/core/agent/src/types.ts:447`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:451`](../../packages/core/agent/src/types.ts) ### `agent/turn-continuation` — waterfall @@ -161,7 +161,7 @@ Waterfall: override the turn-continuation decision via a typed ContinuationDecis Types: [Agent](../core-data-structures/core.md) -Source: [`packages/core/agent/src/types.ts:460`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:464`](../../packages/core/agent/src/types.ts) ## `fs/*` diff --git a/docs/event-producer-consumer.md b/docs/event-producer-consumer.md index 19a2d4b1c0..a3c9cef433 100644 --- a/docs/event-producer-consumer.md +++ b/docs/event-producer-consumer.md @@ -9,16 +9,16 @@ This matrix shows which packages dispatch each harness-owned event and which pac | --- | --- | --- | --- | --- | | `agent/created` | `emit` | [`packages/core/agent/src/types.ts:265`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`emit`) | [`stdio-agent`](../packages/ui/stdio-agent) | | `agent/disposed` | `emit` | [`packages/core/agent/src/types.ts:272`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`emit`) | [`stdio-agent`](../packages/ui/stdio-agent) | -| `agent/error` | `emit` | [`packages/core/agent/src/types.ts:472`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | - | +| `agent/error` | `emit` | [`packages/core/agent/src/types.ts:476`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | - | | `agent/pre-step` | `serial` | [`packages/core/agent/src/types.ts:357`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`compact-basic`](../packages/compact/compact-basic) | | `agent/prompt-submit` | `waterfall` | [`packages/core/agent/src/types.ts:370`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex) | | `agent/queued` | `emit` | [`packages/core/agent/src/types.ts:290`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | - | | `agent/request` | `waterfall` | [`packages/core/agent/src/types.ts:394`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | - | -| `agent/session-prefix` | `waterfall` | [`packages/core/agent/src/types.ts:437`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | - | +| `agent/session-prefix` | `waterfall` | [`packages/core/agent/src/types.ts:441`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | - | | `agent/session-start` | `emit` | [`packages/core/agent/src/types.ts:305`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex) | | `agent/status` | `emit` | [`packages/core/agent/src/types.ts:281`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`acp`](../packages/ui/acp), [`invariants`](../packages/support/invariants), [`stdio-agent`](../packages/ui/stdio-agent) | -| `agent/step-result` | `waterfall` | [`packages/core/agent/src/types.ts:447`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | - | -| `agent/turn-continuation` | `waterfall` | [`packages/core/agent/src/types.ts:460`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex) | +| `agent/step-result` | `waterfall` | [`packages/core/agent/src/types.ts:451`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | - | +| `agent/turn-continuation` | `waterfall` | [`packages/core/agent/src/types.ts:464`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex) | | `fs/edit-intent` | `waterfall` | [`packages/fs/fs/src/index.ts:123`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`waterfall`) | [`fs-policy`](../packages/fs/fs-policy) | | `fs/observed` | `emit` | [`packages/fs/fs/src/index.ts:138`](../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:109`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`waterfall`) | [`fs-policy`](../packages/fs/fs-policy) | diff --git a/packages/core/agent-loop/src/loop.ts b/packages/core/agent-loop/src/loop.ts index 31ebb61380..33ddee8f64 100644 --- a/packages/core/agent-loop/src/loop.ts +++ b/packages/core/agent-loop/src/loop.ts @@ -497,18 +497,27 @@ async function runTurn( // CURRENT request. if (transmission.sessionPrefix === undefined) { const emptyPrefix: Message[] = deepFreeze([]) - transmission.sessionPrefix = deepFreeze(structuredClone(await ctx.waterfall( + const composed = await ctx.waterfall( 'agent/session-prefix', agent, emptyPrefix, abort.signal, () => Promise.resolve(emptyPrefix), - ))) - } + ) - // Interruption landing during prefix composition: mirror the assembly - // window above — drop the about-to-start step without running the seam. - if (handle.isCancelled() || handle.isDisposed()) { - handle.setAbort(undefined) - reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() } - break + // Interruption landing during prefix composition: mirror the assembly + // window above — drop the about-to-start step without running the + // seam, and DISCARD the composition instead of caching it. An + // abort-aware listener may have returned a degraded fallback under + // the firing signal; committing it would ship a prefix no request + // ever used (and no header ever logged) on this instance's next real + // request. The next turn recomposes under a live signal — the cache + // only ever holds a fully composed prefix. The cache-hit path needs + // no such check: nothing awaits between the assembly check above and + // the pre-step seam. + if (handle.isCancelled() || handle.isDisposed()) { + handle.setAbort(undefined) + reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() } + break + } + transmission.sessionPrefix = deepFreeze(structuredClone(composed)) } // Pre-step surface-mutation checkpoint (compaction), fired OUTSIDE the diff --git a/packages/core/agent-loop/tests/cancel.spec.ts b/packages/core/agent-loop/tests/cancel.spec.ts index 4b3be6e1c1..63a75447ca 100644 --- a/packages/core/agent-loop/tests/cancel.spec.ts +++ b/packages/core/agent-loop/tests/cancel.spec.ts @@ -12,7 +12,7 @@ import { describe, expect, it } from 'vitest' import { Context } from 'cordis' -import LlmService from '@deepseek-ai/dsh-llm' +import LlmService, { type Message } from '@deepseek-ai/dsh-llm' import SessionStore, { SessionId, TurnEndReason } from '@deepseek-ai/dsh-session' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRegistry from '@deepseek-ai/dsh-tools' @@ -229,6 +229,40 @@ describe('Agent.cancel()', () => { expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' }) }) + it('a cancel-interrupted prefix composition is discarded: the next send recomposes and ships the fresh prefix (stale-cache guard)', async () => { + const adapter = new MockAdapter([textResponse('reply')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' }) + + // The first composition is interrupted mid-waterfall and — like an + // abort-aware listener bailing on a firing signal — contributes nothing. + // Caching that degraded result would silently strip the prefix from every + // later request of this instance; the loop must discard it and recompose + // on the next send, and the SECOND composition's value must be what the + // wire and the header log carry. + const opener: Message = { role: 'user', content: [{ type: 'text', text: 'fresh opener' }] } + let compositions = 0 + ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next): Promise => { + compositions += 1 + if (compositions === 1) { + agent.cancel('mid-composition') + return next() + } + return [opener, ...await next()] + }) + + send(agent, 'dropped') + await waitForIdle(ctx, agent) + send(agent, 'real prompt') + await waitForIdle(ctx, agent) + + expect(compositions).toBe(2) + expect(adapter.requests).toHaveLength(1) + expect(adapter.requests[0]?.messages[0]).toEqual(opener) + const headerEvent = agent.session.events.find(e => e.type === 'request/header') + expect(headerEvent?.type === 'request/header' && headerEvent.data.header.messagePrefix).toEqual([opener]) + }) + it('cancel from a synchronous turn/start session-event listener drops the step (step-start window)', async () => { const adapter = new MockAdapter([textResponse('should not stream')]) const ctx = await harness(adapter) diff --git a/packages/core/agent/src/types.ts b/packages/core/agent/src/types.ts index f17c666286..2bde463be5 100644 --- a/packages/core/agent/src/types.ts +++ b/packages/core/agent/src/types.ts @@ -407,7 +407,11 @@ declare module 'cordis' { * or `ctx.agents.resume()` is a new instance: it recomposes, and any * drift lands attributably on the `'resume'` snapshot). Composition runs * outside the step, before the boundary snapshot: a composing listener's - * session append joins the CURRENT request's derived history. + * session append joins the CURRENT request's derived history. A + * composition interrupted by a cancel/dispose landing inside the + * waterfall is discarded — never cached, logged, or sent — and the next + * turn recomposes under a live signal, so an abort-aware listener's + * degraded fallback cannot leak into later requests. * * This is the home for session-stable openers the model must always see * but that must NOT become durable history — a skills catalog, an