package daemon

// Production full-chain test for the authoritative observation fencing: a REAL
// daemon.Hub (the same type server wiring constructs), the HubWorkingReader, and
// the RecoveryRunner. It proves:
//   - a nonzero current connection/session generation and per-Agent turn-state
//     revision reach RecoveryTickParams.Observation through the runner;
//   - the machine observation guard blocks a turn-state replacement while a
//     recovery observation is in use, so a stale pre-guard observation cannot
//     transition park state;
//   - the guard is released on every path.

import (
	"context"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	"github.com/tidbcloud/AgentTeam/internal/workflow/agentruntime"
	"github.com/tidbcloud/AgentTeam/internal/workflow/orchestrator"
	"github.com/tidbcloud/AgentTeam/internal/workflow/plan"
	runkernel "github.com/tidbcloud/AgentTeam/internal/workflow/run"
)

// fullChainResolver resolves the fixed machine for the full-chain test.
type fullChainResolver struct{}

func (fullChainResolver) MachineIDForAgent(context.Context, string, string) (string, error) {
	return "mach_full", nil
}

// fullChainRepository captures the observation that reached the row-locked tick
// and drives the park FSM the way the production store does.
type fullChainRepository struct {
	mu          sync.Mutex
	effect      run.Effect
	observation run.RecoveryObservation
	tickCalled  bool
}

func (r *fullChainRepository) ListRecoveryCandidates(context.Context, run.ListRecoveryCandidatesParams) ([]run.Effect, error) {
	return []run.Effect{r.effect}, nil
}

func (r *fullChainRepository) RecoveryTick(_ context.Context, params run.RecoveryTickParams) (run.RecoveryTickResult, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.observation = params.Observation
	r.tickCalled = true
	if !params.Observation.Working && r.effect.ParkState == run.EffectParkStateArmOnIdle {
		r.effect.ParkState = run.EffectParkStateParked
	}
	return run.RecoveryTickResult{Effect: r.effect, Send: false}, nil
}

type fullChainWriter struct {
	mu     sync.Mutex
	writes int
}

func (w *fullChainWriter) WriteRunMessage(context.Context, orchestrator.MessageRequest) (orchestrator.MessageResult, error) {
	w.mu.Lock()
	defer w.mu.Unlock()
	w.writes++
	return orchestrator.MessageResult{}, nil
}

