package openai

import (
	"context"
	"sync"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"go.uber.org/zap"

	"iop/packages/go/streamgate"
)

// This file implements the request-local liveness recovery observation
// projection (SDD S06). It sits between StreamGate's immutable
// FilterObservation timeline and the configured downstream sink. For each
// request runtime, Server.observationSink() returns one fresh
// openAILivenessObservationSink. That wrapper watches the predecessor-owned
// private liveness filter (openAIStallRecoveryFilterID / openai.liveness) plus
// the ExactReplay recovery lifecycle it arms, and emits exactly one bounded
// eligibility observation and at most one final result per liveness cycle. It
// never changes filter arbitration, recovery budgets, dispatch, or any Core
// observation; sink/metric/log failures stay observation-only.

// liveness_recovery_observability metric label vocabularies. All values are
// closed and low-cardinality: correlation/request/attempt/run/session/model/
// provider/node/lease/slot/credential identifiers and raw prompt/response/tool
// content are never used as labels or logged.

const (
	livenessMetricEligibilityName = "iop_edge_liveness_recovery_eligibility_total"
	livenessMetricResultsName     = "iop_edge_liveness_recovery_results_total"

	livenessLogMessage = "edge_liveness_recovery_observation"
)

// execution_path label values.
const (
	livenessPathNormalized     = "normalized"
	livenessPathProviderTunnel = "provider_tunnel"
	livenessPathUnknown        = "unknown"
)

// provider_health label values.
const (
	livenessHealthAvailable   = "available"
	livenessHealthUnavailable = "unavailable"
	livenessHealthUnknown     = "unknown"
)

// commit_state label values (Core's closed CommitState plus unknown fallback).
const (
	livenessCommitUncommitted = "transport_uncommitted"
	livenessCommitStreamOpen  = "stream_open"
	livenessCommitTerminal    = "terminal_committed"
	livenessCommitUnknown     = "unknown"
)

// eligibility label values. Only the values the predecessor filter can produce
// are reachable today; the rest are reserved so the closed vocabulary does not
// have to change if the predecessor's descriptor set grows. See
// classifyLivenessEligibility for the exact descriptor mapping.
const (
	livenessEligibilityEligible              = "eligible"
	livenessEligibilityNoOwner               = "no_owner"
	livenessEligibilityPostCommit            = "post_commit"
	livenessEligibilityUnconfirmedFence      = "unconfirmed_fence"
	livenessEligibilityCallerCancelled       = "caller_cancelled"
	livenessEligibilityToolSideEffect        = "tool_side_effect"
	livenessEligibilityBudgetExhausted       = "budget_exhausted"
	livenessEligibilityNoCandidate           = "no_candidate"
	livenessEligibilitySameProviderForbidden = "same_provider_forbidden"
	livenessEligibilityOther                 = "other"
)

// recovery_result label values.
const (
	livenessResultRedispatched   = "redispatched"
	livenessResultPlanRejected   = "plan_rejected"
	livenessResultAbortFailed    = "abort_failed"
	livenessResultRebuildFailed  = "rebuild_failed"
	livenessResultDispatchFailed = "dispatch_failed"
	livenessResultNotSelected    = "not_selected"
	livenessResultTerminal       = "terminal"
	livenessResultOther          = "other"
)

// Predecessor descriptor codes. These mirror the sanitized evidence descriptor
// strings emitted by openAIStallRecoveryFilter.Evaluate in
// stream_gate_filters.go. They are not exported constants there, so they are
// re-declared here and covered by a test that drives the real filter, so a
// predecessor change is caught rather than silently mismapped.
const (
	livenessDescriptorConfirmed       = "response_stalled_confirmed"
	livenessDescriptorUnconfirmed     = "response_stalled_unconfirmed"
	livenessDescriptorIneligible      = "response_stalled_ineligible"
	livenessDescriptorProviderIgnored = "provider_error_ignored"
)

// livenessRecoveryCollectors is one Prometheus collector set for the liveness
// recovery projection. The production set is registered exactly once at package
// initialization against the default registerer; tests construct isolated sets
// against an explicit registry. NewServer, observationSink(), and the request
// wrapper never register collectors.
type livenessRecoveryCollectors struct {
	eligibility *prometheus.CounterVec
	results     *prometheus.CounterVec
}

