package openai

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"sync"
	"time"
	"unicode"

	"iop/packages/go/config"
	"iop/packages/go/streamgate"
)

// This file defines the caller-neutral, Edge-owned semantic output-validation
// filters that consume the Stream Evidence Gate Core. Each filter reads only the
// immutable FilterContext and EvidenceBatch and returns a sanitized decision plus
// an optional typed RecoveryIntent. Core owns hold, all-complete arbitration,
// commit, rebuild, and the recovery budget; these filters own only the semantic
// judgement and the intent shape.
//
// The three kinds map directly onto the three Core hold shapes the pipeline must
// exercise (SDD S02/S14):
//
//   - repeat_guard:   rolling-window participant -> ready/evaluated each epoch.
//   - schema_gate:    terminal-gate participant  -> blocking-deferred until the
//                     terminal trigger, then evaluated.
//   - provider_error: error-event-only participant -> not-applicable on a clean
//                     epoch and observed-unmatched pass on a provider error.
//                     Matcher and recovery intent construction remain follow-up
//                     work; this foundation never creates an exact replay.
//
// Full repeat detection/repair (repeat-guard Task) and JSON-schema validation
// (schema-contract Task) are out of this milestone Task's scope; here the repeat
// and schema filters are the rolling/terminal pipeline participants that produce
// the correct outcome lifecycle and evidence.

const (
	openAIRepeatGuardFilterID       = "openai.repeat_guard"
	openAIRepeatGuardRuleID         = "openai.repeat_guard.rolling"
	openAIRepeatActionGuardFilterID = "openai.repeat_guard.action"
	openAIRepeatActionGuardRuleID   = "openai.repeat_guard.action.terminal"
	openAISchemaGateFilterID        = "openai.schema_gate"
	openAISchemaGateRuleID          = "openai.schema_gate.terminal"
	openAIProviderErrorFilterID     = "openai.provider_error"
	openAIProviderErrorRuleID       = "openai.provider_error.foundation"

	openAIOutputFilterConsumerID = "openai.output_filters"
)

// openAIOutputFilterKind identifies the semantic behavior of an output filter.
type openAIOutputFilterKind string

const (
	openAIOutputFilterRepeatGuard       openAIOutputFilterKind = config.StreamGateFilterRepeatGuard
	openAIOutputFilterRepeatActionGuard openAIOutputFilterKind = "repeat_action_guard"
	openAIOutputFilterSchemaGate        openAIOutputFilterKind = config.StreamGateFilterSchemaGate
	openAIOutputFilterProviderError     openAIOutputFilterKind = config.StreamGateFilterProviderError
)

// openAIOutputFilter is the single semantic output-validation filter type. Its
// kind selects the hold shape (rolling/terminal/none) and the decision it makes.
// It is request-local: requestRef and priority are captured at construction so a
// provider_error violation intent always carries a priority that matches its
// resolved registration priority.
type openAIOutputFilter struct {
	streamgate.FilterBase
	kind       openAIOutputFilterKind
	ruleID     string
	channel    string
	holdRunes  int
	priority   int
	requestRef string
	history    openAIRepeatHistorySnapshot

	repeatMu               sync.Mutex
	repeatAttemptID        string
	releasedContentBytes   int
	releasedReasoningBytes int
}

// newOpenAIOutputFilter constructs one request-local semantic participant.
// history is optional for direct unit fixtures; production always supplies the
// endpoint decoder's immutable raw-free snapshot.
func newOpenAIOutputFilter(kind openAIOutputFilterKind, holdRunes, priority int, requestRef string, history ...openAIRepeatHistorySnapshot) (*openAIOutputFilter, error) {
	var id, rule string
	switch kind {
	case openAIOutputFilterRepeatGuard:
		id, rule = openAIRepeatGuardFilterID, openAIRepeatGuardRuleID
	case openAIOutputFilterRepeatActionGuard:
		id, rule = openAIRepeatActionGuardFilterID, openAIRepeatActionGuardRuleID
	case openAIOutputFilterSchemaGate:
		id, rule = openAISchemaGateFilterID, openAISchemaGateRuleID
	case openAIOutputFilterProviderError:
		id, rule = openAIProviderErrorFilterID, openAIProviderErrorRuleID
	default:
		return nil, fmt.Errorf("unsupported openai output filter kind %q", kind)
	}
	if holdRunes <= 0 {
		holdRunes = config.DefaultStreamGateFilterHoldEvidenceRunes
	}
	if priority < 0 {
		return nil, fmt.Errorf("openai output filter priority must be non-negative")
	}
	base, err := streamgate.NewFilterBase(id)
	if err != nil {
		return nil, err
	}
	filter := &openAIOutputFilter{
		FilterBase: base,
		kind:       kind,
		ruleID:     rule,
		channel:    streamGateChannelDefault,
		holdRunes:  holdRunes,
		priority:   priority,
		requestRef: requestRef,
	}
	if len(history) > 0 {
		filter.history = history[0]
	}
	return filter, nil
}

