package node

import (
	"context"
	"errors"
	"fmt"
	"io"
	"strings"
	"sync"
	"testing"

	"github.com/prometheus/client_golang/prometheus"
	dto "github.com/prometheus/client_model/go"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"google.golang.org/protobuf/types/known/structpb"

	"iop/apps/node/internal/store"
	runtime "iop/packages/go/execution"
	iop "iop/proto/gen/iop"
)

// TestNodeLivenessObservability proves the bounded Node stall-observability
// contract on deterministic normalized and tunnel fixtures. It covers the four
// path/health outcomes (available/request-stalled and unavailable/provider-
// unhealthy on both paths), verifies exact metric families/labels and allow-
// listed values, asserts one dedicated log per claimed stall, and rejects
// high-cardinality raw values from both the metric labels and the structured
// log. It also proves unknown label normalization, logger panic containment, and
// repeated default Node construction.
func TestNodeLivenessObservability(t *testing.T) {
	t.Run("normalized/request-stalled", testNormalizedRequestStalled)
	t.Run("normalized/provider-unhealthy", testNormalizedProviderUnhealthy)
	t.Run("provider_tunnel/request-stalled", testTunnelRequestStalled)
	t.Run("provider_tunnel/provider-unhealthy", testTunnelProviderUnhealthy)
	t.Run("unknown-normalization", testUnknownNormalization)
	t.Run("failure-isolation", testFailureIsolation)
	t.Run("repeated-default-construction", testRepeatedDefaultConstruction)
}

type evidenceExpectation struct {
	path             string
	health           string
	classification   string
	fence            string
	counter          float64
	histogramCount   uint64
	idleMS           int64
	hostileSentinels []string
}

