feat(agent): add the agent/request-messages request-only message seam
A new waterfall near request construction lets plugins contribute
request-ONLY messages framing the derived history: RequestMessages
{ before, after } with a frozen empty seed, fired inside the open step
after the agent/request config waterfall, so the step/start boundary
snapshot and its same-sync-frame invariant are untouched. The request
becomes messagePrefix + boundary snapshot + messageSuffix.
Contributions never enter session history — deriveMessages() is
unchanged — so the request header is their durable record:
EpochHeader gains messagePrefix/messageSuffix (canonical absence for
empty arrays), request/header-delta replaces either array whole with
an empty array encoding the transition back to absence, and the
dev-mode reconstruction cross-check now expects the folded header's
framing around the boundary derivation.
This is the seam for per-request advisory context that must be
model-visible now without becoming durable history (a skills catalog,
an environment reminder), keeping the base system prompt
workspace-independent and provider prefix caches stable. The docs
carry the channel cost model: session-frozen content belongs in
before, low-frequency change notices belong in durable history via
inject() (paid once, prefix-cached thereafter), and after is reserved
for small frequently-refreshed state snapshots re-paid on every
request they ride. No shipped producer yet, so ACP snapshot fixtures
are byte-identical.
This commit is contained in:
Yichen Jiang
@@ -59,8 +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-messages ⟵ request-only before/after messages; recorded
|
||||
on the header, never session history
|
||||
session('request/header'[-delta]) ⟵ the header event this request owes the log
|
||||
stream llm.stream(freeze({header..., messages: boundary})) → session('assistant/chunk')
|
||||
stream llm.stream(freeze({header..., messages: before+boundary+after})) → session('assistant/chunk')
|
||||
message = waterfall agent/step-result
|
||||
session('assistant/message')
|
||||
each tool-call: session('tool/call')
|
||||
@@ -84,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/step-result`, `tools/pre-execute`, `tools/post-execute`, `agent/turn-continuation`
|
||||
- Hooks: `agent/session-start`, `agent/prompt-submit`, `agent/pre-step`, `agent/request`, `agent/request-messages`, `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.
|
||||
|
||||
@@ -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 } from '@deepseek-ai/dsh-agent'
|
||||
import type { ContinuationDecision, HookContext, PromptDecision, RequestMessages } 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,9 +161,11 @@ 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
|
||||
* reqMsgs = waterfall agent/request-messages ⟵ request-only before/after messages; 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: boundary, sessionId, signal})
|
||||
* req = freeze({header..., messages: before+boundary+after, 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
|
||||
@@ -671,11 +673,11 @@ function drainSteering(agent: ReactLoopAgent, turn: number): boolean {
|
||||
}
|
||||
|
||||
/** One step: build the request from the boundary snapshot + the step's
|
||||
* header → 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. */
|
||||
* header → collect request-only messages → 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. */
|
||||
async function runStep(
|
||||
ctx: Context,
|
||||
agent: ReactLoopAgent,
|
||||
@@ -715,22 +717,43 @@ async function runStep(
|
||||
throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
|
||||
}
|
||||
|
||||
// Collect request-ONLY messages: `before` contributions precede the boundary
|
||||
// snapshot in the request, `after` contributions follow it. They are not
|
||||
// session history — the header event below is their 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. 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 emptyRequestMessages: RequestMessages = deepFreeze({ before: [], after: [] })
|
||||
const requestMessages = await ctx.waterfall(
|
||||
'agent/request-messages', agent, turn, step, emptyRequestMessages,
|
||||
{ system, assembly, boundaryMessages, signal },
|
||||
() => Promise.resolve(emptyRequestMessages),
|
||||
)
|
||||
|
||||
// The request header (the log's request/header* vocabulary): canonical form,
|
||||
// recorded before dispatch so the log always explains the request.
|
||||
// recorded before dispatch so the log always explains the request —
|
||||
// including the request-only messages, which no other event carries.
|
||||
const header = canonicalHeader({
|
||||
config,
|
||||
...system ? { system } : {},
|
||||
...assembly.tools.length > 0 ? { tools: assembly.tools } : {},
|
||||
...requestMessages.before.length > 0 ? { messagePrefix: requestMessages.before } : {},
|
||||
...requestMessages.after.length > 0 ? { messageSuffix: requestMessages.after } : {},
|
||||
})
|
||||
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.
|
||||
// keys on. Message order: header.messagePrefix, then the boundary snapshot,
|
||||
// then header.messageSuffix — the reconstruction equation the invariant
|
||||
// recomputes.
|
||||
const request: GenerateOptions = deepFreeze({
|
||||
model: header.config.model,
|
||||
messages: boundaryMessages,
|
||||
messages: [...header.messagePrefix ?? [], ...boundaryMessages, ...header.messageSuffix ?? []],
|
||||
...header.system !== undefined ? { system: header.system } : {},
|
||||
...header.tools !== undefined ? { tools: header.tools } : {},
|
||||
...header.config.temperature !== undefined ? { temperature: header.config.temperature } : {},
|
||||
|
||||
@@ -1,13 +1,14 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { type SessionEvent, type TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
import LlmService, { CallId, type Message } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { foldRequestHeader, 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 RequestMessages,
|
||||
type SessionStartSource,
|
||||
} from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop, { type ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
@@ -310,6 +311,145 @@ describe('agent/session-start', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('agent/request-messages (RequestMessages)', () => {
|
||||
it('frames the derived history: before precedes it, after follows it, and the header records both', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
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-messages', async (_agent, _turn, _step, _messages, _context, next): Promise<RequestMessages> => {
|
||||
const result = await next()
|
||||
return { before: [...result.before, reminder], after: [...result.after, trailer] }
|
||||
})
|
||||
|
||||
send(agent, 'hi')
|
||||
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' }] },
|
||||
])
|
||||
})
|
||||
|
||||
it('contributions compose across listeners and see the read-only request facts', 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-messages', async (_agent, _turn, _step, _messages, context, next): Promise<RequestMessages> => {
|
||||
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/request-messages', async (_agent, _turn, _step, _messages, _context, next): Promise<RequestMessages> => {
|
||||
const result = await next()
|
||||
return { before: [...result.before, { role: 'user', content: [{ type: 'text', text: 'second' }] }], after: result.after }
|
||||
})
|
||||
|
||||
send(agent, 'hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// Registration order composes: the first listener runs last on the way
|
||||
// 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 () => {
|
||||
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-messages', async (_agent, _turn, _step, _messages, _context, 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' }] }])
|
||||
})
|
||||
|
||||
it('the frozen seed rejects in-place mutation — a contribution is a returned extension', 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-messages', async (_agent, _turn, _step, messages, _context, next): Promise<RequestMessages> => {
|
||||
try {
|
||||
messages.before.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 () => {
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'echo', { text: 'ping' }),
|
||||
textResponse('done'),
|
||||
])
|
||||
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' })
|
||||
|
||||
let step = 0
|
||||
ctx.on('agent/request-messages', async (_agent, _turn, _step, _messages, _context, next): Promise<RequestMessages> => {
|
||||
const result = await next()
|
||||
step += 1
|
||||
return { before: [...result.before, { role: 'user', content: [{ type: 'text', text: `reminder v${step}` }] }], after: result.after }
|
||||
})
|
||||
|
||||
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' }] }])
|
||||
})
|
||||
})
|
||||
|
||||
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')])
|
||||
|
||||
Reference in New Issue
Block a user