package dispatcher

// Joined production-path recovery regression for #t2741/#t2739. These tests
// drive the REAL dispatcher handleProcessExit for a current stopping/SIGTERM
// generation and forward the ACTUAL TurnStateEvent produced by AgentTeamReporterSink
// through the production wire encoding/decoding path (a real
// desktop/internal/wire Protocol over a real httptest WebSocket) into the real
// daemon.Hub, read it back with HubWorkingReader, and run a REAL RecoveryRunner
// against a due waiting callback. They never hand-construct the terminal
// TurnStateEvent under test: the dispatcher's current-generation exit publishes
// it end to end.

import (
	"context"
	"encoding/json"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"sync"
	"testing"
	"time"

	"github.com/gorilla/websocket"
	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/desktop/internal/profile"
	"github.com/tidbcloud/AgentTeam/desktop/internal/types"
	"github.com/tidbcloud/AgentTeam/desktop/internal/wire"
	"github.com/tidbcloud/AgentTeam/internal/api/daemon"
	"github.com/tidbcloud/AgentTeam/internal/api/dto"
	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/store/agent"
	"github.com/tidbcloud/AgentTeam/internal/store/app/runtime"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/rewind"
	"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"
)

// recoveryHubStore is a minimal daemon.Store double that lets the real Hub
// connection handler/read agentteam run without persistence.
type recoveryHubStore struct{}

func (recoveryHubStore) MarkMachineDisconnected(context.Context, string) error { return nil }
func (recoveryHubStore) PrepareAgentStart(context.Context, string, string, *message.ReceiveMessage, bool, bool, bool) (agent.StartCommand, error) {
	return agent.StartCommand{}, nil
}
func (recoveryHubStore) MarkMachineReady(context.Context, agent.MachineIdentity, agent.ReadyPayload) error {
	return nil
}
func (recoveryHubStore) UpdateAgentStatusFromMachine(context.Context, string, string, string) (bool, error) {
	return true, nil
}
func (recoveryHubStore) UpdateAgentActivityFromMachine(context.Context, string, string, string, string) (bool, error) {
	return true, nil
}
func (recoveryHubStore) UpdateAgentSessionFromMachine(context.Context, string, string, string, bool) (bool, error) {
	return true, nil
}
func (recoveryHubStore) AgentBelongsToMachine(context.Context, string, string) (bool, error) {
	return true, nil
}
func (recoveryHubStore) AgentNamesForMachine(context.Context, string, string) (string, string, bool, error) {
	return "agent-name", "Agent Name", true, nil
}
func (recoveryHubStore) AgentPermissionModeForMachine(context.Context, string, string) (string, bool, error) {
	return agent.AgentPermissionModeFullAccess, true, nil
}
func (recoveryHubStore) MachineIDForAgent(context.Context, string, string) (string, error) {
	return "mac_1", nil
}
func (recoveryHubStore) ClearWorkspaceWriteAgentAuthTokensForMachine(context.Context, string) error {
	return nil
}
func (recoveryHubStore) RecordMessageRecoveryPointAndPruneOtherThreads(context.Context, string, string, string, string, string, string) ([]rewind.MessageRecoveryPoint, error) {
	return nil, nil
}
func (recoveryHubStore) ListAgentsForMachine(context.Context, string) ([]agent.MachineAgent, error) {
	return nil, nil
}

// recordingHubStore records every status/activity/session mutation so the joined
// tests can prove a stale terminal stop never reaches the backing store while a
// current-generation stop still does.
type recordingHubStore struct {
	recoveryHubStore
	mu         sync.Mutex
	statuses   []string
	activities []string
	sessions   []string
}

func (store *recordingHubStore) UpdateAgentStatusFromMachine(_ context.Context, _ string, agentID string, status string) (bool, error) {
	store.mu.Lock()
	defer store.mu.Unlock()
	store.statuses = append(store.statuses, agentID+":"+status)
	return true, nil
}

func (store *recordingHubStore) UpdateAgentActivityFromMachine(_ context.Context, _ string, agentID string, activity string, detail string) (bool, error) {
	store.mu.Lock()
	defer store.mu.Unlock()
	store.activities = append(store.activities, agentID+":"+activity)
	return true, nil
}

func (store *recordingHubStore) UpdateAgentSessionFromMachine(_ context.Context, _ string, agentID string, sessionID string, persistLegacySession bool) (bool, error) {
	store.mu.Lock()
	defer store.mu.Unlock()
	store.sessions = append(store.sessions, agentID+":"+sessionID)
	return true, nil
}

