fix(compact): decide step-alignment from surface tool-pairing, fire compaction pre-step (CBR-001)
Codex round 1 CBR-001: a head-anchored compaction checkpoint was mis-classified by the log-position step-alignment scan, so a second auto-compaction over a checkpoint-headed surface silently failed. Root cause: `isStepAlignedStart/End` scanned the LOG by seq, but a `replace` op lands a checkpoint at a high log seq whose SURFACE position is the head — its log neighbours (the open step's assistant/message) are not its surface neighbours, so the forward scan wrongly reported mid-step. Fix, per the agreed direction: - Replace the two log-position predicates with one surface-anchored helper `isToolPairingBalanced(nodes, events, beforeSeq)` in `dsh-session` (renamed step-boundary.ts → tool-pairing.ts). A cut is balanced when no unanswered tool-call precedes it on the surface; a region is collapsible iff both edges are balanced cuts. The open-tail and free-node cases fall out of the same counter. It also throws on a corrupt surface (a tool/result with no matching call). - Move compaction off the in-step seam to a new "pre-step" seam fired after turn/start and before step/start, so a compaction's log-only compact/* records and its replacement node land cleanly OUTSIDE any step (the honest structure crash-safety relies on). Renamed the event agent/pre-request → agent/pre-step and switched its dispatch from parallel → serial (listeners mutate the surface as a side effect; serial isolates them so concurrent appends can't interleave). Extended the catalog generator to accept @mode serial. Regression coverage: a real-loop test driving an auto-compaction asserts the landed checkpoint is a balanced cut on both sides; unit tests pin the checkpoint case, the mid-step injection case, multi-call steps, and the corrupt-surface guard. Proven red on the old log-position logic.
This commit is contained in:
Hypatia May
@@ -12,6 +12,7 @@ import type { FinishReason, GenerateOptions, Message } from '@deepseek-ai/dsh-ll
|
||||
import { BlockAssembler, HarnessError } from '@deepseek-ai/dsh-llm'
|
||||
import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
|
||||
import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
|
||||
import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
|
||||
import type {} from '@deepseek-ai/dsh-tools'
|
||||
import type { ReactLoopAgent } from './agent.ts'
|
||||
|
||||
@@ -147,9 +148,9 @@ export interface LoopHandle {
|
||||
* drain queued → 'turn/start' → session('user/message'…) → emit agent/turn-start
|
||||
* STEP loop:
|
||||
* drain steering → session('steering/message') ⟵ catches late steering
|
||||
* session('step/start'); emit agent/step-start ⟵ append before emit (the event-sourcing RFC)
|
||||
* assembly = ctx.systemPrompt.assemble() ⟵ waterfall system-prompt/assemble
|
||||
* await ctx.parallel('agent/pre-request') ⟵ surface mutation (compaction) BEFORE derive
|
||||
* await ctx.serial('agent/pre-step') ⟵ surface mutation (compaction) OUTSIDE the step
|
||||
* session('step/start'); emit agent/step-start ⟵ append before emit (the event-sourcing RFC)
|
||||
* req = {model, system, tools, messages: session.deriveMessages(), signal}
|
||||
* req = waterfall agent/request ⟵ hooks/model-switch
|
||||
* stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks)
|
||||
@@ -387,20 +388,54 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
|
||||
// (or turn-start listeners on the first step) joins before the request.
|
||||
drainSteering(ctx, agent, turn)
|
||||
|
||||
// Assemble the system prompt for this step. Done HERE (before step/start)
|
||||
// because the pre-step seam needs it: compaction measures token pressure
|
||||
// against the system prompt (it counts toward the budget) and a listener
|
||||
// also receives the model to summarize with. runStep reuses this same
|
||||
// assembly for the request, so the prompt is assembled once per step.
|
||||
const assembly = await ctx.systemPrompt.assemble()
|
||||
const system = [renderPrompt(assembly), agent.options.systemPrompt ?? '']
|
||||
.filter(text => text.length > 0)
|
||||
.join('\n\n')
|
||||
|
||||
// The step's AbortController exists BEFORE the pre-step seam so a cancel()
|
||||
// during the seam aborts any in-flight work a listener started (e.g. a
|
||||
// compaction summarization call). Cleared on every exit path below.
|
||||
const abort = new AbortController()
|
||||
handle.setAbort(abort)
|
||||
|
||||
// Cancel landing before the seam: a synchronous `agent/turn-start` listener
|
||||
// (or the previous step's continuation listeners) can have called
|
||||
// `cancel()`. Drop the about-to-start step WITHOUT running the seam — no
|
||||
// step is open yet, so end the turn `aborted` directly.
|
||||
if (handle.isCancelled()) {
|
||||
handle.setAbort(undefined)
|
||||
reason = { 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
|
||||
// replacement node land cleanly outside any step (honest structure that
|
||||
// crash-safety relies on — a dangling `compact/start` sits before the
|
||||
// synthetic `turn/end` repair appends). Serial (awaited, in order, no
|
||||
// veto): each listener completes its surface mutation before the next, so
|
||||
// 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, system, agent.options.model ?? '', abort.signal)
|
||||
|
||||
session.append('step/start', { turn, step })
|
||||
stepOpen = true
|
||||
ctx.emit('agent/step-start', agent, turn, step)
|
||||
|
||||
const abort = new AbortController()
|
||||
handle.setAbort(abort)
|
||||
|
||||
// Cancel landing in the step-start window: a synchronous `agent/turn-start`
|
||||
// or `agent/step-start` listener (both fire before this point) can have
|
||||
// called `cancel()`, and `runStep` would otherwise run a full extra step
|
||||
// with no AbortController having observed it. Check the marker AFTER
|
||||
// setAbort (so the next-iteration drain sees a clean controller) and before
|
||||
// `runStep`: drop the step, end the turn `aborted`. closeStep balances the
|
||||
// already-appended step/start.
|
||||
// Cancel landing in the seam / step-start window: a `cancel()` during the
|
||||
// pre-step seam (it aborted `abort.signal` above) OR a synchronous
|
||||
// `agent/step-start` listener that cancels. Check AFTER setAbort/step-start
|
||||
// and before `runStep`: drop the step, end the turn `aborted`. closeStep
|
||||
// balances the already-appended step/start.
|
||||
if (handle.isCancelled()) {
|
||||
handle.setAbort(undefined)
|
||||
reason = { kind: 'aborted', reason: handle.cancelReason() }
|
||||
@@ -410,7 +445,7 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
|
||||
|
||||
let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
|
||||
try {
|
||||
stepOutcome = await runStep(ctx, agent, turn, step, abort.signal)
|
||||
stepOutcome = await runStep(ctx, agent, turn, step, assembly, system, abort.signal)
|
||||
} catch (error: unknown) {
|
||||
stepOutcome = { error: toError(error) }
|
||||
} finally {
|
||||
@@ -550,29 +585,22 @@ function drainSteering(ctx: Context, agent: ReactLoopAgent, turn: number): boole
|
||||
return messages.length > 0
|
||||
}
|
||||
|
||||
/** One step: assemble request → stream model → record → execute tools. */
|
||||
/** One step: derive request from the (already pre-step-mutated) surface →
|
||||
* stream model → record → execute tools. The caller assembles the system prompt
|
||||
* and fires the `agent/pre-step` seam BEFORE opening the step, then passes the
|
||||
* resulting `assembly`/`system` here, so the surface this step derives from
|
||||
* already reflects any compaction. */
|
||||
async function runStep(
|
||||
ctx: Context,
|
||||
agent: ReactLoopAgent,
|
||||
turn: number,
|
||||
step: number,
|
||||
assembly: PromptAssembly,
|
||||
system: string,
|
||||
signal: AbortSignal,
|
||||
): Promise<{ hadToolCalls: boolean; finish: FinishReason }> {
|
||||
const { session, options } = agent
|
||||
|
||||
// --- Request assembly ---
|
||||
const assembly = await ctx.systemPrompt.assemble()
|
||||
const system = [renderPrompt(assembly), options.systemPrompt ?? '']
|
||||
.filter(text => text.length > 0)
|
||||
.join('\n\n')
|
||||
|
||||
// Surface-mutation checkpoint BEFORE deriving history: compaction shadows an
|
||||
// older range with a summary node here, and the single derive below reflects
|
||||
// it. Awaited (no veto) — a listener mutates the surface as a side effect.
|
||||
// `model` is resolved to '' when unset; a compaction listener that needs a
|
||||
// model falls back to its own config.
|
||||
await ctx.parallel('agent/pre-request', agent, turn, step, system, options.model ?? '', signal)
|
||||
|
||||
let request: GenerateOptions = {
|
||||
model: options.model ?? '',
|
||||
messages: session.deriveMessages(),
|
||||
|
||||
@@ -194,6 +194,36 @@ describe('Agent.cancel()', () => {
|
||||
expect(reasons).toEqual([{ kind: 'aborted', reason: 'from turn-start' }])
|
||||
})
|
||||
|
||||
it('cancel from a synchronous agent/step-start listener drops the step (post-step-start window)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('should not stream')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// A step-start listener fires AFTER step/start is appended (and after the
|
||||
// pre-step seam), so cancelling there lands in the SECOND cancel check (the
|
||||
// one that must closeStep() to balance the already-open step) — distinct
|
||||
// from a turn-start cancel, which is caught before the step opens.
|
||||
let streamed = false
|
||||
ctx.on('agent/stream-chunk', () => { streamed = true })
|
||||
const dispose = ctx.on('agent/step-start', (subject) => {
|
||||
if (subject === agent) agent.cancel('from step-start')
|
||||
})
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
dispose()
|
||||
|
||||
// No step streamed, the turn ended aborted with the caller's reason, and the
|
||||
// log is balanced (the open step was closed by the cancel branch).
|
||||
expect(streamed).toBe(false)
|
||||
expect(reasons).toEqual([{ kind: 'aborted', reason: 'from step-start' }])
|
||||
const types = agent.session.events.map(e => e.type)
|
||||
expect(types.filter(t => t === 'step/start').length).toBe(types.filter(t => t === 'step/end').length)
|
||||
})
|
||||
|
||||
it('cancel during the continuation window ends the turn aborted and runs no further step', async () => {
|
||||
// A continuation-waterfall listener cancels DURING the continuation decision
|
||||
// (the finished step's AbortController is already cleared), and votes to
|
||||
|
||||
@@ -320,11 +320,11 @@ describe('agent loop', () => {
|
||||
expect(adapter.requests[0]!.model).toBe('other-model')
|
||||
})
|
||||
|
||||
it('agent/pre-request fires once per step before the request is derived', async () => {
|
||||
it('agent/pre-step fires once per step before the step is opened', async () => {
|
||||
// Two steps (a tool call, then a final text turn) → two model calls → two
|
||||
// pre-request fires, each carrying the assembled system + model, BEFORE the
|
||||
// request messages are derived (the request the adapter sees reflects any
|
||||
// surface state at fire time).
|
||||
// pre-step fires, each carrying the assembled system + model, BEFORE the
|
||||
// step is opened and its request is derived (the request the adapter sees
|
||||
// reflects any surface state at fire time).
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'echo', {}, 'calling echo'),
|
||||
textResponse('done'),
|
||||
@@ -337,7 +337,7 @@ describe('agent loop', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const fires: { turn: number; step: number; model: string }[] = []
|
||||
ctx.on('agent/pre-request', (subject, turn, step, _system, model) => {
|
||||
ctx.on('agent/pre-step', (subject, turn, step, _system, model) => {
|
||||
if (subject === agent) fires.push({ turn, step, model })
|
||||
})
|
||||
|
||||
@@ -351,32 +351,77 @@ describe('agent loop', () => {
|
||||
])
|
||||
})
|
||||
|
||||
it('a surface mutation in agent/pre-request is reflected in the derived request (single derive)', async () => {
|
||||
// pre-request fires BEFORE deriveMessages(), so a listener that appends a
|
||||
// surface node there sees it land in the SAME step's request — proving the
|
||||
// loop derives once, after the checkpoint, with no stale pre-derive.
|
||||
it('agent/pre-step fires BEFORE the step it precedes opens (events land outside the step)', async () => {
|
||||
// A listener appending a surface node in pre-step lands it BEFORE step/start
|
||||
// in the log — proving the seam fires outside the step. The node is still in
|
||||
// the derived request for that step (derive happens after step/start).
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let injected = false
|
||||
ctx.on('agent/pre-request', (subject, turn) => {
|
||||
ctx.on('agent/pre-step', (subject) => {
|
||||
if (subject === agent && !injected) {
|
||||
injected = true
|
||||
subject.session.append('context/message', {
|
||||
content: [{ type: 'text', text: 'INJECTED-IN-PRE-REQUEST' }],
|
||||
content: [{ type: 'text', text: 'INJECTED-IN-PRE-STEP' }],
|
||||
source: { kind: 'plugin', plugin: 'test' },
|
||||
}, { surfaceOp: 'append' })
|
||||
void turn
|
||||
}
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// The adapter's request includes the node injected during pre-request.
|
||||
// The adapter's request includes the node injected during pre-step (derive
|
||||
// reflects it).
|
||||
const text = JSON.stringify(adapter.requests[0]!.messages)
|
||||
expect(text).toContain('INJECTED-IN-PRE-REQUEST')
|
||||
expect(text).toContain('INJECTED-IN-PRE-STEP')
|
||||
|
||||
// And the injected event sits BEFORE the first step/start in the log —
|
||||
// the seam fired outside the step.
|
||||
const events = agent.session.events
|
||||
const injectedSeq = events.find(e => e.type === 'context/message')!.seq
|
||||
const firstStepStartSeq = events.find(e => e.type === 'step/start')!.seq
|
||||
expect(injectedSeq).toBeLessThan(firstStepStartSeq)
|
||||
})
|
||||
|
||||
it('a throwing agent/pre-step listener ends the turn (error), not the loop', async () => {
|
||||
// The seam fires before step/start, so a throw escapes to runTurn's outer
|
||||
// catch: the not-yet-open step closes as a no-op, the failure surfaces via
|
||||
// agent/error, and the turn ends `error` (recorded on the durable turn/end).
|
||||
// The loop survives and a follow-up prompt still runs.
|
||||
const adapter = new MockAdapter([textResponse('second turn ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let throwOnce = true
|
||||
ctx.on('agent/pre-step', () => {
|
||||
if (throwOnce) { throwOnce = false; throw new Error('boom in pre-step') }
|
||||
})
|
||||
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
|
||||
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
// The first turn failed at step 1 (no model call happened), surfaced via
|
||||
// agent/error, with the durable failure on turn/end.reason.
|
||||
expect(errors).toHaveLength(1)
|
||||
expect(errors[0]!.message).toContain('boom in pre-step')
|
||||
expect(adapter.requests.length).toBe(0)
|
||||
const firstTurnEnd = agent.session.events.find(e => e.type === 'turn/end')
|
||||
expect(firstTurnEnd?.type === 'turn/end' && firstTurnEnd.data.reason).toMatchObject({ kind: 'error', step: 1 })
|
||||
// The step opened-and-closed count stays balanced even though it never ran.
|
||||
const types = agent.session.events.map(e => e.type)
|
||||
expect(types.filter(t => t === 'step/start').length).toBe(types.filter(t => t === 'step/end').length)
|
||||
|
||||
// The loop survived: a second prompt runs a normal completed turn.
|
||||
send(agent, 'second')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(adapter.requests.length).toBe(1)
|
||||
const lastTurnEnd = agent.session.events.findLast(e => e.type === 'turn/end')
|
||||
expect(lastTurnEnd?.type === 'turn/end' && lastTurnEnd.data.reason).toEqual({ kind: 'completed' })
|
||||
})
|
||||
|
||||
it('cancel() mid-stream ends the turn with reason aborted', async () => {
|
||||
|
||||
Reference in New Issue
Block a user