package streamgate

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"
)

// --- Test helper types for the coordinator tests ---

// recordingArbiter records every arbiter invocation for assertion.
type recordingArbiter struct {
	mu       sync.Mutex
	calls    int
	batches  []EvidenceBatch
	sets     []EvaluationSet
	firstErr error
	// blockUntil signals the arbiter to wait before returning.
	blockUntil chan struct{}
}

func (r *recordingArbiter) Arbitrate(_ context.Context, batch EvidenceBatch, set EvaluationSet) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.calls++
	r.batches = append(r.batches, batch)
	r.sets = append(r.sets, set)
	if r.blockUntil != nil {
		<-r.blockUntil
	}
	return r.firstErr
}

// blockingEvaluator blocks until the shared signal channel is closed.
// After signal it returns the given decision and error.
type blockingEvaluator struct {
	id     string
	signal chan struct{}
	// decision and err are read under no lock; the test populates them
	// before closing the shared signal channel.
	decision FilterDecision
	err      error
}

func (b *blockingEvaluator) ID() string { return b.id }
func (b *blockingEvaluator) Evaluate(_ context.Context, _ EvidenceBatch) (FilterDecision, error) {
	<-b.signal
	return b.decision, b.err
}

// slowEvaluator returns after the given delay.
type slowEvaluator struct {
	id       string
	delay    time.Duration
	decision FilterDecision
	err      error
}

func (s *slowEvaluator) ID() string { return s.id }
func (s *slowEvaluator) Evaluate(_ context.Context, _ EvidenceBatch) (FilterDecision, error) {
	time.Sleep(s.delay)
	return s.decision, s.err
}

// mutationEvaluator captures the batch state at evaluation time and then
// attempts to mutate the batch's internal events slice. Because the test
// file is in the same package it can read EvidenceBatch's unexported fields.
type mutationEvaluator struct {
	id             string
	decision       FilterDecision
	capturedAt     time.Time
	capturedEvents int
}

func (m *mutationEvaluator) ID() string { return m.id }
func (m *mutationEvaluator) Evaluate(_ context.Context, batch EvidenceBatch) (FilterDecision, error) {
	// Capture the batch state atomically so that the test can read it later
	// without a data race. We do NOT mutate batch.events here — the purpose of
	// this test is to verify that the coordinator passes an immutable snapshot,
	// not to exercise mutation.
	m.capturedAt = batch.CapturedAt()
	m.capturedEvents = len(batch.Events())
	return m.decision, nil
}

// panicEvaluator panics on Evaluate to verify that goroutine panic in the
// fan-out does not corrupt the coordinator's lifecycle.
type panicEvaluator struct {
	id string
}

func (p *panicEvaluator) ID() string { return p.id }
func (p *panicEvaluator) Evaluate(_ context.Context, _ EvidenceBatch) (FilterDecision, error) {
	panic("unexpected panic in evaluator")
}

// errorEvaluator always returns a non-nil error.
type errorEvaluator struct {
	id string
}

func (e *errorEvaluator) ID() string { return e.id }
func (e *errorEvaluator) Evaluate(_ context.Context, _ EvidenceBatch) (FilterDecision, error) {
	return FilterDecision{}, errors.New("internal guardrail crash")
}

// deadlineEvaluator returns after a long delay so that the per-filter
// timeout fires.
type deadlineEvaluator struct {
	id string
}

func (d *deadlineEvaluator) ID() string { return d.id }
func (d *deadlineEvaluator) Evaluate(ctx context.Context, _ EvidenceBatch) (FilterDecision, error) {
	select {
	case <-ctx.Done():
		return FilterDecision{}, ctx.Err()
	case <-time.After(10 * time.Second):
		return FilterDecision{}, nil
	}
}

// --- Shared test fixtures ---

var testTS = time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)

func makeTestBatch(t *testing.T) EvidenceBatch {
	t.Helper()
	ev1, err := NewTextDeltaEvent("ch", "hello", testTS)
	if err != nil {
		t.Fatalf("NewTextDeltaEvent 1: %v", err)
	}
	ev2, err := NewTextDeltaEvent("ch", "world", testTS)
	if err != nil {
		t.Fatalf("NewTextDeltaEvent 2: %v", err)
	}
	b, err := NewEvidenceBatch(
		[]NormalizedEvent{ev1, ev2},
		nil, nil, nil, false,
		CommitStateTransportUncommitted, testTS,
	)
	if err != nil {
		t.Fatalf("NewEvidenceBatch: %v", err)
	}
	return b
}

