package streamgate

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"reflect"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"
	"unicode/utf8"
)

func testOccurredAt() time.Time {
	return time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
}

func testObservationAttemptTarget(t *testing.T) ObservationAttemptTarget {
	t.Helper()
	target, err := NewObservationAttemptTarget("model-group", "actual/model", "provider-a", "normalized")
	if err != nil {
		t.Fatalf("NewObservationAttemptTarget: %v", err)
	}
	return target
}

func TestObservationAttemptTarget_BoundedImmutableContract(t *testing.T) {
	target := testObservationAttemptTarget(t)
	if target.ModelGroup() != "model-group" || target.Model() != "actual/model" ||
		target.Provider() != "provider-a" || target.ExecutionPath() != "normalized" {
		t.Fatalf("unexpected target: %+v", target)
	}

	// Existing valid runtime identifiers — including whitespace and Unicode,
	// which the config/runtime contract accepts as long as they are non-empty —
	// are preserved verbatim so an observation never narrows the request
	// identifier contract.
	verbatim := []string{
		strings.Repeat("m", maxObservationAttemptTargetValueBytes), // byte-bound boundary
		"model group with spaces",
		"프로바이더/실측-model",
		"gpt-4o@2026-01-01",
	}
	for _, value := range verbatim {
		got, err := NewObservationAttemptTarget(value, "model", "provider", "path")
		if err != nil {
			t.Fatalf("valid identifier %q rejected: %v", value, err)
		}
		if got.ModelGroup() != value {
			t.Fatalf("identifier %q normalized to %q, want verbatim", value, got.ModelGroup())
		}
	}

	// Empty is the only caller contract error.
	if _, err := NewObservationAttemptTarget("", "model", "provider", "path"); err == nil {
		t.Fatal("empty model group accepted")
	}

	// An over-bound or invalid-UTF-8 identifier is deterministically reduced to
	// a bounded, log-safe "sha256:<hex>" join key rather than rejected, so a
	// still-valid request is never failed by observation construction.
	unsafe := []string{
		strings.Repeat("m", maxObservationAttemptTargetValueBytes+1),
		"invalid-utf8-\xff\xfe-tail",
	}
	for _, value := range unsafe {
		got, err := NewObservationAttemptTarget(value, "model", "provider", "path")
		if err != nil {
			t.Fatalf("unsafe identifier %q rejected instead of normalized: %v", value, err)
		}
		sum := sha256.Sum256([]byte(value))
		want := "sha256:" + hex.EncodeToString(sum[:])
		if got.ModelGroup() != want {
			t.Fatalf("identifier %q normalized to %q, want %q", value, got.ModelGroup(), want)
		}
		if !utf8.ValidString(got.ModelGroup()) || len(got.ModelGroup()) > maxObservationAttemptTargetValueBytes {
			t.Fatalf("normalized identifier %q is not bounded/valid utf-8: %q", value, got.ModelGroup())
		}
		again, err := NewObservationAttemptTarget(value, "model", "provider", "path")
		if err != nil {
			t.Fatalf("re-normalize %q: %v", value, err)
		}
		if again.ModelGroup() != got.ModelGroup() {
			t.Fatalf("normalization not deterministic for %q: %q vs %q", value, got.ModelGroup(), again.ModelGroup())
		}
	}

	input := validFilterEvaluatedInput(t)
	obs, err := newFilterObservation(1, input)
	if err != nil {
		t.Fatalf("newFilterObservation: %v", err)
	}
	input.AttemptTarget = ObservationAttemptTarget{}
	if got := obs.AttemptTarget(); got != target {
		t.Fatalf("observation target changed with input reassignment: %+v", got)
	}
}

