package streamgate

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"strconv"
	"sync/atomic"
	"time"
	"unicode/utf8"
)

const maxObservationAttemptTargetValueBytes = 256

// ObservationAttemptTarget is the immutable, raw-free actual admission target
// associated with one observation. Values are bounded identifiers only; raw
// request, response, authorization, or preparer data cannot be represented.
type ObservationAttemptTarget struct {
	modelGroup    string
	model         string
	provider      string
	executionPath string
}

// NewObservationAttemptTarget constructs a validated actual attempt target.
// Each identifier is normalized so the observation contract never narrows the
// existing request identifier contract, which only requires a non-empty value:
// a value that is already valid UTF-8 within the byte bound (spaces and
// Unicode included) is preserved verbatim, while an over-bound or invalid-UTF-8
// value is deterministically reduced to a bounded, log-safe "sha256:<hex>" join
// key. An empty identifier remains the only caller contract error.
func NewObservationAttemptTarget(modelGroup, model, provider, executionPath string) (ObservationAttemptTarget, error) {
	mg, err := normalizeObservationAttemptTargetValue("model group", modelGroup)
	if err != nil {
		return ObservationAttemptTarget{}, err
	}
	m, err := normalizeObservationAttemptTargetValue("model", model)
	if err != nil {
		return ObservationAttemptTarget{}, err
	}
	p, err := normalizeObservationAttemptTargetValue("provider", provider)
	if err != nil {
		return ObservationAttemptTarget{}, err
	}
	ep, err := normalizeObservationAttemptTargetValue("execution path", executionPath)
	if err != nil {
		return ObservationAttemptTarget{}, err
	}
	target := ObservationAttemptTarget{modelGroup: mg, model: m, provider: p, executionPath: ep}
	if err := target.Validate(); err != nil {
		return ObservationAttemptTarget{}, err
	}
	return target, nil
}

// normalizeObservationAttemptTargetValue preserves any non-empty identifier
// that is valid UTF-8 within the byte bound and otherwise reduces it to a
// deterministic bounded join key. The hash keeps correlation stable while
// guaranteeing the stored value is always log-safe and length-bounded. Empty is
// the only caller contract error.
func normalizeObservationAttemptTargetValue(name, value string) (string, error) {
	if value == "" {
		return "", errors.New("streamgate: observation attempt target " + name + " is required")
	}
	if utf8.ValidString(value) && len(value) <= maxObservationAttemptTargetValueBytes {
		return value, nil
	}
	sum := sha256.Sum256([]byte(value))
	return "sha256:" + hex.EncodeToString(sum[:]), nil
}

// Validate returns nil when all target identifiers are present, valid UTF-8,
// and within the byte bound. Construction normalizes every value into this
// invariant, so Validate only re-checks the invariant rather than any
// character allowlist.
func (t ObservationAttemptTarget) Validate() error {
	for _, item := range []struct{ name, value string }{
		{"model group", t.modelGroup}, {"model", t.model},
		{"provider", t.provider}, {"execution path", t.executionPath},
	} {
		if item.value == "" {
			return errors.New("streamgate: observation attempt target " + item.name + " is required")
		}
		if len(item.value) > maxObservationAttemptTargetValueBytes || !utf8.ValidString(item.value) {
			return errors.New("streamgate: observation attempt target " + item.name + " is invalid or exceeds maximum length")
		}
	}
	return nil
}

func (t ObservationAttemptTarget) ModelGroup() string    { return t.modelGroup }
func (t ObservationAttemptTarget) Model() string         { return t.model }
func (t ObservationAttemptTarget) Provider() string      { return t.provider }
func (t ObservationAttemptTarget) ExecutionPath() string { return t.executionPath }

// ObservationKind identifies which filter evaluation or coordinator lifecycle
// moment a FilterObservation records. The set is closed: Core owns the
// lifecycle vocabulary and consumers cannot introduce new kinds.
type ObservationKind string