func makeTestPassDecision(t *testing.T, filterID string) FilterDecision {
	t.Helper()
	evidence := buildValidEvidence(t, testTS)
	d, err := NewFilterDecision(FilterDecisionKindPass, "consumer.id", filterID, "rule.id", evidence, nil)
	if err != nil {
		t.Fatalf("NewFilterDecision: %v", err)
	}
	return d
}

func mustMakeEpochFilter(t *testing.T, e FilterEvaluator, subscribed, triggerReady, blocksRelease bool, enforcement FilterEnforcement) EpochFilter {
	t.Helper()
	f, err := NewEpochFilter(e, subscribed, triggerReady, blocksRelease, enforcement, 5*time.Second)
	if err != nil {
		t.Fatalf("NewEpochFilter: %v", err)
	}
	return f
}

func mustMakeShortTimeoutEpochFilter(t *testing.T, e FilterEvaluator, subscribed, triggerReady, blocksRelease bool, enforcement FilterEnforcement) EpochFilter {
	t.Helper()
	f, err := NewEpochFilter(e, subscribed, triggerReady, blocksRelease, enforcement, 200*time.Millisecond)
	if err != nil {
		t.Fatalf("NewEpochFilter: %v", err)
	}
	return f
}

// mustNewCoordinator is a test helper that creates a coordinator and returns
// a cleanup function that calls Close.
func mustNewCoordinator(t *testing.T, arbiter EpochArbiter, maxQueued int) (*GateCoordinator, func()) {
	t.Helper()
	hostCtx := context.Background()
	coord, err := NewGateCoordinator(hostCtx, GateCoordinatorOptions{MaxQueuedEpochs: maxQueued}, arbiter)
	if err != nil {
		t.Fatalf("NewGateCoordinator: %v", err)
	}
	return coord, func() { _ = coord.Close() }
}

// --- [REVIEW_REVIEW_API-1] Parallel fan-out와 all-complete Arbiter barrier ---

// TestGateCoordinatorWaitsForCompleteOutcomeSet verifies that the arbiter is
// NOT invoked until every filter outcome is collected. Two ready evaluators
// block on a shared signal channel; they are signalled after a delay. Until
// both complete, the arbiter call count must be zero. After both complete
// (in reverse order), the arbiter must have been called exactly once with
// all four outcomes (two evaluated, one deferred, one not_applicable).
func TestGateCoordinatorWaitsForCompleteOutcomeSet(t *testing.T) {
	batch := makeTestBatch(t)
	signal := make(chan struct{})

	passA := makeTestPassDecision(t, "beta.filter")
	passB := makeTestPassDecision(t, "alpha.filter")

	// Two ready evaluators that block until signal.
	eA := &blockingEvaluator{id: "beta.filter", signal: signal, decision: passA}
	eB := &blockingEvaluator{id: "alpha.filter", signal: signal, decision: passB}

	// One deferred (subscribed + trigger not ready + blocks release).
	eDeferred := &stubEvaluator{id: "gamma.filter", decision: passA}
	// One not-applicable (unsubscribed).
	eNotApplicable := &stubEvaluator{id: "delta.filter", decision: passA}

	fA := mustMakeEpochFilter(t, eA, true, true, false, FilterEnforcementBlocking)
	fB := mustMakeEpochFilter(t, eB, true, true, false, FilterEnforcementBlocking)
	fDeferred := mustMakeEpochFilter(t, eDeferred, true, false, true, FilterEnforcementBlocking)
	fNotApplicable := mustMakeEpochFilter(t, eNotApplicable, false, false, false, FilterEnforcementBlocking)

	ra := &recordingArbiter{}
	coord, cleanup := mustNewCoordinator(t, ra, 0)
	defer cleanup()

	// Evaluate on its own goroutine so we can inspect arbiter state first.
	done := make(chan struct{})
	var result EvaluationSet
	var resultErr error
	go func() {
		result, resultErr = coord.Evaluate(context.Background(), batch,
			[]EpochFilter{fA, fB, fDeferred, fNotApplicable})
		close(done)
	}()

	// Wait briefly to ensure goroutines have started but not completed.
	time.Sleep(100 * time.Millisecond)

	if ra.calls != 0 {
		t.Errorf("arbiter called %d times before both evaluators complete, want 0", ra.calls)
	}

	// Release both evaluators (they complete in arbitrary order).
	close(signal)

	<-done

	if resultErr != nil {
		t.Fatalf("Evaluate error: %v", resultErr)
	}

	// Arbiter should have been called exactly once.
	if ra.calls != 1 {
		t.Errorf("arbiter calls = %d, want 1", ra.calls)
	}

	// Set must contain all four outcomes.
	if result.Len() != 4 {
		t.Fatalf("set len = %d, want 4", result.Len())
	}

	// Stable ascending order by filter id: alpha < beta < delta < gamma.
	wantOrder := []string{"alpha.filter", "beta.filter", "delta.filter", "gamma.filter"}
	for i, want := range wantOrder {
		o, err := result.At(i)
		if err != nil {
			t.Fatalf("At(%d): %v", i, err)
		}
		if o.FilterID() != want {
			t.Errorf("At(%d).FilterID() = %q, want %q", i, o.FilterID(), want)
		}
	}

	// All evaluated outcomes should have kind "evaluated".
	for i := 0; i < result.Len(); i++ {
		o, _ := result.At(i)
		oc := o.Outcome()
		if oc.Kind() == FilterOutcomeKindEvaluated {
			if oc.Decision() == nil {
				t.Errorf("At(%d) evaluated outcome has nil decision", i)
			}
		}
	}
}

