package openai

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"reflect"
	"sort"
	"strings"
	"sync"
	"testing"
	"time"

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

	edgeservice "iop/apps/edge/internal/service"
	"iop/packages/go/config"
)

// ---------------------------------------------------------------------------
// API-1: closed enum / projection / observer contract tests
// ---------------------------------------------------------------------------

func TestHotPathObservationSchema_AllEventClassesAreClosed(t *testing.T) {
	cases := []struct {
		name string
		raw  string
		want hotPathEventClass
	}{
		{"dispatch", "dispatch", hotPathEventClassDispatch},
		{"stage", "stage", hotPathEventClassStage},
		{"light", "light", hotPathEventClassLight},
		{"terminal", "terminal", hotPathEventClassTerminal},
		{"cleanup", "cleanup", hotPathEventClassCleanup},
		{"orphan", "orphan", hotPathEventClassOrphan},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			got := hotPathNormalizeEventClass(c.raw)
			if got != c.want {
				t.Errorf("normalizeEventClass(%q) = %q, want %q", c.raw, got, c.want)
			}
			if !hotPathEventClassIsValid(got) {
				t.Errorf("normalizeEventClass(%q) = %q is not valid", c.raw, got)
			}
		})
	}

	// Unknown values normalize to empty and are not valid.
	unknowns := []string{"dispatch_v2", "request", "metric", "foo", "", "CLEANUP", "Stage"}
	for _, u := range unknowns {
		got := hotPathNormalizeEventClass(u)
		if got != "" {
			t.Errorf("normalizeEventClass(%q) = %q, want empty", u, string(got))
		}
		if hotPathEventClassIsValid(got) {
			t.Errorf("normalizeEventClass(%q) = %q is unexpectedly valid", u, got)
		}
	}
}

func TestHotPathObservationSchema_AllModesAreClosed(t *testing.T) {
	cases := []struct {
		name string
		raw  string
		want hotPathMode
	}{
		{"direct", "direct", hotPathModeDirect},
		{"light", "light", hotPathModeLight},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			got := hotPathNormalizeMode(c.raw)
			if got != c.want {
				t.Errorf("normalizeMode(%q) = %q, want %q", c.raw, got, c.want)
			}
			if !hotPathModeIsValid(got) {
				t.Errorf("normalizeMode(%q) = %q is not valid", c.raw, got)
			}
		})
	}

	unknowns := []string{"heavy", "hybrid", "direct_v2", "", "DIRECT", "light_mode"}
	for _, u := range unknowns {
		got := hotPathNormalizeMode(u)
		if got != "" {
			t.Errorf("normalizeMode(%q) = %q, want empty", u, string(got))
		}
		if hotPathModeIsValid(got) {
			t.Errorf("normalizeMode(%q) = %q is unexpectedly valid", u, got)
		}
	}
}

func TestHotPathObservationSchema_AllStageKindsAreClosed(t *testing.T) {
	cases := []struct {
		name string
		raw  string
		want hotPathStageKind
	}{
		{"selector", "selector", hotPathStageKindSelector},
		{"local", "local", hotPathStageKindLocal},
		{"review", "review", hotPathStageKindReview},
		{"cleanup", "cleanup", hotPathStageKindCleanup},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			got := hotPathNormalizeStageKind(c.raw)
			if got != c.want {
				t.Errorf("normalizeStageKind(%q) = %q, want %q", c.raw, got, c.want)
			}
			if !hotPathStageKindIsValid(got) {
				t.Errorf("normalizeStageKind(%q) = %q is not valid", c.raw, got)
			}
		})
	}

	unknowns := []string{"stage", "ingress", "", "SELECTOR", "local_active"}
	for _, u := range unknowns {
		got := hotPathNormalizeStageKind(u)
		if got != "" {
			t.Errorf("normalizeStageKind(%q) = %q, want empty", u, string(got))
		}
		if hotPathStageKindIsValid(got) {
			t.Errorf("normalizeStageKind(%q) = %q is unexpectedly valid", u, got)
		}
	}
}

func TestHotPathObservationSchema_AllDispositionKindsAreClosed(t *testing.T) {
	cases := []struct {
		name string
		raw  string
		want hotPathTerminalDispositionKind
	}{
		{"success", "success", hotPathTerminalDispositionSuccess},
		{"tool_turn", "tool_turn", hotPathTerminalDispositionToolTurn},
		{"length", "length", hotPathTerminalDispositionLength},
		{"provider_error", "provider_error", hotPathTerminalDispositionProviderError},
		{"validation_error", "validation_error", hotPathTerminalDispositionValidationError},
		{"timeout", "timeout", hotPathTerminalDispositionTimeout},
		{"caller_cancel", "caller_cancel", hotPathTerminalDispositionCallerCancel},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			got := hotPathNormalizeDisposition(c.raw)
			if got != c.want {
				t.Errorf("normalizeDisposition(%q) = %q, want %q", c.raw, got, c.want)
			}
			if !hotPathTerminalDispositionIsValid(got) {
				t.Errorf("normalizeDisposition(%q) = %q is not valid", c.raw, got)
			}
		})
	}

	unknowns := []string{"partial_success", "review_pass", "", "SUCCESS", "error"}
	for _, u := range unknowns {
		got := hotPathNormalizeDisposition(u)
		if got != "" {
			t.Errorf("normalizeDisposition(%q) = %q, want empty", u, string(got))
		}
		if hotPathTerminalDispositionIsValid(got) {
			t.Errorf("normalizeDisposition(%q) = %q is unexpectedly valid", u, got)
		}
	}
}

func TestHotPathObservationSchema_AllRouteReasonsAreClosed(t *testing.T) {
	cases := []struct {
		name string
		raw  string
		want hotPathRouteReason
	}{
		{"mode_disabled", "mode_disabled", hotPathRouteReasonModeDisabled},
		{"artifact_required", "artifact_required", hotPathRouteReasonArtifactReq},
		{"invalid_input", "invalid_input", hotPathRouteReasonInvalidInput},
		{"provider_error", "provider_error", hotPathRouteReasonProviderError},
		{"timeout", "timeout", hotPathRouteReasonTimeout},
		{"caller_cancel", "caller_cancel", hotPathRouteReasonCallerCancel},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			got := hotPathNormalizeRouteReason(c.raw)
			if got != c.want {
				t.Errorf("normalizeRouteReason(%q) = %q, want %q", c.raw, got, c.want)
			}
			if !hotPathRouteReasonIsValid(got) {
				t.Errorf("normalizeRouteReason(%q) = %q is not valid", c.raw, got)
			}
		})
	}

	unknowns := []string{"internal_error", "rate_limit", "", "MODE_DISABLED", "error"}
	for _, u := range unknowns {
		got := hotPathNormalizeRouteReason(u)
		if got != "" {
			t.Errorf("normalizeRouteReason(%q) = %q, want empty", u, string(got))
		}
		if hotPathRouteReasonIsValid(got) {
			t.Errorf("normalizeRouteReason(%q) = %q is unexpectedly valid", u, got)
		}
	}
}

func TestHotPathObservationSchema_AllCleanupOutcomesAreClosed(t *testing.T) {
	cases := []struct {
		name string
		raw  string
		want hotPathCleanupOutcome
	}{
		{"success", "success", hotPathCleanupOutcomeSuccess},
		{"primary_error", "primary_error", hotPathCleanupOutcomePrimaryError},
		{"ttl_expired", "ttl_expired", hotPathCleanupOutcomeTTLExpired},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			got := hotPathNormalizeCleanupOutcome(c.raw)
			if got != c.want {
				t.Errorf("normalizeCleanupOutcome(%q) = %q, want %q", c.raw, got, c.want)
			}
			if !hotPathCleanupOutcomeIsValid(got) {
				t.Errorf("normalizeCleanupOutcome(%q) = %q is not valid", c.raw, got)
			}
		})
	}

	unknowns := []string{"partial", "", "SUCCESS", "cleanup_failed"}
	for _, u := range unknowns {
		got := hotPathNormalizeCleanupOutcome(u)
		if got != "" {
			t.Errorf("normalizeCleanupOutcome(%q) = %q, want empty", u, string(got))
		}
		if hotPathCleanupOutcomeIsValid(got) {
			t.Errorf("normalizeCleanupOutcome(%q) = %q is unexpectedly valid", u, got)
		}
	}
}

func TestHotPathObservationSchema_AllOrphanOutcomesAreClosed(t *testing.T) {
	cases := []struct {
		name string
		raw  string
		want hotPathOrphanOutcome
	}{
		{"ttl_expired", "ttl_expired", hotPathOrphanOutcomeTTLExpired},
		{"cleanup_failed", "cleanup_failed", hotPathOrphanOutcomeCleanupFailed},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			got := hotPathNormalizeOrphanOutcome(c.raw)
			if got != c.want {
				t.Errorf("normalizeOrphanOutcome(%q) = %q, want %q", c.raw, got, c.want)
			}
			if !hotPathOrphanOutcomeIsValid(got) {
				t.Errorf("normalizeOrphanOutcome(%q) = %q is not valid", c.raw, got)
			}
		})
	}

	unknowns := []string{"success", "", "TTL_EXPIRED", "orphan_removed"}
	for _, u := range unknowns {
		got := hotPathNormalizeOrphanOutcome(u)
		if got != "" {
			t.Errorf("normalizeOrphanOutcome(%q) = %q, want empty", u, string(got))
		}
		if hotPathOrphanOutcomeIsValid(got) {
			t.Errorf("normalizeOrphanOutcome(%q) = %q is unexpectedly valid", u, got)
		}
	}
}

func TestHotPathObservationSchema_AllAttemptBucketsAreClosed(t *testing.T) {
	cases := []struct {
		name string
		raw  string
		want hotPathAttemptBucket
	}{
		{"first", "first", hotPathAttemptFirst},
		{"retry", "retry", hotPathAttemptRetry},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			got := hotPathNormalizeAttemptBucket(c.raw)
			if got != c.want {
				t.Errorf("normalizeAttemptBucket(%q) = %q, want %q", c.raw, got, c.want)
			}
			if !hotPathAttemptBucketIsValid(got) {
				t.Errorf("normalizeAttemptBucket(%q) = %q is not valid", c.raw, got)
			}
		})
	}

	unknowns := []string{"third", "last", "", "FIRST", "attempt_1"}
	for _, u := range unknowns {
		got := hotPathNormalizeAttemptBucket(u)
		if got != "" {
			t.Errorf("normalizeAttemptBucket(%q) = %q, want empty", u, string(got))
		}
		if hotPathAttemptBucketIsValid(got) {
			t.Errorf("normalizeAttemptBucket(%q) = %q is unexpectedly valid", u, got)
		}
	}
}

func TestHotPathObservationSchema_LogProjectionKeysAreExact(t *testing.T) {
	keys := logProjectionKeys()
	want := []string{
		"hot_path_event_class",
		"hot_path_mode",
		"hot_path_stage_kind",
		"hot_path_disposition",
		"hot_path_correlation",
		"hot_path_stage_id",
		"hot_path_request_id",
		"hot_path_call_id",
		"hot_path_owner_edge_id",
		"hot_path_reason",
		"hot_path_preset_id",
		"hot_path_attempt_bucket",
		"hot_path_cleanup_outcome",
		"hot_path_orphan_outcome",
	}
	if len(keys) != len(want) {
		t.Fatalf("logProjectionKeys() length = %d, want %d", len(keys), len(want))
	}
	for i := range keys {
		if keys[i] != want[i] {
			t.Errorf("logProjectionKeys()[%d] = %q, want %q", i, keys[i], want[i])
		}
	}

	allowlist := logProjectionAllowlist()
	if len(allowlist) != len(want) {
		t.Errorf("logProjectionAllowlist() size = %d, want %d", len(allowlist), len(want))
	}
	for _, k := range want {
		if _, ok := allowlist[k]; !ok {
			t.Errorf("logProjectionAllowlist() missing key %q", k)
		}
	}
}

func TestHotPathObservationSchema_LogProjectionRejectsNonAllowlistedKeys(t *testing.T) {
	allowlist := logProjectionAllowlist()

	// Every key in the allowlist should be present.
	for k := range allowlist {
		if !hotPathLogProjectionKeyAllowed(k) {
			t.Errorf("allowlisted key %q is not reported as allowed", k)
		}
	}

	// Every known raw field category should be rejected.
	forbidden := []string{
		"prompt", "output", "tool_args", "tool_result",
		"authorization", "preparer_input", "preparer_output",
		"error_text", "raw_body", "content", "reasoning",
		"request_id", "stage_id", "attempt_id", "run_id",
		"provider_id", "node_id", "session_id",
		"header", "bearer_token", "api_key",
	}
	for _, f := range forbidden {
		if hotPathLogProjectionKeyAllowed(f) {
			t.Errorf("forbidden key %q is unexpectedly allowed", f)
		}
		if _, ok := allowlist[f]; ok {
			t.Errorf("forbidden key %q is in the allowlist map", f)
		}
	}
}

