refactor(compact): turn-agnostic retention + dedicated agent/pre-request seam
Reform the compaction blueprint so a runaway turn survives and the design
stops drifting across review rounds:
- Drop in-flight-turn protection ("layer 2"). Retention is a uniform tail→head
whole-unit walk; the only structural guard is step-alignment. A single turn
that alone exceeds the window now compacts its own early closed steps instead
of being retained verbatim (the failure mode that motivated this).
- Move auto-compaction off the agent/request waterfall onto a new awaited
agent/pre-request loop seam, fired before history derivation. Compaction
mutates the surface; the loop derives once from the result — no double-derive,
and a listener structurally cannot act on not-yet-derived messages.
- Tighten compactIfNeeded to required (session, system, model, signal).
- Enforce a single-pass convergence invariant in resolveConfig: reject configs
where summarizationMaxTokens + retainTokens exceeds the threshold, so a
compaction can never immediately re-trigger.
- Document the crash vs recoverable failure taxonomy; core session repair stays
compaction-agnostic (a log-only orphaned compact/start is inert).
- Wire dsh-compact-basic into examples/coding-agent and add a with-key
compaction e2e (compaction's first real-world exercise + runaway net).
- Rewrite the RFC to encode the blueprint and move it to implemented/.
The runaway-turn snapshot is a named deferred follow-up: dsh-llm-replay cannot
yet serve the interleaved summarization model call.
This commit is contained in:
@@ -149,8 +149,9 @@ export interface LoopHandle {
|
||||
* 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
|
||||
* req = {model, system, tools, messages: session.deriveMessages(), signal}
|
||||
* req = waterfall agent/request ⟵ hooks/compaction/model-switch
|
||||
* req = waterfall agent/request ⟵ hooks/model-switch
|
||||
* stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks)
|
||||
* session('assistant/chunk'); emit agent/stream-chunk
|
||||
* msg = waterfall agent/step-result ⟵ BEFORE the log append, so the
|
||||
@@ -565,6 +566,13 @@ async function runStep(
|
||||
.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(),
|
||||
|
||||
@@ -320,6 +320,65 @@ 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 () => {
|
||||
// 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).
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'echo', {}, 'calling echo'),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'echo', description: 'echo', parameters: {},
|
||||
async execute() { return [{ type: 'text', text: 'echoed' }] },
|
||||
}))
|
||||
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) => {
|
||||
if (subject === agent) fires.push({ turn, step, model })
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// One fire per step, in order, each with the agent's model.
|
||||
expect(fires).toEqual([
|
||||
{ turn: 1, step: 1, model: 'mock' },
|
||||
{ turn: 1, step: 2, model: 'mock' },
|
||||
])
|
||||
})
|
||||
|
||||
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.
|
||||
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) => {
|
||||
if (subject === agent && !injected) {
|
||||
injected = true
|
||||
subject.session.append('context/message', {
|
||||
content: [{ type: 'text', text: 'INJECTED-IN-PRE-REQUEST' }],
|
||||
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.
|
||||
const text = JSON.stringify(adapter.requests[0]!.messages)
|
||||
expect(text).toContain('INJECTED-IN-PRE-REQUEST')
|
||||
})
|
||||
|
||||
it('cancel() mid-stream ends the turn with reason aborted', async () => {
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
|
||||
@@ -180,10 +180,32 @@ declare module 'cordis' {
|
||||
'agent/step-end'(agent: Agent, turn: number, step: number): void
|
||||
|
||||
// ---- interception seams (waterfall) ----
|
||||
/**
|
||||
* Awaited surface-mutation checkpoint, fired BEFORE the step's message
|
||||
* history is derived (and thus before {@link agent/request}). The loop
|
||||
* awaits `ctx.parallel('agent/pre-request', …)` after assembling the system
|
||||
* prompt but before `session.deriveMessages()`, then derives ONCE from
|
||||
* whatever the surface now holds. This is where compaction belongs: it
|
||||
* mutates the session surface in place (shadowing an older range with a
|
||||
* summary node), and the single subsequent derive reflects the mutation —
|
||||
* so there is no double-derive and no listener can see (or be expected to
|
||||
* act on) an assembled `messages` array that does not exist yet.
|
||||
*
|
||||
* Awaited (parallel), not a waterfall: a listener mutates the surface as a
|
||||
* side effect; there is nothing to transform or veto, but the loop must wait
|
||||
* for the mutation to complete before deriving. `system`/`model` are the
|
||||
* assembled values a listener needs to measure pressure (system counts
|
||||
* toward the budget) and to summarize (the model). `signal` cancels any
|
||||
* in-flight work a listener starts (e.g. a summarization model call).
|
||||
* @mode parallel
|
||||
*/
|
||||
'agent/pre-request'(agent: Agent, turn: number, step: number, system: string, model: string, signal: AbortSignal): Promise<void> | void
|
||||
/**
|
||||
* Waterfall: mutate the fully-assembled {@link GenerateOptions} before the
|
||||
* model call (hooks, compaction, model switching, tool filtering, …). Call
|
||||
* `next()` to delegate, or return without it to short-circuit.
|
||||
* model call (hooks, model switching, tool filtering, …). Call `next()` to
|
||||
* delegate, or return without it to short-circuit. For surface mutation that
|
||||
* must precede history derivation (compaction), use {@link agent/pre-request}
|
||||
* instead — by the time this fires, `options.messages` is already derived.
|
||||
* @mode waterfall
|
||||
*/
|
||||
'agent/request'(agent: Agent, turn: number, step: number, options: GenerateOptions, next: () => Promise<GenerateOptions>): Promise<GenerateOptions>
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import type { SessionEvent, SurfaceEvent, SurfaceEventType } from '@deepseek-ai/dsh-session'
|
||||
import { Session, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import { Session, SessionId, isSurfaceEvent } from '@deepseek-ai/dsh-session'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm'
|
||||
|
||||
/** Build a minimal session with turn boundaries and a single user message. */
|
||||
@@ -278,4 +278,21 @@ describe('Session.append surface opts', () => {
|
||||
// The string 'append' is a primitive — identity-preserving is fine.
|
||||
expect(event.surfaceOp).toBe('append')
|
||||
})
|
||||
|
||||
it('isSurfaceEvent rejects a surface-eligible type missing its surfaceOp marker', () => {
|
||||
// A raw event (not built via append, which mandates the marker) of a
|
||||
// surface-eligible type but with no surfaceOp must NOT narrow to a
|
||||
// SurfaceEvent — it would otherwise be silently dropped from the surface.
|
||||
const noMarker: SessionEvent = {
|
||||
type: 'user/message', seq: 0, time: 1,
|
||||
data: { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } },
|
||||
}
|
||||
expect(isSurfaceEvent(noMarker)).toBe(false)
|
||||
// A non-surface type is rejected too (the type gate).
|
||||
const boundary: SessionEvent = { type: 'turn/start', seq: 1, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }
|
||||
expect(isSurfaceEvent(boundary)).toBe(false)
|
||||
// A properly-marked surface event narrows.
|
||||
const marked = { ...noMarker, surfaceOp: 'append' } as SurfaceEvent
|
||||
expect(isSurfaceEvent(marked)).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user