const (
	// ObservationKindFilterEvaluated records a single filter's evaluation
	// outcome for one epoch (evaluated, evaluation_error, not_applicable, or
	// deferred).
	ObservationKindFilterEvaluated ObservationKind = "filter_evaluated"

	// ObservationKindResponseStaged records that a response-start was staged
	// pending first safe release.
	ObservationKindResponseStaged ObservationKind = "response_staged"

	// ObservationKindArbitrationDecided records the Arbiter's single
	// synthesized action for an epoch.
	ObservationKindArbitrationDecided ObservationKind = "arbitration_decided"

	// ObservationKindReleaseCommitted records that a safe prefix or
	// replacement was committed to the downstream sink.
	ObservationKindReleaseCommitted ObservationKind = "release_committed"

	// ObservationKindTerminalCommitted records that the single terminal
	// sequence (success or error) was committed to the downstream sink.
	ObservationKindTerminalCommitted ObservationKind = "terminal_committed"

	// ObservationKindRecoveryPlanSelected records that the RecoveryPlan
	// Coordinator selected a RecoveryPlan.
	ObservationKindRecoveryPlanSelected ObservationKind = "recovery_plan_selected"

	// ObservationKindRecoveryAttemptAborted records that current attempt
	// ownership was closed ahead of a recovery cycle.
	ObservationKindRecoveryAttemptAborted ObservationKind = "recovery_attempt_aborted"

	// ObservationKindRecoveryPrepared records the outcome of an optional
	// RecoveryPlanPreparer invocation.
	ObservationKindRecoveryPrepared ObservationKind = "recovery_prepared"

	// ObservationKindRecoveryRebuilt records that a RequestRebuilder produced
	// a rebuilt request for the selected plan.
	ObservationKindRecoveryRebuilt ObservationKind = "recovery_rebuilt"

	// ObservationKindRecoveryDispatched records that a rebuilt request was
	// dispatched for re-admission.
	ObservationKindRecoveryDispatched ObservationKind = "recovery_dispatched"

	// ObservationKindRecoveryPlanRejected records a recovery plan that could
	// not be materialized. It carries a bounded sanitized cause chain and no
	// RecoveryInfo because no valid plan exists to identify.
	ObservationKindRecoveryPlanRejected ObservationKind = "recovery_plan_rejected"

	// ObservationKindRecoveryAttemptAbortFailed records a failure while closing
	// the previous provider attempt before recovery dispatch.
	ObservationKindRecoveryAttemptAbortFailed ObservationKind = "recovery_attempt_abort_failed"

	// ObservationKindRecoveryRebuildFailed records a failure while rebuilding
	// the selected recovery request.
	ObservationKindRecoveryRebuildFailed ObservationKind = "recovery_rebuild_failed"

	// ObservationKindRecoveryDispatchFailed records a failed or invalid
	// recovery dispatch after a selected plan was rebuilt.

	ObservationKindRecoveryDispatchFailed ObservationKind = "recovery_dispatch_failed"
	// ObservationKindFilterEvaluationStarted records that Core began a
	// parallel filter evaluation pass for an epoch. It requires attribution
	// and a non-zero epoch id.
	ObservationKindFilterEvaluationStarted ObservationKind = "filter_evaluation_started"
)

var knownObservationKinds = map[ObservationKind]struct{}{
	ObservationKindFilterEvaluated:            {},
	ObservationKindResponseStaged:             {},
	ObservationKindArbitrationDecided:         {},
	ObservationKindReleaseCommitted:           {},
	ObservationKindTerminalCommitted:          {},
	ObservationKindRecoveryPlanSelected:       {},
	ObservationKindRecoveryAttemptAborted:     {},
	ObservationKindRecoveryPrepared:           {},
	ObservationKindRecoveryRebuilt:            {},
	ObservationKindRecoveryDispatched:         {},
	ObservationKindFilterEvaluationStarted:    {},
	ObservationKindRecoveryPlanRejected:       {},
	ObservationKindRecoveryAttemptAbortFailed: {},
	ObservationKindRecoveryRebuildFailed:      {},
	ObservationKindRecoveryDispatchFailed:     {},
}

// Validate returns nil when the ObservationKind is a known lifecycle value.
func (k ObservationKind) Validate() error {
	if _, ok := knownObservationKinds[k]; ok {
		return nil
	}
	return errors.New("streamgate: unknown observation kind: " + string(k))
}

// ObservationDeadlineOutcome is a closed, sanitized descriptor of how a
// bounded-deadline coordinator operation (currently RecoveryPlanPreparer)
// concluded. It never carries raw error text.
type ObservationDeadlineOutcome string

const (
	// ObservationDeadlineOutcomeCompleted indicates the operation returned
	// successfully within its deadline.
	ObservationDeadlineOutcomeCompleted ObservationDeadlineOutcome = "completed"

	// ObservationDeadlineOutcomeTimeout indicates the operation did not
	// complete before its bounded deadline elapsed.
	ObservationDeadlineOutcomeTimeout ObservationDeadlineOutcome = "timeout"

	// ObservationDeadlineOutcomeFailed indicates the operation returned an
	// error before its deadline elapsed.
	ObservationDeadlineOutcomeFailed ObservationDeadlineOutcome = "failed"
)

var knownObservationDeadlineOutcomes = map[ObservationDeadlineOutcome]struct{}{
	ObservationDeadlineOutcomeCompleted: {},
	ObservationDeadlineOutcomeTimeout:   {},
	ObservationDeadlineOutcomeFailed:    {},
}

// Validate returns nil when the ObservationDeadlineOutcome is a known value.
func (o ObservationDeadlineOutcome) Validate() error {
	if _, ok := knownObservationDeadlineOutcomes[o]; ok {
		return nil
	}
	return errors.New("streamgate: unknown observation deadline outcome: " + string(o))
}

// ObservationAttribution identifies the stable consumer/filter/rule a
// FilterObservation is attributed to. It reuses the same bounded StableToken
// grammar as FilterDecision and RecoveryContributor; no raw content is
// carried.
type ObservationAttribution struct {
	consumerID StableToken
	filterID   StableToken
	ruleID     StableToken
}

// NewObservationAttribution creates a validated ObservationAttribution.
func NewObservationAttribution(consumerID, filterID, ruleID string) (ObservationAttribution, error) {
	c, err := NewStableTokenRequired("consumerID", consumerID)
	if err != nil {
		return ObservationAttribution{}, err
	}
	f, err := NewStableTokenRequired("filterID", filterID)
	if err != nil {
		return ObservationAttribution{}, err
	}
	r, err := NewStableTokenRequired("ruleID", ruleID)
	if err != nil {
		return ObservationAttribution{}, err
	}
	return ObservationAttribution{consumerID: c, filterID: f, ruleID: r}, nil
}