func assertNodeLivenessEvidence(t *testing.T, reg *prometheus.Registry, logs *testLogCore, exp evidenceExpectation) {
	t.Helper()

	gathered, err := reg.Gather()
	if err != nil {
		t.Fatalf("gather error: %v", err)
	}

	// 1. Counter assertion
	wantCounter := findMetric(gathered, "iop_node_response_stalls_total")
	if wantCounter == nil {
		t.Fatal("counter iop_node_response_stalls_total not found")
	}
	if len(wantCounter.GetMetric()) != 1 {
		t.Fatalf("counter metric series count = %d, want 1", len(wantCounter.GetMetric()))
	}
	gotCounter := wantCounter.GetMetric()[0]
	if gotCounter.GetCounter().GetValue() != exp.counter {
		t.Fatalf("counter value = %v, want %v", gotCounter.GetCounter().GetValue(), exp.counter)
	}
	counterLabelMap := dtoLabelMap(gotCounter.GetLabel())
	if len(counterLabelMap) != 4 {
		t.Fatalf("counter label count = %d, want 4 (labels=%v)", len(counterLabelMap), counterLabelMap)
	}
	assertLabel(t, counterLabelMap, "execution_path", exp.path)
	assertLabel(t, counterLabelMap, "provider_health", exp.health)
	assertLabel(t, counterLabelMap, "liveness_classification", exp.classification)
	assertLabel(t, counterLabelMap, "attempt_fence", exp.fence)

	// 2. Histogram assertion
	wantHist := findMetric(gathered, "iop_node_response_stall_duration_seconds")
	if wantHist == nil {
		t.Fatal("histogram iop_node_response_stall_duration_seconds not found")
	}
	if len(wantHist.GetMetric()) != 1 {
		t.Fatalf("histogram metric series count = %d, want 1", len(wantHist.GetMetric()))
	}
	gotHist := wantHist.GetMetric()[0]
	if gotHist.GetHistogram().GetSampleCount() != exp.histogramCount {
		t.Fatalf("histogram sample count = %d, want %d", gotHist.GetHistogram().GetSampleCount(), exp.histogramCount)
	}
	expectedSec := float64(exp.idleMS) / 1000.0
	if gotHist.GetHistogram().GetSampleSum() < expectedSec*0.99 || gotHist.GetHistogram().GetSampleSum() > expectedSec*1.01 {
		t.Fatalf("histogram sample sum = %v, want ~%v", gotHist.GetHistogram().GetSampleSum(), expectedSec)
	}
	histLabelMap := dtoLabelMap(gotHist.GetLabel())
	if len(histLabelMap) != 4 {
		t.Fatalf("histogram label count = %d, want 4 (labels=%v)", len(histLabelMap), histLabelMap)
	}
	assertLabel(t, histLabelMap, "execution_path", exp.path)
	assertLabel(t, histLabelMap, "provider_health", exp.health)
	assertLabel(t, histLabelMap, "liveness_classification", exp.classification)
	assertLabel(t, histLabelMap, "attempt_fence", exp.fence)

	// 3. Log entry assertion
	logs.mu.Lock()
	entries := make([]testLogEntry, len(logs.entries))
	copy(entries, logs.entries)
	logs.mu.Unlock()

	var matching []testLogEntry
	for _, entry := range entries {
		if entry.Message == "node_response_stall_observation" {
			matching = append(matching, entry)
		}
	}
	if len(matching) != 1 {
		t.Fatalf("dedicated stall observation log count = %d, want 1 (total log entries = %d)", len(matching), len(entries))
	}
	entry := matching[0]
	if entry.Level != zapcore.InfoLevel {
		t.Fatalf("log level = %v, want Info", entry.Level)
	}
	if len(entry.Fields) != 5 {
		t.Fatalf("log field count = %d, want 5 (fields=%+v)", len(entry.Fields), entry.Fields)
	}

	var foundPath, foundHealth, foundClass, foundFence bool
	var foundDuration int64
	var durationType zapcore.FieldType
	for _, f := range entry.Fields {
		switch f.Key {
		case "execution_path":
			foundPath = true
			if f.String != exp.path {
				t.Fatalf("field execution_path = %q, want %q", f.String, exp.path)
			}
		case "provider_health":
			foundHealth = true
			if f.String != exp.health {
				t.Fatalf("field provider_health = %q, want %q", f.String, exp.health)
			}
		case "liveness_classification":
			foundClass = true
			if f.String != exp.classification {
				t.Fatalf("field liveness_classification = %q, want %q", f.String, exp.classification)
			}
		case "attempt_fence":
			foundFence = true
			if f.String != exp.fence {
				t.Fatalf("field attempt_fence = %q, want %q", f.String, exp.fence)
			}
		case "idle_duration_ms":
			foundDuration = f.Integer
			durationType = f.Type
		default:
			t.Fatalf("unexpected log field key %q", f.Key)
		}
	}
	if !foundPath || !foundHealth || !foundClass || !foundFence {
		t.Fatalf("missing expected string fields in log entry: %+v", entry.Fields)
	}
	if durationType != zapcore.Int64Type {
		t.Fatalf("idle_duration_ms type = %v, want Int64Type (%v)", durationType, zapcore.Int64Type)
	}
	if foundDuration != exp.idleMS {
		t.Fatalf("idle_duration_ms value = %d, want %d", foundDuration, exp.idleMS)
	}

	// 4. Encoded JSON field assertions
	encoder := zapcore.NewJSONEncoder(zap.NewProductionEncoderConfig())
	item, err := encoder.EncodeEntry(zapcore.Entry{
		Level:   entry.Level,
		Message: entry.Message,
	}, entry.Fields)
	if err != nil {
		t.Fatalf("encode log entry: %v", err)
	}
	encodedJSON := item.String()
	expectedNumJSON := fmt.Sprintf(`"idle_duration_ms":%d`, exp.idleMS)
	if !strings.Contains(encodedJSON, expectedNumJSON) {
		t.Fatalf("encoded JSON log %q does not contain expected numeric field %q", encodedJSON, expectedNumJSON)
	}

	// 5. Hostile sentinel rejection
	for _, sentinel := range exp.hostileSentinels {
		if sentinel == "" {
			continue
		}
		for _, mf := range gathered {
			for _, m := range mf.GetMetric() {
				for _, l := range m.GetLabel() {
					if l.GetName() == sentinel || strings.Contains(l.GetName(), sentinel) {
						t.Fatalf("sentinel %q leaked into metric label name %q", sentinel, l.GetName())
					}
					if l.GetValue() == sentinel || strings.Contains(l.GetValue(), sentinel) {
						t.Fatalf("sentinel %q leaked into metric label value %q", sentinel, l.GetValue())
					}
				}
			}
		}
		if strings.Contains(encodedJSON, sentinel) {
			t.Fatalf("sentinel %q leaked into encoded JSON log %q", sentinel, encodedJSON)
		}
	}
}

