From 217b8ec0e2d0413b5581cec5fe7ed9efb764e90e Mon Sep 17 00:00:00 2001
From: Tianyi Cui <53024+tianyicui@users.noreply.github.com>
Date: Thu, 11 Jun 2026 12:18:52 +0800
Subject: [PATCH] Fix architecture-review findings in the loop and service
 packages

High (loop pipeline): agent/step-result now runs before the
assistant/message append so the session log records what tool dispatch
actually uses; abort is honored between tool calls, not just
mid-stream; steering drains at step start, pending steering overrides
a negative turn-continuation decision (/goal pattern), and leftover
steering is re-enqueued as queued messages so it is never stranded;
exceptions from turn-continuation listeners and session/flush are
contained to the turn (error event + agent/error) instead of killing
the driver loop.

Medium: disposal emits agent/status('disposed') and mid-turn disposal
records reason 'disposed'; duplicate LLM adapter registration throws
(all-or-nothing); SessionEvent is a real discriminated union (casts
removed); model-less agents fail with a clear actionable error unless
agent/request supplies a model.

Low: agent/queued and agent/steering carry the resolved MessageSource;
streamBlocks() yields strictly in stream order and flushes delta-only
blocks (matches generate()); BlockAssembler freezes blocks on
block-end and ignores stragglers from malformed streams; turn
numbering is a counter seeded from the log (fork-safe); LoopAgent's
stop disposer is infallible (a throwing status listener cannot skip
registry cleanup); AgentLoop.create uses a generator effect so stop
and unregister are independent disposables; SessionStore wires
onAppend inside its effect.

21 regression tests added (review-fixes.spec.ts), organized by
finding. Docs updated: loop pseudocode (status emissions, ordering,
error containment, steering guarantees) and waterfall composition
caveat in docs/architecture.md; AGENTS.md notes that excessive tests
are welcome.
---
 .gitignore                                    |   1 +
 AGENTS.md                                     |   5 +-
 docs/architecture.md                          |  31 +-
 packages/agent-loop/src/agent.ts              |  29 +-
 packages/agent-loop/src/index.ts              |  14 +-
 packages/agent-loop/src/loop.ts               | 249 +++++---
 .../agent-loop/tests/review-fixes.spec.ts     | 547 ++++++++++++++++++
 packages/agent/src/types.ts                   |   9 +-
 packages/llm/src/assembler.ts                 | 109 +++-
 packages/llm/src/index.ts                     |  16 +-
 packages/session/src/index.ts                 |  20 +-
 packages/session/src/types.ts                 |  25 +-
 12 files changed, 887 insertions(+), 168 deletions(-)
 create mode 100644 packages/agent-loop/tests/review-fixes.spec.ts