// hotPathLogProjectionKeyAllowed reports whether a key is in the log projection
// allowlist. Exported for tests.
func hotPathLogProjectionKeyAllowed(key string) bool {
	allowlist := logProjectionAllowlist()
	_, ok := allowlist[key]
	return ok
}

// ---------------------------------------------------------------------------
// API-2: correlation id, rejection, observer failure isolation tests
// ---------------------------------------------------------------------------

func TestHotPathObservationRejectsRawValues_EventClass(t *testing.T) {
	raws := []string{
		"dispatch_v2",
		"request",
		"metric",
		"foo",
		"CLEANUP",
		"stage/with/slashes",
		"\x00control",
	}
	for _, r := range raws {
		got := hotPathNormalizeEventClass(r)
		if got != "" {
			t.Errorf("normalizeEventClass(%q) = %q, want empty (raw rejected)", r, string(got))
		}
	}
}

func TestHotPathObservationRejectsRawValues_Mode(t *testing.T) {
	raws := []string{
		"heavy",
		"hybrid",
		"direct_v2",
		"DIRECT",
		"light_mode",
		"light/with/slash",
	}
	for _, r := range raws {
		got := hotPathNormalizeMode(r)
		if got != "" {
			t.Errorf("normalizeMode(%q) = %q, want empty (raw rejected)", r, string(got))
		}
	}
}

func TestHotPathObservationRejectsRawValues_Disposition(t *testing.T) {
	raws := []string{
		"partial_success",
		"review_pass",
		"SUCCESS",
		"error",
		"provider_error/extra",
	}
	for _, r := range raws {
		got := hotPathNormalizeDisposition(r)
		if got != "" {
			t.Errorf("normalizeDisposition(%q) = %q, want empty (raw rejected)", r, string(got))
		}
	}
}

func TestHotPathObservationRejectsRawValues_RouteReason(t *testing.T) {
	raws := []string{
		"internal_error",
		"rate_limit",
		"MODE_DISABLED",
		"error",
	}
	for _, r := range raws {
		got := hotPathNormalizeRouteReason(r)
		if got != "" {
			t.Errorf("normalizeRouteReason(%q) = %q, want empty (raw rejected)", r, string(got))
		}
	}
}

func TestHotPathObservationRejectsRawValues_CleanupOutcome(t *testing.T) {
	raws := []string{
		"partial",
		"SUCCESS",
		"cleanup_failed",
	}
	for _, r := range raws {
		got := hotPathNormalizeCleanupOutcome(r)
		if got != "" {
			t.Errorf("normalizeCleanupOutcome(%q) = %q, want empty (raw rejected)", r, string(got))
		}
	}
}

func TestHotPathObservationRejectsRawValues_OrphanOutcome(t *testing.T) {
	raws := []string{
		"success",
		"TTL_EXPIRED",
		"orphan_removed",
	}
	for _, r := range raws {
		got := hotPathNormalizeOrphanOutcome(r)
		if got != "" {
			t.Errorf("normalizeOrphanOutcome(%q) = %q, want empty (raw rejected)", r, string(got))
		}
	}
}

func TestHotPathObservationRejectsRawValues_StageKind(t *testing.T) {
	raws := []string{
		"stage",
		"ingress",
		"SELECTOR",
		"local_active",
	}
	for _, r := range raws {
		got := hotPathNormalizeStageKind(r)
		if got != "" {
			t.Errorf("normalizeStageKind(%q) = %q, want empty (raw rejected)", r, string(got))
		}
	}
}

func TestHotPathObservationRejectsRawValues_AttemptBucket(t *testing.T) {
	raws := []string{
		"third",
		"last",
		"FIRST",
		"attempt_1",
	}
	for _, r := range raws {
		got := hotPathNormalizeAttemptBucket(r)
		if got != "" {
			t.Errorf("normalizeAttemptBucket(%q) = %q, want empty (raw rejected)", r, string(got))
		}
	}
}

func TestHotPathObservationCorrelationID_BoundsAndSafety(t *testing.T) {
	// Empty segments produce empty id.
	id := newHotPathCorrelationID("", "", "")
	if id != "" {
		t.Errorf("empty segments produced non-empty id: %q", id)
	}

	// Single segment works.
	id = newHotPathCorrelationID("req-1", "", "")
	if string(id) != "hot_path.req.req-1" {
		t.Errorf("single segment id = %q", id)
	}

	// Full correlation is joined with colon.
	id = newHotPathCorrelationID("req-1", "stage-2", "call-3")
	want := "hot_path.req.req-1:hot_path.stage.stage-2:hot_path.call.call-3"
	if string(id) != want {
		t.Errorf("full correlation id = %q, want %q", id, want)
	}

	// Spaces and slashes are sanitized.
	id = newHotPathCorrelationID("req with spaces", "stage/with/slash", "call\twith\ttabs")
	s := string(id)
	if strings.Contains(s, " ") {
		t.Errorf("correlation id contains space: %q", s)
	}
	if strings.Contains(s, "/") {
		t.Errorf("correlation id contains slash: %q", s)
	}
	if strings.Contains(s, "\t") {
		t.Errorf("correlation id contains tab: %q", s)
	}

	// Bounded to 64 runes per segment.
	long := strings.Repeat("X", 300)
	id = newHotPathCorrelationID(long, "", "")
	if len(id) > 64 {
		t.Errorf("correlation id length = %d, want <= 64", len(id))
	}

	// Control characters are sanitized.
	id = newHotPathCorrelationID("req\x00ctrl", "", "")
	if strings.Contains(string(id), "\x00") {
		t.Errorf("correlation id contains control char: %q", id)
	}
}

func TestHotPathObservationNoopObserver_EmitsNilError(t *testing.T) {
	obs := hotPathNoopObserver{}
	ctx := context.Background()
	proj := hotPathLogProjection{
		EventClass:  hotPathEventClassDispatch,
		Mode:        hotPathModeDirect,
		StageKind:   hotPathStageKindSelector,
		Disposition: hotPathTerminalDispositionSuccess,
		Correlation: "corr-1",
		RequestID:   "req-1",
		StageID:     "stage-1",
		CallID:      "call-1",
		OwnerEdgeID: "edge-1",
		Reason:      hotPathRouteReasonModeDisabled,
	}
	if err := obs.Emit(ctx, proj); err != nil {
		t.Errorf("noop observer Emit error = %v, want nil", err)
	}
}

func TestHotPathObservationBoundedObserver_DelegatesToInner(t *testing.T) {
	called := false
	inner := &fakeHotPathObserver{
		emitFn: func(ctx context.Context, p hotPathLogProjection) error {
			called = true
			if p.EventClass != hotPathEventClassDispatch {
				t.Errorf("inner received wrong event class: %q", p.EventClass)
			}
			return nil
		},
	}
	obs := &hotPathBoundedObserver{inner: inner}
	ctx := context.Background()
	proj := hotPathLogProjection{EventClass: hotPathEventClassDispatch}
	if err := obs.Emit(ctx, proj); err != nil {
		t.Errorf("bounded observer Emit error = %v, want nil", err)
	}
	if !called {
		t.Errorf("inner observer was not called")
	}
}

func TestHotPathObservationBoundedObserver_NilInnerIsNoop(t *testing.T) {
	obs := &hotPathBoundedObserver{}
	ctx := context.Background()
	proj := hotPathLogProjection{EventClass: hotPathEventClassDispatch}
	if err := obs.Emit(ctx, proj); err != nil {
		t.Errorf("nil-inner bounded observer Emit error = %v, want nil", err)
	}
}

func TestHotPathObservationSafeObserver_IgnoresInnerError(t *testing.T) {
	expectedErr := errors.New("inner observer failure")
	inner := &fakeHotPathObserver{
		emitErr: expectedErr,
	}
	var hookCalled bool
	var hookProj hotPathLogProjection
	var hookErr error
	hook := func(p hotPathLogProjection, err error) {
		hookCalled = true
		hookProj = p
		hookErr = err
	}
	safe := &hotPathSafeObserver{inner: inner, onFailure: hook}
	ctx := context.Background()
	proj := hotPathLogProjection{EventClass: hotPathEventClassTerminal}

	// Emit returns nil even though inner returned an error.
	if err := safe.Emit(ctx, proj); err != nil {
		t.Errorf("safe observer Emit error = %v, want nil", err)
	}

	if !hookCalled {
		t.Errorf("failure hook was not called")
	}
	if !errors.Is(hookErr, expectedErr) {
		t.Errorf("hook error = %v, want %v", hookErr, expectedErr)
	}
	if hookProj.EventClass != hotPathEventClassTerminal {
		t.Errorf("hook received wrong projection: %v", hookProj)
	}

	if safe.failureCount() != 1 {
		t.Errorf("failure count = %d, want 1", safe.failureCount())
	}
}

func TestHotPathObservationSafeObserver_IgnoresInnerPanic(t *testing.T) {
	inner := &fakeHotPathObserver{
		emitFn: func(ctx context.Context, p hotPathLogProjection) error {
			panic("observer boom")
		},
	}
	var hookCalled bool
	var hookErr error
	hook := func(p hotPathLogProjection, err error) {
		hookCalled = true
		hookErr = err
	}
	safe := &hotPathSafeObserver{inner: inner, onFailure: hook}
	ctx := context.Background()
	proj := hotPathLogProjection{EventClass: hotPathEventClassCleanup}

	// Emit returns nil even though inner panicked.
	if err := safe.Emit(ctx, proj); err != nil {
		t.Errorf("safe observer Emit error = %v, want nil (panic isolated)", err)
	}

	if !hookCalled {
		t.Errorf("failure hook was not called on panic")
	}
	if hookErr == nil {
		t.Errorf("hook error is nil on panic")
	}
	if !strings.Contains(hookErr.Error(), "observer panic") {
		t.Errorf("hook error message = %q, want to contain 'observer panic'", hookErr.Error())
	}

	if safe.failureCount() != 1 {
		t.Errorf("failure count = %d, want 1", safe.failureCount())
	}
}

func TestHotPathObservationSafeObserver_NilObserverIsNoop(t *testing.T) {
	var safe *hotPathSafeObserver
	ctx := context.Background()
	proj := hotPathLogProjection{EventClass: hotPathEventClassDispatch}
	if err := safe.Emit(ctx, proj); err != nil {
		t.Errorf("nil safe observer Emit error = %v, want nil", err)
	}
	if safe.failureCount() != 0 {
		t.Errorf("nil safe observer failure count = %d, want 0", safe.failureCount())
	}
}

func TestHotPathObservationSafeObserver_MultipleFailuresCounted(t *testing.T) {
	inner := &fakeHotPathObserver{emitErr: errors.New("fail")}
	safe := &hotPathSafeObserver{inner: inner}
	ctx := context.Background()
	proj := hotPathLogProjection{EventClass: hotPathEventClassDispatch}

	for i := 0; i < 5; i++ {
		if err := safe.Emit(ctx, proj); err != nil {
			t.Errorf("emit %d: unexpected error = %v", i, err)
		}
	}
	if safe.failureCount() != 5 {
		t.Errorf("failure count = %d, want 5", safe.failureCount())
	}
}

func TestHotPathObservationSafeObserver_SuccessDoesNotIncrement(t *testing.T) {
	inner := &fakeHotPathObserver{}
	safe := &hotPathSafeObserver{inner: inner}
	ctx := context.Background()
	proj := hotPathLogProjection{EventClass: hotPathEventClassDispatch}

	if err := safe.Emit(ctx, proj); err != nil {
		t.Errorf("emit: unexpected error = %v", err)
	}
	if safe.failureCount() != 0 {
		t.Errorf("failure count = %d, want 0 after success", safe.failureCount())
	}
}

func TestHotPathObservationSafeObserver_ConcurrentSafety(t *testing.T) {
	inner := &fakeHotPathObserver{emitErr: errors.New("fail")}
	safe := &hotPathSafeObserver{inner: inner}
	ctx := context.Background()
	proj := hotPathLogProjection{EventClass: hotPathEventClassDispatch}

	var wg sync.WaitGroup
	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = safe.Emit(ctx, proj)
		}()
	}
	wg.Wait()

	if safe.failureCount() != 100 {
		t.Errorf("concurrent failure count = %d, want 100", safe.failureCount())
	}
}

// fakeHotPathObserver is a test double for hotPathObserver.
type fakeHotPathObserver struct {
	mu      sync.Mutex
	emitFn  func(ctx context.Context, p hotPathLogProjection) error
	emitErr error
	calls   int
}

func (f *fakeHotPathObserver) Emit(ctx context.Context, p hotPathLogProjection) error {
	f.mu.Lock()
	f.calls++
	fn := f.emitFn
	err := f.emitErr
	f.mu.Unlock()
	if fn != nil {
		return fn(ctx, p)
	}
	return err
}