// TestRuntimeObservation_IdentifierCompatibilityDoesNotAffectRequestOutcome
// drives a full RequestRuntime.Run with a model group / model / provider /
// execution path that the config and AttemptBinding contracts accept (non-empty)
// but that the former ASCII allowlist would have rejected. It proves the request
// still commits a single successful terminal and that every emitted observation
// carries the identifiers verbatim as its bounded attempt-target correlation.
func TestRuntimeObservation_IdentifierCompatibilityDoesNotAffectRequestOutcome(t *testing.T) {
	sink := newFixtureSink()
	disp := &fixtureDispatcher{}
	rebuilder := &fixtureRebuilder{}

	passF := &customMockFilter{id: "pass-filter"}
	regPass, err := NewFilterRegistration(passF, "cap1", true, FilterEnforcementBlocking, 5*time.Second, 10)
	if err != nil {
		t.Fatalf("NewFilterRegistration: %v", err)
	}

	obsSink := &recordingSink{}
	snap := createTestRuntimeSnapshot(t, []FilterRegistration{regPass}, disp, rebuilder, sink).
		WithObservationSink(obsSink)

	rsEv, err := NewResponseStartEvent("default", 200, map[string]string{"content-type": "text/event-stream"}, time.Now())
	if err != nil {
		t.Fatalf("NewResponseStartEvent: %v", err)
	}
	txtEv, err := NewTextDeltaEvent("default", "hello world", time.Now())
	if err != nil {
		t.Fatalf("NewTextDeltaEvent: %v", err)
	}
	termEv, err := NewTerminalEvent("default", time.Now())
	if err != nil {
		t.Fatalf("NewTerminalEvent: %v", err)
	}

	src := newSliceEventSource([]NormalizedEvent{rsEv, txtEv, termEv})
	ctrl := &fixtureController{}

	// Identifiers valid per the config/AttemptBinding contract (non-empty) but
	// outside the former ASCII allowlist: spaces and Unicode.
	const (
		modelGroup    = "model group with spaces"
		actualModel   = "actual model/v1 실측"
		actualProv    = "provider a"
		executionPath = "primary path"
	)
	initialBinding, err := NewAttemptBinding("att-1", actualModel, actualProv, executionPath, src, ctrl)
	if err != nil {
		t.Fatalf("NewAttemptBinding: %v", err)
	}

	rt, err := NewRequestRuntime(snap, modelGroup, initialBinding)
	if err != nil {
		t.Fatalf("NewRequestRuntime: %v", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	if err := rt.Run(ctx); err != nil {
		t.Fatalf("Run: %v", err)
	}

	// Request outcome is unchanged: exactly one successful terminal committed.
	sink.mu.Lock()
	terminals := append([]TerminalResult(nil), sink.terminals...)
	sink.mu.Unlock()
	if len(terminals) != 1 {
		t.Fatalf("expected 1 terminal result, got %d", len(terminals))
	}
	if !terminals[0].Success() {
		t.Fatalf("expected success terminal result, got %+v", terminals[0])
	}

	// The still-valid identifiers stay verbatim in every observation target.
	wantTarget, err := NewObservationAttemptTarget(modelGroup, actualModel, actualProv, executionPath)
	if err != nil {
		t.Fatalf("NewObservationAttemptTarget: %v", err)
	}
	if wantTarget.ModelGroup() != modelGroup || wantTarget.Model() != actualModel ||
		wantTarget.Provider() != actualProv || wantTarget.ExecutionPath() != executionPath {
		t.Fatalf("target did not preserve identifiers verbatim: %+v", wantTarget)
	}

	obsSink.mu.Lock()
	got := append([]FilterObservation(nil), obsSink.got...)
	obsSink.mu.Unlock()
	if len(got) == 0 {
		t.Fatal("expected at least one observation emitted during the run")
	}
	for i, obs := range got {
		if obs.AttemptTarget() != wantTarget {
			t.Fatalf("observation %d (kind=%s) target = %+v, want %+v", i, obs.Kind(), obs.AttemptTarget(), wantTarget)
		}
	}
}

func validFilterEvaluatedInput(t *testing.T) FilterObservationInput {
	t.Helper()
	attribution, err := NewObservationAttribution("consumer-a", "filter-a", "rule-a")
	if err != nil {
		t.Fatalf("NewObservationAttribution: %v", err)
	}
	policy, err := NewObservationDecisionPolicy(
		FilterOutcomeKindEvaluated,
		FilterDecisionKindPass,
		FilterEnforcementBlocking,
	)
	if err != nil {
		t.Fatalf("NewObservationDecisionPolicy: %v", err)
	}
	return FilterObservationInput{
		Kind:              ObservationKindFilterEvaluated,
		StableCorrelation: "req-1",
		ConfigGeneration:  "gen-1",
		AttemptID:         "attempt-1",
		AttemptTarget:     testObservationAttemptTarget(t),
		EpochID:           1,
		CommitState:       CommitStateTransportUncommitted,
		Attribution:       &attribution,
		DecisionPolicy:    &policy,
		OccurredAt:        testOccurredAt(),
	}
}

// --- Monotonic sequence -----------------------------------------------------

// TestObservationSequencer_MonotonicSequence verifies that sequential Emit
// calls assign strictly increasing sequence numbers starting at 1, with no
// gaps or repeats.
func TestObservationSequencer_MonotonicSequence(t *testing.T) {
	seq := NewObservationSequencer(NoopObservationSink{}, nil)
	var last uint64
	for i := 0; i < 5; i++ {
		obs, err := seq.Emit(context.Background(), validFilterEvaluatedInput(t))
		if err != nil {
			t.Fatalf("Emit %d: %v", i, err)
		}
		if obs.Sequence() != last+1 {
			t.Fatalf("Emit %d: sequence = %d, want %d", i, obs.Sequence(), last+1)
		}
		last = obs.Sequence()
	}
}

// TestObservationSequencer_ConcurrentEmitIsRaceFreeAndUnique launches
// concurrent Emit calls and verifies every assigned sequence number is
// unique and falls within the expected contiguous range. Run with -race to
// prove the shared counter has no data race.
func TestObservationSequencer_ConcurrentEmitIsRaceFreeAndUnique(t *testing.T) {
	const goroutines = 20
	const perGoroutine = 25
	const total = goroutines * perGoroutine

	seq := NewObservationSequencer(NoopObservationSink{}, nil)

	var mu sync.Mutex
	seen := make(map[uint64]bool, total)

	var wg sync.WaitGroup
	wg.Add(goroutines)
	for g := 0; g < goroutines; g++ {
		go func() {
			defer wg.Done()
			for i := 0; i < perGoroutine; i++ {
				obs, err := seq.Emit(context.Background(), validFilterEvaluatedInput(t))
				if err != nil {
					t.Errorf("Emit: %v", err)
					return
				}
				mu.Lock()
				seen[obs.Sequence()] = true
				mu.Unlock()
			}
		}()
	}
	wg.Wait()

	if len(seen) != total {
		t.Fatalf("unique sequence count = %d, want %d", len(seen), total)
	}
	for i := uint64(1); i <= uint64(total); i++ {
		if !seen[i] {
			t.Fatalf("sequence %d was never assigned", i)
		}
	}
}

// TestFilterObservation_SequenceMustBePositive verifies that a zero sequence
// is rejected at construction, since ObservationSequencer starts counting at
// 1 and callers must never forge sequence assignment.
func TestFilterObservation_SequenceMustBePositive(t *testing.T) {
	if _, err := newFilterObservation(0, validFilterEvaluatedInput(t)); err == nil {
		t.Fatal("expected error for zero sequence, got nil")
	}
}

// --- Sink failure isolation -------------------------------------------------

type recordingSink struct {
	mu   sync.Mutex
	fail error
	got  []FilterObservation
}

func (s *recordingSink) Emit(ctx context.Context, obs FilterObservation) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.got = append(s.got, obs)
	return s.fail
}

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

// TestObservationSequencer_SinkFailureIsolation verifies that a failing sink
// never surfaces its error through Emit's return value and never blocks
// sequence progress: sink failure is a best-effort observability boundary
// that must not change runtime correctness.
func TestObservationSequencer_SinkFailureIsolation(t *testing.T) {
	sinkErr := errors.New("observability backend unavailable")
	sink := &recordingSink{fail: sinkErr}

	var hookMu sync.Mutex
	var hookCalls []error
	onError := func(obs FilterObservation, err error) {
		hookMu.Lock()
		defer hookMu.Unlock()
		hookCalls = append(hookCalls, err)
	}

	seq := NewObservationSequencer(sink, onError)

	obs1, err := seq.Emit(context.Background(), validFilterEvaluatedInput(t))
	if err != nil {
		t.Fatalf("Emit must isolate sink failure, got error: %v", err)
	}
	obs2, err := seq.Emit(context.Background(), validFilterEvaluatedInput(t))
	if err != nil {
		t.Fatalf("Emit must isolate sink failure, got error: %v", err)
	}

	if obs1.Sequence() != 1 || obs2.Sequence() != 2 {
		t.Fatalf("sequence progress must continue despite sink failure, got %d, %d", obs1.Sequence(), obs2.Sequence())
	}
	if sink.count() != 2 {
		t.Fatalf("sink should still receive every observation, got %d deliveries", sink.count())
	}

	hookMu.Lock()
	defer hookMu.Unlock()
	if len(hookCalls) != 2 {
		t.Fatalf("onError hook call count = %d, want 2", len(hookCalls))
	}
	for _, e := range hookCalls {
		if !errors.Is(e, sinkErr) {
			t.Fatalf("onError hook error = %v, want %v", e, sinkErr)
		}
	}
}

// TestObservationSequencer_NilSinkDefaultsToNoop verifies a nil sink does not
// panic and Emit still succeeds and assigns sequence numbers.
func TestObservationSequencer_NilSinkDefaultsToNoop(t *testing.T) {
	seq := NewObservationSequencer(nil, nil)
	obs, err := seq.Emit(context.Background(), validFilterEvaluatedInput(t))
	if err != nil {
		t.Fatalf("Emit: %v", err)
	}
	if obs.Sequence() != 1 {
		t.Fatalf("Sequence() = %d, want 1", obs.Sequence())
	}
}

// TestObservationSequencer_ConstructionErrorIsNotSwallowed verifies that a
// genuine caller bug (invalid input) still surfaces as an error, so failure
// isolation only applies to the sink boundary, not to input validation.
func TestObservationSequencer_ConstructionErrorIsNotSwallowed(t *testing.T) {
	sink := &recordingSink{}
	seq := NewObservationSequencer(sink, nil)

	input := validFilterEvaluatedInput(t)
	input.StableCorrelation = ""

	if _, err := seq.Emit(context.Background(), input); err == nil {
		t.Fatal("expected validation error for empty stable correlation, got nil")
	}
	if sink.count() != 0 {
		t.Fatalf("sink must not receive an observation that failed construction, got %d deliveries", sink.count())
	}
}

// --- Bounded cause chain -----------------------------------------------------

func causeAt(t *testing.T, code string) FailureCause {
	t.Helper()
	c, err := NewFailureCause("stage-1", code, "consumer-a", "filter-a", "rule-a")
	if err != nil {
		t.Fatalf("NewFailureCause: %v", err)
	}
	return c
}

// TestFilterObservation_CauseChainAtMaxIsAccepted verifies exactly
// MaxFailureCauses (4) causes is a valid, immutable terminal_committed
// observation.
func TestFilterObservation_CauseChainAtMaxIsAccepted(t *testing.T) {
	chain, err := NewFailureCauseChain([]FailureCause{
		causeAt(t, "code-1"), causeAt(t, "code-2"), causeAt(t, "code-3"), causeAt(t, "code-4"),
	})
	if err != nil {
		t.Fatalf("NewFailureCauseChain: %v", err)
	}
	if chain.Len() != MaxFailureCauses {
		t.Fatalf("chain.Len() = %d, want %d", chain.Len(), MaxFailureCauses)
	}

	input := FilterObservationInput{
		Kind:              ObservationKindTerminalCommitted,
		StableCorrelation: "req-1",
		ConfigGeneration:  "gen-1",
		AttemptID:         "attempt-1",
		AttemptTarget:     testObservationAttemptTarget(t),
		CommitState:       CommitStateTerminalCommitted,
		Causes:            chain,
		OccurredAt:        testOccurredAt(),
	}
	obs, err := newFilterObservation(1, input)
	if err != nil {
		t.Fatalf("newFilterObservation: %v", err)
	}
	if obs.Causes().Len() != MaxFailureCauses {
		t.Fatalf("obs.Causes().Len() = %d, want %d", obs.Causes().Len(), MaxFailureCauses)
	}
}

// TestFilterObservation_CauseChainOverMaxIsRejected proves Validate() itself
// enforces the MaxFailureCauses bound as an invariant of the immutable
// contract, independent of FailureCauseChain's own capping behavior in its
// friendly constructors. It builds an oversized chain via package-internal
// struct construction to bypass NewFailureCauseChain's auto-cap.
func TestFilterObservation_CauseChainOverMaxIsRejected(t *testing.T) {
	oversized := FailureCauseChain{causes: []FailureCause{
		causeAt(t, "code-1"), causeAt(t, "code-2"), causeAt(t, "code-3"),
		causeAt(t, "code-4"), causeAt(t, "code-5"),
	}}
	if oversized.Len() != 5 {
		t.Fatalf("test setup: oversized.Len() = %d, want 5", oversized.Len())
	}

	obs := FilterObservation{
		sequence:          1,
		kind:              ObservationKindTerminalCommitted,
		stableCorrelation: StableToken{value: "req-1"},
		configGeneration:  StableToken{value: "gen-1"},
		attemptID:         StableToken{value: "attempt-1"},
		attemptTarget:     testObservationAttemptTarget(t),
		commitState:       CommitStateTerminalCommitted,
		causes:            oversized,
		occurredAt:        testOccurredAt(),
	}

	if err := obs.Validate(); err == nil {
		t.Fatal("expected error for cause chain exceeding maximum, got nil")
	}
}

// --- Sanitized-evidence-only / raw-free structural contract -----------------

// forbiddenFieldSubstrings lists substrings that must never appear in a
// FilterObservation (or nested value type) field name. Their presence would
// indicate a raw prompt/output/tool/auth/preparer payload leaked into the
// envelope.
var forbiddenFieldSubstrings = []string{
	"raw", "prompt", "output", "auth", "body", "toolarg", "result", "stacktrace",
}

func assertNoForbiddenFields(t *testing.T, typ reflect.Type) {
	t.Helper()
	if typ.Kind() == reflect.Ptr {
		typ = typ.Elem()
	}
	if typ.Kind() != reflect.Struct {
		return
	}
	for i := 0; i < typ.NumField(); i++ {
		field := typ.Field(i)
		lower := strings.ToLower(field.Name)
		for _, forbidden := range forbiddenFieldSubstrings {
			if strings.Contains(lower, forbidden) {
				t.Errorf("%s.%s contains forbidden substring %q; envelope must stay raw-free", typ.Name(), field.Name, forbidden)
			}
		}
	}
}

// TestFilterObservationStructIsRawFree walks FilterObservation and every
// nested value type's field names and fails if any resembles a raw content
// carrier. This is a structural regression guard for the reviewer checkpoint
// that the envelope never exposes raw prompt/output/tool/auth/preparer
// payload fields.
func TestFilterObservationStructIsRawFree(t *testing.T) {
	types := []reflect.Type{
		reflect.TypeOf(FilterObservation{}),
		reflect.TypeOf(FilterObservationInput{}),
		reflect.TypeOf(ObservationAttemptTarget{}),
		reflect.TypeOf(ObservationAttribution{}),
		reflect.TypeOf(ObservationHoldInfo{}),
		reflect.TypeOf(ObservationDecisionPolicy{}),
		reflect.TypeOf(ObservationRecoveryInfo{}),
		reflect.TypeOf(ObservationPreparerInfo{}),
	}
	for _, typ := range types {
		assertNoForbiddenFields(t, typ)
	}
}

// TestFilterObservation_EvidenceIsConsumerProvidedSanitizedOnly verifies that
// the only per-event content an observation may carry is the existing
// SanitizedEvidence type (fingerprint/count/offset/descriptor code), and that
// Evidence() returns a defensive copy.
func TestFilterObservation_EvidenceIsConsumerProvidedSanitizedOnly(t *testing.T) {
	evidence, err := NewSanitizedEvidence(
		EventKindTextDelta, "ch", "rule-a", "desc-code",
		FixedFingerprint{1}, 3, 10,
		FilterOutcomeKindEvaluated, testOccurredAt(),
	)
	if err != nil {
		t.Fatalf("NewSanitizedEvidence: %v", err)
	}

	input := validFilterEvaluatedInput(t)
	input.Evidence = &evidence

	obs, err := newFilterObservation(1, input)
	if err != nil {
		t.Fatalf("newFilterObservation: %v", err)
	}

	got := obs.Evidence()
	if got == nil {
		t.Fatal("Evidence() = nil, want non-nil")
	}
	if got.Fingerprint() != evidence.Fingerprint() || got.Count() != evidence.Count() || got.Offset() != evidence.Offset() {
		t.Fatalf("Evidence() = %+v, want a copy matching input evidence", got)
	}

	// Mutate the returned pointer's backing value; must not affect a second
	// call's result, proving Evidence() returns a defensive copy.
	*got = SanitizedEvidence{}
	again := obs.Evidence()
	if again.Fingerprint() != evidence.Fingerprint() {
		t.Fatal("Evidence() must return a defensive copy; mutation of a prior result leaked into stored state")
	}
}

// --- Kind-specific required fields (closed, immutable contract) -------------

func TestFilterObservation_KindSpecificRequiredFields(t *testing.T) {
	attribution, err := NewObservationAttribution("consumer-a", "filter-a", "rule-a")
	if err != nil {
		t.Fatalf("NewObservationAttribution: %v", err)
	}
	policy, err := NewObservationDecisionPolicy(
		FilterOutcomeKindEvaluated, FilterDecisionKindPass,
		FilterEnforcementBlocking,
	)
	if err != nil {
		t.Fatalf("NewObservationDecisionPolicy: %v", err)
	}
	hold, err := NewObservationHoldInfo(FilterHoldModeRolling, EventKindTextDelta, 500, 120)
	if err != nil {
		t.Fatalf("NewObservationHoldInfo: %v", err)
	}
	recoveryInfo, err := NewObservationRecoveryInfo("plan-1", RecoveryStrategyExactReplay, RecoveryResumeModeReplaceAttempt, "")
	if err != nil {
		t.Fatalf("NewObservationRecoveryInfo: %v", err)
	}
	continuationInfo, err := NewObservationRecoveryInfo("plan-2", RecoveryStrategyContinuationRepair, RecoveryResumeModeContinueStream, "attempt-0")
	if err != nil {
		t.Fatalf("NewObservationRecoveryInfo: %v", err)
	}
	preparerInfo, err := NewObservationPreparerInfo("preparer-1", RecoveryPreparationPrepared, ObservationDeadlineOutcomeCompleted)
	if err != nil {
		t.Fatalf("NewObservationPreparerInfo: %v", err)
	}
	causes, err := NewFailureCauseChain([]FailureCause{causeAt(t, "code-1")})
	if err != nil {
		t.Fatalf("NewFailureCauseChain: %v", err)
	}

	base := func() FilterObservationInput {
		return FilterObservationInput{
			StableCorrelation: "req-1",
			ConfigGeneration:  "gen-1",
			AttemptID:         "attempt-1",
			AttemptTarget:     testObservationAttemptTarget(t),
			EpochID:           1,
			CommitState:       CommitStateTransportUncommitted,
			OccurredAt:        testOccurredAt(),
		}
	}

	tests := []struct {
		name    string
		mutate  func(in *FilterObservationInput)
		wantErr bool
	}{
		{
			name: "filter_evaluated requires attribution and decision policy",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindFilterEvaluated
			},
			wantErr: true,
		},
		{
			name: "filter_evaluated with attribution and decision policy is valid",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindFilterEvaluated
				in.Attribution = &attribution
				in.DecisionPolicy = &policy
			},
			wantErr: false,
		},
		{
			name: "response_staged requires hold info",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindResponseStaged
			},
			wantErr: true,
		},
		{
			name: "response_staged with hold info is valid",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindResponseStaged
				in.Hold = &hold
			},
			wantErr: false,
		},
		{
			name: "terminal_committed requires reason or causes",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindTerminalCommitted
				in.CommitState = CommitStateTerminalCommitted
			},
			wantErr: true,
		},
		{
			name: "terminal_committed rejects both reason and causes",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindTerminalCommitted
				in.CommitState = CommitStateTerminalCommitted
				in.TerminalReason = TerminalReasonCompleted
				in.Causes = causes
			},
			wantErr: true,
		},
		{
			name: "terminal_committed with only reason is valid",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindTerminalCommitted
				in.CommitState = CommitStateTerminalCommitted
				in.TerminalReason = TerminalReasonCompleted
			},
			wantErr: false,
		},
		{
			name: "terminal_committed with only causes is valid",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindTerminalCommitted
				in.CommitState = CommitStateTerminalCommitted
				in.Causes = causes
			},
			wantErr: false,
		},
		{
			name: "recovery_plan_selected requires recovery info",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindRecoveryPlanSelected
			},
			wantErr: true,
		},
		{
			name: "recovery_plan_selected with recovery info is valid",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindRecoveryPlanSelected
				in.Recovery = &recoveryInfo
			},
			wantErr: false,
		},
		{
			name: "recovery_prepared requires recovery and preparer info",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindRecoveryPrepared
				in.Recovery = &recoveryInfo
			},
			wantErr: true,
		},
		{
			name: "recovery_prepared with recovery and preparer info is valid",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindRecoveryPrepared
				in.Recovery = &continuationInfo
				in.Preparer = &preparerInfo
				in.CommitState = CommitStateStreamOpen
			},
			wantErr: false,
		},
		{
			name: "arbitration_decided requires arbitration summary",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindArbitrationDecided
			},
			wantErr: true,
		},
		{
			name: "arbitration_decided with arbitration summary is valid",
			mutate: func(in *FilterObservationInput) {
				in.Kind = ObservationKindArbitrationDecided
				arb, _ := NewObservationArbitrationInfo(ArbitrationActionRelease, BaseDispositionReleaseCandidate)
				in.Arbitration = &arb
			},
			wantErr: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			in := base()
			tt.mutate(&in)
			_, err := newFilterObservation(1, in)
			if tt.wantErr && err == nil {
				t.Fatal("expected error, got nil")
			}
			if !tt.wantErr && err != nil {
				t.Fatalf("unexpected error: %v", err)
			}
		})
	}
}

// TestObservationRecoveryInfo_SharedAttemptCoupledToResumeMode verifies that
// a shared attempt id is required for continue_stream and forbidden for
// replace_attempt, matching the RecoveryPlan resume-mode contract.
func TestObservationRecoveryInfo_SharedAttemptCoupledToResumeMode(t *testing.T) {
	if _, err := NewObservationRecoveryInfo("plan-1", RecoveryStrategyContinuationRepair, RecoveryResumeModeContinueStream, ""); err == nil {
		t.Fatal("expected error: continue_stream without shared attempt id")
	}
	if _, err := NewObservationRecoveryInfo("plan-1", RecoveryStrategyExactReplay, RecoveryResumeModeReplaceAttempt, "attempt-0"); err == nil {
		t.Fatal("expected error: replace_attempt with a shared attempt id")
	}
	if _, err := NewObservationRecoveryInfo("plan-1", RecoveryStrategyContinuationRepair, RecoveryResumeModeContinueStream, "attempt-0"); err != nil {
		t.Fatalf("expected valid continue_stream with shared attempt id, got: %v", err)
	}
}

// TestObservationRecoveryInfo_StrategyResumeModeMatrix verifies that the
// ObservationRecoveryInfo strategy→resume-mode matrix matches the canonical
// RecoveryPlan: exact/schema recovery requires replace_attempt, while
// continuation/managed recovery requires continue_stream. Opposite combinations
// must be rejected.
func TestObservationRecoveryInfo_StrategyResumeModeMatrix(t *testing.T) {
	// --- Exact replay: canonical resume mode (replace_attempt) must succeed ---
	if _, err := NewObservationRecoveryInfo("plan-1", RecoveryStrategyExactReplay, RecoveryResumeModeReplaceAttempt, ""); err != nil {
		t.Fatalf("exact_replay + replace_attempt: unexpected error: %v", err)
	}
	// --- Schema repair: canonical resume mode (replace_attempt) must succeed ---
	if _, err := NewObservationRecoveryInfo("plan-2", RecoveryStrategySchemaRepair, RecoveryResumeModeReplaceAttempt, ""); err != nil {
		t.Fatalf("schema_repair + replace_attempt: unexpected error: %v", err)
	}
	// --- Continuation repair: canonical resume mode (continue_stream) must succeed ---
	if _, err := NewObservationRecoveryInfo("plan-3", RecoveryStrategyContinuationRepair, RecoveryResumeModeContinueStream, "attempt-0"); err != nil {
		t.Fatalf("continuation_repair + continue_stream: unexpected error: %v", err)
	}
	// --- Managed continuation: canonical resume mode (continue_stream) must succeed ---
	if _, err := NewObservationRecoveryInfo("plan-4", RecoveryStrategyManagedContinuation, RecoveryResumeModeContinueStream, "attempt-1"); err != nil {
		t.Fatalf("managed_continuation + continue_stream: unexpected error: %v", err)
	}

	// --- Exact replay + continue_stream (wrong resume mode) must fail ---
	if _, err := NewObservationRecoveryInfo("plan-5", RecoveryStrategyExactReplay, RecoveryResumeModeContinueStream, "attempt-x"); err == nil {
		t.Fatal("expected error: exact_replay + continue_stream")
	}
	// --- Schema repair + continue_stream (wrong resume mode) must fail ---
	if _, err := NewObservationRecoveryInfo("plan-6", RecoveryStrategySchemaRepair, RecoveryResumeModeContinueStream, "attempt-y"); err == nil {
		t.Fatal("expected error: schema_repair + continue_stream")
	}
	// --- Continuation repair + replace_attempt (wrong resume mode) must fail ---
	if _, err := NewObservationRecoveryInfo("plan-7", RecoveryStrategyContinuationRepair, RecoveryResumeModeReplaceAttempt, ""); err == nil {
		t.Fatal("expected error: continuation_repair + replace_attempt")
	}
	// --- Managed continuation + replace_attempt (wrong resume mode) must fail ---
	if _, err := NewObservationRecoveryInfo("plan-8", RecoveryStrategyManagedContinuation, RecoveryResumeModeReplaceAttempt, ""); err == nil {
		t.Fatal("expected error: managed_continuation + replace_attempt")
	}
}

