package dispatcher

import (
	"context"
	"log/slog"
	"reflect"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/desktop/internal/agent"
	"github.com/tidbcloud/AgentTeam/desktop/internal/types"
	"github.com/tidbcloud/AgentTeam/desktop/internal/wire/manager"
	"github.com/tidbcloud/AgentTeam/internal/api/daemon"
)

// TestCurrentGenerationExitPublishesTerminalIdleTurnState drives the confirmed
// defect end to end at the dispatcher boundary: an active session's CURRENT
// Process generation exits, handleProcessExit must publish the terminal
// stopped/idle turn-state to the AgentTeam reporter while the Process is still the
// current generation (before clearing sess.process), so the server Hub overlay
// receives idle instead of dropping an empty snapshot (process == nil) and
// retaining stale active. This is what unblocks Workflow RecoveryRunner attempt
// 1.
func TestCurrentGenerationExitPublishesTerminalIdleTurnState(t *testing.T) {
	d := mustNewDispatcher("")
	mgr := manager.New(nil, nil)
	ch := &reportingStubChannel{id: "agentteam"}
	mgr.Register(ch)
	d.serverConnections = mgr
	d.initServices()

	// A live Process generation that is stopping/stopped: DerivedTurnState
	// resolves idle, and the session snapshot is still resolvable while the
	// terminal report is published.
	exiting := newStoppedProcessForTest(t, "agent-1", "session-1")
	session := &interactiveSession{
		sessionKey:       "agent-1",
		agentID:          "agent-1",
		serverAgentID:    "agent-1",
		workDir:          t.TempDir(),
		binding:          conversationBinding{channelID: "agentteam", serverAgentID: "agent-1"},
		process:          exiting,
		runtimeSessionID: "session-1",
	}
	d.sessionManager.Set("agent-1", session)

	d.handleProcessExit(session, exiting)

	emitted := ch.eventSnapshot()
	var turnState *types.TurnStateEvent
	for _, event := range emitted {
		if turn, ok := event.(types.TurnStateEvent); ok {
			turnState = &turn
			break
		}
	}
	require.NotNil(t, turnState, "the current-generation exit must publish a TurnStateEvent to the AgentTeam reporter")
	require.Equal(t, "agent-1", turnState.AgentID)
	require.Equal(t, types.AgentTurnStateIdle, turnState.TurnState,
		"the current generation must explicitly publish idle so Workflow RecoveryRunner can attempt recovery")
	require.Equal(t, "session-1", turnState.SessionID,
		"the TurnStateEvent must carry the RuntimeSessionID so the Hub session fence can reject a stale event from a replaced session")
}

// TestCurrentGenerationExitReplacementDuringReport is the barrier-controlled
// production race the review required: pause after the old exit observation is
// captured by the AgentTeamReporterSink but before reporter.Emit, commit a
// replacement with a new RuntimeSessionID and a live active Process, then
// release reporting. It proves the emitted event retains the OLD generation's
// session identity (never old idle + new session), the replacement pointer is
// never cleared by the exiting generation's fence, and — after the captured
// stale event passes through the real Hub session fence — the new active
// generation remains authoritative.
func TestCurrentGenerationExitReplacementDuringReport(t *testing.T) {
	d := mustNewDispatcher("")
	mgr := manager.New(nil, nil)
	ch := &barrierTurnStateChannel{id: "agentteam", blocked: make(chan struct{}), release: make(chan struct{})}
	mgr.Register(ch)
	d.serverConnections = mgr
	d.initServices()

	exiting := newStoppedProcessForTest(t, "agent-1", "session-1")
	session := &interactiveSession{
		sessionKey:       "agent-1",
		agentID:          "agent-1",
		serverAgentID:    "agent-1",
		workDir:          t.TempDir(),
		binding:          conversationBinding{channelID: "agentteam", serverAgentID: "agent-1"},
		process:          exiting,
		runtimeSessionID: "session-1",
	}
	// A restart/replacement is in flight while the old generation exits — the
	// production guard that tells the old exit callback to publish the terminal
	// report, preserve a replacement committed during it, and then return without
	// tearing the session down.
	session.restartRetirementInFlight = true
	d.sessionManager.Set("agent-1", session)

	done := make(chan struct{})
	go func() {
		defer close(done)
		d.handleProcessExit(session, exiting)
	}()

	// The sink has captured the CURRENT generation's terminal observation and is
	// holding reporter.Emit. Commit a replacement NOW — during reporting — with a
	// new RuntimeSessionID and a live active Process.
	<-ch.blocked
	replacement := newActiveProcessForTest(t, "agent-1", "session-2")
	session.mu.Lock()
	session.process = replacement
	session.runtimeSessionID = "session-2"
	session.mu.Unlock()
	close(ch.release)
	<-done

	// The emitted TurnStateEvent retained the OLD generation's idle + session-1:
	// a torn old-idle + new-session pair is impossible with the combined resolver.
	emitted := ch.eventSnapshot()
	var turnState *types.TurnStateEvent
	for _, event := range emitted {
		if turn, ok := event.(types.TurnStateEvent); ok {
			turnState = &turn
			break
		}
	}
	require.NotNil(t, turnState, "the current-generation exit must publish a TurnStateEvent to the AgentTeam reporter")
	require.Equal(t, types.AgentTurnStateIdle, turnState.TurnState)
	require.Equal(t, "session-1", turnState.SessionID,
		"the terminal event must retain the old session identity, never old idle + new session")

	// The replacement pointer is never cleared by the exiting generation's
	// identity fence.
	require.Same(t, replacement, session.process,
		"a replacement committed during reporting must not be cleared by the exiting generation")

	// After the captured stale event passes through the real Hub session fence,
	// the new active generation remains authoritative: the Hub knows the current
	// session is session-2 and reports active, and the stale session-1 idle is
	// dropped, never overwriting the new active.
	hub := daemon.New(slog.Default(), recoveryHubStore{}, "", nil)
	server := startRecoveryHubServer(t, hub)
	proto := connectRecoveryProtocol(t, d, server.URL).(types.EventReporter)
	reader := recoveryWorkingReader(hub)
	require.Eventually(t, func() bool {
		working, known := reader.RuntimeWorking(context.Background(), "wsp_1", "agent-1")
		return known && working
	}, 10*time.Second, 20*time.Millisecond,
		"the replacement (session-2 active) reported by the fresh connection's ready snapshot must be working")

	// Replay the captured stale session-1 idle through the real wire encoder into
	// the real Hub: the new-session fence drops it and the new active remains.
	proto.Emit(*turnState)
	require.Eventually(t, func() bool {
		working, known := reader.RuntimeWorking(context.Background(), "wsp_1", "agent-1")
		return known && working
	}, 10*time.Second, 20*time.Millisecond,
		"the stale session-1 idle must be dropped by the Hub session fence and never clear the new active generation")
}