// TestGateCoordinatorPassesSameImmutableBatch verifies that every evaluator
// receives the same immutable batch snapshot: capturedAt matches and the
// batch is not mutated by any evaluator.
func TestGateCoordinatorPassesSameImmutableBatch(t *testing.T) {
	batch := makeTestBatch(t)
	originalEvents := batch.Events()
	originalCapturedAt := batch.CapturedAt()
	originalEventCount := len(originalEvents)

	// Create a pass decision for the filter — the decision's filterID
	// must match the filter's stable id exactly.
	passDecision := makeTestPassDecision(t, "mut.filter")

	// Create evaluators that capture and attempt mutation.
	e1 := &mutationEvaluator{id: "mut.filter", decision: passDecision}
	decision2 := makeTestPassDecision(t, "mut2.filter")
	e2 := &mutationEvaluator{id: "mut2.filter", decision: decision2}

	f1 := mustMakeEpochFilter(t, e1, true, true, false, FilterEnforcementBlocking)
	e2filter, err := NewEpochFilter(e2, true, true, false, FilterEnforcementBlocking, 5*time.Second)
	if err != nil {
		t.Fatalf("NewEpochFilter e2: %v", err)
	}

	ra := &recordingArbiter{}
	coord, cleanup := mustNewCoordinator(t, ra, 0)
	defer cleanup()

	_, err = coord.Evaluate(context.Background(), batch, []EpochFilter{f1, e2filter})
	if err != nil {
		t.Fatalf("Evaluate error: %v", err)
	}

	// Verify both evaluators saw the same capturedAt.
	type snap struct {
		at     time.Time
		events int
	}
	e1Snap := snap{at: e1.capturedAt, events: e1.capturedEvents}
	e2Snap := snap{at: e2.capturedAt, events: e2.capturedEvents}
	if !e1Snap.at.Equal(originalCapturedAt) {
		t.Errorf("e1.capturedAt = %v, want %v", e1Snap.at, originalCapturedAt)
	}
	if !e2Snap.at.Equal(originalCapturedAt) {
		t.Errorf("e2.capturedAt = %v, want %v", e2Snap.at, originalCapturedAt)
	}
	if e1Snap.at != e2Snap.at {
		t.Errorf("e1.capturedAt (%v) != e2.capturedAt (%v)", e1Snap.at, e2Snap.at)
	}

	// Both evaluators saw the same number of events.
	if e1Snap.events != originalEventCount {
		t.Errorf("e1.capturedEvents = %d, want %d", e1Snap.events, originalEventCount)
	}
	if e2Snap.events != originalEventCount {
		t.Errorf("e2.capturedEvents = %d, want %d", e2Snap.events, originalEventCount)
	}

	// The original batch must NOT have been mutated by either evaluator.
	storedEvents := batch.Events()
	if len(storedEvents) != originalEventCount {
		t.Errorf("batch events after Evaluate = %d, want %d", len(storedEvents), originalEventCount)
	}

	// The set's capturedAt should match the original batch capturedAt.
	if ra.calls != 1 {
		t.Fatalf("arbiter calls = %d, want 1", ra.calls)
	}
	capturedInArbiter := ra.batches[0].CapturedAt()
	if !capturedInArbiter.Equal(originalCapturedAt) {
		t.Errorf("arbiter batch.CapturedAt() = %v, want %v", capturedInArbiter, originalCapturedAt)
	}
}