// TestObservationDecisionPolicy_DecisionKindCoupledToOutcome verifies that a
// decision kind is required only when the outcome is evaluated.
func TestObservationDecisionPolicy_DecisionKindCoupledToOutcome(t *testing.T) {
	if _, err := NewObservationDecisionPolicy(
		FilterOutcomeKindEvaluated, "", FilterEnforcementBlocking,
	); err == nil {
		t.Fatal("expected error: evaluated outcome without a decision kind")
	}
	if _, err := NewObservationDecisionPolicy(
		FilterOutcomeKindNotApplicableForEpoch, FilterDecisionKindPass, FilterEnforcementBlocking,
	); err == nil {
		t.Fatal("expected error: non-evaluated outcome carrying a decision kind")
	}
	if _, err := NewObservationDecisionPolicy(
		FilterOutcomeKindNotApplicableForEpoch, "", FilterEnforcementBlocking,
	); err != nil {
		t.Fatalf("expected valid not-applicable policy without a decision kind, got: %v", err)
	}
}

// --- Defensive copies on FilterObservation accessors -------------------------

// TestFilterObservation_AccessorsReturnDefensiveCopies verifies that mutating
// a pointer returned by an accessor does not alter subsequent reads.
func TestFilterObservation_AccessorsReturnDefensiveCopies(t *testing.T) {
	obs, err := newFilterObservation(1, validFilterEvaluatedInput(t))
	if err != nil {
		t.Fatalf("newFilterObservation: %v", err)
	}

	attribution := obs.Attribution()
	if attribution == nil {
		t.Fatal("Attribution() = nil, want non-nil")
	}
	*attribution = ObservationAttribution{}
	again := obs.Attribution()
	if again.ConsumerID() != "consumer-a" {
		t.Fatal("Attribution() must return a defensive copy")
	}

	policy := obs.DecisionPolicy()
	if policy == nil {
		t.Fatal("DecisionPolicy() = nil, want non-nil")
	}
	*policy = ObservationDecisionPolicy{}
	againPolicy := obs.DecisionPolicy()
	if againPolicy.Outcome() != FilterOutcomeKindEvaluated {
		t.Fatal("DecisionPolicy() must return a defensive copy")
	}
}

// TestFilterObservation_EpochIDZeroMeansUnbound verifies that omitting the
// epoch id leaves the observation not epoch-bound, mirroring FilterContext's
// HasEpoch/EpochID contract. EpochID zero is a valid stored value; it is only
// rejected by kind-specific validation for kinds that require a non-zero epoch.
func TestFilterObservation_EpochIDZeroMeansUnbound(t *testing.T) {
	// response_staged does not require a non-zero epoch; it only requires
	// hold info and transport_uncommitted commit state.
	hold, _ := NewObservationHoldInfo(FilterHoldModeRolling, EventKindTextDelta, 500, 120)
	input := FilterObservationInput{
		Kind:              ObservationKindResponseStaged,
		StableCorrelation: "req-1",
		ConfigGeneration:  "gen-1",
		AttemptID:         "attempt-1",
		AttemptTarget:     testObservationAttemptTarget(t),
		EpochID:           0,
		CommitState:       CommitStateTransportUncommitted,
		Hold:              &hold,
		OccurredAt:        testOccurredAt(),
	}
	obs, err := newFilterObservation(1, input)
	if err != nil {
		t.Fatalf("newFilterObservation: %v", err)
	}
	if obs.EpochID() != 0 {
		t.Fatalf("EpochID() = %d, want 0", obs.EpochID())
	}
}

// TestFilterObservation_UnknownKindIsRejected verifies the closed
// ObservationKind set rejects unregistered values.
func TestFilterObservation_UnknownKindIsRejected(t *testing.T) {
	input := validFilterEvaluatedInput(t)
	input.Kind = ObservationKind("not_a_real_kind")
	input.Attribution = nil
	input.DecisionPolicy = nil
	if _, err := newFilterObservation(1, input); err == nil {
		t.Fatal("expected error for unknown observation kind, got nil")
	}
}

// --- CONTRACT-1: Arbitration info required and closed -----------------------

// TestFilterObservation_ArbitrationInfoRequiredAndClosed verifies that
// arbitration_decided requires a non-nil arbitration summary carrying both
// action and base disposition, that unknown actions/dispositions are rejected,
// and that the arbitration accessor returns a defensive copy.
func TestFilterObservation_ArbitrationInfoRequiredAndClosed(t *testing.T) {
	arb, err := NewObservationArbitrationInfo(ArbitrationActionRelease, BaseDispositionReleaseCandidate)
	if err != nil {
		t.Fatalf("NewObservationArbitrationInfo: %v", err)
	}

	// Valid arbitration info round-trips.
	if arb.Action() != ArbitrationActionRelease {
		t.Fatalf("Action() = %s, want release", arb.Action())
	}
	if arb.BaseDisposition() != BaseDispositionReleaseCandidate {
		t.Fatalf("BaseDisposition() = %s, want release_candidate", arb.BaseDisposition())
	}

	// Unknown action is rejected.
	if _, err := NewObservationArbitrationInfo(ArbitrationAction("bogus"), BaseDispositionReleaseCandidate); err == nil {
		t.Fatal("expected error for unknown arbitration action")
	}

	// Unknown base disposition is rejected.
	if _, err := NewObservationArbitrationInfo(ArbitrationActionRelease, BaseEventDisposition("bogus")); err == nil {
		t.Fatal("expected error for unknown base disposition")
	}

	// arbitration_decided without summary is rejected.
	input := FilterObservationInput{
		Kind:              ObservationKindArbitrationDecided,
		StableCorrelation: "req-1",
		ConfigGeneration:  "gen-1",
		AttemptID:         "attempt-1",
		AttemptTarget:     testObservationAttemptTarget(t),
		EpochID:           1,
		CommitState:       CommitStateTransportUncommitted,
		Arbitration:       nil,
		OccurredAt:        testOccurredAt(),
	}
	if _, err := newFilterObservation(1, input); err == nil {
		t.Fatal("expected error: arbitration_decided without summary")
	}

	// arbitration_decided with zero epoch is rejected.
	input.Arbitration = &arb
	input.EpochID = 0
	if _, err := newFilterObservation(1, input); err == nil {
		t.Fatal("expected error: arbitration_decided with zero epoch")
	}

	// Defensive copy: mutating the returned pointer must not affect stored state.
	input.EpochID = 1
	obs, err := newFilterObservation(1, input)
	if err != nil {
		t.Fatalf("newFilterObservation: %v", err)
	}
	got := obs.Arbitration()
	*got = ObservationArbitrationInfo{}
	again := obs.Arbitration()
	if again.Action() != ArbitrationActionRelease {
		t.Fatal("Arbitration() must return a defensive copy")
	}
}

// TestFilterObservation_FilterEvaluationStartedContract verifies that the new
// filter_evaluation_started kind requires attribution and a non-zero epoch,
// and that it is accepted in the closed kind set.
func TestFilterObservation_FilterEvaluationStartedContract(t *testing.T) {
	attribution, err := NewObservationAttribution("consumer-a", "filter-a", "rule-a")
	if err != nil {
		t.Fatalf("NewObservationAttribution: %v", err)
	}

	// Missing attribution is rejected.
	input := FilterObservationInput{
		Kind:              ObservationKindFilterEvaluationStarted,
		StableCorrelation: "req-1",
		ConfigGeneration:  "gen-1",
		AttemptID:         "attempt-1",
		AttemptTarget:     testObservationAttemptTarget(t),
		EpochID:           1,
		CommitState:       CommitStateTransportUncommitted,
		OccurredAt:        testOccurredAt(),
	}
	if _, err := newFilterObservation(1, input); err == nil {
		t.Fatal("expected error: filter_evaluation_started without attribution")
	}

	// Zero epoch is rejected.
	input.Attribution = &attribution
	input.EpochID = 0
	if _, err := newFilterObservation(1, input); err == nil {
		t.Fatal("expected error: filter_evaluation_started with zero epoch")
	}

	// Valid combination is accepted.
	input.EpochID = 5
	obs, err := newFilterObservation(1, input)
	if err != nil {
		t.Fatalf("newFilterObservation: %v", err)
	}
	if obs.Kind() != ObservationKindFilterEvaluationStarted {
		t.Fatalf("Kind() = %s, want filter_evaluation_started", obs.Kind())
	}

	// filter_evaluated also requires non-zero epoch.
	policy, err := NewObservationDecisionPolicy(
		FilterOutcomeKindEvaluated, FilterDecisionKindPass,
		FilterEnforcementBlocking,
	)
	if err != nil {
		t.Fatalf("NewObservationDecisionPolicy: %v", err)
	}
	evalInput := FilterObservationInput{
		Kind:              ObservationKindFilterEvaluated,
		StableCorrelation: "req-1",
		ConfigGeneration:  "gen-1",
		AttemptID:         "attempt-1",
		AttemptTarget:     testObservationAttemptTarget(t),
		EpochID:           0,
		CommitState:       CommitStateTransportUncommitted,
		Attribution:       &attribution,
		DecisionPolicy:    &policy,
		OccurredAt:        testOccurredAt(),
	}
	if _, err := newFilterObservation(1, evalInput); err == nil {
		t.Fatal("expected error: filter_evaluated with zero epoch")
	}
}

// TestFilterObservation_KindStateMatrix verifies the allowed commit state for
// each lifecycle kind: response_staged → transport_uncommitted, release_committed
// → stream_open, terminal_committed → terminal_committed, and the recovery
// resume-mode coupling.
func TestFilterObservation_KindStateMatrix(t *testing.T) {
	attribution, _ := NewObservationAttribution("consumer-a", "filter-a", "rule-a")
	policy, _ := NewObservationDecisionPolicy(
		FilterOutcomeKindEvaluated, FilterDecisionKindPass,
		FilterEnforcementBlocking,
	)
	hold, _ := NewObservationHoldInfo(FilterHoldModeRolling, EventKindTextDelta, 500, 120)
	recoveryReplace, _ := NewObservationRecoveryInfo("plan-1", RecoveryStrategyExactReplay, RecoveryResumeModeReplaceAttempt, "")
	recoverySchemaReplace, _ := NewObservationRecoveryInfo("plan-1b", RecoveryStrategySchemaRepair, RecoveryResumeModeReplaceAttempt, "")
	recoveryContinue, _ := NewObservationRecoveryInfo("plan-2", RecoveryStrategyContinuationRepair, RecoveryResumeModeContinueStream, "attempt-0")
	recoveryManagedContinue, _ := NewObservationRecoveryInfo("plan-2b", RecoveryStrategyManagedContinuation, RecoveryResumeModeContinueStream, "attempt-1")

	tests := []struct {
		name     string
		kind     ObservationKind
		commit   CommitState
		extra    func(*FilterObservationInput)
		wantErr  bool
		errSubst string
	}{
		{
			name:    "response_staged requires transport_uncommitted",
			kind:    ObservationKindResponseStaged,
			commit:  CommitStateStreamOpen,
			wantErr: true,
		},
		{
			name:    "response_staged valid with transport_uncommitted",
			kind:    ObservationKindResponseStaged,
			commit:  CommitStateTransportUncommitted,
			extra:   func(in *FilterObservationInput) { in.Hold = &hold },
			wantErr: false,
		},
		{
			name:    "release_committed requires stream_open",
			kind:    ObservationKindReleaseCommitted,
			commit:  CommitStateTransportUncommitted,
			wantErr: true,
		},
		{
			name:    "release_committed valid with stream_open",
			kind:    ObservationKindReleaseCommitted,
			commit:  CommitStateStreamOpen,
			wantErr: false,
		},
		{
			name:    "terminal_committed requires terminal_committed state",
			kind:    ObservationKindTerminalCommitted,
			commit:  CommitStateTransportUncommitted,
			wantErr: true,
		},
		{
			name:    "terminal_committed valid with terminal_committed state",
			kind:    ObservationKindTerminalCommitted,
			commit:  CommitStateTerminalCommitted,
			extra:   func(in *FilterObservationInput) { in.TerminalReason = TerminalReasonCompleted },
			wantErr: false,
		},
		{
			name:    "exact_replay recovery valid with transport_uncommitted",
			kind:    ObservationKindRecoveryPlanSelected,
			commit:  CommitStateTransportUncommitted,
			extra:   func(in *FilterObservationInput) { in.Recovery = &recoveryReplace },
			wantErr: false,
		},
		{
			name:    "exact_replay recovery requires transport_uncommitted",
			kind:    ObservationKindRecoveryPlanSelected,
			commit:  CommitStateStreamOpen,
			extra:   func(in *FilterObservationInput) { in.Recovery = &recoveryReplace },
			wantErr: true,
		},
		{
			name:    "schema_repair recovery valid with transport_uncommitted",
			kind:    ObservationKindRecoveryPlanSelected,
			commit:  CommitStateTransportUncommitted,
			extra:   func(in *FilterObservationInput) { in.Recovery = &recoverySchemaReplace },
			wantErr: false,
		},
		{
			name:    "schema_repair recovery requires transport_uncommitted",
			kind:    ObservationKindRecoveryPlanSelected,
			commit:  CommitStateStreamOpen,
			extra:   func(in *FilterObservationInput) { in.Recovery = &recoverySchemaReplace },
			wantErr: true,
		},
		{
			name:    "continuation_repair recovery valid with stream_open",
			kind:    ObservationKindRecoveryPlanSelected,
			commit:  CommitStateStreamOpen,
			extra:   func(in *FilterObservationInput) { in.Recovery = &recoveryContinue },
			wantErr: false,
		},
		{
			name:    "continuation_repair recovery requires stream_open",
			kind:    ObservationKindRecoveryPlanSelected,
			commit:  CommitStateTransportUncommitted,
			extra:   func(in *FilterObservationInput) { in.Recovery = &recoveryContinue },
			wantErr: true,
		},
		{
			name:    "managed_continuation recovery valid with stream_open",
			kind:    ObservationKindRecoveryPlanSelected,
			commit:  CommitStateStreamOpen,
			extra:   func(in *FilterObservationInput) { in.Recovery = &recoveryManagedContinue },
			wantErr: false,
		},
		{
			name:    "managed_continuation recovery requires stream_open",
			kind:    ObservationKindRecoveryPlanSelected,
			commit:  CommitStateTransportUncommitted,
			extra:   func(in *FilterObservationInput) { in.Recovery = &recoveryManagedContinue },
			wantErr: true,
		},
		{
			name:   "filter_evaluated valid with non-zero epoch and transport_uncommitted",
			kind:   ObservationKindFilterEvaluated,
			commit: CommitStateTransportUncommitted,
			extra: func(in *FilterObservationInput) {
				in.Attribution = &attribution
				in.DecisionPolicy = &policy
				in.EpochID = 1
			},
			wantErr: false,
		},
	}

	base := func() FilterObservationInput {
		return FilterObservationInput{
			StableCorrelation: "req-1",
			ConfigGeneration:  "gen-1",
			AttemptID:         "attempt-1",
			AttemptTarget:     testObservationAttemptTarget(t),
			EpochID:           1,
			OccurredAt:        testOccurredAt(),
		}
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			in := base()
			in.Kind = tt.kind
			in.CommitState = tt.commit
			if tt.extra != nil {
				tt.extra(&in)
			}
			_, err := newFilterObservation(1, in)
			if tt.wantErr && err == nil {
				t.Fatal("expected error, got nil")
			}
			if !tt.wantErr && err != nil {
				t.Fatalf("unexpected error: %v", err)
			}
		})
	}
}

