package wire

import (
	"encoding/json"
	"testing"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/desktop/internal/types"
)

// L1: the authoritative session snapshot is the TurnState publication point.
// emitAgentTeamSessionTelemetrySnapshot emits a TurnStateEvent whenever
// snapshot.TurnState is non-empty (active/queued/idle), with the same
// RuntimeSessionID as the SessionEvent emitted first; an empty TurnState emits
// none (server Hub stays unknown).
func TestEmitAgentTeamSessionTelemetrySnapshotPublishesTurnState(t *testing.T) {
	cases := []struct {
		name      string
		turnState string
		wantEmit  bool
	}{
		{"idle live session", types.AgentTurnStateIdle, true},
		{"active turn", types.AgentTurnStateActive, true},
		{"real queued turn", types.AgentTurnStateQueued, true},
		{"empty (unknown/old daemon)", "", false},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			proto := &Protocol{
				sendCh: make(chan []byte, 64),
				disp: &agentteamTestDispatcher{sessions: []types.SessionSnapshot{{
					AgentID:          "agt_1",
					RuntimeSessionID: "sess_1",
					Status:           "active",
					TurnState:        tc.turnState,
				}}},
			}
			proto.emitAgentTeamSessionTelemetrySnapshot("agt_1")
			frames := drainSendCh(t, proto.sendCh)
			// First frame is the SessionEvent carrying the RuntimeSessionID.
			require.GreaterOrEqual(t, len(frames), 1)
			var first map[string]any
			require.NoError(t, json.Unmarshal(frames[0], &first))
			require.Equal(t, "agent:session", first["type"])
			require.Equal(t, "sess_1", first["sessionId"])

			var turnFrame *map[string]any
			for _, frame := range frames {
				var decoded map[string]any
				if err := json.Unmarshal(frame, &decoded); err != nil {
					continue
				}
				if decoded["type"] == "agent:turn_state" {
					cp := decoded
					turnFrame = &cp
					break
				}
			}
			if !tc.wantEmit {
				require.Nil(t, turnFrame, "an empty TurnState must not emit agent:turn_state")
				return
			}
			require.NotNil(t, turnFrame, "a non-empty TurnState must be published by the authoritative snapshot")
			require.Equal(t, tc.turnState, (*turnFrame)["turnState"])
			require.Equal(t, "sess_1", (*turnFrame)["sessionId"], "the turn_state must carry the same sessionId as the SessionEvent")
		})
	}
}

// L1: ordering within one snapshot emission — SessionEvent precedes TurnStateEvent
// and both carry the same sessionId, so the server learns the sessionId first
// (FIFO sendCh) and can fence stale turn states from a replaced session.
func TestEmitAgentTeamSessionTelemetrySnapshotOrdersSessionBeforeTurnState(t *testing.T) {
	proto := &Protocol{
		sendCh: make(chan []byte, 64),
		disp: &agentteamTestDispatcher{sessions: []types.SessionSnapshot{{
			AgentID:          "agt_1",
			RuntimeSessionID: "sess_1",
			Status:           "active",
			TurnState:        types.AgentTurnStateQueued,
		}}},
	}
	proto.emitAgentTeamSessionTelemetrySnapshot("agt_1")
	frames := drainSendCh(t, proto.sendCh)
	order := make([]string, 0, len(frames))
	for _, frame := range frames {
		var decoded map[string]any
		if err := json.Unmarshal(frame, &decoded); err != nil {
			continue
		}
		order = append(order, decoded["type"].(string))
	}
	sessionIdx, turnIdx := -1, -1
	for i, eventType := range order {
		if eventType == "agent:session" {
			sessionIdx = i
		}
		if eventType == "agent:turn_state" {
			turnIdx = i
		}
	}
	require.GreaterOrEqual(t, sessionIdx, 0, "the session event must be emitted")
	require.GreaterOrEqual(t, turnIdx, 0, "the turn state event must be emitted")
	require.Less(t, sessionIdx, turnIdx, "the SessionEvent must precede the TurnStateEvent in the FIFO sendCh")
}

