feat(loop,gateway): 13.4.5 目标态 —— 回合内 steer（不打断、新消息进同一上下文流）+ steer_ack 帧

规格＝design/15 末尾「13.4.5 目标态规格」（不变量 I1/I2/I3）。调研依据：豆包无公开技术文档（不编造）；
同类成熟 Agent 产品无一做同会话并行回合，共同范式＝ queue + 回合内 steer（Claude Code / Cline PR #13958 /
Codex Queue·Steer / open-webui / LangGraph）。

- 注入点＝**回合内边界**（上一模型步 + 它那批工具调用都已结束、下一次模型调用还没开始）：
  driveTurn 每 step 组装前调用一次 SteerSource（run.go）——完成 step 只能靠工具批次推进，
  故唯一可注入处就是「工具批次结束之后」，循环里没有任何「起第二个回合」的分支（**I1 结构性成立**）
- **I2 恰好一次**：注入只进**本轮 view、绝不写回 msgs**（checkpoint 与记忆提取面保持干净）；
  被注入的消息从 pending 移除，不会由自己的回合再答一次
- **I3 转向不取消工具**：注入调用点在 runToolCalls **之后**（此刻无工具在飞），循环从不因注入 cancel 任何 ctx；
  边界序列有测试钉死（恰为 [1 2 …]，无批次中途值）
- 判定口径＝ **pending ∩ 转录**：候选是本会话 pending 条目，每个边界重读 history.Recent 验证已落库，
  按**行 id** 去重、按 pending FIFO 注入。**不用时间戳**（Windows time.Now() 步长 ~0.44ms，连发常同毫秒并列，
  任何 ms 基线都会漏注入或重复注入）；跨重启不成立也不需要（在飞回合与 pending 都在内存）
- 新下行帧 `steer_ack`：`{type, sessionId, data:{messageId, atStep}}`，语义＝「已被本回合吃进、尚未被正面回答」
  （驱动前端灰态）；**不进 frameReachesAll** ⇒ fail-closed 只发 owner+admin（测试双钉）；经 hub 广播、不经 bus
- 组合根**零改动**：per-turn 状态用 ctx 承载（loop.WithSteerSource），编排器在回合启动/续跑时挂
- 守卫：session_orchestrator.go 加完接线后 505 行 ⇒ 按职责纯搬出 session_events.go（常量取值未改、行为零变化）
- 复核：build/vet 绿 + **全树 0 FAIL** + linux/darwin 交叉编译绿；loop 6 例（含 3 子测）+ gateway 2 例 E2E；
  **4 条变异反证**（删 consumePending → 模型步数 4≠3；注入写回 msgs → 快照被污染；批次中途注入 → 边界 [1 1 2]；
  把帧加进白名单 → viewer 收到）均已还原
- 未做（登记）：IM 自有 Chatter 落库的消息不参与 steer（有意，避免重复作答）；pending 条目无自落库 id
  ⇒ 配对用 (内容,图片) 逐行认领；同回合超 20 条用户消息时超出者留 pending 由各自回合作答（不丢）

internal/runtime/loop/flow.go

@@ -29,11 +29,11 @@ type allowedSessions func(sessionID string) bool
 // 而是把其 span 发布到所属会话下，因此子代理渲染为
 // 父图的一条泳道，而不是第二张图（design/19 §2）。
 type FlowPublisher struct {
-	mu     sync.Mutex
+	mu      sync.Mutex
 	publish func(subject string, payload any)
-	allow  allowedSessions
-	graphs map[string]*trace.RunGraph
-	rounds map[string]int
+	allow   allowedSessions
+	graphs  map[string]*trace.RunGraph
+	rounds  map[string]int
 }
 
 // NewFlowPublisher 构建一个 FlowPublisher。publish 非 nil 时，会在
@@ -84,6 +84,31 @@ func (p *FlowPublisher) EndRound(sessionID, outputText string) {
 	p.publishLocked(g)
 }
 