func assertNoAdditionalTerminal[T any](t *testing.T, ch <-chan T) {
	t.Helper()
	select {
	case msg := <-ch:
		t.Fatalf("unexpected additional terminal message: %+v", msg)
	default:
	}
}

func testNormalizedRequestStalled(t *testing.T) {
	reg := prometheus.NewRegistry()
	logger, logs := newTestLogger()
	adapterName := "hostile-adapter-norm-avail"
	target := "hostile-target-norm-avail"
	runID := "obs-norm-avail-spoof-run-id"
	sessionID := "spoof-session-norm-avail"
	requestID := "spoof-request-id-norm-avail"
	prompt := "raw-prompt-norm-avail"
	response := "raw-response-norm-avail"
	credential := "raw-credential-norm-avail"

	sentinels := []string{runID, sessionID, adapterName, target, requestID, prompt, response, credential}

	adapter := newProbingWatchdogAdapter(adapterName)
	n := newNodeWithObserver(t, adapter, reg, logger)
	pipe := newWatchdogPipe(t)

	done := make(chan error, 1)
	go func() {
		done <- n.OnRunRequest(context.Background(), pipe.sess, &iop.RunRequest{
			RunId:                  runID,
			Adapter:                adapter.Name(),
			Target:                 target,
			SessionId:              sessionID,
			ResponseStallTimeoutMs: 500,
			Input:                  &structpb.Struct{Fields: map[string]*structpb.Value{"prompt": structpb.NewStringValue(prompt)}},
			Metadata:               map[string]string{"request_id": requestID, "response": response, "credential": credential},
		})
	}()
	call := <-adapter.runCalls
	clock := n.watchdogClock.(*manualAttemptClock)
	clock.waitTimer(t, 0).fire()
	waitContextCanceled(t, call.ctx)
	clock.waitTimer(t, 1)
	adapter.probeReturn <- probeReply{result: runtime.ProviderProbeResult{AdapterName: adapter.Name(), 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()["attempt_fence"] != "confirmed" {
		t.Fatalf("terminal = %+v", terminal)
	}

	assertNodeLivenessEvidence(t, reg, logs, evidenceExpectation{
		path:             "normalized",
		health:           "available",
		classification:   "request_stalled",
		fence:            "confirmed",
		counter:          1,
		histogramCount:   1,
		idleMS:           500,
		hostileSentinels: sentinels,
	})
}

func testNormalizedProviderUnhealthy(t *testing.T) {
	reg := prometheus.NewRegistry()
	logger, logs := newTestLogger()
	adapterName := "hostile-adapter-norm-unavail"
	target := "hostile-target-norm-unavail"
	runID := "obs-norm-unavail-spoof-run-id"
	sessionID := "spoof-session-norm-unavail"
	requestID := "spoof-request-id-norm-unavail"
	prompt := "raw-prompt-norm-unavail"
	response := "raw-response-norm-unavail"
	credential := "raw-credential-norm-unavail"

	sentinels := []string{runID, sessionID, adapterName, target, requestID, prompt, response, credential}

	adapter := newProbingWatchdogAdapter(adapterName)
	n := newNodeWithObserver(t, adapter, reg, logger)
	pipe := newWatchdogPipe(t)

	done := make(chan error, 1)
	go func() {
		done <- n.OnRunRequest(context.Background(), pipe.sess, &iop.RunRequest{
			RunId:                  runID,
			Adapter:                adapter.Name(),
			Target:                 target,
			SessionId:              sessionID,
			ResponseStallTimeoutMs: 500,
			Input:                  &structpb.Struct{Fields: map[string]*structpb.Value{"prompt": structpb.NewStringValue(prompt)}},
			Metadata:               map[string]string{"request_id": requestID, "response": response, "credential": credential},
		})
	}()
	call := <-adapter.runCalls
	clock := n.watchdogClock.(*manualAttemptClock)
	clock.waitTimer(t, 0).fire()
	waitContextCanceled(t, call.ctx)
	clock.waitTimer(t, 1)
	adapter.probeReturn <- probeReply{result: runtime.ProviderProbeResult{AdapterName: adapter.Name(), Target: target, Status: runtime.ProviderStatusUnavailable}}

	grace := clock.waitTimer(t, 1)
	grace.fire()

	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("run result = %v", err)
	}
	terminal := waitRunEvent(t, pipe.events)
	if terminal.GetType() != string(runtime.EventTypeError) || terminal.GetMetadata()["attempt_fence"] != "unconfirmed" {
		t.Fatalf("terminal = %+v", terminal)
	}

	assertNodeLivenessEvidence(t, reg, logs, evidenceExpectation{
		path:             "normalized",
		health:           "unavailable",
		classification:   "provider_unhealthy",
		fence:            "unconfirmed",
		counter:          1,
		histogramCount:   1,
		idleMS:           500,
		hostileSentinels: sentinels,
	})
}

func testTunnelRequestStalled(t *testing.T) {
	reg := prometheus.NewRegistry()
	logger, logs := newTestLogger()
	adapterName := "hostile-adapter-tun-avail"
	target := "hostile-target-tun-avail"
	runID := "obs-tun-avail-spoof-run-id"
	tunnelID := "tunnel-obs-spoof-id"
	sessionID := "spoof-session-tun-avail"
	requestID := "spoof-request-id-tun-avail"
	headerVal := "raw-header-tun-avail"
	bodyVal := "raw-body-tun-avail"
	responseVal := "raw-response-tun-avail"
	credentialVal := "raw-credential-tun-avail"

	sentinels := []string{runID, tunnelID, adapterName, target, sessionID, requestID, headerVal, bodyVal, responseVal, credentialVal}

	adapter := newProbingWatchdogAdapter(adapterName)
	n := newNodeWithObserver(t, adapter, reg, logger)
	pipe := newWatchdogPipe(t)

	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,
			SessionId:              sessionID,
			Headers:                map[string]string{"authorization": credentialVal, "request_id": requestID, "x-header": headerVal},
			Body:                   []byte(bodyVal),
			Metadata:               map[string]string{"response": responseVal},
			ResponseStallTimeoutMs: 500,
		})
	}()
	call := <-adapter.tunnelCalls
	if call.req.RunID != runID || call.req.TunnelID != tunnelID || call.req.Adapter != adapter.Name() || call.req.Target != target || call.req.SessionID != sessionID || call.req.Headers["authorization"] != credentialVal || call.req.Headers["request_id"] != requestID || call.req.Headers["x-header"] != headerVal || string(call.req.Body) != bodyVal || call.req.Metadata["response"] != responseVal {
		t.Fatalf("captured tunnel request mismatch: %#v", call.req)
	}
	clock := n.watchdogClock.(*manualAttemptClock)
	clock.waitTimer(t, 0).fire()
	waitContextCanceled(t, call.ctx)
	clock.waitTimer(t, 1)
	adapter.probeReturn <- probeReply{result: runtime.ProviderProbeResult{AdapterName: adapter.Name(), Target: target, Status: runtime.ProviderStatusAvailable}}
	adapter.tunnelReturn <- nil
	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("tunnel result = %v", err)
	}
	terminal := waitTunnelFrame(t, pipe.frames)
	if terminal.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
		t.Fatalf("terminal = %+v", terminal)
	}

	assertNodeLivenessEvidence(t, reg, logs, evidenceExpectation{
		path:             "provider_tunnel",
		health:           "available",
		classification:   "request_stalled",
		fence:            "confirmed",
		counter:          1,
		histogramCount:   1,
		idleMS:           500,
		hostileSentinels: sentinels,
	})
}

