Cyrene_v2/internal/runtime/loop/stream_events_test.go

package loop

import (
	"context"
	"errors"
	"strings"
	"sync"
	"testing"

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

// captureFlow 返回一个 FlowPublisher 与读取其已发布快照的 reader，
// 使测试能断言工具节点的 status 如何随 phase 演进（design/19 §1）。
func captureFlow() (*FlowPublisher, func() []*trace.FlowGraph) {
	var mu sync.Mutex
	var seen []*trace.FlowGraph
	flow := NewFlowPublisher(func(subject string, payload any) {
		if subject != graphUpdateSubject {
			return
		}
		if g, ok := payload.(*trace.FlowGraph); ok {
			mu.Lock()
			seen = append(seen, g)
			mu.Unlock()
		}
	}, func(string) bool { return true })
	return flow, func() []*trace.FlowGraph {
		mu.Lock()
		defer mu.Unlock()
		return append([]*trace.FlowGraph(nil), seen...)
	}
}

// toolExecEvents 过滤出 tool_execution 三事件，保持发出顺序。
func toolExecEvents(evs []model.StreamEvent) []model.StreamEvent {
	var out []model.StreamEvent
	for _, ev := range evs {
		switch ev.Kind {
		case model.StreamKindToolExecutionStart, model.StreamKindToolExecutionUpdate, model.StreamKindToolExecutionEnd:
			out = append(out, ev)
		}
	}
	return out
}

// toolNodeStatus 返回图中所有 tool 节点的当前 status。
func toolNodeStatus(g *trace.FlowGraph) []string {
	var out []string
	for _, n := range g.Nodes {
		if n.Kind == "tool" {
			out = append(out, n.Status)
		}
	}
	return out
}

// runToolTurn 用「模型请求 datetime → 工具执行 → 模型收尾」的两步流
// 驱动一个带流式回调与 flow 图的 Run，返回收集到的事件与图快照。
func runToolTurn(t *testing.T, toolResult any, toolErr error) ([]model.StreamEvent, []*trace.FlowGraph) {
	t.Helper()
	ad := &mockAdapter{stream: []mgResp{
		{resp: nativeDatetimeCall()},
		{resp: model.Response{Model: "m", Content: "done"}},
	}}
	td := dispatcherWith(t, &mockTool{name: "datetime", result: toolResult, err: toolErr})
	a := newTestAgent(t, ad, td)
	onStream, streams := collectStream()
	flow, graphs := captureFlow()
	l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8, Flow: flow, Tracer: trace.New(nil)})

	if _, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "time?")}); err != nil {
		t.Fatalf("Run: %v", err)
	}
	return streams(), graphs()
}

// TestRun_ToolExecutionThreeEvents 锁住裁决 01.4.1：一次工具执行发出
// start 与 end 两帧（顺序固定），phase 是 running/done 的结构化字段，
// 绝不再塞进 Delta；工具名走 Tool 字段，ID 复用 provider 的
// tool call id，使执行帧可与 tool_call.* 帧关联。
func TestRun_ToolExecutionThreeEvents(t *testing.T) {
	t.Parallel()
	evs, _ := runToolTurn(t, map[string]any{"now": "2026-09-08 22:34:00"}, nil)
	ex := toolExecEvents(evs)
	if len(ex) != 2 {
		t.Fatalf("tool_execution events = %d, want 2 (start+end): %+v", len(ex), ex)
	}
	start, end := ex[0], ex[1]
	if start.Kind != model.StreamKindToolExecutionStart || start.Phase != model.ToolPhaseRunning {
		t.Fatalf("start event = %+v, want kind start phase running", start)
	}
	if end.Kind != model.StreamKindToolExecutionEnd || end.Phase != model.ToolPhaseDone {
		t.Fatalf("end event = %+v, want kind end phase done", end)
	}
	for _, ev := range ex {
		if ev.Tool != "datetime" {
			t.Fatalf("event tool = %q, want datetime (%+v)", ev.Tool, ev)
		}
		if ev.ID != "call_native_1" {
			t.Fatalf("event id = %q, want the provider tool call id (%+v)", ev.ID, ev)
		}
		if ev.Delta != "" {
			t.Fatalf("event delta = %q, want empty (phase must not ride in Delta)", ev.Delta)
		}
		if ev.StreamID != "" {
			t.Fatalf("event streamId = %q, want empty (tool execution is not a model stream)", ev.StreamID)
		}
	}
}