// newLivenessRecoveryCollectors registers the eligibility and result counters
// against reg. A nil reg falls back to the default registerer.
func newLivenessRecoveryCollectors(reg prometheus.Registerer) *livenessRecoveryCollectors {
	if reg == nil {
		reg = prometheus.DefaultRegisterer
	}
	factory := promauto.With(reg)
	return &livenessRecoveryCollectors{
		eligibility: factory.NewCounterVec(prometheus.CounterOpts{
			Name: livenessMetricEligibilityName,
			Help: "Private OpenAI liveness recovery eligibility decisions by execution path, provider health, commit state, and sanitized eligibility.",
		}, []string{"execution_path", "provider_health", "commit_state", "eligibility"}),
		results: factory.NewCounterVec(prometheus.CounterOpts{
			Name: livenessMetricResultsName,
			Help: "Final OpenAI liveness recovery results by execution path, provider health, and recovery result.",
		}, []string{"execution_path", "provider_health", "recovery_result"}),
	}
}

// defaultLivenessRecoveryCollectors is the process-global production collector
// set. It is created exactly once here and shared by every default Server.
var defaultLivenessRecoveryCollectors = newLivenessRecoveryCollectors(prometheus.DefaultRegisterer)

// livenessPhase is the bounded request-local cycle phase.
type livenessPhase int

const (
	livenessPhaseIdle livenessPhase = iota
	livenessPhaseEligiblePending
)

func (p livenessPhase) String() string {
	if p == livenessPhaseEligiblePending {
		return "eligible_pending"
	}
	return "idle"
}

// openAILivenessObservationSink is the request-local wrapper around the
// configured downstream observation sink. It retains only a mutex-protected
// bounded phase plus the current cycle's closed classification values; no raw
// identifiers are held. Every method is safe for concurrent use because
// parallel filter evaluation can emit observations from multiple goroutines.
type openAILivenessObservationSink struct {
	downstream      streamgate.ObservationSink
	logger          *zap.Logger
	suppressDefault bool
	collectors      *livenessRecoveryCollectors

	mu           sync.Mutex
	phase        livenessPhase
	cyclePath    string
	cycleHealth  string
	cycleCommit  string
	cycleElig    string
	recoverySeen bool
	resultDone   bool
}

var _ streamgate.ObservationSink = (*openAILivenessObservationSink)(nil)

// newOpenAILivenessObservationSink builds a fresh request-local wrapper. A nil
// downstream defaults to NoopObservationSink; a nil logger defaults to a no-op
// logger; a nil collector set defaults to the process-global production set.
// suppressDefault is true only when downstream is the Server's constructor-owned
// default generic zap sink.
func newOpenAILivenessObservationSink(downstream streamgate.ObservationSink, logger *zap.Logger, suppressDefault bool, collectors *livenessRecoveryCollectors) *openAILivenessObservationSink {
	if downstream == nil {
		downstream = streamgate.NoopObservationSink{}
	}
	if logger == nil {
		logger = zap.NewNop()
	}
	if collectors == nil {
		collectors = defaultLivenessRecoveryCollectors
	}
	return &openAILivenessObservationSink{
		downstream:      downstream,
		logger:          logger,
		suppressDefault: suppressDefault,
		collectors:      collectors,
	}
}

// Emit projects the observation into the liveness metrics and, when this
// wrapper owns the constructor-default generic sink, replaces the suppressed
// high-cardinality generic log with the safe edge_liveness_recovery_observation
// entry for consumed private-liveness/ExactReplay rows. Every other observation
// is forwarded unchanged to the downstream sink. Metric projection always runs;
// suppression and the safe log only apply on the default-sink path.
func (s *openAILivenessObservationSink) Emit(ctx context.Context, obs streamgate.FilterObservation) error {
	s.mu.Lock()
	consumed, elig, result := s.project(obs)
	phase := s.phase.String()
	s.mu.Unlock()

	if s.suppressDefault && consumed {
		s.writeSafeLog(phase, obs, elig, result)
		return nil
	}
	// Observation delivery is deliberately best-effort. A custom sink failure
	// must not feed back into filter arbitration or recovery ownership.
	_ = s.downstream.Emit(ctx, obs)
	return nil
}