// A provider may coalesce one valid content/reasoning delta above Core's
// default hold size. This bound is measured in runes and does not widen evidence.
const openAIRepeatHoldMaxBufferRunes = 1 << 20

// Applies is execution-path-neutral because endpoint codecs expose the same
// semantic event kinds for normalized and provider-tunnel attempts.
func (f *openAIOutputFilter) Applies(streamgate.FilterContext) bool {
	return true
}

// HoldRequirement selects the Core hold shape for this filter kind.
func (f *openAIOutputFilter) HoldRequirement(streamgate.FilterContext) streamgate.FilterHoldRequirement {
	switch f.kind {
	case openAIOutputFilterRepeatGuard:
		req, err := streamgate.NewFilterHoldRequirementRollingWithMaxBuffer(
			f.channel,
			[]streamgate.EventKind{
				streamgate.EventKindTextDelta,
				streamgate.EventKindReasoningDelta,
			},
			f.holdRunes,
			openAIRepeatHoldMaxBufferRunes,
		)
		if err != nil {
			req, _ = streamgate.NewFilterHoldRequirementRollingWithMaxBuffer(
				f.channel,
				[]streamgate.EventKind{streamgate.EventKindTextDelta},
				f.holdRunes,
				openAIRepeatHoldMaxBufferRunes,
			)
		}
		return req
	case openAIOutputFilterRepeatActionGuard:
		req, err := streamgate.NewFilterHoldRequirementTerminalGateWithMaxBuffer(
			f.channel,
			[]streamgate.EventKind{
				streamgate.EventKindToolCallFragment,
				streamgate.EventKindTerminal,
			},
			streamgate.EventKindTerminal,
			openAIRepeatHoldMaxBufferRunes,
		)
		if err != nil {
			req, _ = streamgate.NewFilterHoldRequirementTerminalGateWithMaxBuffer(
				f.channel,
				[]streamgate.EventKind{streamgate.EventKindToolCallFragment},
				streamgate.EventKindTerminal,
				openAIRepeatHoldMaxBufferRunes,
			)
		}
		return req
	case openAIOutputFilterSchemaGate:
		req, err := streamgate.NewFilterHoldRequirementTerminalGate(
			f.channel,
			[]streamgate.EventKind{streamgate.EventKindTextDelta, streamgate.EventKindTerminal},
			streamgate.EventKindTerminal,
		)
		if err != nil {
			req, _ = streamgate.NewFilterHoldRequirementTerminalGate(f.channel, []streamgate.EventKind{streamgate.EventKindTextDelta}, streamgate.EventKindTerminal)
		}
		return req
	default: // provider_error: none mode is trigger-ready only on provider_error.
		req, err := streamgate.NewFilterHoldRequirementNone(
			f.channel,
			[]streamgate.EventKind{streamgate.EventKindProviderError},
		)
		if err != nil {
			req, _ = streamgate.NewFilterHoldRequirementNone(f.channel, []streamgate.EventKind{streamgate.EventKindProviderError})
		}
		return req
	}
}