func testTunnelProviderUnhealthy(t *testing.T) {
	reg := prometheus.NewRegistry()
	logger, logs := newTestLogger()
	adapterName := "hostile-adapter-tun-unavail"
	target := "hostile-target-tun-unavail"
	runID := "obs-tun-unavail-spoof-run-id"
	tunnelID := "tunnel-unavail-spoof-id"
	sessionID := "spoof-session-tun-unavail"
	requestID := "spoof-request-id-tun-unavail"
	headerVal := "raw-header-tun-unavail"
	bodyVal := "raw-body-tun-unavail"
	responseVal := "raw-response-tun-unavail"
	credentialVal := "raw-credential-tun-unavail"

	sentinels := []string{runID, tunnelID, adapterName, target, sessionID, requestID, headerVal, bodyVal, responseVal, credentialVal}

	adapter := newProbingWatchdogAdapter(adapterName)
	n := newNodeWithObserver(t, adapter, reg, logger)
	pipe := newWatchdogPipe(t)

	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,
			SessionId:              sessionID,
			Headers:                map[string]string{"authorization": credentialVal, "request_id": requestID, "x-header": headerVal},
			Body:                   []byte(bodyVal),
			Metadata:               map[string]string{"response": responseVal},
			ResponseStallTimeoutMs: 500,
		})
	}()
	call := <-adapter.tunnelCalls
	if call.req.RunID != runID || call.req.TunnelID != tunnelID || call.req.Adapter != adapter.Name() || call.req.Target != target || call.req.SessionID != sessionID || call.req.Headers["authorization"] != credentialVal || call.req.Headers["request_id"] != requestID || call.req.Headers["x-header"] != headerVal || string(call.req.Body) != bodyVal || call.req.Metadata["response"] != responseVal {
		t.Fatalf("captured tunnel request mismatch: %#v", call.req)
	}
	clock := n.watchdogClock.(*manualAttemptClock)
	clock.waitTimer(t, 0).fire()
	waitContextCanceled(t, call.ctx)
	clock.waitTimer(t, 1)
	adapter.probeReturn <- probeReply{result: runtime.ProviderProbeResult{AdapterName: adapter.Name(), Target: target, Status: runtime.ProviderStatusUnavailable}}

	grace := clock.waitTimer(t, 1)
	grace.fire()

	if err := <-done; err != errProviderResponseStalled {
		t.Fatalf("tunnel result = %v", err)
	}
	terminal := waitTunnelFrame(t, pipe.frames)
	if terminal.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR || terminal.GetMetadata()["attempt_fence"] != "unconfirmed" {
		t.Fatalf("terminal = %+v", terminal)
	}

	assertNodeLivenessEvidence(t, reg, logs, evidenceExpectation{
		path:             "provider_tunnel",
		health:           "unavailable",
		classification:   "provider_unhealthy",
		fence:            "unconfirmed",
		counter:          1,
		histogramCount:   1,
		idleMS:           500,
		hostileSentinels: sentinels,
	})
}