// L1: every turn_state increment carries a strictly newer revision and the ready
// snapshot carries the watermark at capture time, so the server can order them.
func TestEmitAgentTeamSessionTelemetrySnapshotStrictlyNewerRevisions(t *testing.T) {
	proto := &Protocol{
		sendCh: make(chan []byte, 64),
		disp: &agentteamTestDispatcher{sessions: []types.SessionSnapshot{{
			AgentID:          "agt_1",
			RuntimeSessionID: "sess_1",
			Status:           "active",
			TurnState:        types.AgentTurnStateIdle,
		}}},
	}
	// Ready snapshot first: watermark equals the revision counter at capture.
	msg, _ := proto.readyMessage()
	require.Equal(t, int64(0), msg["runningAgentTurnStatesWatermark"], "first ready snapshot watermark is the base revision")

	// Two incremental publications carry strictly newer revisions 1 then 2.
	proto.emitAgentTeamSessionTelemetrySnapshot("agt_1")
	proto.emitAgentTeamSessionTelemetrySnapshot("agt_1")
	frames := drainSendCh(t, proto.sendCh)
	revisions := make([]int64, 0, 2)
	for _, frame := range frames {
		var decoded map[string]any
		if err := json.Unmarshal(frame, &decoded); err != nil {
			continue
		}
		if decoded["type"] == "agent:turn_state" {
			revisions = append(revisions, int64(decoded["revision"].(float64)))
		}
	}
	require.Len(t, revisions, 2)
	require.Equal(t, int64(1), revisions[0])
	require.Equal(t, int64(2), revisions[1])

	// A later ready snapshot carries the current watermark (2) so every future
	// increment is strictly newer.
	msg, _ = proto.readyMessage()
	require.Equal(t, int64(2), msg["runningAgentTurnStatesWatermark"], "second ready watermark reflects the current revision")
}

// L1: every modern ready publication carries an explicit watermark even when
// runningAgentTurnStates is empty/omitted and even when runningAgents is empty,
// so an empty snapshot is still an authoritative current-epoch snapshot whose
// watermark can retire prior per-agent state on the server.
func TestReadyMessageAlwaysEmitsWatermarkEvenEmptySnapshot(t *testing.T) {
	// No running sessions at all -> zero running agents and no turn states.
	emptyProto := &Protocol{sendCh: make(chan []byte, 64), disp: &agentteamTestDispatcher{}}
	msg, _ := emptyProto.readyMessage()
	require.Contains(t, msg, "runningAgentTurnStatesWatermark",
		"a ready with zero running agents must still carry an explicit watermark")
	require.Equal(t, int64(0), msg["runningAgentTurnStatesWatermark"])
	_, hasTurnStates := msg["runningAgentTurnStates"]
	require.False(t, hasTurnStates, "no turn-state entries expected for an empty snapshot")

	// A running agent with an empty (or absent) TurnState.
	emptyTurnState := &Protocol{
		sendCh: make(chan []byte, 64),
		disp: &agentteamTestDispatcher{sessions: []types.SessionSnapshot{{
			AgentID:   "agt_1",
			Status:    "active",
			TurnState: "",
		}}},
	}
	msg, _ = emptyTurnState.readyMessage()
	require.Contains(t, msg, "runningAgentTurnStatesWatermark",
		"a ready with a running agent but no turn-state entry must still carry an explicit watermark")
	require.Equal(t, int64(0), msg["runningAgentTurnStatesWatermark"])
	_, hasTurnStates = msg["runningAgentTurnStates"]
	require.False(t, hasTurnStates, "an empty TurnState must not emit runningAgentTurnStates but must still emit the watermark")
}

func drainSendCh(t *testing.T, sendCh chan []byte) [][]byte {
	t.Helper()
	var frames [][]byte
	for {
		select {
		case frame := <-sendCh:
			frames = append(frames, frame)
		default:
			return frames
		}
	}
}