package service

import (
	"context"
	"fmt"
	"net"
	"sync"
	"testing"
	"time"

	toki "git.toki-labs.com/toki/proto-socket/go"
	"google.golang.org/protobuf/proto"

	edgeevents "iop/apps/edge/internal/events"
	edgenode "iop/apps/edge/internal/node"
	"iop/packages/go/config"
	runtime "iop/packages/go/execution"
	iop "iop/proto/gen/iop"
)

const (
	overlayNodeID     = "node-overlay"
	overlayProviderID = "provider-overlay"
	overlayAdapter    = "vllm-overlay"
	overlayTarget     = "model-overlay"
	overlayGroup      = "group-overlay"
)

func newProviderHealthOverlayService(t *testing.T, client *toki.TcpClient) (*Service, *edgenode.NodeEntry, *edgenode.NodeRecord) {
	t.Helper()
	registry := edgenode.NewRegistry()
	entry := &edgenode.NodeEntry{NodeID: overlayNodeID, Alias: "overlay", Client: client}
	registry.Register(entry)

	record := &edgenode.NodeRecord{
		ID: overlayNodeID,
		Adapters: config.AdaptersConf{VllmInstances: []config.VllmInstanceConf{{
			Name: overlayAdapter, Enabled: true, Capacity: 1,
		}}},
		Providers: []config.NodeProviderConf{{
			ID: overlayProviderID, Type: "vllm", Category: config.CategoryAPI,
			Adapter: overlayAdapter, Models: []string{overlayTarget}, Health: "available", Capacity: 1,
		}},
	}
	store := edgenode.NewNodeStore()
	store.Add(record)
	svc := New(registry, edgeevents.NewBus())
	svc.SetRuntimeConfig(store, []config.ModelCatalogEntry{{
		ID: overlayGroup, Providers: map[string]string{overlayProviderID: overlayTarget},
	}}, NewGroupPolicy(16, 30*time.Second))
	svc.HandleNodeConnect(entry.NodeID, entry.ConnectionGeneration)
	return svc, entry, record
}

func addBoundOverlayLease(t *testing.T, queue *modelQueueManager, runID, providerID, adapter, target string, generation uint64) {
	t.Helper()
	queue.mu.Lock()
	defer queue.mu.Unlock()
	group := queue.getOrCreateGroupLocked(overlayGroup, NewGroupPolicy(16, 30*time.Second))
	group.adapter = adapter
	group.target = target
	queue.leaseSeq++
	leaseID := queue.leaseSeq
	lease := &providerLease{
		id: leaseID, groupKey: overlayGroup, nodeID: overlayNodeID, providerID: providerID,
		generation: generation, adapter: adapter, target: target,
		state: leaseStateTracked, runID: runID,
	}
	queue.leases[leaseID] = lease
	queue.leaseByRun[runID] = leaseID
	if providerID == "" {
		group.inflight[overlayNodeID]++
		return
	}
	key := providerResourceKey{nodeID: overlayNodeID, providerID: providerID}
	resource := queue.resources[key]
	if resource == nil {
		resource = &providerResourceState{
			nodeID: overlayNodeID, providerID: providerID, capacity: 1, enabled: true, generation: generation,
		}
		queue.resources[key] = resource
	}
	resource.reserve(false)
}

func stallFailure(runID, adapter, target, providerHealth, classification string, sequence uint64) *iop.ExecutionFailure {
	return &iop.ExecutionFailure{
		Code: string(runtime.FailureCodeResponseStalled), Message: "provider response stalled", Retryable: true,
		Metadata: map[string]string{
			"failure_code": string(runtime.FailureCodeResponseStalled), "provider_health": providerHealth,
			"liveness_classification": classification, "idle_duration_ms": "300000",
			"run_id": runID, "attempt_id": runID, "attempt_fence": "confirmed",
			"adapter": adapter, "target": target, "health_observation_seq": fmt.Sprint(sequence),
		},
	}
}

func unavailableRunEvent(runID, adapter, target string, sequence uint64) *iop.RunEvent {
	failure := stallFailure(runID, adapter, target, "unavailable", "provider_unhealthy", sequence)
	metadata := make(map[string]string, len(failure.GetMetadata()))
	for key, value := range failure.GetMetadata() {
		metadata[key] = value
	}
	return &iop.RunEvent{RunId: runID, Type: "error", Failure: failure, Metadata: metadata}
}