func testUnknownNormalization(t *testing.T) {
	labels := normalizeNodeLivenessLabels("invalid_path", stallObservation{
		health: HealthProbeEvidence{
			Status: runtime.ProviderStatus("invalid_status"),
			Health: runtime.ProviderHealth("invalid_health"),
		},
		fence: "invalid_fence",
	})
	want := [4]string{"unknown", "unknown", "health_unknown", "unknown"}
	if labels != want {
		t.Fatalf("normalizeNodeLivenessLabels = %v, want %v", labels, want)
	}
}

type panickingLogCore struct{}

func (p *panickingLogCore) Enabled(zapcore.Level) bool    { return true }
func (p *panickingLogCore) With([]zap.Field) zapcore.Core { return p }
func (p *panickingLogCore) Check(e zapcore.Entry, ce *zapcore.CheckedEntry) *zapcore.CheckedEntry {
	return ce.AddCore(e, p)
}
func (p *panickingLogCore) Write(zapcore.Entry, []zap.Field) error {
	panic("simulated logger panic")
}
func (p *panickingLogCore) Sync() error { return nil }

func testFailureIsolation(t *testing.T) {
	t.Run("normalized", func(t *testing.T) {
		reg := prometheus.NewRegistry()
		panickingLogger := zap.New(&panickingLogCore{})
		adapter := newProbingWatchdogAdapter("obs-panic-norm")
		n := newNodeWithObserver(t, adapter, reg, zap.NewNop())
		n.liveness.logger = panickingLogger
		pipe := newWatchdogPipe(t)

		done := make(chan error, 1)
		go func() {
			done <- n.OnRunRequest(context.Background(), pipe.sess, &iop.RunRequest{
				RunId:                  "obs-panic-norm",
				Adapter:                adapter.Name(),
				Target:                 "target",
				ResponseStallTimeoutMs: 500,
			})
		}()
		call := <-adapter.runCalls
		clock := n.watchdogClock.(*manualAttemptClock)
		clock.waitTimer(t, 0).fire()
		waitContextCanceled(t, call.ctx)
		clock.waitTimer(t, 1)
		adapter.probeReturn <- probeReply{result: runtime.ProviderProbeResult{AdapterName: adapter.Name(), Target: "target", Status: runtime.ProviderStatusAvailable}}
		adapter.runReturn <- nil
		if err := <-done; err != errProviderResponseStalled {
			t.Fatalf("run result = %v, want errProviderResponseStalled", err)
		}
		terminal := waitRunEvent(t, pipe.events)
		if terminal.GetType() != string(runtime.EventTypeError) || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
			t.Fatalf("terminal = %+v", terminal)
		}
		assertNoAdditionalTerminal(t, pipe.events)
	})

	t.Run("tunnel", func(t *testing.T) {
		reg := prometheus.NewRegistry()
		panickingLogger := zap.New(&panickingLogCore{})
		adapter := newProbingWatchdogAdapter("obs-panic-tun")
		n := newNodeWithObserver(t, adapter, reg, zap.NewNop())
		n.liveness.logger = panickingLogger
		pipe := newWatchdogPipe(t)

		done := make(chan error, 1)
		go func() {
			done <- n.OnProviderTunnelRequest(context.Background(), pipe.sess, &iop.ProviderTunnelRequest{
				RunId:                  "obs-panic-tun",
				TunnelId:               "tunnel-panic",
				Adapter:                adapter.Name(),
				Target:                 "target",
				ResponseStallTimeoutMs: 500,
			})
		}()
		call := <-adapter.tunnelCalls
		clock := n.watchdogClock.(*manualAttemptClock)
		clock.waitTimer(t, 0).fire()
		waitContextCanceled(t, call.ctx)
		clock.waitTimer(t, 1)
		adapter.probeReturn <- probeReply{result: runtime.ProviderProbeResult{AdapterName: adapter.Name(), Target: "target", Status: runtime.ProviderStatusAvailable}}
		adapter.tunnelReturn <- nil
		if err := <-done; err != errProviderResponseStalled {
			t.Fatalf("tunnel result = %v, want errProviderResponseStalled", err)
		}
		terminal := waitTunnelFrame(t, pipe.frames)
		if terminal.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
			t.Fatalf("terminal = %+v", terminal)
		}
		assertNoAdditionalTerminal(t, pipe.frames)
	})
}

