package daemon

import (
	"context"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/workflow/agentruntime"
)

// These tests drive the REAL production pipeline for the turn-state recovery
// signal: the wire `ready` / `agent:session` / `agent:turn_state` JSON frames
// (byte-for-byte the shape the daemon wire encoder emits, see the desktop
// internal/wire turn_state_snapshot_test.go) through hub.handleMessage — the real
// production command router/decoder — into the real clientState merge
// (epoch/session/revision ordering) and the real HubWorkingReader consumed by the
// RecoveryRunner. The only production boundary NOT exercised here is the raw
// WebSocket transport framing itself (hub.ServeConnection read agentteam + ping), which
// is covered by the transport-level websocket tests in this package. No fake
// RuntimeWorkingReader is used: every assertion reads the real Hub overlay.

func restartReader(t *testing.T) (*Hub, *clientState, *agentruntime.HubWorkingReader) {
	t.Helper()
	hub := &Hub{store: &recordingStartStore{}, clients: map[string]*clientState{}}
	client := newOrderingClient("mac_1")
	hub.clients["mac_1"] = client
	reader := agentruntime.NewHubWorkingReader(fakeAgentMachineResolver{machineID: "mac_1"}, hub)
	return hub, client, reader
}

// restartRoute routes one production wire frame through the real decoder/router.
func restartRoute(ctx context.Context, hub *Hub, client *clientState, frame string) error {
	return hub.handleMessage(ctx, client, []byte(frame))
}

// active/queued never wake across a reconnect: a ready snapshot reporting active,
// then per-agent turn_state events, must keep the reader working; a reorder of a
// newer active increment before an older idle snapshot must not make it
// recoverable.
func TestRestartReadyAndIncrementsActiveNeverRecoverable(t *testing.T) {
	ctx := context.Background()
	hub, client, reader := restartReader(t)

	// Reconnect: ready snapshot with active turn state at base watermark.
	require.NoError(t, restartRoute(ctx, hub, client, `{"type":"ready","capabilities":["agent:permission-mode"],"runningAgents":["agt_1"],"runningAgentTurnStates":{"agt_1":"active"},"runningAgentTurnStatesWatermark":0}`))
	working, known := reader.RuntimeWorking(ctx, "wsp_1", "agt_1")
	require.True(t, known)
	require.True(t, working, "active from the reconnect ready snapshot must never be recoverable")

	// A newer active increment (revision 1) keeps it working.
	require.NoError(t, restartRoute(ctx, hub, client, `{"type":"agent:session","agentId":"agt_1","sessionId":"sess_1"}`))
	require.NoError(t, restartRoute(ctx, hub, client, `{"type":"agent:turn_state","agentId":"agt_1","turnState":"active","sessionId":"sess_1","revision":1}`))
	working, known = reader.RuntimeWorking(ctx, "wsp_1", "agt_1")
	require.True(t, known)
	require.True(t, working)

	// A delayed OLDER idle snapshot (watermark 0 <= per-agent revision 1) arrives;
	// the merge must NOT overwrite the newer active increment.
	require.NoError(t, restartRoute(ctx, hub, client, `{"type":"ready","capabilities":["agent:permission-mode"],"runningAgents":["agt_1"],"runningAgentTurnStates":{"agt_1":"idle"},"runningAgentTurnStatesWatermark":0}`))
	working, known = reader.RuntimeWorking(ctx, "wsp_1", "agt_1")
	require.True(t, known)
	require.True(t, working, "an older idle snapshot must never make an active responder recoverable")
	require.Equal(t, "active", client.agentTurnStateRecords["agt_1"].State)
}

// Missing-state grace then snapshot timeout: an online responder with no current
// snapshot defers within the bounded grace, then becomes recoverable — recovery
// can never stall on an empty Hub after a daemon restart.
func TestRestartMissingSnapshotGraceThenRecoverable(t *testing.T) {
	ctx := context.Background()
	now := time.Date(2026, 8, 16, 12, 0, 0, 0, time.UTC)
	hub := &Hub{clients: map[string]*clientState{}}
	client := newOrderingClient("mac_1")
	client.connectedAt = now.Add(-30 * time.Second)
	hub.clients["mac_1"] = client
	reader := agentruntime.NewHubWorkingReaderWithOptions(
		fakeAgentMachineResolver{machineID: "mac_1"},
		hub,
		agentruntime.HubWorkingReaderOptions{Now: func() time.Time { return now }, Grace: 60 * time.Second},
	)

	// Online but no snapshot yet: within grace, deferred (no send, no timing).
	working, known := reader.RuntimeWorking(ctx, "wsp_1", "agt_1")
	require.False(t, known, "within grace the state is deferred")
	require.False(t, working)

	// Grace elapses (connection 90s old > 60s grace): recoverable.
	client.connectedAt = now.Add(-90 * time.Second)
	working, known = reader.RuntimeWorking(ctx, "wsp_1", "agt_1")
	require.True(t, known, "after the bounded grace an online responder with no snapshot becomes recoverable")
	require.False(t, working)
}