// project updates the request-local phase and records metrics for one
// observation. It returns whether the observation belongs to the private
// liveness cycle (and must be kept off the generic writer) plus the eligibility
// and result recorded on this call (empty when none). It must be called with
// s.mu held.
func (s *openAILivenessObservationSink) project(obs streamgate.FilterObservation) (consumed bool, elig string, result string) {
	kind := obs.Kind()

	if isLivenessFilterObservation(obs) {
		if kind != streamgate.ObservationKindFilterEvaluated {
			// filter_evaluation_started or any other private filter row: keep it
			// off the generic writer but record no metric.
			return true, "", ""
		}
		descriptor := livenessDescriptor(obs)
		eligibility, cycle := classifyLivenessEligibility(descriptor)
		if !cycle {
			// A provider error the liveness filter did not treat as a stall.
			return true, "", ""
		}
		if s.phase == livenessPhaseEligiblePending {
			// Deduplicate a second eligibility while a cycle is still open.
			return true, "", ""
		}
		s.cyclePath = classifyLivenessPath(obs.AttemptTarget().ExecutionPath())
		s.cycleHealth = classifyLivenessHealth(livenessProviderHealth(obs))
		s.cycleCommit = classifyLivenessCommit(obs.CommitState())
		s.cycleElig = eligibility
		s.recoverySeen = false
		s.resultDone = false
		s.collectors.eligibility.WithLabelValues(s.cyclePath, s.cycleHealth, s.cycleCommit, eligibility).Inc()
		if eligibility == livenessEligibilityEligible {
			s.phase = livenessPhaseEligiblePending
			return true, eligibility, ""
		}
		// Ineligible decisions finish immediately with a terminal result.
		return true, eligibility, s.recordResult(livenessResultTerminal)
	}

	if s.phase != livenessPhaseEligiblePending {
		return false, "", ""
	}

	switch kind {
	case streamgate.ObservationKindRecoveryPlanSelected:
		if !isExactReplayRecovery(obs) {
			// A different recovery strategy won arbitration. This private
			// liveness cycle was not selected, but the unrelated lifecycle
			// observation must retain its normal downstream visibility.
			return false, "", s.recordResult(livenessResultNotSelected)
		}
		s.recoverySeen = true
		return true, "", ""
	case streamgate.ObservationKindRecoveryAttemptAborted,
		streamgate.ObservationKindRecoveryPrepared,
		streamgate.ObservationKindRecoveryRebuilt:
		if !isExactReplayRecovery(obs) {
			return false, "", ""
		}
		s.recoverySeen = true
		return true, "", ""
	case streamgate.ObservationKindRecoveryDispatched:
		if !isExactReplayRecovery(obs) {
			return false, "", ""
		}
		return true, "", s.recordResult(livenessResultRedispatched)
	case streamgate.ObservationKindRecoveryPlanRejected:
		// Core intentionally omits Recovery from plan_rejected observations;
		// while this private cycle is pending, the row is its final rejection.
		return true, "", s.recordResult(livenessResultPlanRejected)
	case streamgate.ObservationKindRecoveryAttemptAbortFailed:
		if !isExactReplayRecovery(obs) {
			return false, "", ""
		}
		return true, "", s.recordResult(livenessResultAbortFailed)
	case streamgate.ObservationKindRecoveryRebuildFailed:
		if !isExactReplayRecovery(obs) {
			return false, "", ""
		}
		return true, "", s.recordResult(livenessResultRebuildFailed)
	case streamgate.ObservationKindRecoveryDispatchFailed:
		if !isExactReplayRecovery(obs) {
			return false, "", ""
		}
		return true, "", s.recordResult(livenessResultDispatchFailed)
	case streamgate.ObservationKindTerminalCommitted:
		// The terminal itself stays on the generic writer; it only finalizes the
		// liveness result when recovery ended without an explicit lifecycle row.
		final := livenessResultTerminal
		if !s.recoverySeen {
			final = livenessResultNotSelected
		}
		return false, "", s.recordResult(final)
	}
	return false, "", ""
}

// isExactReplayRecovery reports whether a lifecycle row belongs to the
// private liveness strategy. Plan rejection is the one Core lifecycle row
// without Recovery metadata and is handled explicitly in project.
func isExactReplayRecovery(obs streamgate.FilterObservation) bool {
	recovery := obs.Recovery()
	return recovery != nil && recovery.Strategy() == streamgate.RecoveryStrategyExactReplay
}