// ---------------------------------------------------------------------------
// Metric label allowlist / cardinality tests
// ---------------------------------------------------------------------------

func TestHotPathMetricLabels_FixedLabelNames(t *testing.T) {
	names := hotPathMetricLabelNamesSnapshot()
	want := []string{
		"edge_id",
		"hot_path_event_class",
		"hot_path_mode",
		"hot_path_stage_kind",
		"hot_path_disposition",
		"hot_path_duration_bucket",
		"hot_path_usage_bucket",
		"hot_path_attempt_bucket",
		"hot_path_reason",
		"hot_path_cleanup_outcome",
		"hot_path_orphan_outcome",
	}
	if len(names) != len(want) {
		t.Fatalf("metric label names count = %d, want %d", len(names), len(want))
	}
	for i := range names {
		if names[i] != want[i] {
			t.Errorf("metric label names[%d] = %q, want %q", i, names[i], want[i])
		}
	}
}

func TestHotPathMetricLabels_NoHighCardinalityNames(t *testing.T) {
	names := hotPathMetricLabelNamesSnapshot()
	forbidden := []string{
		"request_id", "stage_id", "attempt_id", "run_id",
		"provider_id", "node_id", "session_id", "correlation_id",
		"content", "reasoning", "tool_args", "tool_result",
		"authorization", "bearer_token", "api_key",
		"error_text", "raw_body", "header",
	}
	for _, f := range forbidden {
		for _, n := range names {
			if n == f {
				t.Errorf("metric label %q is high-cardinality and should not be present", f)
			}
		}
	}
}

func TestHotPathMetricLabels_CardinalityBudget(t *testing.T) {
	card := hotPathMetricLabelCardinalitySnapshot()
	if len(card) != len(hotPathMetricLabelNamesSnapshot()) {
		t.Errorf("cardinality map size = %d, want %d", len(card), len(hotPathMetricLabelNamesSnapshot()))
	}
	total := hotPathMetricLabelCardinalityTotal()
	if total > hotPathMetricLabelCardinalityBudget {
		t.Errorf("cardinality total = %d exceeds budget %d", total, hotPathMetricLabelCardinalityBudget)
	}
}

func TestHotPathMetricLabels_DurationBucketNormalization(t *testing.T) {
	cases := []struct {
		name string
		raw  string
		want hotPathDurationBucket
	}{
		{"sub_ms", "sub_ms", hotPathDurationSubMS},
		{"1_to_10ms", "1_to_10ms", hotPathDuration1to10MS},
		{"10_to_100ms", "10_to_100ms", hotPathDuration10to100MS},
		{"100ms_to_1s", "100ms_to_1s", hotPathDuration100to1S},
		{"1_to_10s", "1_to_10s", hotPathDuration1to10S},
		{"10_to_60s", "10_to_60s", hotPathDuration10to60S},
		{"over_60s", "over_60s", hotPathDurationOver60S},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			got := hotPathNormalizeDurationBucket(c.raw)
			if got != c.want {
				t.Errorf("normalizeDurationBucket(%q) = %q, want %q", c.raw, got, c.want)
			}
			if !hotPathDurationBucketIsValid(got) {
				t.Errorf("normalizeDurationBucket(%q) = %q is not valid", c.raw, got)
			}
		})
	}

	unknowns := []string{"fast", "slow", "", "SUB_MS", "1ms", "100us"}
	for _, u := range unknowns {
		got := hotPathNormalizeDurationBucket(u)
		if got != "" {
			t.Errorf("normalizeDurationBucket(%q) = %q, want empty", u, string(got))
		}
	}
}

func TestHotPathMetricLabels_UsageBucketNormalization(t *testing.T) {
	cases := []struct {
		name string
		raw  string
		want hotPathUsageBucket
	}{
		{"prompt", "prompt", hotPathUsagePrompt},
		{"completion", "completion", hotPathUsageCompletion},
		{"reasoning", "reasoning", hotPathUsageReasoning},
		{"cached_input", "cached_input", hotPathUsageCachedInput},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			got := hotPathNormalizeUsageBucket(c.raw)
			if got != c.want {
				t.Errorf("normalizeUsageBucket(%q) = %q, want %q", c.raw, got, c.want)
			}
			if !hotPathUsageBucketIsValid(got) {
				t.Errorf("normalizeUsageBucket(%q) = %q is not valid", c.raw, got)
			}
		})
	}

	unknowns := []string{"total", "", "PROMPT", "input_tokens"}
	for _, u := range unknowns {
		got := hotPathNormalizeUsageBucket(u)
		if got != "" {
			t.Errorf("normalizeUsageBucket(%q) = %q, want empty", u, string(got))
		}
	}
}

func TestHotPathMetricLabels_DurationBucketFromSeconds(t *testing.T) {
	cases := []struct {
		name     string
		seconds  float64
		expected hotPathDurationBucket
	}{
		{"sub_ms", 0.0005, hotPathDurationSubMS},
		{"1_to_10ms", 0.005, hotPathDuration1to10MS},
		{"10_to_100ms", 0.05, hotPathDuration10to100MS},
		{"100ms_to_1s", 0.5, hotPathDuration100to1S},
		{"1_to_10s", 5.0, hotPathDuration1to10S},
		{"10_to_60s", 30.0, hotPathDuration10to60S},
		{"over_60s", 120.0, hotPathDurationOver60S},
		{"boundary_1ms", 0.001, hotPathDuration1to10MS},
		{"boundary_10ms", 0.01, hotPathDuration10to100MS},
		{"boundary_100ms", 0.1, hotPathDuration100to1S},
		{"boundary_1s", 1.0, hotPathDuration1to10S},
		{"boundary_10s", 10.0, hotPathDuration10to60S},
		{"boundary_60s", 60.0, hotPathDurationOver60S},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			got := hotPathDurationBucketFromSeconds(c.seconds)
			if got != c.expected {
				t.Errorf("durationBucketFromSeconds(%v) = %q, want %q", c.seconds, got, c.expected)
			}
		})
	}
}

func TestHotPathMetricLabels_MetricsInitializeOnce(t *testing.T) {
	m1 := initHotPathMetrics()
	m2 := initHotPathMetrics()
	if m1 != m2 {
		t.Errorf("initHotPathMetrics() returned different instances")
	}
	if m1 == nil {
		t.Errorf("initHotPathMetrics() returned nil")
	}
}

func TestHotPathMetricLabels_RecordFunctionsDoNotPanic(t *testing.T) {
	m := initHotPathMetrics()
	ctx := context.Background()
	_ = ctx

	// Every record function should be callable without panic.
	m.recordStageDuration("edge-1", hotPathModeDirect, hotPathStageKindSelector, hotPathAttemptFirst, 0.05)
	m.recordTerminal("edge-1", hotPathModeDirect, hotPathTerminalDispositionSuccess)
	m.recordUsage("edge-1", hotPathModeDirect, hotPathUsagePrompt, 100)
	m.recordUsage("edge-1", hotPathModeDirect, hotPathUsageCompletion, 50)
	m.recordUsage("edge-1", hotPathModeDirect, hotPathUsageReasoning, 0) // zero count is skipped
	m.recordDispatch("edge-1", hotPathModeLight, hotPathRouteReasonModeDisabled)
	m.recordCleanup("edge-1", hotPathCleanupOutcomeSuccess)
	m.recordOrphan("edge-1", hotPathOrphanOutcomeTTLExpired)
	m.recordObserverFailure("edge-1")

	// Nil metrics should also be safe.
	var nilM *hotPathMetrics
	nilM.recordStageDuration("edge-1", hotPathModeDirect, hotPathStageKindSelector, hotPathAttemptFirst, 0.05)
	nilM.recordTerminal("edge-1", hotPathModeDirect, hotPathTerminalDispositionSuccess)
	nilM.recordUsage("edge-1", hotPathModeDirect, hotPathUsagePrompt, 100)
	nilM.recordDispatch("edge-1", hotPathModeLight, hotPathRouteReasonModeDisabled)
	nilM.recordCleanup("edge-1", hotPathCleanupOutcomeSuccess)
	nilM.recordOrphan("edge-1", hotPathOrphanOutcomeTTLExpired)
	nilM.recordObserverFailure("edge-1")
}

// ---------------------------------------------------------------------------
// Observer failure isolation end-to-end
// ---------------------------------------------------------------------------

func TestHotPathObserverFailureIsolation_EndToEnd(t *testing.T) {
	// Build a chain: safe -> bounded -> failing inner.
	failingInner := &fakeHotPathObserver{
		emitFn: func(ctx context.Context, p hotPathLogProjection) error {
			return fmt.Errorf("failing inner observer")
		},
	}
	bounded := &hotPathBoundedObserver{inner: failingInner}

	var failures []error
	var mu sync.Mutex
	hook := func(p hotPathLogProjection, err error) {
		mu.Lock()
		failures = append(failures, err)
		mu.Unlock()
	}
	safe := &hotPathSafeObserver{inner: bounded, onFailure: hook}

	ctx := context.Background()
	proj := hotPathLogProjection{
		EventClass:  hotPathEventClassDispatch,
		Mode:        hotPathModeDirect,
		Disposition: hotPathTerminalDispositionSuccess,
	}

	// Emit should not propagate the error.
	if err := safe.Emit(ctx, proj); err != nil {
		t.Errorf("safe.Emit error = %v, want nil (failure isolated)", err)
	}

	mu.Lock()
	if len(failures) != 1 {
		t.Errorf("hook called %d times, want 1", len(failures))
	}
	if len(failures) > 0 && !strings.Contains(failures[0].Error(), "failing inner observer") {
		t.Errorf("hook error = %v, want to contain 'failing inner observer'", failures[0])
	}
	mu.Unlock()

	if safe.failureCount() != 1 {
		t.Errorf("failure count = %d, want 1", safe.failureCount())
	}
}

func TestHotPathObserverFailureIsolation_PanicIsolation(t *testing.T) {
	panicInner := &fakeHotPathObserver{
		emitFn: func(ctx context.Context, p hotPathLogProjection) error {
			panic("observer panic in production")
		},
	}
	bounded := &hotPathBoundedObserver{inner: panicInner}

	var panicErr error
	hook := func(p hotPathLogProjection, err error) {
		panicErr = err
	}
	safe := &hotPathSafeObserver{inner: bounded, onFailure: hook}

	ctx := context.Background()
	proj := hotPathLogProjection{EventClass: hotPathEventClassCleanup}

	// Emit should not propagate the panic.
	if err := safe.Emit(ctx, proj); err != nil {
		t.Errorf("safe.Emit error = %v, want nil (panic isolated)", err)
	}

	if panicErr == nil {
		t.Errorf("hook was not called on panic")
	}
	if panicErr != nil && !strings.Contains(panicErr.Error(), "observer panic") {
		t.Errorf("hook error = %v, want to contain 'observer panic'", panicErr)
	}
}

// ---------------------------------------------------------------------------
// Server seam tests
// ---------------------------------------------------------------------------

func TestHotPathObserver_ServerDefaultIsZap(t *testing.T) {
	s := newTestServer(t)
	obs := s.HotPathObserver()
	if obs == nil {
		t.Fatal("HotPathObserver() returned nil")
	}
	if _, ok := obs.(*zapHotPathObserver); !ok {
		t.Fatalf("default observer type=%T, want *zapHotPathObserver", obs)
	}
	ctx := context.Background()
	proj := hotPathLogProjection{EventClass: hotPathEventClassDispatch}
	if err := obs.Emit(ctx, proj); err != nil {
		t.Errorf("default observer Emit error = %v, want nil", err)
	}
}

func TestHotPathObserver_ServerSetAndRetrieve(t *testing.T) {
	s := newTestServer(t)

	called := false
	inner := &fakeHotPathObserver{
		emitFn: func(ctx context.Context, p hotPathLogProjection) error {
			called = true
			return nil
		},
	}
	s.SetHotPathObserver(inner)

	obs := s.HotPathObserver()
	if obs != inner {
		t.Errorf("HotPathObserver() did not return the installed observer: got %T, want %T", obs, inner)
	}

	ctx := context.Background()
	proj := hotPathLogProjection{EventClass: hotPathEventClassDispatch}
	_ = obs.Emit(ctx, proj)
	if !called {
		t.Errorf("installed observer was not called")
	}
}

func TestHotPathObserver_ServerSetNilInstallsNoop(t *testing.T) {
	s := newTestServer(t)

	inner := &fakeHotPathObserver{}
	s.SetHotPathObserver(inner)
	s.SetHotPathObserver(nil)

	obs := s.HotPathObserver()
	if _, ok := obs.(hotPathNoopObserver); !ok {
		t.Errorf("SetHotPathObserver(nil) did not install noop observer, got %T", obs)
	}
}

