package openai

import (
	"context"
	"encoding/hex"
	"errors"
	"reflect"
	"sort"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/testutil"
	dto "github.com/prometheus/client_model/go"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"go.uber.org/zap/zaptest/observer"

	"iop/packages/go/streamgate"
	iop "iop/proto/gen/iop"
)

var livenessTestTime = time.Date(2026, 8, 5, 12, 0, 0, 0, time.UTC)

// capturingObservationSink records every forwarded observation so tests can
// assert whether the wrapper suppressed or forwarded a row.
type capturingObservationSink struct {
	mu  sync.Mutex
	got []streamgate.FilterObservation
}

type failingObservationSink struct{}

func (failingObservationSink) Emit(context.Context, streamgate.FilterObservation) error {
	return errors.New("observation sink unavailable")
}

func (c *capturingObservationSink) Emit(_ context.Context, obs streamgate.FilterObservation) error {
	c.mu.Lock()
	c.got = append(c.got, obs)
	c.mu.Unlock()
	return nil
}

func (c *capturingObservationSink) kinds() []streamgate.ObservationKind {
	c.mu.Lock()
	defer c.mu.Unlock()
	out := make([]streamgate.ObservationKind, 0, len(c.got))
	for _, obs := range c.got {
		out = append(out, obs.Kind())
	}
	return out
}

func (c *capturingObservationSink) count() int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return len(c.got)
}

// livenessHarness wires a request-local wrapper to an isolated collector set and
// an in-memory logger so metric and log projections can be asserted directly.
type livenessHarness struct {
	sink       *openAILivenessObservationSink
	seq        *streamgate.ObservationSequencer
	collectors *livenessRecoveryCollectors
	reg        *prometheus.Registry
	logs       *observer.ObservedLogs
	spy        *capturingObservationSink
}

func newLivenessHarness(t *testing.T, suppressDefault bool, downstreamZap bool) *livenessHarness {
	t.Helper()
	reg := prometheus.NewRegistry()
	coll := newLivenessRecoveryCollectors(reg)
	core, logs := observer.New(zapcore.InfoLevel)
	logger := zap.New(core)

	var downstream streamgate.ObservationSink
	var spy *capturingObservationSink
	if downstreamZap {
		downstream = newZapFilterObservationSink(logger)
	} else {
		spy = &capturingObservationSink{}
		downstream = spy
	}
	sink := newOpenAILivenessObservationSink(downstream, logger, suppressDefault, coll)
	return &livenessHarness{
		sink:       sink,
		seq:        streamgate.NewObservationSequencer(sink, nil),
		collectors: coll,
		reg:        reg,
		logs:       logs,
		spy:        spy,
	}
}

func (h *livenessHarness) emit(t *testing.T, input streamgate.FilterObservationInput) {
	t.Helper()
	if _, err := h.seq.Emit(context.Background(), input); err != nil {
		t.Fatalf("emit observation kind=%s: %v", input.Kind, err)
	}
}

func (h *livenessHarness) eligibility(path, health, commit, elig string) float64 {
	return testutil.ToFloat64(h.collectors.eligibility.WithLabelValues(path, health, commit, elig))
}

func (h *livenessHarness) result(path, health, result string) float64 {
	return testutil.ToFloat64(h.collectors.results.WithLabelValues(path, health, result))
}

// --- observation input builders (mirror the real emission sites) -------------

func livenessTarget(t *testing.T, provider, model, path string) streamgate.ObservationAttemptTarget {
	t.Helper()
	tgt, err := streamgate.NewObservationAttemptTarget("matrix-model", model, provider, path)
	if err != nil {
		t.Fatalf("NewObservationAttemptTarget: %v", err)
	}
	return tgt
}

// livenessEvalInput builds a private-liveness filter_evaluated observation
// exactly as parallel_evaluation.go would: attributed to the liveness filter,
// with a decision policy and a sanitized evidence descriptor.
func livenessEvalInput(t *testing.T, descriptor string, decision streamgate.FilterDecisionKind, target streamgate.ObservationAttemptTarget, commit streamgate.CommitState, correlation string) streamgate.FilterObservationInput {
	t.Helper()
	attr, err := streamgate.NewObservationAttribution(openAIStallRecoveryConsumerID, openAIStallRecoveryFilterID, openAIStallRecoveryFilterRuleID)
	if err != nil {
		t.Fatalf("NewObservationAttribution: %v", err)
	}
	dp, err := streamgate.NewObservationDecisionPolicy(streamgate.FilterOutcomeKindEvaluated, decision, streamgate.FilterEnforcementBlocking)
	if err != nil {
		t.Fatalf("NewObservationDecisionPolicy: %v", err)
	}
	ev, err := streamgate.NewSanitizedEvidence(
		streamgate.EventKindProviderError, streamGateChannelDefault, openAIStallRecoveryFilterRuleID,
		descriptor, openAIOutputFilterFingerprint(openAIStallRecoveryFilterRuleID, descriptor), 1, 0,
		streamgate.FilterOutcomeKindEvaluated, livenessTestTime,
	)
	if err != nil {
		t.Fatalf("NewSanitizedEvidence: %v", err)
	}
	return streamgate.FilterObservationInput{
		Kind:              streamgate.ObservationKindFilterEvaluated,
		StableCorrelation: correlation,
		ConfigGeneration:  "gen",
		AttemptID:         "att-x",
		AttemptTarget:     target,
		EpochID:           1,
		CommitState:       commit,
		Attribution:       &attr,
		DecisionPolicy:    &dp,
		Evidence:          &ev,
		OccurredAt:        livenessTestTime,
	}
}

func exactReplayRecoveryInfo(t *testing.T) *streamgate.ObservationRecoveryInfo {
	t.Helper()
	ri, err := streamgate.NewObservationRecoveryInfo("plan-x", streamgate.RecoveryStrategyExactReplay, streamgate.RecoveryResumeModeReplaceAttempt, "")
	if err != nil {
		t.Fatalf("NewObservationRecoveryInfo: %v", err)
	}
	return &ri
}