// TestRun_ToolExecutionErrorPhase 覆盖失败路径：end 帧的 phase 为
// error，且 flow 图中该 tool 节点最终也是 error —— 同一 phase 驱动
// 两处（裁决 03.4.4）。
func TestRun_ToolExecutionErrorPhase(t *testing.T) {
	t.Parallel()
	evs, graphs := runToolTurn(t, nil, errors.New("boom"))
	ex := toolExecEvents(evs)
	if len(ex) != 2 {
		t.Fatalf("tool_execution events = %d, want 2: %+v", len(ex), ex)
	}
	if ex[1].Kind != model.StreamKindToolExecutionEnd || ex[1].Phase != model.ToolPhaseError {
		t.Fatalf("end event = %+v, want kind end phase error", ex[1])
	}
	last := graphs[len(graphs)-1]
	statuses := toolNodeStatus(last)
	if len(statuses) == 0 || statuses[0] != trace.FlowStatusError {
		t.Fatalf("final tool node status = %v, want [error] (phase-driven)", statuses)
	}
}

// TestRun_ToolPhaseDrivesFlowNodeStatus 锁住裁决 03.4.4：工具执行期间
// flow 图出现 running 的 tool 节点，执行结束后翻转为 done —— 节点状态
// 由细粒度 phase 驱动，取值与事件 phase 完全一致。
func TestRun_ToolPhaseDrivesFlowNodeStatus(t *testing.T) {
	t.Parallel()
	evs, graphs := runToolTurn(t, map[string]any{"now": "x"}, nil)
	sawRunning, sawDone := false, false
	for _, g := range graphs {
		for _, s := range toolNodeStatus(g) {
			switch s {
			case trace.FlowStatusRunning:
				sawRunning = true
			case trace.FlowStatusDone:
				sawDone = true
			}
		}
	}
	if !sawRunning || !sawDone {
		t.Fatalf("tool node statuses over the round: running=%v done=%v, want both (phase-driven)", sawRunning, sawDone)
	}
	ex := toolExecEvents(evs)
	if len(ex) != 2 || ex[0].Phase != model.ToolPhaseRunning || ex[1].Phase != model.ToolPhaseDone {
		t.Fatalf("phases = %+v, want [running done] matching the node statuses", ex)
	}
}

// TestRun_StreamIDPerModelStream 锁住裁决 01.4.2：同一次模型流的多帧
// 共享同一 streamId；多步工具轮的第二次模型调用取新的 streamId；
// turn 与 tool_execution 帧不属于任何模型流，不带 streamId。
func TestRun_StreamIDPerModelStream(t *testing.T) {
	t.Parallel()
	ad := &mockAdapter{stream: []mgResp{
		{resp: nativeDatetimeCall()},
		{resp: model.Response{Model: "m", Content: "done"}},
	}, chunks: [][]string{{"Hel"}, {"lo"}}}
	td := dispatcherWith(t, &mockTool{name: "datetime", result: "now"})
	a := newTestAgent(t, ad, td)
	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, "time?")}); err != nil {
		t.Fatalf("Run: %v", err)
	}
	var firstStream, secondStream string
	for _, ev := range streams() {
		if ev.Kind != model.StreamKindTextDelta {
			continue
		}
		if !strings.HasPrefix(ev.StreamID, "stream-") {
			t.Fatalf("delta streamId = %q, want a stream-* id", ev.StreamID)
		}
		switch ev.Delta {
		case "Hel":
			firstStream = ev.StreamID
		case "lo":
			secondStream = ev.StreamID
		}
	}
	if firstStream == "" || secondStream == "" {
		t.Fatalf("missing deltas for the two model streams (first=%q second=%q)", firstStream, secondStream)
	}
	if firstStream == secondStream {
		t.Fatalf("the two model streams share streamId %q; each ChatStream call must mint a fresh one", firstStream)
	}
	for _, ev := range streams() {
		switch ev.Kind {
		case model.StreamKindTurnStart, model.StreamKindTurnEnd,
			model.StreamKindToolExecutionStart, model.StreamKindToolExecutionEnd:
			if ev.StreamID != "" {
				t.Fatalf("%s carries streamId %q; only model-stream frames do", ev.Kind, ev.StreamID)
			}
		}
	}
}