func TestHotPathObserver_ServerPreservesObsSink(t *testing.T) {
	s := newTestServer(t)
	// obsSink should still be the default zap filter sink, not affected by
	// hot path observer changes.
	if s.obsSink == nil {
		t.Errorf("obsSink was nil after construction, expected default sink")
	}
}

// newTestServer constructs a minimal Server for observer seam tests.
func newTestServer(t *testing.T) *Server {
	t.Helper()
	return NewServer(
		defaultTestEdgeOpenAIConf(),
		nil,
		nil,
	)
}

// defaultTestEdgeOpenAIConf returns a minimal config for server construction.
func defaultTestEdgeOpenAIConf() config.EdgeOpenAIConf {
	return config.EdgeOpenAIConf{Enabled: false}
}

// ---------------------------------------------------------------------------
// Focused Boundary & Production Seam Tests (REVIEW_API-1 & REVIEW_API-2)
// ---------------------------------------------------------------------------

func TestHotPathObservationProjectionBoundary(t *testing.T) {
	t.Run("valid projection passes to inner sink with exact allowlisted fields", func(t *testing.T) {
		var captured hotPathLogProjection
		called := false
		inner := &fakeHotPathObserver{
			emitFn: func(ctx context.Context, p hotPathLogProjection) error {
				called = true
				captured = p
				return nil
			},
		}
		obs := &hotPathBoundedObserver{inner: inner}
		proj := hotPathLogProjection{
			EventClass:     hotPathEventClassDispatch,
			Mode:           hotPathModeDirect,
			StageKind:      hotPathStageKindSelector,
			Disposition:    hotPathTerminalDispositionSuccess,
			RequestID:      "req-123",
			StageID:        "stage-456",
			CallID:         "call-789",
			OwnerEdgeID:    "edge-1",
			Reason:         hotPathRouteReasonModeDisabled,
			PresetID:       "preset-standard",
			AttemptBucket:  hotPathAttemptFirst,
			CleanupOutcome: hotPathCleanupOutcomeSuccess,
			OrphanOutcome:  hotPathOrphanOutcomeTTLExpired,
		}
		if err := obs.Emit(context.Background(), proj); err != nil {
			t.Fatalf("Emit error = %v, want nil", err)
		}
		if !called {
			t.Fatalf("inner sink was not called for valid projection")
		}
		if captured.EventClass != hotPathEventClassDispatch || captured.Mode != hotPathModeDirect {
			t.Errorf("captured projection mismatch: %+v", captured)
		}
		if captured.Correlation == "" {
			t.Errorf("expected correlation id to be generated, got empty")
		}
	})

	t.Run("invalid enum values produce no sink emission", func(t *testing.T) {
		called := false
		inner := &fakeHotPathObserver{
			emitFn: func(ctx context.Context, p hotPathLogProjection) error {
				called = true
				return nil
			},
		}
		obs := &hotPathBoundedObserver{inner: inner}
		proj := hotPathLogProjection{
			EventClass: hotPathEventClass("invalid_class"),
			Mode:       hotPathModeDirect,
		}
		if err := obs.Emit(context.Background(), proj); err != nil {
			t.Fatalf("Emit error = %v, want nil", err)
		}
		if called {
			t.Errorf("inner sink was unexpectedly called for invalid EventClass")
		}

		projBadMode := hotPathLogProjection{
			EventClass: hotPathEventClassDispatch,
			Mode:       hotPathMode("unknown_mode"),
		}
		called = false
		_ = obs.Emit(context.Background(), projBadMode)
		if called {
			t.Errorf("inner sink was unexpectedly called for invalid Mode")
		}
	})

	t.Run("secret sentinels cannot reach captured sink", func(t *testing.T) {
		called := false
		inner := &fakeHotPathObserver{
			emitFn: func(ctx context.Context, p hotPathLogProjection) error {
				called = true
				return nil
			},
		}
		obs := &hotPathBoundedObserver{inner: inner}
		projSecret := hotPathLogProjection{
			EventClass: hotPathEventClassDispatch,
			Mode:       hotPathModeDirect,
			PresetID:   "SECRET_API_KEY_VAL",
		}
		_ = obs.Emit(context.Background(), projSecret)
		if called {
			t.Errorf("inner sink was unexpectedly called when projection contained secret sentinel")
		}
	})
}

func TestHotPathMetricProjectionBoundary(t *testing.T) {
	m := initHotPathMetrics()

	t.Run("valid metrics record without panic", func(t *testing.T) {
		m.recordDispatch("edge-1", hotPathModeDirect, hotPathRouteReasonModeDisabled)
		m.recordTerminal("edge-1", hotPathModeDirect, hotPathTerminalDispositionSuccess)
		m.recordCleanup("edge-1", hotPathCleanupOutcomeSuccess)
		m.recordOrphan("edge-1", hotPathOrphanOutcomeTTLExpired)
		m.recordStageDuration("edge-1", hotPathModeDirect, hotPathStageKindSelector, hotPathAttemptFirst, 0.05)
	})

	t.Run("invalid typed-string casts create no new series", func(t *testing.T) {
		// Snapshot the current series count on each vec. Deltas are robust to
		// series accumulated by earlier tests on the shared package collectors.
		dispatchBefore := testutil.CollectAndCount(m.dispatchCounter)
		terminalBefore := testutil.CollectAndCount(m.terminalCounter)
		cleanupBefore := testutil.CollectAndCount(m.cleanupCounter)
		orphanBefore := testutil.CollectAndCount(m.orphanCounter)
		stageBefore := testutil.CollectAndCount(m.stageDuration)

		// Invalid casts across every closed dimension must be rejected before
		// WithLabelValues, so no new series is created on any collector.
		m.recordDispatch("edge-invalid", hotPathMode("unknown_mode"), hotPathRouteReasonModeDisabled)
		m.recordDispatch("edge-invalid", hotPathModeDirect, hotPathRouteReason("invalid_reason"))
		m.recordTerminal("edge-invalid", hotPathMode("unknown_mode"), hotPathTerminalDispositionSuccess)
		m.recordTerminal("edge-invalid", hotPathModeDirect, hotPathTerminalDispositionKind("invalid_disp"))
		m.recordCleanup("edge-invalid", hotPathCleanupOutcome("invalid_cleanup"))
		m.recordOrphan("edge-invalid", hotPathOrphanOutcome("invalid_orphan"))
		m.recordStageDuration("edge-invalid", hotPathMode("unknown_mode"), hotPathStageKindSelector, hotPathAttemptFirst, 0.05)
		m.recordStageDuration("edge-invalid", hotPathModeDirect, hotPathStageKind("invalid_stage"), hotPathAttemptFirst, 0.05)

		if got := testutil.CollectAndCount(m.dispatchCounter) - dispatchBefore; got != 0 {
			t.Errorf("invalid dispatch casts created %d new series, want 0", got)
		}
		if got := testutil.CollectAndCount(m.terminalCounter) - terminalBefore; got != 0 {
			t.Errorf("invalid terminal casts created %d new series, want 0", got)
		}
		if got := testutil.CollectAndCount(m.cleanupCounter) - cleanupBefore; got != 0 {
			t.Errorf("invalid cleanup casts created %d new series, want 0", got)
		}
		if got := testutil.CollectAndCount(m.orphanCounter) - orphanBefore; got != 0 {
			t.Errorf("invalid orphan casts created %d new series, want 0", got)
		}
		if got := testutil.CollectAndCount(m.stageDuration) - stageBefore; got != 0 {
			t.Errorf("invalid stage casts created %d new series, want 0", got)
		}
	})

	t.Run("distinct route reasons and cleanup/orphan outcomes create distinct series", func(t *testing.T) {
		dispatchBefore := testutil.CollectAndCount(m.dispatchCounter)
		cleanupBefore := testutil.CollectAndCount(m.cleanupCounter)
		orphanBefore := testutil.CollectAndCount(m.orphanCounter)

		// Two distinct dispatch reasons with the same edge/mode produce two
		// distinct label series instead of being discarded.
		m.recordDispatch("edge-distinct", hotPathModeDirect, hotPathRouteReasonModeDisabled)
		m.recordDispatch("edge-distinct", hotPathModeDirect, hotPathRouteReasonTimeout)
		// Distinct cleanup outcomes produce distinct series.
		m.recordCleanup("edge-distinct", hotPathCleanupOutcomeSuccess)
		m.recordCleanup("edge-distinct", hotPathCleanupOutcomePrimaryError)
		// The two closed orphan outcomes produce distinct series.
		m.recordOrphan("edge-distinct", hotPathOrphanOutcomeTTLExpired)
		m.recordOrphan("edge-distinct", hotPathOrphanOutcomeCleanupFailed)

		if got := testutil.CollectAndCount(m.dispatchCounter) - dispatchBefore; got != 2 {
			t.Errorf("distinct dispatch reasons created %d new series, want 2", got)
		}
		if got := testutil.CollectAndCount(m.cleanupCounter) - cleanupBefore; got != 2 {
			t.Errorf("distinct cleanup outcomes created %d new series, want 2", got)
		}
		if got := testutil.CollectAndCount(m.orphanCounter) - orphanBefore; got != 2 {
			t.Errorf("distinct orphan outcomes created %d new series, want 2", got)
		}
	})

	t.Run("edgeID containing secret sentinel is normalized to edge-local", func(t *testing.T) {
		// A secret sentinel in the edge id collapses to the single "edge-local"
		// label and must not leak the raw value as a distinct series.
		before := testutil.CollectAndCount(m.dispatchCounter)
		m.recordDispatch("edge-SECRET-token", hotPathModeDirect, hotPathRouteReasonModeDisabled)
		m.recordDispatch("edge-bearer-value", hotPathModeDirect, hotPathRouteReasonModeDisabled)
		if got := testutil.CollectAndCount(m.dispatchCounter) - before; got > 1 {
			t.Errorf("secret edge ids created %d new series, want at most 1 (collapsed to edge-local)", got)
		}
	})
}

func TestHotPathObserverProductionFailureIsolation(t *testing.T) {
	t.Run("table of failure isolation behaviors through server seam", func(t *testing.T) {
		tests := []struct {
			name        string
			innerFn     func(ctx context.Context, p hotPathLogProjection) error
			hookFn      func(p hotPathLogProjection, err error)
			wantCalled  bool
			wantHookErr string
		}{
			{
				name: "success case",
				innerFn: func(ctx context.Context, p hotPathLogProjection) error {
					return nil
				},
				hookFn:     nil,
				wantCalled: true,
			},
			{
				name: "sink error isolated",
				innerFn: func(ctx context.Context, p hotPathLogProjection) error {
					return errors.New("sink failure")
				},
				hookFn:      nil,
				wantCalled:  true,
				wantHookErr: "sink failure",
			},
			{
				name: "sink panic isolated",
				innerFn: func(ctx context.Context, p hotPathLogProjection) error {
					panic("sink panic occurred")
				},
				hookFn:      nil,
				wantCalled:  true,
				wantHookErr: "observer panic",
			},
		}

		for _, tt := range tests {
			t.Run(tt.name, func(t *testing.T) {
				s := newTestServer(t)
				called := false
				var hookErrCaptured error

				obs := &fakeHotPathObserver{
					emitFn: func(ctx context.Context, p hotPathLogProjection) error {
						called = true
						if tt.innerFn != nil {
							return tt.innerFn(ctx, p)
						}
						return nil
					},
				}
				s.SetHotPathObserver(obs)
				s.SetHotPathObserverHook(func(p hotPathLogProjection, err error) {
					hookErrCaptured = err
					if tt.hookFn != nil {
						tt.hookFn(p, err)
					}
				})

				proj := hotPathLogProjection{
					EventClass: hotPathEventClassDispatch,
					Mode:       hotPathModeDirect,
				}

				// Must not panic or return error
				s.emitHotPathObservation(context.Background(), proj)

				if tt.wantCalled && !called {
					t.Errorf("expected inner observer to be called")
				}
				if tt.wantHookErr != "" {
					if hookErrCaptured == nil || !strings.Contains(hookErrCaptured.Error(), tt.wantHookErr) {
						t.Errorf("hook err = %v, want substring %q", hookErrCaptured, tt.wantHookErr)
					}
				}
			})
		}
	})

	t.Run("hook panic is isolated and does not interrupt execution", func(t *testing.T) {
		s := newTestServer(t)
		obs := &fakeHotPathObserver{
			emitFn: func(ctx context.Context, p hotPathLogProjection) error {
				return errors.New("sink error")
			},
		}
		s.SetHotPathObserver(obs)
		s.SetHotPathObserverHook(func(p hotPathLogProjection, err error) {
			panic("hook panic occurred")
		})

		proj := hotPathLogProjection{
			EventClass: hotPathEventClassDispatch,
			Mode:       hotPathModeDirect,
		}

		// Must not panic even though both sink and hook panic
		s.emitHotPathObservation(context.Background(), proj)
	})

	t.Run("concurrent observer replacement and emission under race detector", func(t *testing.T) {
		s := newTestServer(t)
		ctx := context.Background()
		proj := hotPathLogProjection{
			EventClass: hotPathEventClassDispatch,
			Mode:       hotPathModeDirect,
		}

		var wg sync.WaitGroup
		for i := 0; i < 50; i++ {
			wg.Add(2)
			go func() {
				defer wg.Done()
				s.SetHotPathObserver(&fakeHotPathObserver{})
			}()
			go func() {
				defer wg.Done()
				s.emitHotPathObservation(ctx, proj)
			}()
		}
		wg.Wait()
	})
}