func continuationRecoveryInfo(t *testing.T) *streamgate.ObservationRecoveryInfo {
	t.Helper()
	ri, err := streamgate.NewObservationRecoveryInfo("plan-other", streamgate.RecoveryStrategyContinuationRepair, streamgate.RecoveryResumeModeContinueStream, "att-x")
	if err != nil {
		t.Fatalf("NewObservationRecoveryInfo: %v", err)
	}
	return &ri
}

func livenessCause(t *testing.T, stage, code string) streamgate.FailureCauseChain {
	t.Helper()
	cause, err := streamgate.NewFailureCause(stage, code, "", "", "")
	if err != nil {
		t.Fatalf("NewFailureCause: %v", err)
	}
	chain, err := streamgate.NewFailureCauseChain([]streamgate.FailureCause{cause})
	if err != nil {
		t.Fatalf("NewFailureCauseChain: %v", err)
	}
	return chain
}

// livenessRecoveryInput builds a recovery-lifecycle observation for the armed
// ExactReplay cycle, mirroring recovery_coordinator.go's emissions.
func livenessRecoveryInput(t *testing.T, kind streamgate.ObservationKind, target streamgate.ObservationAttemptTarget, correlation string) streamgate.FilterObservationInput {
	t.Helper()
	in := streamgate.FilterObservationInput{
		Kind:              kind,
		StableCorrelation: correlation,
		ConfigGeneration:  "gen",
		AttemptID:         "att-x",
		AttemptTarget:     target,
		EpochID:           1,
		CommitState:       streamgate.CommitStateTransportUncommitted,
		OccurredAt:        livenessTestTime,
	}
	switch kind {
	case streamgate.ObservationKindRecoveryPlanRejected:
		in.Causes = livenessCause(t, "recovery", "plan_ineligible")
	case streamgate.ObservationKindRecoveryAttemptAbortFailed,
		streamgate.ObservationKindRecoveryRebuildFailed,
		streamgate.ObservationKindRecoveryDispatchFailed:
		in.Recovery = exactReplayRecoveryInfo(t)
		in.Causes = livenessCause(t, "recovery", "attempt_failed")
	default:
		in.Recovery = exactReplayRecoveryInfo(t)
	}
	return in
}

func livenessTerminalInput(t *testing.T, target streamgate.ObservationAttemptTarget, correlation string) streamgate.FilterObservationInput {
	t.Helper()
	return streamgate.FilterObservationInput{
		Kind:              streamgate.ObservationKindTerminalCommitted,
		StableCorrelation: correlation,
		ConfigGeneration:  "gen",
		AttemptID:         "att-x",
		AttemptTarget:     target,
		CommitState:       streamgate.CommitStateTerminalCommitted,
		TerminalReason:    streamgate.TerminalReasonCompleted,
		OccurredAt:        livenessTestTime,
	}
}

// unrelatedEvalInput builds a filter_evaluated observation for a non-liveness
// filter so the wrapper must ignore and forward it.
func unrelatedEvalInput(t *testing.T, target streamgate.ObservationAttemptTarget, correlation string) streamgate.FilterObservationInput {
	t.Helper()
	attr, err := streamgate.NewObservationAttribution("semantic.consumer", "semantic.filter", "semantic.rule")
	if err != nil {
		t.Fatalf("NewObservationAttribution: %v", err)
	}
	dp, err := streamgate.NewObservationDecisionPolicy(streamgate.FilterOutcomeKindEvaluated, streamgate.FilterDecisionKindPass, streamgate.FilterEnforcementObserveOnly)
	if err != nil {
		t.Fatalf("NewObservationDecisionPolicy: %v", err)
	}
	return streamgate.FilterObservationInput{
		Kind:              streamgate.ObservationKindFilterEvaluated,
		StableCorrelation: correlation,
		ConfigGeneration:  "gen",
		AttemptID:         "att-x",
		AttemptTarget:     target,
		EpochID:           1,
		CommitState:       streamgate.CommitStateTransportUncommitted,
		Attribution:       &attr,
		DecisionPolicy:    &dp,
		OccurredAt:        livenessTestTime,
	}
}

const (
	livTestPathNormalized = "normalized"
	livTestPathTunnel     = "provider_tunnel"
)