// TestRun_TurnBoundaryFrames 锁住裁决 01.4.3：回合以 turn.start 开始、
// turn.end 结束（无论正常落定还是预算耗尽），两帧携带会话与 turn
// 作用域、无载荷。
func TestRun_TurnBoundaryFrames(t *testing.T) {
	t.Parallel()
	ad := &mockAdapter{stream: []mgResp{{resp: model.Response{Content: "ok"}}}, chunks: [][]string{{"ok"}}}
	a := newTestAgent(t, ad, nil)
	onStream, streams := collectStream()
	l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8})

	ctx := WithTurnID(context.Background(), "turn-b")
	if _, err := l.Run(ctx, "web", []model.Message{model.TextMessage(model.RoleUser, "hi")}); err != nil {
		t.Fatalf("Run: %v", err)
	}
	evs := streams()
	if len(evs) < 3 {
		t.Fatalf("events = %+v, want at least turn.start + one delta + turn.end", evs)
	}
	if evs[0].Kind != model.StreamKindTurnStart || evs[len(evs)-1].Kind != model.StreamKindTurnEnd {
		t.Fatalf("boundary kinds = %q…%q, want turn_start…turn_end", evs[0].Kind, evs[len(evs)-1].Kind)
	}
	for _, boundary := range []model.StreamEvent{evs[0], evs[len(evs)-1]} {
		if boundary.SessionID != "web" || boundary.TurnID != "turn-b" {
			t.Fatalf("boundary scope = session %q turn %q, want web turn-b", boundary.SessionID, boundary.TurnID)
		}
		if boundary.Delta != "" || boundary.Phase != "" || boundary.Tool != "" {
			t.Fatalf("boundary carries payload: %+v", boundary)
		}
	}
}

// TestRun_TurnEndEmittedOnStepError 覆盖错误出口：模型步失败时 Run
// 返回错误，但回合已经以 turn.start 开过，因此 turn.end 必须照样
// 发出，客户端不会留下一个永远开着的回合。
func TestRun_TurnEndEmittedOnStepError(t *testing.T) {
	t.Parallel()
	ad := &mockAdapter{stream: []mgResp{{err: errors.New("model down")}}}
	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.Fatal("Run: expected the step error to surface")
	}
	evs := streams()
	if len(evs) != 2 || evs[0].Kind != model.StreamKindTurnStart || evs[1].Kind != model.StreamKindTurnEnd {
		t.Fatalf("events = %+v, want exactly turn.start + turn.end around the failed run", evs)
	}
}

// TestRun_TurnBoundaryMutedForInternalRuns 内部自主流程（backthink /
// 子代理）经 WithMutedBroadcast 驱动同一个循环：回合边界帧与流帧
// 一样被静音，中间过程绝不泄露进面向用户的聊天流。
func TestRun_TurnBoundaryMutedForInternalRuns(t *testing.T) {
	t.Parallel()
	ad := &mockAdapter{stream: []mgResp{{resp: model.Response{Content: "ok"}}}, chunks: [][]string{{"o"}}}
	a := newTestAgent(t, ad, nil)
	onStream, streams := collectStream()
	l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8})

	ctx := WithMutedBroadcast(context.Background())
	if _, err := l.Run(ctx, "sess-backthink", []model.Message{model.TextMessage(model.RoleUser, "reflect")}); err != nil {
		t.Fatalf("Run: %v", err)
	}
	if evs := streams(); len(evs) != 0 {
		t.Fatalf("muted run emitted %d events, want none: %+v", len(evs), evs)
	}
}