func testRepeatedDefaultConstruction(t *testing.T) {
	st, err := store.New(":memory:", zap.NewNop())
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = st.Close() }()

	for i := 0; i < 50; i++ {
		_ = New("node-dup-"+string(rune('a'+i%26)), &noopRouter{}, st, 0, io.Discard, zap.NewNop(), nil)
	}
}

// --- Test helpers ---

type testLogEntry struct {
	Level   zapcore.Level
	Message string
	Fields  []zap.Field
}

type testLogCore struct {
	mu      sync.Mutex
	entries []testLogEntry
}

func newTestLogCore() *testLogCore {
	return &testLogCore{}
}

func (c *testLogCore) Enabled(lvl zapcore.Level) bool {
	return true
}

func (c *testLogCore) With(fields []zap.Field) zapcore.Core {
	return c
}

func (c *testLogCore) Check(entry zapcore.Entry, ce *zapcore.CheckedEntry) *zapcore.CheckedEntry {
	if c.Enabled(entry.Level) {
		return ce.AddCore(entry, c)
	}
	return ce
}

func (c *testLogCore) Write(entry zapcore.Entry, fields []zap.Field) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.entries = append(c.entries, testLogEntry{
		Level:   entry.Level,
		Message: entry.Message,
		Fields:  fields,
	})
	return nil
}

