Merge remote-tracking branch 'origin/master' into codex/skill-system

This commit is contained in:
Yichen Jiang
2026-07-09 23:11:10 +08:00
438 changed files with 24091 additions and 1723 deletions

View File

@@ -56,12 +56,15 @@ forever:
STEP loop:
drain steering
assembly = systemPrompt.assemble({agent}) ⟵ renderPrompt(assembly) IS the full prompt
await serial agent/pre-step ⟵ surface mutation (compaction) outside the step
prefix ??= waterfall agent/session-prefix ⟵ once per instance (first step): frozen
session prefix; on the header, never history
await serial agent/pre-step(…, prefix) ⟵ surface mutation (compaction) outside the step;
pressure gates see the prefix the request carries
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
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: prefix+boundary})) → session('assistant/chunk')
message = waterfall agent/step-result
session('assistant/message')
each tool-call: session('tool/call')
@@ -85,7 +88,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/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

@@ -22,6 +22,10 @@ import { isTurnOpen, lastTurnNumber, runLoop } from './loop.ts'
* the agent/* event taxonomy — plugins never need this class.
*/
export class ReactLoopAgent implements Agent {
/**
* The queued + steering FIFOs behind {@link send}/{@link steer}. Public so
* the driver loop can drain it; {@link cancel} clears it wholesale.
*/
readonly inbox = new Inbox()
private _status: AgentStatus = 'idle'
@@ -256,6 +260,8 @@ export class ReactLoopAgent implements Agent {
* promise (unblocking the idle wait), releases any `whenIdle` waiters, and
* aborts the current request if any. The returned `agent.done` promise
* resolves once the loop exits.
* @returns the disposer — idempotent and infallible (it runs inside the
* fiber's LIFO disposal chain, where a throw would skip later disposers).
*/
start(): () => void {
this.done = runLoop(this.ctx, this, {

View File

@@ -24,30 +24,47 @@ export class Inbox {
private steeringMessages: InboxMessage[] = []
private wakeup: (() => void) | undefined
/** Resolves when a queued message arrives (used by the idle loop). */
/** True while queued messages are pending — read by the idle wait's fast path and the loop's turn-start checks. */
get hasQueued(): boolean {
return this.queuedMessages.length > 0
}
/** True while steering messages are pending — read by `cancel()`'s arm gate and the loop's stop-override check. */
get hasSteering(): boolean {
return this.steeringMessages.length > 0
}
/**
* Add a message to the queued FIFO and wake a parked {@link waitForQueued}.
* @param message - the message to queue for the next turn start.
*/
enqueue(message: InboxMessage): void {
this.queuedMessages.push(message)
this.wakeup?.()
}
/**
* Add a message to the steering FIFO. Deliberately no wakeup: steering is
* drained between steps of a running turn, never by the idle wait —
* `Agent.steer()` on an idle agent falls back to `send()` instead.
* @param message - the message to inject between steps of the running turn.
*/
steer(message: InboxMessage): void {
this.steeringMessages.push(message)
}
/** Drain all queued messages (turn start). */
/**
* Drain all queued messages (turn start).
* @returns the drained messages in arrival order; the queued FIFO is left empty.
*/
drainQueued(): InboxMessage[] {
return this.queuedMessages.splice(0)
}
/** Drain all steering messages (between steps). */
/**
* Drain all steering messages (between steps).
* @returns the drained messages in arrival order; the steering FIFO is left empty.
*/
drainSteering(): InboxMessage[] {
return this.steeringMessages.splice(0)
}
@@ -62,7 +79,12 @@ export class Inbox {
this.steeringMessages.length = 0
}
/** Wait until a queued message arrives or `cancel` resolves. */
/**
* Wait until a queued message arrives or `cancel` resolves.
* @param cancel - a promise whose resolution abandons the wait without a
* message (the driver loop passes the agent's disposed promise so a parked
* loop can exit).
*/
waitForQueued(cancel: Promise<void>): Promise<void> {
if (this.hasQueued) return Promise.resolve()
const { promise, resolve } = Promise.withResolvers<void>()

View File

@@ -29,9 +29,14 @@ declare module 'cordis' {
}
}
/**
* Plugin config: the agents to create — or resume, via `resumeSessionId` —
* declaratively at startup, so a cordis.yml deployment needs no code.
*/
export interface Config {
/** Agents created from configuration at startup. */
agents: (AgentOptions & {
/** Agent id to register under; also seeds the fresh per-run session id (`${id}-session-<uuid>`). */
id: AgentId
/** Optional workspace cwd for the config-created fresh session. */
cwd?: string

View File

@@ -157,13 +157,17 @@ export interface LoopHandle {
* drain steering → session('steering/message') ⟵ catches late steering
* assembly = ctx.systemPrompt.assemble({agent}) ⟵ waterfall system-prompt/assemble; renderPrompt
* (persona section + {{variables}}) IS the full prompt
* await ctx.serial('agent/pre-step') ⟵ surface mutation (compaction) OUTSIDE the step
* prefix ??= waterfall agent/session-prefix ⟵ once per loop instance (first step): frozen
* session prefix; logged on the header, never
* session history
* await ctx.serial('agent/pre-step', …, prefix) ⟵ surface mutation (compaction) OUTSIDE the step;
* pressure gates see the prefix the request carries
* 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
* 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: 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
@@ -185,6 +189,9 @@ export interface LoopHandle {
* re-enqueue leftover steering as queued ⟵ steering is never stranded
* idle (emit agent/status) unless more queued
* ```
* @param ctx - the plugin context the loop reaches events (agent/…, session/flush) and services (systemPrompt, llm, tools) through.
* @param agent - the agent this invocation drives for its whole lifetime (its inbox, session, and options).
* @param handle - the bridge to the agent's mutable state: status/abort setters plus the disposal and cancel-marker reads.
*/
export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle): Promise<void> {
// Per-instance transmission bookkeeping: whether THIS loop instance has
@@ -469,6 +476,50 @@ async function runTurn(
break
}
// Compose the session prefix ONCE per loop instance, lazily before the
// instance's first pre-step: request-only messages placed in front of
// the ENTIRE derived history on every request this instance sends. It
// MUST precede the pre-step seam so compaction gates on THIS instance's
// prefix — reading a previous instance's logged prefix would let a
// resumed/forked instance whose contributor grew skip compaction and
// ship an over-window first request. 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 in runStep is its only durable record
// (EpochHeader.messagePrefix). 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. This
// runs OUTSIDE the step, before the boundary snapshot: a composing
// listener's session append lands before the boundary and joins the
// CURRENT request.
if (transmission.sessionPrefix === undefined) {
const emptyPrefix: Message[] = deepFreeze([])
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, 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
// step: after `turn/start` (and the prior step's close) but before
// `step/start`, so a compaction's log-only `compact/*` records and its
@@ -479,8 +530,10 @@ async function runTurn(
// concurrent listeners cannot interleave their `session.append`s. A
// 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, fullSystemPrompt, abort.signal)
// pre-step plugin ends the turn, not the loop. The composed session
// prefix rides along so token-pressure listeners count everything the
// request will actually carry.
await ctx.serial('agent/pre-step', agent, turn, step, fullSystemPrompt, transmission.sessionPrefix, abort.signal)
// Interruption landing during the pre-step seam: do not open an empty step.
if (handle.isCancelled() || handle.isDisposed()) {
@@ -671,11 +724,12 @@ 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 → 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. */
async function runStep(
ctx: Context,
agent: ReactLoopAgent,
@@ -715,22 +769,30 @@ async function runStep(
throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
}
// The session prefix was composed (once per instance) before this step's
// pre-step seam — the caller guarantees it, so the cache is always set here.
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- runTurn composes the prefix before every runStep call
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.
// recorded before dispatch so the log always explains the request
// including the session prefix, which no other event carries.
const header = canonicalHeader({
config,
...system ? { system } : {},
...assembly.tools.length > 0 ? { tools: assembly.tools } : {},
...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.
// 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: boundaryMessages,
messages: [...header.messagePrefix ?? [], ...boundaryMessages],
...header.system !== undefined ? { system: header.system } : {},
...header.tools !== undefined ? { tools: header.tools } : {},
...header.config.temperature !== undefined ? { temperature: header.config.temperature } : {},
@@ -875,7 +937,11 @@ function withoutToolCalls(message: Message): Message {
return { ...message, content: message.content.filter(block => block.type !== 'tool-call') }
}
/** The last turn number in a (possibly seeded) session log, or 0. */
/**
* The last turn number in a (possibly seeded) session log, or 0.
* @param session - the session whose log is scanned for the latest `turn/start`.
* @returns the latest `turn/start`'s turn number, or 0 when the log has none (the next turn is this plus one).
*/
export function lastTurnNumber(session: Session): number {
const lastStart = session.events.findLast(event => event.type === 'turn/start')
return lastStart?.data.turn ?? 0
@@ -889,6 +955,8 @@ export function lastTurnNumber(session: Session): number {
* returns to idle), so status is not a reliable open-turn signal. Used by
* `inject()` to choose between appending into an open turn vs. wrapping the
* injection in its own one-shot turn (the turn-enclosure RFC).
* @param session - the session whose log is inspected.
* @returns true when the log's last turn boundary is a `turn/start` with no matching `turn/end` yet.
*/
export function isTurnOpen(session: Session): boolean {
const last = session.events.findLast(e => e.type === 'turn/start' || e.type === 'turn/end')

View File

@@ -12,14 +12,26 @@
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[]
}
/** Fresh bookkeeping for a newly-started loop instance. */
/**
* Fresh bookkeeping for a newly-started loop instance.
* @returns state with `loggedHeader` false, so the instance's first request appends an anchoring snapshot.
*/
export function createTransmissionLog(): TransmissionLog {
return { loggedHeader: false }
}
@@ -58,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

@@ -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'
@@ -166,6 +166,103 @@ describe('Agent.cancel()', () => {
expect(reasons.length).toBe(2)
})
it('cancel from inside the agent/session-prefix waterfall drops the step (prefix-composition window)', async () => {
const adapter = new MockAdapter([textResponse('should not stream')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
// Prefix composition runs before the pre-step seam on the instance's first
// step; a cancel landing inside it must drop the about-to-start step
// without running the seam or the model.
let streamed = false
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next) => {
agent.cancel('from prefix composition')
return next()
})
const reasons: TurnEndReason[] = []
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
send(agent, 'go')
await waitForIdle(ctx, agent)
expect(streamed).toBe(false)
expect(reasons).toEqual([{ kind: 'aborted', reason: 'from prefix composition' }])
})
it('disposal from inside the agent/session-prefix waterfall ends the turn disposed (prefix-composition window)', async () => {
const adapter = new MockAdapter([textResponse('should not stream')])
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: [] })
ctx.llm.registerAdapter(['mock'], adapter)
const handle = ctx.agents.create({
agentId: AgentId('a-dispose-prefix'),
sessionId: SessionId('dispose-prefix-session'),
agentOptions: { model: 'mock' },
})
const agent = handle.agent as ReactLoopAgent
let disposalDone: Promise<void> | undefined
let streamed = false
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next) => {
disposalDone = handle.dispose()
return next()
})
send(agent, 'go')
await new Promise(resolve => setTimeout(resolve, 0))
await disposalDone
await agent.done
// No step opened, no model call ran, and the turn closed disposed.
expect(streamed).toBe(false)
expect(adapter.requests).toHaveLength(0)
const turnEnd = agent.session.events.findLast(e => e.type === 'turn/end')
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<Message[]> => {
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)

View File

@@ -1,6 +1,6 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService, { CallId } from '@deepseek-ai/dsh-llm'
import LlmService, { CallId, type Message } from '@deepseek-ai/dsh-llm'
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'
@@ -310,6 +310,162 @@ describe('agent/session-start', () => {
})
})
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>' }] }
let composed = 0
ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next): Promise<Message[]> => {
composed += 1
return [...await next(), reminder]
})
send(agent, 'go')
await waitForIdle(ctx, agent)
send(agent, 'next turn')
await waitForIdle(ctx, agent)
// 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('composes before the first pre-step and hands the prefix to the seam (pressure gates see the real value)', 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: 'opener' }] }
const order: string[] = []
ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next): Promise<Message[]> => {
order.push('compose')
return [reminder, ...await next()]
})
const seen: (readonly Message[])[] = []
ctx.on('agent/pre-step', (_agent, _turn, _step, _system, sessionPrefix) => {
order.push('pre-step')
seen.push(sessionPrefix)
})
send(agent, 'hi')
await waitForIdle(ctx, agent)
// Composition precedes the pre-step seam, and the seam receives THIS
// instance's composed prefix — a token-pressure gate (compaction) counts
// what the request will actually carry, never a stale logged prefix.
expect(order).toEqual(['compose', 'pre-step'])
expect(seen[0]).toEqual([reminder])
})
it('the canonical prepend pattern composes contributions in registration order', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
// Both listeners use the canonical `[mine, ...await next()]` prepend: the
// waterfall unwinds innermost-first (the second listener's array is built
// first), so prepending puts the FIRST-registered contribution first.
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/session-prefix', async (_agent, _prefix, _signal, next): Promise<Message[]> => {
return [{ role: 'user', content: [{ type: 'text', text: 'second' }] }, ...await next()]
})
send(agent, 'hi')
await waitForIdle(ctx, agent)
const texts = adapter.requests[0]!.messages.map(m => m.content[0]?.type === 'text' ? m.content[0].text : '')
expect(texts).toEqual(['first', 'second', 'hi'])
})
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/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(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/session-prefix', async (_agent, prefix, _signal, next): Promise<Message[]> => {
try {
prefix.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('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'),
])
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 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)
// 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

@@ -0,0 +1,117 @@
/**
* Loop-level tool-order determinism: the request/header event — and therefore
* the frozen request the adapter receives — carries the assembly's canonical
* tool order (system-prompt's `toolOrder` config, or lexicographic name
* order), regardless of the order tool plugins happened to register in.
* Registration order is a plugin-load artifact (concurrent dynamic imports
* race), so nothing downstream of the registry may depend on it.
*/
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService from '@deepseek-ai/dsh-llm'
import SessionStore, { foldRequestHeader } from '@deepseek-ai/dsh-session'
import SystemPrompt, { TOOL_ORDER_REST } from '@deepseek-ai/dsh-system-prompt'
import type { Config as SystemPromptConfig } from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent'
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
import { MockAdapter, textResponse } from './mock-adapter.ts'
async function harness(adapter: MockAdapter, toolOrder?: SystemPromptConfig['toolOrder']) {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt, { persona: 'stable base', ...toolOrder !== undefined ? { toolOrder } : {} })
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(AgentLoop, { agents: [] })
ctx.llm.registerAdapter(['mock'], adapter)
return ctx
}
function waitForIdle(ctx: Context, agent: ReactLoopAgent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()
}
})
})
}
function registerNamed(ctx: Context, name: string) {
ctx.tools.register(defineTool({
name,
description: `the ${name} tool`,
parameters: {},
async execute() {
return [{ type: 'text', text: name }]
},
}))
}
/** Run one text-only turn and return the harness context + agent. */
async function runTurn(registrationOrder: string[], toolOrder?: SystemPromptConfig['toolOrder']) {
const adapter = new MockAdapter([textResponse('done')])
const ctx = await harness(adapter, toolOrder)
for (const name of registrationOrder) registerNamed(ctx, name)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await waitForIdle(ctx, agent)
return { ctx, agent, adapter }
}
describe('loop-level canonical tool order', () => {
it('logs the request/header with tools in canonical order, not registration order', async () => {
const { agent, adapter } = await runTurn(['zulu', 'alpha', 'mike'])
const header = foldRequestHeader(agent.session.events)
expect(header?.tools?.map(tool => tool.name)).toEqual(['alpha', 'mike', 'zulu'])
// The dispatched request is built FROM the logged header (whose tools the
// assembly already canonicalized) and reaches the adapter deep-frozen —
// the marker the reconstruction invariant keys on.
expect(adapter.requests[0]?.tools?.map(tool => tool.name)).toEqual(['alpha', 'mike', 'zulu'])
expect(Object.isFrozen(adapter.requests[0])).toBe(true)
expect(adapter.requests[0]?.sessionId).toBe(agent.session.id)
})
it('produces the same header order for any registration order', async () => {
const first = await runTurn(['alpha', 'mike', 'zulu'])
const second = await runTurn(['zulu', 'mike', 'alpha'])
const names = (run: typeof first) => foldRequestHeader(run.agent.session.events)?.tools?.map(tool => tool.name)
expect(names(first)).toEqual(['alpha', 'mike', 'zulu'])
expect(names(second)).toEqual(names(first))
})
it('honors a configured toolOrder in the logged header and the dispatched request', async () => {
const { agent, adapter } = await runTurn(['alpha', 'zulu', 'mike'], ['zulu', TOOL_ORDER_REST])
const header = foldRequestHeader(agent.session.events)
expect(header?.tools?.map(tool => tool.name)).toEqual(['zulu', 'alpha', 'mike'])
expect(adapter.requests[0]?.tools?.map(tool => tool.name)).toEqual(['zulu', 'alpha', 'mike'])
expect(Object.isFrozen(adapter.requests[0])).toBe(true)
})
it('fails the turn — no model request — when toolOrder names an unregistered tool', async () => {
// The assemble rejection escapes to runTurn's outer catch: the open turn
// closes with an `error` reason (agent/error mirrors it), no step opens,
// no request/header is logged, the adapter never sees a request, and the
// agent returns to idle — a misconfigured deployment fails every turn
// deterministically instead of silently reordering nothing.
const adapter = new MockAdapter([textResponse('never sent')])
const ctx = await harness(adapter, ['ghost', TOOL_ORDER_REST])
registerNamed(ctx, 'alpha')
const errors: Error[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(0)
expect(errors.map(e => e.message)).toEqual(['toolOrder lists unregistered tool "ghost"; registered tools: alpha'])
expect(foldRequestHeader(agent.session.events)).toBeUndefined()
const end = agent.session.events.find(e => e.type === 'turn/end')
expect(end?.type === 'turn/end' && end.data.reason).toMatchObject({ kind: 'error', step: 1 })
// The turn is balanced (turn/start → turn/end) with no step events inside.
expect(agent.session.events.some(e => e.type === 'step/start')).toBe(false)
})
})