// --- CONTRACT-2: Canonical failure disposition matrix -----------------------

// TestObservationDecisionPolicy_CanonicalDispositionMatrix verifies that the
// failure disposition is derived canonically from outcome × decisionKind ×
// enforcement and matches the epoch filter outcome truth table. Every
// combination is covered so a future truth-table change is caught here.
func TestObservationDecisionPolicy_CanonicalDispositionMatrix(t *testing.T) {
	tests := []struct {
		name         string
		outcome      FilterOutcomeKind
		decisionKind FilterDecisionKind
		enforcement  FilterEnforcement
		expectedDisp EvaluationFailureDisposition
	}{
		// evaluated paths
		{"evaluated pass blocking → none", FilterOutcomeKindEvaluated, FilterDecisionKindPass, FilterEnforcementBlocking, EvaluationFailureDispositionNone},
		{"evaluated pass observe_only → none", FilterOutcomeKindEvaluated, FilterDecisionKindPass, FilterEnforcementObserveOnly, EvaluationFailureDispositionNone},
		{"evaluated observe blocking → none", FilterOutcomeKindEvaluated, FilterDecisionKindObserve, FilterEnforcementBlocking, EvaluationFailureDispositionNone},
		{"evaluated observe observe_only → none", FilterOutcomeKindEvaluated, FilterDecisionKindObserve, FilterEnforcementObserveOnly, EvaluationFailureDispositionNone},
		{"evaluated replacement blocking → none", FilterOutcomeKindEvaluated, FilterDecisionKindReplacement, FilterEnforcementBlocking, EvaluationFailureDispositionNone},
		{"evaluated replacement observe_only → none", FilterOutcomeKindEvaluated, FilterDecisionKindReplacement, FilterEnforcementObserveOnly, EvaluationFailureDispositionNone},
		{"evaluated violation blocking → blocking_fatal", FilterOutcomeKindEvaluated, FilterDecisionKindViolation, FilterEnforcementBlocking, EvaluationFailureDispositionBlockingFatal},
		{"evaluated violation observe_only → observe_error", FilterOutcomeKindEvaluated, FilterDecisionKindViolation, FilterEnforcementObserveOnly, EvaluationFailureDispositionObserveError},
		{"evaluated fatal blocking → blocking_fatal", FilterOutcomeKindEvaluated, FilterDecisionKindFatal, FilterEnforcementBlocking, EvaluationFailureDispositionBlockingFatal},
		{"evaluated fatal observe_only → observe_error", FilterOutcomeKindEvaluated, FilterDecisionKindFatal, FilterEnforcementObserveOnly, EvaluationFailureDispositionObserveError},
		// evaluation_error paths
		{"eval_error blocking → blocking_fatal", FilterOutcomeKindEvaluationError, "", FilterEnforcementBlocking, EvaluationFailureDispositionBlockingFatal},
		{"eval_error observe_only → observe_error", FilterOutcomeKindEvaluationError, "", FilterEnforcementObserveOnly, EvaluationFailureDispositionObserveError},
		// not_applicable / deferred paths
		{"not_applicable → none", FilterOutcomeKindNotApplicableForEpoch, "", FilterEnforcementBlocking, EvaluationFailureDispositionNone},
		{"not_applicable → none observe", FilterOutcomeKindNotApplicableForEpoch, "", FilterEnforcementObserveOnly, EvaluationFailureDispositionNone},
		{"deferred → none", FilterOutcomeKindDeferredByRequirement, "", FilterEnforcementBlocking, EvaluationFailureDispositionNone},
		{"deferred → none observe", FilterOutcomeKindDeferredByRequirement, "", FilterEnforcementObserveOnly, EvaluationFailureDispositionNone},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			policy, err := NewObservationDecisionPolicy(tt.outcome, tt.decisionKind, tt.enforcement)
			if err != nil {
				t.Fatalf("NewObservationDecisionPolicy: %v", err)
			}
			got := policy.FailureDisposition()
			if got != tt.expectedDisp {
				t.Fatalf("FailureDisposition() = %s, want %s", got, tt.expectedDisp)
			}
		})
	}
}

// TestObservationDecisionPolicy_CanonicalDispositionConsistency verifies that
// the derived disposition is consistent across multiple construction calls
// with the same inputs, proving the constructor and accessor agree on the
// truth table. This guards against drift between deriveObservationDecisionPolicyDisposition
// and the constructor validation.
func TestObservationDecisionPolicy_CanonicalDispositionConsistency(t *testing.T) {
	// evaluated + violation + blocking → blocking_fatal
	p1, err := NewObservationDecisionPolicy(
		FilterOutcomeKindEvaluated, FilterDecisionKindViolation,
		FilterEnforcementBlocking,
	)
	if err != nil {
		t.Fatalf("NewObservationDecisionPolicy: %v", err)
	}
	if p1.FailureDisposition() != EvaluationFailureDispositionBlockingFatal {
		t.Fatalf("expected blocking_fatal, got %s", p1.FailureDisposition())
	}

	// evaluated + violation + observe_only → observe_error
	p2, err := NewObservationDecisionPolicy(
		FilterOutcomeKindEvaluated, FilterDecisionKindViolation,
		FilterEnforcementObserveOnly,
	)
	if err != nil {
		t.Fatalf("NewObservationDecisionPolicy: %v", err)
	}
	if p2.FailureDisposition() != EvaluationFailureDispositionObserveError {
		t.Fatalf("expected observe_error, got %s", p2.FailureDisposition())
	}

	// Same inputs produce the same disposition.
	for i := 0; i < 100; i++ {
		p, err := NewObservationDecisionPolicy(
			FilterOutcomeKindEvaluated, FilterDecisionKindViolation,
			FilterEnforcementBlocking,
		)
		if err != nil {
			t.Fatalf("iteration %d: %v", i, err)
		}
		if p.FailureDisposition() != p1.FailureDisposition() {
			t.Fatalf("iteration %d: disposition = %s, want %s", i, p.FailureDisposition(), p1.FailureDisposition())
		}
	}
}

// --- CONTRACT-3: Raw-free carrier graph and defensive copies ---------------

// carrierTypesForGraph returns every value type defined in filter_observation
// that forms the carrier graph. Standard library types like time.Time are
// excluded as leaf nodes.
func carrierTypesForGraph() []reflect.Type {
	return []reflect.Type{
		reflect.TypeOf(FilterObservation{}),
		reflect.TypeOf(FilterObservationInput{}),
		reflect.TypeOf(ObservationAttemptTarget{}),
		reflect.TypeOf(ObservationAttribution{}),
		reflect.TypeOf(ObservationHoldInfo{}),
		reflect.TypeOf(ObservationDecisionPolicy{}),
		reflect.TypeOf(ObservationArbitrationInfo{}),
		reflect.TypeOf(ObservationRecoveryInfo{}),
		reflect.TypeOf(ObservationPreparerInfo{}),
		reflect.TypeOf(SanitizedEvidence{}),
		reflect.TypeOf(FailureCauseChain{}),
		reflect.TypeOf(FailureCause{}),
	}
}

// TestFilterObservation_RawFreeCarrierGraph verifies that every type in the
// filter observation carrier graph is free of raw content fields. This is a
// structural regression guard.
func TestFilterObservation_RawFreeCarrierGraph(t *testing.T) {
	for _, typ := range carrierTypesForGraph() {
		assertNoForbiddenFields(t, typ)
	}
}

// TestFilterObservation_AllAccessorsDefensive verifies that every accessor on
// FilterObservation that returns a mutable pointer or value returns a
// defensive copy. Mutating the returned object must not affect subsequent
// reads.
func TestFilterObservation_AllAccessorsDefensive(t *testing.T) {
	attribution, _ := NewObservationAttribution("consumer-a", "filter-a", "rule-a")
	policy, _ := NewObservationDecisionPolicy(
		FilterOutcomeKindEvaluated, FilterDecisionKindPass,
		FilterEnforcementBlocking,
	)
	hold, _ := NewObservationHoldInfo(FilterHoldModeRolling, EventKindTextDelta, 500, 120)
	arb, _ := NewObservationArbitrationInfo(ArbitrationActionRelease, BaseDispositionReleaseCandidate)
	recoveryInfo, _ := NewObservationRecoveryInfo("plan-1", RecoveryStrategyExactReplay, RecoveryResumeModeReplaceAttempt, "")
	preparerInfo, _ := NewObservationPreparerInfo("preparer-1", RecoveryPreparationPrepared, ObservationDeadlineOutcomeCompleted)
	causes, _ := NewFailureCauseChain([]FailureCause{causeAt(t, "code-1")})
	evidence, _ := NewSanitizedEvidence(
		EventKindTextDelta, "ch", "rule-a", "desc-code",
		FixedFingerprint{1}, 3, 10,
		FilterOutcomeKindEvaluated, testOccurredAt(),
	)

	obs, err := newFilterObservation(1, FilterObservationInput{
		Kind:              ObservationKindFilterEvaluated,
		StableCorrelation: "req-1",
		ConfigGeneration:  "gen-1",
		AttemptID:         "attempt-1",
		AttemptTarget:     testObservationAttemptTarget(t),
		EpochID:           1,
		CommitState:       CommitStateTransportUncommitted,
		Attribution:       &attribution,
		Hold:              &hold,
		DecisionPolicy:    &policy,
		Arbitration:       &arb,
		Recovery:          &recoveryInfo,
		Preparer:          &preparerInfo,
		Causes:            causes,
		Evidence:          &evidence,
		TerminalReason:    TerminalReasonCompleted,
		OccurredAt:        testOccurredAt(),
	})
	if err != nil {
		t.Fatalf("newFilterObservation: %v", err)
	}

	// Attribution
	orig := obs.Attribution()
	*orig = ObservationAttribution{}
	if obs.Attribution().ConsumerID() != "consumer-a" {
		t.Fatal("Attribution() must return a defensive copy")
	}

	// Hold
	origHold := obs.Hold()
	*origHold = ObservationHoldInfo{}
	if obs.Hold().Mode() != FilterHoldModeRolling {
		t.Fatal("Hold() must return a defensive copy")
	}

	// DecisionPolicy
	origPol := obs.DecisionPolicy()
	*origPol = ObservationDecisionPolicy{}
	if obs.DecisionPolicy().Outcome() != FilterOutcomeKindEvaluated {
		t.Fatal("DecisionPolicy() must return a defensive copy")
	}

	// Arbitration
	origArb := obs.Arbitration()
	*origArb = ObservationArbitrationInfo{}
	if obs.Arbitration().Action() != ArbitrationActionRelease {
		t.Fatal("Arbitration() must return a defensive copy")
	}

	// Recovery
	origRec := obs.Recovery()
	*origRec = ObservationRecoveryInfo{}
	if obs.Recovery().PlanID() != "plan-1" {
		t.Fatal("Recovery() must return a defensive copy")
	}

	// Preparer
	origPrep := obs.Preparer()
	*origPrep = ObservationPreparerInfo{}
	if obs.Preparer().PreparerID() != "preparer-1" {
		t.Fatal("Preparer() must return a defensive copy")
	}

	// Causes
	origCauses := obs.Causes()
	if origCauses.Len() != causes.Len() {
		t.Fatalf("Causes().Len() = %d, want %d", origCauses.Len(), causes.Len())
	}
	// Append to the returned chain should not affect the observation.
	updated, _ := origCauses.Append(causeAt(t, "code-2"))
	if obs.Causes().Len() != causes.Len() {
		t.Fatal("Causes() must return a defensive copy; append must not affect stored state")
	}
	_ = updated

	// Evidence
	origEvi := obs.Evidence()
	*origEvi = SanitizedEvidence{}
	if obs.Evidence().Count() != 3 {
		t.Fatal("Evidence() must return a defensive copy")
	}
}

// TestFilterObservation_TerminalReasonAndOccurrenceAreImmutable verifies that
// TerminalReason (string) and OccurredAt (time.Time) are value types and
// cannot be mutated through the accessor.
func TestFilterObservation_TerminalReasonAndOccurrenceAreImmutable(t *testing.T) {
	obs, err := newFilterObservation(1, FilterObservationInput{
		Kind:              ObservationKindTerminalCommitted,
		StableCorrelation: "req-1",
		ConfigGeneration:  "gen-1",
		AttemptID:         "attempt-1",
		AttemptTarget:     testObservationAttemptTarget(t),
		CommitState:       CommitStateTerminalCommitted,
		TerminalReason:    TerminalReasonCompleted,
		OccurredAt:        testOccurredAt(),
	})
	if err != nil {
		t.Fatalf("newFilterObservation: %v", err)
	}

	reason := obs.TerminalReason()
	if reason != TerminalReasonCompleted {
		t.Fatalf("TerminalReason() = %s, want completed", reason)
	}

	ts := obs.OccurredAt()
	if ts.Unix() != testOccurredAt().Unix() {
		t.Fatalf("OccurredAt() = %v, want %v", ts, testOccurredAt())
	}
}

// --- OBS-CORE-4: Runtime Observation Tests -----------------------------------

type testObservationRecordingSink struct {
	mu   sync.Mutex
	got  []FilterObservation
	fail error
}

func (s *testObservationRecordingSink) Emit(ctx context.Context, obs FilterObservation) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.got = append(s.got, obs)
	return s.fail
}

func (s *testObservationRecordingSink) observations() []FilterObservation {
	s.mu.Lock()
	defer s.mu.Unlock()
	cp := make([]FilterObservation, len(s.got))
	copy(cp, s.got)
	return cp
}