func (store *recordingHubStore) statusSnapshot() []string {
	store.mu.Lock()
	defer store.mu.Unlock()
	return append([]string(nil), store.statuses...)
}

func (store *recordingHubStore) activitySnapshot() []string {
	store.mu.Lock()
	defer store.mu.Unlock()
	return append([]string(nil), store.activities...)
}

func (store *recordingHubStore) sessionSnapshot() []string {
	store.mu.Lock()
	defer store.mu.Unlock()
	return append([]string(nil), store.sessions...)
}

// recordingHubEmitter records the realtime telemetry the Hub emits so the joined
// tests can prove a stale terminal stop never reaches realtime while a
// current-generation stop still does.
type recordingHubEmitter struct {
	mu            sync.Mutex
	statuses      []string
	activities    []string
	sessions      []string
	agentSessions []string
}

func (emitter *recordingHubEmitter) EmitMachineStatus(string, string, string) {}
func (emitter *recordingHubEmitter) EmitMachineCapabilities(string, string, []string, map[string]runtime.RuntimeDefaults, string, string, string) {
}
func (emitter *recordingHubEmitter) EmitDaemonStatus(string, string, string) {}
func (emitter *recordingHubEmitter) EmitAgentWakeStatus(string, string, string, string, string, string, string, string, string, string, string) {
}
func (emitter *recordingHubEmitter) EmitAgentTrajectory(string, string, string, string, int64, json.RawMessage) {
}

func (emitter *recordingHubEmitter) EmitAgentStatusWithInstance(_ string, _ string, agentID string, status string, _ *dto.AppAgentInstance) {
	emitter.mu.Lock()
	defer emitter.mu.Unlock()
	emitter.statuses = append(emitter.statuses, agentID+":"+status)
}

func (emitter *recordingHubEmitter) EmitAgentActivityWithInstance(_ string, _ string, agentID string, activity string, detail string, _ *dto.AppAgentInstance) {
	emitter.mu.Lock()
	defer emitter.mu.Unlock()
	emitter.activities = append(emitter.activities, agentID+":"+activity)
}

func (emitter *recordingHubEmitter) EmitAgentSessionWithInstance(_ string, _ string, agentID string, sessionID string, _ *dto.AppAgentInstance) {
	emitter.mu.Lock()
	defer emitter.mu.Unlock()
	emitter.sessions = append(emitter.sessions, agentID+":"+sessionID)
}

func (emitter *recordingHubEmitter) statusSnapshot() []string {
	emitter.mu.Lock()
	defer emitter.mu.Unlock()
	return append([]string(nil), emitter.statuses...)
}

func (emitter *recordingHubEmitter) activitySnapshot() []string {
	emitter.mu.Lock()
	defer emitter.mu.Unlock()
	return append([]string(nil), emitter.activities...)
}

func (emitter *recordingHubEmitter) sessionSnapshot() []string {
	emitter.mu.Lock()
	defer emitter.mu.Unlock()
	return append([]string(nil), emitter.sessions...)
}
func (recoveryHubStore) GetTaskActiveTarget(context.Context, string, string, string) (message.ReceiveActiveTaskTarget, bool, error) {
	return message.ReceiveActiveTaskTarget{}, false, nil
}

// recoveryMachineResolver resolves every agent to the harness machine, mirroring
// the production ServiceAdapter the RecoveryRunner wiring uses.
type recoveryMachineResolver struct{}

func (recoveryMachineResolver) MachineIDForAgent(context.Context, string, string) (string, error) {
	return "mac_1", nil
}

// startRecoveryHubServer mounts the real daemon Hub connection handler over a
// real httptest WebSocket endpoint at the exact daemon connect path.
func startRecoveryHubServer(t *testing.T, hub *daemon.Hub) *httptest.Server {
	t.Helper()
	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/api/v1/workspaces/wsp_1/daemon/connect" {
			http.NotFound(w, r)
			return
		}
		conn, err := upgrader.Upgrade(w, r, nil)
		if err != nil {
			return
		}
		identity := agent.MachineIdentity{MachineID: "mac_1", WorkspaceID: "wsp_1", CreatedByUserID: "usr_1"}
		_ = hub.ServeConnection(r.Context(), identity, conn)
	}))
	t.Cleanup(server.Close)
	return server
}