// newStoppedProcessForTest builds a Process whose lifecycle state is stopped so
// DerivedTurnState returns idle, and ExitSummary reports a clean exit.
func newStoppedProcessForTest(t *testing.T, agentID, sessionID string) *agent.Process {
	t.Helper()
	ap := &agent.Process{AgentID: agentID, SessionID: sessionID}
	setProcessLifecycleStopped(t, ap)
	setProcessExitSummary(t, ap, 0, true)
	return ap
}

// newActiveProcessForTest builds a Process whose lifecycle is a live (running,
// non-idle) generation so State() reports a running status and DerivedTurnState
// resolves active.
func newActiveProcessForTest(t *testing.T, agentID, sessionID string) *agent.Process {
	t.Helper()
	ap := &agent.Process{AgentID: agentID, SessionID: sessionID}
	value := reflect.ValueOf(ap).Elem()
	lifecycle := value.FieldByName("processLifecycleState")
	setUnexportedString(t, lifecycle.FieldByName("state"), "running")
	return ap
}

// barrierTurnStateChannel is a reporting channel that records every event but
// HOLDS the first TurnStateEvent Emit until release is closed. It lets a test
// commit a generation replacement while the sink's terminal observation is in
// flight between the resolver capture and reporter.Emit — the exact window the
// combined resolver and the identity-fenced clear must make race-free.
type barrierTurnStateChannel struct {
	mu        sync.Mutex
	id        string
	events    []types.AgentEvent
	blocked   chan struct{} // closed when a TurnStateEvent is being held
	release   chan struct{} // close to release the held TurnStateEvent
	blockedOn bool
}

func (c *barrierTurnStateChannel) ID() string { return c.id }

func (c *barrierTurnStateChannel) Enabled() bool { return true }

func (c *barrierTurnStateChannel) Start(context.Context) error { return nil }

func (c *barrierTurnStateChannel) Stop(context.Context) error { return nil }

func (c *barrierTurnStateChannel) Send(context.Context, types.Metadata, types.OutboundMessage) error {
	return nil
}

func (c *barrierTurnStateChannel) Emit(event types.AgentEvent) {
	c.mu.Lock()
	c.events = append(c.events, event)
	if _, isTurnState := event.(types.TurnStateEvent); isTurnState && !c.blockedOn {
		c.blockedOn = true
		blocked, release := c.blocked, c.release
		c.mu.Unlock()
		close(blocked)
		<-release
		return
	}
	c.mu.Unlock()
}

func (c *barrierTurnStateChannel) eventSnapshot() []types.AgentEvent {
	c.mu.Lock()
	defer c.mu.Unlock()
	return append([]types.AgentEvent(nil), c.events...)
}

// setProcessLifecycleStopped sets the unexported lifecycle state to stopped so
// DerivedTurnState resolves idle.
func setProcessLifecycleStopped(t *testing.T, ap *agent.Process) {
	t.Helper()
	value := reflect.ValueOf(ap).Elem()
	lifecycle := value.FieldByName("processLifecycleState")
	setUnexportedString(t, lifecycle.FieldByName("state"), "stopped")
}