func (f *openAIOutputFilter) Evaluate(ctx context.Context, fctx streamgate.FilterContext, batch streamgate.EvidenceBatch) (streamgate.FilterDecision, error) {
	if err := ctx.Err(); err != nil {
		return streamgate.FilterDecision{}, err
	}
	if f.kind == openAIOutputFilterRepeatGuard {
		return f.evaluateRepeatGuard(fctx, batch)
	}
	if f.kind == openAIOutputFilterRepeatActionGuard {
		return f.evaluateRepeatActionGuard(batch)
	}
	events := batch.Events()
	kind := streamgate.EventKindTextDelta
	if len(events) > 0 {
		kind = events[len(events)-1].Kind()
	}
	ts := batch.CapturedAt()
	if ts.IsZero() {
		ts = time.Now()
	}

	descriptor := "repeat_rolling_clear"
	switch f.kind {
	case openAIOutputFilterSchemaGate:
		descriptor = "schema_terminal_clear"
	case openAIOutputFilterProviderError:
		if batchHasProviderError(batch) {
			descriptor = "provider_error_observed_unmatched"
		} else {
			descriptor = "provider_error_absent"
		}
	}
	evidence, err := streamgate.NewSanitizedEvidence(
		kind, f.channel, f.ruleID, descriptor,
		openAIOutputFilterFingerprint(f.ruleID, descriptor), len(events), 0,
		streamgate.FilterOutcomeKindEvaluated, ts,
	)
	if err != nil {
		return streamgate.FilterDecision{}, err
	}
	return streamgate.NewFilterDecision(streamgate.FilterDecisionKindPass, openAIOutputFilterConsumerID, f.ID(), f.ruleID, evidence, nil)
}

type openAIRepeatTextMatch struct {
	channel     openAIRepeatChannel
	cursorBytes int
	count       int
	fingerprint streamgate.FixedFingerprint
	descriptor  string
}

func (f *openAIOutputFilter) evaluateRepeatGuard(fctx streamgate.FilterContext, batch streamgate.EvidenceBatch) (streamgate.FilterDecision, error) {
	f.repeatMu.Lock()
	defer f.repeatMu.Unlock()
	if f.repeatAttemptID != fctx.AttemptID() {
		f.repeatAttemptID = fctx.AttemptID()
		f.releasedContentBytes = 0
		f.releasedReasoningBytes = 0
	}

	pending := batch.ChannelPending()[f.channel]
	lookBehind := batch.CommittedLookBehind()[f.channel]
	events := batch.Events()
	kind := streamgate.EventKindTextDelta
	if len(events) > 0 {
		kind = events[len(events)-1].Kind()
	}
	ts := batch.CapturedAt()
	if ts.IsZero() {
		ts = time.Now()
	}

	contentLookBehind, reasoningLookBehind, releasedTool := openAIRepeatEventText(lookBehind)
	contentPending, reasoningPending, pendingTool := openAIRepeatEventText(pending)
	contentTail := contentLookBehind + contentPending
	reasoningTail := reasoningLookBehind + reasoningPending
	contentBase := f.releasedContentBytes - len(contentLookBehind)
	reasoningBase := f.releasedReasoningBytes - len(reasoningLookBehind)
	if contentBase < 0 {
		contentBase = 0
	}
	if reasoningBase < 0 {
		reasoningBase = 0
	}

	match, matched := f.findRepeatTextMatch(
		openAIRepeatChannelContent,
		contentTail,
		contentBase,
		f.releasedContentBytes+len(contentPending),
	)
	if !matched {
		match, matched = f.findRepeatTextMatch(
			openAIRepeatChannelReasoning,
			reasoningTail,
			reasoningBase,
			f.releasedContentBytes+len(contentPending),
		)
	}
	if matched {
		// A rolling suffix match that uses committed look-behind must resume at
		// the released boundary. The bounded suffix can otherwise identify a
		// later overlapping period inside pending bytes and duplicate part of
		// the safe prefix. A history anchor keeps its exact pending prefix when
		// it starts after the boundary, but cannot retract committed bytes.
		if match.channel == openAIRepeatChannelContent &&
			(len(contentLookBehind) > 0 &&
				match.descriptor == "repeat_content_detected" ||
				match.cursorBytes < f.releasedContentBytes) {
			match.cursorBytes = f.releasedContentBytes
		}
		reasoningReleasedBoundary := f.releasedContentBytes + len(contentPending) + 1 + f.releasedReasoningBytes
		if match.channel == openAIRepeatChannelReasoning &&
			(len(reasoningLookBehind) > 0 &&
				match.descriptor == "repeat_content_detected" ||
				match.cursorBytes < reasoningReleasedBoundary) {
			match.cursorBytes = reasoningReleasedBoundary
		}
		if fctx.HasToolSideEffect() || releasedTool || pendingTool {
			return f.repeatDecision(
				streamgate.FilterDecisionKindFatal,
				kind,
				"repeat_after_tool_boundary",
				match.fingerprint,
				match.count,
				match.cursorBytes,
				nil,
				ts,
			)
		}
		if f.requestRef == "" {
			return f.repeatDecision(
				streamgate.FilterDecisionKindFatal,
				kind,
				"repeat_recovery_source_unavailable",
				match.fingerprint,
				match.count,
				match.cursorBytes,
				nil,
				ts,
			)
		}
		directive, err := streamgate.NewRecoveryDirectiveContinuation(match.cursorBytes, f.requestRef)
		if err != nil {
			return streamgate.FilterDecision{}, err
		}
		intent, err := streamgate.NewRecoveryIntent(
			streamgate.RecoveryStrategyContinuationRepair,
			directive,
			"repeat_content_detected",
			f.priority,
		)
		if err != nil {
			return streamgate.FilterDecision{}, err
		}
		return f.repeatDecision(
			streamgate.FilterDecisionKindViolation,
			kind,
			match.descriptor,
			match.fingerprint,
			match.count,
			match.cursorBytes,
			&intent,
			ts,
		)
	}

	// A successful epoch releases every pending event captured by the Core.
	// Tracking only byte counts keeps the next committed-look-behind cursor
	// absolute without retaining any model output beyond the Core's bounded tail.
	if len(pending) > 0 {
		f.releasedContentBytes += len(contentPending)
		f.releasedReasoningBytes += len(reasoningPending)
	}
	return f.repeatDecision(
		streamgate.FilterDecisionKindPass,
		kind,
		"repeat_rolling_clear",
		openAIOutputFilterFingerprint(f.ruleID, "repeat_rolling_clear"),
		len(events),
		0,
		nil,
		ts,
	)
}

