package node

import (
	"context"
	"errors"
	"testing"
	"time"

	"google.golang.org/protobuf/proto"

	runtime "iop/packages/go/execution"
	iop "iop/proto/gen/iop"
)

// TestStalledTerminalsCloneSafeMetadata proves the normalized stall terminal
// clones Node-owned metadata into the Failure map, the event map, and the
// protobuf map without sharing a mutable alias, and that caller-provided
// spoof values in the execution spec never leak into the terminal.
func TestStalledTerminalsCloneSafeMetadata(t *testing.T) {
	spec := runtime.ExecutionSpec{RunID: "node-run", Adapter: "adapter", Target: "target", Metadata: map[string]string{"run_id": "spoof", "attempt_id": "spoof", "provider_health": "spoof", "liveness_classification": "spoof", "health_observation_seq": "spoof", "recovery_eligible": "true", "secret": "leak"}}
	obs := stallObservation{fence: "confirmed", idle: 2 * time.Second, health: HealthProbeEvidence{Health: runtime.RequestStalled, Status: runtime.ProviderStatusAvailable}, seq: 7, hasSeq: true}
	event := stalledRuntimeEvent(spec, obs)
	if event.Failure.Code != runtime.FailureCodeResponseStalled || !event.Failure.Retryable {
		t.Fatalf("failure = %#v", event.Failure)
	}
	if event.Metadata["run_id"] != "node-run" || event.Metadata["attempt_id"] != "node-run" || event.Metadata["recovery_eligible"] != "" || event.Metadata["secret"] != "" {
		t.Fatalf("unsafe normalized metadata = %#v", event.Metadata)
	}
	// Node-owned health evidence and the connection-scoped observation sequence
	// overwrite any caller-provided spoof values.
	if event.Metadata["provider_health"] != "available" || event.Metadata["liveness_classification"] != "request_stalled" || event.Metadata["health_observation_seq"] != "7" {
		t.Fatalf("health evidence not applied to normalized metadata = %#v", event.Metadata)
	}
	sender := &recordingProtoSender{}
	sink := &sessionSink{sess: sender}
	if err := sink.Emit(context.Background(), event); err != nil {
		t.Fatal(err)
	}
	wire := sender.snapshot()[0].(*iop.RunEvent)
	// The Failure map, event map, and protobuf map must carry identical safe
	// values without sharing a mutable alias.
	for _, key := range []string{"attempt_fence", "provider_health", "liveness_classification", "health_observation_seq"} {
		if event.Failure.Metadata[key] != event.Metadata[key] || wire.GetMetadata()[key] != event.Metadata[key] {
			t.Fatalf("normalized failure/event/protobuf disagree on %q: %q / %q / %q", key, event.Failure.Metadata[key], event.Metadata[key], wire.GetMetadata()[key])
		}
	}
	if wire.GetFailure() == nil || wire.GetFailure().GetCode() != "response_stalled" || !wire.GetFailure().GetRetryable() {
		t.Fatalf("wire.Failure mismatch: %#v", wire.GetFailure())
	}
	if wire.GetFailure().GetMetadata()["recovery_eligible"] != "" || wire.GetFailure().GetMetadata()["secret"] != "" {
		t.Fatalf("wire.Failure contains unsafe metadata: %#v", wire.GetFailure().GetMetadata())
	}
	event.Metadata["attempt_fence"] = "mutated"
	if event.Failure.Metadata["attempt_fence"] != "confirmed" || wire.GetMetadata()["attempt_fence"] != "confirmed" || wire.GetFailure().GetMetadata()["attempt_fence"] != "confirmed" {
		t.Fatal("normalized failure, event, and protobuf metadata alias")
	}

	tunnelObs := stallObservation{fence: "unconfirmed", idle: 2 * time.Second, health: HealthProbeEvidence{Health: runtime.ProviderUnhealthy, Status: runtime.ProviderStatusUnavailable}, seq: 8, hasSeq: true}
	frame := stalledTunnelFrame(runtime.ProviderTunnelRequest{RunID: "node-run", Adapter: "adapter", Target: "target", Metadata: spec.Metadata}, tunnelObs)
	protoFrame := tunnelFrameToProto(frame, "node", "alias")
	if protoFrame.GetMetadata()["provider_health"] != "unavailable" || protoFrame.GetMetadata()["liveness_classification"] != "provider_unhealthy" || protoFrame.GetMetadata()["health_observation_seq"] != "8" {
		t.Fatalf("tunnel health evidence not applied = %#v", protoFrame.GetMetadata())
	}
	if protoFrame.GetFailure() == nil || protoFrame.GetFailure().GetCode() != "response_stalled" || protoFrame.GetFailure().GetRetryable() {
		t.Fatalf("protoFrame.Failure mismatch: %#v", protoFrame.GetFailure())
	}
	if protoFrame.GetFailure().GetMetadata()["provider_health"] != "unavailable" || protoFrame.GetFailure().GetMetadata()["liveness_classification"] != "provider_unhealthy" || protoFrame.GetFailure().GetMetadata()["health_observation_seq"] != "8" {
		t.Fatalf("protoFrame.Failure metadata mismatch = %#v", protoFrame.GetFailure().GetMetadata())
	}
	if protoFrame.GetFailure().GetMetadata()["recovery_eligible"] != "" || protoFrame.GetFailure().GetMetadata()["secret"] != "" {
		t.Fatalf("protoFrame.Failure contains unsafe metadata = %#v", protoFrame.GetFailure().GetMetadata())
	}
	frame.Metadata["attempt_fence"] = "mutated"
	if protoFrame.GetMetadata()["attempt_fence"] != "unconfirmed" || protoFrame.GetMetadata()["recovery_eligible"] != "" || protoFrame.GetMetadata()["secret"] != "" {
		t.Fatalf("unsafe or aliased tunnel metadata = %#v", protoFrame.GetMetadata())
	}
}