// TestRuntimeObservation_LifecycleOrderAndCorrelation verifies that a full
// request execution emits observations in strict monotonic sequence order, with
// correct request/attempt correlation, exact lifecycle counts, and valid commit
// state transitions.
func TestRuntimeObservation_LifecycleOrderAndCorrelation(t *testing.T) {
	obsSink := &testObservationRecordingSink{}
	releaseSink := newFixtureSink()

	f1 := &customMockFilter{id: "filter.alpha"}
	f2 := &customMockFilter{id: "filter.beta"}
	reg1, err := NewFilterRegistration(f1, "cap.alpha", true, FilterEnforcementBlocking, 5*time.Second, 10)
	if err != nil {
		t.Fatalf("NewFilterRegistration f1: %v", err)
	}
	reg2, err := NewFilterRegistration(f2, "cap.alpha", true, FilterEnforcementBlocking, 5*time.Second, 20)
	if err != nil {
		t.Fatalf("NewFilterRegistration f2: %v", err)
	}

	snap := createTestRuntimeSnapshot(t, []FilterRegistration{reg1, reg2}, &fixtureDispatcher{}, &fixtureRebuilder{}, releaseSink).
		WithObservationSink(obsSink)

	evStart := mustResponseStart("default", 200, nil, time.Now())
	evText := mustTextDelta("default", "hello", time.Now())
	evTerm := mustTerminal("default", time.Now())

	source := newSliceEventSource([]NormalizedEvent{evStart, evText, evTerm})
	ctrl := &fixtureController{name: "ctrl-1"}
	binding, err := NewAttemptBinding("attempt.1", "mod", "prov", "path", source, ctrl)
	if err != nil {
		t.Fatalf("NewAttemptBinding: %v", err)
	}

	rt, err := NewRequestRuntime(snap, "group-a", binding)
	if err != nil {
		t.Fatalf("NewRequestRuntime: %v", err)
	}

	if err := rt.Run(context.Background()); err != nil {
		t.Fatalf("rt.Run: %v", err)
	}

	obss := obsSink.observations()
	if len(obss) == 0 {
		t.Fatal("expected observations to be recorded, got 0")
	}

	// 1. Verify sequence numbers start at 1 and are strictly monotonic.
	type expectedObservationRow struct {
		kind     ObservationKind
		filterID string
		epochID  uint64
	}

	wantRows := []expectedObservationRow{
		{ObservationKindResponseStaged, "", 0},
		{ObservationKindFilterEvaluationStarted, "filter.alpha", 1},
		{ObservationKindFilterEvaluationStarted, "filter.beta", 1},
		{ObservationKindFilterEvaluated, "filter.alpha", 1},
		{ObservationKindFilterEvaluated, "filter.beta", 1},
		{ObservationKindArbitrationDecided, "", 1},
		{ObservationKindReleaseCommitted, "", 1},
		{ObservationKindFilterEvaluationStarted, "filter.alpha", 2},
		{ObservationKindFilterEvaluationStarted, "filter.beta", 2},
		{ObservationKindFilterEvaluated, "filter.alpha", 2},
		{ObservationKindFilterEvaluated, "filter.beta", 2},
		{ObservationKindArbitrationDecided, "", 2},
		{ObservationKindReleaseCommitted, "", 2},
		{ObservationKindTerminalCommitted, "", 2},
	}

	if len(obss) != len(wantRows) {
		t.Fatalf("total observation count = %d, want %d", len(obss), len(wantRows))
	}

	for i, obs := range obss {
		expectedSeq := uint64(i + 1)
		if obs.Sequence() != expectedSeq {
			t.Errorf("observation[%d] sequence = %d, want %d", i, obs.Sequence(), expectedSeq)
		}
		if obs.StableCorrelation() != snap.RequestID() {
			t.Errorf("observation[%d] correlation = %s, want %s", i, obs.StableCorrelation(), snap.RequestID())
		}
		if obs.ConfigGeneration() != snap.ConfigGeneration() {
			t.Errorf("observation[%d] configGen = %s, want %s", i, obs.ConfigGeneration(), snap.ConfigGeneration())
		}
		if obs.AttemptID() != "attempt.1" {
			t.Errorf("observation[%d] attemptID = %s, want attempt.1", i, obs.AttemptID())
		}
		wantTarget, _ := NewObservationAttemptTarget("group-a", "mod", "prov", "path")
		if obs.AttemptTarget() != wantTarget {
			t.Errorf("observation[%d] target = %+v, want %+v", i, obs.AttemptTarget(), wantTarget)
		}

		w := wantRows[i]
		if obs.Kind() != w.kind {
			t.Errorf("observation[%d] kind = %s, want %s", i, obs.Kind(), w.kind)
		}
		if obs.EpochID() != w.epochID {
			t.Errorf("observation[%d] epoch = %d, want %d", i, obs.EpochID(), w.epochID)
		}
		if w.filterID != "" {
			if obs.Attribution() == nil {
				t.Errorf("observation[%d] attribution is nil, want filter %s", i, w.filterID)
			} else if obs.Attribution().FilterID() != w.filterID {
				t.Errorf("observation[%d] filterID = %s, want %s", i, obs.Attribution().FilterID(), w.filterID)
			}
		} else {
			if obs.Attribution() != nil {
				t.Errorf("observation[%d] unexpected attribution: %v", i, obs.Attribution())
			}
		}
	}

	// 2. Count exact observation kinds.
	kindCounts := make(map[ObservationKind]int)
	for _, obs := range obss {
		kindCounts[obs.Kind()]++
	}

	wantCounts := map[ObservationKind]int{
		ObservationKindResponseStaged:          1,
		ObservationKindFilterEvaluationStarted: 4,
		ObservationKindFilterEvaluated:         4,
		ObservationKindArbitrationDecided:      2,
		ObservationKindReleaseCommitted:        2,
		ObservationKindTerminalCommitted:       1,
	}

	for kind, wantCount := range wantCounts {
		if got := kindCounts[kind]; got != wantCount {
			t.Errorf("kind %s count = %d, want %d", kind, got, wantCount)
		}
	}

	// 3. Verify exact lifecycle ordering and positional attributes.
	if obss[0].Hold() == nil {
		t.Errorf("observation[0] hold info is nil")
	}

	// Epoch 1 arbitration (index 5)
	if obss[5].Arbitration() == nil {
		t.Errorf("observation[5] arbitration info is nil")
	}

	// Epoch 1 release (index 6)
	if obss[6].CommitState() != CommitStateStreamOpen {
		t.Errorf("observation[6] commitState = %s, want stream_open", obss[6].CommitState())
	}

	// Epoch 2 arbitration (index 11)
	if obss[11].Arbitration() == nil {
		t.Errorf("observation[11] arbitration info is nil")
	}

	// Terminal success releases its safe pending prefix before committing the terminal.
	if obss[12].CommitState() != CommitStateStreamOpen {
		t.Errorf("observation[12] commitState = %s, want stream_open", obss[12].CommitState())
	}
	if obss[13].CommitState() != CommitStateTerminalCommitted {
		t.Errorf("observation[13] commitState = %s, want terminal_committed", obss[13].CommitState())
	}
	if obss[13].TerminalReason() == "" {
		t.Errorf("observation[13] terminalReason is empty")
	}
}

// TestRuntimeObservation_DeltaSuppression verifies that non-ready / un-evaluated
// filters do not emit filter_evaluation_started or filter_evaluated observations.
func TestRuntimeObservation_DeltaSuppression(t *testing.T) {
	obsSink := &testObservationRecordingSink{}
	releaseSink := newFixtureSink()

	fEvaluated := &customMockFilter{id: "filter.evaluated"}
	fUnready := &customMockFilter{
		id: "filter.unready",
		appliesFn: func(c FilterContext) bool {
			return false // not applicable for request/attempt context
		},
	}

	reg1, _ := NewFilterRegistration(fEvaluated, "cap.alpha", true, FilterEnforcementBlocking, 5*time.Second, 10)
	reg2, _ := NewFilterRegistration(fUnready, "cap.alpha", true, FilterEnforcementBlocking, 5*time.Second, 20)

	snap := createTestRuntimeSnapshot(t, []FilterRegistration{reg1, reg2}, &fixtureDispatcher{}, &fixtureRebuilder{}, releaseSink).
		WithObservationSink(obsSink)

	evText := mustTextDelta("default", "hello", time.Now())
	evTerm := mustTerminal("default", time.Now())

	source := newSliceEventSource([]NormalizedEvent{evText, evTerm})
	ctrl := &fixtureController{name: "ctrl-1"}
	binding, _ := NewAttemptBinding("attempt.1", "mod", "prov", "path", source, ctrl)

	rt, err := NewRequestRuntime(snap, "group-a", binding)
	if err != nil {
		t.Fatalf("NewRequestRuntime: %v", err)
	}

	if err := rt.Run(context.Background()); err != nil {
		t.Fatalf("rt.Run: %v", err)
	}

	obss := obsSink.observations()
	for _, obs := range obss {
		if obs.Kind() == ObservationKindFilterEvaluationStarted || obs.Kind() == ObservationKindFilterEvaluated {
			if obs.Attribution() != nil && obs.Attribution().FilterID() == "filter.unready" {
				t.Errorf("unready/un-evaluated filter emitted observation kind %s", obs.Kind())
			}
		}
	}
}

type mockGateArbiter struct {
	mu                sync.Mutex
	calls             int
	lastBatch         EvidenceBatch
	lastSet           EvaluationSet
	requiredCompleted []<-chan struct{}
}