type openAIRepeatPendingAction struct {
	id        string
	name      string
	arguments []byte
}

func (f *openAIOutputFilter) evaluateRepeatActionGuard(batch streamgate.EvidenceBatch) (streamgate.FilterDecision, error) {
	ts := batch.CapturedAt()
	if ts.IsZero() {
		ts = time.Now()
	}

	actions := make(map[string]*openAIRepeatPendingAction)
	order := make([]string, 0)
	for _, ev := range batch.ChannelPending()[f.channel] {
		if ev.Kind() != streamgate.EventKindToolCallFragment {
			continue
		}
		call, err := ev.AsToolCallFragment()
		if err != nil {
			return f.repeatActionFatal("action_fragment_invalid", "", ts)
		}
		id := call.ID
		if id == "" {
			return f.repeatActionFatal("action_call_id_missing", "", ts)
		}
		action := actions[id]
		if action == nil {
			action = &openAIRepeatPendingAction{id: id}
			actions[id] = action
			order = append(order, id)
		}
		if call.Name != "" {
			if action.name != "" && action.name != call.Name {
				return f.repeatActionFatal("action_name_conflict", id, ts)
			}
			action.name = call.Name
		}
		action.arguments = append(action.arguments, call.Arguments...)
	}

	for _, id := range order {
		action := actions[id]
		if action.name == "" {
			return f.repeatActionFatal("action_name_missing", id, ts)
		}
		if len(action.arguments) == 0 || !json.Valid(action.arguments) {
			return f.repeatActionFatal("action_arguments_incomplete", id, ts)
		}
		fingerprint, ok := openAIRepeatActionFingerprint(action.name, json.RawMessage(action.arguments))
		if !ok {
			return f.repeatActionFatal("action_fingerprint_unavailable", id, ts)
		}
		if count, repeated := f.history.repeatedAction(fingerprint); repeated {
			return f.repeatDecision(
				streamgate.FilterDecisionKindFatal,
				streamgate.EventKindToolCallFragment,
				"repeated_action_no_progress",
				fingerprint,
				count+1,
				0,
				nil,
				ts,
			)
		}
	}

	kind := streamgate.EventKindTerminal
	if len(order) > 0 {
		kind = streamgate.EventKindToolCallFragment
	}
	return f.repeatDecision(
		streamgate.FilterDecisionKindPass,
		kind,
		"repeat_action_clear",
		openAIOutputFilterFingerprint(f.ruleID, "repeat_action_clear"),
		len(order),
		0,
		nil,
		ts,
	)
}

