From 3d8822c9222d0053f6a3eb8235ab7bee8a8d3544 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 14:53:40 +0800 Subject: [PATCH] =?UTF-8?q?feat(loop,gateway):=2013.4.5=20=E7=9B=AE?= =?UTF-8?q?=E6=A0=87=E6=80=81=20=E2=80=94=E2=80=94=20=E5=9B=9E=E5=90=88?= =?UTF-8?q?=E5=86=85=20steer=EF=BC=88=E4=B8=8D=E6=89=93=E6=96=AD=E3=80=81?= =?UTF-8?q?=E6=96=B0=E6=B6=88=E6=81=AF=E8=BF=9B=E5=90=8C=E4=B8=80=E4=B8=8A?= =?UTF-8?q?=E4=B8=8B=E6=96=87=E6=B5=81=EF=BC=89+=20steer=5Fack=20=E5=B8=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 规格＝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 | 33 +- internal/runtime/loop/run.go | 14 + internal/runtime/loop/steer.go | 103 ++++ internal/runtime/loop/steer_test.go | 395 +++++++++++++++ internal/services/gateway/session_events.go | 28 ++ .../services/gateway/session_orchestrator.go | 38 +- internal/services/gateway/session_steer.go | 236 +++++++++ .../services/gateway/session_steer_test.go | 466 ++++++++++++++++++ internal/services/gateway/ws_bridge.go | 15 + 9 files changed, 1306 insertions(+), 22 deletions(-) create mode 100644 internal/runtime/loop/steer.go create mode 100644 internal/runtime/loop/steer_test.go create mode 100644 internal/services/gateway/session_events.go create mode 100644 internal/services/gateway/session_steer.go create mode 100644 internal/services/gateway/session_steer_test.go diff --git a/internal/runtime/loop/flow.go b/internal/runtime/loop/flow.go index d2901e2..ef530dd 100644 --- a/internal/runtime/loop/flow.go +++ b/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) { diff --git a/internal/runtime/loop/run.go b/internal/runtime/loop/run.go index b0df1f6..e629a0d 100644 --- a/internal/runtime/loop/run.go +++ b/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) diff --git a/internal/runtime/loop/steer.go b/internal/runtime/loop/steer.go new file mode 100644 index 0000000..1a5f0b8 --- /dev/null +++ b/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...) +} diff --git a/internal/runtime/loop/steer_test.go b/internal/runtime/loop/steer_test.go new file mode 100644 index 0000000..0642678 --- /dev/null +++ b/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 +} diff --git a/internal/services/gateway/session_events.go b/internal/services/gateway/session_events.go new file mode 100644 index 0000000..3cf8b3c --- /dev/null +++ b/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" +) diff --git a/internal/services/gateway/session_orchestrator.go b/internal/services/gateway/session_orchestrator.go index 967373c..45110f6 100644 --- a/internal/services/gateway/session_orchestrator.go +++ b/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) diff --git a/internal/services/gateway/session_steer.go b/internal/services/gateway/session_steer.go new file mode 100644 index 0000000..5e9991f --- /dev/null +++ b/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 +} diff --git a/internal/services/gateway/session_steer_test.go b/internal/services/gateway/session_steer_test.go new file mode 100644 index 0000000..56ae72a --- /dev/null +++ b/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) + } +} diff --git a/internal/services/gateway/ws_bridge.go b/internal/services/gateway/ws_bridge.go index abcd365..f39e730 100644 --- a/internal/services/gateway/ws_bridge.go +++ b/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.* 类型），使前端既有的 // 消费方亮起来，而生产者无需改变其主题名。