Cyrene_v2/internal/services/gateway/session_parallel_protocol_test.go

package gateway

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"sync"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)

// 本文件是 design/15 §2.5 基础管线的**协议面**回归：turnId 一致性
// （编排器生成 → loop 复用 → 流帧帧壳同值）、带 turnId 的状态通道变体、
// 上行 `parallel` 的解码，以及 REST 三条入口的应答形状。夹具留在
// session_parallel_test.go（同包），两半共用。

// ---------------------------------------------------------------------------
// ⑥ turnId 一致性（编排器生成 → loop 复用 → 流帧帧壳同值）
// ---------------------------------------------------------------------------

// TestRunTurnReusesOrchestratorTurnID 钉住①的硬前提：hub 的 runTurn 必须**复用**
// 编排器已经生成并交给 Run 的那个 turnId（否则 `turn_begin` 与流帧各生成一个，
// 客户端两级分桶立即失配）；未挂载时才新铸一个，因此单路路径无可见差异。
func TestRunTurnReusesOrchestratorTurnID(t *testing.T) {
	t.Parallel()
	h := newTestHub(&stubRunner{res: &loop.Result{Text: "ok", StopReason: "stop"}}, nil)
	conn := newOwnerConn(h, 16)
	h.register(conn)

	const want = "turn-orchestrator-made"
	ctx := loop.WithTurnID(context.Background(), want)
	if _, err := h.runTurn(ctx, "web", []model.Message{userMsg("你好")}, loop.OwnerEnv()); err != nil {
		t.Fatalf("runTurn: %v", err)
	}
	// 一回合发两帧（stream_chunk + stream_end），两帧的帧壳都必须带上同一个 id。
	for _, typ := range []string{"stream_chunk", "stream_end"} {
		if ev := waitFrame(t, conn, typ); ev.TurnID != want {
			t.Fatalf("%s 帧壳 turnId = %q, want %q（编排器生成的那个）", typ, ev.TurnID, want)
		}
	}

	// 未挂载 ⇒ 每回合一个全新的随机 id（老契约不变）。
	if _, err := h.runTurn(context.Background(), "web", []model.Message{userMsg("再来")}, loop.OwnerEnv()); err != nil {
		t.Fatalf("runTurn: %v", err)
	}
	waitFrame(t, conn, "stream_chunk")
	ev := waitFrame(t, conn, "stream_end")
	if ev.TurnID == "" || ev.TurnID == want {
		t.Fatalf("未挂载时应新铸一个 turnId，得到 %q", ev.TurnID)
	}
}

// TestEmitTurnVariantCarriesTurnID 钉住②：带 turnId 的状态通道变体把 turnId 盖到
// 帧壳上；未接线该变体时退化为普通状态帧（绝不因此丢掉一条状态迁移）。
func TestEmitTurnVariantCarriesTurnID(t *testing.T) {
	t.Parallel()
	sink := &turnSink{}
	o := NewOrchestrator(OrchestratorOptions{Run: fixedRunner(nil), Emit: sink.Emit, EmitTurn: sink.EmitTurn})
	o.emitTurnState("web", "turn-1", EventSessionBusy, busyPayload("web", ModeThinking, 2))
	got := sink.snapshot()
	if len(got) != 1 || got[0].turnID != "turn-1" || got[0].typ != EventSessionBusy {
		t.Fatalf("带 turnId 的变体 = %+v, want turnId=turn-1 的 session_busy", got)
	}
	if stateInt(got[0], "activeTurns") != 2 {
		t.Fatalf("activeTurns = %v, want 2", got[0].data["activeTurns"])
	}

	plain := &turnSink{}
	o2 := NewOrchestrator(OrchestratorOptions{Run: fixedRunner(nil), Emit: plain.Emit})
	o2.emitTurnState("web", "turn-1", EventSessionBusy, busyPayload("web", ModeThinking, 1))
	got = plain.snapshot()
	if len(got) != 1 || got[0].turnID != "" {
		t.Fatalf("未接线变体时应退化为普通帧（无 turnId），得到 %+v", got)
	}
}

// TestChatStreamRequestDecodesParallel 钉住 WS 上行帧的解码：`parallel:true` 被
// 读懂，省略时为零值（因此省略该字段的老客户端行为逐字不变）。
func TestChatStreamRequestDecodesParallel(t *testing.T) {
	t.Parallel()
	var on chatStreamRequest
	if err := json.Unmarshal([]byte(`{"sessionId":"web","text":"hi","parallel":true}`), &on); err != nil {
		t.Fatalf("解码: %v", err)
	}
	if !on.Parallel {
		t.Fatal("parallel:true 没有被解码到")
	}
	var off chatStreamRequest
	if err := json.Unmarshal([]byte(`{"sessionId":"web","text":"hi"}`), &off); err != nil {
		t.Fatalf("解码: %v", err)
	}
	if off.Parallel {
		t.Fatal("省略 parallel 时必须为零值")
	}
}

// TestRESTParallelAckShapes 钉住 REST 两条入口的应答形状：stream 路径在照准并行时
// 明确回报 parallel+turnId（而不是一个什么都没说的 200），同步路径回 202（第二路在
// 后台跑，拿不到同步回复）。
func TestRESTParallelAckShapes(t *testing.T) {
	t.Parallel()
	s := &Server{}
	rec := httptest.NewRecorder()
	s.writeChatStreamAck(rec, SubmitResult{Accepted: true, Immediate: true, Parallel: true, TurnID: "turn-9"}, "web")
	if rec.Code != http.StatusOK {
		t.Fatalf("stream 并行 ack 状态码 = %d, want 200", rec.Code)
	}
	var body struct {
		Data map[string]any `json:"data"`
	}
	if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
		t.Fatalf("解码 ack: %v", err)
	}
	if body.Data["stream"] != true || body.Data["parallel"] != true || body.Data["turnId"] != "turn-9" {
		t.Fatalf("stream 并行 ack = %v, want {stream:true, parallel:true, turnId:turn-9}", body.Data)
	}

	// 现状不动：排队仍是 202 + queued。
	rec = httptest.NewRecorder()
	s.writeChatStreamAck(rec, SubmitResult{Accepted: true, Pending: true, QueueDepth: 3}, "web")
	if rec.Code != http.StatusAccepted {
		t.Fatalf("排队 ack 状态码 = %d, want 202", rec.Code)
	}
}