// ---------------------------------------------------------------------------
// API-2: actual-path lifecycle emission tests.
//
// These tests drive the real Hot Path lifecycle through the existing scripted
// fixtures and assert that the joined observation lifecycle (dispatch, stage,
// transition, terminal, cleanup, orphan) is emitted exactly-once per request,
// with bounded raw-free projections and observer failure isolation (SDD S15).
// ---------------------------------------------------------------------------

// recordingHotPathObserver captures every validated projection that reaches the
// installed sink. It optionally delegates to emitFn so failure-isolation paths
// can be exercised on actual lifecycle flows.
type recordingHotPathObserver struct {
	mu          sync.Mutex
	emitFn      func(context.Context, hotPathLogProjection) error
	projections []hotPathLogProjection
}

func (r *recordingHotPathObserver) Emit(ctx context.Context, p hotPathLogProjection) error {
	r.mu.Lock()
	r.projections = append(r.projections, p)
	fn := r.emitFn
	r.mu.Unlock()
	if fn != nil {
		return fn(ctx, p)
	}
	return nil
}

func (r *recordingHotPathObserver) snapshot() []hotPathLogProjection {
	r.mu.Lock()
	defer r.mu.Unlock()
	return append([]hotPathLogProjection(nil), r.projections...)
}

type hotPathTracePoint struct {
	Event       hotPathEventClass
	Stage       hotPathStageKind
	Attempt     hotPathAttemptBucket
	Disposition hotPathTerminalDispositionKind
	Cleanup     hotPathCleanupOutcome
	Orphan      hotPathOrphanOutcome
}

func projectHotPathTrace(projections []hotPathLogProjection, requestID string) []hotPathTracePoint {
	out := make([]hotPathTracePoint, 0, len(projections))
	for _, projection := range projections {
		if projection.RequestID != requestID {
			continue
		}
		out = append(out, hotPathTracePoint{
			Event: projection.EventClass, Stage: projection.StageKind, Attempt: projection.AttemptBucket,
			Disposition: projection.Disposition, Cleanup: projection.CleanupOutcome, Orphan: projection.OrphanOutcome,
		})
	}
	return out
}

func assertHotPathTraceEqual(t *testing.T, got, want []hotPathTracePoint) {
	t.Helper()
	if !reflect.DeepEqual(got, want) {
		t.Fatalf("hot path trace mismatch:\n got: %#v\nwant: %#v", got, want)
	}
}

func hotPathPassTrace() []hotPathTracePoint {
	return []hotPathTracePoint{
		{Event: hotPathEventClassDispatch},
		{Event: hotPathEventClassStage, Stage: hotPathStageKindLocal, Attempt: hotPathAttemptFirst, Disposition: hotPathTerminalDispositionToolTurn},
		{Event: hotPathEventClassStage, Stage: hotPathStageKindLocal, Attempt: hotPathAttemptRetry, Disposition: hotPathTerminalDispositionToolTurn},
		{Event: hotPathEventClassStage, Stage: hotPathStageKindLocal, Attempt: hotPathAttemptRetry, Disposition: hotPathTerminalDispositionSuccess},
		{Event: hotPathEventClassLight, Stage: hotPathStageKindReview, Attempt: hotPathAttemptFirst},
		{Event: hotPathEventClassStage, Stage: hotPathStageKindReview, Attempt: hotPathAttemptFirst, Disposition: hotPathTerminalDispositionToolTurn},
		{Event: hotPathEventClassStage, Stage: hotPathStageKindReview, Attempt: hotPathAttemptRetry, Disposition: hotPathTerminalDispositionSuccess},
		{Event: hotPathEventClassLight, Stage: hotPathStageKindCleanup, Attempt: hotPathAttemptFirst},
		{Event: hotPathEventClassCleanup, Cleanup: hotPathCleanupOutcomeSuccess},
		{Event: hotPathEventClassTerminal, Disposition: hotPathTerminalDispositionSuccess},
	}
}

func hotPathMetricValue(t *testing.T, name string, labels map[string]string) float64 {
	t.Helper()
	families, err := prometheus.DefaultGatherer.Gather()
	if err != nil {
		t.Fatalf("gather metrics: %v", err)
	}
	var total float64
	for _, family := range families {
		if family.GetName() != name {
			continue
		}
		for _, metric := range family.Metric {
			matched := true
			for key, want := range labels {
				found := false
				for _, pair := range metric.Label {
					if pair.GetName() == key && pair.GetValue() == want {
						found = true
						break
					}
				}
				if !found {
					matched = false
					break
				}
			}
			if !matched {
				continue
			}
			switch {
			case metric.Counter != nil:
				total += metric.Counter.GetValue()
			case metric.Histogram != nil:
				total += float64(metric.Histogram.GetSampleCount())
			}
		}
	}
	return total
}

type hotPathRawSeed struct {
	Prompt        string
	Output        string
	Reasoning     string
	ToolArguments string
	ToolResult    string
	Authorization string
	Credential    string
	Provider      string
	Target        string
	ProviderError string
}

func newHotPathRawSeed(t *testing.T) hotPathRawSeed {
	t.Helper()
	suffix := strings.NewReplacer("/", "-", " ", "-").Replace(t.Name())
	return hotPathRawSeed{
		Prompt: "raw-prompt-" + suffix, Output: "raw-output-" + suffix,
		Reasoning: "raw-reasoning-" + suffix, ToolArguments: "raw-tool-args-" + suffix,
		ToolResult: "raw-tool-result-" + suffix, Authorization: "raw-auth-" + suffix,
		Credential: "raw-credential-" + suffix, Provider: "raw-provider-" + suffix,
		Target:        "raw-target-" + suffix,
		ProviderError: "raw-provider-error-" + suffix,
	}
}

func (s hotPathRawSeed) values() []string {
	return []string{s.Prompt, s.Output, s.Reasoning, s.ToolArguments, s.ToolResult, s.Authorization, s.Credential, s.Provider, s.Target, s.ProviderError}
}

func assertHotPathSeedAbsent(t *testing.T, seed hotPathRawSeed, projections []hotPathLogProjection, entries []observer.LoggedEntry) {
	t.Helper()
	serialized := fmt.Sprint(projections)
	for _, entry := range entries {
		serialized += entry.Message + fmt.Sprint(entry.ContextMap())
	}
	for _, value := range seed.values() {
		if strings.Contains(serialized, value) {
			t.Fatalf("Hot Path observation leaked seeded value %q: %s", value, serialized)
		}
	}

	families, err := prometheus.DefaultGatherer.Gather()
	if err != nil {
		t.Fatalf("gather metrics: %v", err)
	}
	for _, family := range families {
		if !strings.HasPrefix(family.GetName(), "iop_hot_path_") {
			continue
		}
		for _, metric := range family.Metric {
			for _, pair := range metric.Label {
				for _, value := range seed.values() {
					if strings.Contains(pair.GetValue(), value) {
						t.Fatalf("Hot Path metric %s label %s leaked seeded value %q", family.GetName(), pair.GetName(), value)
					}
				}
			}
		}
	}
}

type failingHotPathStageService struct {
	*scriptedLightPoolService
	mu     sync.Mutex
	calls  int
	failAt int
	fail   func(context.Context) error
}

func (s *failingHotPathStageService) SubmitProviderPool(ctx context.Context, req edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
	s.mu.Lock()
	index := s.calls
	s.calls++
	s.mu.Unlock()
	if index == s.failAt {
		return nil, s.fail(ctx)
	}
	return s.scriptedLightPoolService.SubmitProviderPool(ctx, req)
}

// hotPathRawSentinels is the set of seeded raw values that must never reach a
// Hot Path log projection or metric label on an actual lifecycle path.
var hotPathRawSentinels = []string{
	"prompt", "output", "tool_args", "tool_result",
	"authorization", "bearer", "api_key", "secret",
	"credential", "raw_body", "content", "reasoning",
	"provider_error_detail", "header",
}

// projectionLeakSentinel reports whether any captured projection field contains
// a raw sentinel. Captured projections are already validated and sanitized by
// the bounded observer, so this asserts the contract holds on actual paths.
func projectionLeakSentinel(p hotPathLogProjection) string {
	fields := []string{
		string(p.EventClass), string(p.Mode), string(p.StageKind), string(p.Disposition),
		p.Correlation, p.StageID, p.RequestID, p.CallID, p.OwnerEdgeID,
		string(p.Reason), p.PresetID, string(p.AttemptBucket),
		string(p.CleanupOutcome), string(p.OrphanOutcome),
	}
	for _, sentinel := range hotPathRawSentinels {
		needle := strings.ToLower(sentinel)
		for _, f := range fields {
			if strings.Contains(strings.ToLower(f), needle) {
				return sentinel
			}
		}
	}
	return ""
}

// firstDispatchRequestID returns the request id carried by the first dispatch
// observation. The dispatch admission emit is the lifecycle join root.
func firstDispatchRequestID(projs []hotPathLogProjection) string {
	for _, p := range projs {
		if p.EventClass == hotPathEventClassDispatch && p.RequestID != "" {
			return p.RequestID
		}
	}
	return ""
}

// eventClassCounts groups captured projections by event class for one request.
func eventClassCounts(projs []hotPathLogProjection, requestID string) map[hotPathEventClass]int {
	out := make(map[hotPathEventClass]int)
	for _, p := range projs {
		if p.RequestID == requestID {
			out[p.EventClass]++
		}
	}
	return out
}

// assertProjectionsRawFree fails the test if any captured projection carries a
// raw sentinel in any field.
func assertProjectionsRawFree(t *testing.T, projs []hotPathLogProjection) {
	t.Helper()
	for i, p := range projs {
		if leak := projectionLeakSentinel(p); leak != "" {
			t.Fatalf("projection %d leaked raw sentinel %q: %+v", i, leak, p)
		}
	}
}

// assertProjectionsUseClosedEnums fails if any captured projection carries a
// non-empty enum field that is not a closed value.
func assertProjectionsUseClosedEnums(t *testing.T, projs []hotPathLogProjection) {
	t.Helper()
	for i, p := range projs {
		if p.Mode != "" && !hotPathModeIsValid(p.Mode) {
			t.Fatalf("projection %d has unclosed mode %q", i, p.Mode)
		}
		if p.StageKind != "" && !hotPathStageKindIsValid(p.StageKind) {
			t.Fatalf("projection %d has unclosed stage kind %q", i, p.StageKind)
		}
		if p.Disposition != "" && !hotPathTerminalDispositionIsValid(p.Disposition) {
			t.Fatalf("projection %d has unclosed disposition %q", i, p.Disposition)
		}
		if p.Reason != "" && !hotPathRouteReasonIsValid(p.Reason) {
			t.Fatalf("projection %d has unclosed reason %q", i, p.Reason)
		}
		if p.AttemptBucket != "" && !hotPathAttemptBucketIsValid(p.AttemptBucket) {
			t.Fatalf("projection %d has unclosed attempt bucket %q", i, p.AttemptBucket)
		}
		if p.CleanupOutcome != "" && !hotPathCleanupOutcomeIsValid(p.CleanupOutcome) {
			t.Fatalf("projection %d has unclosed cleanup outcome %q", i, p.CleanupOutcome)
		}
		if p.OrphanOutcome != "" && !hotPathOrphanOutcomeIsValid(p.OrphanOutcome) {
			t.Fatalf("projection %d has unclosed orphan outcome %q", i, p.OrphanOutcome)
		}
	}
}

// driveScriptedLightPass drives a full non-repair light lifecycle through the
// scripted fixture and returns the final response. It mirrors the proven
// TestHotPathCleanupTerminalMatrix pattern.
func driveScriptedLightPass(t *testing.T, fixture *scriptedLightFixture) *httptest.ResponseRecorder {
	t.Helper()
	cleanup := fixture.runToCleanup()
	fixture.consumeToolResponse(cleanup, []string{`{"written":true}`})
	return fixture.request()
}