// TestGateCoordinatorRejectsInvalidInput verifies that Evaluate and Submit
// reject invalid batches and that the coordinator constructor rejects
// nil arbiter and negative MaxQueuedEpochs.
func TestGateCoordinatorRejectsInvalidInput(t *testing.T) {
	// --- Nil arbiter ---
	_, err := NewGateCoordinator(context.Background(), GateCoordinatorOptions{MaxQueuedEpochs: 0}, nil)
	if err == nil {
		t.Error("NewGateCoordinator with nil arbiter should return error")
	}

	// --- Negative MaxQueuedEpochs ---
	ra := &recordingArbiter{}
	_, err = NewGateCoordinator(context.Background(), GateCoordinatorOptions{MaxQueuedEpochs: -1}, ra)
	if err == nil {
		t.Error("NewGateCoordinator with negative MaxQueuedEpochs should return error")
	}

	// --- Invalid batch ---
	badBatch := EvidenceBatch{capturedAt: time.Time{}, commitState: CommitStateTransportUncommitted}
	coord, cleanup := mustNewCoordinator(t, ra, 0)
	defer cleanup()

	_, err = coord.Evaluate(context.Background(), badBatch, nil)
	if err == nil {
		t.Error("Evaluate with invalid batch should return error")
	}

	// --- Cancelled context ---
	cancelCtx, cancel := context.WithCancel(context.Background())
	cancel()
	_, err = coord.Evaluate(cancelCtx, makeTestBatch(t), nil)
	if err == nil {
		t.Error("Evaluate with cancelled context should return error")
	}
}

// --- [REVIEW_REVIEW_API-2] Single-flight, bounded backpressure와 cancel/failure policy ---

