Cyrene_v2/internal/runtime/loop/stream_scope_test.go

package loop

import (
	"context"
	"sync"
	"testing"

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

// collectStream 返回一个 OnStream 回调，外加一个读取它所见过
// 事件的 reader，使测试能断言循环给每个发出的事件打上的作用域。
func collectStream() (func(model.StreamEvent), func() []model.StreamEvent) {
	var mu sync.Mutex
	var events []model.StreamEvent
	on := func(ev model.StreamEvent) {
		mu.Lock()
		events = append(events, ev)
		mu.Unlock()
	}
	read := func() []model.StreamEvent {
		mu.Lock()
		defer mu.Unlock()
		return append([]model.StreamEvent(nil), events...)
	}
	return on, read
}

// TestRun_StreamCarriesSessionAndTurn 钉住 A1 契约：循环给每个
// 流式事件打上它被驱动的会话，以及经 WithTurnID 挂载的 turn id，
// 使 WS hub 能把 message.delta/thinking.delta/tool_call.*
// 路由到正确的会话与 turn（design/16 §4.2）。
func TestRun_StreamCarriesSessionAndTurn(t *testing.T) {
	t.Parallel()
	ad := &mockAdapter{
		stream: []mgResp{{resp: model.Response{Content: "Hello world"}}},
		chunks: [][]string{{"Hel", "lo"}},
	}
	a := newTestAgent(t, ad, nil)
	onStream, streams := collectStream()
	l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8})

	ctx := WithTurnID(context.Background(), "turn-xyz")
	if _, err := l.Run(ctx, "web", []model.Message{model.TextMessage(model.RoleUser, "hi")}); err != nil {
		t.Fatalf("Run: %v", err)
	}
	events := streams()
	if len(events) != 2 {
		t.Fatalf("stream events = %d, want 2", len(events))
	}
	for i, ev := range events {
		if ev.SessionID != "web" || ev.TurnID != "turn-xyz" {
			t.Fatalf("event[%d] scope = session %q turn %q, want session web turn turn-xyz", i, ev.SessionID, ev.TurnID)
		}
	}
	if events[0].Delta != "Hel" || events[1].Delta != "lo" {
		t.Fatalf("deltas = [%q %q], want [Hel lo]", events[0].Delta, events[1].Delta)
	}
}

// TestRun_StreamWithoutTurnIDLeavesItEmpty 保持向后兼容的默认行为：
// 未挂载 turn id 的调用方在每个事件上仍拿到会话，但
// TurnID 为空 —— 该字段的零值即旧行为。
func TestRun_StreamWithoutTurnIDLeavesItEmpty(t *testing.T) {
	t.Parallel()
	ad := &mockAdapter{
		stream: []mgResp{{resp: model.Response{Content: "ok"}}},
		chunks: [][]string{{"o", "k"}},
	}
	a := newTestAgent(t, ad, nil)
	onStream, streams := collectStream()
	l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8})

	if _, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "hi")}); err != nil {
		t.Fatalf("Run: %v", err)
	}
	events := streams()
	if len(events) == 0 {
		t.Fatal("no stream events were emitted")
	}
	for i, ev := range events {
		if ev.SessionID != "web" || ev.TurnID != "" {
			t.Fatalf("event[%d] scope = session %q turn %q, want session web turn \"\"", i, ev.SessionID, ev.TurnID)
		}
	}
}

// TestRun_ToolExecutionCarriesScope 覆盖循环的另一个流式发出点：
// tool_execution_update 进度帧也必须带作用域，否则它就会成为
// 多会话客户端唯一无法路由的帧。
func TestRun_ToolExecutionCarriesScope(t *testing.T) {
	t.Parallel()
	ad := &mockAdapter{stream: []mgResp{
		{resp: nativeDatetimeCall()},
		{resp: model.Response{Model: "m", Content: "done"}},
	}}
	td := dispatcherWith(t, &mockTool{name: "datetime", result: map[string]any{"now": "2026-09-08 22:34:00"}})
	a := newTestAgent(t, ad, td)
	onStream, streams := collectStream()
	l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8})

	ctx := WithTurnID(context.Background(), "turn-tool")
	if _, err := l.Run(ctx, "web", []model.Message{model.TextMessage(model.RoleUser, "time?")}); err != nil {
		t.Fatalf("Run: %v", err)
	}
	found := 0
	for _, ev := range streams() {
		if ev.Kind != model.StreamKindToolExecution {
			continue
		}
		found++
		if ev.SessionID != "web" || ev.TurnID != "turn-tool" {
			t.Fatalf("tool_execution scope = session %q turn %q, want session web turn turn-tool", ev.SessionID, ev.TurnID)
		}
	}
	if found == 0 {
		t.Fatal("no tool_execution_update event was emitted")
	}
}