func (c *testLogCore) Sync() error { return nil }

func newTestLogger() (*zap.Logger, *testLogCore) {
	core := newTestLogCore()
	logger := zap.New(core)
	return logger, core
}

func findMetric(gathered []*dto.MetricFamily, name string) *dto.MetricFamily {
	for _, mf := range gathered {
		if mf.GetName() == name {
			return mf
		}
	}
	return nil
}

func dtoLabelMap(labels []*dto.LabelPair) map[string]string {
	m := make(map[string]string, len(labels))
	for _, l := range labels {
		m[l.GetName()] = l.GetValue()
	}
	return m
}

func assertLabel(t *testing.T, labels map[string]string, name, want string) {
	t.Helper()
	got, ok := labels[name]
	if !ok {
		t.Fatalf("label %q missing, labels=%v", name, labels)
	}
	if got != want {
		t.Fatalf("label %s = %q, want %q", name, got, want)
	}
}

type noopRouter struct{}

func (r *noopRouter) Resolve(_ context.Context, _ runtime.RunRequest) (runtime.ExecutionSpec, error) {
	return runtime.ExecutionSpec{}, errors.New("noop")
}
func (r *noopRouter) ResolveAdapter(_ context.Context, _ runtime.RunRequest) (runtime.ExecutionSpec, runtime.Provider, error) {
	return runtime.ExecutionSpec{}, nil, errors.New("noop")
}
func (r *noopRouter) LookupAdapter(_ string) (runtime.Provider, error) {
	return nil, errors.New("noop")
}
func (r *noopRouter) GetAdapter(_ string) (runtime.Provider, bool) {
	return nil, false
}

func newNodeWithObserver(t *testing.T, adapter runtime.ProviderTunnelAdapter, reg prometheus.Registerer, logger *zap.Logger) *Node {
	t.Helper()
	st, err := store.New(":memory:", zap.NewNop())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = st.Close() })
	n := New("node-obs", &watchdogRouter{adapter: adapter}, st, 0, io.Discard, logger, nil)
	n.watchdogClock = newManualAttemptClock()
	n.liveness = newNodeLivenessObserverForTest(logger, reg)
	return n
}