// TestStallMetadataMapsThreeWayHealthEvidence proves the joined metadata carries
// each of the three stable health outcomes, fails closed to unknown on zero
// evidence, and includes the connection-scoped sequence only when one was
// allocated.
func TestStallMetadataMapsThreeWayHealthEvidence(t *testing.T) {
	cases := []struct {
		name           string
		obs            stallObservation
		wantHealth     string
		wantClass      string
		wantSeqPresent bool
		wantSeq        string
	}{
		{"available maps to request_stalled", stallObservation{fence: "confirmed", health: HealthProbeEvidence{Health: runtime.RequestStalled, Status: runtime.ProviderStatusAvailable}, seq: 1, hasSeq: true}, "available", "request_stalled", true, "1"},
		{"unavailable maps to provider_unhealthy", stallObservation{fence: "unconfirmed", health: HealthProbeEvidence{Health: runtime.ProviderUnhealthy, Status: runtime.ProviderStatusUnavailable}, seq: 2, hasSeq: true}, "unavailable", "provider_unhealthy", true, "2"},
		{"unknown status maps to health_unknown", stallObservation{fence: "confirmed", health: HealthProbeEvidence{Health: runtime.HealthUnknown, Status: runtime.ProviderStatusUnknown}, seq: 3, hasSeq: true}, "unknown", "health_unknown", true, "3"},
		{"zero evidence fails closed and omits seq", stallObservation{fence: "unconfirmed"}, "unknown", "health_unknown", false, ""},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			metadata := stallMetadata("run", "adapter", "target", tc.obs)
			if metadata["failure_code"] != string(runtime.FailureCodeResponseStalled) {
				t.Fatalf("failure_code = %q", metadata["failure_code"])
			}
			if metadata["provider_health"] != tc.wantHealth || metadata["liveness_classification"] != tc.wantClass {
				t.Fatalf("health = %q, classification = %q", metadata["provider_health"], metadata["liveness_classification"])
			}
			if metadata["attempt_fence"] != tc.obs.fence || metadata["run_id"] != "run" || metadata["attempt_id"] != "run" || metadata["adapter"] != "adapter" || metadata["target"] != "target" {
				t.Fatalf("ownership metadata = %#v", metadata)
			}
			seq, present := metadata["health_observation_seq"]
			if present != tc.wantSeqPresent || seq != tc.wantSeq {
				t.Fatalf("health_observation_seq present=%v value=%q, want present=%v value=%q", present, seq, tc.wantSeqPresent, tc.wantSeq)
			}
		})
	}
}