// Validate returns nil when the ObservationAttribution is in a consistent
// state.
func (a ObservationAttribution) Validate() error {
	if err := validateStableTokenRequired("consumerID", a.consumerID.value); err != nil {
		return err
	}
	if err := validateStableTokenRequired("filterID", a.filterID.value); err != nil {
		return err
	}
	return validateStableTokenRequired("ruleID", a.ruleID.value)
}

// ConsumerID returns the stable consumer id token.
func (a ObservationAttribution) ConsumerID() string { return a.consumerID.value }

// FilterID returns the stable filter id token.
func (a ObservationAttribution) FilterID() string { return a.filterID.value }

// RuleID returns the stable rule id token.
func (a ObservationAttribution) RuleID() string { return a.ruleID.value }

// ObservationHoldInfo carries the hold-mode/event-kind/rune accounting for a
// filter evaluation or response-staging observation. Rune counts are bounded,
// non-negative counters only; no buffered text is retained.
type ObservationHoldInfo struct {
	mode           FilterHoldMode
	eventKind      EventKind
	effectiveRunes int
	pendingRunes   int
}

// NewObservationHoldInfo creates a validated ObservationHoldInfo.
func NewObservationHoldInfo(mode FilterHoldMode, eventKind EventKind, effectiveRunes, pendingRunes int) (ObservationHoldInfo, error) {
	info := ObservationHoldInfo{
		mode:           mode,
		eventKind:      eventKind,
		effectiveRunes: effectiveRunes,
		pendingRunes:   pendingRunes,
	}
	if err := info.Validate(); err != nil {
		return ObservationHoldInfo{}, err
	}
	return info, nil
}

// Validate returns nil when the ObservationHoldInfo is in a consistent state.
func (h ObservationHoldInfo) Validate() error {
	if err := h.mode.Validate(); err != nil {
		return err
	}
	if err := h.eventKind.Validate(); err != nil {
		return err
	}
	if h.effectiveRunes < 0 {
		return errors.New("streamgate: observation hold info effective runes must be non-negative")
	}
	if h.pendingRunes < 0 {
		return errors.New("streamgate: observation hold info pending runes must be non-negative")
	}
	return nil
}

// Mode returns the filter hold mode.
func (h ObservationHoldInfo) Mode() FilterHoldMode { return h.mode }

// EventKind returns the subscribed event kind.
func (h ObservationHoldInfo) EventKind() EventKind { return h.eventKind }

// EffectiveRunes returns the policy evidence window size in Unicode runes.
func (h ObservationHoldInfo) EffectiveRunes() int { return h.effectiveRunes }

// PendingRunes returns the current pending tail size in Unicode runes.
func (h ObservationHoldInfo) PendingRunes() int { return h.pendingRunes }

// ObservationDecisionPolicy carries a filter's evaluation outcome kind, the
// decision kind when evaluated, and its enforcement mode. The failure
// disposition is derived canonically from the outcome and enforcement so the
// observation contract cannot drift from the epoch filter outcome truth
// table. Raw error text is never carried; only the closed enums already used
// by FilterOutcome and EpochFilterOutcome are reused.
type ObservationDecisionPolicy struct {
	outcome      FilterOutcomeKind
	decisionKind FilterDecisionKind
	enforcement  FilterEnforcement
}

// NewObservationDecisionPolicy creates a validated ObservationDecisionPolicy.
// decisionKind must be empty unless outcome is FilterOutcomeKindEvaluated.
// The failure disposition is derived canonically and cannot be overridden.
func NewObservationDecisionPolicy(
	outcome FilterOutcomeKind,
	decisionKind FilterDecisionKind,
	enforcement FilterEnforcement,
) (ObservationDecisionPolicy, error) {
	policy := ObservationDecisionPolicy{
		outcome:      outcome,
		decisionKind: decisionKind,
		enforcement:  enforcement,
	}
	if err := policy.Validate(); err != nil {
		return ObservationDecisionPolicy{}, err
	}
	return policy, nil
}

// Validate returns nil when the ObservationDecisionPolicy is in a consistent
// state. The failure disposition is derived canonically from the outcome and
// enforcement and must match the epoch filter outcome truth table.
func (p ObservationDecisionPolicy) Validate() error {
	if err := p.outcome.Validate(); err != nil {
		return err
	}
	if p.outcome == FilterOutcomeKindEvaluated {
		if _, ok := knownFilterDecisionKinds[p.decisionKind]; !ok {
			return errors.New("streamgate: observation decision policy requires a known decision kind when evaluated")
		}
	} else if p.decisionKind != "" {
		return errors.New("streamgate: observation decision policy must not carry a decision kind unless evaluated")
	}
	if err := p.enforcement.Validate(); err != nil {
		return err
	}
	expected := deriveObservationDecisionPolicyDisposition(p.outcome, p.decisionKind, p.enforcement)
	if expected == "" {
		return errors.New("streamgate: observation decision policy cannot derive a canonical failure disposition")
	}
	return nil
}