func fullChainEffect(effectID string) run.Effect {
	request, _ := orchestrator.EncodeEffectRequest(orchestrator.EffectRequest{
		Intent: runkernel.Intent{
			Kind: runkernel.IntentSendMessage, ActivationID: "a1", NodeID: "say", Callback: true,
			Retry: plan.ArtifactRetryPolicy{MaxAttempts: 1, BackoffNanos: int64(time.Second)},
		},
		Message: &orchestrator.MessageRequest{
			WorkspaceID: "wsp_full", PlanID: "wfp_1", PlanVersionID: "wfv_1", RunID: "wri_full",
			ActivationID: "a1", EffectID: effectID, NodeID: "say", SourceKind: "manual",
			OwnerUserID: "usr_1", AuthorMemberID: "mem_1", TargetRef: "#game",
			Content: "original", ClientMessageID: "workflow-run:" + effectID, Callback: true,
		},
	})
	receipt, _ := orchestrator.EncodeEffectResult(orchestrator.EffectResult{
		MessageID: "msg_" + effectID, WakeAgentIDs: []string{"agt_full"},
		CallbackResponderAgentID: "agt_full",
		DeliveryScope:            &domain.MessageScopeValue{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_full"},
	})
	return run.Effect{
		ID: effectID, WorkspaceID: "wsp_full", RunID: "wri_full", ActivationID: "wra_1",
		Kind: orchestrator.EffectKindMessage, Status: run.EffectStatusAwaiting,
		CallbackResponderAgentID: "agt_full", RequestJSON: request, ResultJSON: receipt,
	}
}

// newFullChainHub builds a real Hub with one connected client whose current
// connection carries a session generation, a turn-state revision, and a
// high-water watermark.
func newFullChainHub(t *testing.T) (*Hub, *clientState) {
	t.Helper()
	hub := &Hub{clients: map[string]*clientState{}}
	client := &clientState{
		machineID:             "mach_full",
		workspaceID:           "wsp_full",
		sessionGeneration:     "sess_full_1",
		runningAgents:         map[string]bool{"agt_full": true},
		startingAgents:        map[string]bool{},
		stoppedAgents:         map[string]bool{},
		agentStatuses:         map[string]string{},
		agentActivities:       map[string]agentActivityState{},
		agentSessions:         map[string]string{"agt_full": "sess_full_1"},
		agentTurnStateRecords: map[string]agentTurnStateRecord{"agt_full": {State: "idle", Revision: 7, Generation: "sess_full_1", SessionID: "sess_full_1"}},
		turnStateHighWater:    7,
		connectedAt:           time.Now().UTC(),
	}
	client.guard = hub.observationGuard("mach_full")
	hub.clients["mach_full"] = client
	return hub, client
}

// TestFullChainNonzeroObservationThroughHubWorkingReader proves the real Hub
// observation carries a nonzero session generation, revision, and high-water.
func TestFullChainNonzeroObservationThroughHubWorkingReader(t *testing.T) {
	hub, _ := newFullChainHub(t)
	reader := agentruntime.NewHubWorkingReader(fullChainResolver{}, hub)
	obs, release, ok := reader.AcquireRuntimeObservation(context.Background(), "wsp_full", "agt_full")
	require.True(t, ok)
	defer release()
	require.True(t, obs.Connected)
	require.Equal(t, "sess_full_1", obs.SessionGeneration, "the current session generation must reach the observation")
	require.Equal(t, int64(7), obs.Revision, "the per-Agent turn-state revision must reach the observation")
	require.Equal(t, int64(7), obs.HighWater, "the connection high-water must reach the observation")
	require.Equal(t, "idle", obs.State)
	require.True(t, obs.StateKnown)
	require.False(t, obs.Working, "an idle turn is recoverable, not working")
	require.True(t, obs.Known)
}

// TestFullChainRecoveryRunnerCarriesNonzeroObservation proves the RecoveryRunner
// passes the real Hub observation (nonzero generation + revision) into
// RecoveryTickParams and releases the guard after the tick.
func TestFullChainRecoveryRunnerCarriesNonzeroObservation(t *testing.T) {
	hub, _ := newFullChainHub(t)
	reader := agentruntime.NewHubWorkingReader(fullChainResolver{}, hub)
	effect := fullChainEffect("wfx_full_chain")
	effect.ParkState = run.EffectParkStateArmOnIdle
	repository := &fullChainRepository{effect: effect}
	writer := &fullChainWriter{}
	runner, err := orchestrator.NewRecoveryRunner(repository, writer, orchestrator.RecoveryRunnerOptions{
		WorkerID:       "recovery-full-chain",
		RuntimeWorking: reader,
	})
	require.NoError(t, err)

	outcome, err := runner.RecoverOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, 1, outcome.Candidates)
	require.Zero(t, outcome.Working, "an idle turn is not counted working")
	require.Zero(t, outcome.Sent)

	repository.mu.Lock()
	obs := repository.observation
	tickCalled := repository.tickCalled
	repository.mu.Unlock()
	require.True(t, tickCalled, "the tick must be reached for a known idle observation")
	require.True(t, obs.Known)
	require.False(t, obs.Working)
	require.Equal(t, "sess_full_1", obs.SessionGeneration, "the real session generation must reach the tick")
	require.Equal(t, int64(7), obs.Revision, "the real turn-state revision must reach the tick")
	require.Equal(t, run.EffectParkStateParked, repository.effect.ParkState, "the idle observation confirms parked")
}

// TestFullChainGuardBlocksTurnStateReplacement proves the machine observation
// guard held across a recovery transition blocks a concurrent turn-state
// replacement (a state update attempted after the observation but before the
// row-locked transition commits), so a stale pre-guard observation cannot
// transition park state: the replacement lands only after the guard is
// released, and a fresh observation then reports the post-replacement state.
func TestFullChainGuardBlocksTurnStateReplacement(t *testing.T) {
	hub, client := newFullChainHub(t)
	reader := agentruntime.NewHubWorkingReader(fullChainResolver{}, hub)

	// Acquire the observation guard (idle state) and hold it.
	obs, release, ok := reader.AcquireRuntimeObservation(context.Background(), "wsp_full", "agt_full")
	require.True(t, ok)
	require.False(t, obs.Working, "the seeded state is idle for this barrier")

	// Attempt a turn-state replacement while the guard is held: it must block
	// until the guard is released. This is the "state update attempted after
	// validation but before DB commit" barrier — the guard fences it.
	replaced := make(chan struct{})
	go func() {
		client.setAgentTurnState("agt_full", "active", "sess_full_1", 8, "")
		close(replaced)
	}()
	select {
	case <-replaced:
		t.Fatal("a turn-state replacement must be blocked while the observation guard is held")
	case <-time.After(50 * time.Millisecond):
		// blocked as expected
	}

	// Release the guard (the recovery transition would have committed by now,
	// using the pre-guard idle observation): the blocked replacement lands.
	release()
	select {
	case <-replaced:
	case <-time.After(time.Second):
		t.Fatal("the blocked turn-state replacement must proceed after the guard is released")
	}

	// A fresh observation now reports the post-replacement working state, so a
	// later transition cannot park based on the stale idle.
	fresh, release2, ok := reader.AcquireRuntimeObservation(context.Background(), "wsp_full", "agt_full")
	require.True(t, ok)
	defer release2()
	require.True(t, fresh.Working, "the replacement lands and a fresh observation sees it")
	require.Equal(t, int64(8), fresh.Revision, "the fresh observation carries the new revision")
}