func assertOverlayUnavailable(t *testing.T, queue *modelQueueManager, generation uint64, want bool, wantSequence uint64) {
	t.Helper()
	queue.mu.Lock()
	defer queue.mu.Unlock()
	overlay := queue.runtimeHealth[providerRuntimeHealthKey{
		nodeID: overlayNodeID, generation: generation, providerID: overlayProviderID,
	}]
	if overlay == nil {
		if want || wantSequence != 0 {
			t.Fatalf("runtime overlay missing, want unavailable=%v sequence=%d", want, wantSequence)
		}
		return
	}
	if overlay.unavailable != want || overlay.observationSeq != wantSequence {
		t.Fatalf("runtime overlay=(unavailable=%v sequence=%d), want (%v,%d)", overlay.unavailable, overlay.observationSeq, want, wantSequence)
	}
}

func TestReceivedRunFailureHealthOverlayTable(t *testing.T) {
	t.Run("missing provider identity releases but cannot project", func(t *testing.T) {
		svc, entry, _ := newProviderHealthOverlayService(t, nil)
		addBoundOverlayLease(t, svc.queue, "run-missing-provider", "", overlayAdapter, overlayTarget, entry.ConnectionGeneration)
		event := unavailableRunEvent("run-missing-provider", overlayAdapter, overlayTarget, 1)
		svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, event)
		if leaseCount(svc.queue) != 0 {
			t.Fatal("valid terminal did not release its provider-less lease")
		}
		if event.GetMetadata()["recovery_handoff"] != "" || len(svc.queue.runtimeHealth) != 0 {
			t.Fatalf("provider-less evidence affected handoff/overlay: event=%#v overlay=%#v", event.GetMetadata(), svc.queue.runtimeHealth)
		}
	})

	for _, tc := range []struct {
		name       string
		nodeID     string
		generation func(uint64) uint64
	}{
		{name: "wrong reception node", nodeID: "other-node", generation: func(generation uint64) uint64 { return generation }},
		{name: "stale reception generation", nodeID: overlayNodeID, generation: func(generation uint64) uint64 { return generation + 1 }},
	} {
		t.Run(tc.name, func(t *testing.T) {
			svc, entry, _ := newProviderHealthOverlayService(t, nil)
			runID := "run-" + tc.name
			addBoundOverlayLease(t, svc.queue, runID, overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
			event := unavailableRunEvent(runID, overlayAdapter, overlayTarget, 1)
			svc.HandleReceivedRunLifecycleEvent(tc.nodeID, tc.generation(entry.ConnectionGeneration), event)
			if leaseCount(svc.queue) != 1 || len(svc.queue.runtimeHealth) != 0 {
				t.Fatalf("wrong reception changed correctness state: leases=%d overlay=%#v", leaseCount(svc.queue), svc.queue.runtimeHealth)
			}
			svc.HandleRunLifecycleEvent(event)
		})
	}

	for _, tc := range []struct {
		name    string
		adapter string
		target  string
	}{
		{name: "adapter binding mismatch", adapter: "other-adapter", target: overlayTarget},
		{name: "target binding mismatch", adapter: overlayAdapter, target: "other-target"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			svc, entry, _ := newProviderHealthOverlayService(t, nil)
			runID := "run-" + tc.name
			addBoundOverlayLease(t, svc.queue, runID, overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
			event := unavailableRunEvent(runID, tc.adapter, tc.target, 1)
			svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, event)
			if leaseCount(svc.queue) != 0 || len(svc.queue.runtimeHealth) != 0 || event.GetMetadata()["recovery_handoff"] != "" {
				t.Fatalf("mismatched binding changed overlay/handoff: event=%#v overlay=%#v", event.GetMetadata(), svc.queue.runtimeHealth)
			}
		})
	}

	t.Run("fresh unavailable lowers admission and snapshot without config mutation", func(t *testing.T) {
		svc, entry, record := newProviderHealthOverlayService(t, nil)
		addBoundOverlayLease(t, svc.queue, "run-unavailable", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
		event := unavailableRunEvent("run-unavailable", overlayAdapter, overlayTarget, 3)
		svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, event)

		if event.GetMetadata()["recovery_handoff"] != "confirmed" || event.GetMetadata()["provider_id"] != overlayProviderID ||
			event.GetFailure().GetMetadata()["recovery_handoff"] != "confirmed" {
			t.Fatalf("confirmed handoff annotation missing: event=%#v failure=%#v", event.GetMetadata(), event.GetFailure().GetMetadata())
		}
		if event.GetMetadata()["recovery_eligible"] != "" || event.GetFailure().GetMetadata()["recovery_eligible"] != "" {
			t.Fatal("Edge handoff invented recovery eligibility")
		}
		assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, true, 3)
		if record.Providers[0].Health != "available" {
			t.Fatalf("config health mutated to %q", record.Providers[0].Health)
		}
		snapshot := svc.queue.getSnapshotForNode(entry.NodeID, record, true)[0]
		if snapshot.GetStatus() != "unavailable" || snapshot.GetHealth() != "unavailable" || snapshot.GetCapacity() != 0 {
			t.Fatalf("effective snapshot did not project overlay: %#v", snapshot)
		}
		candidate := candidateNode{
			entry: entry, capacity: 1, providerID: overlayProviderID, adapter: overlayAdapter,
			servedTarget: overlayTarget, generation: entry.ConnectionGeneration,
		}
		if _, err := svc.queue.admit(context.Background(), overlayGroup, overlayAdapter, overlayTarget, []candidateNode{candidate}, NewGroupPolicy(16, time.Second), nil, false, true); err == nil {
			t.Fatal("runtime-unavailable provider remained admissible")
		}
	})

	t.Run("available stall advances fence but does not recover", func(t *testing.T) {
		svc, entry, _ := newProviderHealthOverlayService(t, nil)
		addBoundOverlayLease(t, svc.queue, "run-lower", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
		svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, unavailableRunEvent("run-lower", overlayAdapter, overlayTarget, 5))

		addBoundOverlayLease(t, svc.queue, "run-available-stall", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
		failure := stallFailure("run-available-stall", overlayAdapter, overlayTarget, "available", "request_stalled", 6)
		event := &iop.RunEvent{RunId: "run-available-stall", Type: "error", Failure: failure, Metadata: map[string]string{}}
		svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, event)
		assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, true, 6)
	})

	t.Run("unknown stall advances fence but does not recover", func(t *testing.T) {
		svc, entry, _ := newProviderHealthOverlayService(t, nil)
		addBoundOverlayLease(t, svc.queue, "run-lower-unknown", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
		svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, unavailableRunEvent("run-lower-unknown", overlayAdapter, overlayTarget, 9))

		addBoundOverlayLease(t, svc.queue, "run-unknown-stall", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
		failure := stallFailure("run-unknown-stall", overlayAdapter, overlayTarget, "unknown", "health_unknown", 10)
		event := &iop.RunEvent{RunId: "run-unknown-stall", Type: "error", Failure: failure, Metadata: map[string]string{}}
		svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, event)
		assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, true, 10)
	})

	t.Run("equal or lower sequence is a projection no-op", func(t *testing.T) {
		svc, entry, _ := newProviderHealthOverlayService(t, nil)
		addBoundOverlayLease(t, svc.queue, "run-first", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
		svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, unavailableRunEvent("run-first", overlayAdapter, overlayTarget, 8))
		for _, sequence := range []uint64{8, 7} {
			runID := fmt.Sprintf("run-stale-%d", sequence)
			addBoundOverlayLease(t, svc.queue, runID, overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
			event := unavailableRunEvent(runID, overlayAdapter, overlayTarget, sequence)
			svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, event)
			if event.GetMetadata()["recovery_handoff"] != "confirmed" {
				t.Fatalf("stale sequence %d lost its request-local handoff", sequence)
			}
		}
		assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, true, 8)
		if leaseCount(svc.queue) != 0 {
			t.Fatal("stale-but-valid terminals did not release exactly once")
		}
	})

	t.Run("new generation does not inherit unavailable overlay", func(t *testing.T) {
		svc, entry, _ := newProviderHealthOverlayService(t, nil)
		addBoundOverlayLease(t, svc.queue, "run-old-generation", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
		svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, unavailableRunEvent("run-old-generation", overlayAdapter, overlayTarget, 4))
		svc.queue.mu.Lock()
		svc.queue.activateNodeGenerationLocked(entry.NodeID, entry.ConnectionGeneration+1)
		candidate := &candidateNode{entry: &edgenode.NodeEntry{NodeID: entry.NodeID}, providerID: overlayProviderID, generation: entry.ConnectionGeneration + 1}
		healthy := svc.queue.candidateRuntimeHealthyLocked(candidate)
		svc.queue.mu.Unlock()
		if !healthy {
			t.Fatal("new connection generation inherited old runtime health")
		}
	})
}