// recordResult increments the result counter once per cycle and resets the
// phase so a later provider stall can open a new bounded cycle. It must be
// called with s.mu held. The returned value is the recorded result, or "" when
// a result was already recorded for this cycle.
func (s *openAILivenessObservationSink) recordResult(result string) string {
	if s.resultDone {
		return ""
	}
	s.collectors.results.WithLabelValues(s.cyclePath, s.cycleHealth, result).Inc()
	s.resultDone = true
	s.phase = livenessPhaseIdle
	return result
}

// writeSafeLog writes the bounded replacement for the suppressed generic log.
// Only phase and the closed classification labels are recorded; no identifiers
// or raw content are ever present. Values are recomputed from the observation
// through closed maps so nothing high-cardinality can leak.
func (s *openAILivenessObservationSink) writeSafeLog(phase string, obs streamgate.FilterObservation, elig, result string) {
	s.logger.Info(livenessLogMessage,
		zap.String("phase", phase),
		zap.String("execution_path", classifyLivenessPath(obs.AttemptTarget().ExecutionPath())),
		zap.String("provider_health", classifyLivenessHealth(livenessProviderHealth(obs))),
		zap.String("commit_state", classifyLivenessCommit(obs.CommitState())),
		zap.String("eligibility", elig),
		zap.String("recovery_result", result),
	)
}

// isLivenessFilterObservation reports whether obs was attributed to the
// predecessor-owned private liveness filter.
func isLivenessFilterObservation(obs streamgate.FilterObservation) bool {
	attr := obs.Attribution()
	if attr == nil {
		return false
	}
	return attr.FilterID() == openAIStallRecoveryFilterID
}

// livenessDescriptor returns the sanitized evidence descriptor code carried by a
// private liveness filter_evaluated observation, or "" when absent.
func livenessDescriptor(obs streamgate.FilterObservation) string {
	ev := obs.Evidence()
	if ev == nil {
		return ""
	}
	return ev.DescriptorCode()
}

// livenessProviderHealth returns the raw provider-health signal carried by the
// observation. The predecessor's private filter_evaluated observation does not
// carry provider health (health lives only in the request-local recovery state
// bridge, never in the immutable timeline), so this is currently always empty
// and classifyLivenessHealth resolves it to unknown. The seam is kept so a
// future health-bearing observation maps without a projection change.
func livenessProviderHealth(_ streamgate.FilterObservation) string {
	return ""
}

// classifyLivenessEligibility maps a sanitized descriptor to a closed
// eligibility value and reports whether the descriptor opens a liveness cycle.
// provider_error_ignored (and any unrecognized descriptor that is not a stall)
// does not open a cycle.
func classifyLivenessEligibility(descriptor string) (eligibility string, cycle bool) {
	switch descriptor {
	case livenessDescriptorConfirmed:
		return livenessEligibilityEligible, true
	case livenessDescriptorUnconfirmed:
		return livenessEligibilityUnconfirmedFence, true
	case livenessDescriptorIneligible:
		// The predecessor collapses post-commit, tool-side-effect, caller-cancel,
		// and missing-request-ref into one ineligible descriptor, so the exact
		// reason is not recoverable from the immutable timeline.
		return livenessEligibilityOther, true
	case livenessDescriptorProviderIgnored, "":
		return "", false
	default:
		return livenessEligibilityOther, true
	}
}

// classifyLivenessPath maps an execution path to the closed path vocabulary.
func classifyLivenessPath(path string) string {
	switch path {
	case livenessPathNormalized:
		return livenessPathNormalized
	case livenessPathProviderTunnel:
		return livenessPathProviderTunnel
	default:
		return livenessPathUnknown
	}
}

// classifyLivenessHealth maps a raw provider-health signal to the closed health
// vocabulary with an unknown fallback.
func classifyLivenessHealth(health string) string {
	switch health {
	case livenessHealthAvailable:
		return livenessHealthAvailable
	case livenessHealthUnavailable:
		return livenessHealthUnavailable
	default:
		return livenessHealthUnknown
	}
}

// classifyLivenessCommit maps Core's commit state to the closed commit
// vocabulary with an unknown fallback.
func classifyLivenessCommit(cs streamgate.CommitState) string {
	switch cs {
	case streamgate.CommitStateTransportUncommitted:
		return livenessCommitUncommitted
	case streamgate.CommitStateStreamOpen:
		return livenessCommitStreamOpen
	case streamgate.CommitStateTerminalCommitted:
		return livenessCommitTerminal
	default:
		return livenessCommitUnknown
	}
}