func (m *mockGateArbiter) Arbitrate(_ context.Context, b EvidenceBatch, s EvaluationSet) (ArbitrationResult, error) {
	for _, done := range m.requiredCompleted {
		select {
		case <-done:
		default:
			return ArbitrationResult{}, errors.New("arbiter invoked before all ready evaluators completed")
		}
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	m.calls++
	m.lastBatch = b
	m.lastSet = s
	baseDisp, _ := b.BaseDisposition()
	return NewArbitrationResult(ArbitrationActionRelease, baseDisp, "", "", nil, nil)
}

func (m *mockGateArbiter) count() int {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.calls
}

type controlledGateEvaluator struct {
	id        string
	enteredCh chan string
	waitCh    chan struct{}
	doneCh    chan struct{}
}

func (e *controlledGateEvaluator) ID() string { return e.id }
func (e *controlledGateEvaluator) Evaluate(ctx context.Context, batch EvidenceBatch) (FilterDecision, error) {
	if e.enteredCh != nil {
		select {
		case e.enteredCh <- e.id:
		case <-ctx.Done():
			return FilterDecision{}, ctx.Err()
		}
	}
	if e.waitCh != nil {
		select {
		case <-e.waitCh:
		case <-ctx.Done():
			return FilterDecision{}, ctx.Err()
		}
	}
	if e.doneCh != nil {
		defer close(e.doneCh)
	}
	evi, _ := NewSanitizedEvidence(EventKindTextDelta, "ch", "rule.1", "desc", FixedFingerprint{1}, 1, 0, FilterOutcomeKindEvaluated, testTS)
	return NewFilterDecision(FilterDecisionKindPass, "consumer.1", e.id, "rule.1", evi, nil)
}

// TestGateCoordinatorObservation_ParallelCompletionAndDeltaSuppression verifies
// that GateCoordinator.Evaluate waits for all ready filters to complete before
// invoking Arbiter, includes pre-evaluation outcomes for non-ready filters in
// EvaluationSet, and suppresses observation emission for non-ready filters.
func TestGateCoordinatorObservation_ParallelCompletionAndDeltaSuppression(t *testing.T) {
	enteredCh := make(chan string, 4)
	ch1 := make(chan struct{})
	ch2 := make(chan struct{})
	done1 := make(chan struct{})
	done2 := make(chan struct{})

	obsSink := &testObservationRecordingSink{}
	arb := &mockGateArbiter{
		requiredCompleted: []<-chan struct{}{done1, done2},
	}

	coord, err := NewGateCoordinator(context.Background(), GateCoordinatorOptions{
		MaxQueuedEpochs: 0,
		ObservationSink: obsSink,
	}, arb)
	if err != nil {
		t.Fatalf("NewGateCoordinator: %v", err)
	}
	defer coord.Close()

	eval1 := &controlledGateEvaluator{id: "filter.ready.1", enteredCh: enteredCh, waitCh: ch1, doneCh: done1}
	eval2 := &controlledGateEvaluator{id: "filter.ready.2", enteredCh: enteredCh, waitCh: ch2, doneCh: done2}
	evalUnready1 := &controlledGateEvaluator{id: "filter.unready.unsubscribed"}
	evalUnready2 := &controlledGateEvaluator{id: "filter.unready.deferred"}

	f1, err := NewEpochFilter(eval1, 10, true /*subscribed*/, true /*triggerReady*/, true /*blocksRelease*/, FilterEnforcementBlocking, 5*time.Second)
	if err != nil {
		t.Fatalf("NewEpochFilter f1: %v", err)
	}
	f2, err := NewEpochFilter(eval2, 20, true /*subscribed*/, true /*triggerReady*/, true /*blocksRelease*/, FilterEnforcementBlocking, 5*time.Second)
	if err != nil {
		t.Fatalf("NewEpochFilter f2: %v", err)
	}
	// Non-ready filter 1: unsubscribed
	fUnready1, err := NewEpochFilter(evalUnready1, 30, false /*subscribed*/, true /*triggerReady*/, true /*blocksRelease*/, FilterEnforcementBlocking, 5*time.Second)
	if err != nil {
		t.Fatalf("NewEpochFilter fUnready1: %v", err)
	}
	// Non-ready filter 2: trigger not ready + blocks release -> deferred
	fUnready2, err := NewEpochFilter(evalUnready2, 40, true /*subscribed*/, false /*triggerReady*/, true /*blocksRelease*/, FilterEnforcementBlocking, 5*time.Second)
	if err != nil {
		t.Fatalf("NewEpochFilter fUnready2: %v", err)
	}

	evText, _ := NewTextDeltaEvent("ch", "data", testTS)
	batch, err := NewEvidenceBatch([]NormalizedEvent{evText}, nil, nil, nil, false, CommitStateTransportUncommitted, testTS)
	if err != nil {
		t.Fatalf("NewEvidenceBatch: %v", err)
	}

	obsCtx := ObservationContext{
		Sequencer:         coord.sequencer,
		StableCorrelation: "req-gate-1",
		ConfigGeneration:  "gen-gate-1",
		AttemptID:         "attempt-gate-1",
		AttemptTarget:     testObservationAttemptTarget(t),
		EpochID:           1,
	}
	ctx := WithObservationContext(context.Background(), obsCtx)

	evalDone := make(chan struct{})
	var evalSet EvaluationSet
	var evalRes ArbitrationResult
	var evalErr error

	go func() {
		evalSet, evalRes, evalErr = coord.Evaluate(ctx, batch, []EpochFilter{f1, f2, fUnready1, fUnready2})
		close(evalDone)
	}()

	// Handshake: Wait for both ready evaluators to enter concurrently
	entered := make(map[string]bool)
	for len(entered) < 2 {
		select {
		case id := <-enteredCh:
			entered[id] = true
		case <-time.After(5 * time.Second):
			t.Fatal("timed out waiting for ready evaluators to enter")
		}
	}
	if !entered["filter.ready.1"] || !entered["filter.ready.2"] {
		t.Fatalf("expected filter.ready.1 and filter.ready.2 to enter, got %v", entered)
	}

	// 1. Verify Arbiter has NOT been called while ready filters are waiting on channels
	if got := arb.count(); got != 0 {
		t.Fatalf("arbiter count = %d before filters complete, want 0", got)
	}

	// Release second ready filter first (reverse order completion)
	close(ch2)
	select {
	case <-done2:
	case <-time.After(5 * time.Second):
		t.Fatal("timed out waiting for filter.ready.2 completion")
	}

	// 2. Verify Arbiter STILL has not been called (all-complete barrier waiting for second filter)
	if got := arb.count(); got != 0 {
		t.Fatalf("arbiter count = %d after only filter 2 completes, want 0", got)
	}

	// Release first ready filter
	close(ch1)
	select {
	case <-done1:
	case <-time.After(5 * time.Second):
		t.Fatal("timed out waiting for filter.ready.1 completion")
	}

	select {
	case <-evalDone:
	case <-time.After(5 * time.Second):
		t.Fatal("Evaluate did not complete")
	}

	if evalErr != nil {
		t.Fatalf("Evaluate returned error: %v", evalErr)
	}
	_ = evalRes

	// 3. Verify Arbiter called EXACTLY ONCE after all ready filters complete
	if arb.count() != 1 {
		t.Fatalf("arbiter count = %d after both complete, want 1", arb.count())
	}

	// 4. Verify EvaluationSet contains outcomes for all 4 filters
	if evalSet.Len() != 4 {
		t.Fatalf("evalSet len = %d, want 4", evalSet.Len())
	}

	outUnready1, err := evalSet.ByID("filter.unready.unsubscribed")
	if err != nil {
		t.Fatalf("evalSet.ByID filter.unready.unsubscribed: %v", err)
	}
	if outUnready1.Outcome().Kind() != FilterOutcomeKindNotApplicableForEpoch {
		t.Errorf("unready1 outcome kind = %s, want not_applicable_for_epoch", outUnready1.Outcome().Kind())
	}

	outUnready2, err := evalSet.ByID("filter.unready.deferred")
	if err != nil {
		t.Fatalf("evalSet.ByID filter.unready.deferred: %v", err)
	}
	if outUnready2.Outcome().Kind() != FilterOutcomeKindDeferredByRequirement {
		t.Errorf("unready2 outcome kind = %s, want deferred_by_requirement", outUnready2.Outcome().Kind())
	}

	if _, err := evalSet.ByID("filter.ready.1"); err != nil {
		t.Fatalf("evalSet.ByID filter.ready.1: %v", err)
	}
	if _, err := evalSet.ByID("filter.ready.2"); err != nil {
		t.Fatalf("evalSet.ByID filter.ready.2: %v", err)
	}

	// 5. Verify observation sink received events ONLY for ready filters with exact row matching
	obss := obsSink.observations()
	if len(obss) != 4 {
		t.Fatalf("total observation count = %d, want 4 (2 start + 2 evaluated)", len(obss))
	}

	wantObss := []struct {
		seq      uint64
		kind     ObservationKind
		filterID string
	}{
		{1, ObservationKindFilterEvaluationStarted, "filter.ready.1"},
		{2, ObservationKindFilterEvaluationStarted, "filter.ready.2"},
		{3, ObservationKindFilterEvaluated, "filter.ready.1"},
		{4, ObservationKindFilterEvaluated, "filter.ready.2"},
	}

	for i, obs := range obss {
		w := wantObss[i]
		if obs.Sequence() != w.seq {
			t.Errorf("observation[%d] sequence = %d, want %d", i, obs.Sequence(), w.seq)
		}
		if obs.Kind() != w.kind {
			t.Errorf("observation[%d] kind = %s, want %s", i, obs.Kind(), w.kind)
		}
		if obs.EpochID() != 1 {
			t.Errorf("observation[%d] epochID = %d, want 1", i, obs.EpochID())
		}
		if obs.StableCorrelation() != "req-gate-1" {
			t.Errorf("observation[%d] correlation = %s, want req-gate-1", i, obs.StableCorrelation())
		}
		if obs.ConfigGeneration() != "gen-gate-1" {
			t.Errorf("observation[%d] configGen = %s, want gen-gate-1", i, obs.ConfigGeneration())
		}
		if obs.AttemptID() != "attempt-gate-1" {
			t.Errorf("observation[%d] attemptID = %s, want attempt-gate-1", i, obs.AttemptID())
		}
		attr := obs.Attribution()
		if attr == nil {
			t.Errorf("observation[%d] attribution is nil", i)
		} else if attr.FilterID() != w.filterID {
			t.Errorf("observation[%d] filterID = %s, want %s", i, attr.FilterID(), w.filterID)
		}
		if attr != nil && (attr.FilterID() == "filter.unready.unsubscribed" || attr.FilterID() == "filter.unready.deferred") {
			t.Errorf("non-ready filter %s emitted observation kind %s", attr.FilterID(), obs.Kind())
		}
	}
}

// TestRuntimeObservation_SinkFailureIsolation verifies that a failing observation
// sink never alters request execution, stream release, or terminal commit.
func TestRuntimeObservation_SinkFailureIsolation(t *testing.T) {
	failingObsSink := &testObservationRecordingSink{
		fail: errors.New("observability backend write failed"),
	}
	releaseSink := newFixtureSink()

	f := &customMockFilter{id: "filter.alpha"}
	reg, _ := NewFilterRegistration(f, "cap.alpha", true, FilterEnforcementBlocking, 5*time.Second, 10)

	snap := createTestRuntimeSnapshot(t, []FilterRegistration{reg}, &fixtureDispatcher{}, &fixtureRebuilder{}, releaseSink).
		WithObservationSink(failingObsSink)

	evText := mustTextDelta("default", "hello", time.Now())
	evTerm := mustTerminal("default", time.Now())

	source := newSliceEventSource([]NormalizedEvent{evText, evTerm})
	ctrl := &fixtureController{name: "ctrl-1"}
	binding, _ := NewAttemptBinding("attempt.1", "mod", "prov", "path", source, ctrl)

	rt, err := NewRequestRuntime(snap, "group-a", binding)
	if err != nil {
		t.Fatalf("NewRequestRuntime: %v", err)
	}

	// Must succeed without propagating sink failure
	if err := rt.Run(context.Background()); err != nil {
		t.Fatalf("rt.Run failed despite sink error isolation: %v", err)
	}

	if len(releaseSink.terminals) != 1 || !releaseSink.terminals[0].Success() {
		t.Fatalf("expected successful terminal release downstream, got %+v", releaseSink.terminals)
	}

	// Verify sink received observations despite erroring internally
	if len(failingObsSink.observations()) == 0 {
		t.Fatal("expected failing sink to receive observations before erroring")
	}
}

// --- OBS-CORE-4: Recovery Lifecycle and Complete Timeline Closure Tests ---

// TestRecoveryCoordinator_ObservationHooks verifies that RecoveryCoordinator.Execute
// emits the full set of recovery lifecycle observations (selected, aborted, prepared,
// rebuilt, dispatched) in strict monotonic sequence order with non-nil recovery
// and preparer info.
func TestRecoveryCoordinator_ObservationHooks(t *testing.T) {
	recorder := &coordinatorCallRecorder{}
	preparedDirective, err := NewRecoveryDirectiveContinuation(10, "prep.snap")
	if err != nil {
		t.Fatalf("NewRecoveryDirectiveContinuation: %v", err)
	}
	preparer := &coordinatorPreparer{recorder: recorder, id: "preparer-obs-1", directive: preparedDirective}
	snapshot := &coordinatorPreparationSnapshot{recorder: recorder, ref: "prep.snapshot.ref"}
	rebuilder := &coordinatorRebuilder{recorder: recorder, promptTokens: 30}
	dispatcher := &coordinatorDispatcher{recorder: recorder}

	obsSink := &testObservationRecordingSink{}
	coordinator, err := NewRecoveryCoordinator(RecoveryCoordinatorOptions{
		Policy:             coordinatorPolicy(t, 3, nil),
		Usage:              coordinatorUsage(t, 0, nil),
		RequestSnapshot:    coordinatorRequestSnapshot(t),
		CurrentBinding:     coordinatorBinding(t, recorder, nil),
		Rebuilder:          rebuilder,
		Dispatcher:         dispatcher,
		Preparers:          map[string]RecoveryPlanPreparer{"preparer-obs-1": preparer},
		PreparationTimeout: 5 * time.Second,
		ObservationSink:    obsSink,
	})
	if err != nil {
		t.Fatalf("NewRecoveryCoordinator: %v", err)
	}

	input := coordinatorCycleInput(t, "plan.obs.test", RecoveryStrategyContinuationRepair)
	input.PreparerID = "preparer-obs-1"
	input.PreparationSnapshot = snapshot

	initialTarget, _ := NewObservationAttemptTarget("model-group", "model.initial", "provider.initial", "tunnel")
	reboundTarget, _ := NewObservationAttemptTarget("model-group", "model.actual", "provider.actual", "normalized")
	obsCtx := ObservationContext{
		Sequencer:         NewObservationSequencer(obsSink, nil),
		StableCorrelation: "req-obs-123",
		ConfigGeneration:  "gen-obs-123",
		AttemptID:         "attempt-initial-1",
		AttemptTarget:     initialTarget,
		EpochID:           1,
	}
	ctx := WithObservationContext(context.Background(), obsCtx)

	res, err := coordinator.Execute(ctx, input)
	if err != nil {
		t.Fatalf("Execute: %v", err)
	}
	if !res.PreviousAttemptClosed() {
		t.Fatal("expected previous attempt closed")
	}

	obss := obsSink.observations()
	if len(obss) != 5 {
		t.Fatalf("observation count = %d, want 5 (selected, aborted, prepared, rebuilt, dispatched)", len(obss))
	}

	wantKinds := []ObservationKind{
		ObservationKindRecoveryPlanSelected,
		ObservationKindRecoveryAttemptAborted,
		ObservationKindRecoveryPrepared,
		ObservationKindRecoveryRebuilt,
		ObservationKindRecoveryDispatched,
	}

	for i, obs := range obss {
		wantSeq := uint64(i + 1)
		if obs.Sequence() != wantSeq {
			t.Errorf("observation[%d] sequence = %d, want %d", i, obs.Sequence(), wantSeq)
		}
		if obs.Kind() != wantKinds[i] {
			t.Errorf("observation[%d] kind = %s, want %s", i, obs.Kind(), wantKinds[i])
		}
		if obs.StableCorrelation() != "req-obs-123" {
			t.Errorf("observation[%d] correlation = %s, want req-obs-123", i, obs.StableCorrelation())
		}
		if obs.ConfigGeneration() != "gen-obs-123" {
			t.Errorf("observation[%d] configGen = %s, want gen-obs-123", i, obs.ConfigGeneration())
		}
		if obs.EpochID() != 1 {
			t.Errorf("observation[%d] epochID = %d, want 1", i, obs.EpochID())
		}
		if obs.Recovery() == nil {
			t.Fatalf("observation[%d] recovery info is nil", i)
		}
		if obs.Recovery().PlanID() != "plan.obs.test" {
			t.Errorf("observation[%d] planID = %s, want plan.obs.test", i, obs.Recovery().PlanID())
		}
		if obs.Recovery().Strategy() != RecoveryStrategyContinuationRepair {
			t.Errorf("observation[%d] strategy = %s, want continuation_repair", i, obs.Recovery().Strategy())
		}
		if obs.Recovery().ResumeMode() != RecoveryResumeModeContinueStream {
			t.Errorf("observation[%d] resumeMode = %s, want continue_stream", i, obs.Recovery().ResumeMode())
		}
		if obs.Recovery().SharedAttemptID() != "attempt.initial" {
			t.Errorf("observation[%d] sharedAttemptID = %s, want attempt.initial", i, obs.Recovery().SharedAttemptID())
		}
		wantAttemptID, wantTarget := "attempt-initial-1", initialTarget
		if obs.Kind() == ObservationKindRecoveryDispatched {
			wantAttemptID, wantTarget = "attempt.recovery.a", reboundTarget
		}
		if obs.AttemptID() != wantAttemptID || obs.AttemptTarget() != wantTarget {
			t.Errorf("observation[%d] attempt/target = (%s,%+v), want (%s,%+v)", i, obs.AttemptID(), obs.AttemptTarget(), wantAttemptID, wantTarget)
		}

		if obs.Kind() == ObservationKindRecoveryPrepared {
			prep := obs.Preparer()
			if prep == nil {
				t.Fatalf("observation[%d] preparer info is nil", i)
			}
			if prep.PreparerID() != "preparer-obs-1" {
				t.Errorf("preparerID = %s, want preparer-obs-1", prep.PreparerID())
			}
			if prep.Status() != RecoveryPreparationPrepared {
				t.Errorf("preparer status = %s, want prepared", prep.Status())
			}
			if prep.DeadlineOutcome() != ObservationDeadlineOutcomeCompleted {
				t.Errorf("preparer deadlineOutcome = %s, want completed", prep.DeadlineOutcome())
			}
		} else {
			if obs.Preparer() != nil {
				t.Errorf("observation[%d] unexpected preparer info for kind %s", i, obs.Kind())
			}
		}
	}
}

type matrixChangingPreparationSnapshot struct {
	recorder *coordinatorCallRecorder
	refs     []string
	calls    int
}

func (s *matrixChangingPreparationSnapshot) SnapshotRef() string {
	idx := s.calls
	s.calls++
	if idx >= len(s.refs) {
		idx = len(s.refs) - 1
	}
	return s.refs[idx]
}

func (s *matrixChangingPreparationSnapshot) Release() error {
	s.recorder.add("snapshot.release:" + s.refs[len(s.refs)-1])
	return nil
}

type observationRow struct {
	kind     ObservationKind
	causes   []string
	status   *RecoveryPreparationStatus
	deadline *ObservationDeadlineOutcome
}

func assertObservationRows(t *testing.T, got []FilterObservation, want []observationRow) {
	t.Helper()
	if len(got) != len(want) {
		t.Fatalf("observation count = %d, want %d", len(got), len(want))
	}
	for i, row := range want {
		obs := got[i]
		if obs.Sequence() != uint64(i+1) {
			t.Errorf("observation[%d] sequence = %d, want %d", i, obs.Sequence(), i+1)
		}
		if obs.Kind() != row.kind {
			t.Errorf("observation[%d] kind = %s, want %s", i, obs.Kind(), row.kind)
		}
		var codes []string
		for j := 0; j < obs.Causes().Len(); j++ {
			cause, err := obs.Causes().At(j)
			if err != nil {
				t.Fatalf("observation[%d] cause[%d]: %v", i, j, err)
			}
			codes = append(codes, cause.Code())
		}
		if !reflect.DeepEqual(codes, row.causes) {
			t.Errorf("observation[%d] causes = %v, want %v", i, codes, row.causes)
		}
		if row.status == nil {
			if obs.Preparer() != nil {
				t.Errorf("observation[%d] unexpected preparer", i)
			}
			continue
		}
		prep := obs.Preparer()
		if prep == nil {
			t.Errorf("observation[%d] preparer is nil", i)
			continue
		}
		if prep.Status() != *row.status || prep.DeadlineOutcome() != *row.deadline {
			t.Errorf("observation[%d] preparer = %s/%s, want %s/%s", i, prep.Status(), prep.DeadlineOutcome(), *row.status, *row.deadline)
		}
	}
}

// TestRecoveryCoordinator_ObservationResultMatrix exercises RecoveryCoordinator.Execute
// itself across every result boundary and asserts the entire emitted lifecycle rather
// than a synthetic observation stream or only its final row.
func TestRecoveryCoordinator_ObservationResultMatrix(t *testing.T) {
	hostErr := errors.New("host failure")
	required := RecoveryPreparationRequired
	failed, timeout := ObservationDeadlineOutcomeFailed, ObservationDeadlineOutcomeTimeout
	tests := []struct {
		name           string
		strategy       RecoveryStrategy
		preparer       *coordinatorPreparer
		snapshot       RecoveryPreparationSnapshot
		rebuildErr     error
		dispatchErr    error
		invalidBinding bool
		controllerErr  error
		primeUsedPlan  bool
		setup          func(*RecoveryCycleInput)
		wantErr        error
		wantAnyErr     bool
		wantRows       []observationRow
		wantCalls      []string
		wantClosed     bool
		wantUsage      int
	}{
		{
			name: "snapshot missing", strategy: RecoveryStrategyContinuationRepair,
			setup: func(in *RecoveryCycleInput) { in.PreparerID = "preparer.matrix" }, wantErr: ErrRecoveryPreparationInvalid,
			wantRows: []observationRow{{kind: ObservationKindRecoveryPlanRejected, causes: []string{"preparation_snapshot_missing"}}},
		},
		{
			name: "snapshot unexpected release failure", strategy: RecoveryStrategyExactReplay,
			snapshot: &coordinatorPreparationSnapshot{ref: "snapshot.unexpected", err: hostErr}, wantErr: ErrRecoveryPreparationInvalid,
			wantRows:  []observationRow{{kind: ObservationKindRecoveryPlanRejected, causes: []string{"preparer_missing", "preparation_snapshot_release_failed"}}},
			wantCalls: []string{"snapshot.release:snapshot.unexpected"},
		},
		{
			name: "invalid contributor", strategy: RecoveryStrategyExactReplay,
			setup: func(in *RecoveryCycleInput) { in.ConsumerID = "" }, wantAnyErr: true,
			wantRows: []observationRow{{kind: ObservationKindRecoveryPlanRejected, causes: []string{"plan_ineligible"}}},
		},
		{
			name: "plan reentry", strategy: RecoveryStrategyContinuationRepair,
			snapshot:      &coordinatorPreparationSnapshot{ref: "snapshot.reentry"},
			setup:         func(in *RecoveryCycleInput) { in.PreparerID = "preparer.matrix" },
			primeUsedPlan: true,
			wantErr:       ErrRecoveryPlanReentry,
			wantRows:      []observationRow{{kind: ObservationKindRecoveryPlanRejected, causes: []string{"plan_reentry"}}},
			wantCalls:     []string{"snapshot.release:snapshot.reentry"},
		},
		{
			name: "preparer snapshot mismatch", strategy: RecoveryStrategyContinuationRepair,
			preparer: &coordinatorPreparer{id: "preparer.matrix"},
			snapshot: &matrixChangingPreparationSnapshot{refs: []string{"snapshot.first", "snapshot.second"}},
			setup:    func(in *RecoveryCycleInput) { in.PreparerID = "preparer.matrix" }, wantErr: ErrRecoveryPreparationInvalid,
			wantRows:  []observationRow{{kind: ObservationKindRecoveryPlanSelected}, {kind: ObservationKindRecoveryAttemptAborted}, {kind: ObservationKindRecoveryPrepared, causes: []string{"preparation_snapshot_mismatch"}, status: &required, deadline: &failed}},
			wantCalls: []string{"abort:attempt.initial", "snapshot.release:snapshot.second"}, wantClosed: true,
		},
		{
			name: "preparer unavailable", strategy: RecoveryStrategyContinuationRepair,
			snapshot: &coordinatorPreparationSnapshot{ref: "snapshot.unavailable"},
			setup:    func(in *RecoveryCycleInput) { in.PreparerID = "preparer.matrix" }, wantErr: ErrRecoveryPreparerUnavailable,
			wantRows:  []observationRow{{kind: ObservationKindRecoveryPlanSelected}, {kind: ObservationKindRecoveryAttemptAborted}, {kind: ObservationKindRecoveryPrepared, causes: []string{"preparer_unavailable"}, status: &required, deadline: &failed}},
			wantCalls: []string{"abort:attempt.initial", "snapshot.release:snapshot.unavailable"}, wantClosed: true,
		},
		{
			name: "preparer failure", strategy: RecoveryStrategyContinuationRepair,
			preparer: &coordinatorPreparer{id: "preparer.matrix", err: hostErr}, snapshot: &coordinatorPreparationSnapshot{ref: "snapshot.failure"},
			setup: func(in *RecoveryCycleInput) { in.PreparerID = "preparer.matrix" }, wantErr: ErrRecoveryPreparationFailed,
			wantRows:  []observationRow{{kind: ObservationKindRecoveryPlanSelected}, {kind: ObservationKindRecoveryAttemptAborted}, {kind: ObservationKindRecoveryPrepared, causes: []string{"preparation_failed"}, status: &required, deadline: &failed}},
			wantCalls: []string{"abort:attempt.initial", "prepare:preparer.matrix", "snapshot.release:snapshot.failure"}, wantClosed: true,
		},
		{
			name: "preparer deadline", strategy: RecoveryStrategyContinuationRepair,
			preparer: &coordinatorPreparer{id: "preparer.matrix", wait: true}, snapshot: &coordinatorPreparationSnapshot{ref: "snapshot.deadline"},
			setup: func(in *RecoveryCycleInput) { in.PreparerID = "preparer.matrix" }, wantErr: ErrRecoveryPreparationDeadline,
			wantRows:  []observationRow{{kind: ObservationKindRecoveryPlanSelected}, {kind: ObservationKindRecoveryAttemptAborted}, {kind: ObservationKindRecoveryPrepared, causes: []string{"preparation_deadline"}, status: &required, deadline: &timeout}},
			wantCalls: []string{"abort:attempt.initial", "prepare:preparer.matrix", "snapshot.release:snapshot.deadline"}, wantClosed: true,
		},
		{
			name: "preparer invalid", strategy: RecoveryStrategyContinuationRepair,
			preparer: &coordinatorPreparer{id: "preparer.matrix"}, snapshot: &coordinatorPreparationSnapshot{ref: "snapshot.invalid"},
			setup: func(in *RecoveryCycleInput) { in.PreparerID = "preparer.matrix" }, wantErr: ErrRecoveryPreparationInvalid,
			wantRows:  []observationRow{{kind: ObservationKindRecoveryPlanSelected}, {kind: ObservationKindRecoveryAttemptAborted}, {kind: ObservationKindRecoveryPrepared, causes: []string{"preparation_invalid"}, status: &required, deadline: &failed}},
			wantCalls: []string{"abort:attempt.initial", "prepare:preparer.matrix", "snapshot.release:snapshot.invalid"}, wantClosed: true,
		},
		{
			name: "prepared snapshot release failure", strategy: RecoveryStrategyContinuationRepair,
			preparer: &coordinatorPreparer{id: "preparer.matrix", directive: mustCoordinatorContinuationDirective(t)}, snapshot: &coordinatorPreparationSnapshot{ref: "snapshot.release", err: hostErr},
			setup: func(in *RecoveryCycleInput) { in.PreparerID = "preparer.matrix" }, wantErr: ErrRecoveryPreparationSnapshotRelease,
			wantRows:  []observationRow{{kind: ObservationKindRecoveryPlanSelected}, {kind: ObservationKindRecoveryAttemptAborted}, {kind: ObservationKindRecoveryPrepared, causes: []string{"preparation_snapshot_release_failed"}, status: &required, deadline: &failed}},
			wantCalls: []string{"abort:attempt.initial", "prepare:preparer.matrix", "snapshot.release:snapshot.release"}, wantClosed: true,
		},
		{
			name: "abort failure", strategy: RecoveryStrategyExactReplay, controllerErr: hostErr, wantErr: ErrRecoveryAbortFailed,
			wantRows:  []observationRow{{kind: ObservationKindRecoveryPlanSelected}, {kind: ObservationKindRecoveryAttemptAbortFailed, causes: []string{"attempt_abort_failed"}}},
			wantCalls: []string{"abort:attempt.initial"},
		},
		{
			name: "rebuild failure", strategy: RecoveryStrategyExactReplay, rebuildErr: hostErr, wantErr: ErrRecoveryRebuildFailed,
			wantRows:  []observationRow{{kind: ObservationKindRecoveryPlanSelected}, {kind: ObservationKindRecoveryAttemptAborted}, {kind: ObservationKindRecoveryRebuildFailed, causes: []string{"request_rebuild_failed"}}},
			wantCalls: []string{"abort:attempt.initial", "rebuild:plan.matrix"}, wantClosed: true,
		},
		{
			name: "dispatch failure", strategy: RecoveryStrategyExactReplay, dispatchErr: hostErr, wantErr: ErrRecoveryDispatchFailed,
			wantRows:  []observationRow{{kind: ObservationKindRecoveryPlanSelected}, {kind: ObservationKindRecoveryAttemptAborted}, {kind: ObservationKindRecoveryRebuilt}, {kind: ObservationKindRecoveryDispatchFailed, causes: []string{"attempt_dispatch_failed"}}},
			wantCalls: []string{"abort:attempt.initial", "rebuild:plan.matrix", "dispatch:plan.matrix"}, wantClosed: true, wantUsage: 1,
		},
		{
			name: "invalid binding", strategy: RecoveryStrategyExactReplay, invalidBinding: true, wantErr: ErrRecoveryDispatchFailed,
			wantRows:  []observationRow{{kind: ObservationKindRecoveryPlanSelected}, {kind: ObservationKindRecoveryAttemptAborted}, {kind: ObservationKindRecoveryRebuilt}, {kind: ObservationKindRecoveryDispatchFailed, causes: []string{"attempt_binding_invalid"}}},
			wantCalls: []string{"abort:attempt.initial", "rebuild:plan.matrix", "dispatch:plan.matrix"}, wantClosed: true, wantUsage: 1,
		},
		{
			name: "success", strategy: RecoveryStrategyExactReplay,
			wantRows:  []observationRow{{kind: ObservationKindRecoveryPlanSelected}, {kind: ObservationKindRecoveryAttemptAborted}, {kind: ObservationKindRecoveryRebuilt}, {kind: ObservationKindRecoveryDispatched}},
			wantCalls: []string{"abort:attempt.initial", "rebuild:plan.matrix", "dispatch:plan.matrix"}, wantClosed: true, wantUsage: 1,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			recorder := &coordinatorCallRecorder{}
			rebuilder := &coordinatorRebuilder{recorder: recorder, err: tt.rebuildErr}
			dispatcher := &coordinatorDispatcher{recorder: recorder, err: tt.dispatchErr, invalid: tt.invalidBinding}
			controller := &coordinatorController{recorder: recorder, attempt: "attempt.initial", err: tt.controllerErr}
			preparers := map[string]RecoveryPlanPreparer{}
			if tt.preparer != nil {
				tt.preparer.recorder = recorder
				preparers["preparer.matrix"] = tt.preparer
			}
			obsSink := &testObservationRecordingSink{}
			coordinator, err := NewRecoveryCoordinator(RecoveryCoordinatorOptions{Policy: coordinatorPolicy(t, 3, nil), Usage: coordinatorUsage(t, 0, nil), RequestSnapshot: coordinatorRequestSnapshot(t), CurrentBinding: coordinatorBinding(t, recorder, controller), Rebuilder: rebuilder, Dispatcher: dispatcher, Preparers: preparers, PreparationTimeout: time.Millisecond, ObservationSink: obsSink})
			if err != nil {
				t.Fatalf("NewRecoveryCoordinator: %v", err)
			}
			input := coordinatorCycleInput(t, "plan.matrix", tt.strategy)
			if tt.snapshot != nil {
				switch snapshot := tt.snapshot.(type) {
				case *coordinatorPreparationSnapshot:
					snapshot.recorder = recorder
				case *matrixChangingPreparationSnapshot:
					snapshot.recorder = recorder
				}
				input.PreparationSnapshot = tt.snapshot
			}
			if tt.setup != nil {
				tt.setup(&input)
			}
			if tt.primeUsedPlan {
				coordinator.markPlanUsed(input.PlanID)
			}
			result, err := coordinator.Execute(context.Background(), input)
			if tt.wantAnyErr {
				if err == nil {
					t.Fatal("Execute error = nil, want a validation error")
				}
			} else if tt.wantErr == nil {
				if err != nil {
					t.Fatalf("Execute: %v", err)
				}
			} else if !errors.Is(err, tt.wantErr) {
				t.Fatalf("Execute error = %v, want %v", err, tt.wantErr)
			}
			assertObservationRows(t, obsSink.observations(), tt.wantRows)
			if !reflect.DeepEqual(recorder.snapshot(), tt.wantCalls) {
				t.Fatalf("host calls = %v, want %v", recorder.snapshot(), tt.wantCalls)
			}
			if result.PreviousAttemptClosed() != tt.wantClosed {
				t.Errorf("PreviousAttemptClosed = %t, want %t", result.PreviousAttemptClosed(), tt.wantClosed)
			}
			if got := result.UsageSnapshot().RequestFaultRecoveries(); got != tt.wantUsage {
				t.Errorf("fault usage = %d, want %d", got, tt.wantUsage)
			}
			if len(tt.wantRows) > 0 {
				last := obsSink.observations()[len(tt.wantRows)-1]
				if !reflect.DeepEqual(result.FailureCauses(), last.Causes()) {
					t.Errorf("final result causes = %v, observation causes = %v", result.FailureCauses(), last.Causes())
				}
			}
		})
	}
}