func TestReceivedNormalizedAndTunnelFailureReleaseOnce(t *testing.T) {
	for _, executionPath := range []string{"normalized", "tunnel"} {
		t.Run(executionPath, func(t *testing.T) {
			svc, entry, _ := newProviderHealthOverlayService(t, nil)
			addBoundOverlayLease(t, svc.queue, "run-release-once", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
			const racers = 16
			var wg sync.WaitGroup
			wg.Add(racers)
			for i := 0; i < racers; i++ {
				go func() {
					defer wg.Done()
					if executionPath == "normalized" {
						svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, unavailableRunEvent("run-release-once", overlayAdapter, overlayTarget, 1))
						return
					}
					failure := stallFailure("run-release-once", overlayAdapter, overlayTarget, "unavailable", "provider_unhealthy", 1)
					svc.HandleReceivedProviderTunnelFrame(entry.NodeID, entry.ConnectionGeneration, &iop.ProviderTunnelFrame{
						RunId: "run-release-once", TunnelId: "tunnel-release-once",
						Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR, Failure: failure,
					})
				}()
			}
			wg.Wait()
			if leaseCount(svc.queue) != 0 {
				t.Fatalf("%s lease remained after terminal race", executionPath)
			}
			inFlight, longInFlight := providerResourceCounts(svc.queue, entry.NodeID, overlayProviderID)
			if inFlight != 0 || longInFlight != 0 {
				t.Fatalf("%s counters=(%d,%d), want zero", executionPath, inFlight, longInFlight)
			}
		})
	}
}

