refactor(agent): replace the per-step advice seam with agent/session-prefix

Review discussion converged on the industry shape (Claude Code caches
user context per conversation; Codex separates initial context from
diffs; Kimi appends at continuation boundaries to protect prompt
caching): stable openers belong in a compose-once prefix, mid-session
changes belong in append-only history — not in a per-request slot.

agent/session-prefix fires ONCE per loop instance, lazily on its first
request-building step: the composed Message[] is deep-frozen, cached on
the transmission bookkeeping, recorded as EpochHeader.messagePrefix on
the anchoring 'initial'/'resume' snapshot, and reused verbatim for
every request the instance sends — prefix stability is structural, not
a producer discipline, and a resume recomposes with attributable drift.
The request is messagePrefix + boundary snapshot.

The per-step RequestAdvice/RequestAdviceContext surface and the
messageSuffix header field are dropped: the tail slot had no consumer,
and every current update pattern (new AGENTS.md discovered, memory
update, skills change) routes through the existing append-only history
channels — inject(), tools/post-execute additionalContext,
prompt-submit additionalContext — each paid once and prefix-cached
thereafter. The messagePrefix delta arm stays for codec totality; the
loop never produces one in practice.
This commit is contained in:
Yichen Jiang
2026-07-08 15:44:30 +08:00
parent 2cbdeb0872
commit ea4c10d753
21 changed files with 260 additions and 380 deletions

View File

@@ -59,10 +59,10 @@ forever:
boundary = session.deriveMessages() ⟵ reconstruction boundary: same sync frame,
session('step/start') strictly before step/start
config = waterfall agent/request ⟵ frozen seed; return a replacement to switch
reqMsgs = waterfall agent/request-advice ⟵ request-only before/after messages; recorded
on the header, never session history
prefix ??= waterfall agent/session-prefix ⟵ once per instance (first request): frozen
session prefix; on the header, never history
session('request/header'[-delta]) ⟵ the header event this request owes the log
stream llm.stream(freeze({header..., messages: before+boundary+after})) → session('assistant/chunk')
stream llm.stream(freeze({header..., messages: prefix+boundary})) → session('assistant/chunk')
message = waterfall agent/step-result
session('assistant/message')
each tool-call: session('tool/call')
@@ -86,7 +86,7 @@ Cancellation: `agent.cancel()` is the single public stop primitive — it clears
### What is NOT here
Everything that goes beyond "call the model, run the tools, repeat" belongs to plugins listening on the event taxonomy:
- Hooks: `agent/session-start`, `agent/prompt-submit`, `agent/pre-step`, `agent/request`, `agent/request-advice`, `agent/step-result`, `tools/pre-execute`, `tools/post-execute`, `agent/turn-continuation`
- Hooks: `agent/session-start`, `agent/prompt-submit`, `agent/pre-step`, `agent/request`, `agent/session-prefix`, `agent/step-result`, `tools/pre-execute`, `tools/post-execute`, `agent/turn-continuation`
- Compaction: `agent/pre-step`
- Sandbox, permission, plan mode: `tools/pre-execute` (deny/ask gate), `tools/post-execute`
- Sub-agents: implemented outside the loop as `ctx.subagents` providers; in-process providers use `ctx.agents.create()` and owned `AgentHandle` teardown, while child streaming/progress and background/poll collection remain deferred.

View File

