Cyrene_v2/internal/services/gateway/ws_hub_stream_scope_test.go

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)
	}
}