func TestReceivedTunnelFailureHandoffBeforeRoute(t *testing.T) {
	svc, entry, _ := newProviderHealthOverlayService(t, nil)
	addBoundOverlayLease(t, svc.queue, "run-tunnel-handoff", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
	frames, unsubscribe := svc.tunnels.subscribe("tunnel-handoff", 1)
	defer unsubscribe()
	failure := stallFailure("run-tunnel-handoff", overlayAdapter, overlayTarget, "unavailable", "provider_unhealthy", 1)
	svc.HandleReceivedProviderTunnelFrame(entry.NodeID, entry.ConnectionGeneration, &iop.ProviderTunnelFrame{
		RunId: "run-tunnel-handoff", TunnelId: "tunnel-handoff",
		Kind:    iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR,
		Failure: failure, Metadata: map[string]string{},
	})
	select {
	case frame := <-frames:
		if frame.GetMetadata()["recovery_handoff"] != "confirmed" || frame.GetFailure().GetMetadata()["provider_id"] != overlayProviderID {
			t.Fatalf("routed terminal missed validated annotation: %#v", frame)
		}
	case <-time.After(time.Second):
		t.Fatal("validated tunnel terminal was not routed")
	}
}

func TestProviderHealthOverlayCapabilitiesRecovery(t *testing.T) {
	edgeConn, nodeConn := net.Pipe()
	t.Cleanup(func() {
		_ = edgeConn.Close()
		_ = nodeConn.Close()
	})
	parserMap := toki.ParserMap{
		toki.TypeNameOf(&iop.NodeCommandRequest{}): func(data []byte) (proto.Message, error) {
			message := &iop.NodeCommandRequest{}
			return message, proto.Unmarshal(data, message)
		},
		toki.TypeNameOf(&iop.NodeCommandResponse{}): func(data []byte) (proto.Message, error) {
			message := &iop.NodeCommandResponse{}
			return message, proto.Unmarshal(data, message)
		},
	}
	edgeClient := toki.NewTcpClient(edgeConn, 0, 0, parserMap)
	nodeClient := toki.NewTcpClient(nodeConn, 0, 0, parserMap)
	toki.AddRequestListenerTyped(&nodeClient.Communicator, func(request *iop.NodeCommandRequest) (*iop.NodeCommandResponse, error) {
		return &iop.NodeCommandResponse{
			RequestId: request.GetRequestId(), Type: request.GetType(),
			Adapter: request.GetAdapter(), Target: request.GetTarget(), SessionId: request.GetSessionId(),
			Result: map[string]string{
				"adapter_key": request.GetAdapter(), "target": request.GetTarget(),
				"provider_status": "available", "health_observation_seq": "2",
			},
		}, nil
	})

	svc, entry, record := newProviderHealthOverlayService(t, edgeClient)
	addBoundOverlayLease(t, svc.queue, "run-needs-recovery", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
	svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, unavailableRunEvent("run-needs-recovery", overlayAdapter, overlayTarget, 1))
	assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, true, 1)

	if _, err := svc.Capabilities(context.Background(), NodeCommandRequestSpec{
		NodeRef: entry.NodeID, Adapter: overlayAdapter, Target: overlayTarget,
	}); err != nil {
		t.Fatalf("CAPABILITIES recovery probe: %v", err)
	}
	assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, false, 2)
	if record.Providers[0].Health != "available" {
		t.Fatalf("recovery mutated config health to %q", record.Providers[0].Health)
	}
	snapshot := svc.queue.getSnapshotForNode(entry.NodeID, record, true)[0]
	if snapshot.GetStatus() != "available" || snapshot.GetHealth() != "available" || snapshot.GetCapacity() != 1 {
		t.Fatalf("recovered snapshot=%#v", snapshot)
	}
}