+// MarkInputSteered 给本轮的 input 节点打上 `steered:true` 标记并发布一帧
+// （design/15 §5 / 裁决 13.4.5「注入时给 input 节点打 steered:true 标记」）。
+// 它让链路图如实表达「这一轮的输入在回合中途被追加过」——图仍然只有一条主干
+// （I1：同会话恒只有一个在飞回合），而不是被画成第二轮。
+//
+// nil 接收者、空会话、发布白名单外的会话、或该会话还没有图时都是空操作；标记
+// 只并入既有 input 节点（Upsert 的合并语义），绝不新建节点、绝不改动它的标签。
+func (p *FlowPublisher) MarkInputSteered(sessionID string) {
+	if p == nil || sessionID == "" || !p.allowed(sessionID) {
+		return
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	g := p.graphs[sessionID]
+	if g == nil {
+		return
+	}
+	g.Upsert(trace.NodeSpec{
+		ID:    "input",
+		Kind:  "input",
+		Props: map[string]any{"steered": true},
+	})
+	p.publishLocked(g)
+}
+
 // SpanStarted 为一个刚开始的 span 记录一个运行中的占位
 // 节点。state_transition span 会被图构建器省略。
 func (p *FlowPublisher) SpanStarted(sessionID string, sp *trace.Span) {

internal/runtime/loop/run.go

@@ -85,6 +85,12 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
 func (l *Loop) driveTurn(ctx context.Context, sessionID string, msgs []model.Message, startStep int, res *Result) (*Result, error) {
 	toolState := &toolLoopState{prevFP: make(map[string]bool)}
 	step := startStep
+	// steered 累积本回合在**回合内边界**注入的用户消息（design/15 §2.4，裁决
+	// 13.4.5）。它们只并入每一步的模型可见视图，绝不写回 msgs —— checkpoint 快照
+	// 与后台记忆提取因此保持干净；又因为每个后续 step 都重新并入，注入内容对模型
+	// 而言始终是**同一个窗口**的一部分（一致性），后到的消息也绝不会顶掉旧窗口
+	// 已产出的内容（完整性，I1）。
+	var steered []model.Message
 	// 未落定退出的回合不留下「可恢复的在途态」（裁决 13.4.5：下一条消息必须
 	// 看到自己的最新上下文，而不是被放弃回合的旧窗口顶掉）。判别口径（主人
 	// 2026-10-04 裁决，「只有崩溃才允许恢复」）：**崩溃**时本 defer 根本不执行，
@@ -111,7 +117,15 @@ func (l *Loop) driveTurn(ctx context.Context, sessionID string, msgs []model.Mes
 		// 然后为本轮组装模型可见上下文（全新记忆
 		// 注入 + transform 钩子），使每一步都看到最新上下文。
 		msgs = l.compactIfNeeded(ctx, msgs)
+		// 回合内边界（design/15 §2.0 I3）：上一个模型步与它那一批工具调用都已结束、
+		// 下一次模型调用还没开始 —— 此刻没有任何工具在飞，因此在这里注入不可能取消
+		// 任何已发起的工具；新到的用户消息则对**同一回合**的下一个模型步立即可见
+		//（design/15 §2.4 最新性）。循环绝不因此新开第二个回合（I1）。
+		steered = l.steerAtBoundary(ctx, sessionID, step, steered)
 		view := l.assembleContext(ctx, msgs, step)
+		if len(steered) > 0 {
+			view = append(view, steered...)
+		}
 
 		rc, roundSpan := l.span(ctx, "loop.step", trace.SpanKindStateTransition)
 		outcome, err := l.step(rc, sessionID, view)

internal/runtime/loop/steer.go

@@ -0,0 +1,103 @@
+package loop
+
+import (
+	"context"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
+)
+
+// 本文件实现「回合内转向注入」（design/15 §2.0/§2.4，裁决 13.4.5 目标态）。
+//
+// 不变量按 design/15 §2.0 实现，语义不在循环里自创：
+//
+//   - **I1（同会话在途回合数恒为 1）**：转向只在**同一回合的边界**注入，绝不取消
+//     当前回合、也绝不新开第二个回合。循环对这一点的保证是结构性的——它没有任何
+//     「起新回合」的分支，注入只是把消息并进当前回合下一次模型调用的视图。
+//   - **I2（消息恰好被处理一次）**：注入＝已消费。循环看不到消息 id
+//     （model.Message 没有 id 字段），因此「按 created_at + id 去重」与「从待处理
+//     队列移除」都归 SteerSource 的实现方（网关会话编排器，它同时持有转录与
+//     pending 队列）。循环只保证同一个边界只问一次、按返回顺序追加。
+//   - **I3（转向不取消已发起的工具）**：注入点只在**回合内边界**——一次模型步结束
+//     与它那一批工具调用结束**之后**、下一次模型调用**之前**。此刻没有任何工具在
+//     飞，因此注入不可能取消任何已发起的工具；正在执行的工具永远跑完（与 Cline
+//     PR #13958「只打断在飞的模型请求、工具与整轮继续」同口径）。
+//
+// 为什么注入只进**本轮视图**而不写回运行窗口：写回会让注入内容进入 checkpoint
+// 快照与后台记忆提取的输入面（两者都读该回合的 msgs），而它们本应只看到该回合真正
+// 的输入与工具结果。注入在每一步重新并入视图，因此对模型而言它仍是**同一个窗口**
+// （一致性＝同一窗口，design/15 §2.4），只是不落盘、不出现在记忆面。
+
+// SteerRequest 是循环在回合内边界向 SteerSource 发起的一次「重读转录」请求。
+type SteerRequest struct {
+	// SessionID 是本次 Run 驱动的会话。
+	SessionID string
+	// Step 是**即将开始**的模型步序号（从 1 开始）：被返回的消息会出现在该步的
+	// 模型可见视图里，因此它就是网关 `steer_ack` 帧上报的 `atStep`。
+	Step int
+}
+
+// SteerSource 由调用方（网关会话编排器）提供，在指定的回合内边界重读该会话转录，
+// 返回「本回合开始后新到、且尚未被本回合吃进」的用户消息，供循环追加进本轮视图。
+// 返回空切片表示本次边界没有可注入的消息，是正常情形而非错误。
+//
+// 实现方负责去重与「已消费」口径（见文件头 I2）：循环按返回顺序原样追加，不做也
+// 无法做 id 去重。它必须是非阻塞的——它在 Run 路径上被调用；网络/存储读取有界，
+// 读失败时返回空（该消息留在待处理队列，照旧作为自己的回合被回答，绝不静默丢失）。
+type SteerSource interface {
+	Steer(ctx context.Context, req SteerRequest) []model.Message
+}
+
+// steerSourceKey 是回合内转向源的 context key。
+type steerSourceKey struct{}
+
+// WithSteerSource 把转向源挂到 ctx 上，使本次 Run 的每个回合内边界都能重读转录并
+// 注入新到的用户消息（design/15 §2.4）。调用方（网关编排器）在**每个回合启动之前**
+// 挂一份新的源：转向游标是按回合的状态，跨回合复用会让上一个回合已吃进的消息被
+// 过滤掉或重复注入。
+//
+// nil 源是空操作，ctx 原样返回——未接线的调用方（子代理、后台思考、未接网关的循环）
+// 行为逐字不变。
+func WithSteerSource(ctx context.Context, src SteerSource) context.Context {
+	if ctx == nil || src == nil {
+		return ctx
+	}
+	return context.WithValue(ctx, steerSourceKey{}, src)
+}
+
+// SteerSourceFrom 返回 ctx 上挂的转向源，未挂载时为 nil。它是 WithSteerSource 的
+// 读侧，供拿到 context 的调用方（网关测试、断言注入边界的实现）观察本次 Run 的
+// 注入能力；nil 表示本次运行不做任何转向注入。
+func SteerSourceFrom(ctx context.Context) SteerSource {
+	if ctx == nil {
+		return nil
+	}
+	src, _ := ctx.Value(steerSourceKey{}).(SteerSource)
+	return src
+}
+
+// steerAtBoundary 在一个回合内边界重读该会话转录，并把新到的用户消息合并进本回合
+// 已注入列表。未接线转向源、ctx 已取消、或本次边界无新消息时，它原样返回 steered
+// （无新消息是常态：工具批次之间通常没有用户插话）。
+//
+// 它绝不返回错误：转向是增益，任何一个边界的读取失败都不该让用户已经在跑的回合
+// 失败——失败时返回空（见 SteerSource 契约），该消息留在待处理队列里等下一个回合。
+func (l *Loop) steerAtBoundary(ctx context.Context, sessionID string, step int, steered []model.Message) []model.Message {
+	src := SteerSourceFrom(ctx)
+	if src == nil || ctx.Err() != nil {
+		return steered
+	}
+	msgs := src.Steer(ctx, SteerRequest{SessionID: sessionID, Step: step})
+	if len(msgs) == 0 {
+		return steered
+	}
+	// 注入的图片消息走与回合开始处同一条视觉改写（design/06 §0）：否则一条中途
+	// 注入的图片会绕过 resolver，直接落到不支持视觉的主模型面前。
+	msgs = l.resolveImages(ctx, msgs)
+	// 链路图如实标记「本轮输入被中途追加过」（design/15 §5）。子代理 Run 不拥有
+	// 自己的图（它的 span 并入派生它的会话），因此分支 Run 不打这个标记。
+	if _, isBranch := flowBranch(ctx); !isBranch {
+		l.flow.MarkInputSteered(sessionID)
+	}
+	l.log.Info("steer injected", "session", sessionID, "step", step, "messages", len(msgs))
+	return append(steered, msgs...)
+}

internal/runtime/loop/steer_test.go

@@ -0,0 +1,395 @@
+package loop
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
+)
+
+// 本文件锁住裁决 13.4.5 的**目标态**在 loop 侧的三条不变量（design/15 §2.0/§2.4）：
+//
+//	I1 同会话在途回合数恒为 1：忙时新消息在**下一个回合内边界**被吃进**同一回合**
+//	   （模型在该步看到的窗口里有它），循环绝不因此新开第二个回合；
+//	I2 恰好被处理一次：注入只并进本轮视图，不写回运行窗口 —— checkpoint 快照与
+//	   记忆提取面因此看不到它，它也不会被当成新输入再答一遍；
+//	I3 转向不取消已发起的工具：注入点只在**一批工具调用结束之后**、下一次模型调用
+//	   之前；正在执行的工具跑完，其结果仍在窗口里。
+//
+// 这些测试用真的 mockAdapter/mockTool 装配真循环，断言的是**模型实际收到的消息**
+// 与**真的落下的 checkpoint/提取面**，不是内部字段的回读。
+
+// steerTrace 是工具与转向源共享的**加锁**事件序：I3 要证明的「工具跑完 → 边界注入」
+// 这个先后关系，在 Windows 上无法用 time.Now() 证明（粒度粗到两者拿到同一时间戳），
+// 只能靠共享的单调事件序；加锁是为了让 -race 也保持干净。
+type steerTrace struct {
+	mu     sync.Mutex
+	events []string
+}
+
+func (t *steerTrace) add(ev string) {
+	if t == nil {
+		return
+	}
+	t.mu.Lock()
+	t.events = append(t.events, ev)
+	t.mu.Unlock()
+}
+
+func (t *steerTrace) snapshot() []string {
+	if t == nil {
+		return nil
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return append([]string(nil), t.events...)
+}
+
+// steerScript 是一份脚本化转向源：按 step 返回要注入的消息，并记录被问到的边界。
+type steerScript struct {
+	mu       sync.Mutex
+	at       map[int][]model.Message
+	steps    []int
+	returned [][]model.Message
+	// trace 非 nil 时把「真的注入了」记进共享事件序。
+	trace *steerTrace
+}
+
+func newSteerScript(at map[int][]model.Message) *steerScript {
+	return &steerScript{at: at}
+}
+
+// withTrace 让本转向源把「真的注入了」记进共享事件序。
+func (s *steerScript) withTrace(tr *steerTrace) *steerScript {
+	s.trace = tr
+	return s
+}
+
+func (s *steerScript) Steer(_ context.Context, req SteerRequest) []model.Message {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.steps = append(s.steps, req.Step)
+	msgs := s.at[req.Step]
+	if len(msgs) == 0 {
+		return nil
+	}
+	s.returned = append(s.returned, msgs)
+	s.trace.add("steer:" + messageText(msgs[0]))
+	return msgs
+}
+
+// askedSteps 返回转向源被问到的边界序号序列。
+func (s *steerScript) askedSteps() []int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]int(nil), s.steps...)
+}
+
+// boundaryTool 是一个记录自己是否完整跑完、以及 ctx 是否在运行中被取消的工具，
+// 供 I3 断言「注入没有取消已发起的工具」。
+type boundaryTool struct {
+	mu    sync.Mutex
+	calls int
+	// canceled 记录这次调用是否因 ctx 在运行中被取消而中断。
+	canceled bool
+	delay    time.Duration
+	// trace 非 nil 时在完整跑完之际记下 "tool:done"。
+	trace *steerTrace
+}
+
+func (b *boundaryTool) Name() string        { return "slow" }
+func (b *boundaryTool) Description() string { return "slow tool for steering tests" }
+func (b *boundaryTool) Parameters() (map[string]any, error) {
+	return nil, nil
+}
+
+func (b *boundaryTool) Execute(ctx context.Context, _ any) (any, error) {
+	if b.delay > 0 {
+		select {
+		case <-time.After(b.delay):
+		case <-ctx.Done():
+			b.mu.Lock()
+			b.canceled = true
+			b.mu.Unlock()
+			return nil, ctx.Err()
+		}
+	}
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.calls++
+	b.trace.add("tool:done")
+	return "slow-done", nil
+}
+
+// counts 返回该工具被完整执行的次数与被取消的次数。
+func (b *boundaryTool) counts() (calls int, canceled bool) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.calls, b.canceled
+}
+
+// steerTestCall 是一次「一步工具 + 一步收尾」的脚本：第 1 步请求工具调用，第 2 步
+// 给出最终答复。它是 I1/I3 的公共前置（一个回合至少要有两条模型步，才存在边界）。
+const steerTestToolCall = `{"tool_calls":[{"name":"slow","args":{}}]}`
+
+// steerTestAdapter 返回一个两步脚本的适配器与它的工具。
+func steerTestAdapter(tool *boundaryTool) *mockAdapter {
+	return &mockAdapter{chat: []mgResp{
+		{resp: model.Response{Content: steerTestToolCall}},
+		{resp: model.Response{Content: "两步收尾"}},
+	}}
+}
+
+// TestRun_SteerInjectedAtStepBoundary_SingleTurn 钉住 I1 与 §2.4 的最新性：忙时新消息
+// 在**下一个 step 边界**被吃进同一回合 —— 第 2 步的模型窗口里有它（且排在工具结果
+// 之后），而整轮只有 2 次模型调用、只有 1 个回合结果。
+func TestRun_SteerInjectedAtStepBoundary_SingleTurn(t *testing.T) {
+	t.Parallel()
+	tool := &boundaryTool{}
+	ad := steerTestAdapter(tool)
+	td := dispatcherWith(t, tool)
+	l := New(Options{Agent: newTestAgent(t, ad, td), Tools: td, MaxRounds: 8})
+
+	src := newSteerScript(map[int][]model.Message{
+		2: {model.TextMessage(model.RoleUser, "忙时新消息")},
+	})
+	res, err := l.Run(WithSteerSource(WithSessionEnv(context.Background(), envOwner()), src), "sess-steer", []model.Message{model.TextMessage(model.RoleUser, "第一条")})
+	if err != nil {
+		t.Fatalf("Run: %v", err)
+	}
+
+	// I1：只有一个回合（Steps=2 的两步同一回合），不是「注入后又开了一回合」。
+	if res.Steps != 2 || res.StopReason != agent.StopReasonStop {
+		t.Fatalf("result = %+v, want a single 2-step turn ending with stop", res)
+	}
+	if len(ad.reqs) != 2 {
+		t.Fatalf("模型调用 %d 次, want 2（一次工具步 + 一次收尾步；注入绝不新开回合）", len(ad.reqs))
+	}
+	// 边界：每个 step 组装前问一次，正好 [1 2]。
+	if got := src.askedSteps(); len(got) != 2 || got[0] != 1 || got[1] != 2 {
+		t.Fatalf("转向边界 = %v, want [1 2]", got)
+	}
+
+	step2 := ad.reqs[1].Messages
+	last := step2[len(step2)-1]
+	if last.Role != model.RoleUser || messageText(last) != "忙时新消息" {
+		t.Fatalf("第 2 步窗口末条 = %+v, want 注入的忙碌消息（它必须对同一回合的下一个模型步立即可见）", last)
+	}
+	// 旧窗口已产出的内容（工具结果）完好：注入是追加，不是替换。
+	if !windowContainsRoleText(step2, model.RoleTool, "slow-done") {
+		t.Fatalf("第 2 步窗口丢了工具结果（注入不得顶掉旧窗口内容）：%+v", windowTexts(step2))
+	}
+	// 第一条用户消息仍在窗口里（同一窗口，不是新回合）。
+	if !windowContainsRoleText(step2, model.RoleUser, "第一条") {
+		t.Fatalf("第 2 步窗口丢了本轮第一条用户消息：%+v", windowTexts(step2))
+	}
+}
+
+// TestRun_SteerDoesNotCancelInFlightTools 钉住 I3：注入发生在**一批工具调用结束之后**
+// —— 工具完整跑完（未被取消），其结果在窗口里，且注入**在事件序上晚于**工具的完成
+// （Windows 时间戳粒度粗到两者同值，故用共享事件序而不是时间比较）。
+func TestRun_SteerDoesNotCancelInFlightTools(t *testing.T) {
+	t.Parallel()
+	tr := &steerTrace{}
+	tool := &boundaryTool{delay: 20 * time.Millisecond, trace: tr}
+	ad := steerTestAdapter(tool)
+	td := dispatcherWith(t, tool)
+	l := New(Options{Agent: newTestAgent(t, ad, td), Tools: td, MaxRounds: 8})
+
+	src := newSteerScript(map[int][]model.Message{
+		2: {model.TextMessage(model.RoleUser, "工具跑着的时候我插一句")},
+	}).withTrace(tr)
+	res, err := l.Run(WithSteerSource(context.Background(), src), "sess-tools", []model.Message{model.TextMessage(model.RoleUser, "去跑那个慢工具")})
+	if err != nil {
+		t.Fatalf("Run: %v", err)
+	}
+
+	calls, canceled := tool.counts()
+	if calls != 1 || canceled {
+		t.Fatalf("工具 calls=%d canceled=%v, want 1/false（转向绝不取消已发起的工具）", calls, canceled)
+	}
+	if res.ToolCallsExecuted != 1 {
+		t.Fatalf("ToolCallsExecuted = %d, want 1", res.ToolCallsExecuted)
+	}
+	// 事件序必须恰好是「工具跑完」→「边界注入」：注入点落在工具批次**之后**。
+	if got := tr.snapshot(); len(got) != 2 || got[0] != "tool:done" || got[1] != "steer:工具跑着的时候我插一句" {
+		t.Fatalf("事件序 = %v, want [tool:done steer:…]（注入必须在工具批次结束之后的边界上）", got)
+	}
+	// 注入只能发生在边界上：整个回合的边界恰好是 [1 2]，没有任何「批次中途」的调用。
+	if got := src.askedSteps(); len(got) != 2 || got[0] != 1 || got[1] != 2 {
+		t.Fatalf("转向边界 = %v, want [1 2]（批次中途绝不注入）", got)
+	}
+}
+
+// TestRun_SteerDoesNotPolluteCheckpointOrExtraction 钉住 I2 的另一半：注入只进本轮
+// 视图，**不进**该回合的运行窗口 —— checkpoint 快照与后台记忆提取的输入面里都不该
+// 出现它，而该回合真正的输入与工具结果必须在。
+func TestRun_SteerDoesNotPolluteCheckpointOrExtraction(t *testing.T) {
+	t.Parallel()
+	tool := &boundaryTool{}
+	ad := steerTestAdapter(tool)
+	td := dispatcherWith(t, tool)
+	cp := newTestCheckpointStore(t)
+	extractor := &knowledgeCaptureExtractor{}
+	l := New(Options{Agent: newTestAgent(t, ad, td), Tools: td, MaxRounds: 8, Checkpoint: cp, Memory: newTestMemoryStore(t), Extractor: extractor})
+
+	const busy = "忙时插进来的一句"
+	src := newSteerScript(map[int][]model.Message{2: {model.TextMessage(model.RoleUser, busy)}})
+	if _, err := l.Run(WithSteerSource(context.Background(), src), "sess-cp", []model.Message{model.TextMessage(model.RoleUser, "第一条")}); err != nil {
+		t.Fatalf("Run: %v", err)
+	}
+
+	// 模型确实看到了它（否则这条断言就没有意义）。
+	if !windowContainsRoleText(ad.reqs[len(ad.reqs)-1].Messages, model.RoleUser, busy) {
+		t.Fatal("测试前提不成立：注入的消息没有出现在模型窗口里")
+	}
+
+	latest, err := cp.LoadLatest(context.Background(), "sess-cp")
+	if err != nil {
+		t.Fatalf("LoadLatest: %v", err)
+	}
+	var st checkpointState
+	if err := json.Unmarshal(latest.State, &st); err != nil {
+		t.Fatalf("unmarshal checkpoint: %v", err)
+	}
+	if msgInAny(st.Messages, busy) {
+		t.Fatalf("注入的消息进了 checkpoint 快照（只应影响本轮视图）：%+v", windowTexts(st.Messages))
+	}
+	if !windowContainsRoleText(st.Messages, model.RoleTool, "slow-done") {
+		t.Fatalf("快照里丢了工具结果：%+v", windowTexts(st.Messages))
+	}
+	if len(extractor.saw) == 0 {
+		t.Fatal("测试前提不成立：提取器没有被调用")
+	}
+	if msgInAny(extractor.saw, busy) {
+		t.Fatalf("注入的消息进了记忆提取面（会被写成记忆）：%+v", windowTexts(extractor.saw))
+	}
+}
+
+// TestRun_SteerRestrictedSessionStaysRestricted 是红线回归：群聊/陌生人会话**不因
+// 转向注入**而拿到主人身份知识（design/11 §0）—— 注入的只是那条用户消息，人格知识
+// 块仍按会话环境门控（主人会话有、受限会话一个字都没有），且注入不改变本轮 env。
+func TestRun_SteerRestrictedSessionStaysRestricted(t *testing.T) {
+	t.Parallel()
+	for _, tc := range []struct {
+		name     string
+		env      SessionEnv
+		wantKnow bool
+	}{
+		{name: "主人会话", env: envOwner(), wantKnow: true},
+		{name: "群聊", env: envGroup(), wantKnow: false},
+		{name: "陌生人私聊", env: envStranger(), wantKnow: false},
+	} {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			tool := &boundaryTool{}
+			ad := steerTestAdapter(tool)
+			td := dispatcherWith(t, tool)
+			l := New(Options{Agent: newTestAgent(t, ad, td), Tools: td, MaxRounds: 8, PersonaKnowledgeFor: gateKnowledge})
+
+			src := newSteerScript(map[int][]model.Message{2: {model.TextMessage(model.RoleUser, "群里插一句")}})
+			ctx := WithSteerSource(WithSessionEnv(context.Background(), tc.env), src)
+			if _, err := l.Run(ctx, "sess-env", []model.Message{model.TextMessage(model.RoleUser, "第一条")}); err != nil {
+				t.Fatalf("Run: %v", err)
+			}
+			step2 := ad.reqs[1].Messages
+			if !windowContainsRoleText(step2, model.RoleUser, "群里插一句") {
+				t.Fatal("测试前提不成立：注入的消息没有出现在模型窗口里")
+			}
+			if got := knowledgeIn(step2); got != tc.wantKnow {
+				t.Fatalf("身份知识块在窗口里 = %v, want %v（注入不得绕过会话门）", got, tc.wantKnow)
+			}
+			if env := SessionEnvFrom(ctx); env != tc.env {
+				t.Fatalf("本轮 env 被注入改变了：%+v, want %+v", env, tc.env)
+			}
+		})
+	}
+}
+
+// TestRun_SteerAbsent_ViewUnchanged 钉住向后兼容：没有转向源的 Run，模型窗口与接线
+// 之前逐条一致（前导系统块 + 对话 + 工具结果），绝无多余消息。
+func TestRun_SteerAbsent_ViewUnchanged(t *testing.T) {
+	t.Parallel()
+	tool := &boundaryTool{}
+	ad := steerTestAdapter(tool)
+	td := dispatcherWith(t, tool)
+	l := New(Options{Agent: newTestAgent(t, ad, td), Tools: td, MaxRounds: 8})
+
+	if _, err := l.Run(context.Background(), "sess-plain", []model.Message{model.TextMessage(model.RoleUser, "第一条")}); err != nil {
+		t.Fatalf("Run: %v", err)
+	}
+	step2 := ad.reqs[1].Messages
+	if len(step2) != 3 {
+		t.Fatalf("未接线转向源时第 2 步窗口 = %d 条 (%v), want 3（原用户消息 + 助手工具调用轮 + 工具结果）",
+			len(step2), windowTexts(step2))
+	}
+	if msgInAny(step2, "忙时新消息") {
+		t.Fatalf("未接线却出现了注入内容：%+v", windowTexts(step2))
+	}
+}
+
+// TestFlowPublisher_MarkInputSteered 钉住 design/15 §5 的 flow 标记：注入时给 input
+// 节点打 steered:true，且**不新建节点**（图仍是一条主干，与 I1 一致）。
+func TestFlowPublisher_MarkInputSteered(t *testing.T) {
+	t.Parallel()
+	var mu sync.Mutex
+	var published []any
+	p := NewFlowPublisher(func(_ string, payload any) {
+		mu.Lock()
+		published = append(published, payload)
+		mu.Unlock()
+	}, nil)
+	p.StartRound("web", "第一条")
+	p.MarkInputSteered("web")
+
+	mu.Lock()
+	last := published[len(published)-1]
+	mu.Unlock()
+	graph, ok := last.(*trace.FlowGraph)
+	if !ok {
+		t.Fatalf("published payload = %T", last)
+	}
+	if len(graph.Nodes) != 1 {
+		t.Fatalf("节点数 = %d, want 1（标记只并入既有 input 节点）", len(graph.Nodes))
+	}
+	got := graph.Nodes[0]
+	if got.ID != "input" || got.Props["steered"] != true || got.Label != "第一条" {
+		t.Fatalf("input 节点 = %+v, want 保留标签并带上 steered:true", got)
+	}
+}
+
+// windowTexts 把一份模型窗口渲染为 role:文本 列表，供失败信息定位。
+func windowTexts(msgs []model.Message) []string {
+	out := make([]string, 0, len(msgs))
+	for _, m := range msgs {
+		out = append(out, string(m.Role)+":"+messageText(m))
+	}
+	return out
+}
+
+// windowContainsRoleText 报告窗口里是否存在该角色、且文本包含 want 的消息。
+func windowContainsRoleText(msgs []model.Message, role model.Role, want string) bool {
+	for _, m := range msgs {
+		if m.Role == role && strings.Contains(messageText(m), want) {
+			return true
+		}
+	}
+	return false
+}
+
+// msgInAny 报告任何一条消息的文本里是否含 probe。
+func msgInAny(msgs []model.Message, probe string) bool {
+	for _, m := range msgs {
+		if strings.Contains(messageText(m), probe) {
+			return true
+		}
+	}
+	return false
+}