// connectRecoveryProtocol builds a REAL production wire.Protocol bound to the
// given dispatcher (so its ready snapshot reflects the dispatcher's live
// sessions), connects it to the given server, and waits for the transport to be
// connected. It does NOT register the Protocol in the dispatcher manager — the
// caller registers it only when the AgentTeamReporterSink must emit into it. The
// returned value implements both types.ServerConnection and types.EventReporter.
func connectRecoveryProtocol(t *testing.T, d *Dispatcher, serverURL string) types.ServerConnection {
	t.Helper()
	conn, err := wire.New(profile.AgentTeamServer{
		ID:          "agentteam",
		Enabled:     true,
		ServerURL:   serverURL,
		WorkspaceID: "wsp_1",
		APIKey:      "machine-key",
		AuthToken:   "machine-key",
	}, d)
	require.NoError(t, err, "the real wire Protocol must construct against the dispatcher runtime")
	waiter, ok := conn.(interface{ WaitConnected(context.Context) error })
	require.True(t, ok)
	ctx, cancel := context.WithCancel(context.Background())
	require.NoError(t, conn.Start(ctx), "the real wire Protocol must start")
	require.NoError(t, waiter.WaitConnected(ctx), "the real wire Protocol must connect to the Hub")
	t.Cleanup(func() {
		_ = conn.Stop(context.Background())
		cancel()
	})
	return conn
}

// recoveryWorkingReader builds the production HubWorkingReader over the given
// Hub with a short grace so the recoverable judgment is reached quickly.
func recoveryWorkingReader(hub *daemon.Hub) *agentruntime.HubWorkingReader {
	return agentruntime.NewHubWorkingReaderWithOptions(
		recoveryMachineResolver{},
		hub,
		agentruntime.HubWorkingReaderOptions{Now: time.Now, Grace: 500 * time.Millisecond},
	)
}

// turnStateGateReporter wraps an inner ServerConnection+EventReporter (the real
// Protocol) and HOLDS the first TurnStateEvent until release is closed,
// forwarding every other event immediately. It lets a test publish a
// new-generation active turn-state while the old generation's terminal stopped
// event is still in flight, then release the stale event so the Hub's session
// fence must reject it. It is itself a full types.ServerConnection so the
// dispatcher manager can register it as the AgentTeam reporter.
type turnStateGateReporter struct {
	inner   types.ServerConnection
	mu      sync.Mutex
	blocked chan struct{}
	release chan struct{}
	held    bool
}

func (g *turnStateGateReporter) ID() string    { return g.inner.ID() }
func (g *turnStateGateReporter) Enabled() bool { return g.inner.Enabled() }
func (g *turnStateGateReporter) Start(ctx context.Context) error {
	return g.inner.Start(ctx)
}
func (g *turnStateGateReporter) Stop(ctx context.Context) error {
	return g.inner.Stop(ctx)
}
func (g *turnStateGateReporter) Send(ctx context.Context, meta types.Metadata, msg types.OutboundMessage) error {
	return g.inner.Send(ctx, meta, msg)
}

func (g *turnStateGateReporter) Emit(event types.AgentEvent) {
	g.mu.Lock()
	if _, isTurnState := event.(types.TurnStateEvent); isTurnState && !g.held {
		g.held = true
		blocked, release := g.blocked, g.release
		g.mu.Unlock()
		close(blocked)
		<-release
		g.mu.Lock()
	}
	g.mu.Unlock()
	g.inner.(types.EventReporter).Emit(event)
}