// deriveObservationDecisionPolicyDisposition derives the canonical failure
// disposition from the outcome, decision kind, and enforcement mode. The
// rules mirror deriveEvaluationFailureDisposition so the observation
// contract stays aligned with the epoch filter outcome truth table.
func deriveObservationDecisionPolicyDisposition(
	outcome FilterOutcomeKind,
	decisionKind FilterDecisionKind,
	enforcement FilterEnforcement,
) EvaluationFailureDisposition {
	switch outcome {
	case FilterOutcomeKindEvaluated:
		switch decisionKind {
		case FilterDecisionKindPass, FilterDecisionKindObserve, FilterDecisionKindReplacement:
			return EvaluationFailureDispositionNone
		case FilterDecisionKindViolation, FilterDecisionKindFatal:
			if enforcement == FilterEnforcementBlocking {
				return EvaluationFailureDispositionBlockingFatal
			}
			return EvaluationFailureDispositionObserveError
		default:
			return EvaluationFailureDispositionNone
		}
	case FilterOutcomeKindEvaluationError:
		if enforcement == FilterEnforcementBlocking {
			return EvaluationFailureDispositionBlockingFatal
		}
		return EvaluationFailureDispositionObserveError
	case FilterOutcomeKindNotApplicableForEpoch, FilterOutcomeKindDeferredByRequirement:
		return EvaluationFailureDispositionNone
	default:
		return ""
	}
}

// Outcome returns the filter outcome kind.
func (p ObservationDecisionPolicy) Outcome() FilterOutcomeKind { return p.outcome }

// DecisionKind returns the filter decision kind, or "" if not evaluated.
func (p ObservationDecisionPolicy) DecisionKind() FilterDecisionKind { return p.decisionKind }

// Enforcement returns the filter enforcement mode.
func (p ObservationDecisionPolicy) Enforcement() FilterEnforcement { return p.enforcement }

// FailureDisposition returns the canonical evaluation failure disposition
// derived from the outcome, decision kind, and enforcement mode. It cannot
// be overridden by the caller.
func (p ObservationDecisionPolicy) FailureDisposition() EvaluationFailureDisposition {
	return deriveObservationDecisionPolicyDisposition(p.outcome, p.decisionKind, p.enforcement)
}

// ObservationRecoveryInfo carries the selected RecoveryPlan's stable
// identity, strategy, and resume mode. The shared attempt id is set only for
// continue_stream plans and identifies the attempt the plan continues rather
// than replaces.
type ObservationRecoveryInfo struct {
	planID          StableToken
	strategy        RecoveryStrategy
	resumeMode      RecoveryResumeMode
	sharedAttemptID StableToken
}

// NewObservationRecoveryInfo creates a validated ObservationRecoveryInfo.
// sharedAttemptID may be empty for replace_attempt plans and must be set for
// continue_stream plans.
func NewObservationRecoveryInfo(planID string, strategy RecoveryStrategy, resumeMode RecoveryResumeMode, sharedAttemptID string) (ObservationRecoveryInfo, error) {
	p, err := NewStableTokenRequired("planID", planID)
	if err != nil {
		return ObservationRecoveryInfo{}, err
	}
	shared, err := NewStableTokenOptional("sharedAttemptID", sharedAttemptID)
	if err != nil {
		return ObservationRecoveryInfo{}, err
	}
	info := ObservationRecoveryInfo{
		planID:          p,
		strategy:        strategy,
		resumeMode:      resumeMode,
		sharedAttemptID: shared,
	}
	if err := info.Validate(); err != nil {
		return ObservationRecoveryInfo{}, err
	}
	return info, nil
}

// Validate returns nil when the ObservationRecoveryInfo is in a consistent
// state.
func (r ObservationRecoveryInfo) Validate() error {
	if err := validateStableTokenRequired("planID", r.planID.value); err != nil {
		return err
	}
	if _, ok := knownRecoveryStrategies[r.strategy]; !ok {
		return errors.New("streamgate: observation recovery info unknown strategy")
	}
	if err := r.resumeMode.Validate(); err != nil {
		return err
	}
	// Enforce the canonical strategy→resume-mode matrix matching RecoveryPlan.
	// exact/schema recovery must pair with replace_attempt; continuation/managed
	// recovery must pair with continue_stream.
	switch r.strategy {
	case RecoveryStrategyExactReplay, RecoveryStrategySchemaRepair:
		if r.resumeMode != RecoveryResumeModeReplaceAttempt {
			return errors.New("streamgate: observation recovery info exact/schema recovery requires replace_attempt resume mode")
		}
	case RecoveryStrategyContinuationRepair, RecoveryStrategyManagedContinuation:
		if r.resumeMode != RecoveryResumeModeContinueStream {
			return errors.New("streamgate: observation recovery info continuation/managed recovery requires continue_stream resume mode")
		}
	}
	if err := validateStableTokenIfSet("sharedAttemptID", r.sharedAttemptID.value); err != nil {
		return err
	}
	if r.resumeMode == RecoveryResumeModeContinueStream && r.sharedAttemptID.value == "" {
		return errors.New("streamgate: observation recovery info requires a shared attempt id for continue_stream")
	}
	if r.resumeMode == RecoveryResumeModeReplaceAttempt && r.sharedAttemptID.value != "" {
		return errors.New("streamgate: observation recovery info must not carry a shared attempt id for replace_attempt")
	}
	return nil
}

// PlanID returns the stable recovery plan id token.
func (r ObservationRecoveryInfo) PlanID() string { return r.planID.value }

// Strategy returns the recovery strategy.
func (r ObservationRecoveryInfo) Strategy() RecoveryStrategy { return r.strategy }

// ResumeMode returns the recovery resume mode.
func (r ObservationRecoveryInfo) ResumeMode() RecoveryResumeMode { return r.resumeMode }