internal/services/gateway/session_events.go

@@ -0,0 +1,28 @@
+package gateway
+
+// 本文件归属编排器推给前端的每会话状态事件类型（design/15 §5/§2.1）。
+// 它们从 session_orchestrator.go 拆出，使编排器本体只留消息状态机，并保持在
+// 「单文件 ≤500 行」的硬约束内。帧名即契约：客户端按 type 分发，因此这些取值
+// 绝不随重构改名（要改就是契约变更，须先登记）。
+//
+// 这里只放**状态机信号**（不含消息内容）。带内容的帧（message.*、steer_ack、
+// chat_duplicate 等）由 design/11 §7.1 的帧门按 fail-closed 默认档只投给
+// owner + admin，见 ws_frame_scope.go。
+
+// 经 WS 通道推给前端的状态事件类型。每个的 data map 都
+// 携带 sessionId，使客户端能按会话关联状态。
+const (
+	EventSessionBusy      = "session_busy"
+	EventPendingAck       = "pending_ack"
+	EventPendingDrain     = "pending_drain"
+	EventSessionIdle      = "session_idle"
+	EventSessionInterrupt = "session_interrupted"
+	EventPendingFull      = "pending_full"
+	// EventChatDuplicate 是一次重复到达的明确应答（G4 幂等）：本条 clientMsgId
+	// 已有落库的消息，本次没有新回合、也不会有新帧。它的 data 携带
+	// {sessionId, clientMsgId, messageId}，让客户端把本地乐观气泡对齐到既有
+	// 消息；没有发过这个 clientMsgId 的其他客户端自行忽略即可。它不是状态机
+	// 迁移，因此不在 design/11 §7.1 的「状态信号」白名单里 —— 帧门默认只把它
+	// 投给 owner/admin（fail-closed）。
+	EventChatDuplicate = "chat_duplicate"
+)