// TestStallMetadataFailsClosedOnContradictoryProbeStatus proves the terminal
// pair never emits a definitive provider status paired with an inconclusive
// classification. When the raw probe reports available or unavailable but the
// normalized classification is HealthUnknown (identity mismatch, timeout, or
// probe error), both provider_health and liveness_classification must resolve
// to unknown/health_unknown on both the normalized and tunnel terminal paths.
func TestStallMetadataFailsClosedOnContradictoryProbeStatus(t *testing.T) {
	contradictory := []struct {
		name string
		obs  stallObservation
	}{
		{"raw available with unknown classification", stallObservation{fence: "confirmed", health: HealthProbeEvidence{Health: runtime.HealthUnknown, Status: runtime.ProviderStatusAvailable}, seq: 10, hasSeq: true}},
		{"raw unavailable with unknown classification", stallObservation{fence: "unconfirmed", health: HealthProbeEvidence{Health: runtime.HealthUnknown, Status: runtime.ProviderStatusUnavailable}, seq: 11, hasSeq: true}},
	}
	for _, tc := range contradictory {
		t.Run(tc.name, func(t *testing.T) {
			// Normalized terminal path.
			metadata := stallMetadata("run", "adapter", "target", tc.obs)
			if metadata["provider_health"] != string(runtime.ProviderStatusUnknown) {
				t.Fatalf("normalized provider_health = %q, want %q", metadata["provider_health"], runtime.ProviderStatusUnknown)
			}
			if metadata["liveness_classification"] != string(runtime.HealthUnknown) {
				t.Fatalf("normalized liveness_classification = %q, want %q", metadata["liveness_classification"], runtime.HealthUnknown)
			}
			if metadata["failure_code"] != string(runtime.FailureCodeResponseStalled) {
				t.Fatalf("failure_code = %q", metadata["failure_code"])
			}

			// Tunnel terminal path via stalledTunnelFrame.
			tunnelObs := tc.obs
			req := runtime.ProviderTunnelRequest{RunID: "run", Adapter: "adapter", Target: "target"}
			frame := stalledTunnelFrame(req, tunnelObs)
			protoFrame := tunnelFrameToProto(frame, "node", "alias")
			if protoFrame.GetMetadata()["provider_health"] != string(runtime.ProviderStatusUnknown) {
				t.Fatalf("tunnel provider_health = %q, want %q", protoFrame.GetMetadata()["provider_health"], runtime.ProviderStatusUnknown)
			}
			if protoFrame.GetMetadata()["liveness_classification"] != string(runtime.HealthUnknown) {
				t.Fatalf("tunnel liveness_classification = %q, want %q", protoFrame.GetMetadata()["liveness_classification"], runtime.HealthUnknown)
			}
		})
	}
}

// TestRunWatchdogJoinsHealthEvidence proves the normalized stall terminal joins
// the bounded exact-target probe result. The probe runs on an independent,
// still-live context after the request was canceled, and its three-way outcome
// reaches the terminal without changing the confirmed fence or reviving the run.
func TestRunWatchdogJoinsHealthEvidence(t *testing.T) {
	cases := []struct {
		name       string
		reply      probeReply
		wantHealth string
		wantClass  string
	}{
		{"available maps to request_stalled", probeReply{result: runtime.ProviderProbeResult{AdapterName: "run-health", Target: "target", Status: runtime.ProviderStatusAvailable}}, "available", "request_stalled"},
		{"unavailable maps to provider_unhealthy", probeReply{result: runtime.ProviderProbeResult{AdapterName: "run-health", Target: "target", Status: runtime.ProviderStatusUnavailable}}, "unavailable", "provider_unhealthy"},
		{"probe error fails closed to unknown", probeReply{err: errors.New("probe transport failure")}, "unknown", "health_unknown"},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			clock := newManualAttemptClock()
			adapter := newProbingWatchdogAdapter("run-health")
			n := newWatchdogNode(t, adapter, clock)
			pipe := newWatchdogPipe(t)
			done := make(chan error, 1)
			go func() {
				done <- n.OnRunRequest(context.Background(), pipe.sess, &iop.RunRequest{RunId: "run-health", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
			}()
			call := <-adapter.runCalls
			clock.waitTimer(t, 0).fire()
			waitContextCanceled(t, call.ctx)
			probe := <-adapter.probeCalls
			if probe.target != "target" {
				t.Fatalf("probe target = %q", probe.target)
			}
			if probe.ctx.Err() != nil {
				t.Fatal("health probe inherited the canceled request context")
			}
			grace := clock.waitTimer(t, 1)
			requireTimerDurations(t, grace, defaultAttemptCloseGrace)
			adapter.probeReturn <- tc.reply
			adapter.runReturn <- nil // provider returns within grace -> confirmed
			if err := <-done; err != errProviderResponseStalled {
				t.Fatalf("run result = %v", err)
			}
			terminal := waitRunEvent(t, pipe.events)
			meta := terminal.GetMetadata()
			if terminal.GetType() != string(runtime.EventTypeError) || meta["attempt_fence"] != "confirmed" {
				t.Fatalf("terminal = %+v", terminal)
			}
			if meta["provider_health"] != tc.wantHealth || meta["liveness_classification"] != tc.wantClass {
				t.Fatalf("health evidence = %q/%q, want %q/%q", meta["provider_health"], meta["liveness_classification"], tc.wantHealth, tc.wantClass)
			}
			if meta["health_observation_seq"] != "1" {
				t.Fatalf("health_observation_seq = %q, want 1", meta["health_observation_seq"])
			}
			// Exactly one terminal; late provider output remains fenced.
			_ = call.sink.Emit(context.Background(), runtime.RuntimeEvent{RunID: "run-health", Type: runtime.EventTypeDelta, Delta: "late"})
			select {
			case extra := <-pipe.events:
				t.Fatalf("late or duplicate event = %+v", extra)
			default:
			}
		})
	}
}