func scriptedRawDirectTool(endpoint string, seed hotPathRawSeed) string {
	if endpoint == "anthropic" {
		return fmt.Sprintf(`{"id":"msg-raw-seed","type":"message","role":"assistant","content":[{"type":"thinking","thinking":%q,"signature":"sig"},{"type":"text","text":%q},{"type":"tool_use","id":"provider-raw-tool","name":"run_command","input":{"command":%q}}],"stop_reason":"tool_use"}`,
			seed.Reasoning, seed.Output, seed.ToolArguments)
	}
	arguments, _ := json.Marshal(map[string]string{"command": seed.ToolArguments})
	return fmt.Sprintf(`{"id":"chatcmpl-raw-seed","created":1,"choices":[{"message":{"role":"assistant","content":%q,"reasoning_content":%q,"tool_calls":[{"id":"provider-raw-tool","type":"function","function":{"name":"run_command","arguments":%q}}]},"finish_reason":"tool_calls"}]}`,
		seed.Output, seed.Reasoning, string(arguments))
}

func serveRawSeededRequest(t *testing.T, srv *Server, endpoint string, body []byte, seed hotPathRawSeed, writer http.ResponseWriter, ctx context.Context) {
	t.Helper()
	path := "/v1/chat/completions"
	if endpoint == "anthropic" {
		path = "/v1/messages"
	}
	request := httptest.NewRequest(http.MethodPost, path, strings.NewReader(string(body))).WithContext(ctx)
	request.Header.Set("Authorization", "Bearer "+seed.Authorization)
	request.Header.Set("X-Api-Key", seed.Authorization)
	request.Header.Set("X-Raw-Observation", seed.Output)
	request.Header.Set("X-IOP-Provider-Authorization", seed.Credential)
	if endpoint == "anthropic" {
		request.Header.Set(anthropicVersionHeader, anthropicSupportedVersion)
	}
	srv.routes().ServeHTTP(writer, request)
}

func TestHotPathObservationLifecycle_ProductionZapObserver(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		t.Run(endpoint, func(t *testing.T) {
			seed := newHotPathRawSeed(t)
			fixture := newScriptedLightFixture(t, endpoint, false)
			fixture.history = []any{map[string]any{"role": "user", "content": seed.Prompt}}
			fixture.service.responses[0] = func(string) string { return scriptedRawDirectTool(endpoint, seed) }
			oldProvider := fixture.service.candidate.ProviderID
			fixture.service.candidate.ProviderID = seed.Provider
			fixture.service.candidate.ActualModel = seed.Target

			catalog := fixture.server.modelCatalogSnapshot()
			for index := range catalog {
				if _, ok := catalog[index].Providers[oldProvider]; ok {
					delete(catalog[index].Providers, oldProvider)
					catalog[index].Providers[seed.Provider] = seed.Target
				}
			}
			core, observed := observer.New(zap.InfoLevel)
			cfg := config.EdgeOpenAIConf{
				BearerToken: seed.Authorization,
				ProviderAuth: config.EdgeOpenAIProviderAuthConf{
					Enabled: true, FromHeader: "X-IOP-Provider-Authorization",
					TargetHeader: "Authorization", Scheme: "Bearer", Required: true,
				},
			}
			server := NewServer(cfg, fixture.service, zap.New(core))
			server.SetEdgeID("edge-production-zap-" + endpoint)
			server.SetExecutionPresets(fixture.server.ExecutionPresetsSnapshot())
			server.SetModelCatalog(catalog)
			fixture.server = server

			body := scriptedArtifactRequestBodyWithOptions(t, endpoint, fixture.tools, fixture.history, 0, false)
			response := httptest.NewRecorder()
			serveRawSeededRequest(t, server, endpoint, body, seed, response, context.Background())
			if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), seed.ToolArguments) {
				t.Fatalf("seeded direct response status=%d body=%s", response.Code, response.Body.String())
			}

			entries := observed.FilterMessage(hotPathObservationMessage).All()
			if len(entries) != 1 {
				t.Fatalf("production Hot Path log entries=%d, want 1: %+v", len(entries), entries)
			}
			keys := make([]string, 0, len(entries[0].ContextMap()))
			for key := range entries[0].ContextMap() {
				keys = append(keys, key)
			}
			sort.Strings(keys)
			wantKeys := logProjectionKeys()
			sort.Strings(wantKeys)
			if !reflect.DeepEqual(keys, wantKeys) {
				t.Fatalf("production zap keys=%v, want exact allowlist %v", keys, wantKeys)
			}
			if entries[0].ContextMap()["hot_path_event_class"] != string(hotPathEventClassDispatch) {
				t.Fatalf("production zap entry=%v, want initial dispatch", entries[0].ContextMap())
			}
			requests := fixture.service.snapshots()
			if len(requests) != 1 || requests[0].Run.ModelGroupKey != "selector-model" {
				t.Fatalf("seeded selector requests=%+v", requests)
			}
			prepared, err := requests[0].PrepareProtocolTunnel(requests[0].Tunnel, fixture.service.candidate)
			if err != nil || prepared.BuildBody == nil {
				t.Fatalf("prepare seeded provider tunnel: err=%v request=%+v", err, prepared)
			}
			providerPrompt := requests[0].Run.Prompt
			if endpoint == "anthropic" {
				providerBody, buildErr := prepared.BuildBody(fixture.service.candidate.ActualModel)
				if buildErr != nil {
					t.Fatalf("build seeded Anthropic provider body: %v", buildErr)
				}
				providerPrompt = string(providerBody)
			}
			if !strings.Contains(providerPrompt, seed.Prompt) || fixture.service.candidate.ProviderID != seed.Provider || fixture.service.candidate.ActualModel != seed.Target {
				t.Fatalf("raw prompt/provider fixtures were not inserted: prompt=%q provider=%q target=%q", providerPrompt, fixture.service.candidate.ProviderID, fixture.service.candidate.ActualModel)
			}
			if !strings.Contains(fmt.Sprint(prepared.Headers), seed.Credential) {
				t.Fatalf("provider credential fixture was not forwarded: headers=%v", prepared.Headers)
			}
			assertHotPathSeedAbsent(t, seed, nil, entries)
		})
	}
}

func TestHotPathObservationLifecycle_LightPass(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		t.Run(endpoint, func(t *testing.T) {
			fixture := newScriptedLightFixture(t, endpoint, false)
			edgeID := "edge-observation-pass-" + endpoint
			fixture.server.SetEdgeID(edgeID)
			rec := &recordingHotPathObserver{}
			fixture.server.SetHotPathObserver(rec)
			stageBefore := hotPathMetricValue(t, "iop_hot_path_stage_duration_seconds", map[string]string{"edge_id": edgeID})
			terminalBefore := hotPathMetricValue(t, "iop_hot_path_terminal_total", map[string]string{"edge_id": edgeID})
			cleanupBefore := hotPathMetricValue(t, "iop_hot_path_cleanup_total", map[string]string{"edge_id": edgeID})

			final := driveScriptedLightPass(t, fixture)
			if final.Code != http.StatusOK {
				t.Fatalf("light pass final status=%d body=%s", final.Code, final.Body.String())
			}

			projs := rec.snapshot()
			assertProjectionsRawFree(t, projs)
			assertProjectionsUseClosedEnums(t, projs)

			requestID := firstDispatchRequestID(projs)
			if requestID == "" {
				t.Fatalf("no dispatch admission observation emitted; projs=%v", projs)
			}
			assertHotPathTraceEqual(t, projectHotPathTrace(projs, requestID), hotPathPassTrace())
			if delta := hotPathMetricValue(t, "iop_hot_path_stage_duration_seconds", map[string]string{"edge_id": edgeID}) - stageBefore; delta != 5 {
				t.Fatalf("stage metric delta=%v, want 5", delta)
			}
			if delta := hotPathMetricValue(t, "iop_hot_path_terminal_total", map[string]string{"edge_id": edgeID, "hot_path_disposition": "success"}) - terminalBefore; delta != 1 {
				t.Fatalf("terminal metric delta=%v, want 1", delta)
			}
			if delta := hotPathMetricValue(t, "iop_hot_path_cleanup_total", map[string]string{"edge_id": edgeID, "hot_path_cleanup_outcome": "success"}) - cleanupBefore; delta != 1 {
				t.Fatalf("cleanup metric delta=%v, want 1", delta)
			}
		})
	}
}

func TestHotPathObservationLifecycle_LightRepair(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		t.Run(endpoint, func(t *testing.T) {
			fixture := newScriptedLightFixture(t, endpoint, true)
			rec := &recordingHotPathObserver{}
			fixture.server.SetHotPathObserver(rec)

			final := driveScriptedLightPass(t, fixture)
			if final.Code != http.StatusOK {
				t.Fatalf("light repair final status=%d body=%s", final.Code, final.Body.String())
			}

			projs := rec.snapshot()
			assertProjectionsRawFree(t, projs)
			requestID := firstDispatchRequestID(projs)
			want := hotPathPassTrace()
			want[6].Disposition = hotPathTerminalDispositionToolTurn
			want = append(want[:7], append([]hotPathTracePoint{
				{Event: hotPathEventClassLight, Stage: hotPathStageKindReview, Attempt: hotPathAttemptRetry},
				{Event: hotPathEventClassStage, Stage: hotPathStageKindReview, Attempt: hotPathAttemptRetry, Disposition: hotPathTerminalDispositionSuccess},
			}, want[7:]...)...)
			assertHotPathTraceEqual(t, projectHotPathTrace(projs, requestID), want)
		})
	}
}

func TestHotPathObservationLifecycle_CleanupFailure(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		t.Run(endpoint, func(t *testing.T) {
			fixture := newScriptedLightFixture(t, endpoint, false)
			rec := &recordingHotPathObserver{}
			fixture.server.SetHotPathObserver(rec)

			cleanup := fixture.runToCleanup()
			// Cleanup delete result mismatches the receipt: the primary success
			// is converted to a primary-error cleanup.
			fixture.consumeToolResponse(cleanup, []string{`{"written":false}`})
			final := fixture.request()
			if final.Code != http.StatusBadGateway {
				t.Fatalf("cleanup failure final status=%d body=%s", final.Code, final.Body.String())
			}

			projs := rec.snapshot()
			assertProjectionsRawFree(t, projs)
			requestID := firstDispatchRequestID(projs)
			want := hotPathPassTrace()
			want[len(want)-2].Cleanup = hotPathCleanupOutcomePrimaryError
			want[len(want)-1].Disposition = hotPathTerminalDispositionProviderError
			assertHotPathTraceEqual(t, projectHotPathTrace(projs, requestID), want)
		})
	}
}

func TestHotPathObservationLifecycle_ObserverFailureMetric(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		for _, kind := range []string{"error", "panic"} {
			kind := kind
			t.Run(endpoint+"/"+kind, func(t *testing.T) {
				fixture := newScriptedLightFixture(t, endpoint, false)
				edgeID := "edge-observer-failure-" + endpoint + "-" + kind
				fixture.server.SetEdgeID(edgeID)
				rec := &recordingHotPathObserver{emitFn: func(context.Context, hotPathLogProjection) error {
					if kind == "panic" {
						panic("observer panic on actual path")
					}
					return errors.New("observer sink unavailable")
				}}
				fixture.server.SetHotPathObserver(rec)
				fixture.server.SetHotPathObserverHook(func(hotPathLogProjection, error) {})
				before := hotPathMetricValue(t, "iop_hot_path_observer_failures_total", map[string]string{"edge_id": edgeID})

				final := driveScriptedLightPass(t, fixture)
				if final.Code != http.StatusOK {
					t.Fatalf("observer %s altered response: status=%d body=%s", kind, final.Code, final.Body.String())
				}
				projections := rec.snapshot()
				if len(projections) != len(hotPathPassTrace()) {
					t.Fatalf("observer %s calls=%d, want %d", kind, len(projections), len(hotPathPassTrace()))
				}
				after := hotPathMetricValue(t, "iop_hot_path_observer_failures_total", map[string]string{"edge_id": edgeID})
				if delta := after - before; delta != float64(len(projections)) {
					t.Fatalf("observer failure metric delta=%v, want %d", delta, len(projections))
				}
			})
		}
	}
}

func TestHotPathObservationLifecycle_OrphanTTL(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		t.Run(endpoint, func(t *testing.T) {
			fixture := newScriptedLightFixture(t, endpoint, false)
			edgeID := "edge-orphan-" + endpoint
			fixture.server.SetEdgeID(edgeID)
			rec := &recordingHotPathObserver{}
			fixture.server.SetHotPathObserver(rec)
			before := hotPathMetricValue(t, "iop_hot_path_orphan_total", map[string]string{
				"edge_id": edgeID, "hot_path_orphan_outcome": "ttl_expired",
			})

			prepare := fixture.request()
			fixture.consumeToolResponse(prepare, []string{`{"written":true}`})

			fixture.server.lightFlows.mu.Lock()
			var requestID string
			for id := range fixture.server.lightFlows.records {
				requestID = id
			}
			fixture.server.lightFlows.mu.Unlock()
			if requestID == "" {
				t.Fatal("no light record admitted for orphan test")
			}

			// Force the request into a sweepable detached state, then advance the
			// coordinator clock past TTL and sweep. The workspace stores remain
			// populated, so the TTL handoff emits an orphan observation.
			_ = fixture.server.requestCoordinator.disconnect(requestID, fixture.server.edgeIDValue(), "cancelled")
			fixture.server.requestCoordinator.mu.Lock()
			expireAt := fixture.server.requestCoordinator.now().Add(fixture.server.requestCoordinator.ttl + time.Second)
			fixture.server.requestCoordinator.now = func() time.Time { return expireAt }
			fixture.server.requestCoordinator.mu.Unlock()
			fixture.server.sweepLogicalRequestTTL()

			projs := rec.snapshot()
			assertProjectionsRawFree(t, projs)
			assertHotPathTraceEqual(t, projectHotPathTrace(projs, requestID), []hotPathTracePoint{
				{Event: hotPathEventClassDispatch},
				{Event: hotPathEventClassOrphan, Orphan: hotPathOrphanOutcomeTTLExpired},
			})
			after := hotPathMetricValue(t, "iop_hot_path_orphan_total", map[string]string{"edge_id": edgeID, "hot_path_orphan_outcome": "ttl_expired"})
			if delta := after - before; delta != 1 {
				t.Fatalf("orphan metric delta=%v, want 1", delta)
			}
		})
	}
}

