Cyrene_v2/internal/services/gateway/session_steer_test.go

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 收不到
// （负例以白名单哨兵确实到达它为前提，否则「什么都没收到」不可判定）。
//
// 同时断言帧层的两件事（design/15 §2.5.9）：这一路恰好一帧 `turn_begin`（把本路与
// 第一条用户消息钉死），且 steer_ack 的**帧壳 turnId** 与它同值 —— 真并行下客户端
// 就是靠这个值把灰态配到正确的 `turn.end`。
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)
		ack := findFrame(t, string(role), frames, wsSteerAck)
		assertSteerAckFrame(t, string(role), ack, users[1])
		assertTurnBeginBindsAck(t, string(role), frames, ack, users[0])
	}
	viewerFrames := drainFrames(t, f.conns[auth.RoleViewer])
	assertFrameCount(t, string(auth.RoleViewer), viewerFrames, wsSteerAck, 0)
	// turn_begin 携带消息身份 ⇒ fail-closed 默认档，viewer 同样收不到。
	assertFrameCount(t, string(auth.RoleViewer), viewerFrames, EventTurnBegin, 0)
	if frameCountOf(viewerFrames, scopeSentinel) == 0 {
		t.Fatalf("viewer 未收到白名单哨兵 %q，负例不可判定（收到的帧：%v）", scopeSentinel, frameTypesOf(viewerFrames))
	}
}

// assertTurnBeginBindsAck 断言本路的 `turn_begin` 与其 `steer_ack` 在帧层对得上：
// 帧壳 turnId 同值、kind/leg/activeTurns 取规范草案 §2.4 的值、messageIds 是这一路
// 归属的那条用户消息的权威 id。
func assertTurnBeginBindsAck(t *testing.T, role string, frames []subagentFrame, ack subagentFrame, owner history.Message) {
	t.Helper()
	assertFrameCount(t, role, frames, EventTurnBegin, 1)
	begin := findFrame(t, role, frames, EventTurnBegin)
	if begin.TurnID == "" || begin.TurnID != ack.TurnID {
		t.Fatalf("%s: turn_begin 帧壳 turnId = %q，steer_ack 的 = %q：两者必须同值（客户端靠它把灰态配到正确的 turn.end）",
			role, begin.TurnID, ack.TurnID)
	}
	var data struct {
		Kind        string   `json:"kind"`
		MessageIDs  []string `json:"messageIds"`
		Leg         int      `json:"leg"`
		ActiveTurns int      `json:"activeTurns"`
	}
	if err := json.Unmarshal(begin.Data, &data); err != nil {
		t.Fatalf("%s: turn_begin data = %s: %v", role, begin.Data, err)
	}
	if data.Kind != turnBeginKindTurn || data.Leg != 1 || data.ActiveTurns != 1 {
		t.Fatalf("%s: turn_begin = %+v, want {kind:turn, leg:1, activeTurns:1}", role, data)
	}
	if len(data.MessageIDs) != 1 || data.MessageIDs[0] != owner.ID {
		t.Fatalf("%s: turn_begin messageIds = %v, want [%s]（本路归属的那条用户消息）", role, data.MessageIDs, owner.ID)
	}
}

// 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)
	}
}