// TestTunnelWatchdogJoinsHealthEvidence proves the tunnel ERROR terminal joins
// the bounded probe result under an unconfirmed close fence while retaining
// provider-owned cleanup until the provider actually returns.
func TestTunnelWatchdogJoinsHealthEvidence(t *testing.T) {
	clock := newManualAttemptClock()
	adapter := newProbingWatchdogAdapter("tunnel-health")
	n := newWatchdogNode(t, adapter, clock)
	pipe := newWatchdogPipe(t)
	done := make(chan error, 1)
	go func() {
		done <- n.OnProviderTunnelRequest(context.Background(), pipe.sess, &iop.ProviderTunnelRequest{RunId: "tunnel-health", TunnelId: "tunnel", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
	}()
	call := <-adapter.tunnelCalls
	clock.waitTimer(t, 0).fire()
	waitContextCanceled(t, call.ctx)
	probe := <-adapter.probeCalls
	if probe.ctx.Err() != nil {
		t.Fatal("tunnel health probe inherited the canceled request context")
	}
	grace := clock.waitTimer(t, 1)
	requireTimerDurations(t, grace, defaultAttemptCloseGrace)
	adapter.probeReturn <- probeReply{result: runtime.ProviderProbeResult{AdapterName: "tunnel-health", Target: "target", Status: runtime.ProviderStatusAvailable}}
	grace.fire() // provider does not return within grace -> unconfirmed
	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("tunnel result = %v", err)
	}
	terminal := waitTunnelFrame(t, pipe.frames)
	meta := terminal.GetMetadata()
	if terminal.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR || meta["attempt_fence"] != "unconfirmed" {
		t.Fatalf("terminal = %+v", terminal)
	}
	if meta["provider_health"] != "available" || meta["liveness_classification"] != "request_stalled" || meta["health_observation_seq"] != "1" {
		t.Fatalf("tunnel health evidence = %#v", meta)
	}
	if activeAdapterAttempts(n, adapter.Name()) != 1 || !n.runs.hasAnyActiveRuns() {
		t.Fatal("unconfirmed tunnel released ownership before provider return")
	}
	adapter.tunnelReturn <- nil
	waitForOwnershipRelease(t, n, adapter.Name(), "tunnel provider return did not release ownership")
	select {
	case extra := <-pipe.frames:
		t.Fatalf("late or duplicate frame = %+v", extra)
	default:
	}
}

// TestRunWatchdogProbeEvidenceDoesNotResetProgress proves a positive
// availability probe is evidence only: it never suppresses the stall terminal,
// arms another activity timer, or revives local ownership.
func TestRunWatchdogProbeEvidenceDoesNotResetProgress(t *testing.T) {
	clock := newManualAttemptClock()
	adapter := newProbingWatchdogAdapter("run-noreset")
	n := newWatchdogNode(t, adapter, clock)
	pipe := newWatchdogPipe(t)
	done := make(chan error, 1)
	go func() {
		done <- n.OnRunRequest(context.Background(), pipe.sess, &iop.RunRequest{RunId: "run-noreset", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
	}()
	call := <-adapter.runCalls
	clock.waitTimer(t, 0).fire()
	waitContextCanceled(t, call.ctx)
	<-adapter.probeCalls
	clock.waitTimer(t, 1)
	adapter.probeReturn <- probeReply{result: runtime.ProviderProbeResult{AdapterName: "run-noreset", Target: "target", Status: runtime.ProviderStatusAvailable}}
	adapter.runReturn <- nil
	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("run result = %v", err)
	}
	terminal := waitRunEvent(t, pipe.events)
	if terminal.GetType() != string(runtime.EventTypeError) || terminal.GetMetadata()["failure_code"] != string(runtime.FailureCodeResponseStalled) {
		t.Fatalf("available probe suppressed the stall terminal: %+v", terminal)
	}
	// Only the stall and close-grace timers were armed; probe evidence never reset
	// the activity watchdog.
	if clock.count() != 2 {
		t.Fatalf("probe evidence armed an extra timer: %d timers", clock.count())
	}
	if activeAdapterAttempts(n, adapter.Name()) != 0 || n.runs.hasAnyActiveRuns() {
		t.Fatal("available probe revived local ownership")
	}
}

// TestWatchdogHealthObservationSeqIsConnectionScoped proves the sequence source
// is shared by normalized and tunnel attempts on one Session, increases per
// finalized observation, and resets on a new connection.
func TestWatchdogHealthObservationSeqIsConnectionScoped(t *testing.T) {
	clock := newManualAttemptClock()
	adapter := newControlledWatchdogAdapter("seq-adapter")
	n := newWatchdogNode(t, adapter, clock)
	pipe := newWatchdogPipe(t)

	runSeq := driveNormalizedConfirmedStall(t, n, pipe, adapter, clock, "seq-run", 0)
	if runSeq != "1" {
		t.Fatalf("first normalized observation seq = %q, want 1", runSeq)
	}
	tunnelSeq := driveTunnelConfirmedStall(t, n, pipe, adapter, clock, "seq-tunnel", "tunnel", 2)
	if tunnelSeq != "2" {
		t.Fatalf("tunnel observation seq on same connection = %q, want 2", tunnelSeq)
	}

	pipe2 := newWatchdogPipe(t)
	resetSeq := driveNormalizedConfirmedStall(t, n, pipe2, adapter, clock, "seq-run-2", 4)
	if resetSeq != "1" {
		t.Fatalf("new-connection observation seq = %q, want 1", resetSeq)
	}
}

func driveNormalizedConfirmedStall(t *testing.T, n *Node, pipe *watchdogPipe, adapter *controlledWatchdogAdapter, clock *manualAttemptClock, runID string, firstTimer int) string {
	t.Helper()
	done := make(chan error, 1)
	go func() {
		done <- n.OnRunRequest(context.Background(), pipe.sess, &iop.RunRequest{RunId: runID, Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
	}()
	call := <-adapter.runCalls
	clock.waitTimer(t, firstTimer).fire()
	waitContextCanceled(t, call.ctx)
	clock.waitTimer(t, firstTimer+1)
	adapter.runReturn <- nil
	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("run result = %v", err)
	}
	return waitRunEvent(t, pipe.events).GetMetadata()["health_observation_seq"]
}

func driveTunnelConfirmedStall(t *testing.T, n *Node, pipe *watchdogPipe, adapter *controlledWatchdogAdapter, clock *manualAttemptClock, runID, tunnelID string, firstTimer int) string {
	t.Helper()
	done := make(chan error, 1)
	go func() {
		done <- n.OnProviderTunnelRequest(context.Background(), pipe.sess, &iop.ProviderTunnelRequest{RunId: runID, TunnelId: tunnelID, Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
	}()
	call := <-adapter.tunnelCalls
	clock.waitTimer(t, firstTimer).fire()
	waitContextCanceled(t, call.ctx)
	clock.waitTimer(t, firstTimer+1)
	adapter.tunnelReturn <- nil
	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("tunnel result = %v", err)
	}
	return waitTunnelFrame(t, pipe.frames).GetMetadata()["health_observation_seq"]
}

// TestWatchdogOmitsHealthObservationSeqWithoutBoundSession proves an internal or
// unbound execution path omits the sequence key entirely while health evidence
// still fails closed to unknown.
func TestWatchdogOmitsHealthObservationSeqWithoutBoundSession(t *testing.T) {
	clock := newManualAttemptClock()
	adapter := newControlledWatchdogAdapter("tunnel-nilseq")
	n := newWatchdogNode(t, adapter, clock)
	ticket, err := n.admissionFor(adapter.Name(), runtime.Capabilities{MaxConcurrency: 1}).acquire()
	if err != nil {
		t.Fatal(err)
	}
	tr := runtime.ProviderTunnelRequest{RunID: "tunnel-nilseq", TunnelID: "tunnel", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMS: 1000}
	execCtx, cancel := context.WithCancel(context.Background())
	h := &runHandle{runID: tr.RunID, adapter: tr.Adapter, target: tr.Target, cancel: cancel, done: make(chan struct{})}
	n.runs.register(h)
	sender := &recordingProtoSender{}
	sink := &tunnelSink{sess: sender, observer: newAttemptObserver(clock, time.Second)}
	done := make(chan error, 1)
	go func() { done <- n.executeTunnelAttempt(execCtx, cancel, adapter, tr, sink, ticket, h, nil, nil, nil) }()
	call := <-adapter.tunnelCalls
	clock.waitTimer(t, 0).fire()
	waitContextCanceled(t, call.ctx)
	clock.waitTimer(t, 1)
	adapter.tunnelReturn <- nil // confirmed
	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("tunnel result = %v", err)
	}
	frames := sender.snapshot()
	if len(frames) != 1 {
		t.Fatalf("emitted frames = %d, want 1", len(frames))
	}
	meta := frames[0].(*iop.ProviderTunnelFrame).GetMetadata()
	if _, present := meta["health_observation_seq"]; present {
		t.Fatalf("unbound-session terminal carried a sequence: %#v", meta)
	}
	if meta["provider_health"] != "unknown" || meta["liveness_classification"] != "health_unknown" {
		t.Fatalf("nil-probe health = %#v", meta)
	}
}

func TestStallMetadataNormalizedAndTunnelParity(t *testing.T) {
	spec := runtime.ExecutionSpec{RunID: "parity-run", Adapter: "ollama", Target: "llama3"}
	obs := stallObservation{fence: "confirmed", idle: 3 * time.Second, health: HealthProbeEvidence{Health: runtime.RequestStalled, Status: runtime.ProviderStatusAvailable}, seq: 10, hasSeq: true}

	normEvent := stalledRuntimeEvent(spec, obs)
	normProto := runEventToProto(normEvent, "node-1", "session-1", false)

	tunnelReq := runtime.ProviderTunnelRequest{RunID: "parity-run", TunnelID: "tunnel-1", Adapter: "ollama", Target: "llama3"}
	tunnelFrame := stalledTunnelFrame(tunnelReq, obs)
	tunnelProto := tunnelFrameToProto(tunnelFrame, "node-1", "alias-1")

	if normProto.GetFailure() == nil || tunnelProto.GetFailure() == nil {
		t.Fatalf("expected non-nil failure on both paths: norm=%#v tunnel=%#v", normProto.GetFailure(), tunnelProto.GetFailure())
	}
	if normProto.GetFailure().GetCode() != tunnelProto.GetFailure().GetCode() {
		t.Fatalf("code mismatch: norm=%q tunnel=%q", normProto.GetFailure().GetCode(), tunnelProto.GetFailure().GetCode())
	}
	if normProto.GetFailure().GetRetryable() != tunnelProto.GetFailure().GetRetryable() {
		t.Fatalf("retryable mismatch: norm=%v tunnel=%v", normProto.GetFailure().GetRetryable(), tunnelProto.GetFailure().GetRetryable())
	}
	for _, key := range []string{"failure_code", "provider_health", "liveness_classification", "idle_duration_ms", "run_id", "attempt_id", "attempt_fence", "adapter", "target", "health_observation_seq"} {
		if normProto.GetFailure().GetMetadata()[key] != tunnelProto.GetFailure().GetMetadata()[key] {
			t.Fatalf("metadata key %q mismatch: norm=%q tunnel=%q", key, normProto.GetFailure().GetMetadata()[key], tunnelProto.GetFailure().GetMetadata()[key])
		}
	}
}

// Ensure proto import is used by the test file (kept for compatibility).
var _ = proto.Clone