@@ -10,7 +10,7 @@
import type { Context } from 'cordis'
import type { FinishReason, GenerateOptions, LlmCallConfig, Message } from '@deepseek-ai/dsh-llm'
import { BlockAssembler, HarnessError, deepFreeze } from '@deepseek-ai/dsh-llm'
import type { ContinuationDecision, HookContext, PromptDecision, RequestAdvice } from '@deepseek-ai/dsh-agent'
import type { ContinuationDecision, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
import { canonicalHeader } from '@deepseek-ai/dsh-session'
import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
import { createTransmissionLog, recordRequestHeader } from './request-log.ts'
@@ -161,11 +161,12 @@ export interface LoopHandle {
* boundary = session.deriveMessages() ⟵ the reconstruction boundary: snapshot in the
* session('step/start') same sync frame, strictly before step/start
* config = waterfall agent/request(config) ⟵ frozen seed; a returned replacement switches
* advice = waterfall agent/request-advice ⟵ request-only before/after advice; logged on
* the header, never session history
* prefix ??= waterfall agent/session-prefix ⟵ once per loop instance (first request):
* frozen session prefix; logged on the header,
* never session history
* session('request/header'|'request/header-delta') ⟵ the header event this request owes the
* log (initial/resume anchor, delta, fallback)
* req = freeze({header..., messages: before+boundary+after, sessionId, signal})
* req = freeze({header..., messages: prefix+boundary, sessionId, signal})
* stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks, frozen req)
* session('assistant/chunk')
* msg = waterfall agent/step-result ⟵ BEFORE the log append, so the
@@ -676,8 +677,9 @@ function drainSteering(agent: ReactLoopAgent, turn: number): boolean {
}
/** One step: build the request from the boundary snapshot + the step's
* header → collect request-only messages → log the header event the request
* owes → stream model → record → execute tools. The caller assembles the
* header → compose the session prefix if this instance has none yet → log
* the header event the request owes → stream model → record → execute
* tools. The caller assembles the
* system prompt, fires the `agent/pre-step` seam, snapshots the derivation,
* and opens the step BEFORE calling this, so `boundaryMessages` is exactly
* the surface prefix at step/start and already reflects any compaction. */
@@ -720,47 +722,47 @@ async function runStep(
throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
}
// Collect the request-ONLY advice: `before` messages go in front of the
// entire boundary snapshot, `after` messages follow its last message. Advice
// is not session history — the header event below is its only durable
// record (EpochHeader.messagePrefix/messageSuffix), which keeps the request
// a pure function of the log. The frozen empty seed serves both the
// listener chain and the no-listener fallback: a contribution is a RETURNED
// extension of `await next()`, never an in-place push. The context gets a
// frozen COPY of the boundary (the request is built from the internal
// snapshot), so a listener cannot smuggle unlogged content into the request
// by mutating what it was shown. Fired AFTER the boundary snapshot, so a
// listener's session append lands past the boundary and joins the NEXT
// request — the same window rule as the `agent/request` waterfall.
const emptyRequestAdvice: RequestAdvice = deepFreeze({ before: [], after: [] })
const requestAdviceBoundary = deepFreeze([...boundaryMessages])
const requestAdvice = await ctx.waterfall(
'agent/request-advice', agent, turn, step, emptyRequestAdvice,
{ system, assembly, boundaryMessages: requestAdviceBoundary, signal },
() => Promise.resolve(emptyRequestAdvice),
)
// Compose the session prefix ONCE per loop instance, lazily on its first
// request-building step: request-only messages placed in front of the
// ENTIRE derived history on every request this instance sends. The result
// is deep-cloned (decoupled from listener-held references), deep-frozen,
// and cached on the transmission bookkeeping, so reuse is structural — the
// prefix cannot change mid-session and the provider prefix cache holds by
// construction (resume = a new instance = a recompose, anchored by its
// 'resume' snapshot). The prefix is not session history — the header event
// below is its only durable record (EpochHeader.messagePrefix), which
// keeps the request a pure function of the log. The frozen empty seed
// serves both the listener chain and the no-listener fallback: a
// contribution is a RETURNED extension of `await next()`, never an
// in-place push.
if (transmission.sessionPrefix === undefined) {
const emptyPrefix: Message[] = deepFreeze([])
transmission.sessionPrefix = deepFreeze(structuredClone(await ctx.waterfall(
'agent/session-prefix', agent, emptyPrefix, signal,
() => Promise.resolve(emptyPrefix),
)))
}
const sessionPrefix = transmission.sessionPrefix
// The request header (the log's request/header* vocabulary): canonical form,
// recorded before dispatch so the log always explains the request —
// including the request-only advice, which no other event carries.
// including the session prefix, which no other event carries.
const header = canonicalHeader({
config,
...system ? { system } : {},
...assembly.tools.length > 0 ? { tools: assembly.tools } : {},
...requestAdvice.before.length > 0 ? { messagePrefix: requestAdvice.before } : {},
...requestAdvice.after.length > 0 ? { messageSuffix: requestAdvice.after } : {},
...sessionPrefix.length > 0 ? { messagePrefix: sessionPrefix } : {},
})
recordRequestHeader(session, transmission, header)
// Build and freeze: the request is a pure function of (boundary snapshot,
// logged header) — llm/stream listeners and adapters read it, mutation
// throws. sessionId + frozen is the loop-built marker the dev invariant
// keys on. Message order: header.messagePrefix, then the boundary snapshot,
// then header.messageSuffix — the reconstruction equation the invariant
// recomputes.
// keys on. Message order: header.messagePrefix, then the boundary
// snapshot — the reconstruction equation the invariant recomputes.
const request: GenerateOptions = deepFreeze({
model: header.config.model,
messages: [...header.messagePrefix ?? [], ...boundaryMessages, ...header.messageSuffix ?? []],
messages: [...header.messagePrefix ?? [], ...boundaryMessages],
...header.system !== undefined ? { system: header.system } : {},
...header.tools !== undefined ? { tools: header.tools } : {},
...header.config.temperature !== undefined ? { temperature: header.config.temperature } : {},

View File

@@ -12,11 +12,20 @@
import { diffHeader, headerEquals, applyHeaderDelta } from '@deepseek-ai/dsh-session'
import type { EpochHeader, Session } from '@deepseek-ai/dsh-session'
import type { Message } from '@deepseek-ai/dsh-llm'
/** Per-loop-instance bookkeeping: whether THIS instance has logged a header yet. */
export interface TransmissionLog {
/** True once this loop instance appended its anchoring `request/header` snapshot. */
loggedHeader: boolean
/**
* The instance's composed session prefix (the `agent/session-prefix`
* waterfall's deep-frozen product), cached on the instance's first
* request-building step and reused verbatim for every request it sends —
* the structural guarantee that the prefix never changes mid-session.
* `undefined` until composed.
*/
sessionPrefix?: Message[]
}
/**
@@ -61,7 +70,7 @@ export function recordRequestHeader(session: Session, state: TransmissionLog, he
const baseline = session.requestHeader()!
if (headerEquals(baseline, header)) return
const delta = diffHeader(baseline, header)
/* v8 ignore next -- headerEquals false ⟹ diffHeader defined: both compare the same three parts */
/* v8 ignore next -- headerEquals false ⟹ diffHeader defined: both compare the same four parts */
if (delta === undefined) return
if (headerEquals(applyHeaderDelta(baseline, delta), header)) {
session.append('request/header-delta', delta)

View File

@@ -1,14 +1,13 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService, { CallId, type Message } from '@deepseek-ai/dsh-llm'
import SessionStore, { foldRequestHeader, type SessionEvent, type TurnEndReason } from '@deepseek-ai/dsh-session'
import SessionStore, { type SessionEvent, type TurnEndReason } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineTool, type PostToolDecision, type PreToolDecision } from '@deepseek-ai/dsh-tools'
import AgentRegistry, {
AgentId,
type ContinuationDecision,
type PromptDecision,
type RequestAdvice,
type SessionStartSource,
} from '@deepseek-ai/dsh-agent'
import AgentLoop, { type ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
@@ -311,58 +310,58 @@ describe('agent/session-start', () => {
})
})
describe('agent/request-advice (RequestAdvice)', () => {
it('frames the derived history: before precedes it, after follows it, and the header records both', async () => {
const adapter = new MockAdapter([textResponse('ok')])
describe('agent/session-prefix', () => {
it('composes once per loop instance and fronts every request; the header records it; history stays untouched', async () => {
const adapter = new MockAdapter([
toolCallResponse('c1', 'echo', { text: 'ping' }),
textResponse('done'),
textResponse('again'),
])
const ctx = await harness(adapter)
ctx.tools.register(defineTool({
name: 'echo', description: 'echo', parameters: { text: { type: 'string' } },
async execute(args) { return [{ type: 'text', text: String(args.text) }] },
}))
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
const reminder: Message = { role: 'user', content: [{ type: 'text', text: '<system-reminder>catalog</system-reminder>' }] }
const trailer: Message = { role: 'user', content: [{ type: 'text', text: 'trailing note' }] }
ctx.on('agent/request-advice', async (_agent, _turn, _step, _messages, _context, next): Promise<RequestAdvice> => {
const result = await next()
return { before: [...result.before, reminder], after: [...result.after, trailer] }
let composed = 0
ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next): Promise<Message[]> => {
composed += 1
return [...await next(), reminder]
})
send(agent, 'hi')
send(agent, 'go')
await waitForIdle(ctx, agent)
send(agent, 'next turn')
await waitForIdle(ctx, agent)
// The request carries before + derived history + after, in that order…
const request = adapter.requests[0]!
expect(request.messages).toEqual([
reminder,
{ role: 'user', content: [{ type: 'text', text: 'hi' }] },
trailer,
])
// …the header event is their durable record…
const headerEvent = events(agent).find(e => e.type === 'request/header')
expect(headerEvent?.type === 'request/header' && headerEvent.data.header.messagePrefix).toEqual([reminder])
expect(headerEvent?.type === 'request/header' && headerEvent.data.header.messageSuffix).toEqual([trailer])
// …and they never become session history.
expect(agent.session.deriveMessages()).toEqual([
{ role: 'user', content: [{ type: 'text', text: 'hi' }] },
{ role: 'assistant', content: [{ type: 'text', text: 'ok' }] },
])
// Three requests (two turns), ONE composition: the frozen product is
// reused verbatim, so the prefix cannot drift mid-session.
expect(adapter.requests).toHaveLength(3)
expect(composed).toBe(1)
for (const request of adapter.requests) {
expect(request.messages[0]).toEqual(reminder)
}
// The anchoring snapshot is the prefix's durable record — and the ONLY
// header event: reuse means no request/header-delta ever.
const headerEvents = events(agent).filter(e => e.type === 'request/header' || e.type === 'request/header-delta')
expect(headerEvents).toHaveLength(1)
expect(headerEvents[0]?.type === 'request/header' && headerEvents[0].data.header.messagePrefix).toEqual([reminder])
// Never session history: the derivation starts at the real user prompt.
expect(agent.session.deriveMessages()[0]).toEqual({ role: 'user', content: [{ type: 'text', text: 'go' }] })
})
it('contributions compose across listeners and see the read-only request facts', async () => {
it('contributions compose across listeners in registration order', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
const seen: { system: string; boundaryRoles: string[]; sectionCount: number }[] = []
ctx.on('agent/request-advice', async (_agent, _turn, _step, _messages, context, next): Promise<RequestAdvice> => {
const result = await next()
seen.push({
system: context.system,
boundaryRoles: context.boundaryMessages.map(m => m.role),
sectionCount: context.assembly.sections.length,
})
return { before: [{ role: 'user', content: [{ type: 'text', text: 'first' }] }, ...result.before], after: result.after }
ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next): Promise<Message[]> => {
return [{ role: 'user', content: [{ type: 'text', text: 'first' }] }, ...await next()]
})
ctx.on('agent/request-advice', async (_agent, _turn, _step, _messages, _context, next): Promise<RequestAdvice> => {
const result = await next()
return { before: [...result.before, { role: 'user', content: [{ type: 'text', text: 'second' }] }], after: result.after }
ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next): Promise<Message[]> => {
return [...await next(), { role: 'user', content: [{ type: 'text', text: 'second' }] }]
})
send(agent, 'hi')
@@ -372,27 +371,21 @@ describe('agent/request-advice (RequestAdvice)', () => {
// out (waterfall), so its prepend lands first.
const texts = adapter.requests[0]!.messages.map(m => m.content[0]?.type === 'text' ? m.content[0].text : '')
expect(texts).toEqual(['first', 'second', 'hi'])
// The context carried the request facts: the rendered system prompt, the
// boundary snapshot (exactly the drained user prompt), and the assembly.
expect(seen).toHaveLength(1)
expect(seen[0]!.boundaryRoles).toEqual(['user'])
expect(typeof seen[0]!.system).toBe('string')
})
it('with no contributions the header omits both fields and the request is the bare derivation', async () => {
it('with no contributions the header omits messagePrefix and the request is the bare derivation', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
// A listener that delegates without contributing — the canonical no-op.
ctx.on('agent/request-advice', async (_agent, _turn, _step, _messages, _context, next) => next())
ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next) => next())
send(agent, 'hi')
await waitForIdle(ctx, agent)
const headerEvent = events(agent).find(e => e.type === 'request/header')
expect(headerEvent?.type === 'request/header' && 'messagePrefix' in headerEvent.data.header).toBe(false)
expect(headerEvent?.type === 'request/header' && 'messageSuffix' in headerEvent.data.header).toBe(false)
expect(adapter.requests[0]!.messages).toEqual([{ role: 'user', content: [{ type: 'text', text: 'hi' }] }])
})
@@ -402,9 +395,9 @@ describe('agent/request-advice (RequestAdvice)', () => {
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
let mutationError: unknown
ctx.on('agent/request-advice', async (_agent, _turn, _step, messages, _context, next): Promise<RequestAdvice> => {
ctx.on('agent/session-prefix', async (_agent, prefix, _signal, next): Promise<Message[]> => {
try {
messages.before.push({ role: 'user', content: [{ type: 'text', text: 'smuggled' }] })
prefix.push({ role: 'user', content: [{ type: 'text', text: 'smuggled' }] })
} catch (error: unknown) {
mutationError = error
}
@@ -418,30 +411,7 @@ describe('agent/request-advice (RequestAdvice)', () => {
expect(adapter.requests[0]!.messages).toEqual([{ role: 'user', content: [{ type: 'text', text: 'hi' }] }])
})
it('the read-only boundary context rejects in-place mutation before the request is built', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
let mutationError: unknown
ctx.on('agent/request-advice', async (_agent, _turn, _step, _messages, context, next): Promise<RequestAdvice> => {
try {
const mutableBoundary = context.boundaryMessages as Message[]
mutableBoundary.push({ role: 'user', content: [{ type: 'text', text: 'smuggled' }] })
} catch (error: unknown) {
mutationError = error
}
return next()
})
send(agent, 'hi')
await waitForIdle(ctx, agent)
expect(mutationError).toBeInstanceOf(TypeError)
expect(adapter.requests[0]!.messages).toEqual([{ role: 'user', content: [{ type: 'text', text: 'hi' }] }])
})
it('a per-step contribution change is logged as a header delta, so every request stays reconstructable', async () => {
it('mutating a listener-held reference after composition cannot alter later requests (the cache is a frozen clone)', async () => {
const adapter = new MockAdapter([
toolCallResponse('c1', 'echo', { text: 'ping' }),
textResponse('done'),
@@ -453,26 +423,21 @@ describe('agent/request-advice (RequestAdvice)', () => {
}))
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
let step = 0
ctx.on('agent/request-advice', async (_agent, _turn, _step, _messages, _context, next): Promise<RequestAdvice> => {
const result = await next()
step += 1
return { before: [...result.before, { role: 'user', content: [{ type: 'text', text: `reminder v${step}` }] }], after: result.after }
})
const held: Message = { role: 'user', content: [{ type: 'text', text: 'v1' }] }
ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next): Promise<Message[]> => [...await next(), held])
send(agent, 'go')
await waitForIdle(ctx, agent)
expect(adapter.requests[0]!.messages[0]).toEqual({ role: 'user', content: [{ type: 'text', text: 'reminder v1' }] })
expect(adapter.requests[1]!.messages[0]).toEqual({ role: 'user', content: [{ type: 'text', text: 'reminder v2' }] })
// Step 2's changed prefix rides a request/header-delta whose fold matches
// what the second request actually sent.
const delta = events(agent).find(e => e.type === 'request/header-delta')
expect(delta?.type === 'request/header-delta' && delta.data.messagePrefix).toEqual([{ role: 'user', content: [{ type: 'text', text: 'reminder v2' }] }])
expect(foldRequestHeader(agent.session.events)?.messagePrefix).toEqual([{ role: 'user', content: [{ type: 'text', text: 'reminder v2' }] }])
// The listener mutates the object it contributed AFTER composition; the
// cached prefix is a deep-frozen clone, so step 2's request is unchanged.
held.content = [{ type: 'text', text: 'v2' }]
expect(adapter.requests[1]!.messages[0]).toEqual({ role: 'user', content: [{ type: 'text', text: 'v1' }] })
expect(events(agent).filter(e => e.type === 'request/header-delta')).toHaveLength(0)
})
})
describe('agent/turn-continuation (ContinuationDecision)', () => {
it('a continue decision with a reason records next-step steering in the same turn', async () => {
const adapter = new MockAdapter([textResponse('step 1 no tools'), textResponse('step 2')])

View File

@@ -45,7 +45,7 @@ Turn and step boundaries are NOT mirrored as `agent/*` emits: a consumer that ne
- `agent/prompt-submit` — decide what happens to one drained queued message before it becomes a `user/message`: `PromptDecision` = `allow` (optionally rewriting the prompt `content` or attaching `additionalContext`) or `block` (drop it; a batch whose every prompt is blocked opens a zero-step turn that ends `rejected`). Maps onto Claude Code's `UserPromptSubmit`.
- `agent/pre-step` (serial) — mutate the session surface before the step opens and history is derived (compaction). Fires after `turn/start` and before `step/start`, so a listener's appended events land outside the step.
- `agent/request` — shape the call config before the model call: a frozen `LlmCallConfig` seed in, a replacement out (model switching, sampling overrides). Content is not shapeable here — every request is a pure function of the session log ([reconstructability RFC](../../../docs/rfc/implemented/architecture/2026-07-05-reconstructable-requests.md)); the loop logs whatever config the request actually uses as a `request/header*` event
- `agent/request-advice` — contribute request-ONLY messages around the derived history: a frozen empty `RequestAdvice` seed in, an extension of `await next()` out (`before` messages precede the boundary snapshot in the request, `after` messages follow it). For per-request advisory context the model must see now but that must not become durable history; the loop records the contributions on the request's `request/header*` event (`EpochHeader.messagePrefix`/`messageSuffix`), so `deriveMessages()` stays untouched and the request stays reconstructable. Cost model: contributions ride the request's uncached tail and are re-paid at full price on every request they appear in — put session-frozen content in `before` (cacheable prefix; a mid-session change busts the cache for everything after it), route low-frequency change notices through `agent.inject()` instead (paid once, prefix-cached thereafter), and reserve `after` for small, frequently refreshed state snapshots
- `agent/session-prefix` — compose the session prefix: request-only messages placed in front of the ENTIRE derived history on every request. Fired ONCE per loop instance, lazily on its first request; the composed result is deep-frozen, recorded as `EpochHeader.messagePrefix` on the anchoring `request/header` snapshot, and reused verbatim afterwards — the prefix cannot change mid-session, so the provider prefix cache holds by construction (resume = a new instance = a recompose, attributably anchored by its `'resume'` snapshot). The home for session-stable openers that must not become durable history (a skills catalog, an AGENTS.md digest); `deriveMessages()` never returns it. Content that CHANGES mid-session belongs in the append-only history channels instead — `agent.inject()`, `tools/post-execute` `additionalContext`, prompt-submit `additionalContext` — each a durable `context/message` paid once and prefix-cached thereafter
- `agent/step-result` — post-process the assembled assistant message before tool dispatch (validates what the log records)
- `agent/turn-continuation` — override the continue/stop decision via `ContinuationDecision` = `{action:'stop'}` or `{action:'continue', reason?}` (a `continue` `reason` is recorded as next-step steering in the same turn — the typed `/goal` pattern). Force-continue `/loop`, force-stop budget guard.

View File

@@ -17,7 +17,7 @@
* consumer that wants the live transcript subscribes here.
* - **`agent/*`** (this module) — the LIVE runtime surface. Always carries the
* live `Agent`. Two shapes: INTERCEPTION seams (the `agent/prompt-submit`/
* `agent/request`/`agent/request-advice`/`agent/step-result`/
* `agent/request`/`agent/session-prefix`/`agent/step-result`/
* `agent/turn-continuation` waterfalls and
* the serial `agent/pre-step`) that mutate/veto, and TRANSIENT emits
* (`agent/status`, `agent/error`, `agent/created`/
@@ -46,7 +46,7 @@
import type { Branded } from '@deepseek-ai/dsh-brand'
import type { ContentBlock, LlmCallConfig, Message, MessageSource } from '@deepseek-ai/dsh-llm'
import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
import type {} from '@deepseek-ai/dsh-system-prompt'
/** Identifies one live agent in the registry. */
export type AgentId = Branded<'AgentId'>
@@ -155,54 +155,6 @@ export type ContinuationDecision =
| { action: 'stop' }
| { action: 'continue'; reason?: HookContext }
/**
* The request-only ADVICE an `agent/request-advice` waterfall listener weaves
* around the derived history of ONE LLM request — advice in both senses:
* advisory content for the model, attached before/after the join point like
* AOP advice, never modifying the history itself. In
* `GenerateOptions.messages` the `before` messages sit in front of the ENTIRE
* derived history (directly after the provider's system slot) and the `after`
* messages follow its last message (the newest user prompt on a turn's first
* step, the previous step's tool results afterwards). Advice is NOT session
* state — nothing here enters the session log as durable history,
* `Session.deriveMessages()` never returns it, and the next step recomputes
* it from scratch. The loop records the non-empty arrays on the request's
* `request/header*` event (`EpochHeader.messagePrefix` / `messageSuffix`), so
* the request stays reconstructable from the log (the reconstructability
* RFC). For content that must become durable conversation history, use the
* log channels instead: `agent.inject()`, steering, or prompt-submit
* `additionalContext`.
*/
export interface RequestAdvice {
/** Before-advice: messages placed ahead of the entire derived history. */
before: Message[]
/** After-advice: messages placed after the derived history's last message. */
after: Message[]
}
/**
* Read-only facts about the request an `agent/request-advice` listener is
* contributing to. Everything here is already fixed when the seam fires: the
* step is open, the boundary snapshot is taken, and the system prompt is
* assembled — a listener uses these to DECIDE what to contribute (e.g. render
* a workspace-dependent reminder, or skip one already present in history),
* never to mutate them.
*/
export interface RequestAdviceContext {
/** The rendered system prompt this request will carry. */
system: string
/** The prompt assembly the system prompt was rendered from (sections + tools). */
assembly: PromptAssembly
/**
* The boundary snapshot: the derived history this request will carry between
* `before` and `after`. A frozen snapshot — treat it as read-only; content
* for the NEXT request flows through the log channels.
*/
boundaryMessages: readonly Message[]
/** Aborts in-flight listener work when the step is torn down. */
signal: AbortSignal
}
/**
* Why an agent's session lifecycle began, carried by `agent/session-start`. A
* bridge keys its SessionStart hook's matcher on this (Claude Code's
@@ -417,7 +369,7 @@ declare module 'cordis' {
* session log (the reconstructability RFC), so model-visible content
* flows through the log channels — `inject()`, steering, prompt-submit
* `additionalContext`, prompt sections via `system-prompt/assemble`, or
* header-logged request-only messages via {@link agent/request-advice}
* the header-logged session prefix via {@link agent/session-prefix}
* — never through request mutation, and the loop records whatever config
* the request actually uses as a `request/header*` event before dispatch.
* The step's messages are already snapshotted when this fires (the
@@ -434,47 +386,38 @@ declare module 'cordis' {
*/
'agent/request'(agent: Agent, turn: number, step: number, config: LlmCallConfig, next: () => Promise<LlmCallConfig>): Promise<LlmCallConfig>
/**
* Waterfall: weave request-ONLY advice around the derived history — a
* {@link RequestAdvice} whose `before` messages sit in front of the
* ENTIRE boundary snapshot in `GenerateOptions.messages` and whose
* `after` messages follow its last message. Fires once per step, inside
* the open step, after the
* {@link agent/request} config waterfall and before the loop logs the
* request header. This is the seam for per-request advisory context the
* model must see NOW but that must NOT become durable history (a skills
* catalog, an environment reminder): contributions are recorded on the
* request's `request/header*` event (`EpochHeader.messagePrefix` /
* `messageSuffix`) — never as session messages — so
* `Session.deriveMessages()` stays untouched and the request remains
* reconstructable from the log.
* 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 on its first request-building step; 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).
*
* The seed is frozen and empty; a contributing listener returns a NEW
* {@link RequestAdvice} extending `await next()` (spread its arrays —
* never mutate them), so contributions compose across plugins in
* registration order. The boundary snapshot is already taken when this
* fires: a `session.append`/`inject()` from a listener here lands in the
* log but joins the NEXT request — contribute through the returned value,
* not the session. Call `next()` to delegate, or return a
* {@link RequestAdvice} without it to short-circuit.
* 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.
*
* Pick the channel by change frequency (the cost model): a contribution
* rides the request's uncached tail, re-tokenized at full price on EVERY
* request it appears in — cheap only while small. Session-FROZEN content
* belongs in `before`, where it extends the cacheable prefix at zero
* marginal cost (but changing it mid-session invalidates the provider
* cache for the entire history after it). A LOW-FREQUENCY change notice
* belongs in durable history via `agent.inject()` — appended once,
* prefix-cached thereafter. Reserve `after` for small, frequently
* refreshed state snapshots, where a durable chain of stale copies would
* bloat the log and mislead the model.
* @param agent - the agent making the model call.
* @param turn - the open turn number.
* @param step - the step whose request this is.
* @param advice - the frozen empty seed; return an extended replacement to contribute.
* @param context - read-only request facts ({@link RequestAdviceContext}).
* The seed is a frozen empty list; a contributing listener returns a NEW
* array extending `await next()` (`[...prefix, mine]` — never an in-place
* push), so contributions compose across plugins in registration order
* and compose deterministically for a fixed plugin set. Call `next()` to
* delegate, or return a list without it to short-circuit.
* @param agent - the agent whose session prefix is being composed.
* @param prefix - the frozen empty seed; return an extended replacement to contribute.
* @param signal - aborts in-flight listener work (e.g. a discovery scan) when the step is torn down.
* @mode waterfall
*/
'agent/request-advice'(agent: Agent, turn: number, step: number, advice: RequestAdvice, context: RequestAdviceContext, next: () => Promise<RequestAdvice>): Promise<RequestAdvice>
'agent/session-prefix'(agent: Agent, prefix: Message[], signal: AbortSignal, next: () => Promise<Message[]>): Promise<Message[]>
/**
* Waterfall: post-process the assembled assistant {@link Message} before
* tool dispatch (validation, content rewriting, …).

View File

@@ -51,7 +51,7 @@ Plain class (not a Cordis Service). Create via `ctx.sessions.create()`.
### Request-header reconstruction (`request-header.ts`)
The `request/header` (full `EpochHeader` snapshot with a `RequestHeaderReason`) and `request/header-delta` (system line-trim / name-keyed tools delta / whole config / whole request-only message arrays) events make the request envelope logged session state, so every conversation request is a pure function of the log. The pure trio reconstructs it: `foldRequestHeader(events)` folds a log (or any prefix) into the header in force; `diffHeader(prev, next)` encodes a change (undefined when equal); `applyHeaderDelta(prev, delta)` replays one. Writer contract: every logged delta is round-trip-verified (`apply(prev, delta)` deep-equals the new header) with a `'fallback'` snapshot when the encoding cannot express the change (a pure tool reordering), so folding never needs error recovery on a well-formed log. `canonicalHeader` pins the one representation of absence (empty system/tools/messagePrefix/messageSuffix ≡ absent fields; a delta's EMPTY message array encodes the transition back to absence). `EpochHeader.messagePrefix`/`messageSuffix` are the durable record of the `agent/request-advice` waterfall's request-only contributions — the request is `messagePrefix + derived history + messageSuffix`, and `deriveMessages()` never returns them.
The `request/header` (full `EpochHeader` snapshot with a `RequestHeaderReason`) and `request/header-delta` (system line-trim / name-keyed tools delta / whole config / whole session prefix) events make the request envelope logged session state, so every conversation request is a pure function of the log. The pure trio reconstructs it: `foldRequestHeader(events)` folds a log (or any prefix) into the header in force; `diffHeader(prev, next)` encodes a change (undefined when equal); `applyHeaderDelta(prev, delta)` replays one. Writer contract: every logged delta is round-trip-verified (`apply(prev, delta)` deep-equals the new header) with a `'fallback'` snapshot when the encoding cannot express the change (a pure tool reordering), so folding never needs error recovery on a well-formed log. `canonicalHeader` pins the one representation of absence (empty system/tools/messagePrefix ≡ absent fields; a delta's EMPTY prefix array encodes the transition back to absence). `EpochHeader.messagePrefix` is the durable record of the `agent/session-prefix` waterfall's product — composed once per loop instance, the request is `messagePrefix + derived history`, and `deriveMessages()` never returns it.
### Session event vocabulary (`types.ts`)

View File

@@ -22,16 +22,14 @@ type HeaderDelta = {
tools?: ToolsDelta
config?: LlmCallConfig
messagePrefix?: Message[]
messageSuffix?: Message[]
}
/**
* Normalize a header to canonical form: an empty system prompt, an empty
* tool list, and empty request-only message arrays become ABSENT fields,
* matching how requests are built (the request-build spreads skip empty
* values). Diff, fold, and comparison all operate on canonical headers, so
* "no system prompt" (and "no request-only messages") has exactly one
* representation.
* tool list, and an empty session prefix become ABSENT fields, matching how
* requests are built (the request-build spreads skip empty values). Diff,
* fold, and comparison all operate on canonical headers, so "no system
* prompt" (and "no session prefix") has exactly one representation.
* @param header - the header to normalize (not mutated).
* @returns the canonical header.
*/
@@ -41,7 +39,6 @@ export function canonicalHeader(header: EpochHeader): EpochHeader {
...header.system !== undefined && header.system.length > 0 ? { system: header.system } : {},
...header.tools !== undefined && header.tools.length > 0 ? { tools: header.tools } : {},
...header.messagePrefix !== undefined && header.messagePrefix.length > 0 ? { messagePrefix: header.messagePrefix } : {},
...header.messageSuffix !== undefined && header.messageSuffix.length > 0 ? { messageSuffix: header.messageSuffix } : {},
}
}
@@ -121,22 +118,22 @@ function applyTools(prev: readonly ToolSchema[], delta: ToolsDelta): ToolSchema[
* writer's round-trip guard runs (`applyHeaderDelta(prev, delta)` must equal
* the intended header) and the loop runs to skip logging an unchanged header.
* Tools compare per-schema IN ORDER (canonical JSON), so a pure reordering is
* correctly unequal; request-only message arrays compare as canonical JSON
* (both sides come from the same build path, so key order matches when the
* values do).
* correctly unequal; the session prefix compares as canonical JSON (both
* sides come from the same build path, so key order matches when the values
* do).
* @param a - one canonical header.
* @param b - the other.
* @returns whether config, system, tools (in order), and request-only messages all match.
* @returns whether config, system, tools (in order), and the session prefix all match.
*/
export function headerEquals(a: EpochHeader, b: EpochHeader): boolean {
if (!callConfigEquals(a.config, b.config) || a.system !== b.system) return false
if (!sameMessages(a.messagePrefix, b.messagePrefix) || !sameMessages(a.messageSuffix, b.messageSuffix)) return false
if (!sameMessages(a.messagePrefix, b.messagePrefix)) return false
const at = a.tools ?? []
const bt = b.tools ?? []
return at.length === bt.length && at.every((tool, i) => sameSchema(tool, bt[i] as ToolSchema))
}
/** Canonical JSON equality over request-only message arrays; absence equals the empty array. */
/** Canonical JSON equality over session-prefix arrays; absence equals the empty array. */
function sameMessages(a: readonly Message[] | undefined, b: readonly Message[] | undefined): boolean {
return JSON.stringify(a ?? []) === JSON.stringify(b ?? [])
}
@@ -147,7 +144,7 @@ function sameMessages(a: readonly Message[] | undefined, b: readonly Message[] |
* ({@link applyHeaderDelta} on `prev` deep-equals `next`) before logging it —
* the encoding cannot express every change (a pure tool reordering) — and
* fall back to a full `request/header` snapshot when the check fails.
* Request-only messages are replaced whole (small advisory content, not worth
* The session prefix is replaced whole (small advisory content, not worth
* diffing); an empty replacement array encodes the transition to "none".
* @param prev - the folded header the log currently implies.
* @param next - the header the next request will actually use.
@@ -161,7 +158,6 @@ export function diffHeader(prev: EpochHeader, next: EpochHeader): HeaderDelta |
if (JSON.stringify(prevTools) !== JSON.stringify(nextTools)) delta.tools = diffTools(prevTools, nextTools)
if (!callConfigEquals(prev.config, next.config)) delta.config = next.config
if (!sameMessages(prev.messagePrefix, next.messagePrefix)) delta.messagePrefix = next.messagePrefix ?? []
if (!sameMessages(prev.messageSuffix, next.messageSuffix)) delta.messageSuffix = next.messageSuffix ?? []
return Object.keys(delta).length > 0 ? delta : undefined
}
@@ -177,13 +173,11 @@ export function applyHeaderDelta(prev: EpochHeader, delta: HeaderDelta): EpochHe
const system = delta.system !== undefined ? applySystem(prev.system, delta.system) : prev.system
const tools = delta.tools !== undefined ? applyTools(prev.tools ?? [], delta.tools) : prev.tools
const messagePrefix = delta.messagePrefix ?? prev.messagePrefix
const messageSuffix = delta.messageSuffix ?? prev.messageSuffix
return canonicalHeader({
config: delta.config ?? prev.config,
...system !== undefined ? { system } : {},
...tools !== undefined ? { tools } : {},
...messagePrefix !== undefined ? { messagePrefix } : {},
...messageSuffix !== undefined ? { messageSuffix } : {},
})
}

View File

@@ -185,14 +185,13 @@ export interface TodoItem {
/**
* The request header: everything about an LLM request besides its derived
* message history — the call configuration plus the rendered system prompt,
* tool schemas, and any request-only messages. Logged session state (the
* tool schemas, and the session prefix. Logged session state (the
* reconstructability RFC): a
* {@link SessionEventMap} `request/header` snapshot installs one, a
* `request/header-delta` amends it, and folding those events over the log
* (`foldRequestHeader`) reconstructs the header any request was built under.
* Canonical form: an empty system prompt, an empty tool list, and empty
* request-only message arrays are ABSENT fields, matching how requests are
* built.
* Canonical form: an empty system prompt, an empty tool list, and an empty
* prefix are ABSENT fields, matching how requests are built.
*/
export interface EpochHeader {
/** The conversation's call configuration (model + sampling scalars). */
@@ -202,14 +201,13 @@ export interface EpochHeader {
/** Assembled tool schemas; absent for a tool-less request. */
tools?: ToolSchema[]
/**
* Request-only messages sent BEFORE the derived history (the
* `agent/request-advice` waterfall's `before` contributions). Not session
* history — `deriveMessages()` never returns them — so the header is their
* only durable record; absent when the request carried none.
* The session prefix: request-only messages sent BEFORE the entire derived
* history (the `agent/session-prefix` waterfall's product, composed once
* per loop instance and reused for every request it sends). Not session
* history — `deriveMessages()` never returns it — so the header is its
* only durable record; absent when the instance composed none.
*/
messagePrefix?: Message[]
/** Request-only messages sent AFTER the derived history; absent when none. */
messageSuffix?: Message[]
}
/**
@@ -369,9 +367,11 @@ export interface SessionEventMap {
* Amendment to the folded {@link EpochHeader}: at least one of a
* {@link SystemDelta}, a {@link ToolsDelta}, a whole replacement
* {@link LlmCallConfig} (four scalars — not worth diffing), or a whole
* replacement request-only message array (`messagePrefix`/`messageSuffix` —
* small advisory content, replaced whole; an EMPTY array encodes the
* transition to "none", mirroring the canonical form's absent field).
* replacement session prefix (`messagePrefix` — small advisory content,
* replaced whole; an EMPTY array encodes the transition to "none",
* mirroring the canonical form's absent field — the loop never produces
* one in practice: the prefix is composed once per instance and anchored
* by that instance's snapshot, so this arm exists for codec totality).
* Appended by the
* loop inside the step, before dispatch, when the header for this request
* differs from the fold of the log so far; the writer verifies
@@ -379,7 +379,7 @@ export interface SessionEventMap {
* falls back to a `'fallback'` `request/header` snapshot when it cannot, so
* a logged delta ALWAYS round-trips. NOT a {@link SurfaceEventType}.
*/
'request/header-delta': { system?: SystemDelta; tools?: ToolsDelta; config?: LlmCallConfig; messagePrefix?: Message[]; messageSuffix?: Message[] }
'request/header-delta': { system?: SystemDelta; tools?: ToolsDelta; config?: LlmCallConfig; messagePrefix?: Message[] }
}
/** The appendable event-type keys of {@link SessionEventMap}, plugin-merged extensions included. */

View File

@@ -107,38 +107,37 @@ describe('diffHeader / applyHeaderDelta', () => {
})
})
describe('request-only messages (messagePrefix / messageSuffix)', () => {
it('canonicalHeader normalizes empty arrays to absent fields', () => {
expect(canonicalHeader({ config: CONFIG, messagePrefix: [], messageSuffix: [] })).toEqual({ config: CONFIG })
const full = canonicalHeader({ config: CONFIG, messagePrefix: [msg('p')], messageSuffix: [msg('s')] })
describe('the session prefix (messagePrefix)', () => {
it('canonicalHeader normalizes an empty prefix to an absent field', () => {
expect(canonicalHeader({ config: CONFIG, messagePrefix: [] })).toEqual({ config: CONFIG })
const full = canonicalHeader({ config: CONFIG, messagePrefix: [msg('p')] })
expect(full.messagePrefix).toEqual([msg('p')])
expect(full.messageSuffix).toEqual([msg('s')])
})
it('headerEquals treats absence and empty as one representation, content differences as unequal', () => {
expect(headerEquals(canonicalHeader({ config: CONFIG }), { config: CONFIG, messagePrefix: [] })).toBe(true)
expect(headerEquals({ config: CONFIG, messagePrefix: [msg('a')] }, { config: CONFIG, messagePrefix: [msg('b')] })).toBe(false)
expect(headerEquals({ config: CONFIG, messageSuffix: [msg('a')] }, { config: CONFIG })).toBe(false)
expect(headerEquals({ config: CONFIG, messagePrefix: [msg('a')] }, { config: CONFIG })).toBe(false)
})
it('replaces a changed prefix whole and leaves an untouched suffix alone', () => {
const prev = canonicalHeader({ config: CONFIG, messagePrefix: [msg('old')], messageSuffix: [msg('keep')] })
const next = canonicalHeader({ config: CONFIG, messagePrefix: [msg('new'), msg('more')], messageSuffix: [msg('keep')] })
it('replaces a changed prefix whole and leaves untouched parts alone', () => {
const prev = canonicalHeader({ config: CONFIG, system: 'keep', messagePrefix: [msg('old')] })
const next = canonicalHeader({ config: CONFIG, system: 'keep', messagePrefix: [msg('new'), msg('more')] })
const delta = roundTrip(prev, next)
expect(delta).toEqual({ messagePrefix: [msg('new'), msg('more')] })
})
it('round-trips framing gained from a bare header and lost back to one (empty array encodes absence)', () => {
it('round-trips a prefix gained from a bare header and lost back to one (empty array encodes absence)', () => {
const none = canonicalHeader({ config: CONFIG })
const some = canonicalHeader({ config: CONFIG, messagePrefix: [msg('p')], messageSuffix: [msg('s')] })
const some = canonicalHeader({ config: CONFIG, messagePrefix: [msg('p')] })
const gained = roundTrip(none, some)
expect(gained).toEqual({ messagePrefix: [msg('p')], messageSuffix: [msg('s')] })
expect(gained).toEqual({ messagePrefix: [msg('p')] })
const lost = roundTrip(some, none)
expect(lost).toEqual({ messagePrefix: [], messageSuffix: [] })
expect(lost).toEqual({ messagePrefix: [] })
})
it('folds framing deltas over the log like any other header amendment', () => {
const session = new Session(SessionId('fold-framing'))
it('folds prefix deltas over the log like any other header amendment', () => {
const session = new Session(SessionId('fold-prefix'))
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
const first = canonicalHeader({ config: CONFIG, messagePrefix: [msg('catalog v1')] })
session.append('request/header', { header: first, reason: 'initial' })