func mustCoordinatorContinuationDirective(t *testing.T) RecoveryDirective {
	t.Helper()
	directive, err := NewRecoveryDirectiveContinuation(1, "snapshot.directive")
	if err != nil {
		t.Fatalf("NewRecoveryDirectiveContinuation: %v", err)
	}
	return directive
}

// TestRequestRuntime_CompleteTimelineOrder verifies the complete request timeline order
// spanning filter evaluation, arbitration, recovery lifecycle, stream release, and terminal commit.
func TestRequestRuntime_CompleteTimelineOrder(t *testing.T) {
	obsSink := &testObservationRecordingSink{}
	releaseSink := newFixtureSink()

	var attemptCount int
	var countMu sync.Mutex

	attemptFilter := &customMockFilter{
		id: "filter.recoverable",
		evaluateFn: func(ctx context.Context, fc FilterContext, batch EvidenceBatch) (FilterDecision, error) {
			countMu.Lock()
			count := attemptCount
			attemptCount++
			countMu.Unlock()

			if count == 0 {
				dir := mustRecoveryDirectiveExact("req-ref-1")
				intent := mustRecoveryIntent(RecoveryStrategyExactReplay, dir, "recoverable_violation", 10)
				evi, _ := NewSanitizedEvidence(EventKindTextDelta, "ch", "rule.rec", "desc.rec", FixedFingerprint{1}, 1, 0, FilterOutcomeKindEvaluated, testTS)
				return NewFilterDecision(FilterDecisionKindViolation, "consumer.1", "filter.recoverable", "rule.rec", evi, &intent)
			}
			evi, _ := NewSanitizedEvidence(EventKindTextDelta, "ch", "rule.pass", "desc.pass", FixedFingerprint{2}, 1, 0, FilterOutcomeKindEvaluated, testTS)
			return NewFilterDecision(FilterDecisionKindPass, "consumer.1", "filter.recoverable", "rule.pass", evi, nil)
		},
	}
	reg1, _ := NewFilterRegistration(attemptFilter, "cap.alpha", true, FilterEnforcementBlocking, 5*time.Second, 10)

	rebuilder := &fixtureRebuilder{}
	dispatcher := &fixtureDispatcher{}

	snap := createTestRuntimeSnapshot(t, []FilterRegistration{reg1}, dispatcher, rebuilder, releaseSink).
		WithObservationSink(obsSink)

	evText1 := mustTextDelta("default", "bad payload", time.Now())
	source1 := newSliceEventSource([]NormalizedEvent{evText1})
	ctrl1 := &fixtureController{name: "ctrl-1"}
	binding1, _ := NewAttemptBinding("attempt.1", "mod", "prov", "path", source1, ctrl1)

	rt, err := NewRequestRuntime(snap, "group-a", binding1)
	if err != nil {
		t.Fatalf("NewRequestRuntime: %v", err)
	}

	evText2 := mustTextDelta("default", "good payload", time.Now())
	evTerm2 := mustTerminal("default", time.Now())
	source2 := newSliceEventSource([]NormalizedEvent{evText2, evTerm2})
	ctrl2 := &fixtureController{name: "ctrl-2"}
	binding2, _ := NewAttemptBinding("attempt.2", "mod", "prov", "path", source2, ctrl2)
	dispatcher.handler = func(ctx context.Context, request RebuiltRequest) (AttemptBinding, error) {
		return binding2, nil
	}

	if err := rt.Run(context.Background()); err != nil {
		t.Fatalf("rt.Run: %v", err)
	}

	obss := obsSink.observations()
	if len(obss) == 0 {
		t.Fatal("expected recorded observations, got 0")
	}

	var lastSeq uint64
	for i, obs := range obss {
		if obs.Sequence() <= lastSeq {
			t.Errorf("observation[%d] sequence %d not strictly greater than previous %d", i, obs.Sequence(), lastSeq)
		}
		lastSeq = obs.Sequence()
	}

	var kinds []ObservationKind
	for _, obs := range obss {
		kinds = append(kinds, obs.Kind())
	}

	hasRecoverSelected := slices.Contains(kinds, ObservationKindRecoveryPlanSelected)
	hasRecoverAborted := slices.Contains(kinds, ObservationKindRecoveryAttemptAborted)
	hasRecoverRebuilt := slices.Contains(kinds, ObservationKindRecoveryRebuilt)
	hasRecoverDispatched := slices.Contains(kinds, ObservationKindRecoveryDispatched)
	hasReleaseCommitted := slices.Contains(kinds, ObservationKindReleaseCommitted)
	hasTerminalCommitted := slices.Contains(kinds, ObservationKindTerminalCommitted)

	if !hasRecoverSelected || !hasRecoverAborted || !hasRecoverRebuilt || !hasRecoverDispatched {
		t.Fatalf("timeline missing recovery lifecycle observations: %v", kinds)
	}
	if !hasReleaseCommitted || !hasTerminalCommitted {
		t.Fatalf("timeline missing release or terminal committed observations: %v", kinds)
	}

	selectedIdx := slices.Index(kinds, ObservationKindRecoveryPlanSelected)
	abortedIdx := slices.Index(kinds, ObservationKindRecoveryAttemptAborted)
	rebuiltIdx := slices.Index(kinds, ObservationKindRecoveryRebuilt)
	dispatchedIdx := slices.Index(kinds, ObservationKindRecoveryDispatched)
	terminalIdx := slices.Index(kinds, ObservationKindTerminalCommitted)

	if !(selectedIdx < abortedIdx && abortedIdx < rebuiltIdx && rebuiltIdx < dispatchedIdx && dispatchedIdx < terminalIdx) {
		t.Fatalf("recovery timeline order violated: selected=%d aborted=%d rebuilt=%d dispatched=%d terminal=%d",
			selectedIdx, abortedIdx, rebuiltIdx, dispatchedIdx, terminalIdx)
	}
	if obss[selectedIdx].AttemptID() != "attempt.1" || obss[abortedIdx].AttemptID() != "attempt.1" || obss[rebuiltIdx].AttemptID() != "attempt.1" {
		t.Fatalf("pre-dispatch recovery observations must retain attempt.1")
	}
	firstAfterDispatch := func(kind ObservationKind) int {
		for i := dispatchedIdx + 1; i < len(obss); i++ {
			if obss[i].Kind() == kind {
				return i
			}
		}
		return -1
	}
	started2Idx := firstAfterDispatch(ObservationKindFilterEvaluationStarted)
	evaluated2Idx := firstAfterDispatch(ObservationKindFilterEvaluated)
	arbitration2Idx := firstAfterDispatch(ObservationKindArbitrationDecided)
	release2Idx := firstAfterDispatch(ObservationKindReleaseCommitted)
	if started2Idx < 0 || evaluated2Idx < 0 || arbitration2Idx < 0 || release2Idx < 0 {
		t.Fatalf("post-dispatch attempt.2 lifecycle is incomplete: started=%d evaluated=%d arbitration=%d release=%d", started2Idx, evaluated2Idx, arbitration2Idx, release2Idx)
	}
	if !(dispatchedIdx < started2Idx && started2Idx < evaluated2Idx && evaluated2Idx < arbitration2Idx && arbitration2Idx < release2Idx && release2Idx < terminalIdx) {
		t.Fatalf("post-dispatch timeline order violated: dispatched=%d started=%d evaluated=%d arbitration=%d release=%d terminal=%d", dispatchedIdx, started2Idx, evaluated2Idx, arbitration2Idx, release2Idx, terminalIdx)
	}
	for _, idx := range []int{dispatchedIdx, started2Idx, evaluated2Idx, arbitration2Idx, release2Idx, terminalIdx} {
		if obss[idx].AttemptID() != "attempt.2" {
			t.Errorf("observation[%d] attempt = %s, want attempt.2", idx, obss[idx].AttemptID())
		}
	}

}