diff --git a/.gitignore b/.gitignore
index 09538dc27c..636393047d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,3 +5,4 @@ lib/
 .yarn/
 yarn-error.log
 examples/*/*.jsonl
+.claude/
diff --git a/AGENTS.md b/AGENTS.md
index 25c3902ea7..50e65bea21 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -73,7 +73,10 @@ only needed for publishing/consumption outside the repo.
   docs/architecture.md). Changing `agent-loop` requires updating that doc.
 - **Tests**: vitest, colocated under `packages/<name>/tests/*.spec.ts`. Every
   registry needs an HMR-safety test (dispose the contributing fiber, assert
-  cleanup).
+  cleanup). **Excessive tests are welcome** — when in doubt, write the test;
+  err on the side of covering edge cases, error paths, event ordering, and
+  concurrency races even if they seem unlikely. Review findings get regression
+  tests (see `packages/agent-loop/tests/review-fixes.spec.ts`).
 
 ## Vendoring Policy
 
diff --git a/docs/architecture.md b/docs/architecture.md
index fe64413633..e35ee710cc 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -154,28 +154,40 @@ deferred.
 ```
 forever:
   wait for queued messages (idle)
-  TURN: drain queued → session('user/message'…) → 'turn/start' → emit agent/turn-start
+  emit agent/status(running)
+  TURN (error-contained — a throwing plugin ends the turn, never the loop):
+    drain queued → session('user/message'…) → 'turn/start' → emit agent/turn-start
     STEP loop:
+      drain steering (late steering from previous step's listeners)
       emit agent/step-start
       assembly = ctx.systemPrompt.assemble()          ⟵ waterfall system-prompt/assemble
       req = {model, system, tools, messages: session.deriveMessages(), signal}
       req = waterfall agent/request                   ⟵ hooks, compaction, model switch
       stream ctx.llm.stream(req)                      ⟵ waterfall llm/stream (raw chunks)
         session('assistant/chunk'); emit agent/stream-chunk
-      session('assistant/message', 'usage')
-      msg = waterfall agent/step-result               ⟵ post-process before tool dispatch
-      each tool-call (sequential):
+      msg = waterfall agent/step-result               ⟵ runs BEFORE the log append, so the
+      session('assistant/message', 'usage')              log records what tool dispatch uses
+      each tool-call (sequential, abort-checked between calls):
         session('tool/call'); ctx.tools.execute()     ⟵ waterfall tools/execute
         session('tool/result')
       drain steering → session('steering/message'); emit agent/steering
       emit agent/step-end
       cont = waterfall agent/turn-continuation(default = hadToolCalls || steered)
+      steering pending from step-end/continuation listeners forces cont = true
       if !cont: break
   session('turn/end'); emit agent/turn-end
-  await ctx.parallel('session/flush', session)        ⟵ durability checkpoint
-  idle (emit agent/status) unless more queued
+  await ctx.parallel('session/flush', session)        ⟵ durability checkpoint (failure
+                                                         reported via agent/error, not fatal)
+  leftover steering re-enqueued as queued messages    ⟵ steering is never stranded
+  emit agent/status(idle) unless more queued
 ```
 
+Error containment: a throwing `agent/turn-continuation` listener or a
+rejecting `session/flush` ends the **turn** with an `error` event — never the
+driver loop. `abort()` is honored mid-stream **and** between tool calls;
+disposal mid-turn ends the turn with reason `disposed` and emits
+`agent/status('disposed')`.
+
 ### Event taxonomy
 
 Declared in `@deepseek-ai/dsh-agent` (so nothing depends on the loop package).
@@ -206,8 +218,11 @@ receives `(...args, next)`:
 - return a value **without** calling `next()` to short-circuit (veto);
 - listeners run in registration order; `prepend: true` jumps the queue.
 
-Mutate the passed-in object (e.g. `options.model = '…'`) or return a
-replacement value — both compose.
+Composition caveat: values propagate through `next()`'s **return value**.
+Mutating the passed-in object works when later listeners receive the same
+reference, but a listener that returns a *new* object makes earlier mutations
+invisible downstream. Prefer mutate-then-`next()` for cooperative middleware;
+return a replacement only when you mean to take over the result.
 
 ## Plugin sanity checklist
 
diff --git a/packages/agent-loop/src/agent.ts b/packages/agent-loop/src/agent.ts
index 79e269fc4c..f109f2cc61 100644
--- a/packages/agent-loop/src/agent.ts
+++ b/packages/agent-loop/src/agent.ts
@@ -1,7 +1,7 @@
 import type { Context } from 'cordis'
 import type { AgentOptions, AgentStatus, SendOptions } from '@deepseek-ai/dsh-agent'
 import type { Agent } from '@deepseek-ai/dsh-agent'
-import type { ContentBlock } from '@deepseek-ai/dsh-llm'
+import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
 import type { Session } from '@deepseek-ai/dsh-session'
 import { Inbox } from './inbox.ts'
 import { runLoop } from './loop.ts'
@@ -44,25 +44,28 @@ export class LoopAgent implements Agent {
     this.ctx.emit('agent/status', this, status)
   }
 
+  private resolveSource(options?: SendOptions): MessageSource {
+    return options?.source ?? { kind: 'user' }
+  }
+
   send(content: ContentBlock[], options?: SendOptions): void {
     if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
-    const source = options?.source ?? { kind: 'user' as const }
+    const source = this.resolveSource(options)
     this.inbox.enqueue({ content, source })
-    this.ctx.emit('agent/queued', this, content, { ...options, steering: false })
+    this.ctx.emit('agent/queued', this, content, { source, steering: false })
   }
 
   steer(content: ContentBlock[], options?: SendOptions): void {
     if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
     if (this._status !== 'running') return this.send(content, options)
-    const source = options?.source ?? { kind: 'user' as const }
+    const source = this.resolveSource(options)
     this.inbox.steer({ content, source })
-    this.ctx.emit('agent/queued', this, content, { ...options, steering: true })
+    this.ctx.emit('agent/queued', this, content, { source, steering: true })
   }
 
   inject(content: ContentBlock[], options?: SendOptions): void {
     if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
-    const source = options?.source ?? { kind: 'user' as const }
-    this.session.append('context/message', { content, source })
+    this.session.append('context/message', { content, source: this.resolveSource(options) })
   }
 
   abort(reason?: string): void {
@@ -77,10 +80,22 @@ export class LoopAgent implements Agent {
       disposed: this.disposed,
       isDisposed: () => this._status === 'disposed',
     })
+    // The disposer must be infallible: it runs inside the fiber's LIFO
+    // disposal chain, where a throw would skip later disposers (e.g. the
+    // registry unregistration) and leave `done` pending forever.
     return () => {
+      if (this._status === 'disposed') return
       this._status = 'disposed'
       this.resolveDisposed()
       this.currentAbort?.abort('disposed')
+      // setStatus refuses transitions out of 'disposed', so emit directly —
+      // 'disposed' is part of the agent/status contract. Guarded: a throwing
+      // listener must not break the disposal chain.
+      try {
+        this.ctx.emit('agent/status', this, 'disposed')
+      } catch {
+        // listener error during disposal — nothing safe left to do with it
+      }
     }
   }
 }
diff --git a/packages/agent-loop/src/index.ts b/packages/agent-loop/src/index.ts
index 1098420725..345380052f 100644
--- a/packages/agent-loop/src/index.ts
+++ b/packages/agent-loop/src/index.ts
@@ -59,14 +59,12 @@ export class AgentLoop extends Service {
   create(id: string, options: AgentOptions = {}): LoopAgent {
     const session = this.ctx.sessions.create(`${id}-session`)
     const agent = new LoopAgent(this.ctx, id, options, session)
-    this.ctx.effect(() => {
-      const stop = agent.start()
-      const unregister = this.ctx.agents.register(agent)
-      return () => {
-        stop()
-        unregister()
-      }
-    }, 'agentLoop.create()')
+    // Generator effect: stop and unregister are independent disposables
+    // (LIFO), so a throwing stop() cannot leak the registry entry.
+    this.ctx.effect(function* (this: AgentLoop) {
+      yield this.ctx.agents.register(agent)
+      yield agent.start()
+    }.bind(this), 'agentLoop.create()')
     return agent
   }
 }
diff --git a/packages/agent-loop/src/loop.ts b/packages/agent-loop/src/loop.ts
index 1f045e21b1..ebcc92f15d 100644
--- a/packages/agent-loop/src/loop.ts
+++ b/packages/agent-loop/src/loop.ts
@@ -1,7 +1,7 @@
 import type { Context } from 'cordis'
 import type { GenerateOptions, Message } from '@deepseek-ai/dsh-llm'
 import { BlockAssembler } from '@deepseek-ai/dsh-llm'
-import type { TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
+import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
 import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
 import type {} from '@deepseek-ai/dsh-tools'
 import type { LoopAgent } from './agent.ts'
@@ -20,110 +20,178 @@ export interface LoopHandle {
  * ```
  * forever:
  *   wait for queued messages (idle)
- *   TURN: drain queued → session('turn/start') → emit agent/turn-start
+ *   TURN (error-contained — a throwing plugin ends the turn, never the loop):
+ *     drain queued → session('user/message'…) → 'turn/start' → emit agent/turn-start
  *     STEP loop:
+ *       drain steering → session('steering/message')  ⟵ catches late steering
  *       emit agent/step-start
- *       assembly = ctx.systemPrompt.assemble()          ⟵ waterfall system-prompt/assemble
+ *       assembly = ctx.systemPrompt.assemble()        ⟵ waterfall system-prompt/assemble
  *       req = {model, system, tools, messages: session.deriveMessages(), signal}
- *       req = waterfall agent/request                   ⟵ hooks/compaction/model-switch
- *       stream ctx.llm.stream(req)                      ⟵ waterfall llm/stream (raw chunks)
- *         session('assistant/chunk'); emit agent/stream-chunk; assembler.push
- *       session('assistant/message','usage')
- *       msg = waterfall agent/step-result               ⟵ post-process before tool dispatch
- *       each tool-call (sequential):
- *         session('tool/call'); ctx.tools.execute()     ⟵ waterfall tools/execute
+ *       req = waterfall agent/request                 ⟵ hooks/compaction/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
+ *       session('assistant/message','usage')             session records what actually ran
+ *       each tool-call in msg (sequential, abort-checked):
+ *         session('tool/call'); ctx.tools.execute()   ⟵ waterfall tools/execute
  *         session('tool/result')
  *       drain steering → session('steering/message'); emit agent/steering
  *       emit agent/step-end
  *       cont = waterfall agent/turn-continuation(default = hadToolCalls || steered)
+ *       if !cont && steering arrived from step-end/continuation listeners: cont = true
+ *       if !cont: break
  *     session('turn/end'); emit agent/turn-end
- *     await ctx.parallel('session/flush', session)      ⟵ durability checkpoint
+ *     await ctx.parallel('session/flush', session)    ⟵ durability checkpoint
+ *     re-enqueue leftover steering as queued          ⟵ steering is never stranded
+ *   idle (emit agent/status) unless more queued
  * ```
  */
 export async function runLoop(ctx: Context, agent: LoopAgent, handle: LoopHandle): Promise<void> {
   const { session } = agent
+  let turn = lastTurnNumber(session) // seeded/forked sessions continue numbering
 
   while (!handle.isDisposed()) {
     await agent.inbox.waitForQueued(handle.disposed)
     if (handle.isDisposed()) break
 
     handle.setStatus('running')
-    const turn = nextTurnNumber(session)
-
-    // Drain queued messages into the session — they trigger this turn.
-    const queued = agent.inbox.drainQueued()
-    const trigger: TurnTrigger = { kind: 'message', source: queued[0]!.source }
-    for (const message of queued) {
-      session.append('user/message', { content: message.content, source: message.source })
-    }
-
-    session.append('turn/start', { turn, trigger })
-    ctx.emit('agent/turn-start', agent, turn)
-
-    let reason: TurnEndReason = { kind: 'completed' }
-    let step = 0
-
-    while (true) {
-      step += 1
-      ctx.emit('agent/step-start', agent, turn, step)
-      session.append('step/start', { turn, step })
-
-      const abort = new AbortController()
-      handle.setAbort(abort)
-
-      let stepOutcome: { hadToolCalls: boolean } | { error: Error }
+    turn += 1
+    try {
+      await runTurn(ctx, agent, handle, turn)
+    } catch (error: any) {
+      // Backstop: a throwing emit listener (turn boundaries) or a broken
+      // finalizer must not kill the driver. Record what we can and move on.
       try {
-        stepOutcome = await runStep(ctx, agent, turn, step, abort.signal)
-      } catch (error: any) {
-        stepOutcome = { error: error instanceof Error ? error : new Error(String(error)) }
-      } finally {
-        handle.setAbort(undefined)
-      }
-
-      // Steering arrives between steps: drain before deciding continuation
-      // so the decision (and the next request) sees it.
-      const steered = agent.inbox.drainSteering()
-      for (const message of steered) {
-        session.append('steering/message', { turn, content: message.content, source: message.source })
-        ctx.emit('agent/steering', agent, turn, message.content)
-      }
-
-      session.append('step/end', { turn, step })
-      ctx.emit('agent/step-end', agent, turn, step)
-
-      if ('error' in stepOutcome) {
-        const { error } = stepOutcome
-        if (abort.signal.aborted || handle.isDisposed()) {
-          reason = { kind: 'aborted', reason: String(abort.signal.reason ?? 'aborted') }
-        } else {
-          session.append('error', { turn, step, message: error.message, code: (error as any).code })
-          ctx.emit('agent/error', agent, turn, step, error)
-          reason = { kind: 'error', message: error.message, code: (error as any).code }
-        }
-        break
-      }
-
-      const defaultDecision = stepOutcome.hadToolCalls || steered.length > 0
-      const shouldContinue = await ctx.waterfall(
-        'agent/turn-continuation', agent, turn, defaultDecision,
-        async () => defaultDecision,
-      )
-      if (!shouldContinue || handle.isDisposed()) break
+        const err = error instanceof Error ? error : new Error(String(error))
+        session.append('error', { turn, step: 0, message: err.message, code: (err as any).code })
+        ctx.emit('agent/error', agent, turn, 0, err)
+      } catch { /* the error path itself is broken; nothing left to do */ }
     }
 
-    if (handle.isDisposed() && reason.kind === 'completed') {
-      reason = { kind: 'disposed' }
+    // Steering that arrived too late to join this turn (turn-end listeners,
+    // flush) becomes a queued message — it must never be stranded.
+    for (const message of agent.inbox.drainSteering()) {
+      agent.inbox.enqueue(message)
     }
-    session.append('turn/end', { turn, reason })
-    ctx.emit('agent/turn-end', agent, turn, reason)
-
-    // Durability checkpoint: persistence plugins drain write-behind buffers.
-    await ctx.parallel('session/flush', session)
 
     if (!agent.inbox.hasQueued) handle.setStatus('idle')
   }
 }
 
+async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: number): Promise<void> {
+  const { session } = agent
+
+  // Drain queued messages into the session — they trigger this turn.
+  const queued = agent.inbox.drainQueued()
+  const trigger: TurnTrigger = { kind: 'message', source: queued[0]!.source }
+  for (const message of queued) {
+    session.append('user/message', { content: message.content, source: message.source })
+  }
+
+  session.append('turn/start', { turn, trigger })
+  ctx.emit('agent/turn-start', agent, turn)
+
+  let reason: TurnEndReason = { kind: 'completed' }
+  let step = 0
+
+  while (true) {
+    step += 1
+
+    // Steering from the previous round's step-end/continuation listeners
+    // (or turn-start listeners on the first step) joins before the request.
+    drainSteering(ctx, agent, turn)
+
+    ctx.emit('agent/step-start', agent, turn, step)
+    session.append('step/start', { turn, step })
+
+    const abort = new AbortController()
+    handle.setAbort(abort)
+
+    let stepOutcome: { hadToolCalls: boolean } | { error: Error }
+    try {
+      stepOutcome = await runStep(ctx, agent, turn, step, abort.signal)
+    } catch (error: any) {
+      stepOutcome = { error: error instanceof Error ? error : new Error(String(error)) }
+    } finally {
+      handle.setAbort(undefined)
+    }
+
+    if ('error' in stepOutcome) {
+      // Steering that arrived during the failed step stays in the inbox —
+      // runLoop re-enqueues it as a queued message, so an abort-then-steer
+      // starts a fresh turn instead of being silently consumed.
+      session.append('step/end', { turn, step })
+      ctx.emit('agent/step-end', agent, turn, step)
+      const { error } = stepOutcome
+      if (handle.isDisposed()) {
+        reason = { kind: 'disposed' }
+      } else if (abort.signal.aborted) {
+        reason = { kind: 'aborted', reason: String(abort.signal.reason ?? 'aborted') }
+      } else {
+        session.append('error', { turn, step, message: error.message, code: (error as any).code })
+        ctx.emit('agent/error', agent, turn, step, error)
+        reason = { kind: 'error', message: error.message, code: (error as any).code }
+      }
+      break
+    }
+
+    // Steering that arrived during streaming/tool execution.
+    const steered = drainSteering(ctx, agent, turn)
+
+    session.append('step/end', { turn, step })
+    ctx.emit('agent/step-end', agent, turn, step)
+
+    const defaultDecision = stepOutcome.hadToolCalls || steered
+    let shouldContinue: boolean
+    try {
+      shouldContinue = await ctx.waterfall(
+        'agent/turn-continuation', agent, turn, defaultDecision,
+        async () => defaultDecision,
+      )
+    } catch (error: any) {
+      // A broken continuation plugin ends the turn, not the loop.
+      const err = error instanceof Error ? error : new Error(String(error))
+      session.append('error', { turn, step, message: err.message, code: (err as any).code })
+      ctx.emit('agent/error', agent, turn, step, err)
+      reason = { kind: 'error', message: err.message, code: (err as any).code }
+      break
+    }
+
+    // Steering from step-end/continuation listeners (the /goal pattern)
+    // demands the model see it — it overrides a negative decision; the
+    // next iteration's drain records it.
+    if (!shouldContinue && agent.inbox.hasSteering) shouldContinue = true
+
+    if (!shouldContinue || handle.isDisposed()) {
+      if (handle.isDisposed()) reason = { kind: 'disposed' }
+      break
+    }
+  }
+
+  session.append('turn/end', { turn, reason })
+  ctx.emit('agent/turn-end', agent, turn, reason)
+
+  // Durability checkpoint: persistence plugins drain write-behind buffers.
+  // A failing persistence plugin is reported but doesn't kill the agent.
+  try {
+    await ctx.parallel('session/flush', session)
+  } catch (error: any) {
+    const err = error instanceof Error ? error : new Error(String(error))
+    session.append('error', { turn, step, message: err.message, code: (err as any).code })
+    ctx.emit('agent/error', agent, turn, step, err)
+  }
+}
+
+/** Drain the steering queue into the session. Returns whether any arrived. */
+function drainSteering(ctx: Context, agent: LoopAgent, turn: number): boolean {
+  const messages = agent.inbox.drainSteering()
+  for (const message of messages) {
+    agent.session.append('steering/message', { turn, content: message.content, source: message.source })
+    ctx.emit('agent/steering', agent, turn, message.content, message.source)
+  }
+  return messages.length > 0
+}
+
 /** One step: assemble request → stream model → record → execute tools. */
 async function runStep(
   ctx: Context,
@@ -141,13 +209,16 @@ async function runStep(
     .join('\n\n')
 
   let request: GenerateOptions = {
-    model: options.model ?? 'default',
+    model: options.model ?? '',
     messages: session.deriveMessages(),
     system: system || undefined,
     tools: assembly.tools.length > 0 ? assembly.tools : undefined,
     signal,
   }
   request = await ctx.waterfall('agent/request', agent, turn, step, request, async () => request)
+  if (!request.model) {
+    throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
+  }
 
   // --- Model call (streaming-first; raw chunks are the replay record) ---
   const assembler = new BlockAssembler()
@@ -158,17 +229,23 @@ async function runStep(
     assembler.push(chunk)
   }
 
+  // The step-result waterfall runs BEFORE the session append so the log (the
+  // source of truth for derived history and replay) records the message that
+  // tool dispatch actually uses.
   let message: Message = assembler.message()
+  message = await ctx.waterfall('agent/step-result', agent, turn, step, message, async () => message)
+
   session.append('assistant/message', { turn, step, content: message.content })
   if (assembler.usage) {
     session.append('usage', { turn, step, usage: assembler.usage })
   }
 
-  message = await ctx.waterfall('agent/step-result', agent, turn, step, message, async () => message)
-
   // --- Tool execution (sequential; parallel execution is a TODO) ---
+  // ToolRegistry.execute converts tool failures (including aborts) into
+  // isError results, so abort is re-checked around every call here.
   const toolCalls = message.content.filter(block => block.type === 'tool-call')
   for (const call of toolCalls) {
+    if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
     session.append('tool/call', { turn, step, callId: call.id, name: call.name, arguments: call.arguments })
     let parsedArguments: unknown
     try {
@@ -189,17 +266,17 @@ async function runStep(
       content: result.content,
       isError: result.isError,
     })
+    if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
   }
 
   return { hadToolCalls: toolCalls.length > 0 }
 }
 
-function nextTurnNumber(session: LoopAgent['session']): number {
+/** The last turn number in a (possibly seeded) session log, or 0. */
+function lastTurnNumber(session: Session): number {
   for (let index = session.events.length - 1; index >= 0; index--) {
     const event = session.events[index]
-    if (event.type === 'turn/start') {
-      return (event.data as { turn: number }).turn + 1
-    }
+    if (event.type === 'turn/start') return event.data.turn
   }
-  return 1
+  return 0
 }
diff --git a/packages/agent-loop/tests/review-fixes.spec.ts b/packages/agent-loop/tests/review-fixes.spec.ts
new file mode 100644
index 0000000000..5a7af9058d
--- /dev/null
+++ b/packages/agent-loop/tests/review-fixes.spec.ts
@@ -0,0 +1,547 @@
+import { describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+import LlmService, { ContentBlock, StreamChunk } from '@deepseek-ai/dsh-llm'
+import SessionStore, { Session, SessionEvent } from '@deepseek-ai/dsh-session'
+import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
+import ToolRegistry from '@deepseek-ai/dsh-tools'
+import AgentRegistry from '@deepseek-ai/dsh-agent'
+import AgentLoop, { LoopAgent } from '@deepseek-ai/dsh-agent-loop'
+import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
+
+/**
+ * Regression tests for the findings of the first architecture review
+ * (Codex + sub-agent, post phase-1). Each describe block names the finding.
+ */
+
+async function harness(adapter: MockAdapter) {
+  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)
+  return ctx
+}
+
+function waitForIdle(ctx: Context, agent: LoopAgent): Promise<void> {
+  return new Promise((resolve) => {
+    const dispose = ctx.on('agent/status', (subject, status) => {
+      if (subject === agent && status === 'idle') {
+        dispose()
+        resolve()
+      }
+    })
+  })
+}
+
+function send(agent: LoopAgent, text: string) {
+  agent.send([{ type: 'text', text }])
+}
+
+describe('HIGH: session log records what agent/step-result actually produced', () => {
+  it('a step-result rewrite is what the log, derived history, and tool dispatch all see', async () => {
+    const adapter = new MockAdapter([textResponse('original'), textResponse('done')])
+    const ctx = await harness(adapter)
+    const executed: string[] = []
+    ctx.tools.register({
+      name: 'injected-tool',
+      description: '',
+      parameters: { type: 'object' },
+      async execute() {
+        executed.push('injected-tool')
+        return [{ type: 'text', text: 'ran' }]
+      },
+    })
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    // Plugin rewrites the message: replaces the text AND adds a tool call.
+    let rewritten = false
+    ctx.on('agent/step-result', async (_agent, _turn, _step, message, next) => {
+      if (rewritten) return next()
+      rewritten = true
+      return {
+        role: 'assistant' as const,
+        content: [
+          { type: 'text' as const, text: 'rewritten' },
+          { type: 'tool-call' as const, id: 'c-injected', name: 'injected-tool', arguments: '{}' },
+        ],
+      }
+    })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    // the injected tool call was dispatched…
+    expect(executed).toEqual(['injected-tool'])
+    // …and the session log recorded the REWRITTEN message, not the original
+    const recorded = agent.session.events.find(e => e.type === 'assistant/message')!
+    expect(JSON.stringify(recorded.data)).toContain('rewritten')
+    expect(JSON.stringify(recorded.data)).not.toContain('original')
+    // tool/call + tool/result correlate with the injected call id
+    const callEvent = agent.session.events.find(e => e.type === 'tool/call')!
+    expect((callEvent.data as any).callId).toBe('c-injected')
+    // derived history shows the rewritten message (replay-correct)
+    const derived = agent.session.deriveMessages()
+    expect(JSON.stringify(derived)).toContain('rewritten')
+    expect(JSON.stringify(derived)).not.toContain('original')
+  })
+})
+
+describe('HIGH: abort during tool execution ends the turn', () => {
+  it('abort() inside a tool prevents both remaining tools and the next model step', async () => {
+    const adapter = new MockAdapter([
+      // model asks for two tool calls in one step
+      [
+        { type: 'block-start', index: 0, blockType: 'tool-call' },
+        { type: 'block-end', index: 0, block: { type: 'tool-call', id: 'c1', name: 'aborter', arguments: '{}' } },
+        { type: 'block-start', index: 1, blockType: 'tool-call' },
+        { type: 'block-end', index: 1, block: { type: 'tool-call', id: 'c2', name: 'second', arguments: '{}' } },
+        { type: 'finish', reason: { kind: 'tool-calls' } },
+      ] satisfies StreamChunk[],
+      textResponse('should never be requested'),
+    ])
+    const ctx = await harness(adapter)
+    const executed: string[] = []
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+    ctx.tools.register({
+      name: 'aborter',
+      description: '',
+      parameters: { type: 'object' },
+      async execute() {
+        executed.push('aborter')
+        agent.abort('user interrupt')
+        return [{ type: 'text', text: 'done' }]
+      },
+    })
+    ctx.tools.register({
+      name: 'second',
+      description: '',
+      parameters: { type: 'object' },
+      async execute() {
+        executed.push('second')
+        return [{ type: 'text', text: 'done' }]
+      },
+    })
+
+    const reasons: any[] = []
+    ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    expect(executed).toEqual(['aborter'])           // second tool never ran
+    expect(adapter.requests).toHaveLength(1)        // no follow-up model call
+    expect(reasons).toEqual([{ kind: 'aborted', reason: 'user interrupt' }])
+  })
+})
+
+describe('HIGH: steering from late extension points is never stranded', () => {
+  it('steer() from an agent/step-end listener reaches the next request (/goal pattern)', async () => {
+    const adapter = new MockAdapter([
+      toolCallResponse('c1', 'echo', { text: 'x' }),
+      textResponse('after steering'),
+    ])
+    const ctx = await harness(adapter)
+    ctx.tools.register({
+      name: 'echo',
+      description: '',
+      parameters: { type: 'object' },
+      async execute(args: any) {
+        return [{ type: 'text', text: String(args.text) }]
+      },
+    })
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    let steeredOnce = false
+    ctx.on('agent/step-end', () => {
+      if (steeredOnce) return
+      steeredOnce = true
+      agent.steer([{ type: 'text', text: 'goal reminder from step-end' }])
+    })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    expect(adapter.requests).toHaveLength(2)
+    expect(JSON.stringify(adapter.requests[1].messages)).toContain('goal reminder from step-end')
+  })
+
+  it('steer() from an agent/turn-continuation listener overrides a stop decision', async () => {
+    const adapter = new MockAdapter([
+      textResponse('no tools, would stop here'),
+      textResponse('continued because of steering'),
+    ])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    let steeredOnce = false
+    ctx.on('agent/turn-continuation', async (_agent, _turn, _decision, next) => {
+      if (!steeredOnce) {
+        steeredOnce = true
+        agent.steer([{ type: 'text', text: 'one more thing' }])
+      }
+      return next()
+    })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    // the default decision was false (no tools), but steering forced step 2
+    expect(adapter.requests).toHaveLength(2)
+    expect(JSON.stringify(adapter.requests[1].messages)).toContain('one more thing')
+  })
+
+  it('steer() from an agent/turn-end listener becomes a queued message for the next turn', async () => {
+    const adapter = new MockAdapter([textResponse('turn 1'), textResponse('turn 2')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    let steeredOnce = false
+    ctx.on('agent/turn-end', () => {
+      if (steeredOnce) return
+      steeredOnce = true
+      agent.steer([{ type: 'text', text: 'too late for this turn' }])
+    })
+
+    const turns: number[] = []
+    ctx.on('agent/turn-start', (_agent, turn) => void turns.push(turn))
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+    // the loop chains directly into turn 2 (status never returns to idle in
+    // between), so the first idle transition means both turns are complete
+
+    expect(turns).toEqual([1, 2])
+    expect(adapter.requests).toHaveLength(2)
+    expect(JSON.stringify(adapter.requests[1].messages)).toContain('too late for this turn')
+  })
+
+  it('steering queued during an aborted step is re-delivered, not silently consumed', async () => {
+    const adapter = new MockAdapter(['hang', textResponse('recovered')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    send(agent, 'go')
+    await new Promise(r => setTimeout(r, 30))
+    agent.steer([{ type: 'text', text: 'redirect' }])
+    agent.abort('user interrupt')
+    await waitForIdle(ctx, agent)
+
+    // a new turn ran with the steering content delivered as a message
+    expect(adapter.requests).toHaveLength(2)
+    expect(JSON.stringify(adapter.requests[1].messages)).toContain('redirect')
+  })
+})
+
+describe('HIGH: plugin exceptions are contained', () => {
+  it('a throwing agent/turn-continuation listener ends the turn with an error, loop survives', async () => {
+    const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    let threwOnce = false
+    ctx.on('agent/turn-continuation', async (): Promise<boolean> => {
+      if (!threwOnce) {
+        threwOnce = true
+        throw new Error('broken continuation plugin')
+      }
+      return false
+    })
+
+    const errors: Error[] = []
+    ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
+
+    send(agent, 'first')
+    await waitForIdle(ctx, agent)
+    expect(errors.map(e => e.message)).toEqual(['broken continuation plugin'])
+
+    // the loop is still alive: a second send works normally
+    send(agent, 'second')
+    await waitForIdle(ctx, agent)
+    expect(adapter.requests).toHaveLength(2)
+    expect(agent.status).toBe('idle')
+  })
+
+  it('a rejecting session/flush listener is reported but does not kill the agent', async () => {
+    const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    let rejectedOnce = false
+    ctx.on('session/flush', async () => {
+      if (!rejectedOnce) {
+        rejectedOnce = true
+        throw new Error('disk full')
+      }
+    })
+
+    const errors: Error[] = []
+    ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
+
+    send(agent, 'first')
+    await waitForIdle(ctx, agent)
+    expect(errors.map(e => e.message)).toEqual(['disk full'])
+
+    send(agent, 'second')
+    await waitForIdle(ctx, agent)
+    expect(adapter.requests).toHaveLength(2)
+  })
+})
+
+describe('MEDIUM: disposed status is part of the agent/status contract', () => {
+  it('disposing the fiber emits agent/status(disposed) and ends the turn with reason disposed', async () => {
+    const adapter = new MockAdapter(['hang'])
+    const ctx = await harness(adapter)
+
+    let agent!: LoopAgent
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      agent = inner.agentLoop.create('scoped', { model: 'mock' })
+    }, { inject: ['agentLoop'] }))
+
+    const statuses: string[] = []
+    const reasons: any[] = []
+    ctx.on('agent/status', (_agent, status) => void statuses.push(status))
+    ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
+
+    send(agent, 'go')
+    await new Promise(r => setTimeout(r, 30))
+    await fiber.dispose()
+    await agent.done
+
+    expect(statuses).toEqual(['running', 'disposed'])
+    expect(reasons).toEqual([{ kind: 'disposed' }])
+  })
+
+  it('a throwing agent/status listener cannot break disposal or leak the registry entry', async () => {
+    const adapter = new MockAdapter(['hang'])
+    const ctx = await harness(adapter)
+
+    let agent!: LoopAgent
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      agent = inner.agentLoop.create('scoped', { model: 'mock' })
+    }, { inject: ['agentLoop'] }))
+
+    ctx.on('agent/status', (_agent, status) => {
+      if (status === 'disposed') throw new Error('broken status listener')
+    })
+
+    send(agent, 'go')
+    await new Promise(r => setTimeout(r, 30))
+    await fiber.dispose()
+    await agent.done // must not hang
+
+    expect(agent.status).toBe('disposed')
+    expect(ctx.agents.get('scoped')).toBeUndefined() // unregistered despite the throw
+  })
+})
+
+describe('MEDIUM: misc registry and config fixes', () => {
+  it('duplicate adapter registration is rejected', async () => {
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    const adapter = new MockAdapter([])
+    ctx.llm.registerAdapter(['m1'], adapter)
+    expect(() => ctx.llm.registerAdapter(['m1'], new MockAdapter([])))
+      .toThrow('already registered')
+    // the original registration survives the failed attempt
+    expect(ctx.llm.models()).toEqual(['m1'])
+  })
+
+  it('an agent without a model fails the step with a clear error (not NO_ADAPTER for "default")', async () => {
+    const adapter = new MockAdapter([textResponse('never')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', {}) // no model
+
+    const errors: Error[] = []
+    ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+    expect(errors).toHaveLength(1)
+    expect(errors[0].message).toContain('has no model')
+    expect(errors[0].message).toContain('agent/request')
+  })
+
+  it('the agent/request waterfall can supply the model for a model-less agent', async () => {
+    const adapter = new MockAdapter([textResponse('routed')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', {}) // no model — router plugin decides
+
+    ctx.on('agent/request', async (_agent, _turn, _step, options, next) => {
+      options.model = 'mock'
+      return next()
+    })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+    expect(adapter.requests).toHaveLength(1)
+    expect(agent.session.deriveMessages().at(-1)?.content).toEqual([{ type: 'text', text: 'routed' }])
+  })
+
+  it('agent/queued carries the resolved source; agent/steering carries its source', async () => {
+    const adapter = new MockAdapter([toolCallResponse('c1', 'noop', {}), textResponse('done')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+    ctx.tools.register({
+      name: 'noop',
+      description: '',
+      parameters: { type: 'object' },
+      async execute() {
+        agent.steer([{ type: 'text', text: 's' }], { source: { kind: 'plugin', plugin: 'goal' } })
+        return []
+      },
+    })
+
+    const queuedSources: any[] = []
+    const steeringSources: any[] = []
+    ctx.on('agent/queued', (_agent, _content, info) => void queuedSources.push(info))
+    ctx.on('agent/steering', (_agent, _turn, _content, source) => void steeringSources.push(source))
+
+    send(agent, 'go') // no explicit source → default {kind:'user'} must be visible
+    await waitForIdle(ctx, agent)
+
+    expect(queuedSources[0]).toEqual({ source: { kind: 'user' }, steering: false })
+    expect(queuedSources[1]).toEqual({ source: { kind: 'plugin', plugin: 'goal' }, steering: true })
+    expect(steeringSources).toEqual([{ kind: 'plugin', plugin: 'goal' }])
+  })
+})
+
+describe('MEDIUM: turn numbering continues across seeded (forked) sessions', () => {
+  it('a forked agent continues turn numbers after the seed log', async () => {
+    const first = new MockAdapter([textResponse('turn one')])
+    const ctx = await harness(first)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+    send(agent, 'first')
+    await waitForIdle(ctx, agent)
+
+    // fork: seed a second context's agent with the first session's log
+    const second = new MockAdapter([textResponse('turn two')])
+    const ctx2 = new Context()
+    await ctx2.plugin(LlmService)
+    await ctx2.plugin(SessionStore)
+    await ctx2.plugin(SystemPrompt)
+    await ctx2.plugin(ToolRegistry)
+    await ctx2.plugin(AgentRegistry)
+    await ctx2.plugin(AgentLoop, { agents: [] })
+    ctx2.llm.registerAdapter(['mock'], second)
+
+    const seeded = ctx2.sessions.create('forked', [...agent.session.events])
+    const forked = new LoopAgent(ctx2, 'forked-agent', { model: 'mock' }, seeded)
+    ctx2.effect(() => forked.start())
+
+    const turns: number[] = []
+    ctx2.on('agent/turn-start', (_agent, turn) => void turns.push(turn))
+    forked.send([{ type: 'text', text: 'continue' }])
+    await new Promise<void>((resolve) => {
+      ctx2.on('agent/status', (subject, status) => {
+        if (subject === forked && status === 'idle') resolve()
+      })
+    })
+
+    expect(turns).toEqual([2])
+  })
+})
+
+describe('LOW: BlockAssembler and streamBlocks edge cases', () => {
+  it('ignores deltas arriving after block-end for the same index (malformed stream)', async () => {
+    const { BlockAssembler } = await import('@deepseek-ai/dsh-llm')
+    const assembler = new BlockAssembler()
+    assembler.push({ type: 'block-start', index: 0, blockType: 'text' })
+    assembler.push({ type: 'text-delta', index: 0, text: 'good' })
+    assembler.push({ type: 'block-end', index: 0, block: { type: 'text', text: 'good' } })
+    assembler.push({ type: 'text-delta', index: 0, text: ' straggler' })
+    expect(assembler.blocks()).toEqual([{ type: 'text', text: 'good' }])
+  })
+
+  it('assembles tool-call blocks from deltas without block-end', async () => {
+    const { BlockAssembler } = await import('@deepseek-ai/dsh-llm')
+    const assembler = new BlockAssembler()
+    assembler.push({ type: 'tool-call-delta', index: 0, id: 'c9', name: 'echo', argumentsDelta: '{"a"' })
+    assembler.push({ type: 'tool-call-delta', index: 0, id: 'c9', argumentsDelta: ':1}' })
+    expect(assembler.blocks()).toEqual([
+      { type: 'tool-call', id: 'c9', name: 'echo', arguments: '{"a":1}' },
+    ])
+  })
+
+  it('streamBlocks flushes delta-only blocks at end of stream (matches generate())', async () => {
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    const deltaOnly: StreamChunk[] = [
+      { type: 'text-delta', index: 0, text: 'no ' },
+      { type: 'text-delta', index: 0, text: 'block-end' },
+      { type: 'finish', reason: { kind: 'stop' } },
+    ]
+    ctx.llm.registerAdapter(['m'], new MockAdapter([deltaOnly, deltaOnly]))
+
+    const blocks: ContentBlock[] = []
+    for await (const block of ctx.llm.streamBlocks({ model: 'm', messages: [] })) blocks.push(block)
+    expect(blocks).toEqual([{ type: 'text', text: 'no block-end' }])
+
+    const generated = await ctx.llm.generate({ model: 'm', messages: [] })
+    expect(generated.message.content).toEqual(blocks)
+  })
+
+  it('streamBlocks preserves stream order when an open block precedes a closed one', async () => {
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    // index 0 never gets block-end (delta-only); index 1 closes mid-stream.
+    const interleaved: StreamChunk[] = [
+      { type: 'text-delta', index: 0, text: 'first, open' },
+      { type: 'block-start', index: 1, blockType: 'text' },
+      { type: 'text-delta', index: 1, text: 'second, closed' },
+      { type: 'block-end', index: 1, block: { type: 'text', text: 'second, closed' } },
+      { type: 'finish', reason: { kind: 'stop' } },
+    ]
+    ctx.llm.registerAdapter(['m'], new MockAdapter([interleaved, interleaved]))
+
+    const blocks: ContentBlock[] = []
+    for await (const block of ctx.llm.streamBlocks({ model: 'm', messages: [] })) blocks.push(block)
+    expect(blocks).toEqual([
+      { type: 'text', text: 'first, open' },
+      { type: 'text', text: 'second, closed' },
+    ])
+
+    // identical to generate()'s assembled order
+    const generated = await ctx.llm.generate({ model: 'm', messages: [] })
+    expect(generated.message.content).toEqual(blocks)
+  })
+
+  it('streamBlocks yields closed blocks incrementally once preceding blocks close', async () => {
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    const script: StreamChunk[] = [
+      { type: 'block-start', index: 0, blockType: 'text' },
+      { type: 'text-delta', index: 0, text: 'a' },
+      { type: 'block-end', index: 0, block: { type: 'text', text: 'a' } },
+      { type: 'block-start', index: 1, blockType: 'text' },
+      { type: 'text-delta', index: 1, text: 'b' },
+      { type: 'block-end', index: 1, block: { type: 'text', text: 'b' } },
+      { type: 'finish', reason: { kind: 'stop' } },
+    ]
+    ctx.llm.registerAdapter(['m'], new MockAdapter([script]))
+
+    const blocks: ContentBlock[] = []
+    for await (const block of ctx.llm.streamBlocks({ model: 'm', messages: [] })) blocks.push(block)
+    expect(blocks).toEqual([{ type: 'text', text: 'a' }, { type: 'text', text: 'b' }])
+  })
+})
+
+describe('LOW: discriminated SessionEvent narrows without casts', () => {
+  it('narrows event.data from event.type', () => {
+    const session = new Session('s')
+    const appended: SessionEvent = session.append('tool/call', {
+      turn: 1, step: 1, callId: 'c1', name: 'echo', arguments: '{}',
+    })
+    // compile-time: this switch narrows; runtime: values flow through
+    switch (appended.type) {
+      case 'tool/call': {
+        expect(appended.data.callId).toBe('c1')
+        expect(appended.data.name).toBe('echo')
+        break
+      }
+      default: throw new Error('wrong narrow')
+    }
+  })
+})
diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts
index f66137cdc7..b531776d78 100644
--- a/packages/agent/src/types.ts
+++ b/packages/agent/src/types.ts
@@ -69,8 +69,11 @@ declare module 'cordis' {
     'agent/disposed'(agent: Agent): void
     /** Agent status changed (idle/running/disposed). */
     'agent/status'(agent: Agent, status: AgentStatus): void
-    /** A message entered the agent's inbox (queued or steering). */
-    'agent/queued'(agent: Agent, content: ContentBlock[], options: SendOptions & { steering: boolean }): void
+    /**
+     * A message entered the agent's inbox (queued or steering). `source` is
+     * the resolved source (defaults applied), not the caller's raw options.
+     */
+    'agent/queued'(agent: Agent, content: ContentBlock[], info: { source: MessageSource; steering: boolean }): void
 
     // ---- turn/step boundaries (emit) ----
     'agent/turn-start'(agent: Agent, turn: number): void
@@ -100,7 +103,7 @@ declare module 'cordis' {
     /** A raw stream chunk arrived (token-level UI/log feed). */
     'agent/stream-chunk'(agent: Agent, turn: number, step: number, chunk: StreamChunk): void
     /** Steering content was injected into a running turn. */
-    'agent/steering'(agent: Agent, turn: number, content: ContentBlock[]): void
+    'agent/steering'(agent: Agent, turn: number, content: ContentBlock[], source: MessageSource): void
     /** A step or turn errored. */
     'agent/error'(agent: Agent, turn: number, step: number, error: Error): void
   }
diff --git a/packages/llm/src/assembler.ts b/packages/llm/src/assembler.ts
index 3cd087dbc9..c7f23a3eda 100644
--- a/packages/llm/src/assembler.ts
+++ b/packages/llm/src/assembler.ts
@@ -1,5 +1,15 @@
 import type { ContentBlock, FinishReason, GenerateResult, Message, StreamChunk, TokenUsage } from './types.ts'
 
+interface PartialBlock {
+  blockType: string
+  text: string
+  toolCallId?: string
+  toolCallName?: string
+  toolCallArguments: string
+  /** Set by `block-end` — authoritative, and freezes the partial. */
+  block?: ContentBlock
+}
+
 /**
  * Incrementally assembles raw {@link StreamChunk}s into complete
  * {@link ContentBlock}s and a final assistant {@link Message}.
@@ -7,44 +17,45 @@ import type { ContentBlock, FinishReason, GenerateResult, Message, StreamChunk,
  * This is the single shared assembly implementation: the agent loop feeds it
  * while logging raw chunks for replay fidelity, and `LlmService.generate()` /
  * `streamBlocks()` use it to offer assembled views of the same stream.
+ *
+ * Tolerant of delta-only protocols (no block-start/end); deltas arriving for
+ * an index already closed by `block-end` are ignored (malformed stream) so a
+ * misbehaving adapter cannot grow memory or corrupt a completed block.
  */
 export class BlockAssembler {
-  private partials = new Map<number, {
-    blockType: string
-    text: string
-    toolCallId?: string
-    toolCallName?: string
-    toolCallArguments: string
-    block?: ContentBlock
-  }>()
-
+  private partials = new Map<number, PartialBlock>()
   private order: number[] = []
+  private flushed = 0
   private _usage: TokenUsage | undefined
   private _finish: FinishReason | undefined
 
   /**
    * Feed one chunk. Returns the completed block when the chunk closes one
-   * (either an explicit `block-end` or an implicit close), otherwise undefined.
+   * (an explicit `block-end`), otherwise undefined.
    */
   push(chunk: StreamChunk): ContentBlock | undefined {
     switch (chunk.type) {
       case 'block-start': {
-        if (!this.partials.has(chunk.index)) this.order.push(chunk.index)
-        this.partials.set(chunk.index, {
-          blockType: chunk.blockType,
-          text: '',
-          toolCallArguments: '',
-        })
+        if (!this.partials.has(chunk.index)) {
+          this.order.push(chunk.index)
+          this.partials.set(chunk.index, {
+            blockType: chunk.blockType,
+            text: '',
+            toolCallArguments: '',
+          })
+        }
         return
       }
       case 'text-delta':
       case 'reasoning-delta': {
         const partial = this.ensure(chunk.index, chunk.type === 'text-delta' ? 'text' : 'reasoning')
+        if (partial.block) return // closed by block-end; ignore stragglers
         partial.text += chunk.text
         return
       }
       case 'tool-call-delta': {
         const partial = this.ensure(chunk.index, 'tool-call')
+        if (partial.block) return // closed by block-end; ignore stragglers
         partial.toolCallId = chunk.id
         if (chunk.name) partial.toolCallName = chunk.name
         partial.toolCallArguments += chunk.argumentsDelta
@@ -66,7 +77,7 @@ export class BlockAssembler {
     }
   }
 
-  private ensure(index: number, blockType: string) {
+  private ensure(index: number, blockType: string): PartialBlock {
     let partial = this.partials.get(index)
     if (!partial) {
       partial = { blockType, text: '', toolCallArguments: '' }
@@ -76,23 +87,57 @@ export class BlockAssembler {
     return partial
   }
 
+  private assemble(partial: PartialBlock, index: number): ContentBlock {
+    if (partial.block) return partial.block
+    switch (partial.blockType) {
+      case 'text': return { type: 'text', text: partial.text }
+      case 'reasoning': return { type: 'reasoning', text: partial.text }
+      case 'tool-call': return {
+        type: 'tool-call',
+        id: partial.toolCallId ?? `call-${index}`,
+        name: partial.toolCallName ?? '',
+        arguments: partial.toolCallArguments,
+      }
+      default: throw new Error(`cannot assemble incomplete block of type "${partial.blockType}"`)
+    }
+  }
+
   /** Assemble all blocks seen so far, in stream order. */
   blocks(): ContentBlock[] {
-    return this.order.map((index) => {
-      const partial = this.partials.get(index)!
-      if (partial.block) return partial.block
-      switch (partial.blockType) {
-        case 'text': return { type: 'text', text: partial.text }
-        case 'reasoning': return { type: 'reasoning', text: partial.text }
-        case 'tool-call': return {
-          type: 'tool-call',
-          id: partial.toolCallId ?? `call-${index}`,
-          name: partial.toolCallName ?? '',
-          arguments: partial.toolCallArguments,
-        }
-        default: throw new Error(`cannot assemble incomplete block of type "${partial.blockType}"`)
-      }
-    })
+    return this.order.map(index => this.assemble(this.partials.get(index)!, index))
+  }
+
+  /**
+   * Streaming flush: returns (once) every block that is complete AND has no
+   * incomplete block before it in stream order. Call after each `push()`;
+   * blocks come out strictly in stream order, so a streaming consumer sees
+   * exactly the sequence `blocks()` would produce.
+   */
+  flushReady(): ContentBlock[] {
+    const ready: ContentBlock[] = []
+    while (this.flushed < this.order.length) {
+      const partial = this.partials.get(this.order[this.flushed])!
+      if (!partial.block) break
+      ready.push(partial.block)
+      this.flushed += 1
+    }
+    return ready
+  }
+
+  /**
+   * End-of-stream flush: returns (once) all not-yet-flushed blocks, in stream
+   * order, assembling still-open ones from their deltas (delta-only
+   * protocols). After this, `flushReady()` + `flushRemaining()` together have
+   * yielded exactly `blocks()`.
+   */
+  flushRemaining(): ContentBlock[] {
+    const remaining: ContentBlock[] = []
+    while (this.flushed < this.order.length) {
+      const index = this.order[this.flushed]
+      remaining.push(this.assemble(this.partials.get(index)!, index))
+      this.flushed += 1
+    }
+    return remaining
   }
 
   get usage(): TokenUsage | undefined {
diff --git a/packages/llm/src/index.ts b/packages/llm/src/index.ts
index 6d838cd07e..65165d48f8 100644
--- a/packages/llm/src/index.ts
+++ b/packages/llm/src/index.ts
@@ -56,6 +56,11 @@ export class LlmService extends Service {
   /** Register an adapter for the given model names. Disposed with the fiber. */
   registerAdapter(models: string[], adapter: LlmAdapter): () => void {
     return this.ctx.effect(() => {
+      for (const model of models) {
+        if (this.adapters.has(model)) {
+          throw new LlmError(`an adapter for model "${model}" is already registered`, 'DUPLICATE_ADAPTER')
+        }
+      }
       for (const model of models) this.adapters.set(model, adapter)
       this.ctx.emit('llm/adapter-change')
       return () => {
@@ -85,14 +90,19 @@ export class LlmService extends Service {
 
   /**
    * Stream one model call as completed content blocks — a convenience view
-   * for consumers that don't care about token-level deltas.
+   * for consumers that don't care about token-level deltas. Blocks are
+   * yielded strictly in stream order as soon as they (and everything before
+   * them) complete; blocks left open at end of stream (delta-only protocols)
+   * are assembled and flushed last, so the sequence always equals
+   * `generate()`'s `message.content`.
    */
   async * streamBlocks(options: GenerateOptions): AsyncIterable<ContentBlock> {
     const assembler = new BlockAssembler()
     for await (const chunk of this.stream(options)) {
-      const block = assembler.push(chunk)
-      if (block) yield block
+      assembler.push(chunk)
+      yield * assembler.flushReady()
     }
+    yield * assembler.flushRemaining()
   }
 
   /** One model call, fully assembled (drains the chunk stream). */
diff --git a/packages/session/src/index.ts b/packages/session/src/index.ts
index d5c6b2fba2..b5dabf8854 100644
--- a/packages/session/src/index.ts
+++ b/packages/session/src/index.ts
@@ -66,9 +66,9 @@ export class Session {
 
   /** Append one event. Synchronous — the hot path never blocks on I/O. */
   append<T extends SessionEventType>(type: T, data: SessionEventMap[T]): SessionEvent<T> {
-    const event: SessionEvent<T> = { type, seq: this.log.length, time: Date.now(), data }
-    this.log.push(event as SessionEvent)
-    this.onAppend?.(event as SessionEvent)
+    const event = { type, seq: this.log.length, time: Date.now(), data } as SessionEvent<T>
+    this.log.push(event)
+    this.onAppend?.(event)
     return event
   }
 
@@ -87,17 +87,15 @@ export class Session {
     for (const event of this.log) {
       switch (event.type) {
         case 'user/message': {
-          const { content } = event.data as SessionEventMap['user/message']
-          messages.push({ role: 'user', content })
+          messages.push({ role: 'user', content: event.data.content })
           break
         }
         case 'assistant/message': {
-          const { content } = event.data as SessionEventMap['assistant/message']
-          messages.push({ role: 'assistant', content })
+          messages.push({ role: 'assistant', content: event.data.content })
           break
         }
         case 'tool/result': {
-          const { callId, content, isError } = event.data as SessionEventMap['tool/result']
+          const { callId, content, isError } = event.data
           messages.push({
             role: 'user',
             content: [{ type: 'tool-result', toolCallId: callId, content, isError }],
@@ -105,12 +103,12 @@ export class Session {
           break
         }
         case 'context/message': {
-          const { content, source } = event.data as SessionEventMap['context/message']
+          const { content, source } = event.data
           messages.push({ role: 'user', content: renderTagged('context', content, source) })
           break
         }
         case 'steering/message': {
-          const { content, source } = event.data as SessionEventMap['steering/message']
+          const { content, source } = event.data
           messages.push({ role: 'user', content: renderTagged('steering', content, source) })
           break
         }
@@ -139,8 +137,8 @@ export class SessionStore extends Service {
     id ??= `session-${++this.counter}`
     if (this.store.has(id)) throw new Error(`session "${id}" already exists`)
     const session = new Session(id, seed)
-    session.onAppend = (event) => this.ctx.emit('session/event', session, event)
     this.ctx.effect(() => {
+      session.onAppend = (event) => this.ctx.emit('session/event', session, event)
       this.store.set(id, session)
       this.ctx.emit('session/created', session)
       return () => {
diff --git a/packages/session/src/types.ts b/packages/session/src/types.ts
index c108019f07..44e904a55c 100644
--- a/packages/session/src/types.ts
+++ b/packages/session/src/types.ts
@@ -63,12 +63,19 @@ export interface SessionEventMap {
 
 export type SessionEventType = keyof SessionEventMap
 
-/** One immutable entry in the session log. */
-export interface SessionEvent<T extends SessionEventType = SessionEventType> {
-  type: T
-  /** Monotonic sequence number within the session. */
-  seq: number
-  /** Unix epoch milliseconds. */
-  time: number
-  data: SessionEventMap[T]
-}
+/**
+ * One immutable entry in the session log.
+ *
+ * A proper discriminated union over `type` (not independent `type`/`data`
+ * unions), so `switch (event.type)` narrows `event.data` without casts.
+ */
+export type SessionEvent<T extends SessionEventType = SessionEventType> = {
+  [K in SessionEventType]: {
+    type: K
+    /** Monotonic sequence number within the session. */
+    seq: number
+    /** Unix epoch milliseconds. */
+    time: number
+    data: SessionEventMap[K]
+  }
+}[T]