// TestOpenAILivenessObservationSink covers request-local sequencing,
// deduplication, default-sink suppression, custom/Noop forwarding, closed
// classification, and repeated construction for the liveness projection.
func TestOpenAILivenessObservationSink(t *testing.T) {
	target := func(t *testing.T) streamgate.ObservationAttemptTarget {
		return livenessTarget(t, "provider-a", "served-a", livTestPathNormalized)
	}

	t.Run("eligible_redispatched", func(t *testing.T) {
		h := newLivenessHarness(t, true, false)
		tg := target(t)
		h.emit(t, livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1"))
		h.emit(t, livenessRecoveryInput(t, streamgate.ObservationKindRecoveryPlanSelected, tg, "req-1"))
		h.emit(t, livenessRecoveryInput(t, streamgate.ObservationKindRecoveryDispatched, tg, "req-1"))
		if got := h.eligibility(livTestPathNormalized, livenessHealthUnknown, livenessCommitUncommitted, livenessEligibilityEligible); got != 1 {
			t.Fatalf("eligibility=%v want 1", got)
		}
		if got := h.result(livTestPathNormalized, livenessHealthUnknown, livenessResultRedispatched); got != 1 {
			t.Fatalf("redispatched=%v want 1", got)
		}
	})

	failureCases := []struct {
		name string
		kind streamgate.ObservationKind
		want string
	}{
		{"plan_rejected", streamgate.ObservationKindRecoveryPlanRejected, livenessResultPlanRejected},
		{"abort_failed", streamgate.ObservationKindRecoveryAttemptAbortFailed, livenessResultAbortFailed},
		{"rebuild_failed", streamgate.ObservationKindRecoveryRebuildFailed, livenessResultRebuildFailed},
		{"dispatch_failed", streamgate.ObservationKindRecoveryDispatchFailed, livenessResultDispatchFailed},
	}
	for _, tc := range failureCases {
		t.Run("eligible_"+tc.name, func(t *testing.T) {
			h := newLivenessHarness(t, true, false)
			tg := target(t)
			h.emit(t, livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1"))
			if tc.kind != streamgate.ObservationKindRecoveryPlanRejected {
				h.emit(t, livenessRecoveryInput(t, streamgate.ObservationKindRecoveryPlanSelected, tg, "req-1"))
			}
			h.emit(t, livenessRecoveryInput(t, tc.kind, tg, "req-1"))
			if got := h.result(livTestPathNormalized, livenessHealthUnknown, tc.want); got != 1 {
				t.Fatalf("%s=%v want 1", tc.want, got)
			}
		})
	}

	t.Run("terminal_after_recovery", func(t *testing.T) {
		h := newLivenessHarness(t, true, false)
		tg := target(t)
		h.emit(t, livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1"))
		h.emit(t, livenessRecoveryInput(t, streamgate.ObservationKindRecoveryPlanSelected, tg, "req-1"))
		h.emit(t, livenessTerminalInput(t, tg, "req-1"))
		if got := h.result(livTestPathNormalized, livenessHealthUnknown, livenessResultTerminal); got != 1 {
			t.Fatalf("terminal=%v want 1", got)
		}
	})

	t.Run("not_selected", func(t *testing.T) {
		h := newLivenessHarness(t, true, false)
		tg := target(t)
		h.emit(t, livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1"))
		h.emit(t, livenessTerminalInput(t, tg, "req-1"))
		if got := h.result(livTestPathNormalized, livenessHealthUnknown, livenessResultNotSelected); got != 1 {
			t.Fatalf("not_selected=%v want 1", got)
		}
	})

	t.Run("non_liveness_plan_is_forwarded_and_not_selected", func(t *testing.T) {
		h := newLivenessHarness(t, true, false)
		tg := target(t)
		h.emit(t, livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1"))
		other := livenessRecoveryInput(t, streamgate.ObservationKindRecoveryPlanSelected, tg, "req-1")
		other.Recovery = continuationRecoveryInfo(t)
		other.CommitState = streamgate.CommitStateStreamOpen
		h.emit(t, other)
		if got := h.result(livTestPathNormalized, livenessHealthUnknown, livenessResultNotSelected); got != 1 {
			t.Fatalf("not_selected=%v want 1", got)
		}
		if h.spy.count() != 1 {
			t.Fatalf("non-liveness recovery was suppressed: forwarded=%d want 1", h.spy.count())
		}
	})

	ineligibleCases := []struct {
		name       string
		descriptor string
		decision   streamgate.FilterDecisionKind
		wantElig   string
	}{
		{"ineligible", livenessDescriptorIneligible, streamgate.FilterDecisionKindPass, livenessEligibilityOther},
		{"unconfirmed", livenessDescriptorUnconfirmed, streamgate.FilterDecisionKindPass, livenessEligibilityUnconfirmedFence},
	}
	for _, tc := range ineligibleCases {
		t.Run(tc.name+"_immediate_terminal", func(t *testing.T) {
			h := newLivenessHarness(t, true, false)
			tg := target(t)
			commit := streamgate.CommitStateTransportUncommitted
			if tc.name == "ineligible" {
				commit = streamgate.CommitStateStreamOpen
			}
			h.emit(t, livenessEvalInput(t, tc.descriptor, tc.decision, tg, commit, "req-1"))
			wantCommit := livenessCommitUncommitted
			if tc.name == "ineligible" {
				wantCommit = livenessCommitStreamOpen
			}
			if got := h.eligibility(livTestPathNormalized, livenessHealthUnknown, wantCommit, tc.wantElig); got != 1 {
				t.Fatalf("eligibility=%v want 1", got)
			}
			if got := h.result(livTestPathNormalized, livenessHealthUnknown, livenessResultTerminal); got != 1 {
				t.Fatalf("terminal=%v want 1", got)
			}
		})
	}

	t.Run("provider_error_ignored_no_cycle", func(t *testing.T) {
		h := newLivenessHarness(t, true, false)
		tg := target(t)
		h.emit(t, livenessEvalInput(t, livenessDescriptorProviderIgnored, streamgate.FilterDecisionKindPass, tg, streamgate.CommitStateTransportUncommitted, "req-1"))
		if got := testutil.CollectAndCount(h.collectors.eligibility); got != 0 {
			t.Fatalf("eligibility series=%d want 0", got)
		}
		if got := testutil.CollectAndCount(h.collectors.results); got != 0 {
			t.Fatalf("results series=%d want 0", got)
		}
		// Still consumed (kept off the generic writer) and safe-logged.
		if entries := h.logs.FilterMessage(livenessLogMessage).All(); len(entries) != 1 {
			t.Fatalf("safe log entries=%d want 1", len(entries))
		}
	})

	t.Run("dedup_second_eligibility", func(t *testing.T) {
		h := newLivenessHarness(t, true, false)
		tg := target(t)
		h.emit(t, livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1"))
		h.emit(t, livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1"))
		h.emit(t, livenessRecoveryInput(t, streamgate.ObservationKindRecoveryDispatched, tg, "req-1"))
		if got := h.eligibility(livTestPathNormalized, livenessHealthUnknown, livenessCommitUncommitted, livenessEligibilityEligible); got != 1 {
			t.Fatalf("eligibility=%v want 1 (deduplicated)", got)
		}
		if got := h.result(livTestPathNormalized, livenessHealthUnknown, livenessResultRedispatched); got != 1 {
			t.Fatalf("redispatched=%v want 1", got)
		}
	})

	t.Run("new_cycle_after_redispatch", func(t *testing.T) {
		h := newLivenessHarness(t, true, false)
		tg := target(t)
		h.emit(t, livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1"))
		h.emit(t, livenessRecoveryInput(t, streamgate.ObservationKindRecoveryDispatched, tg, "req-1"))
		// A later stall opens a fresh bounded cycle.
		h.emit(t, livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1"))
		h.emit(t, livenessRecoveryInput(t, streamgate.ObservationKindRecoveryPlanRejected, tg, "req-1"))
		if got := h.eligibility(livTestPathNormalized, livenessHealthUnknown, livenessCommitUncommitted, livenessEligibilityEligible); got != 2 {
			t.Fatalf("eligibility=%v want 2", got)
		}
		if got := h.result(livTestPathNormalized, livenessHealthUnknown, livenessResultRedispatched); got != 1 {
			t.Fatalf("redispatched=%v want 1", got)
		}
		if got := h.result(livTestPathNormalized, livenessHealthUnknown, livenessResultPlanRejected); got != 1 {
			t.Fatalf("plan_rejected=%v want 1", got)
		}
	})

	t.Run("unrelated_observations_forwarded", func(t *testing.T) {
		h := newLivenessHarness(t, true, false)
		tg := target(t)
		h.emit(t, unrelatedEvalInput(t, tg, "req-1"))
		if got := testutil.CollectAndCount(h.collectors.eligibility); got != 0 {
			t.Fatalf("eligibility series=%d want 0", got)
		}
		if h.spy.count() != 1 {
			t.Fatalf("downstream forwarded=%d want 1", h.spy.count())
		}
	})

	t.Run("default_suppression_and_forwarding", func(t *testing.T) {
		h := newLivenessHarness(t, true, false)
		tg := target(t)
		// Private-liveness rows are suppressed from downstream.
		h.emit(t, livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1"))
		h.emit(t, livenessRecoveryInput(t, streamgate.ObservationKindRecoveryPlanSelected, tg, "req-1"))
		h.emit(t, livenessRecoveryInput(t, streamgate.ObservationKindRecoveryDispatched, tg, "req-1"))
		// An unrelated row is forwarded.
		h.emit(t, unrelatedEvalInput(t, tg, "req-1"))
		if h.spy.count() != 1 {
			t.Fatalf("forwarded=%d want 1 (only unrelated)", h.spy.count())
		}
		if got := h.spy.kinds()[0]; got != streamgate.ObservationKindFilterEvaluated {
			t.Fatalf("forwarded kind=%s", got)
		}
		// Safe log written for each suppressed row (eval + plan_selected + dispatched).
		if entries := h.logs.FilterMessage(livenessLogMessage).All(); len(entries) != 3 {
			t.Fatalf("safe log entries=%d want 3", len(entries))
		}
	})

	t.Run("custom_sink_receives_originals", func(t *testing.T) {
		h := newLivenessHarness(t, false, false)
		tg := target(t)
		h.emit(t, livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1"))
		h.emit(t, livenessRecoveryInput(t, streamgate.ObservationKindRecoveryDispatched, tg, "req-1"))
		// Custom (non-default) downstream receives every original observation.
		if h.spy.count() != 2 {
			t.Fatalf("forwarded=%d want 2", h.spy.count())
		}
		// The safe projection is not disabled.
		if got := h.result(livTestPathNormalized, livenessHealthUnknown, livenessResultRedispatched); got != 1 {
			t.Fatalf("redispatched=%v want 1", got)
		}
		// No safe replacement log when not owning the default sink.
		if entries := h.logs.FilterMessage(livenessLogMessage).All(); len(entries) != 0 {
			t.Fatalf("safe log entries=%d want 0", len(entries))
		}
	})

	t.Run("noop_sink_projects_without_forwarding_side_effects", func(t *testing.T) {
		reg := prometheus.NewRegistry()
		coll := newLivenessRecoveryCollectors(reg)
		sink := newOpenAILivenessObservationSink(streamgate.NoopObservationSink{}, zap.NewNop(), false, coll)
		seq := streamgate.NewObservationSequencer(sink, nil)
		tg := target(t)
		if _, err := seq.Emit(context.Background(), livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1")); err != nil {
			t.Fatalf("emit: %v", err)
		}
		if _, err := seq.Emit(context.Background(), livenessRecoveryInput(t, streamgate.ObservationKindRecoveryDispatched, tg, "req-1")); err != nil {
			t.Fatalf("emit: %v", err)
		}
		if got := testutil.ToFloat64(coll.results.WithLabelValues(livTestPathNormalized, livenessHealthUnknown, livenessResultRedispatched)); got != 1 {
			t.Fatalf("redispatched=%v want 1 (projection not disabled by Noop)", got)
		}
	})

	t.Run("downstream_failure_is_observation_only", func(t *testing.T) {
		reg := prometheus.NewRegistry()
		coll := newLivenessRecoveryCollectors(reg)
		sink := newOpenAILivenessObservationSink(failingObservationSink{}, zap.NewNop(), false, coll)
		seq := streamgate.NewObservationSequencer(sink, nil)
		tg := target(t)
		if _, err := seq.Emit(context.Background(), unrelatedEvalInput(t, tg, "req-1")); err != nil {
			t.Fatalf("downstream failure escaped the observer: %v", err)
		}
	})

	t.Run("predecessor_descriptor_contract", func(t *testing.T) {
		ctx := context.Background()
		// Confirmed, eligible.
		confirmed := func(t *testing.T, commit streamgate.CommitState, sideEffect bool) streamgate.FilterDecision {
			t.Helper()
			state := &openAIStallRecoveryState{}
			filter, err := newOpenAIStallRecoveryFilter("openai.ingress.1", state)
			if err != nil {
				t.Fatalf("new filter: %v", err)
			}
			event, err := newOpenAIProviderErrorEventFromFailure(confirmedStallFailure("available"), streamGateErrorRunFailed)
			if err != nil {
				t.Fatalf("map failure: %v", err)
			}
			dec, err := filter.Evaluate(ctx, stallFilterContext(t, commit, sideEffect), stallBatch(t, event, commit))
			if err != nil {
				t.Fatalf("evaluate: %v", err)
			}
			return dec
		}

		eligibleDec := confirmed(t, streamgate.CommitStateTransportUncommitted, false)
		if code := eligibleDec.Evidence().DescriptorCode(); code != livenessDescriptorConfirmed {
			t.Fatalf("confirmed descriptor=%q want %q", code, livenessDescriptorConfirmed)
		}
		if elig, cycle := classifyLivenessEligibility(eligibleDec.Evidence().DescriptorCode()); !cycle || elig != livenessEligibilityEligible {
			t.Fatalf("confirmed classify=%q,%v", elig, cycle)
		}

		ineligibleDec := confirmed(t, streamgate.CommitStateStreamOpen, false)
		if code := ineligibleDec.Evidence().DescriptorCode(); code != livenessDescriptorIneligible {
			t.Fatalf("ineligible descriptor=%q want %q", code, livenessDescriptorIneligible)
		}

		// Non-stall provider error is ignored (no cycle).
		state := &openAIStallRecoveryState{}
		filter, _ := newOpenAIStallRecoveryFilter("openai.ingress.1", state)
		generic, err := newOpenAIProviderErrorEventFromFailure(&iop.ExecutionFailure{Code: "other", Message: "raw"}, streamGateErrorRunFailed)
		if err != nil {
			t.Fatalf("map generic: %v", err)
		}
		ignoredDec, err := filter.Evaluate(ctx, stallFilterContext(t, streamgate.CommitStateTransportUncommitted, false), stallBatch(t, generic, streamgate.CommitStateTransportUncommitted))
		if err != nil {
			t.Fatalf("evaluate generic: %v", err)
		}
		if code := ignoredDec.Evidence().DescriptorCode(); code != livenessDescriptorProviderIgnored {
			t.Fatalf("ignored descriptor=%q want %q", code, livenessDescriptorProviderIgnored)
		}
		if _, cycle := classifyLivenessEligibility(ignoredDec.Evidence().DescriptorCode()); cycle {
			t.Fatalf("provider_error_ignored opened a cycle")
		}

		// Unconfirmed stall descriptor (stall code without confirmed handoff).
		desc, _ := streamgate.NewExternalDescriptor("provider_error", openAIStallFailureCode, openAIStallFailureCode, "")
		unconfirmedEvent, err := streamgate.NewProviderErrorEvent(streamGateChannelDefault, desc, livenessCause(t, openAIStallHealthStage, "unknown"), livenessTestTime)
		if err != nil {
			t.Fatalf("build unconfirmed event: %v", err)
		}
		unconfirmedFilter, _ := newOpenAIStallRecoveryFilter("openai.ingress.1", &openAIStallRecoveryState{})
		unconfirmedDec, err := unconfirmedFilter.Evaluate(ctx, stallFilterContext(t, streamgate.CommitStateTransportUncommitted, false), stallBatch(t, unconfirmedEvent, streamgate.CommitStateTransportUncommitted))
		if err != nil {
			t.Fatalf("evaluate unconfirmed: %v", err)
		}
		if code := unconfirmedDec.Evidence().DescriptorCode(); code != livenessDescriptorUnconfirmed {
			t.Fatalf("unconfirmed descriptor=%q want %q", code, livenessDescriptorUnconfirmed)
		}
	})

	t.Run("repeated_construction_shares_collectors", func(t *testing.T) {
		conf := stallMatrixServer(nil, false, 1)
		conf2 := stallMatrixServer(nil, false, 1)
		if conf.livenessCollectors != defaultLivenessRecoveryCollectors || conf2.livenessCollectors != defaultLivenessRecoveryCollectors {
			t.Fatal("servers do not share the process-global collector set")
		}
		if !conf.obsSinkIsDefault {
			t.Fatal("NewServer did not mark its own sink as default")
		}
		// observationSink() returns a fresh wrapper without registering collectors.
		s1, ok := conf.observationSink().(*openAILivenessObservationSink)
		if !ok {
			t.Fatal("observationSink() did not return the liveness wrapper")
		}
		if !s1.suppressDefault {
			t.Fatal("default server wrapper must suppress the generic writer")
		}
		conf.SetObservationSink(&capturingObservationSink{})
		s2 := conf.observationSink().(*openAILivenessObservationSink)
		if s2.suppressDefault {
			t.Fatal("explicitly installed sink must not be suppressed")
		}
	})

	t.Run("explicit_same_type_zap_sink_preserves_originals", func(t *testing.T) {
		tg := target(t)

		// Explicit same-concrete-type zap sink installed via SetObservationSink
		srvExplicit := stallMatrixServer(nil, false, 1)
		regExplicit := prometheus.NewRegistry()
		collExplicit := newLivenessRecoveryCollectors(regExplicit)
		coreExplicit, logsExplicit := observer.New(zapcore.InfoLevel)
		loggerExplicit := zap.New(coreExplicit)
		srvExplicit.logger = loggerExplicit
		srvExplicit.livenessCollectors = collExplicit

		explicitSink := newZapFilterObservationSink(loggerExplicit)
		srvExplicit.SetObservationSink(explicitSink)
		sinkExplicit := srvExplicit.observationSink()
		seqExplicit := streamgate.NewObservationSequencer(sinkExplicit, nil)

		if _, err := seqExplicit.Emit(context.Background(), livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1")); err != nil {
			t.Fatalf("emit eval explicit: %v", err)
		}
		if _, err := seqExplicit.Emit(context.Background(), livenessRecoveryInput(t, streamgate.ObservationKindRecoveryDispatched, tg, "req-1")); err != nil {
			t.Fatalf("emit dispatched explicit: %v", err)
		}

		if got := testutil.ToFloat64(collExplicit.eligibility.WithLabelValues(livTestPathNormalized, livenessHealthUnknown, livenessCommitUncommitted, livenessEligibilityEligible)); got != 1 {
			t.Fatalf("explicit eligibility=%v want 1", got)
		}
		if got := testutil.ToFloat64(collExplicit.results.WithLabelValues(livTestPathNormalized, livenessHealthUnknown, livenessResultRedispatched)); got != 1 {
			t.Fatalf("explicit redispatched=%v want 1", got)
		}

		if len(logsExplicit.FilterMessage(livenessLogMessage).All()) != 0 {
			t.Fatalf("explicit path produced safe replacement log, want 0")
		}

		genericLogs := logsExplicit.FilterMessage(filterObservationLogMessage).All()
		if len(genericLogs) != 2 {
			t.Fatalf("explicit path generic logs count=%d want 2", len(genericLogs))
		}
		fp := openAIOutputFilterFingerprint(openAIStallRecoveryFilterRuleID, livenessDescriptorConfirmed)
		fpBytes := fp.Bytes()
		assertExactObservationFields(t, genericLogs[0], map[string]any{
			"sequence":                 uint64(1),
			"observation_kind":         string(streamgate.ObservationKindFilterEvaluated),
			"correlation_id":           "req-1",
			"config_generation":        "gen",
			"attempt_id":               "att-x",
			"model_group":              "matrix-model",
			"actual_model":             "served-a",
			"actual_provider":          "provider-a",
			"execution_path":           "normalized",
			"epoch_id":                 uint64(1),
			"commit_state":             string(streamgate.CommitStateTransportUncommitted),
			"consumer_id":              openAIStallRecoveryConsumerID,
			"filter_id":                openAIStallRecoveryFilterID,
			"rule_id":                  openAIStallRecoveryFilterRuleID,
			"filter_outcome":           string(streamgate.FilterOutcomeKindEvaluated),
			"decision_kind":            string(streamgate.FilterDecisionKindViolation),
			"enforcement":              string(streamgate.FilterEnforcementBlocking),
			"failure_disposition":      string(streamgate.EvaluationFailureDispositionBlockingFatal),
			"evidence_event_kind":      string(streamgate.EventKindProviderError),
			"evidence_channel":         streamGateChannelDefault,
			"evidence_filter_rule":     openAIStallRecoveryFilterRuleID,
			"evidence_outcome":         string(streamgate.FilterOutcomeKindEvaluated),
			"evidence_descriptor_code": livenessDescriptorConfirmed,
			"evidence_fingerprint":     hex.EncodeToString(fpBytes[:]),
			"evidence_count":           1,
			"evidence_offset":          0,
			"evidence_timestamp":       livenessTestTime,
		})

		// Contrast with constructor-default server path
		srvDef := stallMatrixServer(nil, false, 1)
		regDef := prometheus.NewRegistry()
		collDef := newLivenessRecoveryCollectors(regDef)
		coreDef, logsDef := observer.New(zapcore.InfoLevel)
		loggerDef := zap.New(coreDef)
		srvDef.logger = loggerDef
		srvDef.livenessCollectors = collDef

		sinkDef := srvDef.observationSink()
		seqDef := streamgate.NewObservationSequencer(sinkDef, nil)

		if _, err := seqDef.Emit(context.Background(), livenessEvalInput(t, livenessDescriptorConfirmed, streamgate.FilterDecisionKindViolation, tg, streamgate.CommitStateTransportUncommitted, "req-1")); err != nil {
			t.Fatalf("emit eval default: %v", err)
		}
		if _, err := seqDef.Emit(context.Background(), livenessRecoveryInput(t, streamgate.ObservationKindRecoveryDispatched, tg, "req-1")); err != nil {
			t.Fatalf("emit dispatched default: %v", err)
		}

		if got := testutil.ToFloat64(collDef.eligibility.WithLabelValues(livTestPathNormalized, livenessHealthUnknown, livenessCommitUncommitted, livenessEligibilityEligible)); got != 1 {
			t.Fatalf("default eligibility=%v want 1", got)
		}
		if got := testutil.ToFloat64(collDef.results.WithLabelValues(livTestPathNormalized, livenessHealthUnknown, livenessResultRedispatched)); got != 1 {
			t.Fatalf("default redispatched=%v want 1", got)
		}

		if len(logsDef.FilterMessage(filterObservationLogMessage).All()) != 0 {
			t.Fatalf("default path produced generic log, want 0")
		}

		safeLogsDef := logsDef.FilterMessage(livenessLogMessage).All()
		if len(safeLogsDef) != 2 {
			t.Fatalf("default path safe logs count=%d want 2", len(safeLogsDef))
		}
		expectedFields := []string{"phase", "execution_path", "provider_health", "commit_state", "eligibility", "recovery_result"}
		for _, entry := range safeLogsDef {
			if len(entry.Context) != len(expectedFields) {
				t.Fatalf("safe log context len=%d want %d", len(entry.Context), len(expectedFields))
			}
			ctxMap := entry.ContextMap()
			for _, key := range expectedFields {
				if _, ok := ctxMap[key]; !ok {
					t.Fatalf("safe log missing field key=%q", key)
				}
			}
		}
	})

	t.Run("closed_classifiers", func(t *testing.T) {
		if got := classifyLivenessPath("normalized"); got != livenessPathNormalized {
			t.Errorf("path normalized=%q", got)
		}
		if got := classifyLivenessPath("provider_tunnel"); got != livenessPathProviderTunnel {
			t.Errorf("path tunnel=%q", got)
		}
		if got := classifyLivenessPath("weird"); got != livenessPathUnknown {
			t.Errorf("path unknown=%q", got)
		}
		for raw, want := range map[string]string{
			"available":   livenessHealthAvailable,
			"unavailable": livenessHealthUnavailable,
			"unknown":     livenessHealthUnknown,
			"":            livenessHealthUnknown,
			"garbage":     livenessHealthUnknown,
		} {
			if got := classifyLivenessHealth(raw); got != want {
				t.Errorf("health %q=%q want %q", raw, got, want)
			}
		}
		if got := classifyLivenessCommit(streamgate.CommitStateStreamOpen); got != livenessCommitStreamOpen {
			t.Errorf("commit stream_open=%q", got)
		}
		if got := classifyLivenessCommit(streamgate.CommitState("odd")); got != livenessCommitUnknown {
			t.Errorf("commit unknown=%q", got)
		}
	})
}

// TestOpenAILivenessRecoveryObservability drives the production handlers so
// the request-local wrapper is proven across both OpenAI endpoints, both
// execution paths, and all outcome variants (redispatched, plan_rejected,
// terminal, dispatch_failed).
func TestOpenAILivenessRecoveryObservability(t *testing.T) {
	surfaces := []struct {
		endpoint string
		path     string
		wantPath string
	}{
		{openAIRebuildEndpointChat, livTestPathNormalized, livenessPathNormalized},
		{openAIRebuildEndpointChat, livTestPathTunnel, livenessPathProviderTunnel},
		{openAIRebuildEndpointResponses, livTestPathNormalized, livenessPathNormalized},
		{openAIRebuildEndpointResponses, livTestPathTunnel, livenessPathProviderTunnel},
	}

	outcomes := []struct {
		name         string
		wantElig     string
		wantHealth   string
		wantCommit   string
		wantResult   string
		wantCode     int
		wantSubmits  int
		setupService func(endpoint, path string) *scriptedPoolRunService
		setupBudget  int
	}{
		{
			name:        "redispatched",
			wantElig:    livenessEligibilityEligible,
			wantHealth:  livenessHealthUnknown,
			wantCommit:  livenessCommitUncommitted,
			wantResult:  livenessResultRedispatched,
			wantCode:    200,
			wantSubmits: 2,
			setupService: func(endpoint, path string) *scriptedPoolRunService {
				return newScriptedPoolRunService(
					stallMatrixFailureAttempt(path, "attempt-sentinel", "provider-sentinel", "unavailable"),
					stallMatrixSuccessAttempt(endpoint, path, false, "replacement-sentinel", "provider-replacement", "recovered-sentinel"),
				)
			},
			setupBudget: 1,
		},
		{
			name:        "plan_rejected",
			wantElig:    livenessEligibilityEligible,
			wantHealth:  livenessHealthUnknown,
			wantCommit:  livenessCommitUncommitted,
			wantResult:  livenessResultPlanRejected,
			wantCode:    502,
			wantSubmits: 1,
			setupService: func(endpoint, path string) *scriptedPoolRunService {
				return newScriptedPoolRunService(
					stallMatrixFailureAttempt(path, "attempt-sentinel", "provider-sentinel", "unavailable"),
				)
			},
			setupBudget: 0,
		},
		{
			name:        "terminal",
			wantElig:    "",
			wantHealth:  livenessHealthUnknown,
			wantCommit:  livenessCommitUncommitted,
			wantResult:  "",
			wantCode:    502,
			wantSubmits: 1,
			setupService: func(endpoint, path string) *scriptedPoolRunService {
				f := confirmedStallFailure("unavailable")
				f.Metadata["recovery_handoff"] = "unconfirmed"
				attempt := scriptedPoolAttempt{
					path:     path,
					runID:    "attempt-sentinel",
					provider: "provider-sentinel",
					target:   "served-provider-sentinel",
				}
				if path == livTestPathNormalized {
					attempt.runEvents = bufferedRunEvents(&iop.RunEvent{Type: "error", Failure: f})
				} else {
					attempt.frames = bufferedTunnelFrames(&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR, Failure: f})
				}
				return newScriptedPoolRunService(attempt)
			},
			setupBudget: 1,
		},
		{
			name:        "dispatch_failed",
			wantElig:    livenessEligibilityEligible,
			wantHealth:  livenessHealthUnknown,
			wantCommit:  livenessCommitUncommitted,
			wantResult:  livenessResultDispatchFailed,
			wantCode:    502,
			wantSubmits: 2,
			setupService: func(endpoint, path string) *scriptedPoolRunService {
				return newScriptedPoolRunService(
					stallMatrixFailureAttempt(path, "attempt-sentinel", "provider-sentinel", "unavailable"),
					scriptedPoolAttempt{
						path:     path,
						runID:    "replacement-sentinel",
						provider: "provider-replacement",
						target:   "served-provider-replacement",
						err:      errors.New("dispatch error sentinel"),
					},
				)
			},
			setupBudget: 1,
		},
	}

	for _, tc := range surfaces {
		for _, outcome := range outcomes {
			t.Run(tc.endpoint+"/"+tc.path+"/"+outcome.name, func(t *testing.T) {
				reg := prometheus.NewRegistry()
				collectors := newLivenessRecoveryCollectors(reg)
				core, logs := observer.New(zapcore.InfoLevel)
				service := outcome.setupService(tc.endpoint, tc.path)
				srv := stallMatrixServer(service, true, outcome.setupBudget)
				srv.logger = zap.New(core)
				srv.livenessCollectors = collectors

				response := runStallMatrixHandler(t, srv, tc.endpoint, false, nil)
				if response.Code != outcome.wantCode {
					t.Fatalf("response code=%d want %d", response.Code, outcome.wantCode)
				}
				if outcome.name == "redispatched" {
					if !strings.Contains(response.Body.String(), "recovered-sentinel") {
						t.Fatalf("redispatched response body=%q want 200 with recovered-sentinel", response.Body.String())
					}
				} else {
					if strings.Contains(response.Body.String(), "recovered-sentinel") {
						t.Fatalf("non-redispatched response body unexpectedly contains recovered-sentinel: %q", response.Body.String())
					}
				}

				if service.poolSubmits() != outcome.wantSubmits {
					t.Fatalf("service submits=%d want %d", service.poolSubmits(), outcome.wantSubmits)
				}

				assertGatheredLivenessMetrics(t, reg, tc.wantPath, outcome.wantHealth, outcome.wantCommit, outcome.wantElig, outcome.wantResult)

				safeLogs := logs.FilterMessage(livenessLogMessage).All()
				if len(safeLogs) == 0 {
					t.Fatal("default liveness route produced no safe replacement log")
				}
				assertSafeLogSchemaAndValues(t, safeLogs, tc.wantPath, outcome.name)

				if len(logs.FilterMessage(filterObservationLogMessage).All()) != 0 {
					t.Fatal("constructor-default generic log received liveness observations")
				}
			})
		}
	}
}

const (
	wantLivenessEligibilityFamily = "iop_edge_liveness_recovery_eligibility_total"
	wantLivenessResultsFamily     = "iop_edge_liveness_recovery_results_total"
)

func assertGatheredLivenessMetrics(t *testing.T, reg *prometheus.Registry, wantPath, wantHealth, wantCommit, wantElig, wantResult string) {
	t.Helper()
	families, err := reg.Gather()
	if err != nil {
		t.Fatalf("reg.Gather error: %v", err)
	}

	if wantElig == "" && wantResult == "" {
		if len(families) != 0 {
			t.Fatalf("unexpected metric families count=%d for negative row (want 0): %v", len(families), families)
		}
		return
	}

	if len(families) != 2 {
		t.Fatalf("gathered metric families count=%d want 2: %v", len(families), families)
	}

	gotNames := []string{families[0].GetName(), families[1].GetName()}
	sort.Strings(gotNames)
	wantNames := []string{wantLivenessEligibilityFamily, wantLivenessResultsFamily}
	if !reflect.DeepEqual(gotNames, wantNames) {
		t.Fatalf("gathered metric family names=%v want %v", gotNames, wantNames)
	}

	var eligFam, resultFam *dto.MetricFamily
	for _, fam := range families {
		switch fam.GetName() {
		case wantLivenessEligibilityFamily:
			eligFam = fam
		case wantLivenessResultsFamily:
			resultFam = fam
		}
	}

	if eligFam == nil || len(eligFam.GetMetric()) != 1 {
		t.Fatalf("eligibility metric family missing or metric count != 1: %v", eligFam)
	}
	eligMetric := eligFam.GetMetric()[0]
	if got := eligMetric.GetCounter().GetValue(); got != 1.0 {
		t.Fatalf("eligibility counter=%v want 1.0", got)
	}
	wantEligLabels := map[string]string{
		"execution_path":  wantPath,
		"provider_health": wantHealth,
		"commit_state":    wantCommit,
		"eligibility":     wantElig,
	}
	checkMetricLabelSchema(t, eligMetric, []string{"commit_state", "eligibility", "execution_path", "provider_health"}, wantEligLabels)

	if resultFam == nil || len(resultFam.GetMetric()) != 1 {
		t.Fatalf("results metric family missing or metric count != 1: %v", resultFam)
	}
	resultMetric := resultFam.GetMetric()[0]
	if got := resultMetric.GetCounter().GetValue(); got != 1.0 {
		t.Fatalf("results counter=%v want 1.0", got)
	}
	wantResultLabels := map[string]string{
		"execution_path":  wantPath,
		"provider_health": wantHealth,
		"recovery_result": wantResult,
	}
	checkMetricLabelSchema(t, resultMetric, []string{"execution_path", "provider_health", "recovery_result"}, wantResultLabels)
}

func checkMetricLabelSchema(t *testing.T, metric *dto.Metric, wantKeys []string, wantMap map[string]string) {
	t.Helper()
	labels := metric.GetLabel()
	if len(labels) != len(wantKeys) {
		t.Fatalf("metric label count=%d want %d", len(labels), len(wantKeys))
	}
	for i, lp := range labels {
		if lp.GetName() != wantKeys[i] {
			t.Fatalf("metric label key[%d]=%q want %q", i, lp.GetName(), wantKeys[i])
		}
		val, ok := wantMap[lp.GetName()]
		if !ok {
			t.Fatalf("unexpected metric label key %q", lp.GetName())
		}
		if lp.GetValue() != val {
			t.Fatalf("metric label %q value=%q want %q", lp.GetName(), lp.GetValue(), val)
		}
	}
}

func expectedSafeLogSequence(wantPath, outcomeName string) []map[string]string {
	makeRow := func(phase, elig, result string) map[string]string {
		return map[string]string{
			"phase":           phase,
			"execution_path":  wantPath,
			"provider_health": livenessHealthUnknown,
			"commit_state":    livenessCommitUncommitted,
			"eligibility":     elig,
			"recovery_result": result,
		}
	}

	switch outcomeName {
	case "terminal":
		return []map[string]string{
			makeRow("idle", "", ""),
			makeRow("idle", "", ""),
		}
	case "plan_rejected":
		return []map[string]string{
			makeRow("idle", "", ""),
			makeRow("eligible_pending", livenessEligibilityEligible, ""),
			makeRow("idle", "", livenessResultPlanRejected),
		}
	case "redispatched":
		return []map[string]string{
			makeRow("idle", "", ""),
			makeRow("eligible_pending", livenessEligibilityEligible, ""),
			makeRow("eligible_pending", "", ""),
			makeRow("eligible_pending", "", ""),
			makeRow("eligible_pending", "", ""),
			makeRow("idle", "", livenessResultRedispatched),
		}
	case "dispatch_failed":
		return []map[string]string{
			makeRow("idle", "", ""),
			makeRow("eligible_pending", livenessEligibilityEligible, ""),
			makeRow("eligible_pending", "", ""),
			makeRow("eligible_pending", "", ""),
			makeRow("eligible_pending", "", ""),
			makeRow("idle", "", livenessResultDispatchFailed),
		}
	default:
		panic("unknown outcome name: " + outcomeName)
	}
}

func assertSafeLogSchemaAndValues(t *testing.T, safeLogs []observer.LoggedEntry, wantPath, outcomeName string) {
	t.Helper()
	expectedFields := []string{"phase", "execution_path", "provider_health", "commit_state", "eligibility", "recovery_result"}
	unsafeKeys := map[string]bool{
		"correlation_id": true, "attempt_id": true, "run_id": true, "session_id": true,
		"model": true, "provider": true, "node_id": true, "plan_id": true,
		"shared_attempt_id": true, "credential": true, "slot": true,
	}

	expectedSeq := expectedSafeLogSequence(wantPath, outcomeName)
	if len(safeLogs) != len(expectedSeq) {
		t.Fatalf("safe log count=%d want %d for outcome %q", len(safeLogs), len(expectedSeq), outcomeName)
	}

	for i, entry := range safeLogs {
		if len(entry.Context) != len(expectedFields) {
			t.Fatalf("safe log[%d] field count=%d want %d", i, len(entry.Context), len(expectedFields))
		}
		ctxMap := entry.ContextMap()
		for _, key := range expectedFields {
			if _, ok := ctxMap[key]; !ok {
				t.Fatalf("safe log[%d] missing key %q", i, key)
			}
		}
		for key, val := range ctxMap {
			if unsafeKeys[key] {
				t.Fatalf("safe log[%d] contained unsafe key %q", i, key)
			}
			strVal, ok := val.(string)
			if !ok {
				t.Fatalf("safe log[%d] key %q value is not string: %v", i, key, val)
			}
			if strings.Contains(strVal, "sentinel") {
				t.Fatalf("safe log[%d] key %q contains sentinel value: %q", i, key, strVal)
			}
		}

		wantMap := expectedSeq[i]
		for key, wantVal := range wantMap {
			gotVal, _ := ctxMap[key].(string)
			if gotVal != wantVal {
				t.Fatalf("safe log[%d] key %q = %q want %q", i, key, gotVal, wantVal)
			}
		}
	}
}