// SharedAttemptID returns the continued attempt id, or "" for replace_attempt.
func (r ObservationRecoveryInfo) SharedAttemptID() string { return r.sharedAttemptID.value }

// ObservationArbitrationInfo carries the Arbiter's single synthesized action
// and the base event disposition for an epoch. It is immutable and required
// for every arbitration_decided observation so consumers can reconstruct the
// full decision without ambiguity.
type ObservationArbitrationInfo struct {
	action          ArbitrationAction
	baseDisposition BaseEventDisposition
}

// NewObservationArbitrationInfo creates a validated ObservationArbitrationInfo.
func NewObservationArbitrationInfo(action ArbitrationAction, baseDisp BaseEventDisposition) (ObservationArbitrationInfo, error) {
	if err := action.Validate(); err != nil {
		return ObservationArbitrationInfo{}, errors.New("streamgate: observation arbitration info action: " + err.Error())
	}
	if err := baseDisp.Validate(); err != nil {
		return ObservationArbitrationInfo{}, errors.New("streamgate: observation arbitration info base disposition: " + err.Error())
	}
	info := ObservationArbitrationInfo{
		action:          action,
		baseDisposition: baseDisp,
	}
	if err := info.Validate(); err != nil {
		return ObservationArbitrationInfo{}, err
	}
	return info, nil
}

// Validate returns nil when the ObservationArbitrationInfo is in a consistent
// state.
func (a ObservationArbitrationInfo) Validate() error {
	if err := a.action.Validate(); err != nil {
		return err
	}
	return a.baseDisposition.Validate()
}

// Action returns the arbitration action.
func (a ObservationArbitrationInfo) Action() ArbitrationAction { return a.action }

// BaseDisposition returns the base event disposition.
func (a ObservationArbitrationInfo) BaseDisposition() BaseEventDisposition { return a.baseDisposition }

// cloneObservationArbitrationInfo returns a defensive copy of the arbitration info.
func cloneObservationArbitrationInfo(a *ObservationArbitrationInfo) *ObservationArbitrationInfo {
	if a == nil {
		return nil
	}
	cp := *a
	return &cp
}

// ObservationPreparerInfo carries the optional RecoveryPlanPreparer's stable
// identity, preparation status, and bounded deadline outcome. Preparer
// input/output is never carried.
type ObservationPreparerInfo struct {
	preparerID      StableToken
	status          RecoveryPreparationStatus
	deadlineOutcome ObservationDeadlineOutcome
}

// NewObservationPreparerInfo creates a validated ObservationPreparerInfo.
func NewObservationPreparerInfo(preparerID string, status RecoveryPreparationStatus, deadlineOutcome ObservationDeadlineOutcome) (ObservationPreparerInfo, error) {
	id, err := NewStableTokenRequired("preparerID", preparerID)
	if err != nil {
		return ObservationPreparerInfo{}, err
	}
	info := ObservationPreparerInfo{
		preparerID:      id,
		status:          status,
		deadlineOutcome: deadlineOutcome,
	}
	if err := info.Validate(); err != nil {
		return ObservationPreparerInfo{}, err
	}
	return info, nil
}

// Validate returns nil when the ObservationPreparerInfo is in a consistent
// state.
func (p ObservationPreparerInfo) Validate() error {
	if err := validateStableTokenRequired("preparerID", p.preparerID.value); err != nil {
		return err
	}
	if err := p.status.Validate(); err != nil {
		return err
	}
	return p.deadlineOutcome.Validate()
}

// PreparerID returns the stable preparer id token.
func (p ObservationPreparerInfo) PreparerID() string { return p.preparerID.value }

// Status returns the recovery preparation status.
func (p ObservationPreparerInfo) Status() RecoveryPreparationStatus { return p.status }

// DeadlineOutcome returns the bounded deadline outcome.
func (p ObservationPreparerInfo) DeadlineOutcome() ObservationDeadlineOutcome {
	return p.deadlineOutcome
}

// FilterObservationInput is the mutable call-site input to
// newFilterObservation. Every field is validated and defensively copied into
// the returned immutable FilterObservation. Sequence is deliberately absent:
// it is assigned only by an ObservationSequencer so callers cannot forge
// ordering.
type FilterObservationInput struct {
	Kind              ObservationKind
	StableCorrelation string
	ConfigGeneration  string
	AttemptID         string
	AttemptTarget     ObservationAttemptTarget
	EpochID           uint64
	CommitState       CommitState
	Attribution       *ObservationAttribution
	Hold              *ObservationHoldInfo
	DecisionPolicy    *ObservationDecisionPolicy
	Arbitration       *ObservationArbitrationInfo
	TerminalReason    TerminalReason
	Recovery          *ObservationRecoveryInfo
	Preparer          *ObservationPreparerInfo
	Causes            FailureCauseChain
	Evidence          *SanitizedEvidence
	OccurredAt        time.Time
}