// recoveryCandidateEffect builds a durable awaiting send_message callback effect
// for the given responder, with valid frozen request/result payloads the real
// RecoveryRunner decodes to send recovery attempt 1.
func recoveryCandidateEffect(t *testing.T, responder, runID, effectID string) run.Effect {
	t.Helper()
	request, err := orchestrator.EncodeEffectRequest(orchestrator.EffectRequest{
		Intent: runkernel.Intent{
			Kind: runkernel.IntentSendMessage, ActivationID: "a1", NodeID: "say", Callback: true,
			Responder: responder,
			Retry:     plan.ArtifactRetryPolicy{MaxAttempts: 1, BackoffNanos: int64(time.Second)},
		},
		Message: &orchestrator.MessageRequest{
			WorkspaceID: "wsp_1", PlanID: "wfp_1", PlanVersionID: "wfv_1", RunID: runID,
			ActivationID: "a1", EffectID: effectID, NodeID: "say", SourceKind: "manual",
			OwnerUserID: "usr_1", AuthorMemberID: "mem_1", TargetRef: "#game",
			ResponderAgentRef: responder,
			Content:           "original", ClientMessageID: "workflow-run:" + effectID, Callback: true,
		},
	})
	require.NoError(t, err)
	receipt, err := orchestrator.EncodeEffectResult(orchestrator.EffectResult{
		MessageID:                "msg_" + effectID,
		WakeAgentIDs:             []string{responder},
		CallbackResponderAgentID: responder,
		DeliveryScope:            &domain.MessageScopeValue{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_game"},
	})
	require.NoError(t, err)
	return run.Effect{
		ID: effectID, WorkspaceID: "wsp_1", RunID: runID, ActivationID: "wra_1",
		Kind: orchestrator.EffectKindMessage, Status: run.EffectStatusAwaiting,
		CallbackResponderAgentID: responder, RequestJSON: request, ResultJSON: receipt,
	}
}

// joinedRecoveryRunner wires a REAL RecoveryRunner against the production
// HubWorkingReader and a recording message writer, over a fake repository that
// returns one due candidate and mirrors the store's conditional transition
// (working observations never send).
func joinedRecoveryRunner(t *testing.T, candidate run.Effect, reader *agentruntime.HubWorkingReader) (*orchestrator.RecoveryRunner, *recordingRecoveryWriter) {
	t.Helper()
	repository := &fakeJoinedRecoveryRepository{candidate: candidate}
	writer := &recordingRecoveryWriter{}
	runner, err := orchestrator.NewRecoveryRunner(repository, writer, orchestrator.RecoveryRunnerOptions{
		WorkerID:         "wkr_rec_test",
		WorkspaceGroupID: "wgrp_rec",
		Now:              func() time.Time { return time.Now() },
		RuntimeWorking:   reader,
	})
	require.NoError(t, err)
	return runner, writer
}

type fakeJoinedRecoveryRepository struct {
	candidate run.Effect
}

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

func (r *fakeJoinedRecoveryRepository) RecoveryTick(_ context.Context, params run.RecoveryTickParams) (run.RecoveryTickResult, error) {
	effect := r.candidate
	send := !params.Observation.Working
	if send {
		effect.RecoveryAttemptCount = 1
	}
	return run.RecoveryTickResult{
		Effect: effect, Send: send, DisplayAttemptIndex: 1,
	}, nil
}

type recordingRecoveryWriter struct {
	mu     sync.Mutex
	writes []orchestrator.MessageRequest
}

func (w *recordingRecoveryWriter) WriteRunMessage(_ context.Context, request orchestrator.MessageRequest) (orchestrator.MessageResult, error) {
	w.mu.Lock()
	defer w.mu.Unlock()
	w.writes = append(w.writes, request)
	return orchestrator.MessageResult{MessageID: "msg_rec"}, nil
}

func (w *recordingRecoveryWriter) snapshot() []orchestrator.MessageRequest {
	w.mu.Lock()
	defer w.mu.Unlock()
	return append([]orchestrator.MessageRequest(nil), w.writes...)
}

// containsStr reports whether the list contains the exact string.
func containsStr(list []string, want string) bool {
	for _, s := range list {
		if s == want {
			return true
		}
	}
	return false
}

// TestJoinedRecoveryCurrentExitTriggersAttempt1 is the one joined
// production-path regression the review required: a Desktop active generation
// receives SIGTERM/current exit -> handleProcessExit publishes the terminal idle
// through the real AgentTeamReporterSink -> the real wire encoder/websocket delivers
// agent:turn_state to the real daemon.Hub -> HubWorkingReader reports idle -> the
// real RecoveryRunner delivers recovery attempt 1. It also proves the FULL
// current-generation stop lifecycle lands: idle + inactive/offline telemetry +
// the session clear, through the backing store and realtime.
func TestJoinedRecoveryCurrentExitTriggersAttempt1(t *testing.T) {
	d := mustNewDispatcher("")
	t.Cleanup(d.Close)

	// A live active generation (SIGTERM is about to stop it). Register before the
	// transport connects so the ready snapshot reports active (recovery must NOT
	// fire while the agent is working).
	ap := newActiveProcessForTest(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:          ap,
		runtimeSessionID: "session-1",
	}
	d.sessionManager.Set("agent-1", session)

	store := &recordingHubStore{}
	emitter := &recordingHubEmitter{}
	hub := daemon.New(slog.Default(), store, "", emitter)
	server := startRecoveryHubServer(t, hub)
	reader := recoveryWorkingReader(hub)

	// The real Protocol is the dispatcher's reporter: the AgentTeamReporterSink emits
	// into the production wire encoder into the Hub.
	conn := connectRecoveryProtocol(t, d, server.URL)
	d.serverConnections.Register(conn)

	// Ready snapshot carries the live active generation: recovery must be working.
	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 live active generation must be working (not recoverable)")

	// SIGTERM: the current Process generation stops and exits. handleProcessExit
	// publishes the terminal stopped/idle turn-state while the Process is still
	// current, so the Hub receives idle and RecoveryRunner can attempt recovery.
	setProcessLifecycleStopped(t, ap)
	d.handleProcessExit(session, ap)

	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 current-generation exit must publish idle to the Hub")

	// The full matching current-generation stop lifecycle landed: the Hub applied
	// the inactive status, the offline activity, and the session clear — through
	// the backing store and realtime, leaving the agent idle (recoverable). The
	// frames arrive asynchronously over the wire FIFO after the idle, so poll for
	// each mutation.
	require.Eventually(t, func() bool {
		return containsStr(store.statusSnapshot(), "agent-1:inactive") &&
			containsStr(store.activitySnapshot(), "agent-1:offline") &&
			containsStr(store.sessionSnapshot(), "agent-1:")
	}, 10*time.Second, 20*time.Millisecond,
		"a matching current-generation stop must reach the backing store for inactive status, offline activity, and the session clear")
	require.Eventually(t, func() bool {
		return containsStr(emitter.statusSnapshot(), "agent-1:inactive") &&
			containsStr(emitter.activitySnapshot(), "agent-1:offline") &&
			containsStr(emitter.sessionSnapshot(), "agent-1:")
	}, 10*time.Second, 20*time.Millisecond,
		"a matching current-generation stop must reach realtime for inactive status, offline activity, and the session clear")

	// The HubWorkingReader must observe idle before the recovery runner decides
	// whether to send: the store/emitter telemetry above can land while the
	// in-memory Hub working state has not yet flipped, and RecoveryTick sends
	// only when !Working. Guard the reader directly to remove the race.
	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 Hub reader must observe idle before recovery attempt 1")

	// A real RecoveryRunner against a due waiting callback delivers attempt 1.
	candidate := recoveryCandidateEffect(t, "agent-1", "wri_joined", "wfx_joined")
	runner, writer := joinedRecoveryRunner(t, candidate, reader)
	outcome, err := runner.RecoverOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, 1, outcome.Sent, "the idle Hub state must drive recovery attempt 1")
	require.Equal(t, 1, outcome.Candidates)
	writes := writer.snapshot()
	require.Len(t, writes, 1, "recovery attempt 1 must be delivered to the recording writer")
	require.Equal(t, "agent-1", writes[0].ResponderAgentRef)
}