internal/services/gateway/session_orchestrator.go

@@ -43,23 +43,8 @@ const (
 	ModeSubagent Mode = "subagent"
 )
 
-// 经 WS 通道推给前端的状态事件类型。每个的 data map 都
-// 携带 sessionId，使客户端能按会话关联状态。
-const (
-	EventSessionBusy      = "session_busy"
-	EventPendingAck       = "pending_ack"
-	EventPendingDrain     = "pending_drain"
-	EventSessionIdle      = "session_idle"
-	EventSessionInterrupt = "session_interrupted"
-	EventPendingFull      = "pending_full"
-	// EventChatDuplicate 是一次重复到达的明确应答（G4 幂等）：本条 clientMsgId
-	// 已有落库的消息，本次没有新回合、也不会有新帧。它的 data 携带
-	// {sessionId, clientMsgId, messageId}，让客户端把本地乐观气泡对齐到既有
-	// 消息；没有发过这个 clientMsgId 的其他客户端自行忽略即可。它不是状态机
-	// 迁移，因此不在 design/11 §7.1 的「状态信号」白名单里 —— 帧门默认只把它
-	// 投给 owner/admin（fail-closed）。
-	EventChatDuplicate = "chat_duplicate"
-)
+// 经 WS 通道推给前端的状态事件类型见 session_events.go（它们从本文件拆出，
+// 使编排器本体保持在「单文件 ≤500 行」的硬约束内）。
 
 const (
 	// defaultMaxPending 限定每会话的待处理队列（design/15 §4）。
@@ -153,6 +138,10 @@ type sessionState struct {
 	// clientMsgOrder 是 clientMsgs 的插入顺序，用于有界淘汰
 	// （maxClientMsgsPerSession）：淘汰只丢快路径，存储仍是权威。
 	clientMsgOrder []string
+	// pendingSeq 是待处理队列的单调递增序号源（design/15 §2.4 的转向认领键）。
+	// 队列是一条只在队尾追加、可能被整体弹出/清空的先进先出队列，因此「第几条」
+	// 不稳定；按 seq 认领已注入的消息，才不会被同一时刻的追加或清空误伤。
+	pendingSeq uint64
 }
 
 // pendingMsg 是一条已排队用户回合。它携带自己的 env，使续跑
@@ -168,6 +157,10 @@ type pendingMsg struct {
 	deviceID string
 	mode     string
 	at       time.Time
+	// seq 是这条待处理回合在**本会话**队列里的单调递增序号（design/15 §2.4）。
+	// 它只在状态锁下分配，是转向注入「注入＝已消费」认领的依据：队列会被追加、
+	// 被弹出、被中断清空，序号却能唯一定位这一条。
+	seq uint64
 }
 
 // Orchestrator 拥有每会话的消息状态机。它对并发安全
@@ -269,7 +262,8 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes
 	// 加入该字段之前逐字相同。
 	ctx = withClientMsgHint(ctx, a.hint)
 	if st.mode != ModeIdle {
-		st.pending = append(st.pending, pendingMsg{messages: messages, env: env, deviceID: deviceID, mode: msgMode, at: time.Now()})
+		st.pendingSeq++
+		st.pending = append(st.pending, pendingMsg{messages: messages, env: env, deviceID: deviceID, mode: msgMode, at: time.Now(), seq: st.pendingSeq})
 		depth := len(st.pending)
 		mode := st.mode
 		o.mu.Unlock()
@@ -299,6 +293,11 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes
 	st.cancel = cancel
 	o.mu.Unlock()
 
+	// 本回合的转向源（design/15 §2.4）：loop 在**回合内边界**回调它重读转录，把忙时
+	// 新到、已入 pending 的用户消息并入**同一回合**的下一次模型调用（I1：绝不新开
+	// 第二回合）。它挂在 runCtx 上，因此只作用于本回合；回合结算或被打断后随 ctx 消亡。
+	runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, gen))
+
 	o.emitState(sessionID, EventSessionBusy, map[string]any{"sessionId": sessionID, "mode": ModeThinking})
 	// 只有此刻到达的消息会被持久化：已播种的转录
 	// 已经持久，重新追加它会重复整个对话。