// TestGateCoordinatorSingleFlightAndBoundedBackpressure verifies that the
// coordinator processes epochs one at a time (single-flight) through a
// bounded queue with FIFO ordering.
func TestGateCoordinatorSingleFlightAndBoundedBackpressure(t *testing.T) {
	const maxQueued = 2
	const totalEpochs = 4

	passDecision := makeTestPassDecision(t, "slow.filter")

	// Count active epochs concurrently.

	// Use a slow evaluator that holds the worker.
	type countingEvaluator struct {
		id       string
		decision FilterDecision
	}
	countEval := &countingEvaluator{id: "sf.filter", decision: passDecision}

	// Override the counting evaluator to track active epochs.
	originalID := countEval.id
	_ = originalID

	// Actually let's use a simpler approach: just measure that we can submit
	// and that max concurrent is 1.

	ra := &recordingArbiter{}
	coord, cleanup := mustNewCoordinator(t, ra, maxQueued)
	defer cleanup()

	// Submit all epochs sequentially. The first one will block the worker.
	// The next two will fill the queue. The fourth will block on send.
	// Use short timeouts to avoid test hanging.
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	// Create a batch.
	batch := makeTestBatch(t)

	// Use a slow evaluator so the first epoch takes time.
	slowEval := &slowEvaluator{
		id:       "slow.filter",
		delay:    200 * time.Millisecond,
		decision: passDecision,
	}
	fSlow := mustMakeEpochFilter(t, slowEval, true, true, false, FilterEnforcementBlocking)

	// Create fast evaluators for subsequent epochs.
	fastPassDecision := makeTestPassDecision(t, "fast.filter")
	fastEval := &stubEvaluator{id: "fast.filter", decision: fastPassDecision}
	fFast := mustMakeEpochFilter(t, fastEval, true, true, false, FilterEnforcementBlocking)

	filters := []EpochFilter{fSlow, fFast, fFast, fFast}

	// Submit all four epochs.
	errCh := make([]chan error, totalEpochs)
	for i := 0; i < totalEpochs; i++ {
		errCh[i] = make(chan error, 1)
		go func(idx int) {
			errCh[idx] <- coord.Submit(ctx, batch, []EpochFilter{filters[idx]})
		}(i)
	}

	// Wait for all submitters to complete.
	for i := 0; i < totalEpochs; i++ {
		select {
		case err := <-errCh[i]:
			if err != nil {
				t.Errorf("Submit[%d] error: %v", i, err)
			}
		case <-time.After(10 * time.Second):
			t.Fatalf("Submit[%d] timed out", i)
		}
	}

	// Arbiter should have been called exactly totalEpochs times.
	if ra.calls != totalEpochs {
		t.Errorf("arbiter calls = %d, want %d", ra.calls, totalEpochs)
	}

	// The epoch batches should arrive in FIFO order.
	for i := 0; i < totalEpochs; i++ {
		// The first epoch was the slow one, the rest are fast.
		// Verify they completed successfully by checking set len.
		if ra.sets[i].Len() < 1 {
			t.Errorf("epoch[%d] set empty", i)
		}
	}
}