func TestHotPathObservationLifecycle_DirectToolContinuation(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		t.Run(endpoint, func(t *testing.T) {
			seed := newHotPathRawSeed(t)
			candidate := anthropicTestCandidate(t, map[string]string{"openai": "openai", "anthropic": "anthropic"}[endpoint])
			candidate.ProviderID = seed.Provider
			candidate.ActualModel = seed.Target
			service := &scriptedArtifactPoolService{endpoint: endpoint, candidate: candidate}
			service.response = func(_ string, call int) string {
				if call == 1 {
					return scriptedRawDirectTool(endpoint, seed)
				}
				return scriptedLightCompletion(endpoint, seed.Output+"-final")
			}
			server := newScriptedArtifactHandlerServer(t, service)
			server.SetEdgeID("edge-direct-continuation-" + endpoint)
			recorder := &recordingHotPathObserver{}
			server.SetHotPathObserver(recorder)
			tools := scriptedLightTools(endpoint)
			history := []any{map[string]any{"role": "user", "content": seed.Prompt}}

			first := serveScriptedArtifactRequest(t, server, endpoint, scriptedArtifactRequestBody(t, endpoint, tools, history))
			assistant, ids, err := artifactAssistantFromResponse(endpoint, first.Body.Bytes())
			if first.Code != http.StatusOK || err != nil || len(ids) != 1 {
				t.Fatalf("direct tool turn status=%d ids=%v err=%v body=%s", first.Code, ids, err, first.Body.String())
			}
			history = append(history, assistant)
			history = scriptedArtifactAppendResults(endpoint, history, ids, []string{seed.ToolResult})
			continuationBody := scriptedArtifactRequestBody(t, endpoint, tools, history)
			if !strings.Contains(string(continuationBody), seed.ToolResult) {
				t.Fatalf("tool-result seed was not inserted into continuation: %s", continuationBody)
			}
			final := serveScriptedArtifactRequest(t, server, endpoint, continuationBody)
			if final.Code != http.StatusOK || !strings.Contains(final.Body.String(), seed.Output+"-final") {
				t.Fatalf("direct continuation status=%d body=%s", final.Code, final.Body.String())
			}

			projections := recorder.snapshot()
			requestID := firstDispatchRequestID(projections)
			assertHotPathTraceEqual(t, projectHotPathTrace(projections, requestID), []hotPathTracePoint{
				{Event: hotPathEventClassDispatch},
				{Event: hotPathEventClassTerminal, Disposition: hotPathTerminalDispositionSuccess},
			})
			assertHotPathSeedAbsent(t, seed, projections, nil)
		})
	}
}

func driveScriptedLightToFirstLocal(t *testing.T, fixture *scriptedLightFixture, toolResult string) *httptest.ResponseRecorder {
	t.Helper()
	prepare := fixture.request()
	fixture.consumeToolResponse(prepare, []string{fmt.Sprintf(`{"written":true,"raw":%q}`, toolResult)})
	pair := fixture.request()
	fixture.consumeToolResponse(pair, []string{
		fmt.Sprintf(`{"written":true,"raw":%q}`, toolResult),
		fmt.Sprintf(`{"written":true,"raw":%q}`, toolResult),
	})
	return fixture.request()
}

func TestHotPathObservationLifecycle_ProviderError(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		t.Run(endpoint, func(t *testing.T) {
			seed := newHotPathRawSeed(t)
			fixture := newScriptedLightFixture(t, endpoint, false)
			edgeID := "edge-stage-provider-error-" + endpoint
			fixture.server.SetEdgeID(edgeID)
			fixture.server.service = &failingHotPathStageService{
				scriptedLightPoolService: fixture.service, failAt: 2,
				fail: func(context.Context) error { return errors.New(seed.ProviderError) },
			}
			recorder := &recordingHotPathObserver{}
			fixture.server.SetHotPathObserver(recorder)
			stageBefore := hotPathMetricValue(t, "iop_hot_path_stage_duration_seconds", map[string]string{"edge_id": edgeID})

			cleanup := driveScriptedLightToFirstLocal(t, fixture, seed.ToolResult)
			fixture.consumeToolResponse(cleanup, []string{fmt.Sprintf(`{"written":true,"raw":%q}`, seed.ToolResult)})
			final := fixture.request()
			if final.Code != http.StatusBadGateway {
				t.Fatalf("provider-error final status=%d body=%s", final.Code, final.Body.String())
			}
			projections := recorder.snapshot()
			requestID := firstDispatchRequestID(projections)
			assertHotPathTraceEqual(t, projectHotPathTrace(projections, requestID), []hotPathTracePoint{
				{Event: hotPathEventClassDispatch},
				{Event: hotPathEventClassStage, Stage: hotPathStageKindLocal, Attempt: hotPathAttemptFirst, Disposition: hotPathTerminalDispositionProviderError},
				{Event: hotPathEventClassLight, Stage: hotPathStageKindCleanup, Attempt: hotPathAttemptFirst},
				{Event: hotPathEventClassCleanup, Cleanup: hotPathCleanupOutcomePrimaryError},
				{Event: hotPathEventClassTerminal, Disposition: hotPathTerminalDispositionProviderError},
			})
			if delta := hotPathMetricValue(t, "iop_hot_path_stage_duration_seconds", map[string]string{"edge_id": edgeID}) - stageBefore; delta != 1 {
				t.Fatalf("failed stage metric delta=%v, want 1", delta)
			}
			assertHotPathSeedAbsent(t, seed, projections, nil)
		})
	}
}

func TestHotPathObservationLifecycle_Timeout(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		t.Run(endpoint, func(t *testing.T) {
			fixture := newScriptedLightFixture(t, endpoint, false)
			edgeID := "edge-stage-timeout-" + endpoint
			fixture.server.SetEdgeID(edgeID)
			fixture.server.service = &failingHotPathStageService{
				scriptedLightPoolService: fixture.service, failAt: 2,
				fail: func(context.Context) error { return context.DeadlineExceeded },
			}
			recorder := &recordingHotPathObserver{}
			fixture.server.SetHotPathObserver(recorder)

			cleanup := driveScriptedLightToFirstLocal(t, fixture, "timeout-tool-result")
			fixture.consumeToolResponse(cleanup, []string{`{"written":true}`})
			final := fixture.request()
			if final.Code != http.StatusBadGateway {
				t.Fatalf("timeout final status=%d body=%s", final.Code, final.Body.String())
			}
			projections := recorder.snapshot()
			requestID := firstDispatchRequestID(projections)
			assertHotPathTraceEqual(t, projectHotPathTrace(projections, requestID), []hotPathTracePoint{
				{Event: hotPathEventClassDispatch},
				{Event: hotPathEventClassStage, Stage: hotPathStageKindLocal, Attempt: hotPathAttemptFirst, Disposition: hotPathTerminalDispositionTimeout},
				{Event: hotPathEventClassLight, Stage: hotPathStageKindCleanup, Attempt: hotPathAttemptFirst},
				{Event: hotPathEventClassCleanup, Cleanup: hotPathCleanupOutcomePrimaryError},
				{Event: hotPathEventClassTerminal, Disposition: hotPathTerminalDispositionTimeout},
			})
		})
	}
}

func TestHotPathObservationLifecycle_CallerCancel(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		t.Run(endpoint, func(t *testing.T) {
			fixture := newScriptedLightFixture(t, endpoint, false)
			edgeID := "edge-stage-caller-cancel-" + endpoint
			fixture.server.SetEdgeID(edgeID)
			recorder := &recordingHotPathObserver{}
			fixture.server.SetHotPathObserver(recorder)

			prepare := fixture.request()
			fixture.consumeToolResponse(prepare, []string{`{"written":true}`})
			pair := fixture.request()
			fixture.consumeToolResponse(pair, []string{`{"written":true}`, `{"written":true}`})
			cancelled, cancel := context.WithCancel(context.Background())
			cancel()
			_ = fixture.requestWithContext(cancelled, 0)

			requestID := firstDispatchRequestID(recorder.snapshot())
			fixture.server.requestCoordinator.mu.Lock()
			expireAt := fixture.server.requestCoordinator.now().Add(fixture.server.requestCoordinator.ttl + time.Second)
			fixture.server.requestCoordinator.now = func() time.Time { return expireAt }
			fixture.server.requestCoordinator.mu.Unlock()
			fixture.server.sweepLogicalRequestTTL()

			assertHotPathTraceEqual(t, projectHotPathTrace(recorder.snapshot(), requestID), []hotPathTracePoint{
				{Event: hotPathEventClassDispatch},
				{Event: hotPathEventClassStage, Stage: hotPathStageKindLocal, Attempt: hotPathAttemptFirst, Disposition: hotPathTerminalDispositionCallerCancel},
				{Event: hotPathEventClassOrphan, Orphan: hotPathOrphanOutcomeTTLExpired},
			})
		})
	}
}

type cancelingHotPathResponseWriter struct {
	header http.Header
	writes int
}

func (w *cancelingHotPathResponseWriter) Header() http.Header {
	if w.header == nil {
		w.header = make(http.Header)
	}
	return w.header
}

func (*cancelingHotPathResponseWriter) WriteHeader(int) {}

func (w *cancelingHotPathResponseWriter) Write([]byte) (int, error) {
	w.writes++
	return 0, context.Canceled
}

func serveHotPathWriteFailureRequest(t *testing.T, server *Server, endpoint, body string, writer http.ResponseWriter) {
	t.Helper()
	path := "/v1/chat/completions"
	if endpoint == "anthropic" {
		path = "/v1/messages"
	}
	request := httptest.NewRequest(http.MethodPost, path, strings.NewReader(body))
	if endpoint == "anthropic" {
		request.Header.Set(anthropicVersionHeader, anthropicSupportedVersion)
	}
	server.routes().ServeHTTP(writer, request)
}

func hotPathTerminalMetricLabels(edgeID string, mode hotPathMode, disposition hotPathTerminalDispositionKind) map[string]string {
	return map[string]string{
		"edge_id": edgeID, "hot_path_mode": string(mode), "hot_path_disposition": string(disposition),
	}
}

func assertHotPathCallerCancelTerminalMetricDelta(t *testing.T, edgeID string, mode hotPathMode, callerCancelBefore, lengthBefore, providerErrorBefore float64) {
	t.Helper()
	callerCancelAfter := hotPathMetricValue(t, "iop_hot_path_terminal_total", hotPathTerminalMetricLabels(edgeID, mode, hotPathTerminalDispositionCallerCancel))
	if delta := callerCancelAfter - callerCancelBefore; delta != 1 {
		t.Fatalf("caller_cancel terminal metric delta=%v, want 1", delta)
	}
	lengthAfter := hotPathMetricValue(t, "iop_hot_path_terminal_total", hotPathTerminalMetricLabels(edgeID, mode, hotPathTerminalDispositionLength))
	if delta := lengthAfter - lengthBefore; delta != 0 {
		t.Fatalf("length terminal metric delta=%v, want 0", delta)
	}
	providerErrorAfter := hotPathMetricValue(t, "iop_hot_path_terminal_total", hotPathTerminalMetricLabels(edgeID, mode, hotPathTerminalDispositionProviderError))
	if delta := providerErrorAfter - providerErrorBefore; delta != 0 {
		t.Fatalf("provider_error terminal metric delta=%v, want 0", delta)
	}
}