@@ -372,6 +371,9 @@ func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Resu
 		gen = st.gen
 		runCtx, cancel := context.WithCancel(context.Background())
 		st.cancel = cancel
+		// 续跑回合同样带自己的转向源（design/15 §2.4）：严格 FIFO 弹出下一条之后，
+		// 期间**仍在 pending** 的后到消息会在本回合的下一个边界被注入，而不是各起一回合。
+		runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, gen))
 		// 本回合的这次 Run 已经返回（我们正处在它的结算里），因此被续跑
 		// 的回合无需再等任何在途回合：围栏只负责放行它后面的那一个。
 		fence := newTurnFence(nil)

internal/services/gateway/session_steer.go

@@ -0,0 +1,236 @@
+package gateway
+
+import (
+	"context"
+	"log/slog"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
+)
+
+// 本文件实现「忙时新消息在回合内边界注入同一回合」的网关侧
+// （design/15 §2.0/§2.1/§2.4/§5，裁决 13.4.5 目标态）。它是 loop.SteerSource
+// 的生产实现，与 loop/steer.go 的调用点合成一条链：
+//
+//	loop 在每个回合内边界（一批工具调用结束之后、下一次模型调用之前）
+//	  → 本文件重读该会话转录，把已落库、且在本会话 pending 里的用户消息交出去
+//	  → 循环把它们并入**同一回合**的下一份模型视图（绝不新开第二回合，I1）
+//	  → 本文件把这些消息从 pending 移除并下发 steer_ack（I2：注入＝已消费）
+//
+// 候选集必须**同时**满足两个条件，缺一不可：
+//
+//   - **在转录里**（history.Recent 重读）＝最新性与可去重：消息真的落库了，我们才
+//     能拿到它的权威 id 与 created_at。模型消息本身没有 id 字段，因此「按
+//     created_at + id 去重」只能在持有转录的这一侧做（design/15 §2.4）。
+//   - **在 pending 里**＝恰好一次（I2）：只有本编排器排队过的消息才允许被本回合
+//     接手；注入后立刻从 pending 移除，因此它既不会被第二个回合再答一遍，也不会丢。
+//     由别的写入方（IM 适配器自有路径）落库、本编排器从未排队的消息**保持不注入**
+//     —— 那条路径正在自己作答它，注入会让同一条消息被答两次。
+//
+// 未接历史存储、读转录失败、消息尚未落库、或本回合的世代已被打断/新回合取代时一律
+// 返回空（该消息留在 pending，照旧作为自己的回合被回答）：转向是增益，任何一处失败
+// 都不该让用户已经在跑的回合失败，也绝不该让一条消息静默消失。
+
+// steerLog 返回标记来源 gateway/session-steer 的 logger（design/15 转向链的日志
+// 能按子功能定位，而不是混在 gateway 的裸日志里）。
+func steerLog(o *Orchestrator) *slog.Logger {
+	return logger.Scoped(o.log, source, "session-steer")
+}
+
+// steerTurn 是一个在飞回合的转向游标。**每个回合一份**，绝不跨回合复用：它记录该
+// 回合已吃进的消息 id（I2 的去重键）与回合世代号（被中断/被新回合取代后立即失效）。
+//
+// 它只在回合自己的 goroutine 上被调用（loop 的边界调用是同步的），因此自身的
+// consumed 无需加锁；pending 的读写与世代校验则一律走编排器互斥锁。
+type steerTurn struct {
+	orchestrator *Orchestrator
+	sessionID    string
+	gen          uint64
+	consumed     map[string]bool
+}
+
+// beginSteer 为一个刚启动的回合建一份转向游标。编排器把它挂到**该回合自己的 ctx**
+// 上（loop.WithSteerSource），因此每个回合、每次续跑各有一份独立游标，上一个回合
+// 已吃进的消息绝不会被下一个回合重复注入或漏掉。
+func (o *Orchestrator) beginSteer(sessionID string, gen uint64) *steerTurn {
+	return &steerTurn{
+		orchestrator: o,
+		sessionID:    sessionID,
+		gen:          gen,
+		consumed:     make(map[string]bool),
+	}
+}
+
+// steerCandidate 是本次边界要注入的一条消息：它的 pending 队列序号（移除认领用）
+// 与它已落库的转录行（id/created_at/内容/说话人归属都从行上来）。
+type steerCandidate struct {
+	seq uint64
+	row history.Message
+}
+
+// Steer 是 loop.SteerSource 的实现（design/15 §2.4）：在一次回合内边界重读该会话
+// 转录，返回本次要注入同一回合的用户消息，并把它们从 pending 移除（注入＝已消费）。
+// 它绝不返回错误——失败一律退化为「本次边界没有可注入的消息」。
+func (s *steerTurn) Steer(ctx context.Context, req loop.SteerRequest) []model.Message {
+	if s == nil || s.orchestrator == nil || ctx.Err() != nil {
+		return nil
+	}
+	sessionID := req.SessionID
+	if sessionID == "" {
+		sessionID = s.sessionID
+	}
+	o := s.orchestrator
+	reader, ok := o.history.(transcriptReader)
+	if !ok {
+		return nil
+	}
+	pending := o.pendingSnapshot(sessionID, s.gen)
+	if len(pending) == 0 {
+		return nil
+	}
+	// 每个边界都重读转录：设计要求的「最新性＝下一个边界即生效」正是靠这次重读，
+	// 而不是靠回合开始时的那一次播种（design/15 §2.4）。
+	recent, err := reader.Recent(ctx, sessionID, transcriptContextCount)
+	if err != nil {
+		steerLog(o).Debug("steer transcript read", "session", sessionID, "step", req.Step, "err", err)
+		return nil
+	}
+	candidates := s.match(pending, recent)
+	if len(candidates) == 0 {
+		return nil
+	}
+	// 认领失败（世代已被中断/被更新的顶层回合取代）时把这条消息留给那条新路径：
+	// 既不注入、也不发 steer_ack —— 否则它会被本回合与下一个回合各答一次（I2）。
+	if !o.consumePending(sessionID, s.gen, candidateSeqs(candidates)) {
+		return nil
+	}
+	out := make([]model.Message, 0, len(candidates))
+	for _, c := range candidates {
+		s.consumed[c.row.ID] = true
+		// steer_ack：该消息已被本回合吃进、尚未被正面回答（前端据此把它渲染成
+		// 灰态，design/15 §5）。它携带具体消息身份，因此不进 frameReachesAll 的
+		// 状态机信号白名单，落 fail-closed 默认档：只发 owner + admin。
+		o.emitState(sessionID, wsSteerAck, map[string]any{"messageId": c.row.ID, "atStep": req.Step})
+		out = append(out, transcriptMessage(c.row))
+	}
+	steerLog(o).Info("pending steered", "session", sessionID, "step", req.Step, "messages", len(out))
+	return out
+}
+
+// match 把 pending 里的用户消息与转录行配对（design/15 §2.4 的「按 created_at + id
+// 去重」）。它按 pending 的 FIFO 顺序（＝到达顺序，与转录的 created_at/rowid 顺序
+// 一致）逐条找**最旧的、尚未被认领的**同行；找不到（尚未落库、或落库失败、或该行
+// 已在本回合更早的边界被吃进）就跳过，让那条消息留在 pending 里等自己的回合。
+//
+// 用 (内容, 图片) 配对而不是 id 配对的原因：编排器不持有自己在 pending 路径上写入
+// 的行的 id（写入在持久化钩子内部完成，本回合不该为了拿到 id 去改那条路径的契约）。
+// 同一文本连发两条仍能正确配对——配对逐行认领、且已吃进的行记在 consumed 里，
+// 因此第二条只会配到第二行。
+func (s *steerTurn) match(pending []pendingMsg, recent []history.Message) []steerCandidate {
+	rows := make([]history.Message, 0, len(recent))
+	for _, r := range recent {
+		if r.Role != history.RoleUser || r.ID == "" || s.consumed[r.ID] {
+			continue
+		}
+		rows = append(rows, r)
+	}
+	if len(rows) == 0 {
+		return nil
+	}
+	claimed := make([]bool, len(rows))
+	out := make([]steerCandidate, 0, len(pending))
+	for _, p := range pending {
+		text, image := pendingUserContent(p)
+		if text == "" && image == "" {
+			continue
+		}
+		idx := -1
+		for i, r := range rows {
+			if !claimed[i] && r.Content == text && r.Image == image {
+				idx = i
+				break
+			}
+		}
+		if idx < 0 {
+			continue
+		}
+		claimed[idx] = true
+		out = append(out, steerCandidate{seq: p.seq, row: rows[idx]})
+	}
+	return out
+}
+
+// pendingUserContent 返回一条待处理消息里第一条用户消息的文本与图片，供与转录行
+// 配对。没有用户内容（理论上不可达：pending 只收非空用户回合）时返回两个空串。
+func pendingUserContent(p pendingMsg) (text, image string) {
+	for _, m := range p.messages {
+		if m.Role != model.RoleUser {
+			continue
+		}
+		if t, img := messageTextImage(m); t != "" || img != "" {
+			return t, img
+		}
+	}
+	return "", ""
+}
+
+// candidateSeqs 把候选消息收集为 pending 序号集合，供锁内一次性认领移除。
+func candidateSeqs(cs []steerCandidate) map[uint64]bool {
+	seqs := make(map[uint64]bool, len(cs))
+	for _, c := range cs {
+		seqs[c.seq] = true
+	}
+	return seqs
+}
+
+// pendingSnapshot 在状态锁下取该会话待处理队列的一份副本（世代不符时返回 nil：
+// 那份队列已不属于本回合，例如被中断清空或被更新的顶层回合取代）。取副本而不是
+// 持锁做转录读取，是为了不让有界存储读取堵住 Submit。
+func (o *Orchestrator) pendingSnapshot(sessionID string, gen uint64) []pendingMsg {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	st := o.states[sessionID]
+	if st == nil || st.gen != gen || len(st.pending) == 0 {
+		return nil
+	}
+	return append([]pendingMsg(nil), st.pending...)
+}
+
+// consumePending 在状态锁下把已注入的消息从待处理队列移除（I2：注入＝已消费，绝不
+// 被第二个回合再答一遍）。它按 pendingMsg.seq 认领，因此即便队列在同一时刻被追加
+// （只会加在队尾）或被中断清空，也绝不会误删别的消息。
+//
+// 返回 false 表示这次认领**没有生效**（队列里已无这些序号，或世代已被中断/更新的
+// 顶层回合取代）——调用方此时必须放弃注入：否则这条消息会被本回合与那条新路径各答
+// 一次。它必须在把消息交给循环**之前**调用。
+func (o *Orchestrator) consumePending(sessionID string, gen uint64, seqs map[uint64]bool) bool {
+	if len(seqs) == 0 {
+		return false
+	}
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	st := o.states[sessionID]
+	if st == nil || st.gen != gen {
+		return false
+	}
+	kept := make([]pendingMsg, 0, len(st.pending))
+	removed := make([]model.Message, 0, len(seqs))
+	for _, p := range st.pending {
+		if seqs[p.seq] {
+			removed = append(removed, p.messages...)
+			continue
+		}
+		kept = append(kept, p)
+	}
+	st.pending = kept
+	if len(removed) == 0 {
+		return false
+	}
+	// 被吃进的消息同时并入续跑尾部：本回合的回答紧随其后，因此若有下一条消息接着
+	// 被续跑，它看得到「这句问过、她在这里答了」，而不是一条无头的回答（design/15
+	// §2.2 的增量上下文续接）。尾部有界，绝不因此无限增长。
+	st.tail = o.boundedTail(concatMessages(st.tail, removed...))
+	return true
+}

