package daemon

// Atomic old-epoch rejection with the turn-state mutation. The event transport
// epoch, per-Agent session, and monotonic revision/watermark are all validated
// inside the single guarded setAgentTurnState critical section (machine guard
// write side + client.mu). The barrier pauses an event after decode/ownership
// validation but before guarded apply, changes the transport generation, resumes
// the event, and proves it is dropped as a true zero-write (state, revision,
// high-water, and diagnostics all unchanged). Lower/equal revision and duplicate
// events are covered as well.

import (
	"context"
	"encoding/json"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
)

// atomicEpochBarrierHub builds a Hub whose testTurnStateApplyHook blocks until
// release, letting the test change the transport generation mid-event.
func atomicEpochBarrierHub(t *testing.T, entered chan struct{}, release chan struct{}) *Hub {
	t.Helper()
	return NewWithOptions(nil, nil, "", nil, HubOptions{
		TestTurnStateApplyHook: func(context.Context) {
			close(entered)
			<-release
		},
	})
}

func turnStatePayload(agentID, state, sessionID string, revision int64, epoch string) []byte {
	payload, _ := json.Marshal(map[string]any{
		"type":      "agent:turn_state",
		"agentId":   agentID,
		"turnState": state,
		"sessionId": sessionID,
		"revision":  revision,
		"epoch":     epoch,
	})
	return payload
}

// TestTurnStateOldEpochRejectedAtomically proves an event whose transport epoch
// does not match the current generation is dropped as a true zero-write: the
// barrier pauses the event after decode/ownership validation, the transport
// generation changes, and the resumed event leaves state, revision, high-water,
// and diagnostics unchanged.
func TestTurnStateOldEpochRejectedAtomically(t *testing.T) {
	entered := make(chan struct{})
	release := make(chan struct{})
	hub := atomicEpochBarrierHub(t, entered, release)

	client := &clientState{
		machineID:         "mac_epoch",
		workspaceID:       "wsp_epoch",
		sessionGeneration: "gen_old",
		runningAgents:     map[string]bool{"agt_1": true},
		startingAgents:    map[string]bool{},
		stoppedAgents:     map[string]bool{},
		agentStatuses:     map[string]string{},
		agentActivities:   map[string]agentActivityState{},
		agentSessions:     map[string]string{"agt_1": "sess_1"},
		agentTurnStateRecords: map[string]agentTurnStateRecord{
			"agt_1": {State: "idle", Revision: 5, Generation: "gen_old", SessionID: "sess_1"},
		},
		turnStateHighWater: 5,
		connectedAt:        epochTestTime(),
	}
	client.guard = hub.observationGuard("mac_epoch")

	// Start the event apply: it decodes/validates then blocks at the hook.
	done := make(chan error, 1)
	go func() {
		done <- handleSessionLinkAgentTurnStateCommand(hub, context.Background(), client, turnStatePayload("agt_1", "active", "sess_1", 6, "gen_old"))
	}()
	<-entered

	// The transport generation changes to a fresh epoch while the event is paused.
	client.setDaemonTransportReady(daemonTransportReady{SchemaVersion: 2, SessionGeneration: "gen_new"})

	// Resume the event: it applies setAgentTurnState with the OLD epoch, which
	// must be a zero-write and must NOT resurrect the generation-retired record.
	close(release)
	require.NoError(t, <-done)

	// Zero-write: the old-epoch event does not re-create the record nor advance
	// the high-water on the fresh generation.
	_, ok := client.agentTurnStateRecords["agt_1"]
	require.False(t, ok, "an old-epoch event must not resurrect a generation-retired record")
	require.Equal(t, int64(0), client.turnStateHighWater, "an old-epoch event must not advance the fresh high-water")
	require.Equal(t, "gen_new", client.sessionGeneration)
}

// TestTurnStateLowerEqualRevisionAndDuplicateDropped proves lower/equal revision
// and duplicate events are dropped as zero-writes: only a strictly-newer revision
// advances the record.
func TestTurnStateLowerEqualRevisionAndDuplicateDropped(t *testing.T) {
	client := &clientState{
		machineID:         "mac_rev",
		workspaceID:       "wsp_rev",
		sessionGeneration: "gen_rev",
		runningAgents:     map[string]bool{"agt_1": true},
		startingAgents:    map[string]bool{},
		stoppedAgents:     map[string]bool{},
		agentStatuses:     map[string]string{},
		agentActivities:   map[string]agentActivityState{},
		agentSessions:     map[string]string{"agt_1": "sess_1"},
		agentTurnStateRecords: map[string]agentTurnStateRecord{
			"agt_1": {State: "active", Revision: 5, Generation: "gen_rev", SessionID: "sess_1"},
		},
		turnStateHighWater: 5,
		connectedAt:        epochTestTime(),
	}

	// Lower revision is dropped.
	client.setAgentTurnState("agt_1", "idle", "sess_1", 4, "gen_rev")
	require.Equal(t, "active", client.agentTurnStateRecords["agt_1"].State, "a lower revision must not overwrite the newer state")
	require.Equal(t, int64(5), client.agentTurnStateRecords["agt_1"].Revision)

	// Equal revision (duplicate) is dropped.
	client.setAgentTurnState("agt_1", "idle", "sess_1", 5, "gen_rev")
	require.Equal(t, "active", client.agentTurnStateRecords["agt_1"].State, "an equal-revision duplicate must not overwrite")
	require.Equal(t, int64(5), client.agentTurnStateRecords["agt_1"].Revision)
	require.Equal(t, int64(5), client.turnStateHighWater)

	// A strictly-newer revision applies.
	client.setAgentTurnState("agt_1", "queued", "sess_1", 6, "gen_rev")
	require.Equal(t, "queued", client.agentTurnStateRecords["agt_1"].State)
	require.Equal(t, int64(6), client.agentTurnStateRecords["agt_1"].Revision)
	require.Equal(t, int64(6), client.turnStateHighWater)
}

// TestTurnStateLegacyEventBoundToCurrentGeneration proves a legacy event with an
// empty epoch is bound to the current generation at application and is observable
// only while that generation owns the connection.
func TestTurnStateLegacyEventBoundToCurrentGeneration(t *testing.T) {
	client := &clientState{
		machineID:             "mac_legacy",
		workspaceID:           "wsp_legacy",
		sessionGeneration:     "gen_legacy",
		runningAgents:         map[string]bool{"agt_1": true},
		startingAgents:        map[string]bool{},
		stoppedAgents:         map[string]bool{},
		agentStatuses:         map[string]string{},
		agentActivities:       map[string]agentActivityState{},
		agentSessions:         map[string]string{},
		agentTurnStateRecords: map[string]agentTurnStateRecord{},
		connectedAt:           epochTestTime(),
	}

	// A legacy event (no epoch) is bound to the current generation.
	client.setAgentTurnState("agt_1", "idle", "", 0, "")
	require.Equal(t, "idle", client.agentTurnStateRecords["agt_1"].State)
	require.Equal(t, "gen_legacy", client.agentTurnStateRecords["agt_1"].Generation, "a legacy event must be bound to the current generation")

	// After a transport generation change the legacy-bound record is retired.
	client.setDaemonTransportReady(daemonTransportReady{SchemaVersion: 2, SessionGeneration: "gen_new"})
	_, ok := client.agentTurnStateRecords["agt_1"]
	require.False(t, ok, "a legacy event bound to an old generation must be retired on generation change")
}

func epochTestTime() time.Time {
	return time.Date(2026, 8, 17, 12, 0, 0, 0, time.UTC)
}