// TestObservationSequencer_RecoveryObservationRace verifies concurrent observation sequencing.
// and recovery execution under the race detector.
func TestObservationSequencer_RecoveryObservationRace(t *testing.T) {
	obsSink := &testObservationRecordingSink{}
	seq := NewObservationSequencer(obsSink, nil)

	const goroutines = 10
	const iterations = 20

	var wg sync.WaitGroup
	wg.Add(goroutines)

	for g := 0; g < goroutines; g++ {
		go func(gID int) {
			defer wg.Done()
			for i := 0; i < iterations; i++ {
				recInfo, err := NewObservationRecoveryInfo("plan-race", RecoveryStrategyExactReplay, RecoveryResumeModeReplaceAttempt, "")
				if err != nil {
					t.Errorf("NewObservationRecoveryInfo: %v", err)
					return
				}
				input := FilterObservationInput{
					Kind:              ObservationKindRecoveryPlanSelected,
					StableCorrelation: "req-race-1",
					ConfigGeneration:  "gen-race-1",
					AttemptID:         "attempt-race-1",
					AttemptTarget:     testObservationAttemptTarget(t),
					EpochID:           uint64(i + 1),
					CommitState:       CommitStateTransportUncommitted,
					Recovery:          &recInfo,
					OccurredAt:        time.Now(),
				}
				_, err = seq.Emit(context.Background(), input)
				if err != nil {
					t.Errorf("Emit: %v", err)
					return
				}
			}
		}(g)
	}

	wg.Wait()

	obss := obsSink.observations()
	const wantTotal = goroutines * iterations
	if len(obss) != wantTotal {
		t.Fatalf("total observations = %d, want %d", len(obss), wantTotal)
	}

	seqMap := make(map[uint64]bool)
	for _, obs := range obss {
		if seqMap[obs.Sequence()] {
			t.Errorf("duplicate sequence number detected: %d", obs.Sequence())
		}
		seqMap[obs.Sequence()] = true
	}
	if len(seqMap) != wantTotal {
		t.Fatalf("unique sequence count = %d, want %d", len(seqMap), wantTotal)
	}
}

// TestRecoveryCoordinator_ObservationConcurrency holds the first abort in
// progress, then proves the second cycle is rejected without a second dispatch.
func TestRecoveryCoordinator_ObservationConcurrency(t *testing.T) {
	recorder := &coordinatorCallRecorder{}
	started := make(chan struct{})
	release := make(chan struct{})
	controller := &coordinatorController{recorder: recorder, attempt: "attempt.initial", started: started, release: release}
	obsSink := &testObservationRecordingSink{}
	dispatcher := &coordinatorDispatcher{recorder: recorder}
	coordinator, err := NewRecoveryCoordinator(RecoveryCoordinatorOptions{
		Policy: coordinatorPolicy(t, 3, nil), Usage: coordinatorUsage(t, 0, nil),
		RequestSnapshot: coordinatorRequestSnapshot(t), CurrentBinding: coordinatorBinding(t, recorder, controller),
		Rebuilder: &coordinatorRebuilder{recorder: recorder}, Dispatcher: dispatcher,
		ObservationSink: obsSink,
	})
	if err != nil {
		t.Fatalf("NewRecoveryCoordinator: %v", err)
	}

	first := coordinatorCycleInput(t, "plan.concurrent.one", RecoveryStrategyExactReplay)
	second := coordinatorCycleInput(t, "plan.concurrent.two", RecoveryStrategyExactReplay)
	firstDone := make(chan error, 1)
	go func() {
		_, err := coordinator.Execute(context.Background(), first)
		firstDone <- err
	}()
	<-started
	if _, err := coordinator.Execute(context.Background(), second); !errors.Is(err, ErrRecoveryCycleInProgress) {
		t.Fatalf("second Execute error = %v, want %v", err, ErrRecoveryCycleInProgress)
	}
	close(release)
	if err := <-firstDone; err != nil {
		t.Fatalf("first Execute: %v", err)
	}
	if dispatcher.callCount() != 1 {
		t.Fatalf("dispatch count = %d, want 1", dispatcher.callCount())
	}

	obss := obsSink.observations()
	assertObservationRows(t, obss, []observationRow{
		{kind: ObservationKindRecoveryPlanSelected},
		{kind: ObservationKindRecoveryPlanRejected, causes: []string{"cycle_in_progress"}},
		{kind: ObservationKindRecoveryAttemptAborted},
		{kind: ObservationKindRecoveryRebuilt},
		{kind: ObservationKindRecoveryDispatched},
	})
	for i := 0; i < 4; i++ {
		if obss[i].AttemptID() != "attempt.initial" {
			t.Errorf("observation[%d] attempt = %s, want attempt.initial", i, obss[i].AttemptID())
		}
		if obss[i].StableCorrelation() != "default" || obss[i].ConfigGeneration() != "default" || obss[i].EpochID() != 0 {
			t.Errorf("observation[%d] correlation/config/epoch = %s/%s/%d, want default/default/0", i, obss[i].StableCorrelation(), obss[i].ConfigGeneration(), obss[i].EpochID())
		}
	}
	if obss[4].AttemptID() != "attempt.recovery.a" {
		t.Errorf("dispatched attempt = %s, want attempt.recovery.a", obss[4].AttemptID())
	}
	if got, want := recorder.snapshot(), []string{"abort:attempt.initial", "rebuild:plan.concurrent.one", "dispatch:plan.concurrent.one"}; !reflect.DeepEqual(got, want) {
		t.Fatalf("host call order = %v, want %v", got, want)
	}
}
func TestFilterObservation_RecoveryFailureContract(t *testing.T) {
	cause := causeAt(t, "recovery_failure")
	causes, err := NewFailureCauseChain([]FailureCause{cause})
	if err != nil {
		t.Fatalf("NewFailureCauseChain: %v", err)
	}
	base := FilterObservationInput{
		Kind:              ObservationKindRecoveryPlanRejected,
		StableCorrelation: "req.failure",
		ConfigGeneration:  "gen.failure",
		AttemptID:         "attempt.failure",
		AttemptTarget:     testObservationAttemptTarget(t),
		CommitState:       CommitStateTransportUncommitted,
		Causes:            causes,
		OccurredAt:        testTS,
	}
	if _, err := newFilterObservation(1, base); err != nil {
		t.Fatalf("plan rejection observation: %v", err)
	}
	recovery, err := NewObservationRecoveryInfo("plan.failure", RecoveryStrategyExactReplay, RecoveryResumeModeReplaceAttempt, "")
	if err != nil {
		t.Fatalf("NewObservationRecoveryInfo: %v", err)
	}
	base.Kind = ObservationKindRecoveryAttemptAbortFailed
	base.Recovery = &recovery
	if _, err := newFilterObservation(2, base); err != nil {
		t.Fatalf("abort failure observation: %v", err)
	}
	base.Causes = FailureCauseChain{}
	if _, err := newFilterObservation(3, base); err == nil {
		t.Fatal("recovery failure observation without a cause chain was accepted")
	}
	base.Kind = ObservationKindRecoveryPlanRejected
	base.Recovery = &recovery
	base.Causes = causes
	if _, err := newFilterObservation(4, base); err == nil {
		t.Fatal("plan rejection observation with recovery info was accepted")
	}
}
