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

# Conflicts:
#	docs/event-producer-consumer.md

packages/core/agent-loop/tests/loop.spec.ts

@@ -230,9 +230,8 @@ describe('agent loop', () => {
       assembly.variables['model'] = 'mock'
       return next()
     })
-    ctx.on('agent/request', async (_agent, _turn, _step, options, next) => {
-      options.model = 'mock'
-      return next()
+    ctx.on('agent/request', async (_agent, _turn, _step, config, _next) => {
+      return { ...config, model: 'mock' }
     })
     const agent = ctx.agentLoop.create(AgentId('a-late-model'), {})
 
@@ -428,20 +427,27 @@ describe('agent loop', () => {
     expect(agent.session.events.some(e => e.type === 'tool/result')).toBe(true)
   })
 
-  it('agent/request waterfall can rewrite the request (model-switch pattern)', async () => {
+  it('agent/request waterfall switches models by returning a replacement config; the switch is logged', async () => {
     const adapter = new MockAdapter([textResponse('ok')])
     const ctx = await harness(adapter)
     ctx.llm.registerAdapter(['other-model'], adapter)
     const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
 
-    ctx.on('agent/request', async (_agent, _turn, _step, options, next) => {
-      options.model = 'other-model'
-      return next()
+    ctx.on('agent/request', async (_agent, _turn, _step, config, _next) => {
+      // The seed is frozen — config is not a mutable per-call knob; a switch
+      // is proposed by returning a replacement, and the loop logs it.
+      expect(Object.isFrozen(config)).toBe(true)
+      expect(() => { (config as { model: string }).model = 'other-model' }).toThrow(TypeError)
+      return { ...config, model: 'other-model' }
     })
 
     send(agent, 'hi')
     await waitForIdle(ctx, agent)
     expect(adapter.requests[0]!.model).toBe('other-model')
+    // The header event records what the request ACTUALLY used — the switch is
+    // a reconstructable fact, not silent drift.
+    const headerEvent = agent.session.events.find(e => e.type === 'request/header')
+    expect(headerEvent?.type === 'request/header' && headerEvent.data.header.config.model).toBe('other-model')
   })
 
   it('agent/pre-step fires once per step before the step is opened', async () => {

packages/core/agent-loop/tests/request-cache.e2e.ts

@@ -0,0 +1,104 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+import LlmService from '@deepseek-ai/dsh-llm'
+import SessionStore from '@deepseek-ai/dsh-session'
+import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
+import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
+import AgentRegistry, { AgentId, type Agent } from '@deepseek-ai/dsh-agent'
+import AgentLoop from '@deepseek-ai/dsh-agent-loop'
+import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
+
+/**
+ * With-key proof that log-derived requests translate into REAL provider cache
+ * hits: a multi-step tool turn (plus a follow-up turn) against the live
+ * DeepSeek API must report `cacheReadTokens > 0` on every request after the
+ * first — the adapter maps the provider's `prompt_cache_hit_tokens`, and the
+ * per-step usage recorded on `assistant/message` events is the production
+ * observable for cache behavior (the reconstructability RFC's measurement
+ * layer: prefix stability is corollary #1). Mocks prove the requests are
+ * append-extensions; only the real API proves those bytes actually hit the
+ * provider cache. Key-gated — skips entirely without $DEEPSEEK_API_KEY.
+ */
+
+// Long enough that the shared request prefix comfortably spans the provider's
+// cache-block granularity (64 tokens) from the very first request.
+const SYSTEM = 'You are a terse coding assistant used in an automated cache test. '
+  + 'Always follow instructions literally and exactly. When the user asks you to look '
+  + 'something up, call the lookup tool with the requested key and wait for its result '
+  + 'before answering. Never invent a value the tool has not returned. After the tool '
+  + 'returns, answer with a single short sentence that repeats the returned value '
+  + 'verbatim. Do not add explanations, do not use markdown, do not ask follow-up '
+  + 'questions. If the user asks anything else, answer in one short sentence.'
+
+let ctx: Context | undefined
+
+afterEach(async () => {
+  await ctx?.fiber.dispose()
+  ctx = undefined
+})
+
+async function loopHarness(): Promise<Context> {
+  const created = new Context()
+  await created.plugin(LlmService)
+  await created.plugin(SessionStore)
+  await created.plugin(SystemPrompt, { persona: SYSTEM })
+  await created.plugin(ToolRegistry)
+  await created.plugin(AgentRegistry)
+  await created.plugin(AgentLoop, { agents: [] })
+  await created.plugin(LlmDeepSeek, { models: ['deepseek-v4-flash'] })
+  created.tools.register(defineTool({
+    name: 'lookup',
+    description: 'Look up the stored value for a key.',
+    parameters: { key: { type: 'string', description: 'The key to look up.' } },
+    async execute(args) {
+      return [{ type: 'text', text: `value(${String(args.key)}) = azure-falcon-42` }]
+    },
+  }))
+  return created
+}
+
+function waitForIdle(context: Context, agent: Agent): Promise<void> {
+  return new Promise((resolve) => {
+    const dispose = context.on('agent/status', (subject, status) => {
+      if (subject === agent && status === 'idle') {
+        dispose()
+        resolve()
+      }
+    })
+  })
+}
+
+describe.skipIf(!process.env.DEEPSEEK_API_KEY)('log-derived request cache hits (real API)', () => {
+  it('every request after the first hits the provider prefix cache', async () => {
+    ctx = await loopHarness()
+    const agent = ctx.agentLoop.create(AgentId('cache-e2e'), { model: 'deepseek-v4-flash' })
+
+    // Turn 1: forces a tool call → at least two steps (two model requests).
+    agent.send([{ type: 'text', text: 'Look up the key "deploy-color" with the lookup tool and tell me the value.' }])
+    await waitForIdle(ctx, agent)
+    // Turn 2: a follow-up over the same (longer) prefix.
+    agent.send([{ type: 'text', text: 'Thanks. Repeat that value one more time.' }])
+    await waitForIdle(ctx, agent)
+
+    const usages = [...agent.session.events]
+      .filter(e => e.type === 'assistant/message')
+      .map(e => e.data.usage)
+    expect(usages.length).toBeGreaterThanOrEqual(3) // 2 steps in turn 1 + ≥1 in turn 2
+    for (const usage of usages) expect(usage).toBeDefined()
+
+    // The first request has nothing to hit; every later one shares its
+    // predecessor as a byte-identical prefix, so the provider must report
+    // cached prompt tokens (prompt_cache_hit_tokens → cacheReadTokens).
+    for (const usage of usages.slice(1)) {
+      expect(usage!.cacheReadTokens ?? 0).toBeGreaterThan(0)
+    }
+
+    // World-verification of the conversation itself: the tool value made it
+    // through the loop into the final answer.
+    const finalText = agent.session.deriveMessages().at(-1)!.content
+      .filter(block => block.type === 'text')
+      .map(block => block.text)
+      .join('')
+    expect(finalText).toContain('azure-falcon-42')
+  }, 180_000)
+})

packages/core/agent-loop/tests/request-log.spec.ts

@@ -0,0 +1,85 @@
+/**
+ * recordRequestHeader unit tests: exactly one of four things per request —
+ * an 'initial' snapshot (log has no header yet), a 'resume' snapshot (fresh
+ * loop instance over a log that has one), nothing (header unchanged), a
+ * round-tripping delta, or a 'fallback' snapshot when the delta encoding
+ * cannot express the change (pure tool reordering).
+ */
+
+import { describe, expect, it } from 'vitest'
+import { Session, SessionId, canonicalHeader } from '@deepseek-ai/dsh-session'
+import type { SessionEvent } from '@deepseek-ai/dsh-session'
+import type { ToolSchema } from '@deepseek-ai/dsh-llm'
+import { createTransmissionLog, recordRequestHeader } from '../src/request-log.ts'
+
+function tool(name: string, description = 'd'): ToolSchema {
+  return { name, description, parameters: { type: 'object' } }
+}
+
+function openSession(id: string): Session {
+  const session = new Session(SessionId(id))
+  session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+  return session
+}
+
+function headerEvents(session: Session): SessionEvent[] {
+  return session.events.filter(e => e.type === 'request/header' || e.type === 'request/header-delta')
+}
+
+describe('recordRequestHeader', () => {
+  it("anchors a new conversation with an 'initial' snapshot, then logs nothing while unchanged", () => {
+    const session = openSession('rl-initial')
+    const state = createTransmissionLog()
+    const header = canonicalHeader({ config: { model: 'm' }, system: 's', tools: [tool('t')] })
+
+    recordRequestHeader(session, state, header)
+    const [first] = headerEvents(session)
+    expect(first?.type === 'request/header' && first.data.reason).toBe('initial')
+
+    recordRequestHeader(session, state, header)
+    expect(headerEvents(session)).toHaveLength(1)
+  })
+
+  it("anchors a fresh loop instance over an anchored log with a 'resume' snapshot, even unchanged", () => {
+    const session = openSession('rl-resume')
+    const header = canonicalHeader({ config: { model: 'm' }, system: 's' })
+    recordRequestHeader(session, createTransmissionLog(), header)
+
+    // A second instance (process restart / fork): the boundary itself is a
+    // recorded fact — snapshot appended even though the header is identical.
+    recordRequestHeader(session, createTransmissionLog(), header)
+    const events = headerEvents(session)
+    expect(events).toHaveLength(2)
+    expect(events[1]?.type === 'request/header' && events[1].data.reason).toBe('resume')
+  })
+
+  it('logs a round-tripping delta for a mid-run change, and the fold reproduces the header', () => {
+    const session = openSession('rl-delta')
+    const state = createTransmissionLog()
+    const first = canonicalHeader({ config: { model: 'm' }, system: 'a\nb', tools: [tool('t')] })
+    recordRequestHeader(session, state, first)
+
+    const second = canonicalHeader({ config: { model: 'm' }, system: 'a\nc', tools: [tool('t'), tool('u')] })
+    recordRequestHeader(session, state, second)
+    const events = headerEvents(session)
+    expect(events).toHaveLength(2)
+    expect(events[1]?.type).toBe('request/header-delta')
+    expect(session.requestHeader()).toEqual(second)
+  })
+
+  it("records a change the delta cannot express (pure reordering) as a 'fallback' snapshot", () => {
+    const session = openSession('rl-fallback')
+    const state = createTransmissionLog()
+    const first = canonicalHeader({ config: { model: 'm' }, tools: [tool('a'), tool('b')] })
+    recordRequestHeader(session, state, first)
+
+    const reordered = canonicalHeader({ config: { model: 'm' }, tools: [tool('b'), tool('a')] })
+    recordRequestHeader(session, state, reordered)
+    const events = headerEvents(session)
+    expect(events).toHaveLength(2)
+    expect(events[1]?.type === 'request/header' && events[1].data.reason).toBe('fallback')
+    // The fold still lands on the exact header — deltas are an encoding
+    // optimization, never a correctness dependency.
+    expect(session.requestHeader()).toEqual(reordered)
+  })
+})

packages/core/agent-loop/tests/request-reconstruction.spec.ts

@@ -0,0 +1,313 @@
+/**
+ * Loop-level reconstructability: every request the loop sends is a pure
+ * function of the session log — messages are the derivation at the step/start
+ * boundary, the header is the fold of request/header* events — and every
+ * request is an append-extension of its predecessor unless a logged event
+ * (compaction replace, header change) explains the difference. The requests
+ * recorded by the mock adapter are the observable; the offline-rebuild test
+ * at the bottom is the theorem stated end-to-end.
+ */
+
+import { describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+import LlmService from '@deepseek-ai/dsh-llm'
+import type { GenerateOptions } from '@deepseek-ai/dsh-llm'
+import SessionStore, { Session, SessionId, foldRequestHeader } from '@deepseek-ai/dsh-session'
+import SystemPrompt 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, toolCallResponse } from './mock-adapter.ts'
+
+async function harness(adapter: MockAdapter, persona = 'stable base') {
+  const ctx = new Context()
+  await ctx.plugin(LlmService)
+  await ctx.plugin(SessionStore)
+  await ctx.plugin(SystemPrompt, { persona })
+  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 send(agent: ReactLoopAgent, text: string) {
+  agent.send([{ type: 'text', text }])
+}
+
+/** Assert `previous` is a strict value-prefix of `current`. */
+function expectPrefixExtension(previous: GenerateOptions, current: GenerateOptions) {
+  expect(current.messages.length).toBeGreaterThan(previous.messages.length)
+  expect(current.messages.slice(0, previous.messages.length)).toEqual([...previous.messages])
+  expect(current.system).toEqual(previous.system)
+  expect(current.tools).toEqual(previous.tools)
+}
+
+function registerEcho(ctx: Context) {
+  ctx.tools.register(defineTool({
+    name: 'echo',
+    description: 'echo back',
+    parameters: { text: { type: 'string' } },
+    async execute(args) {
+      return [{ type: 'text', text: `echo: ${String(args.text)}` }]
+    },
+  }))
+}
+
+describe('request stability across the loop', () => {
+  it('each step request within a turn append-extends the previous, frozen end to end', async () => {
+    const adapter = new MockAdapter([
+      toolCallResponse('c1', 'echo', { text: 'one' }, 'first'),
+      toolCallResponse('c2', 'echo', { text: 'two' }, 'second'),
+      textResponse('done'),
+    ])
+    const ctx = await harness(adapter)
+    registerEcho(ctx)
+    const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    expect(adapter.requests).toHaveLength(3)
+    expectPrefixExtension(adapter.requests[0]!, adapter.requests[1]!)
+    expectPrefixExtension(adapter.requests[1]!, adapter.requests[2]!)
+    for (const request of adapter.requests) {
+      expect(Object.isFrozen(request)).toBe(true)
+      expect(Object.isFrozen(request.messages)).toBe(true)
+    }
+    // One anchoring header snapshot; no further header events (nothing changed).
+    const headerEvents = agent.session.events.filter(e => e.type === 'request/header' || e.type === 'request/header-delta')
+    expect(headerEvents).toHaveLength(1)
+    expect(headerEvents[0]?.type === 'request/header' && headerEvents[0].data.reason).toBe('initial')
+  })
+
+  it('a later turn append-extends the previous turn (one conversation, one log)', async () => {
+    const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
+
+    send(agent, 'first')
+    await waitForIdle(ctx, agent)
+    send(agent, 'second')
+    await waitForIdle(ctx, agent)
+
+    expect(adapter.requests).toHaveLength(2)
+    expectPrefixExtension(adapter.requests[0]!, adapter.requests[1]!)
+  })
+
+  it('a compaction replace rewrites the resend, and the log explains it', async () => {
+    const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
+
+    send(agent, 'first')
+    await waitForIdle(ctx, agent)
+
+    // A pre-step listener compacts turn 1's history before turn 2's step —
+    // the sanctioned surface rewrite, landing OUTSIDE the step.
+    const preStep = ctx.on('agent/pre-step', () => {
+      preStep()
+      const session = agent.session
+      const nodes = session.surface.nodes
+      session.append('context/message', {
+        content: [{ type: 'text', text: '[summary of turn 1]' }],
+        source: { kind: 'plugin', plugin: 'test-compact' },
+      }, {
+        surfaceOp: { op: 'replace', start: nodes[0]!.seq, end: nodes[1]!.seq },
+        sourceEventSeqs: [nodes[0]!.seq, nodes[1]!.seq],
+      })
+    })
+
+    send(agent, 'second')
+    await waitForIdle(ctx, agent)
+
+    const second = adapter.requests[1]!
+    // The rewritten history: summary replaces turn 1's user+assistant pair.
+    expect(second.messages[0]!.content.some(b => b.type === 'text' && b.text.includes('[summary of turn 1]'))).toBe(true)
+    // No header event beyond the anchor: the replace is itself in the log.
+    expect(agent.session.events.filter(e => e.type === 'request/header')).toHaveLength(1)
+  })
+
+  it('a real system-prompt change is a logged header delta; a stable prompt logs nothing', async () => {
+    const adapter = new MockAdapter([textResponse('one'), textResponse('two'), textResponse('three')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
+
+    send(agent, 'first')
+    await waitForIdle(ctx, agent)
+    send(agent, 'second')
+    await waitForIdle(ctx, agent)
+    // Identical assembly re-rendered per step is NOT a change.
+    expect(agent.session.events.filter(e => e.type === 'request/header-delta')).toHaveLength(0)
+
+    ctx.systemPrompt.section({ name: 'extra', order: 2, text: 'new guidance' })
+    send(agent, 'third')
+    await waitForIdle(ctx, agent)
+
+    const deltas = agent.session.events.filter(e => e.type === 'request/header-delta')
+    expect(deltas).toHaveLength(1)
+    expect(adapter.requests[2]!.system).toContain('new guidance')
+    // History is preserved across the change — only the header moved.
+    expect(adapter.requests[2]!.messages.length).toBeGreaterThan(adapter.requests[1]!.messages.length)
+  })
+
+  it('an inject() during the agent/request waterfall joins the NEXT request (the step/start boundary)', async () => {
+    const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
+
+    let injected = false
+    ctx.on('agent/request', async (_agent, _turn, _step, _config, next) => {
+      if (!injected) {
+        injected = true
+        agent.inject([{ type: 'text', text: '[late context]' }], { source: { kind: 'plugin', plugin: 'test' } })
+      }
+      return next()
+    })
+
+    send(agent, 'first')
+    await waitForIdle(ctx, agent)
+    const first = adapter.requests[0]!
+    // The inject landed in the log after the boundary: not in THIS request…
+    expect(first.messages.some(m => m.content.some(b => b.type === 'text' && b.text.includes('[late context]')))).toBe(false)
+    expect(agent.session.events.some(e => e.type === 'context/message')).toBe(true)
+
+    send(agent, 'second')
+    await waitForIdle(ctx, agent)
+    // …but in the next one, at its logged position.
+    const second = adapter.requests[1]!
+    expect(second.messages.some(m => m.content.some(b => b.type === 'text' && b.text.includes('[late context]')))).toBe(true)
+  })
+
+  it('a mutation attempt on the frozen request content throws into the step (loud, not silent)', async () => {
+    const adapter = new MockAdapter([textResponse('one')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
+
+    const errors: Error[] = []
+    ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
+    ctx.on('llm/stream', (options, next) => {
+      // The historical failure mode this design kills: a listener rewriting
+      // request content in place. The freeze turns it into a loud error.
+      options.messages.push({ role: 'user', content: [{ type: 'text', text: 'sneaky' }] })
+      return next()
+    })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    expect(errors).toHaveLength(1)
+    expect(errors[0]!.message).toMatch(/not extensible|frozen|read only|readonly/i)
+  })
+
+  it('a fresh loop instance over a seeded log anchors with a resume snapshot and stays cache-aligned', async () => {
+    const adapter = new MockAdapter([textResponse('one')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(AgentId('gen1'), { model: 'mock' })
+    send(agent, 'first')
+    await waitForIdle(ctx, agent)
+
+    // Second generation: a new agent whose session is seeded with the first
+    // one's full log (the resume/fork path).
+    const adapter2 = new MockAdapter([textResponse('two')])
+    const ctx2 = await harness(adapter2)
+    const handle = ctx2.agents.create({
+      agentId: AgentId('gen2'),
+      sessionId: SessionId('gen2-session'),
+      seed: [...agent.session.events],
+      agentOptions: { model: 'mock' },
+    })
+    const agent2 = handle.agent as ReactLoopAgent
+    send(agent2, 'second')
+    await waitForIdle(ctx2, agent2)
+
+    const snapshots = agent2.session.events.filter(e => e.type === 'request/header')
+    expect(snapshots).toHaveLength(2)
+    expect(snapshots[1]?.type === 'request/header' && snapshots[1].data.reason).toBe('resume')
+    // Identical header across the restart: byte-identical continuation.
+    expect(adapter2.requests[0]!.system).toEqual(adapter.requests[0]!.system)
+    expectPrefixExtension(adapter.requests[0]!, adapter2.requests[0]!)
+  })
+
+  it('a delegating listener cannot mutate the seed through next() — the fold stays log-true', async () => {
+    const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
+
+    ctx.on('agent/request', async (_agent, _turn, _step, _config, next) => {
+      const config = await next()
+      // next() resolves the SAME frozen seed — in-place shaping after
+      // delegation is unrepresentable, so a "mutate what next() returned"
+      // listener cannot desync the log from the request (nor reach the
+      // session's cached header fold, which is deep-cloned away and itself
+      // frozen).
+      expect(Object.isFrozen(config)).toBe(true)
+      expect(() => { (config as { temperature?: number }).temperature = 0.9 }).toThrow(TypeError)
+      return config
+    })
+
+    send(agent, 'first')
+    await waitForIdle(ctx, agent)
+    send(agent, 'second')
+    await waitForIdle(ctx, agent)
+
+    // No delta was logged (nothing really changed), and the session's own
+    // fold is immutable state.
+    expect(agent.session.events.filter(e => e.type === 'request/header-delta')).toHaveLength(0)
+    expect(Object.isFrozen(agent.session.requestHeader())).toBe(true)
+    expect(adapter.requests[1]!.temperature).toBeUndefined()
+  })
+
+  it('THEOREM: every request rebuilds byte-equal from the session log alone', async () => {
+    const adapter = new MockAdapter([
+      toolCallResponse('c1', 'echo', { text: 'one' }, 'calling'),
+      textResponse('done'),
+      textResponse('after change'),
+    ])
+    const ctx = await harness(adapter)
+    registerEcho(ctx)
+    const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+    ctx.systemPrompt.section({ name: 'extra', order: 2, text: 'now with guidance' })
+    ctx.on('agent/request', async (_agent, _turn, _step, config, _next) => ({ ...config, temperature: 0.5, maxTokens: 99, stop: ['<END>'] }))
+    send(agent, 'again')
+    await waitForIdle(ctx, agent)
+
+    expect(adapter.requests).toHaveLength(3)
+    const events = agent.session.events
+    const stepStarts = events.filter(e => e.type === 'step/start')
+    expect(stepStarts).toHaveLength(3)
+
+    adapter.requests.forEach((request, index) => {
+      const stepStart = stepStarts[index]!
+      // Messages: the derivation over the log prefix strictly before this
+      // step's step/start — rebuilt here through a completely fresh Session.
+      const rebuilt = new Session(SessionId(`rebuild-${index}`), structuredClone(events.slice(0, stepStart.seq)))
+      expect(structuredClone(request.messages)).toEqual(rebuilt.deriveMessages())
+
+      // Header: the fold of request/header* events up to this step's dispatch
+      // (its header event sits between step/start and the first chunk).
+      const firstChunk = events.find(e => e.type === 'assistant/chunk' && e.seq > stepStart.seq)!
+      const header = foldRequestHeader(events.slice(0, firstChunk.seq))!
+      expect(request.model).toBe(header.config.model)
+      expect(request.system).toEqual(header.system)
+      expect(structuredClone(request.tools ?? [])).toEqual(structuredClone(header.tools ?? []))
+      expect(request.temperature).toBe(header.config.temperature)
+      expect(request.maxTokens).toBe(header.config.maxTokens)
+      expect(request.stop).toEqual(header.config.stop)
+    })
+  })
+})

packages/core/agent-loop/tests/review-fixes.spec.ts

@@ -399,9 +399,8 @@ describe('MEDIUM: misc registry and config fixes', () => {
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(AgentId('a1'), {}) // no model — router plugin decides
 
-    ctx.on('agent/request', async (_agent, _turn, _step, options, next) => {
-      options.model = 'mock'
-      return next()
+    ctx.on('agent/request', async (_agent, _turn, _step, config, _next) => {
+      return { ...config, model: 'mock' }
     })
 
     send(agent, 'go')