internal/services/gateway/session_steer_test.go

@@ -0,0 +1,466 @@
+package gateway
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
+	capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool"
+	"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
+)
+
+// 本文件锁定裁决 13.4.5 目标态的**网关侧**（design/15 §2.0/§2.1/§2.4/§5）：
+//
+//	I1 忙时新消息在**下一个回合内边界**被注入**同一回合** —— 模型在该步看到的窗口
+//	   里有它，且整轮只有一次助手回答（不是「注入 + 再开一回合」）；
+//	I2 注入＝已消费 —— 那条消息从 pending 移除，绝不被第二个回合再答一遍；
+//	§5 steer_ack 让前端把插入气泡渲染成灰态，且它只发 owner + admin（fail-closed）。
+//
+// 断言走**真路径**：真 ws chat 帧 → 真编排器（真 pending 队列）→ **真 loop**
+// （真模型步、真工具批次、真转录重读）→ 真 SQLite history 存储 → 真 /ws/chat
+// 连接（真 JWT）。注入链路的每一环都是生产实现，只有模型适配器是脚本化的。
+
+// steerToolCallJSON 返回第 step 步请求的工具调用（每一步产生一次回合内边界：工具
+// 批次结束之后才会走下一步）。**参数按步变化**是刻意的：同一个工具调用跨连续步完全
+// 重复会命中循环自身的「关键工具循环安全网」而被跳过，那样就测不到两个真的工具批次。
+// 两步让测试能同时验证「边界注入」与「跨多个边界不重复注入」（I2）。
+func steerToolCallJSON(step int) string {
+	return fmt.Sprintf(`{"tool_calls":[{"name":"steer_slow","args":{"q":%d}}]}`, step)
+}
+
+// steerScriptedToolSteps 是脚本里请求工具调用的步数：第 1、2 步请求工具，第 3 步收尾。
+const steerScriptedToolSteps = 2
+
+// steerAdapter 是三步脚本的模型适配器：前两步请求工具调用，最后给出最终答复。
+// 它的第一次调用被一道门闸挡住，使测试能确定地在「回合在飞」时投递第二条消息
+// （否则第二条可能在第一个回合开始之前就到达，走的就不是忙碌分支了）。
+type steerAdapter struct {
+	mu       sync.Mutex
+	steps    int
+	reqs     []model.Request
+	entered  chan struct{}
+	enterOne sync.Once
+	gate     chan struct{}
+}
+
+func newSteerAdapter() *steerAdapter {
+	return &steerAdapter{entered: make(chan struct{}), gate: make(chan struct{})}
+}
+
+func (a *steerAdapter) Name() string { return "steer-script" }
+
+func (a *steerAdapter) Chat(ctx context.Context, req model.Request, _ model.Options) (model.Response, error) {
+	a.mu.Lock()
+	a.steps++
+	step := a.steps
+	a.mu.Unlock()
+	if step == 1 {
+		a.enterOne.Do(func() { close(a.entered) })
+		select {
+		case <-a.gate:
+		case <-ctx.Done():
+			return model.Response{}, ctx.Err()
+		}
+	}
+	a.mu.Lock()
+	a.reqs = append(a.reqs, req)
+	a.mu.Unlock()
+	if step <= steerScriptedToolSteps {
+		return model.Response{Content: steerToolCallJSON(step)}, nil
+	}
+	return model.Response{Content: "三步收尾"}, nil
+}
+
+func (a *steerAdapter) ChatStream(ctx context.Context, req model.Request, on func(model.StreamEvent)) (model.Response, error) {
+	return a.Chat(ctx, req, model.Options{})
+}
+
+// steps 返回已发生的模型步数。
+func (a *steerAdapter) stepCount() int {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	return a.steps
+}
+
+// request 返回第 n 次（0 起）模型调用所见到的窗口。
+func (a *steerAdapter) request(n int) model.Request {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	return a.reqs[n]
+}
+
+// enteredCh 暴露门闸前的「已进入第一次模型调用」信号。
+func (a *steerAdapter) enteredCh() <-chan struct{} { return a.entered }
+
+// release 放行第一次模型调用，使回合继续走到工具批次与第 2 步。
+func (a *steerAdapter) release() { close(a.gate) }
+
+// steerRecordingTool 记录自己被完整执行了多少次（I2 的旁证：一次回合里那些工具
+// 只该跑一次）。
+type steerRecordingTool struct {
+	mu    sync.Mutex
+	calls int
+}
+
+func (t *steerRecordingTool) Name() string        { return "steer_slow" }
+func (t *steerRecordingTool) Description() string { return "steering test tool" }
+func (t *steerRecordingTool) Parameters() (map[string]any, error) {
+	return nil, nil
+}
+
+func (t *steerRecordingTool) Execute(context.Context, any) (any, error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.calls++
+	return "steer-tool-done", nil
+}
+
+func (t *steerRecordingTool) count() int {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.calls
+}
+
+// newSteerDispatcher 装配一个注册了给定工具的 dispatcher。
+func newSteerDispatcher(t *testing.T, tools ...sdk.Tool) *tool.Dispatcher {
+	t.Helper()
+	d, err := tool.New(capreg.New())
+	if err != nil {
+		t.Fatalf("tool.New: %v", err)
+	}
+	for _, tk := range tools {
+		if err := d.Register(tk); err != nil {
+			t.Fatalf("register %s: %v", tk.Name(), err)
+		}
+	}
+	return d
+}
+
+// countHistoryRows 按角色统计该会话在真存储里的消息条数。
+func countHistoryRows(t *testing.T, store *history.Store, sessionID, role string) []history.Message {
+	t.Helper()
+	recent, err := store.Recent(context.Background(), sessionID, 50)
+	if err != nil {
+		t.Fatalf("history.Recent: %v", err)
+	}
+	out := make([]history.Message, 0, len(recent))
+	for _, m := range recent {
+		if m.Role == role {
+			out = append(out, m)
+		}
+	}
+	return out
+}
+
+// TestSteer_BusyMessageJoinsSameTurnAndIsNotAnsweredTwice 是本设计的核心端到端断言：
+// 一个真 loop 跑着两步回合时又到一条用户消息 —— 它在**下一个回合内边界**被吃进同一
+// 回合（模型第 2 步的窗口里有它），整轮**只有一次助手回答**，pending 被清空，且
+// steer_ack 只到达 owner + admin。
+func TestSteer_BusyMessageJoinsSameTurnAndIsNotAnsweredTwice(t *testing.T) {
+	t.Parallel()
+	// steer_ack 携带具体消息身份：它绝不属于「状态机信号」白名单（fail-closed）。
+	if frameReachesAll(wsSteerAck) {
+		t.Fatalf("%s reached every connection; steer_ack carries a message identity and must stay owner/admin only", wsSteerAck)
+	}
+
+	f := newSteerFixture(t)
+	orch := f.srv.hub.orchestrator
+	// 第一条消息：正常起一个回合，它停在第一次模型调用上。
+	writeChat(t, f.conns[auth.RoleOwner], `{"sessionId":"web","text":"第一条","deviceId":"web:test:steer"}`)
+	select {
+	case <-f.adapter.enteredCh():
+	case <-time.After(3 * time.Second):
+		t.Fatal("第一个回合没有进入模型调用")
+	}
+
+	// 忙时第二条：不阻塞、即刻落库、入 pending（这一句正是要在**本回合内**被吃进的）。
+	writeChat(t, f.conns[auth.RoleOwner], `{"sessionId":"web","text":"忙时新消息","deviceId":"web:test:steer"}`)
+	waitUntil(t, "第二条进入 pending", func() bool { return orch.QueueDepth("web") == 1 })
+	waitUntil(t, "两条用户消息都落库", func() bool {
+		return len(countHistoryRows(t, f.store, "web", history.RoleUser)) == 2
+	})
+
+	// 放行：第 1 步给出工具调用 → 工具批次结束 → 第 2 步前注入第二条消息 → 收尾。
+	f.adapter.release()
+	waitUntil(t, "会话排空（pending 已被吃进，不再续跑第二个回合）", func() bool {
+		return !orch.Busy("web") && orch.QueueDepth("web") == 0
+	})
+
+	assertSteerSameTurnWindow(t, f)
+	assertSteerExactlyOnce(t, f.store)
+	assertSteerAckFrames(t, f)
+}
+
+// steerFixture 是一次端到端转向场景的全部生产依赖（真 loop、真存储、真 WS 连接）。
+type steerFixture struct {
+	store   *history.Store
+	adapter *steerAdapter
+	tool    *steerRecordingTool
+	srv     *Server
+	bus     bus.Bus
+	conns   map[auth.Role]*websocket.Conn
+}
+
+// newSteerFixture 装配真 loop（只有模型适配器是脚本化的）+ 真 SQLite history +
+// 真 /ws/chat 连接的 owner/admin/viewer 三端。
+func newSteerFixture(t *testing.T) *steerFixture {
+	t.Helper()
+	store := newHistoryStoreForGateway(t)
+	ad := newSteerAdapter()
+	steerTool := &steerRecordingTool{}
+	dispatcher := newSteerDispatcher(t, steerTool)
+	ag, err := agent.New("你是测试助手。", dispatcher, nil, agent.WithModel(ad, nil))
+	if err != nil {
+		t.Fatalf("agent.New: %v", err)
+	}
+	realLoop := loop.New(loop.Options{Agent: ag, Tools: dispatcher, MaxRounds: 8, Log: gwTestLogger()})
+	b := bus.New()
+	srv, ts, jwt := buildEnv(t, nil, Deps{Loop: realLoop, History: store, Bus: b})
+
+	conns := map[auth.Role]*websocket.Conn{}
+	for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin, auth.RoleViewer} {
+		conn := dialWS(t, ts.URL, signAny(t, jwt, "acct", role))
+		t.Cleanup(func() { _ = conn.Close() })
+		conns[role] = conn
+	}
+	waitConnCount(t, srv, len(conns))
+	return &steerFixture{store: store, adapter: ad, tool: steerTool, srv: srv, bus: b, conns: conns}
+}
+
+// assertSteerSameTurnWindow 断言 I1 与 §2.4 的最新性/完整性：整轮只有脚本的 3 步
+// （同一回合，而不是「注入 + 再开一回合」），每一步窗口末条都是被注入的忙碌消息，
+// 且它在后续窗口里**只出现一次**（跨边界不重复注入，I2），旧窗口内容（工具结果、
+// 第一条消息）都还在。
+func assertSteerSameTurnWindow(t *testing.T, f *steerFixture) {
+	t.Helper()
+	wantSteps := steerScriptedToolSteps + 1
+	if got := f.adapter.stepCount(); got != wantSteps {
+		t.Fatalf("模型步数 = %d, want %d（注入必须留在同一回合内，而不是再开一回合）", got, wantSteps)
+	}
+	for i := 1; i < wantSteps; i++ {
+		view := f.adapter.request(i).Messages
+		last := view[len(view)-1]
+		if last.Role != model.RoleUser || messageText(last) != "忙时新消息" {
+			t.Fatalf("第 %d 步窗口末条 = %s:%q, want 忙时新消息（§2.4 最新性 + 一致性：同一窗口）",
+				i+1, last.Role, messageText(last))
+		}
+		if n := occurrences(view, "忙时新消息"); n != 1 {
+			t.Fatalf("第 %d 步窗口里忙碌消息出现 %d 次, want 1（跨边界绝不重复注入）", i+1, n)
+		}
+		if !windowHasText(view, "steer-tool-done") || !windowHasText(view, "第一条") {
+			t.Fatalf("第 %d 步窗口丢了旧内容（注入只能追加，不得顶掉旧窗口）：%v", i+1, windowTextList(view))
+		}
+	}
+	if got := f.tool.count(); got != steerScriptedToolSteps {
+		t.Fatalf("工具执行 %d 次, want %d（注入不取消已发起的工具，也不重跑它们）", got, steerScriptedToolSteps)
+	}
+}
+
+// assertSteerExactlyOnce 断言 I2：恰好被处理一次 —— 一条助手回答、两条用户消息，
+// pending 没有把被吃进的那条当第二个回合再答一遍。
+func assertSteerExactlyOnce(t *testing.T, store *history.Store) {
+	t.Helper()
+	assistants := countHistoryRows(t, store, "web", history.RoleAssistant)
+	if len(assistants) != 1 {
+		t.Fatalf("助手回答 = %d 条, want 1（注入的消息不得被二次作答）", len(assistants))
+	}
+	users := countHistoryRows(t, store, "web", history.RoleUser)
+	if len(users) != 2 {
+		t.Fatalf("用户消息 = %d 条, want 2（不得丢，也不得凭空多出）", len(users))
+	}
+}
+
+// assertSteerAckFrames 断言 design/15 §5 的帧壳/载荷键集/角色门：owner 与 admin 各
+// 收到恰好一帧 steer_ack（messageId = 被吃进那条的权威 id、atStep = 2），viewer 收不到
+// （负例以白名单哨兵确实到达它为前提，否则「什么都没收到」不可判定）。
+func assertSteerAckFrames(t *testing.T, f *steerFixture) {
+	t.Helper()
+	if err := f.bus.Publish(context.Background(), wsPrefix+scopeSentinel, map[string]any{"sessionId": "web"}); err != nil {
+		t.Fatalf("publish sentinel: %v", err)
+	}
+	users := countHistoryRows(t, f.store, "web", history.RoleUser)
+	if len(users) != 2 {
+		t.Fatalf("用户消息 = %d 条, want 2", len(users))
+	}
+	for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin} {
+		frames := drainFrames(t, f.conns[role])
+		assertFrameCount(t, string(role), frames, wsSteerAck, 1)
+		assertSteerAckFrame(t, string(role), findFrame(t, string(role), frames, wsSteerAck), users[1])
+	}
+	viewerFrames := drainFrames(t, f.conns[auth.RoleViewer])
+	assertFrameCount(t, string(auth.RoleViewer), viewerFrames, wsSteerAck, 0)
+	if frameCountOf(viewerFrames, scopeSentinel) == 0 {
+		t.Fatalf("viewer 未收到白名单哨兵 %q，负例不可判定（收到的帧：%v）", scopeSentinel, frameTypesOf(viewerFrames))
+	}
+}
+
+// frameCountOf 统计角色收到的某类型帧条数。
+func frameCountOf(frames []subagentFrame, typ string) int {
+	n := 0
+	for _, f := range frames {
+		if f.Type == typ {
+			n++
+		}
+	}
+	return n
+}
+
+// assertSteerAckFrame 按 design/15 §5 逐项断言一帧 steer_ack：帧壳 sessionId 是该
+// 会话，载荷**键集**恰好是 {messageId, atStep}，且 messageId 是那条被吃进的消息的
+// 权威 id、atStep 是吃进它的模型步序号（第 2 步）。
+func assertSteerAckFrame(t *testing.T, role string, frame subagentFrame, userRow history.Message) {
+	t.Helper()
+	if frame.SessionID != userRow.SessionID {
+		t.Fatalf("%s frame sessionId = %q, want %q", role, frame.SessionID, userRow.SessionID)
+	}
+	var keys map[string]json.RawMessage
+	if err := json.Unmarshal(frame.Data, &keys); err != nil {
+		t.Fatalf("%s frame data is not a JSON object: %v (%s)", role, err, frame.Data)
+	}
+	wantKeys := []string{"messageId", "atStep"}
+	if len(keys) != len(wantKeys) {
+		t.Fatalf("%s frame data keys = %v, want exactly %v (design/15 §5)", role, keys, wantKeys)
+	}
+	for _, k := range wantKeys {
+		if _, ok := keys[k]; !ok {
+			t.Fatalf("%s frame data lacks key %q (design/15 §5): %v", role, k, keys)
+		}
+	}
+	var data struct {
+		MessageID string `json:"messageId"`
+		AtStep    int    `json:"atStep"`
+	}
+	if err := json.Unmarshal(frame.Data, &data); err != nil {
+		t.Fatalf("%s frame data = %s: %v", role, frame.Data, err)
+	}
+	if data.MessageID != userRow.ID {
+		t.Fatalf("%s steer_ack messageId = %q, want the persisted id %q（前端据此把乐观气泡对齐灰态）",
+			role, data.MessageID, userRow.ID)
+	}
+	if data.AtStep != 2 {
+		t.Fatalf("%s steer_ack atStep = %d, want 2（被第 2 个模型步吃进）", role, data.AtStep)
+	}
+}
+
+// TestSteer_UnpersistedPendingStaysPending 是失败路径的反向断言：转录里看不到那条
+// 消息时（这里用不返回任何行的历史替身模拟「尚未落库 / 读转录失败」），注入让路 ——
+// 消息留在 pending，照旧作为自己的回合同答，**绝不静默消失**。
+func TestSteer_UnpersistedPendingStaysPending(t *testing.T) {
+	t.Parallel()
+	run := newGatedRunner(2)
+	hist := &fakeHistory{}
+	o := NewOrchestrator(OrchestratorOptions{Run: run.Run, Log: gwTestLogger(), History: hist, MaxPending: 5})
+	if sr := o.Submit(context.Background(), "web", []model.Message{userMsg("第一条")}, loop.OwnerEnv()); !sr.Immediate {
+		t.Fatalf("submit = %+v, want Immediate", sr)
+	}
+	// 票还没放：回合停在 runner 里，会话稳定处于 busy（这正是「忙时」）。
+	if !waitQueueDepth(o, "web", 0) {
+		t.Fatal("前提不成立：队列一开始必须是空的")
+	}
+	if !waitBusy(o, "web") {
+		t.Fatal("第一个回合没有进入在飞状态")
+	}
+
+	if sr := o.Submit(context.Background(), "web", []model.Message{userMsg("忙时新消息")}, loop.OwnerEnv()); !sr.Pending {
+		t.Fatalf("busy submit = %+v, want Pending", sr)
+	}
+	if !waitQueueDepth(o, "web", 1) {
+		t.Fatal("第二条没有进入 pending")
+	}
+
+	// 真 loop 会在边界上调用转向源；这里也一样（替身给不出转录行 ⇒ 不注入），
+	// 因此第二条必须照旧由**自己的回合**作答 —— 注入让路不等于消息被丢。
+	run.tickets <- struct{}{}
+	run.tickets <- struct{}{}
+	if !waitIdle(o, "web") {
+		t.Fatal("会话没有排空")
+	}
+	got := run.waitCalls(2)
+	if closingText(got[1].messages) != "忙时新消息" {
+		t.Fatalf("第二个回合 = %v, want 忙时新消息（注入让路时消息必须被正常作答，不得丢）",
+			closingText(got[1].messages))
+	}
+}
+
+// waitBusy 轮询直到该会话报告在飞。
+func waitBusy(o *Orchestrator, sessionID string) bool {
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		if o.Busy(sessionID) {
+			return true
+		}
+		time.Sleep(time.Millisecond)
+	}
+	return false
+}
+
+// waitQueueDepth 轮询直到该会话的 pending 深度达到 want。
+func waitQueueDepth(o *Orchestrator, sessionID string, want int) bool {
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		if o.QueueDepth(sessionID) == want {
+			return true
+		}
+		time.Sleep(time.Millisecond)
+	}
+	return false
+}
+
+// windowHasText 报告窗口里是否有任一消息的文本包含 want。
+func windowHasText(msgs []model.Message, want string) bool {
+	for _, m := range msgs {
+		if strings.Contains(messageText(m), want) {
+			return true
+		}
+	}
+	return false
+}
+
+// occurrences 统计窗口里含 want 的消息条数（用于「跨边界只注入一次」的断言）。
+func occurrences(msgs []model.Message, want string) int {
+	n := 0
+	for _, m := range msgs {
+		if strings.Contains(messageText(m), want) {
+			n++
+		}
+	}
+	return n
+}
+
+// windowTextList 把窗口渲染为 role:文本，供失败信息定位。
+func windowTextList(msgs []model.Message) []string {
+	out := make([]string, 0, len(msgs))
+	for _, m := range msgs {
+		out = append(out, string(m.Role)+":"+messageText(m))
+	}
+	return out
+}
+
+// closingText 返回窗口最后一条消息的文本。
+func closingText(msgs []model.Message) string {
+	if len(msgs) == 0 {
+		return ""
+	}
+	return messageText(msgs[len(msgs)-1])
+}
+
+// writeChat 在给定连接上发一条 chat 上行帧（与前端同一条上行契约）。
+func writeChat(t *testing.T, c *websocket.Conn, data string) {
+	t.Helper()
+	if err := c.WriteJSON(wsInbound{Type: "chat", Data: json.RawMessage(data)}); err != nil {
+		t.Fatalf("write chat: %v", err)
+	}
+}

