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 returns an OnStream callback plus a reader for the events it saw,
// so a test can assert the scope the loop stamped on each emitted event.
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 locks the A1 contract: the loop stamps the
// session it was driven for and the turn id attached with WithTurnID on every
// streamed event, so the WS hub can route message.delta/thinking.delta/tool_call.*
// to the right conversation and 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 keeps the backward-compatible default:
// a caller that attaches no turn id still gets the session on each event, but an
// empty TurnID — the field's zero value is the old behavior.
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 covers the loop's other stream emitter: the
// tool_execution_update progress frame must be scoped too, otherwise it would be
// the one frame a multi-session client could not route.
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")
	}
}