func (f *openAIOutputFilter) repeatActionFatal(descriptor, callID string, ts time.Time) (streamgate.FilterDecision, error) {
	fingerprint := openAIOutputFilterFingerprint(f.ruleID, descriptor)
	if callID != "" {
		fingerprint = streamgate.FixedFingerprint(sha256.Sum256([]byte(f.ruleID + "\x00" + descriptor + "\x00" + callID)))
	}
	return f.repeatDecision(
		streamgate.FilterDecisionKindFatal,
		streamgate.EventKindToolCallFragment,
		descriptor,
		fingerprint,
		1,
		0,
		nil,
		ts,
	)
}

func (f *openAIOutputFilter) findRepeatTextMatch(
	channel openAIRepeatChannel,
	tail string,
	baseBytes int,
	contentBytesAtViolation int,
) (openAIRepeatTextMatch, bool) {
	if tail == "" {
		return openAIRepeatTextMatch{}, false
	}
	if !f.history.completedProgress() {
		if candidate, fingerprint, sourceStart, matched := f.findAssistantHistoryAnchor(channel, tail); matched {
			cursor := baseBytes + sourceStart
			if channel == openAIRepeatChannelReasoning {
				cursor = contentBytesAtViolation + 1 + cursor
			}
			return openAIRepeatTextMatch{
				channel: channel, cursorBytes: cursor, count: candidate.count + 1,
				fingerprint: fingerprint, descriptor: "assistant_history_anchor_repeat",
			}, true
		}
	}

	runes := []rune(tail)
	minimum := f.holdRunes / 8
	if minimum < 32 {
		minimum = 32
	}
	if minimum > 128 {
		minimum = 128
	}
	start, count, ok := openAIRepeatedSuffix(runes, minimum)
	if !ok {
		return openAIRepeatTextMatch{}, false
	}
	repeated := string(runes[start:])
	fingerprint, ok := openAIRepeatTextFingerprint(repeated)
	if !ok {
		return openAIRepeatTextMatch{}, false
	}
	cursor := baseBytes + len(string(runes[:start]))
	if channel == openAIRepeatChannelReasoning {
		cursor = contentBytesAtViolation + 1 + cursor
	}
	return openAIRepeatTextMatch{
		channel: channel, cursorBytes: cursor, count: count,
		fingerprint: fingerprint, descriptor: "repeat_content_detected",
	}, true
}

func (f *openAIOutputFilter) findAssistantHistoryAnchor(
	channel openAIRepeatChannel,
	tail string,
) (openAIRepeatHistoryCandidate, streamgate.FixedFingerprint, int, bool) {
	candidates := f.history.assistantCandidates[channel]
	if len(candidates) == 0 {
		return openAIRepeatHistoryCandidate{}, streamgate.FixedFingerprint{}, 0, false
	}
	normalized, sourceByteOffsets := normalizeOpenAIRepeatTail(tail)
	if len(normalized) == 0 {
		return openAIRepeatHistoryCandidate{}, streamgate.FixedFingerprint{}, 0, false
	}

	bestStart := -1
	bestLength := -1
	var bestCandidate openAIRepeatHistoryCandidate
	var bestFingerprint streamgate.FixedFingerprint
	for fingerprint, candidate := range candidates {
		length := candidate.normalizedRunes
		if length <= 0 || length > len(normalized) {
			continue
		}
		for start := 0; start+length <= len(normalized); start++ {
			windowFingerprint := streamgate.FixedFingerprint(sha256.Sum256([]byte(string(normalized[start : start+length]))))
			if windowFingerprint != fingerprint {
				continue
			}
			sourceStart := sourceByteOffsets[start]
			if bestStart < 0 || sourceStart < bestStart ||
				(sourceStart == bestStart && length > bestLength) {
				bestStart = sourceStart
				bestLength = length
				bestCandidate = candidate
				bestFingerprint = fingerprint
			}
			break
		}
	}
	if bestStart < 0 {
		return openAIRepeatHistoryCandidate{}, streamgate.FixedFingerprint{}, 0, false
	}
	return bestCandidate, bestFingerprint, bestStart, true
}