// blockingChatter 是**永不返回**的旧式 Chatter（`Deps.Loop`）：它把会话钉在
// 「忙」上，使黑盒用例能在真实 HTTP 层断言并行分支的应答形状 —— 而不必让第一条
// 请求同步跑完。它尊重 ctx：网关关闭 / 中断会让它立刻返回，不留泄漏的 goroutine。
type blockingChatter struct {
	started chan struct{}
	release chan struct{}
	stopped sync.Once
}

func newBlockingChatter() *blockingChatter {
	return &blockingChatter{started: make(chan struct{}, 8), release: make(chan struct{})}
}

func (b *blockingChatter) Run(ctx context.Context, _ string, _ []model.Message) (*loop.Result, error) {
	b.started <- struct{}{}
	select {
	case <-b.release:
		return &loop.Result{Text: "ok", StopReason: "stop"}, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

func (b *blockingChatter) stop() { b.stopped.Do(func() { close(b.release) }) }

// TestRESTParallelEntriesShareOneShape 是 design/15 §2.5.2 的黑盒回归：REST 的
// 三条入口**同形** ——
//   - 第一条经 `stream:true` 进入（提交即返回，回合在后台被钉住，会话因此忙碌）；
//   - 第二条走默认（同步）入口 + `parallel:true` ⇒ 202 + `parallel:true` + turnId
//     （第二路在后台跑，同步入口拿不到回复，绝不回一个 res 为 nil 的 200）；
//   - 第三条省掉 `parallel` ⇒ 仍是既有的 202 `{queued:true, queueDepth}`。
//
// 它同时覆盖 Deps.MaxParallelTurns → 编排器的接线（第二路真的起来了，用 turnId
// 佐证：只有真的开了一路才会有它）。
func TestRESTParallelEntriesShareOneShape(t *testing.T) {
	t.Parallel()
	ch := newBlockingChatter()
	srv, ts, jwt := buildEnv(t, nil, Deps{Loop: ch, MaxParallelTurns: 2})
	t.Cleanup(func() { ch.stop(); srv.Close() })
	tok := signAny(t, jwt, "acct", auth.RoleOwner)

	resp, _ := doReq(t, http.MethodPost, ts.URL+"/api/v1/chat", map[string]any{
		"sessionId": "web", "text": "第一问", "stream": true,
	}, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("stream 入口状态码 = %d, want 200", resp.StatusCode)
	}
	select {
	case <-ch.started:
	case <-time.After(3 * time.Second):
		t.Fatal("后台回合没有开跑（会话应当是忙碌的）")
	}
	if !srv.hub.orchestrator.Busy("web") {
		t.Fatal("会话没有进入忙碌态，后面的并行断言不成立")
	}

	resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/chat", map[string]any{
		"sessionId": "web", "text": "第二问", "parallel": true,
	}, tok)
	if resp.StatusCode != http.StatusAccepted {
		t.Fatalf("同步入口 + parallel 状态码 = %d, want 202（第二路在后台跑）", resp.StatusCode)
	}
	var accepted map[string]any
	if err := json.Unmarshal(decodeBody(t, body)["data"], &accepted); err != nil {
		t.Fatalf("解码并行 ack: %v (body=%s)", err, body)
	}
	if accepted["parallel"] != true {
		t.Fatalf("同步入口并行 ack = %v, want parallel:true", accepted)
	}
	if turnID, _ := accepted["turnId"].(string); turnID == "" {
		t.Fatalf("并行 ack 必须给出该路的 turnId，得到 %v", accepted)
	}
	waitActiveTurns(t, srv.hub.orchestrator, "web", 2)

	resp, body = doReq(t, http.MethodPost, ts.URL+"/api/v1/chat", map[string]any{
		"sessionId": "web", "text": "第三问",
	}, tok)
	if resp.StatusCode != http.StatusAccepted {
		t.Fatalf("未声明并行的忙时提交状态码 = %d, want 202", resp.StatusCode)
	}
	var queued map[string]any
	if err := json.Unmarshal(decodeBody(t, body)["data"], &queued); err != nil {
		t.Fatalf("解码排队 ack: %v (body=%s)", err, body)
	}
	if queued["queued"] != true || queued["parallel"] != nil {
		t.Fatalf("未声明并行的忙时 ack = %v, want {queued:true}（形状逐字不变）", queued)
	}
}

// TestParallelTurnContextIsAdditive 钉住并行声明的 ctx 通道本身：false 是空操作
// （ctx 上不留痕迹），nil ctx 不 panic。
func TestParallelTurnContextIsAdditive(t *testing.T) {
	t.Parallel()
	if parallelRequested(withParallelTurn(context.Background(), false)) {
		t.Fatal("false 必须是空操作：老客户端的 ctx 上不该留任何痕迹")
	}
	if !parallelRequested(withParallelTurn(context.Background(), true)) {
		t.Fatal("true 必须可被读取侧看到")
	}
	if parallelRequested(nil) {
		t.Fatal("nil ctx 不该被判为声明了并行")
	}
}