// TestGateCoordinatorCancellationSkipsArbiter verifies that caller cancellation
// of queued jobs, active jobs, and host close all skip the arbiter and do not
// increase the arbiter call count.
func TestGateCoordinatorCancellationSkipsArbiter(t *testing.T) {
	const maxQueued = 2

	passDecision := makeTestPassDecision(t, "cancel.filter")
	signal := make(chan struct{})

	// Block the worker with a slow evaluator.
	slowEval := &blockingEvaluator{
		id:       "block.filter",
		signal:   signal,
		decision: passDecision,
	}
	fBlock := mustMakeEpochFilter(t, slowEval, true, true, false, FilterEnforcementBlocking)

	// Fast evaluator for queued epochs.
	fastEval := &stubEvaluator{id: "fast.filter", decision: passDecision}
	fFast := mustMakeEpochFilter(t, fastEval, true, true, false, FilterEnforcementBlocking)

	ra := &recordingArbiter{}
	coord, cleanup := mustNewCoordinator(t, ra, maxQueued)
	defer cleanup()

	batch := makeTestBatch(t)

	// Submit the blocking epoch first to hold the worker.
	blockCtx, blockCancel := context.WithCancel(context.Background())
	blockErr := make(chan error, 1)
	go func() {
		blockErr <- coord.Submit(blockCtx, batch, []EpochFilter{fBlock})
	}()

	// Wait for the blocking epoch to start processing.
	time.Sleep(100 * time.Millisecond)

	// Submit two queued epochs.
	fastCtx1 := context.Background()
	fastErr1 := make(chan error, 1)
	go func() {
		fastErr1 <- coord.Submit(fastCtx1, batch, []EpochFilter{fFast})
	}()

	fastCtx2 := context.Background()
	fastErr2 := make(chan error, 1)
	go func() {
		fastErr2 <- coord.Submit(fastCtx2, batch, []EpochFilter{fFast})
	}()

	// Wait for the queued epochs to be accepted.
	time.Sleep(100 * time.Millisecond)

	// At this point:
	// - The worker is processing the blocking epoch.
	// - Two epochs are in the queue.
	// - The arbiter has not been called (blocking evaluator hasn't returned).

	// Cancel the first queued epoch before it's processed.
	ctx, cancel1 := context.WithCancel(context.Background())
	cancel1()
	err1 := coord.Submit(ctx, batch, []EpochFilter{fFast})
	if !errors.Is(err1, context.Canceled) {
		t.Errorf("Submit with cancelled ctx = %v, want context.Canceled", err1)
	}

	// Cancel the blocking epoch.
	blockCancel()

	// The blocking epoch should complete with a cancel error.
	select {
	case err := <-blockErr:
		if !errors.Is(err, context.Canceled) {
			t.Errorf("blocking epoch error = %v, want context.Canceled", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("blocking epoch did not complete after cancel")
	}

	// The two queued epochs should also complete (with their own errors or success).
	select {
	case err := <-fastErr1:
		_ = err // might succeed or fail depending on timing
	case <-time.After(5 * time.Second):
		t.Fatal("fast epoch 1 did not complete")
	}
	select {
	case err := <-fastErr2:
		_ = err
	case <-time.After(5 * time.Second):
		t.Fatal("fast epoch 2 did not complete")
	}

	// Host close should not panic (idempotent).
	if err := coord.Close(); err != nil {
		t.Errorf("second Close error: %v", err)
	}

	// Arbiter should NOT have been called for the cancelled or host-closed epochs.
	// The blocking epoch was cancelled before completing evaluation, so arbiter
	// should not have been called for it.
	if ra.calls > 0 {
		t.Errorf("arbiter was called %d times during cancellation tests", ra.calls)
	}
}

// TestGateCoordinatorNormalizesDeadlineAndFailurePolicy verifies that evaluator
// errors and per-filter deadlines are normalized to stable outcome codes with
// no raw error payload preserved, and that the failure disposition is
// correctly derived from the enforcement mode.
func TestGateCoordinatorNormalizesDeadlineAndFailurePolicy(t *testing.T) {
	batch := makeTestBatch(t)

	// --- Blocking filter with evaluator error ---
	errEvalBlocking := &errorEvaluator{id: "err.block.filter"}
	fErrBlock := mustMakeEpochFilter(t, errEvalBlocking, true, true, false, FilterEnforcementBlocking)

	// --- Observe-only filter with evaluator error ---
	errEvalObserve := &errorEvaluator{id: "err.obs.filter"}
	fErrObs := mustMakeEpochFilter(t, errEvalObserve, true, true, false, FilterEnforcementObserveOnly)

	// --- Blocking filter that hits deadline ---
	deadlineEvalBlock := &deadlineEvaluator{id: "dl.block.filter"}
	fDLBlock := mustMakeShortTimeoutEpochFilter(t, deadlineEvalBlock, true, true, false, FilterEnforcementBlocking)

	// --- Observe-only filter that hits deadline ---
	deadlineEvalObs := &deadlineEvaluator{id: "dl.obs.filter"}
	fDLObs := mustMakeShortTimeoutEpochFilter(t, deadlineEvalObs, true, true, false, FilterEnforcementObserveOnly)

	// --- A passing filter ---
	passDecision2 := makeTestPassDecision(t, "pass.filter")
	passEval := &stubEvaluator{id: "pass.filter", decision: passDecision2}
	fPass := mustMakeEpochFilter(t, passEval, true, true, false, FilterEnforcementBlocking)

	ra := &recordingArbiter{}
	coord, cleanup := mustNewCoordinator(t, ra, 0)
	defer cleanup()

	set, err := coord.Evaluate(context.Background(), batch,
		[]EpochFilter{fErrBlock, fErrObs, fDLBlock, fDLObs, fPass})
	if err != nil {
		t.Fatalf("Evaluate error: %v", err)
	}

	if set.Len() != 5 {
		t.Fatalf("set len = %d, want 5", set.Len())
	}

	// Verify each outcome.
	for i := 0; i < set.Len(); i++ {
		o, _ := set.At(i)
		fid := o.FilterID()
		oc := o.Outcome()
		disp := o.FailureDisposition()
		enf := o.Enforcement()

		switch fid {
		case "err.block.filter":
			if oc.Kind() != FilterOutcomeKindEvaluationError {
				t.Errorf("[%d] %s kind = %v, want evaluation_error", i, fid, oc.Kind())
			}
			if oc.ErrorCode() != "filter_evaluation_error" {
				t.Errorf("[%d] %s error code = %q, want filter_evaluation_error", i, fid, oc.ErrorCode())
			}
			// Check that the raw error string is NOT in the outcome.
			if oc.ErrorCode() == "internal guardrail crash" {
				t.Errorf("[%d] %s leaked raw error in code", i, fid)
			}
			if enf == FilterEnforcementBlocking {
				if disp != EvaluationFailureDispositionBlockingFatal {
					t.Errorf("[%d] %s disposition = %v, want blocking_fatal", i, fid, disp)
				}
			}

		case "err.obs.filter":
			if oc.Kind() != FilterOutcomeKindEvaluationError {
				t.Errorf("[%d] %s kind = %v, want evaluation_error", i, fid, oc.Kind())
			}
			if enf == FilterEnforcementObserveOnly {
				if disp != EvaluationFailureDispositionObserveError {
					t.Errorf("[%d] %s disposition = %v, want observe_error", i, fid, disp)
				}
			}

		case "dl.block.filter":
			if oc.Kind() != FilterOutcomeKindEvaluationError {
				t.Errorf("[%d] %s kind = %v, want evaluation_error", i, fid, oc.Kind())
			}
			if oc.ErrorCode() != "filter_evaluation_deadline" {
				t.Errorf("[%d] %s error code = %q, want filter_evaluation_deadline", i, fid, oc.ErrorCode())
			}

		case "dl.obs.filter":
			if oc.Kind() != FilterOutcomeKindEvaluationError {
				t.Errorf("[%d] %s kind = %v, want evaluation_error", i, fid, oc.Kind())
			}
			if oc.ErrorCode() != "filter_evaluation_deadline" {
				t.Errorf("[%d] %s error code = %q, want filter_evaluation_deadline", i, fid, oc.ErrorCode())
			}

		case "pass.filter":
			if oc.Kind() != FilterOutcomeKindEvaluated {
				t.Errorf("[%d] %s kind = %v, want evaluated", i, fid, oc.Kind())
			}
		}
	}
}

// TestGateCoordinatorIdempotentClose verifies that Close can be called
// multiple times without panic.
func TestGateCoordinatorIdempotentClose(t *testing.T) {
	ra := &recordingArbiter{}
	coord, _ := mustNewCoordinator(t, ra, 0)

	// First close is fine.
	if err := coord.Close(); err != nil {
		t.Fatalf("first Close: %v", err)
	}

	// Second close must not panic.
	if err := coord.Close(); err != nil {
		t.Fatalf("second Close: %v", err)
	}

	// Third close must not panic.
	if err := coord.Close(); err != nil {
		t.Fatalf("third Close: %v", err)
	}
}

// TestGateCoordinatorSubmitReleaseOnHostClose verifies that when the host
// context is cancelled, blocked submitters are released with an error.
func TestGateCoordinatorSubmitReleaseOnHostClose(t *testing.T) {
	const maxQueued = 0 // Only active epoch allowed.

	passDecision := makeTestPassDecision(t, "host.filter")
	signal := make(chan struct{})

	blockEval := &blockingEvaluator{
		id:       "block.filter",
		signal:   signal,
		decision: passDecision,
	}
	fBlock := mustMakeEpochFilter(t, blockEval, true, true, false, FilterEnforcementBlocking)

	ra := &recordingArbiter{}
	hostCtx, hostCancel := context.WithCancel(context.Background())
	coord, _ := NewGateCoordinator(hostCtx, GateCoordinatorOptions{MaxQueuedEpochs: maxQueued}, ra)

	// Submit a blocking epoch to hold the worker.
	submitDone := make(chan error, 1)
	go func() {
		submitDone <- coord.Submit(context.Background(), makeTestBatch(t), []EpochFilter{fBlock})
	}()

	time.Sleep(100 * time.Millisecond)

	// Cancel the host context — this should release the blocked submitter.
	hostCancel()

	select {
	case err := <-submitDone:
		if err == nil {
			t.Error("submitter should have received an error after host cancel")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("submitter was not released after host cancel")
	}

	// Close must still work after host cancel.
	if err := coord.Close(); err != nil {
		t.Fatalf("Close after host cancel: %v", err)
	}
}

// ensure unused imports are consumed.
var _ = time.Millisecond