func TestHotPathObservationLifecycle_DirectCallerWriteFailure(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		for _, response := range []struct {
			name string
			body string
		}{
			{
				name: "final",
				body: map[string]string{
					"openai":    `{"id":"chatcmpl-write-final","created":1,"choices":[{"message":{"role":"assistant","content":"final"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`,
					"anthropic": `{"id":"msg-write-final","type":"message","role":"assistant","content":[{"type":"text","text":"final"}],"stop_reason":"end_turn","usage":{"input_tokens":1,"output_tokens":1}}`,
				}[endpoint],
			},
			{
				name: "tool",
				body: map[string]string{
					"openai":    `{"id":"chatcmpl-write-tool","created":1,"choices":[{"message":{"role":"assistant","tool_calls":[{"id":"provider-write-tool","type":"function","function":{"name":"read_file","arguments":"{\"path\":\"README.md\"}"}}]},"finish_reason":"tool_calls"}]}`,
					"anthropic": `{"id":"msg-write-tool","type":"message","role":"assistant","content":[{"type":"tool_use","id":"provider-write-tool","name":"read_file","input":{"path":"README.md"}}],"stop_reason":"tool_use"}`,
				}[endpoint],
			},
		} {
			response := response
			t.Run(endpoint+"/"+response.name, func(t *testing.T) {
				candidate := anthropicTestCandidate(t, map[string]string{"openai": "openai", "anthropic": "anthropic"}[endpoint])
				frames := staticProviderTunnelFrames(response.body)
				if endpoint == "anthropic" {
					frames = anthropicTunnelFrames(http.StatusOK, "application/json", []byte(response.body))
				}
				server, _ := newHotPathHandlerServer(t, candidate, frames)
				edgeID := "edge-direct-write-cancel-" + endpoint + "-" + response.name
				server.SetEdgeID(edgeID)
				recorder := &recordingHotPathObserver{}
				server.SetHotPathObserver(recorder)

				callerCancelBefore := hotPathMetricValue(t, "iop_hot_path_terminal_total", hotPathTerminalMetricLabels(edgeID, hotPathModeDirect, hotPathTerminalDispositionCallerCancel))
				lengthBefore := hotPathMetricValue(t, "iop_hot_path_terminal_total", hotPathTerminalMetricLabels(edgeID, hotPathModeDirect, hotPathTerminalDispositionLength))
				providerErrorBefore := hotPathMetricValue(t, "iop_hot_path_terminal_total", hotPathTerminalMetricLabels(edgeID, hotPathModeDirect, hotPathTerminalDispositionProviderError))

				requestBody := map[string]string{
					"openai":    `{"model":"virtual-model","messages":[{"role":"user","content":"write cancellation"}],"tools":[{"type":"function","function":{"name":"read_file","parameters":{"type":"object"}}}]}`,
					"anthropic": `{"model":"virtual-model","max_tokens":64,"messages":[{"role":"user","content":"write cancellation"}],"tools":[{"name":"read_file","description":"read","input_schema":{"type":"object"}}]}`,
				}[endpoint]
				writer := &cancelingHotPathResponseWriter{}
				serveHotPathWriteFailureRequest(t, server, endpoint, requestBody, writer)
				if writer.writes == 0 {
					t.Fatal("caller-write fixture did not exercise ResponseWriter.Write")
				}

				projections := recorder.snapshot()
				requestID := firstDispatchRequestID(projections)
				if requestID == "" {
					t.Fatalf("direct write failure did not emit a dispatch request id: %+v", projections)
				}
				assertHotPathTraceEqual(t, projectHotPathTrace(projections, requestID), []hotPathTracePoint{
					{Event: hotPathEventClassDispatch},
					{Event: hotPathEventClassTerminal, Disposition: hotPathTerminalDispositionCallerCancel},
				})
				assertHotPathCallerCancelTerminalMetricDelta(t, edgeID, hotPathModeDirect, callerCancelBefore, lengthBefore, providerErrorBefore)
			})
		}
	}
}

func TestHotPathObservationLifecycle_LightLengthCallerWriteFailure(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		for _, terminal := range []struct {
			name             string
			outputCap        int
			localResponse    func() string
			stageDisposition hotPathTerminalDispositionKind
		}{
			{
				name: "provider-length",
				localResponse: func() string {
					return map[string]string{
						"openai":    `{"id":"chatcmpl-write-length","created":1,"choices":[{"message":{"role":"assistant","content":"limited"},"finish_reason":"length"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`,
						"anthropic": `{"id":"msg-write-length","type":"message","role":"assistant","content":[{"type":"text","text":"limited"}],"stop_reason":"max_tokens","usage":{"input_tokens":1,"output_tokens":1}}`,
					}[endpoint]
				},
				stageDisposition: hotPathTerminalDispositionLength,
			},
			{
				name:      "output-budget",
				outputCap: 4,
				localResponse: func() string {
					return scriptedLightCompletionWithUsage(endpoint, "limited", "", 1, 4)
				},
				stageDisposition: hotPathTerminalDispositionSuccess,
			},
		} {
			terminal := terminal
			t.Run(endpoint+"/"+terminal.name, func(t *testing.T) {
				fixture := newScriptedLightFixture(t, endpoint, false)
				edgeID := "edge-light-write-cancel-" + endpoint + "-" + terminal.name
				fixture.server.SetEdgeID(edgeID)
				recorder := &recordingHotPathObserver{}
				fixture.server.SetHotPathObserver(recorder)
				fixture.service.responses[3] = func(string) string { return terminal.localResponse() }

				prepare := fixture.request()
				fixture.consumeToolResponse(prepare, []string{`{"written":true}`})
				pair := fixture.request()
				fixture.consumeToolResponse(pair, []string{`{"written":true}`, `{"written":true}`})
				localRead := fixture.request()
				fixture.consumeToolResponse(localRead, []string{`{"written":true}`})

				callerCancelBefore := hotPathMetricValue(t, "iop_hot_path_terminal_total", hotPathTerminalMetricLabels(edgeID, hotPathModeLight, hotPathTerminalDispositionCallerCancel))
				lengthBefore := hotPathMetricValue(t, "iop_hot_path_terminal_total", hotPathTerminalMetricLabels(edgeID, hotPathModeLight, hotPathTerminalDispositionLength))
				providerErrorBefore := hotPathMetricValue(t, "iop_hot_path_terminal_total", hotPathTerminalMetricLabels(edgeID, hotPathModeLight, hotPathTerminalDispositionProviderError))

				body := scriptedArtifactRequestBodyWithOptions(t, endpoint, fixture.tools, fixture.history, terminal.outputCap, false)
				writer := &cancelingHotPathResponseWriter{}
				serveHotPathWriteFailureRequest(t, fixture.server, endpoint, string(body), writer)
				if writer.writes == 0 {
					t.Fatal("caller-write fixture did not exercise ResponseWriter.Write")
				}

				projections := recorder.snapshot()
				requestID := firstDispatchRequestID(projections)
				if requestID == "" {
					t.Fatalf("light write failure did not emit a dispatch request id: %+v", projections)
				}
				assertHotPathTraceEqual(t, projectHotPathTrace(projections, requestID), []hotPathTracePoint{
					{Event: hotPathEventClassDispatch},
					{Event: hotPathEventClassStage, Stage: hotPathStageKindLocal, Attempt: hotPathAttemptFirst, Disposition: hotPathTerminalDispositionToolTurn},
					{Event: hotPathEventClassStage, Stage: hotPathStageKindLocal, Attempt: hotPathAttemptRetry, Disposition: terminal.stageDisposition},
					{Event: hotPathEventClassTerminal, Disposition: hotPathTerminalDispositionCallerCancel},
				})
				assertHotPathCallerCancelTerminalMetricDelta(t, edgeID, hotPathModeLight, callerCancelBefore, lengthBefore, providerErrorBefore)
			})
		}
	}
}

func TestHotPathObservationLifecycle_CallerWriteFailure(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		t.Run(endpoint, func(t *testing.T) {
			fixture := newScriptedLightFixture(t, endpoint, false)
			recorder := &recordingHotPathObserver{}
			fixture.server.SetHotPathObserver(recorder)
			cleanup := fixture.runToCleanup()
			fixture.consumeToolResponse(cleanup, []string{`{"written":true}`})

			body := scriptedArtifactRequestBodyWithOptions(t, endpoint, fixture.tools, fixture.history, 0, false)
			writer := &cancelingHotPathResponseWriter{}
			path := "/v1/chat/completions"
			if endpoint == "anthropic" {
				path = "/v1/messages"
			}
			request := httptest.NewRequest(http.MethodPost, path, strings.NewReader(string(body)))
			if endpoint == "anthropic" {
				request.Header.Set(anthropicVersionHeader, anthropicSupportedVersion)
			}
			fixture.server.routes().ServeHTTP(writer, request)
			if writer.writes == 0 {
				t.Fatal("caller-write fixture did not exercise ResponseWriter.Write")
			}

			projections := recorder.snapshot()
			requestID := firstDispatchRequestID(projections)
			want := hotPathPassTrace()
			want[len(want)-1].Disposition = hotPathTerminalDispositionCallerCancel
			assertHotPathTraceEqual(t, projectHotPathTrace(projections, requestID), want)
		})
	}
}

func TestHotPathObservationLifecycle_DispatchRejectionRecordsReason(t *testing.T) {
	// Drive a valid direct selector result into an artifact frontier that only
	// accepts the exact Plan/Review pair, then assert the rejected admission
	// carries a closed route reason and records the bounded dispatch metric.
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		t.Run(endpoint, func(t *testing.T) {
			fixture := newScriptedLightFixture(t, endpoint, false)
			edgeID := "edge-dispatch-rejection-" + endpoint
			fixture.server.SetEdgeID(edgeID)
			rec := &recordingHotPathObserver{}
			fixture.server.SetHotPathObserver(rec)
			before := hotPathMetricValue(t, "iop_hot_path_dispatch_total", map[string]string{
				"edge_id": edgeID, "hot_path_mode": "direct", "hot_path_reason": "artifact_required",
			})

			// A valid direct selector result is rejected because the retained
			// artifact frontier requires the exact Plan/Review pair. The fake
			// provider advances the already-pinned frontier before returning the
			// selector response, matching a concurrent retained-frontier update.
			fixture.service.responses[0] = func(requestID string) string {
				fixture.server.artifactFrontiers.mu.Lock()
				if record := fixture.server.artifactFrontiers.records[requestID]; record != nil {
					record.phase = artifactPhasePairReady
				}
				fixture.server.artifactFrontiers.mu.Unlock()
				return scriptedLightCompletion(endpoint, "direct selector result")
			}
			response := fixture.request()
			if response.Code == http.StatusOK {
				t.Fatalf("expected rejection response, got 200: %s", response.Body.String())
			}

			projs := rec.snapshot()
			assertProjectionsRawFree(t, projs)
			var rejection hotPathLogProjection
			for _, p := range projs {
				if p.EventClass == hotPathEventClassDispatch && p.Reason != "" {
					rejection = p
					break
				}
			}
			if rejection.EventClass != hotPathEventClassDispatch {
				t.Fatalf("no dispatch rejection observation emitted; projs=%v", projs)
			}
			if !hotPathRouteReasonIsValid(rejection.Reason) {
				t.Errorf("dispatch rejection reason=%q is not a closed value", rejection.Reason)
			}
			if len(projs) != 1 || rejection.Reason != hotPathRouteReasonArtifactReq || rejection.Mode != hotPathModeDirect {
				t.Fatalf("dispatch rejection projections=%+v, want one direct artifact_required dispatch", projs)
			}
			after := hotPathMetricValue(t, "iop_hot_path_dispatch_total", map[string]string{
				"edge_id": edgeID, "hot_path_mode": "direct", "hot_path_reason": "artifact_required",
			})
			if delta := after - before; delta != 1 {
				t.Fatalf("dispatch metric delta=%v, want 1", delta)
			}
		})
	}
}

func TestHotPathObservationLifecycle_BoundedMetricLabelsOnActualPath(t *testing.T) {
	edgeID := "edge-bounded-labels-actual"
	terminalBefore := hotPathMetricValue(t, "iop_hot_path_terminal_total", map[string]string{
		"edge_id": edgeID, "hot_path_mode": "light", "hot_path_disposition": "success",
	})
	cleanupBefore := hotPathMetricValue(t, "iop_hot_path_cleanup_total", map[string]string{
		"edge_id": edgeID, "hot_path_cleanup_outcome": "success",
	})

	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		fixture := newScriptedLightFixture(t, endpoint, false)
		fixture.server.SetEdgeID(edgeID)
		_ = driveScriptedLightPass(t, fixture)
	}

	terminalAfter := hotPathMetricValue(t, "iop_hot_path_terminal_total", map[string]string{
		"edge_id": edgeID, "hot_path_mode": "light", "hot_path_disposition": "success",
	})
	cleanupAfter := hotPathMetricValue(t, "iop_hot_path_cleanup_total", map[string]string{
		"edge_id": edgeID, "hot_path_cleanup_outcome": "success",
	})
	if delta := terminalAfter - terminalBefore; delta != 2 {
		t.Errorf("terminal metric delta=%v, want 2", delta)
	}
	if delta := cleanupAfter - cleanupBefore; delta != 2 {
		t.Errorf("cleanup metric delta=%v, want 2", delta)
	}
}
