fix(agent-loop): compose the session prefix before pre-step; hand it to the pressure gate
ds-review-bot critical (follow-up): on the first step of a resumed or seeded/forked instance, auto-compaction ran before runStep composed this instance's prefix, so the gate read the PREVIOUS instance's logged prefix from the header fold — a contributor that grew across resume/fork (skills added, AGENTS.md grown: exactly the environment-dependent case) could under-gate and ship an over-window first request. The loop now composes agent/session-prefix before the instance's first agent/pre-step (still once per instance; runStep just reads the cache), and agent/pre-step carries the composed prefix to its listeners. CompactService.compactIfNeeded gains the sessionPrefix parameter; BasicCompactService.estimatePressure gates on the handed value — the header-fold read is gone, so the estimate is exact at every step including a resumed/forked instance's first. Composition moving before the boundary snapshot also means a composing listener's session append now joins the CURRENT request (documented on the seam). New coverage: composition precedes pre-step and the seam receives the composed prefix; cancel and disposal landing inside the composition window drop the step cleanly; the compact gate test hands the prefix directly.
This commit is contained in:
@@ -55,12 +55,13 @@ 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
|
||||
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: prefix+boundary})) → session('assistant/chunk')
|
||||
message = waterfall agent/step-result
|
||||
|
||||
@@ -157,13 +157,14 @@ 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
|
||||
* 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: prefix+boundary, sessionId, signal})
|
||||
@@ -475,6 +476,41 @@ 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([])
|
||||
transmission.sessionPrefix = deepFreeze(structuredClone(await ctx.waterfall(
|
||||
'agent/session-prefix', agent, emptyPrefix, abort.signal,
|
||||
() => Promise.resolve(emptyPrefix),
|
||||
)))
|
||||
}
|
||||
|
||||
// Interruption landing during prefix composition: mirror the assembly
|
||||
// window above — drop the about-to-start step without running the seam.
|
||||
if (handle.isCancelled() || handle.isDisposed()) {
|
||||
handle.setAbort(undefined)
|
||||
reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
|
||||
break
|
||||
}
|
||||
|
||||
// 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
|
||||
@@ -485,8 +521,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()) {
|
||||
@@ -722,27 +760,10 @@ async function runStep(
|
||||
throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
|
||||
}
|
||||
|
||||
// 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 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 —
|
||||
|
||||
@@ -166,6 +166,69 @@ 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('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)
|
||||
|
||||
@@ -352,6 +352,33 @@ describe('agent/session-prefix', () => {
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user