// normalizeOpenAIRepeatTail mirrors strings.Fields normalization while retaining
// only transient source byte offsets. The returned offsets let a normalized
// candidate window produce an original UTF-8-safe continuation cursor without
// retaining any request-history text.
func normalizeOpenAIRepeatTail(value string) ([]rune, []int) {
	normalized := make([]rune, 0, len([]rune(value)))
	sourceByteOffsets := make([]int, 0, cap(normalized))
	inField := false
	for byteOffset, r := range value {
		if unicode.IsSpace(r) {
			inField = false
			continue
		}
		if !inField && len(normalized) > 0 {
			normalized = append(normalized, ' ')
			sourceByteOffsets = append(sourceByteOffsets, byteOffset)
		}
		normalized = append(normalized, r)
		sourceByteOffsets = append(sourceByteOffsets, byteOffset)
		inField = true
	}
	return normalized, sourceByteOffsets
}

// openAIRepeatedSuffix finds a rune-safe duplicated suffix. It compares the
// current suffix with an earlier suffix at a non-overlapping period, extending
// backwards to the start of the observed duplicate. The returned start is the
// safe continuation cursor: everything before it is preserved.
func openAIRepeatedSuffix(runes []rune, minimum int) (int, int, bool) {
	if minimum <= 0 || len(runes) < minimum*2 {
		return 0, 0, false
	}
	bestStart, bestCount := 0, 0
	for period := minimum; period <= len(runes)-minimum; period++ {
		match := 0
		for match < period && match < len(runes)-period {
			left := len(runes) - period - 1 - match
			right := len(runes) - 1 - match
			if left < 0 || runes[left] != runes[right] {
				break
			}
			match++
		}
		if match < minimum {
			continue
		}
		start := len(runes) - match
		if match > bestCount || (match == bestCount && start < bestStart) {
			bestStart, bestCount = start, match
		}
	}
	if bestCount < minimum {
		return 0, 0, false
	}
	return bestStart, 2, true
}

func openAIRepeatEventText(events []streamgate.NormalizedEvent) (string, string, bool) {
	var content, reasoning string
	hasTool := false
	for _, ev := range events {
		switch ev.Kind() {
		case streamgate.EventKindTextDelta:
			if value, err := ev.AsTextDelta(); err == nil {
				content += value
			}
		case streamgate.EventKindReasoningDelta:
			if value, err := ev.AsReasoningDelta(); err == nil {
				reasoning += value
			}
		case streamgate.EventKindToolCallFragment:
			hasTool = true
		}
	}
	return content, reasoning, hasTool
}

func (f *openAIOutputFilter) repeatDecision(
	decisionKind streamgate.FilterDecisionKind,
	eventKind streamgate.EventKind,
	descriptor string,
	fingerprint streamgate.FixedFingerprint,
	count, offset int,
	intent *streamgate.RecoveryIntent,
	ts time.Time,
) (streamgate.FilterDecision, error) {
	evidence, err := streamgate.NewSanitizedEvidence(
		eventKind,
		f.channel,
		f.ruleID,
		descriptor,
		fingerprint,
		count,
		offset,
		streamgate.FilterOutcomeKindEvaluated,
		ts,
	)
	if err != nil {
		return streamgate.FilterDecision{}, err
	}
	return streamgate.NewFilterDecision(
		decisionKind,
		openAIOutputFilterConsumerID,
		f.ID(),
		f.ruleID,
		evidence,
		intent,
	)
}

// batchHasProviderError reports whether the terminal batch carries a provider
// error event.
func batchHasProviderError(batch streamgate.EvidenceBatch) bool {
	for _, ev := range batch.Events() {
		if ev.Kind() == streamgate.EventKindProviderError {
			return true
		}
	}
	return false
}

const (
	openAIStallRecoveryFilterID     = "openai.response_stalled"
	openAIStallRecoveryFilterRuleID = "response_stalled_exact_replay"
	openAIStallRecoveryConsumerID   = "openai.liveness"
	openAIStallRecoveryPriority     = 100
)

// openAIStallRecoveryFilter is an internal, always-present liveness owner for
// supported OpenAI ingress. It is deliberately outside configurable semantic
// filter policy and provider capability admission.
type openAIStallRecoveryFilter struct {
	streamgate.FilterBase
	requestRef string
	state      *openAIStallRecoveryState
}

func newOpenAIStallRecoveryFilter(requestRef string, state *openAIStallRecoveryState) (*openAIStallRecoveryFilter, error) {
	base, err := streamgate.NewFilterBase(openAIStallRecoveryFilterID)
	if err != nil {
		return nil, err
	}
	return &openAIStallRecoveryFilter{FilterBase: base, requestRef: requestRef, state: state}, nil
}