func TestProviderHealthOverlayCapabilitiesRecoveryRejectsCatalogAmbiguity(t *testing.T) {
	svc, entry, record := newProviderHealthOverlayService(t, nil)
	record.Providers = append(record.Providers, config.NodeProviderConf{
		ID: "provider-healthy", Type: "vllm", Category: config.CategoryAPI,
		Adapter: overlayAdapter, Models: []string{overlayTarget}, Health: "available", Capacity: 1,
	})
	addBoundOverlayLease(t, svc.queue, "run-catalog-ambiguity", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
	svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, unavailableRunEvent("run-catalog-ambiguity", overlayAdapter, overlayTarget, 1))
	assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, true, 1)

	if recovered := svc.queue.applyProviderProbeEvidence(entry.NodeID, entry.ConnectionGeneration, overlayAdapter, overlayTarget, runtime.ProviderStatusAvailable, 2, func() bool { return true }); recovered {
		t.Fatal("ambiguous current catalog recovered provider")
	}
	assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, true, 1)
}

func TestProviderHealthOverlayCapabilitiesRecoveryPreservesAvailableHighWater(t *testing.T) {
	svc, entry, record := newProviderHealthOverlayService(t, nil)
	if recovered := svc.queue.applyProviderProbeEvidence(entry.NodeID, entry.ConnectionGeneration, overlayAdapter, overlayTarget, runtime.ProviderStatusAvailable, 2, func() bool { return true }); recovered {
		t.Fatal("already available provider reported recovery")
	}
	assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, false, 2)

	addBoundOverlayLease(t, svc.queue, "run-delayed-unavailable", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
	svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, unavailableRunEvent("run-delayed-unavailable", overlayAdapter, overlayTarget, 1))
	assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, false, 2)
	if record.Providers[0].Health != "available" {
		t.Fatalf("delayed terminal mutated config health to %q", record.Providers[0].Health)
	}
	snapshot := svc.queue.getSnapshotForNode(entry.NodeID, record, true)[0]
	if snapshot.GetStatus() != "available" || snapshot.GetHealth() != "available" || snapshot.GetCapacity() != 1 {
		t.Fatalf("available high-water was reversed: %#v", snapshot)
	}
}