// FilterObservation is an immutable, raw-free timeline event emitted by Core
// for each filter evaluation or lifecycle/coordinator decision. It carries
// existing request/run/provider/model correlation, config generation,
// attempt/epoch, stable consumer/filter/rule identity, hold/decision/commit
// state, terminal/recovery/preparer summary, a bounded sanitized cause chain,
// and optional consumer-provided sanitized evidence. Storage and retention
// belong to the existing observability sink; this type never carries
// preparer input/output or raw output/prompt/tool args/result/auth content —
// there is no field for any of it.
type FilterObservation struct {
	sequence          uint64
	kind              ObservationKind
	stableCorrelation StableToken
	configGeneration  StableToken
	attemptID         StableToken
	attemptTarget     ObservationAttemptTarget
	epochID           uint64
	commitState       CommitState
	attribution       *ObservationAttribution
	hold              *ObservationHoldInfo
	decisionPolicy    *ObservationDecisionPolicy
	arbitration       *ObservationArbitrationInfo
	terminalReason    TerminalReason
	recovery          *ObservationRecoveryInfo
	preparer          *ObservationPreparerInfo
	causes            FailureCauseChain
	evidence          *SanitizedEvidence
	occurredAt        time.Time
}

// newFilterObservation validates the input and an explicit, caller-assigned
// sequence number and returns an immutable FilterObservation. Only
// ObservationSequencer.Emit calls this so sequence assignment stays
// request-local and monotonic.
func newFilterObservation(sequence uint64, input FilterObservationInput) (FilterObservation, error) {
	if sequence == 0 {
		return FilterObservation{}, errors.New("streamgate: filter observation sequence must be strictly positive")
	}
	correlation, err := NewStableTokenRequired("stableCorrelation", input.StableCorrelation)
	if err != nil {
		return FilterObservation{}, err
	}
	generation, err := NewStableTokenRequired("configGeneration", input.ConfigGeneration)
	if err != nil {
		return FilterObservation{}, err
	}
	attemptID, err := NewStableTokenRequired("attemptID", input.AttemptID)
	if err != nil {
		return FilterObservation{}, err
	}
	if err := input.AttemptTarget.Validate(); err != nil {
		return FilterObservation{}, err
	}
	if input.OccurredAt.IsZero() {
		return FilterObservation{}, errors.New("streamgate: filter observation occurred timestamp is required")
	}

	obs := FilterObservation{
		sequence:          sequence,
		kind:              input.Kind,
		stableCorrelation: correlation,
		configGeneration:  generation,
		attemptID:         attemptID,
		attemptTarget:     input.AttemptTarget,
		epochID:           input.EpochID,
		commitState:       input.CommitState,
		attribution:       cloneObservationAttribution(input.Attribution),
		hold:              cloneObservationHoldInfo(input.Hold),
		decisionPolicy:    cloneObservationDecisionPolicy(input.DecisionPolicy),
		arbitration:       cloneObservationArbitrationInfo(input.Arbitration),
		terminalReason:    input.TerminalReason,
		recovery:          cloneObservationRecoveryInfo(input.Recovery),
		preparer:          cloneObservationPreparerInfo(input.Preparer),
		causes:            input.Causes.Copy(),
		evidence:          cloneSanitizedEvidence(input.Evidence),
		occurredAt:        input.OccurredAt,
	}
	if err := obs.Validate(); err != nil {
		return FilterObservation{}, err
	}
	return obs, nil
}