internal/services/gateway/ws_bridge.go

@@ -16,6 +16,21 @@ import (
 // 状态机信号白名单）。
 const wsSubagentResult = "subagent_result"
 
+// wsSteerAck 是「忙时新消息已被本回合吃进」的下行帧类型（design/15 §5，裁决
+// 13.4.5 目标态）。语义：该消息已在在飞回合的**下一个回合内边界**被注入同一回合、
+// 并已从待处理队列移除，但**尚未被正面回答** —— 前端据此把它的插入气泡渲染成灰态
+// （「已读、未答」，对齐 Claude Code 的 gray-until-responding）；「正面回答」是随后
+// 抵达的 message.* / turn.end 帧，不需要第二个确认帧。
+//
+// 帧壳：{type:"steer_ack", sessionId}。载荷 data：{messageId, atStep} —— messageId
+// 是那条用户消息的权威 id（客户端据此把本地乐观气泡对齐到它，而不是靠文本相等这类
+// 脆弱匹配），atStep 是吃进它的模型步序号（从 1 开始）。
+//
+// 它携带具体消息身份（客户端能定位到某一条气泡，而不只是会话忙闲），因此**不登记**
+// 进 frameReachesAll 的「状态机信号」白名单，落 design/11 §7.1 的 fail-closed 默认
+// 档：只发 owner + admin。
+const wsSteerAck = "steer_ack"
+
 // eventBridge 把生产者 bus 主题映射为其下行 WS 事件类型。hub
 // 订阅这些（并重新归一化为 ws.* 类型），使前端既有的
 // 消费方亮起来，而生产者无需改变其主题名。