func (f *openAIStallRecoveryFilter) Applies(streamgate.FilterContext) bool { return true }

func (f *openAIStallRecoveryFilter) HoldRequirement(streamgate.FilterContext) streamgate.FilterHoldRequirement {
	req, _ := streamgate.NewFilterHoldRequirementNone(
		streamGateChannelDefault, []streamgate.EventKind{streamgate.EventKindProviderError},
	)
	return req
}

func (f *openAIStallRecoveryFilter) Evaluate(_ context.Context, fctx streamgate.FilterContext, batch streamgate.EvidenceBatch) (streamgate.FilterDecision, error) {
	descriptor := "provider_error_ignored"
	var health string
	for _, event := range batch.Events() {
		if event.Kind() != streamgate.EventKindProviderError {
			continue
		}
		terminal, err := event.AsProviderError()
		external := terminal.ExternalDesc()
		if err != nil || external == nil || external.Code() != openAIStallFailureCode {
			continue
		}
		confirmed := false
		for _, cause := range terminal.FailureCauses().All() {
			if cause.Stage() == openAIStallHandoffStage && cause.Code() == openAIStallHandoffCause {
				confirmed = true
			}
			if cause.Stage() == openAIStallHealthStage {
				health = cause.Code()
			}
		}
		if confirmed && (health == "available" || health == "unavailable" || health == "unknown") {
			descriptor = "response_stalled_confirmed"
			break
		}
		descriptor = "response_stalled_unconfirmed"
	}

	decisionKind := streamgate.FilterDecisionKindPass
	var intent *streamgate.RecoveryIntent
	if descriptor == "response_stalled_confirmed" {
		unsafe := fctx.CommitState() != streamgate.CommitStateTransportUncommitted || fctx.HasToolSideEffect() || f.requestRef == "" || batchHasToolEvidence(batch)
		if unsafe || !f.state.claimRecovery() {
			descriptor = "response_stalled_ineligible"
		} else {
			directive, err := streamgate.NewRecoveryDirectiveExact(f.requestRef)
			if err != nil {
				return streamgate.FilterDecision{}, err
			}
			createdIntent, err := streamgate.NewRecoveryIntent(streamgate.RecoveryStrategyExactReplay, directive, openAIStallFailureCode, openAIStallRecoveryPriority)
			if err != nil {
				return streamgate.FilterDecision{}, err
			}
			intent = &createdIntent
			f.state.arm(fctx.AttemptID(), fctx.ActualProvider(), health)
			decisionKind = streamgate.FilterDecisionKindViolation
		}
	}
	ts := batch.CapturedAt()
	if ts.IsZero() {
		ts = time.Now()
	}
	evidence, err := streamgate.NewSanitizedEvidence(streamgate.EventKindProviderError, streamGateChannelDefault, openAIStallRecoveryFilterRuleID, descriptor, openAIOutputFilterFingerprint(openAIStallRecoveryFilterRuleID, descriptor), 1, 0, streamgate.FilterOutcomeKindEvaluated, ts)
	if err != nil {
		return streamgate.FilterDecision{}, err
	}
	return streamgate.NewFilterDecision(decisionKind, openAIStallRecoveryConsumerID, f.ID(), openAIStallRecoveryFilterRuleID, evidence, intent)
}

func batchHasToolEvidence(batch streamgate.EvidenceBatch) bool {
	for _, events := range [][]streamgate.NormalizedEvent{batch.Events(), batch.ChannelPending()[streamGateChannelDefault], batch.CommittedLookBehind()[streamGateChannelDefault]} {
		for _, event := range events {
			if event.Kind() == streamgate.EventKindToolCallFragment {
				return true
			}
		}
	}
	return false
}

var _ streamgate.Filter = (*openAIStallRecoveryFilter)(nil)

// openAIOutputFilterFingerprint derives a stable, raw-free fingerprint from the
// rule id and a sanitized descriptor so evidence carries no provider text.
func openAIOutputFilterFingerprint(ruleID, descriptor string) streamgate.FixedFingerprint {
	return streamgate.FixedFingerprint(sha256.Sum256([]byte(ruleID + "\x00" + descriptor)))
}

var _ streamgate.Filter = (*openAIOutputFilter)(nil)