// Validate returns nil when the FilterObservation is in a consistent state.
// Each present nested field is re-validated against its own grammar, and
// kind-specific required fields are enforced so the timeline stays a closed,
// immutable contract.
func (o FilterObservation) Validate() error {
	if o.sequence == 0 {
		return errors.New("streamgate: filter observation sequence must be strictly positive")
	}
	if err := o.kind.Validate(); err != nil {
		return err
	}
	if err := validateStableTokenRequired("stableCorrelation", o.stableCorrelation.value); err != nil {
		return err
	}
	if err := validateStableTokenRequired("configGeneration", o.configGeneration.value); err != nil {
		return err
	}
	if err := validateStableTokenRequired("attemptID", o.attemptID.value); err != nil {
		return err
	}
	if err := o.attemptTarget.Validate(); err != nil {
		return err
	}
	if err := o.commitState.Validate(); err != nil {
		return err
	}
	if o.attribution != nil {
		if err := o.attribution.Validate(); err != nil {
			return err
		}
	}
	if o.hold != nil {
		if err := o.hold.Validate(); err != nil {
			return err
		}
	}
	if o.decisionPolicy != nil {
		if err := o.decisionPolicy.Validate(); err != nil {
			return err
		}
	}
	if o.terminalReason != "" {
		if err := o.terminalReason.Validate(); err != nil {
			return err
		}
	}
	if o.recovery != nil {
		if err := o.recovery.Validate(); err != nil {
			return err
		}
	}
	if o.preparer != nil {
		if err := o.preparer.Validate(); err != nil {
			return err
		}
	}
	if o.causes.Len() > MaxFailureCauses {
		return errors.New("streamgate: filter observation cause chain exceeds maximum")
	}
	for i := 0; i < o.causes.Len(); i++ {
		c, err := o.causes.At(i)
		if err != nil {
			return errors.New("streamgate: filter observation cause chain: " + err.Error())
		}
		if err := c.Validate(); err != nil {
			return errors.New("streamgate: filter observation cause chain entry " + strconv.Itoa(i) + ": " + err.Error())
		}
	}
	if o.evidence != nil {
		if err := o.evidence.Validate(); err != nil {
			return err
		}
	}
	if o.occurredAt.IsZero() {
		return errors.New("streamgate: filter observation occurred timestamp is required")
	}

	switch o.kind {
	case ObservationKindFilterEvaluationStarted:
		if o.attribution == nil {
			return errors.New("streamgate: filter_evaluation_started observation requires attribution")
		}
		if o.epochID == 0 {
			return errors.New("streamgate: filter_evaluation_started observation requires a non-zero epoch id")
		}
	case ObservationKindFilterEvaluated:
		if o.attribution == nil {
			return errors.New("streamgate: filter_evaluated observation requires attribution")
		}
		if o.decisionPolicy == nil {
			return errors.New("streamgate: filter_evaluated observation requires a decision policy")
		}
		if o.epochID == 0 {
			return errors.New("streamgate: filter_evaluated observation requires a non-zero epoch id")
		}
	case ObservationKindArbitrationDecided:
		if o.arbitration == nil {
			return errors.New("streamgate: arbitration_decided observation requires an arbitration summary")
		}
		if o.epochID == 0 {
			return errors.New("streamgate: arbitration_decided observation requires a non-zero epoch id")
		}
	case ObservationKindResponseStaged:
		if o.hold == nil {
			return errors.New("streamgate: response_staged observation requires hold info")
		}
		if o.commitState != CommitStateTransportUncommitted {
			return errors.New("streamgate: response_staged observation requires transport_uncommitted commit state")
		}
	case ObservationKindReleaseCommitted:
		if o.commitState != CommitStateStreamOpen {
			return errors.New("streamgate: release_committed observation requires stream_open commit state")
		}
	case ObservationKindTerminalCommitted:
		if o.terminalReason == "" && o.causes.Len() == 0 {
			return errors.New("streamgate: terminal_committed observation requires a terminal reason or a cause chain")
		}
		if o.terminalReason != "" && o.causes.Len() > 0 {
			return errors.New("streamgate: terminal_committed observation must not carry both a terminal reason and a cause chain")
		}
		if o.commitState != CommitStateTerminalCommitted {
			return errors.New("streamgate: terminal_committed observation requires terminal_committed commit state")
		}
	case ObservationKindRecoveryPlanSelected, ObservationKindRecoveryAttemptAborted,
		ObservationKindRecoveryRebuilt, ObservationKindRecoveryDispatched:
		if o.recovery == nil {
			return errors.New("streamgate: recovery observation requires recovery info")
		}
		if err := validateObservationRecoveryCommitState(o.recovery.resumeMode, o.commitState); err != nil {
			return err
		}
	case ObservationKindRecoveryPlanRejected:
		if o.recovery != nil {
			return errors.New("streamgate: recovery_plan_rejected observation must not carry recovery info")
		}
		if o.causes.Len() == 0 {
			return errors.New("streamgate: recovery_plan_rejected observation requires a cause chain")
		}
	case ObservationKindRecoveryAttemptAbortFailed,
		ObservationKindRecoveryRebuildFailed,
		ObservationKindRecoveryDispatchFailed:
		if o.recovery == nil {
			return errors.New("streamgate: recovery failure observation requires recovery info")
		}
		if o.causes.Len() == 0 {
			return errors.New("streamgate: recovery failure observation requires a cause chain")
		}
		if err := validateObservationRecoveryCommitState(o.recovery.resumeMode, o.commitState); err != nil {
			return err
		}
	case ObservationKindRecoveryPrepared:
		if o.recovery == nil {
			return errors.New("streamgate: recovery_prepared observation requires recovery info")
		}
		if o.preparer == nil {
			return errors.New("streamgate: recovery_prepared observation requires preparer info")
		}
		if err := validateObservationRecoveryCommitState(o.recovery.resumeMode, o.commitState); err != nil {
			return err
		}
	}
	return nil
}

// Sequence returns the request-local monotonic sequence number.
func (o FilterObservation) Sequence() uint64 { return o.sequence }

// Kind returns the observation kind.
func (o FilterObservation) Kind() ObservationKind { return o.kind }

// StableCorrelation returns the existing request/run/provider/model
// correlation reference this observation was emitted under.
func (o FilterObservation) StableCorrelation() string { return o.stableCorrelation.value }

// ConfigGeneration returns the request-start config generation.
func (o FilterObservation) ConfigGeneration() string { return o.configGeneration.value }

// AttemptID returns the attempt identifier.
func (o FilterObservation) AttemptID() string { return o.attemptID.value }

// AttemptTarget returns the immutable actual target bound to this observation.
func (o FilterObservation) AttemptTarget() ObservationAttemptTarget { return o.attemptTarget }

// EpochID returns the epoch id, or 0 if this observation is not epoch-bound.
func (o FilterObservation) EpochID() uint64 { return o.epochID }

// CommitState returns the commit state at observation time.
func (o FilterObservation) CommitState() CommitState { return o.commitState }

// Attribution returns a defensive copy of the stable consumer/filter/rule
// attribution, or nil if not set.
func (o FilterObservation) Attribution() *ObservationAttribution {
	return cloneObservationAttribution(o.attribution)
}

// Hold returns a defensive copy of the hold info, or nil if not set.
func (o FilterObservation) Hold() *ObservationHoldInfo {
	return cloneObservationHoldInfo(o.hold)
}

// DecisionPolicy returns a defensive copy of the decision policy, or nil if
// not set.
func (o FilterObservation) DecisionPolicy() *ObservationDecisionPolicy {
	return cloneObservationDecisionPolicy(o.decisionPolicy)
}

