feat(gateway,loop): A1 流式帧附加 sessionId/turnId — 多会话并发流式可路由

- model.StreamEvent 增 SessionID/TurnID（适配器在 loop 之下，恒留零值） - loop 在回合边界注入；覆盖 stream.go 与 tools.go 的 tool_execution 发射点 - 网关 runTurn 每回合生成 turnID 并经 loop.WithTurnID 挂进运行上下文（编排层与 legacy 直连的唯一收口点）；broadcastStream 填充 wsEvent - Docs/api-client v0.36 契约与变更日志；v0.10 条目标注为已被补齐 - 新增 loop/gateway 测试，含变异验证
2026-10-03 16:08:52 +08:00
1 parent 29aa60d0eb
commit 61cc32c44b
@@ -948,8 +948,10 @@ func TestHubBroadcastStream(t *testing.T) {

 	// A text_delta StreamEvent on the OnStream→bus→relay path becomes a typed
 	// message.delta frame carrying the delta text and a monotonically increasing
-	// seq, leaving the legacy stream_chunk shape out of the way.
-	hb.broadcastStream(model.StreamEvent{Kind: model.StreamKindTextDelta, Delta: "Hel", Index: 0})
+	// seq, leaving the legacy stream_chunk shape out of the way. The loop stamps
+	// the session/turn identity on the event and the hub must carry it onto the
+	// frame, so a client can route the delta to the active session (design/16 §4.2).
+	hb.broadcastStream(model.StreamEvent{Kind: model.StreamKindTextDelta, Delta: "Hel", Index: 0, SessionID: "web", TurnID: "turn-abc"})
 	ev := readFakeFrame(t, c)
 	if ev.Type != "message.delta" {
 		t.Fatalf("frame type = %q, want message.delta (frame=%+v)", ev.Type, ev)
@@ -961,6 +963,9 @@ func TestHubBroadcastStream(t *testing.T) {
 	if ev.Seq != 1 {
 		t.Fatalf("frame seq = %d, want 1", ev.Seq)
 	}
+	if ev.SessionID != "web" || ev.TurnID != "turn-abc" {
+		t.Fatalf("frame scope = session %q turn %q, want session web turn turn-abc", ev.SessionID, ev.TurnID)
+	}

 	// A tool_call event maps to tool_call.*; a non-StreamEvent payload is dropped.
 	hb.broadcastStream(model.StreamEvent{Kind: model.StreamKindToolCallEnd, ID: "toolu_1", Index: 0})
@@ -0,0 +1,151 @@
+package gateway
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
+)
+
+// scopeStreamAdapter is a model.Adapter that emits two text deltas so a test can
+// drive the real loop and observe the framed events it produces.
+type scopeStreamAdapter struct{}
+
+func (scopeStreamAdapter) Name() string { return "scope-test" }
+
+func (scopeStreamAdapter) Chat(context.Context, model.Request, model.Options) (model.Response, error) {
+	return model.Response{Content: "Hel lo"}, nil
+}
+
+func (scopeStreamAdapter) ChatStream(_ context.Context, _ model.Request, on func(model.StreamEvent)) (model.Response, error) {
+	if on != nil {
+		on(model.StreamEvent{Kind: model.StreamKindTextDelta, Delta: "Hel"})
+		on(model.StreamEvent{Kind: model.StreamKindTextDelta, Delta: "lo"})
+	}
+	return model.Response{Content: "Hel lo"}, nil
+}
+
+// newScopedStreamHub wires a hub to a real loop whose OnStream closure mirrors
+// cmd/cyrene's streamChunk (publish the whole StreamEvent value to ws.stream), so
+// a test exercises the production path OnStream → bus → relay → frame. It returns
+// the hub and a registered fake connection.
+func newScopedStreamHub(t *testing.T) (*hub, *wsConn) {
+	t.Helper()
+	b := bus.New()
+	ag, err := agent.New("You are a helpful assistant.", nil, nil, agent.WithModel(scopeStreamAdapter{}, nil))
+	if err != nil {
+		t.Fatalf("agent.New: %v", err)
+	}
+	lp := loop.New(loop.Options{
+		Agent:     ag,
+		MaxRounds: 8,
+		OnStream: func(ev model.StreamEvent) {
+			_ = b.Publish(context.Background(), wsStreamSubject, ev)
+		},
+	})
+	hb := newTestHub(lp, b)
+	hb.streaming = true
+	hb.start()
+	t.Cleanup(hb.close)
+	c := newFakeConn(hb, 64)
+	hb.register(c)
+	return hb, c
+}
+
+// readDeltas scans the connection's frames until want message.delta frames have
+// arrived. The deltas reach the conn asynchronously through the bus while
+// runTurn's own stream_end is sent straight to the conn, so a test cannot assume
+// an order.
+func readDeltas(t *testing.T, c *wsConn, want int) []wsEvent {
+	t.Helper()
+	var deltas []wsEvent
+	deadline := time.After(3 * time.Second)
+	for len(deltas) < want {
+		select {
+		case raw := <-c.send:
+			var ev wsEvent
+			if err := json.Unmarshal(raw, &ev); err != nil {
+				t.Fatalf("decode frame %s: %v", raw, err)
+			}
+			if ev.Type == "message.delta" {
+				deltas = append(deltas, ev)
+			}
+		case <-deadline:
+			t.Fatalf("timed out with %d message.delta frames, want %d", len(deltas), want)
+		}
+	}
+	return deltas
+}
+
+// TestStreamFramesCarrySessionScope is the end-to-end lock for A1: a turn driven
+// through the hub's runTurn must surface message.delta frames that carry the
+// session and the turn id, so a client can route them to the active session. It
+// runs the real loop instead of calling broadcastStream directly, which also
+// proves the bus carries the new fields untouched — no publisher change needed.
+func TestStreamFramesCarrySessionScope(t *testing.T) {
+	t.Parallel()
+	hb, c := newScopedStreamHub(t)
+
+	if _, err := hb.runTurn(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "hi")}, loop.OwnerEnv()); err != nil {
+		t.Fatalf("runTurn: %v", err)
+	}
+	deltas := readDeltas(t, c, 2)
+
+	for i, ev := range deltas {
+		if ev.SessionID != "web" {
+			t.Fatalf("delta[%d] sessionId = %q, want web (frame=%+v)", i, ev.SessionID, ev)
+		}
+		if !strings.HasPrefix(ev.TurnID, "turn-") {
+			t.Fatalf("delta[%d] turnId = %q, want a turn-* id (frame=%+v)", i, ev.TurnID, ev)
+		}
+	}
+	if deltas[0].TurnID != deltas[1].TurnID {
+		t.Fatalf("turnId differs within one turn: %q vs %q", deltas[0].TurnID, deltas[1].TurnID)
+	}
+	if deltas[0].Seq == 0 || deltas[1].Seq <= deltas[0].Seq {
+		t.Fatalf("seq not increasing: %d then %d", deltas[0].Seq, deltas[1].Seq)
+	}
+	if data, ok := deltas[0].Data.(map[string]any); !ok || data["delta"] != "Hel" {
+		t.Fatalf("first delta data = %+v, want delta=Hel", deltas[0].Data)
+	}
+}
+
+// TestStreamFramesTurnScopeChangesPerTurn covers the orchestrated resume shape: a
+// pending turn drained after the first one completes is a second runTurn call, so
+// it must take a fresh turnId. Without that, deltas from the resumed turn would
+// be indistinguishable from the previous turn's.
+func TestStreamFramesTurnScopeChangesPerTurn(t *testing.T) {
+	t.Parallel()
+	hb, c := newScopedStreamHub(t)
+
+	msgs := []model.Message{model.TextMessage(model.RoleUser, "hi")}
+	if _, err := hb.runTurn(context.Background(), "web", msgs, loop.OwnerEnv()); err != nil {
+		t.Fatalf("first runTurn: %v", err)
+	}
+	first := readDeltas(t, c, 2)
+	if _, err := hb.runTurn(context.Background(), "web", msgs, loop.OwnerEnv()); err != nil {
+		t.Fatalf("second runTurn: %v", err)
+	}
+	second := readDeltas(t, c, 2)
+
+	for i, ev := range second {
+		if ev.SessionID != "web" {
+			t.Fatalf("second turn delta[%d] sessionId = %q, want web", i, ev.SessionID)
+		}
+		if ev.TurnID != second[0].TurnID {
+			t.Fatalf("second turn delta[%d] turnId = %q, want %q for the whole turn", i, ev.TurnID, second[0].TurnID)
+		}
+	}
+	if first[0].TurnID == second[0].TurnID {
+		t.Fatalf("turnId reused across turns: %q", first[0].TurnID)
+	}
+	if second[0].Seq <= first[1].Seq {
+		t.Fatalf("seq not monotonic across turns: %d then %d", first[1].Seq, second[0].Seq)
+	}
+}
@@ -80,6 +80,12 @@ func (h *hub) streamChat(ctx context.Context, payload any) {
 // emits a single whole-text chunk to hold the stream contract.
 func (h *hub) runTurn(ctx context.Context, sessionID string, msgs []model.Message, env loop.SessionEnv) (*loop.Result, error) {
 	turnID := newTurnID()
+	// Thread the turn identity into the run context so the loop stamps it on every
+	// streamed event it emits (message.*/thinking.*/tool_call.*); the hub then puts
+	// it on the outbound frame. Every turn takes a fresh id — a session has at most
+	// one in-flight turn — so a client can correlate one burst of deltas to one
+	// assistant turn (design/16 §4.2).
+	ctx = loop.WithTurnID(ctx, turnID)
 	if h.runner == nil || len(msgs) == 0 {
 		h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"stopReason": "idle"}})
 		return &loop.Result{StopReason: "idle"}, nil
@@ -117,16 +123,17 @@ func (h *hub) handleInterrupt(payload json.RawMessage) {
 // broadcastStream maps a streamed model.StreamEvent to its downstream WS event
 // type and fans it out to every client. It runs on the bus subscription worker;
 // seq is assigned atomically so a burst stays ordered even if the loop emits
-// events from more than one goroutine. TurnId is left empty here because the
-// OnStream bus payload does not carry the turn identity (future work would
-// thread a turn id through loop → streamChunk → payload).
+// events from more than one goroutine. SessionID/TurnID ride on the event itself
+// (the loop stamps them at the turn boundary from the run context), so the frame
+// carries the scope a client needs to route it to the active session and to one
+// assistant turn.
 func (h *hub) broadcastStream(payload any) {
 	ev, ok := payload.(model.StreamEvent)
 	if !ok {
 		return
 	}
 	typ, data := streamEventData(ev)
-	h.broadcast(wsEvent{Type: typ, Seq: int(h.streamSeq.Add(1)), Data: data})
+	h.broadcast(wsEvent{Type: typ, SessionID: ev.SessionID, TurnID: ev.TurnID, Seq: int(h.streamSeq.Add(1)), Data: data})
 }

 // chatStreamRequest is the decoded ws.chat payload.