func TestProviderHealthOverlayCapabilitiesRecoveryRejectsInconclusiveEvidence(t *testing.T) {
	t.Run("unknown and unavailable do not advance the recovery fence", func(t *testing.T) {
		svc, entry, _ := newProviderHealthOverlayService(t, nil)
		addBoundOverlayLease(t, svc.queue, "run-probe-fence", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
		svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, unavailableRunEvent("run-probe-fence", overlayAdapter, overlayTarget, 4))
		if recovered := svc.queue.applyProviderProbeEvidence(entry.NodeID, entry.ConnectionGeneration, overlayAdapter, overlayTarget, runtime.ProviderStatusUnknown, 6, func() bool { return true }); recovered {
			t.Fatal("unknown probe recovered provider")
		}
		if recovered := svc.queue.applyProviderProbeEvidence(entry.NodeID, entry.ConnectionGeneration, overlayAdapter, overlayTarget, runtime.ProviderStatusUnavailable, 7, func() bool { return true }); recovered {
			t.Fatal("unavailable probe recovered provider")
		}
		if recovered := svc.queue.applyProviderProbeEvidence(entry.NodeID, entry.ConnectionGeneration, overlayAdapter, overlayTarget, runtime.ProviderStatusAvailable, 4, func() bool { return true }); recovered {
			t.Fatal("equal-sequence available probe recovered provider")
		}
		assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, true, 4)
	})

	t.Run("stale generation is rejected", func(t *testing.T) {
		svc, entry, _ := newProviderHealthOverlayService(t, nil)
		addBoundOverlayLease(t, svc.queue, "run-stale-generation", overlayProviderID, overlayAdapter, overlayTarget, entry.ConnectionGeneration)
		svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, unavailableRunEvent("run-stale-generation", overlayAdapter, overlayTarget, 1))
		if recovered := svc.queue.applyProviderProbeEvidence(entry.NodeID, entry.ConnectionGeneration, overlayAdapter, overlayTarget, runtime.ProviderStatusAvailable, 2, func() bool { return false }); recovered {
			t.Fatal("stale generation recovered provider")
		}
		assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, true, 1)
	})

	for _, tc := range []struct {
		name    string
		result  map[string]string
		adapter string
		target  string
	}{
		{name: "missing sequence", result: map[string]string{"adapter_key": overlayAdapter, "target": overlayTarget, "provider_status": "available"}, adapter: overlayAdapter, target: overlayTarget},
		{name: "malformed sequence", result: map[string]string{"adapter_key": overlayAdapter, "target": overlayTarget, "provider_status": "available", "health_observation_seq": "bad"}, adapter: overlayAdapter, target: overlayTarget},
		{name: "binding mismatch", result: map[string]string{"adapter_key": "other", "target": overlayTarget, "provider_status": "available", "health_observation_seq": "2"}, adapter: overlayAdapter, target: overlayTarget},
		{name: "unknown status", result: map[string]string{"adapter_key": overlayAdapter, "target": overlayTarget, "provider_status": "corrupt", "health_observation_seq": "2"}, adapter: overlayAdapter, target: overlayTarget},
	} {
		t.Run(tc.name, func(t *testing.T) {
			request := &iop.NodeCommandRequest{Type: iop.NodeCommandType_NODE_COMMAND_TYPE_CAPABILITIES, Adapter: tc.adapter, Target: tc.target}
			response := &iop.NodeCommandResponse{Type: request.GetType(), Adapter: tc.adapter, Target: tc.target, Result: tc.result}
			if _, ok := capabilitiesProbeEvidenceFromResponse(request, response); ok {
				t.Fatalf("malformed evidence accepted: %#v", response)
			}
		})
	}
}

func TestProviderHealthOverlayCapabilitiesRecoveryRequiresLoweredBinding(t *testing.T) {
	svc, entry, record := newProviderHealthOverlayService(t, nil)
	record.Providers[0].Models = []string{"target-a", "target-b"}

	addBoundOverlayLease(t, svc.queue, "run-lower-b", overlayProviderID, overlayAdapter, "target-b", entry.ConnectionGeneration)
	svc.HandleReceivedRunLifecycleEvent(entry.NodeID, entry.ConnectionGeneration, unavailableRunEvent("run-lower-b", overlayAdapter, "target-b", 1))
	assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, true, 1)

	// Newer available evidence for target-a advances sequence high-water mark to 2 but does not recover target-b lowered overlay.
	if recovered := svc.queue.applyProviderProbeEvidence(entry.NodeID, entry.ConnectionGeneration, overlayAdapter, "target-a", runtime.ProviderStatusAvailable, 2, func() bool { return true }); recovered {
		t.Fatal("cross-target available probe recovered provider lowered for another target")
	}
	assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, true, 2)

	// Matching target-b available evidence at sequence 3 recovers the provider overlay.
	if recovered := svc.queue.applyProviderProbeEvidence(entry.NodeID, entry.ConnectionGeneration, overlayAdapter, "target-b", runtime.ProviderStatusAvailable, 3, func() bool { return true }); !recovered {
		t.Fatal("matching target-b available probe failed to recover provider overlay")
	}
	assertOverlayUnavailable(t, svc.queue, entry.ConnectionGeneration, false, 3)
}