// Arbitration returns a defensive copy of the arbitration summary, or nil if
// not set.
func (o FilterObservation) Arbitration() *ObservationArbitrationInfo {
	return cloneObservationArbitrationInfo(o.arbitration)
}

// TerminalReason returns the normalized terminal reason, or "" if not set.
func (o FilterObservation) TerminalReason() TerminalReason { return o.terminalReason }

// Recovery returns a defensive copy of the recovery info, or nil if not set.
func (o FilterObservation) Recovery() *ObservationRecoveryInfo {
	return cloneObservationRecoveryInfo(o.recovery)
}

// Preparer returns a defensive copy of the preparer info, or nil if not set.
func (o FilterObservation) Preparer() *ObservationPreparerInfo {
	return cloneObservationPreparerInfo(o.preparer)
}

// Causes returns a defensive copy of the bounded sanitized cause chain.
func (o FilterObservation) Causes() FailureCauseChain { return o.causes.Copy() }

// Evidence returns a defensive copy of the consumer-provided sanitized
// evidence, or nil if not set.
func (o FilterObservation) Evidence() *SanitizedEvidence {
	if o.evidence == nil {
		return nil
	}
	cp := *o.evidence
	return &cp
}

// OccurredAt returns the observation timestamp.
func (o FilterObservation) OccurredAt() time.Time { return o.occurredAt }

func cloneObservationAttribution(a *ObservationAttribution) *ObservationAttribution {
	if a == nil {
		return nil
	}
	cp := *a
	return &cp
}

func cloneObservationHoldInfo(h *ObservationHoldInfo) *ObservationHoldInfo {
	if h == nil {
		return nil
	}
	cp := *h
	return &cp
}

func cloneObservationDecisionPolicy(p *ObservationDecisionPolicy) *ObservationDecisionPolicy {
	if p == nil {
		return nil
	}
	cp := *p
	return &cp
}

func cloneObservationRecoveryInfo(r *ObservationRecoveryInfo) *ObservationRecoveryInfo {
	if r == nil {
		return nil
	}
	cp := *r
	return &cp
}

func cloneObservationPreparerInfo(p *ObservationPreparerInfo) *ObservationPreparerInfo {
	if p == nil {
		return nil
	}
	cp := *p
	return &cp
}

func cloneSanitizedEvidence(e *SanitizedEvidence) *SanitizedEvidence {
	if e == nil {
		return nil
	}
	cp := *e
	return &cp
}

// validateObservationRecoveryCommitState enforces the allowed matrix between
// recovery resume mode and observation commit state. replace_attempt plans
// require transport_uncommitted; continue_stream plans require stream_open.
func validateObservationRecoveryCommitState(mode RecoveryResumeMode, cs CommitState) error {
	switch mode {
	case RecoveryResumeModeReplaceAttempt:
		if cs != CommitStateTransportUncommitted {
			return errors.New("streamgate: replace_attempt recovery observation requires transport_uncommitted commit state")
		}
	case RecoveryResumeModeContinueStream:
		if cs != CommitStateStreamOpen {
			return errors.New("streamgate: continue_stream recovery observation requires stream_open commit state")
		}
	}
	return nil
}

// ObservationSink receives immutable FilterObservation events. Implementations
// own storage and retention; Core only owns the envelope and delivery
// attempt. Emit should not block indefinitely — sinks that need bounded work
// should apply their own timeout internally.
type ObservationSink interface {
	Emit(ctx context.Context, obs FilterObservation) error
}

// NoopObservationSink discards every observation. It is the default sink for
// hosts that have not wired an observability backend yet.
type NoopObservationSink struct{}

// Emit discards the observation and always returns nil.
func (NoopObservationSink) Emit(ctx context.Context, obs FilterObservation) error { return nil }

// ObservationSequencer assigns strictly increasing, request-local sequence
// numbers and forwards each resulting FilterObservation to a sink. Sequence
// assignment is safe for concurrent use so parallel filter evaluation can
// emit without racing.
//
// Sink delivery uses best-effort failure isolation: a sink error is reported
// to the optional onError hook (if set) and otherwise dropped. Emit only
// returns a non-nil error when the observation itself fails to construct,
// which indicates a caller bug rather than an observability outage. This
// guarantees an observability sink failure can never alter release, terminal,
// or recovery behavior.
type ObservationSequencer struct {
	sink    ObservationSink
	onError func(FilterObservation, error)
	counter uint64
}

// NewObservationSequencer creates a sequencer bound to the given sink. A nil
// sink defaults to NoopObservationSink. onError is optional; pass nil to
// silently drop sink failures.
func NewObservationSequencer(sink ObservationSink, onError func(FilterObservation, error)) *ObservationSequencer {
	if sink == nil {
		sink = NoopObservationSink{}
	}
	return &ObservationSequencer{sink: sink, onError: onError}
}

// Emit assigns the next sequence number, validates the resulting
// FilterObservation, and forwards it to the sink. Sink emission failures are
// isolated per the type-level contract: they never surface through this
// method's return value.
func (s *ObservationSequencer) Emit(ctx context.Context, input FilterObservationInput) (FilterObservation, error) {
	seq := atomic.AddUint64(&s.counter, 1)
	obs, err := newFilterObservation(seq, input)
	if err != nil {
		return FilterObservation{}, err
	}
	if sinkErr := s.sink.Emit(ctx, obs); sinkErr != nil && s.onError != nil {
		s.onError(obs, sinkErr)
	}
	return obs, nil
}