// TestJoinedRecoveryStaleStopCannotOverwriteNewActive is the tightly paired race
// the review required: while the old generation's terminal stopped event is held
// before emission, a new generation publishes session-2 active; when the stale
// event is released it must be dropped by the Hub's new-session fence so the new
// active generation remains authoritative and recovery does NOT fire.
func TestJoinedRecoveryStaleStopCannotOverwriteNewActive(t *testing.T) {
	d := mustNewDispatcher("")
	t.Cleanup(d.Close)

	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: the old exit publishes its terminal
	// report, preserves the replacement committed during it, and returns without
	// tearing the session down (matching the production replacement guard).
	session.restartRetirementInFlight = true
	d.sessionManager.Set("agent-1", session)

	store := &recordingHubStore{}
	emitter := &recordingHubEmitter{}
	hub := daemon.New(slog.Default(), store, "", emitter)
	server := startRecoveryHubServer(t, hub)
	reader := recoveryWorkingReader(hub)

	// The dispatcher's reporter is a gate that holds the old terminal stopped
	// event before it reaches the wire, so the test can publish the new
	// generation's active state first.
	conn := connectRecoveryProtocol(t, d, server.URL)
	proto := conn.(types.EventReporter)
	gate := &turnStateGateReporter{inner: conn, blocked: make(chan struct{}), release: make(chan struct{})}
	d.serverConnections.Register(gate)

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

	// The old generation's terminal stopped idle is captured by the sink and held
	// at the gate before emission.
	<-gate.blocked

	// A replacement commits DURING reporting: new RuntimeSessionID + live active
	// Process. The dispatcher snapshot now reports the new active generation.
	replacement := newActiveProcessForTest(t, "agent-1", "session-2")
	session.mu.Lock()
	session.process = replacement
	session.runtimeSessionID = "session-2"
	session.mu.Unlock()

	// Publish the new generation's session-2 active through the real production
	// wire path BEFORE releasing the stale stopped event: session, turn-state,
	// status, and activity all carry session-2 / active / working. Re-emit until
	// the reader accepts the active transition (a single wire frame may be shed
	// under backpressure without ever reaching the Hub).
	require.Eventually(t, func() bool {
		proto.Emit(types.SessionEvent{AgentID: "agent-1", SessionID: "session-2"})
		proto.Emit(types.TurnStateEvent{AgentID: "agent-1", TurnState: types.AgentTurnStateActive, SessionID: "session-2"})
		proto.Emit(types.StatusEvent{AgentID: "agent-1", Status: "active"})
		proto.Emit(types.ActivityEvent{AgentID: "agent-1", Activity: "working", Detail: "running session-2"})
		working, known := reader.RuntimeWorking(context.Background(), "wsp_1", "agent-1")
		return known && working
	}, 10*time.Second, 50*time.Millisecond, "the new session-2 active generation must be working")

	// Release the stale session-1 stopped event: the Hub new-session fence must
	// drop it and the new active generation must remain authoritative.
	close(gate.release)
	<-done
	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 stopped event must be dropped and never clear the new active generation")

	// The replacement Process pointer remains current: the exiting generation's
	// identity-fenced clear never touched it.
	require.Same(t, replacement, session.process,
		"the replacement Process pointer must remain current after the stale stop")

	// The stale terminal stop reached NEITHER the backing store NOR realtime: the
	// terminal inactive/offline/clear frames were all rejected. The session-2
	// active/working telemetry published before the release is preserved and no
	// terminal (inactive/offline/empty-clear) mutation landed.
	require.NotContains(t, store.statusSnapshot(), "agent-1:inactive", "a stale terminal stop must not reach the backing store status")
	require.NotContains(t, store.activitySnapshot(), "agent-1:offline", "a stale terminal stop must not reach the backing store activity")
	require.NotContains(t, store.sessionSnapshot(), "agent-1:", "a stale terminal stop must not reach the backing store session clear")
	require.NotContains(t, emitter.statusSnapshot(), "agent-1:inactive", "a stale terminal stop must not reach realtime status")
	require.NotContains(t, emitter.activitySnapshot(), "agent-1:offline", "a stale terminal stop must not reach realtime activity")
	require.NotContains(t, emitter.sessionSnapshot(), "agent-1:", "a stale terminal stop must not reach realtime session")

	// A real RecoveryRunner sees the new active generation: recovery must NOT fire.
	candidate := recoveryCandidateEffect(t, "agent-1", "wri_joined_stale", "wfx_joined_stale")
	runner, writer := joinedRecoveryRunner(t, candidate, reader)
	outcome, err := runner.RecoverOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, 0, outcome.Sent, "the new active generation must prevent a recovery send")
	require.Len(t, writer.snapshot(), 0, "no recovery nudge may be delivered while the new generation is working")

	// Prove the stale clear did NOT delete session-2 from the Hub — not merely
	// that active survived momentarily: a subsequent session-2 idle turn-state
	// through the SAME production connection (no republished agent:session) is
	// accepted by the Hub's session fence (session-2 is still current) and makes
	// the responder recoverable. Re-emit the idle turn-state until the reader
	// accepts the transition: a single wire frame may be shed under backpressure
	// (sendJSON drops on a full queue) without ever reaching the Hub. If the
	// stale clear HAD deleted session-2, every re-emit would be dropped by the
	// session fence and the reader could never accept idle.
	require.Eventually(t, func() bool {
		proto.Emit(types.TurnStateEvent{AgentID: "agent-1", TurnState: types.AgentTurnStateIdle, SessionID: "session-2"})
		working, known := reader.RuntimeWorking(context.Background(), "wsp_1", "agent-1")
		return known && !working
	}, 10*time.Second, 50*time.Millisecond,
		"the session-2 idle turn-state must be accepted (the stale clear must not have deleted session-2)")

	// A real RecoveryRunner observes the session-2 idle transition and recovers
	// from it.
	candidate2 := recoveryCandidateEffect(t, "agent-1", "wri_joined_stale2", "wfx_joined_stale2")
	runner2, writer2 := joinedRecoveryRunner(t, candidate2, reader)
	outcome2, err := runner2.RecoverOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, 1, outcome2.Sent, "the session-2 idle transition must drive recovery after the stale clear was dropped")
	require.Len(t, writer2.snapshot(), 1, "the recovery nudge must be delivered after the session-2 idle transition")
}