// Offline defers then online resumes: disconnect leaves the Hub offline (defer,
// no send, no timing); a reconnect with an idle snapshot resumes the schedule.
func TestRestartOfflineDefersThenOnlineResumes(t *testing.T) {
	ctx := context.Background()
	hub, client, reader := restartReader(t)
	client.setAgentSession("agt_1", "sess_1")
	require.NoError(t, restartRoute(ctx, hub, client, `{"type":"agent:turn_state","agentId":"agt_1","turnState":"idle","sessionId":"sess_1","revision":1}`))
	working, known := reader.RuntimeWorking(ctx, "wsp_1", "agt_1")
	require.True(t, known)
	require.False(t, working, "idle is recoverable")

	// Disconnect: the machine is removed -> offline -> deferred.
	delete(hub.clients, "mac_1")
	working, known = reader.RuntimeWorking(ctx, "wsp_1", "agt_1")
	require.False(t, known, "offline defers (no send, no timing)")
	require.False(t, working)

	// Reconnect: fresh connection, ready snapshot reports idle -> recoverable again.
	hub.clients["mac_1"] = newOrderingClient("mac_1")
	reconnected := hub.clients["mac_1"]
	require.NoError(t, restartRoute(ctx, hub, reconnected, `{"type":"ready","capabilities":["agent:permission-mode"],"runningAgents":["agt_1"],"runningAgentTurnStates":{"agt_1":"idle"},"runningAgentTurnStatesWatermark":0}`))
	working, known = reader.RuntimeWorking(ctx, "wsp_1", "agt_1")
	require.True(t, known, "after reconnect + idle snapshot the responder is recoverable again")
	require.False(t, working)
}

// Stale epoch/session plus same-epoch reorder through the production wire route:
// the epoch fence, session fence, and revision fence compose so a delayed stale
// event can never turn an active/queued responder recoverable.
func TestRestartStaleEpochSessionAndSameEpochReorder(t *testing.T) {
	ctx := context.Background()
	hub := &Hub{store: &recordingStartStore{}, clients: map[string]*clientState{}}
	client := newOrderingClient("mac_1")
	client.sessionGeneration = "epoch_cur"
	hub.clients["mac_1"] = client
	client.setAgentSession("agt_1", "sess_1")
	reader := agentruntime.NewHubWorkingReader(fakeAgentMachineResolver{machineID: "mac_1"}, hub)

	route := func(frame string) {
		require.NoError(t, hub.handleMessage(ctx, client, []byte(frame)))
	}

	// A newer queued increment (revision 5, current epoch/session).
	route(`{"type":"agent:turn_state","agentId":"agt_1","turnState":"queued","sessionId":"sess_1","epoch":"epoch_cur","revision":5}`)
	working, known := reader.RuntimeWorking(ctx, "wsp_1", "agt_1")
	require.True(t, known)
	require.True(t, working, "queued is working")

	// Stale epoch (old connection) with a newer revision: dropped by epoch fence.
	route(`{"type":"agent:turn_state","agentId":"agt_1","turnState":"idle","sessionId":"sess_1","epoch":"epoch_old","revision":9}`)
	// Stale session (replaced session) with a newer revision: dropped by session fence.
	route(`{"type":"agent:turn_state","agentId":"agt_1","turnState":"idle","sessionId":"sess_old","epoch":"epoch_cur","revision":9}`)
	// Same-epoch reorder with an older/equal revision: dropped by the revision fence.
	route(`{"type":"agent:turn_state","agentId":"agt_1","turnState":"idle","sessionId":"sess_1","epoch":"epoch_cur","revision":5}`)
	working, known = reader.RuntimeWorking(ctx, "wsp_1", "agt_1")
	require.True(t, known)
	require.True(t, working, "stale epoch/session and same-epoch reorder must never make a queued responder recoverable")
	require.Equal(t, "queued", client.agentTurnStateRecords["agt_1"].State)
}

// TestOldStopNewActiveRaceLateStaleIdleDropped is the deterministic
// old-stop -> new-active race: a NEW generation has already committed an active
// turn (current session), then a LATE stopped/idle frame from the OLD generation
// (replaced session) arrives. The Hub session fence must drop it so the new
// generation's active/queued state is never overwritten, and the responder is
// never made recoverable by a stale old-generation idle.
func TestOldStopNewActiveRaceLateStaleIdleDropped(t *testing.T) {
	ctx := context.Background()
	hub := &Hub{store: &recordingStartStore{}, clients: map[string]*clientState{}}
	client := newOrderingClient("mac_1")
	client.sessionGeneration = "epoch_cur"
	hub.clients["mac_1"] = client
	reader := agentruntime.NewHubWorkingReader(fakeAgentMachineResolver{machineID: "mac_1"}, hub)

	// New generation commits: session-2 with an active turn.
	client.setAgentSession("agt_1", "sess_2")
	require.NoError(t, restartRoute(ctx, hub, client, `{"type":"agent:turn_state","agentId":"agt_1","turnState":"active","sessionId":"sess_2","epoch":"epoch_cur","revision":6}`))
	working, known := reader.RuntimeWorking(ctx, "wsp_1", "agt_1")
	require.True(t, known)
	require.True(t, working, "the new generation's active turn is working")

	// Late idle from the OLD generation (replaced session-1) arrives. The session
	// fence drops it: the new active is never overwritten.
	require.NoError(t, restartRoute(ctx, hub, client, `{"type":"agent:turn_state","agentId":"agt_1","turnState":"idle","sessionId":"sess_1","epoch":"epoch_cur","revision":7}`))
	working, known = reader.RuntimeWorking(ctx, "wsp_1", "agt_1")
	require.True(t, known)
	require.True(t, working, "a late idle from a replaced session must never make the new active generation recoverable")
	require.Equal(t, "active", client.agentTurnStateRecords["agt_1"].State)
}