// Package streamgate provides transport-agnostic event contract types for the
// stream evidence gate. It owns the normalized event lifecycle, immutable
// evidence/filter decision types, and terminal/failure/release payloads.
//
// This package must not import apps/, proto/, or packages/go/config/.
// It depends only on the Go standard library.
package streamgate

import (
	"errors"
	"fmt"
	"sort"
	"strings"
	"unicode/utf8"
)

// FilterHoldMode identifies how a filter holds pending evidence before
// allowing release. It is a closed set of modes; each mode defines its own
// trigger condition and field requirements.
type FilterHoldMode string

// knownFilterHoldModes is the closed set of valid FilterHoldMode values.
var knownFilterHoldModes = map[FilterHoldMode]struct{}{
	FilterHoldModeNone:         {},
	FilterHoldModeRolling:      {},
	FilterHoldModeTerminalGate: {},
	FilterHoldModeFragmentGate: {},
}

const (
	// FilterHoldModeNone indicates no hold. Events pass through immediately.
	FilterHoldModeNone FilterHoldMode = "none"

	// FilterHoldModeRolling indicates the filter holds until the rolling
	// rune evidence window meets a configured threshold. Time is not a
	// release condition.
	FilterHoldModeRolling FilterHoldMode = "rolling_window"

	// FilterHoldModeTerminalGate indicates the filter holds all events until
	// a terminal event is received. Terminal or provider-error events serve
	// as the trigger.
	FilterHoldModeTerminalGate FilterHoldMode = "terminal_gate"

	// FilterHoldModeFragmentGate indicates the filter holds tool-call
	// fragments until a CompleteFragment signal arrives for the keyed ID.
	FilterHoldModeFragmentGate FilterHoldMode = "fragment_gate"
)

// Validate returns nil when the mode is a known lifecycle value.
func (m FilterHoldMode) Validate() error {
	switch m {
	case FilterHoldModeNone, FilterHoldModeRolling, FilterHoldModeTerminalGate, FilterHoldModeFragmentGate:
		return nil
	}
	return errors.New("streamgate: unknown filter hold mode: " + string(m))
}

// IsValidMode returns true when m is a known hold mode value.
func IsValidMode(m FilterHoldMode) bool {
	_, ok := knownFilterHoldModes[m]
	return ok
}

// defaultEvidenceRunes is the default Unicode rune evidence window used when
// no policy override is provided.
const defaultEvidenceRunes = 500

// minEvidenceRunes is the absolute minimum allowed rune window.
const minEvidenceRunes = 1

// maxEvidenceRunes is the absolute maximum allowed rune window.
const maxEvidenceRunes = 65536

// defaultMaxBufferRunes is the default hard buffer limit in runes.
const defaultMaxBufferRunes = 4096

// minMaxBufferRunes is the absolute minimum allowed buffer limit.
const minMaxBufferRunes = 10

// maxMaxBufferRunes is the absolute maximum allowed buffer limit.
const maxMaxBufferRunes = 1048576

// FilterHoldRequirement is an immutable specification of how a filter holds
// pending evidence before releasing it. The fields match the SDD naming and
// semantics exactly. A requirement is mode-specific: rolling requires a
// positive rune threshold, terminal gate requires a terminal trigger kind,
// fragment gate requires a fragment trigger kind, and none is a no-op.
type FilterHoldRequirement struct {
	channel         string
	mode            FilterHoldMode
	subscribedKinds []EventKind
	evidenceRunes   int
	triggerKind     EventKind
	maxBufferRunes  int
}

// NewFilterHoldRequirementNone creates a FilterHoldRequirement of mode none.
// None mode has no rune threshold, no trigger, and releases immediately.
// Subscribed kinds are required but unused.
func NewFilterHoldRequirementNone(channel string, subscribedKinds []EventKind) (FilterHoldRequirement, error) {
	if err := validateHoldChannel("none", channel); err != nil {
		return FilterHoldRequirement{}, err
	}
	kinds, err := validateSubscribedKinds("none", subscribedKinds)
	if err != nil {
		return FilterHoldRequirement{}, err
	}
	req := FilterHoldRequirement{
		channel:         channel,
		mode:            FilterHoldModeNone,
		subscribedKinds: kinds,
		evidenceRunes:   0,
		triggerKind:     "",
		maxBufferRunes:  0,
	}
	if err := req.Validate(); err != nil {
		return FilterHoldRequirement{}, err
	}
	return req, nil
}

// NewFilterHoldRequirementRollingWithMaxBuffer creates a FilterHoldRequirement
// of mode rolling_window with an explicit max_buffer_runes override. The
// rune threshold must be within the allowed range. The max_buffer_runes must
// be >= evidence_runes to ensure the buffer can hold the evidence window.
func NewFilterHoldRequirementRollingWithMaxBuffer(channel string, subscribedKinds []EventKind, evidenceRunes int, maxBufferRunes int) (FilterHoldRequirement, error) {
	if err := validateHoldChannel("rolling_window", channel); err != nil {
		return FilterHoldRequirement{}, err
	}
	if evidenceRunes < minEvidenceRunes || evidenceRunes > maxEvidenceRunes {
		return FilterHoldRequirement{}, fmt.Errorf("streamgate: rolling_window evidence_runes must be between %d and %d, got %d", minEvidenceRunes, maxEvidenceRunes, evidenceRunes)
	}
	if maxBufferRunes < evidenceRunes {
		return FilterHoldRequirement{}, fmt.Errorf("streamgate: rolling_window max_buffer_runes must be >= evidence_runes, got evidence=%d buffer=%d", evidenceRunes, maxBufferRunes)
	}
	if maxBufferRunes < minMaxBufferRunes || maxBufferRunes > maxMaxBufferRunes {
		return FilterHoldRequirement{}, fmt.Errorf("streamgate: rolling_window max_buffer_runes must be between %d and %d, got %d", minMaxBufferRunes, maxMaxBufferRunes, maxBufferRunes)
	}
	kinds, err := validateSubscribedKinds("rolling_window", subscribedKinds)
	if err != nil {
		return FilterHoldRequirement{}, err
	}
	req := FilterHoldRequirement{
		channel:         channel,
		mode:            FilterHoldModeRolling,
		subscribedKinds: kinds,
		evidenceRunes:   evidenceRunes,
		triggerKind:     "",
		maxBufferRunes:  maxBufferRunes,
	}
	if err := req.Validate(); err != nil {
		return FilterHoldRequirement{}, err
	}
	return req, nil
}

// NewFilterHoldRequirementRolling creates a FilterHoldRequirement of mode
// rolling_window. The rune threshold must be within the allowed range. The
// subscribed kinds define which event types contribute to the rolling count.
func NewFilterHoldRequirementRolling(channel string, subscribedKinds []EventKind, evidenceRunes int) (FilterHoldRequirement, error) {
	return NewFilterHoldRequirementRollingWithMaxBuffer(channel, subscribedKinds, evidenceRunes, defaultMaxBufferRunes)
}

// NewFilterHoldRequirementTerminalGateWithMaxBuffer creates a FilterHoldRequirement
// of mode terminal_gate with an explicit max_buffer_runes override.
func NewFilterHoldRequirementTerminalGateWithMaxBuffer(channel string, subscribedKinds []EventKind, triggerKind EventKind, maxBufferRunes int) (FilterHoldRequirement, error) {
	if err := validateHoldChannel("terminal_gate", channel); err != nil {
		return FilterHoldRequirement{}, err
	}
	if triggerKind != EventKindTerminal && triggerKind != EventKindProviderError {
		return FilterHoldRequirement{}, fmt.Errorf("streamgate: terminal_gate trigger must be terminal or provider_error, got %q", triggerKind)
	}
	if maxBufferRunes < minMaxBufferRunes || maxBufferRunes > maxMaxBufferRunes {
		return FilterHoldRequirement{}, fmt.Errorf("streamgate: terminal_gate max_buffer_runes must be between %d and %d, got %d", minMaxBufferRunes, maxMaxBufferRunes, maxBufferRunes)
	}
	kinds, err := validateSubscribedKinds("terminal_gate", subscribedKinds)
	if err != nil {
		return FilterHoldRequirement{}, err
	}
	req := FilterHoldRequirement{
		channel:         channel,
		mode:            FilterHoldModeTerminalGate,
		subscribedKinds: kinds,
		evidenceRunes:   0,
		triggerKind:     triggerKind,
		maxBufferRunes:  maxBufferRunes,
	}
	if err := req.Validate(); err != nil {
		return FilterHoldRequirement{}, err
	}
	return req, nil
}

// NewFilterHoldRequirementTerminalGate creates a FilterHoldRequirement of
// mode terminal_gate. The trigger kind must be terminal or provider_error.
func NewFilterHoldRequirementTerminalGate(channel string, subscribedKinds []EventKind, triggerKind EventKind) (FilterHoldRequirement, error) {
	return NewFilterHoldRequirementTerminalGateWithMaxBuffer(channel, subscribedKinds, triggerKind, defaultMaxBufferRunes)
}

// NewFilterHoldRequirementFragmentGateWithMaxBuffer creates a FilterHoldRequirement
// of mode fragment_gate with an explicit max_buffer_runes override.
func NewFilterHoldRequirementFragmentGateWithMaxBuffer(channel string, subscribedKinds []EventKind, triggerKind EventKind, maxBufferRunes int) (FilterHoldRequirement, error) {
	if err := validateHoldChannel("fragment_gate", channel); err != nil {
		return FilterHoldRequirement{}, err
	}
	if triggerKind != EventKindToolCallFragment {
		return FilterHoldRequirement{}, fmt.Errorf("streamgate: fragment_gate trigger must be tool_call_fragment, got %q", triggerKind)
	}
	if maxBufferRunes < minMaxBufferRunes || maxBufferRunes > maxMaxBufferRunes {
		return FilterHoldRequirement{}, fmt.Errorf("streamgate: fragment_gate max_buffer_runes must be between %d and %d, got %d", minMaxBufferRunes, maxMaxBufferRunes, maxBufferRunes)
	}
	kinds, err := validateSubscribedKinds("fragment_gate", subscribedKinds)
	if err != nil {
		return FilterHoldRequirement{}, err
	}
	req := FilterHoldRequirement{
		channel:         channel,
		mode:            FilterHoldModeFragmentGate,
		subscribedKinds: kinds,
		evidenceRunes:   0,
		triggerKind:     triggerKind,
		maxBufferRunes:  maxBufferRunes,
	}
	if err := req.Validate(); err != nil {
		return FilterHoldRequirement{}, err
	}
	return req, nil
}

// NewFilterHoldRequirementFragmentGate creates a FilterHoldRequirement of
// mode fragment_gate. The trigger kind must be tool_call_fragment.
func NewFilterHoldRequirementFragmentGate(channel string, subscribedKinds []EventKind, triggerKind EventKind) (FilterHoldRequirement, error) {
	return NewFilterHoldRequirementFragmentGateWithMaxBuffer(channel, subscribedKinds, triggerKind, defaultMaxBufferRunes)
}

func validateHoldChannel(mode, channel string) error {
	if channel == "" {
		return errors.New("streamgate: hold requirement channel is required")
	}
	return nil
}

func validateSubscribedKinds(mode string, kinds []EventKind) ([]EventKind, error) {
	if len(kinds) == 0 {
		return nil, fmt.Errorf("streamgate: %s hold requirement subscribed kinds must not be empty", mode)
	}
	seen := make(map[EventKind]struct{}, len(kinds))
	for _, k := range kinds {
		if err := k.Validate(); err != nil {
			return nil, fmt.Errorf("streamgate: %s hold requirement subscribed kind: %v", mode, err)
		}
		if _, dup := seen[k]; dup {
			return nil, fmt.Errorf("streamgate: %s hold requirement duplicate subscribed kind: %s", mode, k)
		}
		seen[k] = struct{}{}
	}
	out := make([]EventKind, len(kinds))
	copy(out, kinds)
	sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
	return out, nil
}

// Validate returns nil when the requirement is in a consistent state.
// Mode-specific field rules are enforced.
func (r FilterHoldRequirement) Validate() error {
	if r.channel == "" {
		return errors.New("streamgate: hold requirement channel is required")
	}
	if err := r.mode.Validate(); err != nil {
		return err
	}
	if len(r.subscribedKinds) == 0 {
		return errors.New("streamgate: hold requirement subscribed kinds must not be empty")
	}
	switch r.mode {
	case FilterHoldModeNone:
		if r.evidenceRunes != 0 {
			return errors.New("streamgate: none mode must not have evidence_runes")
		}
		if r.triggerKind != "" {
			return errors.New("streamgate: none mode must not have a trigger")
		}
		if r.maxBufferRunes != 0 {
			return errors.New("streamgate: none mode must not have max_buffer_runes")
		}
	case FilterHoldModeRolling:
		if r.evidenceRunes < minEvidenceRunes || r.evidenceRunes > maxEvidenceRunes {
			return errors.New("streamgate: rolling_window evidence_runes out of range")
		}
		if r.triggerKind != "" {
			return errors.New("streamgate: rolling_window mode must not have a trigger")
		}
		if r.maxBufferRunes < minMaxBufferRunes || r.maxBufferRunes > maxMaxBufferRunes {
			return errors.New("streamgate: rolling_window max_buffer_runes out of range")
		}
	case FilterHoldModeTerminalGate:
		if r.evidenceRunes != 0 {
			return errors.New("streamgate: terminal_gate mode must not have evidence_runes")
		}
		if r.triggerKind != EventKindTerminal && r.triggerKind != EventKindProviderError {
			return errors.New("streamgate: terminal_gate trigger must be terminal or provider_error")
		}
		if r.maxBufferRunes < minMaxBufferRunes || r.maxBufferRunes > maxMaxBufferRunes {
			return errors.New("streamgate: terminal_gate max_buffer_runes out of range")
		}
	case FilterHoldModeFragmentGate:
		if r.evidenceRunes != 0 {
			return errors.New("streamgate: fragment_gate mode must not have evidence_runes")
		}
		if r.triggerKind != EventKindToolCallFragment {
			return errors.New("streamgate: fragment_gate trigger must be tool_call_fragment")
		}
		if r.maxBufferRunes < minMaxBufferRunes || r.maxBufferRunes > maxMaxBufferRunes {
			return errors.New("streamgate: fragment_gate max_buffer_runes out of range")
		}
	}
	return nil
}

// Channel returns the requirement channel key.
func (r FilterHoldRequirement) Channel() string { return r.channel }

// Mode returns the hold mode.
func (r FilterHoldRequirement) Mode() FilterHoldMode { return r.mode }

// SubscribedKinds returns a defensive copy of the subscribed event kinds.
func (r FilterHoldRequirement) SubscribedKinds() []EventKind {
	if r.subscribedKinds == nil {
		return nil
	}
	out := make([]EventKind, len(r.subscribedKinds))
	copy(out, r.subscribedKinds)
	return out
}

// EvidenceRunes returns the rolling rune threshold. Returns 0 for non-rolling
// modes.
func (r FilterHoldRequirement) EvidenceRunes() int { return r.evidenceRunes }

// TriggerKind returns the trigger event kind. Returns "" for modes that do
// not use a trigger.
func (r FilterHoldRequirement) TriggerKind() EventKind { return r.triggerKind }

// MaxBufferRunes returns the hard buffer limit in runes.
func (r FilterHoldRequirement) MaxBufferRunes() int { return r.maxBufferRunes }

// IsBlocking returns true when this requirement's mode blocks channel release.
// None and observe-only modes are non-blocking; rolling, terminal_gate, and
// fragment_gate are blocking.
func (r FilterHoldRequirement) IsBlocking() bool {
	switch r.mode {
	case FilterHoldModeRolling, FilterHoldModeTerminalGate, FilterHoldModeFragmentGate:
		return true
	default:
		return false
	}
}

// FilterHoldBinding pairs a stable filter ID with its requirement and an
// explicit FilterEnforcement resolved by the caller. The binding is
// immutable to callers.
type FilterHoldBinding struct {
	filterID      StableToken
	requirement   FilterHoldRequirement
	enforcement   FilterEnforcement
	blocksRelease bool
}

// NewFilterHoldBinding creates a FilterHoldBinding with validation. The
// returned value is immutable to callers.
func NewFilterHoldBinding(filterID string, req FilterHoldRequirement, enforcement FilterEnforcement) (FilterHoldBinding, error) {
	f, err := NewStableTokenRequired("filterID", filterID)
	if err != nil {
		return FilterHoldBinding{}, err
	}
	if err := req.Validate(); err != nil {
		return FilterHoldBinding{}, err
	}
	if err := enforcement.Validate(); err != nil {
		return FilterHoldBinding{}, err
	}
	blocksRelease := (enforcement == FilterEnforcementBlocking && req.Mode() != FilterHoldModeNone)
	return FilterHoldBinding{
		filterID:      f,
		requirement:   req,
		enforcement:   enforcement,
		blocksRelease: blocksRelease,
	}, nil
}

// NewFilterHoldBindingFromResolvedFilter constructs a FilterHoldBinding from a ResolvedFilter.
func NewFilterHoldBindingFromResolvedFilter(rf ResolvedFilter) (FilterHoldBinding, error) {
	return NewFilterHoldBinding(rf.FilterID(), rf.HoldRequirement(), rf.Enforcement())
}

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

// Requirement returns the hold requirement.
func (b FilterHoldBinding) Requirement() FilterHoldRequirement { return b.requirement }

// Enforcement returns the effective enforcement.
func (b FilterHoldBinding) Enforcement() FilterEnforcement { return b.enforcement }

// BlocksRelease returns whether this binding blocks channel release.
func (b FilterHoldBinding) BlocksRelease() bool { return b.blocksRelease }

// FilterApplicability captures typed readiness and subscription status for a single filter in an epoch.
type FilterApplicability struct {
	filterID               StableToken
	subscribedEventPresent bool
	triggerReady           bool
}

// NewFilterApplicability creates a validated FilterApplicability.
func NewFilterApplicability(filterID string, subscribedEventPresent, triggerReady bool) (FilterApplicability, error) {
	st, err := NewStableTokenRequired("filterID", filterID)
	if err != nil {
		return FilterApplicability{}, fmt.Errorf("streamgate: filter applicability id: %w", err)
	}
	return FilterApplicability{
		filterID:               st,
		subscribedEventPresent: subscribedEventPresent,
		triggerReady:           triggerReady,
	}, nil
}

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

// SubscribedEventPresent returns whether subscribed events were present for this filter in the epoch.
func (a FilterApplicability) SubscribedEventPresent() bool { return a.subscribedEventPresent }

// TriggerReady returns whether the filter trigger condition is satisfied in the epoch.
func (a FilterApplicability) TriggerReady() bool { return a.triggerReady }

// evidencePlan holds the compiled per-channel hold plan derived from bindings.
type evidencePlan struct {
	bindingsByChannel    map[string][]FilterHoldBinding
	allBindings          map[string]FilterHoldBinding
	observeOnlyKinds     map[string][]EventKind
	allKinds             map[string][]EventKind
	blockingKinds        map[string][]EventKind
	maxBufferRunes       map[string]int
	rollingEvidenceRunes map[string]int
}

// compileEvidencePlan produces a compiled plan from a slice of bindings.
func compileEvidencePlan(bindings []FilterHoldBinding) (evidencePlan, error) {
	plan := evidencePlan{
		bindingsByChannel:    make(map[string][]FilterHoldBinding),
		allBindings:          make(map[string]FilterHoldBinding),
		observeOnlyKinds:     make(map[string][]EventKind),
		allKinds:             make(map[string][]EventKind),
		blockingKinds:        make(map[string][]EventKind),
		maxBufferRunes:       make(map[string]int),
		rollingEvidenceRunes: make(map[string]int),
	}

	for _, b := range bindings {
		fID := b.FilterID()
		if fID == "" {
			return evidencePlan{}, errors.New("streamgate: binding filter id is required")
		}
		if _, dup := plan.allBindings[fID]; dup {
			return evidencePlan{}, fmt.Errorf("streamgate: duplicate binding filter id: %s", fID)
		}
		req := b.requirement
		if err := req.Validate(); err != nil {
			return evidencePlan{}, fmt.Errorf("streamgate: binding requirement invalid: %w", err)
		}
		ch := req.Channel()

		plan.allBindings[fID] = b
		plan.bindingsByChannel[ch] = append(plan.bindingsByChannel[ch], b)

		plan.allKinds[ch] = mergeKinds(plan.allKinds[ch], req.SubscribedKinds())

		if b.BlocksRelease() && req.IsBlocking() {
			plan.blockingKinds[ch] = mergeKinds(plan.blockingKinds[ch], req.SubscribedKinds())
			plan.maxBufferRunes[ch] = minPositiveBound(plan.maxBufferRunes[ch], req.MaxBufferRunes())
			if req.Mode() == FilterHoldModeRolling {
				plan.rollingEvidenceRunes[ch] = maxInt(plan.rollingEvidenceRunes[ch], req.EvidenceRunes())
			}
		} else {
			plan.observeOnlyKinds[ch] = mergeKinds(plan.observeOnlyKinds[ch], req.SubscribedKinds())
		}
	}

	// Sort bindings per channel by filter id ascending
	for ch := range plan.bindingsByChannel {
		sort.SliceStable(plan.bindingsByChannel[ch], func(i, j int) bool {
			return plan.bindingsByChannel[ch][i].FilterID() < plan.bindingsByChannel[ch][j].FilterID()
		})
	}

	// Validate bounds per channel: max rolling evidence must be <= shared hard bound
	for ch := range plan.bindingsByChannel {
		maxBuf := plan.maxBufferRunes[ch]
		rollEv := plan.rollingEvidenceRunes[ch]
		if rollEv > 0 && maxBuf > 0 && rollEv > maxBuf {
			return evidencePlan{}, fmt.Errorf("streamgate: max rolling evidence %d > shared hard bound %d for channel %s", rollEv, maxBuf, ch)
		}
	}

	return plan, nil
}

// isKindSubscribed returns true when kind is in the subscribed kinds slice.
func isKindSubscribed(kind EventKind, subscribed []EventKind) bool {
	for _, k := range subscribed {
		if k == kind {
			return true
		}
	}
	return false
}

// appendUniqueKind appends kind to existing only if not already present.
func appendUniqueKind(existing []EventKind, kind EventKind) []EventKind {
	for _, k := range existing {
		if k == kind {
			return existing
		}
	}
	return append(existing, kind)
}

// strongerMode returns true when a is strictly stronger than b in the
// hold priority: terminal_gate > fragment_gate > rolling_window.
func strongerMode(a, b FilterHoldMode) bool {
	priority := map[FilterHoldMode]int{
		FilterHoldModeRolling:      1,
		FilterHoldModeFragmentGate: 2,
		FilterHoldModeTerminalGate: 3,
	}
	return priority[a] > priority[b]
}

// minPositiveBound returns the smaller of two positive values; if either is
// zero (e.g. none mode has no hard bound) it returns the other.
func minPositiveBound(a, b int) int {
	if a == 0 {
		return b
	}
	if b == 0 {
		return a
	}
	if a < b {
		return a
	}
	return b
}

// maxInt returns the larger of two ints.
func maxInt(a, b int) int {
	if a > b {
		return a
	}
	return b
}

func mergeKinds(a, b []EventKind) []EventKind {
	seen := make(map[EventKind]struct{}, len(a)+len(b))
	var out []EventKind
	for _, k := range a {
		if _, ok := seen[k]; !ok {
			seen[k] = struct{}{}
			out = append(out, k)
		}
	}
	for _, k := range b {
		if _, ok := seen[k]; !ok {
			seen[k] = struct{}{}
			out = append(out, k)
		}
	}
	sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
	return out
}

// EvidencePlan is the compiled channel plan for evidence holding. It is
// immutable to callers once created. The plan only includes blocking
// requirements; observe-only and none-mode bindings never create a channel
// hold on their own.
type EvidencePlan struct {
	plan evidencePlan
}

// NewEvidencePlan compiles a channel hold plan from the provided bindings.
// Only blocking bindings affect the hold decision. Returns an error if any
// binding's requirement fails validation.
func NewEvidencePlan(bindings []FilterHoldBinding) (EvidencePlan, error) {
	if len(bindings) == 0 {
		return EvidencePlan{plan: evidencePlan{
			bindingsByChannel:    make(map[string][]FilterHoldBinding),
			allBindings:          make(map[string]FilterHoldBinding),
			observeOnlyKinds:     make(map[string][]EventKind),
			allKinds:             make(map[string][]EventKind),
			blockingKinds:        make(map[string][]EventKind),
			maxBufferRunes:       make(map[string]int),
			rollingEvidenceRunes: make(map[string]int),
		}}, nil
	}
	compiled, err := compileEvidencePlan(bindings)
	if err != nil {
		return EvidencePlan{}, err
	}
	return EvidencePlan{plan: compiled}, nil
}

// NewEvidencePlanFromResolvedFilters compiles an EvidencePlan from ResolvedFilters.
func NewEvidencePlanFromResolvedFilters(resolved []ResolvedFilter) (EvidencePlan, error) {
	if len(resolved) == 0 {
		return NewEvidencePlan(nil)
	}
	seen := make(map[string]struct{}, len(resolved))
	bindings := make([]FilterHoldBinding, len(resolved))
	for i, rf := range resolved {
		fID := rf.FilterID()
		if fID == "" {
			return EvidencePlan{}, errors.New("streamgate: resolved filter id is required")
		}
		if _, dup := seen[fID]; dup {
			return EvidencePlan{}, fmt.Errorf("streamgate: duplicate resolved filter id: %s", fID)
		}
		seen[fID] = struct{}{}

		b, err := NewFilterHoldBindingFromResolvedFilter(rf)
		if err != nil {
			return EvidencePlan{}, fmt.Errorf("streamgate: build binding from resolved filter: %w", err)
		}
		bindings[i] = b
	}
	return NewEvidencePlan(bindings)
}

// BindingsForChannel returns a defensive copy of all bindings for the channel in stable filter id order.
func (p EvidencePlan) BindingsForChannel(channel string) []FilterHoldBinding {
	bindings := p.plan.bindingsByChannel[channel]
	if len(bindings) == 0 {
		return nil
	}
	out := make([]FilterHoldBinding, len(bindings))
	copy(out, bindings)
	return out
}

// Bindings returns a defensive copy of all bindings in stable filter id order.
func (p EvidencePlan) Bindings() []FilterHoldBinding {
	var all []FilterHoldBinding
	for _, b := range p.plan.allBindings {
		all = append(all, b)
	}
	sort.SliceStable(all, func(i, j int) bool {
		return all[i].FilterID() < all[j].FilterID()
	})
	return all
}

// HasModeForChannel returns true when channel has at least one binding with the given mode.
func (p EvidencePlan) HasModeForChannel(channel string, mode FilterHoldMode) bool {
	bindings := p.plan.bindingsByChannel[channel]
	for _, b := range bindings {
		if b.Requirement().Mode() == mode {
			return true
		}
	}
	return false
}

// IsTerminalTriggerForChannel returns true when eventKind is a configured terminal trigger for any binding on channel.
func (p EvidencePlan) IsTerminalTriggerForChannel(channel string, kind EventKind) bool {
	bindings := p.plan.bindingsByChannel[channel]
	for _, b := range bindings {
		if b.BlocksRelease() && b.Requirement().Mode() == FilterHoldModeTerminalGate {
			if b.Requirement().TriggerKind() == kind {
				return true
			}
		}
	}
	return false
}

// BlockingRequirementFor returns a representative storage envelope requirement for the channel.
func (p EvidencePlan) BlockingRequirementFor(channel string) (FilterHoldRequirement, bool) {
	bindings := p.plan.bindingsByChannel[channel]
	if len(bindings) == 0 {
		return FilterHoldRequirement{}, false
	}
	var strongestMode FilterHoldMode = FilterHoldModeNone
	var triggerKind EventKind
	for _, b := range bindings {
		if !b.BlocksRelease() || !b.Requirement().IsBlocking() {
			continue
		}
		m := b.Requirement().Mode()
		if strongerMode(m, strongestMode) {
			strongestMode = m
			triggerKind = b.Requirement().TriggerKind()
		}
	}
	if strongestMode == FilterHoldModeNone {
		return FilterHoldRequirement{}, false
	}
	maxBuf := p.plan.maxBufferRunes[channel]
	rollEv := p.plan.rollingEvidenceRunes[channel]
	blockingKinds := p.plan.blockingKinds[channel]

	var req FilterHoldRequirement
	var err error
	switch strongestMode {
	case FilterHoldModeRolling:
		if rollEv <= 0 {
			rollEv = defaultEvidenceRunes
		}
		if maxBuf <= 0 {
			maxBuf = defaultMaxBufferRunes
		}
		if maxBuf < rollEv {
			maxBuf = rollEv
		}
		req, err = NewFilterHoldRequirementRollingWithMaxBuffer(channel, blockingKinds, rollEv, maxBuf)
	case FilterHoldModeTerminalGate:
		if maxBuf <= 0 {
			maxBuf = defaultMaxBufferRunes
		}
		req, err = NewFilterHoldRequirementTerminalGateWithMaxBuffer(channel, blockingKinds, triggerKind, maxBuf)
	case FilterHoldModeFragmentGate:
		if maxBuf <= 0 {
			maxBuf = defaultMaxBufferRunes
		}
		req, err = NewFilterHoldRequirementFragmentGateWithMaxBuffer(channel, blockingKinds, triggerKind, maxBuf)
	}
	if err != nil {
		return FilterHoldRequirement{}, false
	}
	return req, true
}

// HasBlockingRequirement returns true when the given channel has at least one
// blocking requirement in the plan.
func (p EvidencePlan) HasBlockingRequirement(channel string) bool {
	bindings := p.plan.bindingsByChannel[channel]
	for _, b := range bindings {
		if b.BlocksRelease() && b.Requirement().IsBlocking() {
			return true
		}
	}
	return false
}

// BlockingRequirement returns the representative blocking requirement for the
// channel, or an error if no blocking requirement exists.
func (p EvidencePlan) BlockingRequirement(channel string) (FilterHoldRequirement, error) {
	req, ok := p.BlockingRequirementFor(channel)
	if !ok {
		return FilterHoldRequirement{}, errors.New("streamgate: no blocking requirement for channel: " + channel)
	}
	return req, nil
}

// SubscribeKinds returns a defensive copy of the merged subscribed kinds for the channel.
func (p EvidencePlan) SubscribeKinds(channel string) []EventKind {
	kinds := p.plan.allKinds[channel]
	if kinds == nil {
		return nil
	}
	out := make([]EventKind, len(kinds))
	copy(out, kinds)
	return out
}

// BlockingSubscribeKinds returns a defensive copy of the blocking subscribed
// kinds for the channel. These are the kinds that contribute to pending state.
// Returns nil when the channel has no blocking requirement.
func (p EvidencePlan) BlockingSubscribeKinds(channel string) []EventKind {
	kinds := p.plan.blockingKinds[channel]
	if kinds == nil {
		return nil
	}
	out := make([]EventKind, len(kinds))
	copy(out, kinds)
	return out
}

// BlockingKinds returns the union of event kinds that contribute to the
// channel's blocking pending state. Returns nil when the channel has no
// blocking requirement.
func (p EvidencePlan) BlockingKinds(channel string) []EventKind {
	kinds := p.plan.blockingKinds[channel]
	if kinds == nil {
		return nil
	}
	out := make([]EventKind, len(kinds))
	copy(out, kinds)
	return out
}

// ObserveKinds returns the union of event kinds that contribute to
// observation-only subscriptions but never to blocking pending state.
// Returns nil when the channel has no observe-only subscription.
func (p EvidencePlan) ObserveKinds(channel string) []EventKind {
	kinds := p.plan.observeOnlyKinds[channel]
	if kinds == nil {
		return nil
	}
	out := make([]EventKind, len(kinds))
	copy(out, kinds)
	return out
}

// AllBindings returns a defensive copy of all bindings in stable filter ID order.
func (p EvidencePlan) AllBindings() []FilterHoldBinding {
	if p.plan.allBindings == nil {
		return nil
	}
	out := make([]FilterHoldBinding, 0, len(p.plan.allBindings))
	for _, b := range p.plan.allBindings {
		out = append(out, b)
	}
	sort.SliceStable(out, func(i, j int) bool {
		return out[i].FilterID() < out[j].FilterID()
	})
	return out
}

func (p EvidencePlan) bindingIDs() []string {
	bindings := p.AllBindings()
	ids := make([]string, len(bindings))
	for i, b := range bindings {
		ids[i] = b.FilterID()
	}
	return ids
}

// MaxBufferRunes returns the compiled max buffer runes for the channel.
func (p EvidencePlan) MaxBufferRunes(channel string) int {
	return p.plan.maxBufferRunes[channel]
}

// RollingEvidenceRunes returns the compiled rolling evidence runes for the channel.
func (p EvidencePlan) RollingEvidenceRunes(channel string) int {
	return p.plan.rollingEvidenceRunes[channel]
}

// BindEpochFilters evaluates epoch applicability and binds resolved filters to EpochFilters in stable filter ID order.
func (p EvidencePlan) BindEpochFilters(epoch EvidenceEpoch, resolvedFilters []ResolvedFilter) ([]EpochFilter, error) {
	if epoch.ID() == 0 {
		return nil, errors.New("streamgate: epoch id must be positive")
	}
	planIDs := p.bindingIDs()
	if len(planIDs) == 0 && len(resolvedFilters) == 0 {
		return nil, nil
	}

	resolvedByID, err := indexResolvedFilters(resolvedFilters)
	if err != nil {
		return nil, err
	}

	appsByID, err := indexApplicabilities(epoch.Applicabilities())
	if err != nil {
		return nil, err
	}

	if err := requireExactFilterIDs(planIDs, resolvedByID, appsByID); err != nil {
		return nil, err
	}

	out := make([]EpochFilter, 0, len(planIDs))
	for _, id := range planIDs {
		rf := resolvedByID[id]
		app := appsByID[id]
		ef, err := rf.BindEpoch(epoch.ID(), app)
		if err != nil {
			return nil, fmt.Errorf("streamgate: bind epoch filter %s: %w", id, err)
		}
		out = append(out, ef)
	}

	return out, nil
}

func indexResolvedFilters(filters []ResolvedFilter) (map[string]ResolvedFilter, error) {
	out := make(map[string]ResolvedFilter, len(filters))
	for _, rf := range filters {
		fID := rf.FilterID()
		if fID == "" {
			return nil, errors.New("streamgate: resolved filter id is required")
		}
		if _, dup := out[fID]; dup {
			return nil, fmt.Errorf("streamgate: duplicate resolved filter id in bind epoch: %s", fID)
		}
		out[fID] = rf
	}
	return out, nil
}

func indexApplicabilities(apps []FilterApplicability) (map[string]FilterApplicability, error) {
	out := make(map[string]FilterApplicability, len(apps))
	for _, app := range apps {
		fID := app.FilterID()
		if fID == "" {
			return nil, errors.New("streamgate: applicability filter id is required")
		}
		if _, dup := out[fID]; dup {
			return nil, fmt.Errorf("streamgate: duplicate applicability filter id in bind epoch: %s", fID)
		}
		out[fID] = app
	}
	return out, nil
}

func requireExactFilterIDs(planIDs []string, resolvedByID map[string]ResolvedFilter, appsByID map[string]FilterApplicability) error {
	if len(resolvedByID) != len(planIDs) {
		return fmt.Errorf("streamgate: resolved filter count mismatch: got %d, expected %d", len(resolvedByID), len(planIDs))
	}
	if len(appsByID) != len(planIDs) {
		return fmt.Errorf("streamgate: applicability count mismatch: got %d, expected %d", len(appsByID), len(planIDs))
	}
	for _, id := range planIDs {
		if _, ok := resolvedByID[id]; !ok {
			return fmt.Errorf("streamgate: missing resolved filter for id %s", id)
		}
		if _, ok := appsByID[id]; !ok {
			return fmt.Errorf("streamgate: missing applicability for id %s", id)
		}
	}
	return nil
}

// EvidenceEpoch is an immutable snapshot of evidence state at the moment of
// a state transition (e.g. threshold reached, trigger fired). It carries
// only safe, externally-facing identification and filter applicabilities.
type EvidenceEpoch struct {
	id              uint64
	channel         string
	mode            FilterHoldMode
	triggered       bool
	reason          string
	applicabilities map[string]FilterApplicability
}

// NewEvidenceEpoch creates an EvidenceEpoch with the given parameters.
func NewEvidenceEpoch(id uint64, channel string, mode FilterHoldMode, triggered bool, reason string) (EvidenceEpoch, error) {
	return NewEvidenceEpochWithApplicabilities(id, channel, mode, triggered, reason, nil)
}

// NewEvidenceEpochWithApplicabilities creates an EvidenceEpoch carrying filter applicabilities.
func NewEvidenceEpochWithApplicabilities(
	id uint64, channel string, mode FilterHoldMode, triggered bool, reason string,
	apps []FilterApplicability,
) (EvidenceEpoch, error) {
	if channel == "" {
		return EvidenceEpoch{}, errors.New("streamgate: epoch channel is required")
	}
	if err := mode.Validate(); err != nil {
		return EvidenceEpoch{}, err
	}
	appMap := make(map[string]FilterApplicability, len(apps))
	for _, app := range apps {
		fID := app.FilterID()
		if fID == "" {
			return EvidenceEpoch{}, errors.New("streamgate: epoch applicability filter id is required")
		}
		if _, dup := appMap[fID]; dup {
			return EvidenceEpoch{}, fmt.Errorf("streamgate: duplicate epoch applicability filter id: %s", fID)
		}
		appMap[fID] = app
	}
	return EvidenceEpoch{
		id:              id,
		channel:         channel,
		mode:            mode,
		triggered:       triggered,
		reason:          reason,
		applicabilities: appMap,
	}, nil
}

// ID returns the epoch identifier.
func (e EvidenceEpoch) ID() uint64 { return e.id }

// Channel returns the epoch channel.
func (e EvidenceEpoch) Channel() string { return e.channel }

// Mode returns the epoch hold mode.
func (e EvidenceEpoch) Mode() FilterHoldMode { return e.mode }

// Triggered returns whether the epoch was a trigger event.
func (e EvidenceEpoch) Triggered() bool { return e.triggered }

// Reason returns the epoch reason string.
func (e EvidenceEpoch) Reason() string { return e.reason }

// ApplicabilityFor returns the applicability for the filter ID, if present.
func (e EvidenceEpoch) ApplicabilityFor(filterID string) (FilterApplicability, bool) {
	if e.applicabilities == nil {
		return FilterApplicability{}, false
	}
	app, ok := e.applicabilities[filterID]
	return app, ok
}

// Applicabilities returns a defensive copy of all applicabilities in stable filter ID order.
func (e EvidenceEpoch) Applicabilities() []FilterApplicability {
	if e.applicabilities == nil {
		return nil
	}
	out := make([]FilterApplicability, 0, len(e.applicabilities))
	for _, app := range e.applicabilities {
		out = append(out, app)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].filterID.value < out[j].filterID.value })
	return out
}

// EvidenceTailSignal identifies the kind of signal emitted by evidence tail
// operations.
type EvidenceTailSignal string

const (
	// EvidenceTailSignalNone indicates no signal was produced.
	EvidenceTailSignalNone EvidenceTailSignal = "none"

	// EvidenceTailSignalThreshold indicates the rolling rune threshold was
	// reached and a batch is ready.
	EvidenceTailSignalThreshold EvidenceTailSignal = "threshold"

	// EvidenceTailSignalTrigger indicates the trigger event (terminal or
	// fragment) was received.
	EvidenceTailSignalTrigger EvidenceTailSignal = "trigger"

	// EvidenceTailSignalBufferOverflow indicates the hard buffer limit was
	// exceeded. No release is produced.
	EvidenceTailSignalBufferOverflow EvidenceTailSignal = "overflow"

	// EvidenceTailSignalReady indicates the state is ready for release
	// (used when all accumulated events are eligible).
	EvidenceTailSignalReady EvidenceTailSignal = "ready"
)

// EvidenceTailSignalOverflow is a typed signal for buffer overflow. It carries
// only the channel, bound, and cause code; no raw content.
type EvidenceTailSignalOverflow struct {
	channel string
	bound   int
	code    string
}

// NewEvidenceTailSignalOverflow creates a typed overflow signal with the
// given parameters.
func NewEvidenceTailSignalOverflow(channel string, bound int, code string) (EvidenceTailSignalOverflow, error) {
	if channel == "" {
		return EvidenceTailSignalOverflow{}, errors.New("streamgate: overflow signal channel is required")
	}
	if bound <= 0 {
		return EvidenceTailSignalOverflow{}, errors.New("streamgate: overflow signal bound must be positive")
	}
	if code == "" {
		return EvidenceTailSignalOverflow{}, errors.New("streamgate: overflow signal cause code is required")
	}
	return EvidenceTailSignalOverflow{
		channel: channel,
		bound:   bound,
		code:    code,
	}, nil
}

// Channel returns the overflow channel.
func (s EvidenceTailSignalOverflow) Channel() string { return s.channel }

// Bound returns the overflow rune bound.
func (s EvidenceTailSignalOverflow) Bound() int { return s.bound }

// Code returns the overflow cause code.
func (s EvidenceTailSignalOverflow) Code() string { return s.code }

// EvidenceTail manages per-channel pending/look-behind state and tool-call
// fragment state for evidence holding. It is the core state machine that
// accumulates normalized events, validates UTF-8, counts Unicode runes, and
// produces epochs when thresholds or triggers are met.
type EvidenceTail struct {
	plan         EvidencePlan
	epochCounter uint64

	// channelState is keyed by channel name.
	channelState map[string]*channelState

	// epochs tracks epoch records by ID for validation during PrepareRelease
	// and ConfirmRelease. Each epoch is bound to a specific channel and can
	// only be prepared once.
	epochs map[uint64]*epochRecord

	// preparedByChannel tracks which channel has a pending prepared release
	// for each epoch, preventing overlapping prepared tokens. Prepared
	// replacements share this slot so a replacement and a release can never be
	// outstanding on the same channel at once.
	preparedByChannel map[string]string

	// replacements tracks in-flight prepared replacements by token. Each record
	// binds the replacement payload to the exact epoch, channel, and pending
	// sequence range it settles.
	replacements map[string]*replacementRecord

	// tokenNonce is a monotonic counter that ensures every generated release
	// token is globally unique. It prevents stale token reuse when the same
	// epoch is re-prepared after zero or partial confirm.
	tokenNonce uint64
}

// channelState holds the runtime state for a single channel.
type channelState struct {
	// pendingEntries is the ordered list of pending entry records for this channel.
	// Each entry tracks the event, its rune count, and associated fragment ID (if any).
	pendingEntries []pendingEntry

	// nextSequence is a monotonic counter that assigns an immutable identity
	// to each pending entry. It is never decremented or reused, even after
	// confirm removes entries from the front. This allows epoch records to
	// anchor their snapshot to a fixed sequence range rather than a mutable
	// slice index.
	nextSequence int

	// committedLookBehind is the bounded list of events already confirmed
	// released downstream. Used for cross-boundary repeat detection.
	committedLookBehind []NormalizedEvent

	// committedCursor is the total number of events confirmed released
	// downstream. This is a monotonic counter that increases with each
	// successful ConfirmRelease and resets on replace.
	committedCursor int

	// pendingRunes is the cumulative Unicode rune count of pendingEntries.
	pendingRunes int

	// effectiveEvidenceRunes is the rolling evidence window used for
	// look-behind trimming on this channel. For non-rolling modes it equals
	// maxBufferRunes.
	effectiveEvidenceRunes int

	// fragmentState tracks incomplete tool-call fragments keyed by toolCallID.
	fragmentState map[string]*fragmentState

	// maxBufferRunes is the hard buffer limit for this channel.
	maxBufferRunes int
}

// pendingEntry represents a single pending event with its metadata.
type pendingEntry struct {
	event      NormalizedEvent
	runes      int
	kind       EventKind
	toolCallID string // empty if not a fragment
	sequence   int    // monotonic sequence number
}

// fragmentState tracks an incomplete tool-call fragment.
// entries is the ordered list of pending entry sequences for this fragment ID.
// runes is the total rune count of all entries.
type fragmentState struct {
	toolCallID string
	entries    []int // sequence numbers of pending entries
	runes      int
	completed  bool // true when CompleteFragment has been called for this ID
}

// epochRecord tracks an epoch's state through its lifecycle. Each epoch
// is bound to an exact pending-entry sequence range captured at creation
// time; remainingSnapshot returns only entries within that range that
// have not yet been confirmed.
//
// A record is valid for PrepareRelease only when:
//   - consumed is false
//   - invalidated is false
//   - prepared and token carry a pending token whose channel matches
//
// Once invalidated (by recovery/continuation/terminal discard or a newer
// completion epoch), all token/prepared state is cleared and ConfirmRelease
// will reject any stale outstanding token with a deterministic error.
type epochRecord struct {
	epochID        uint64
	channel        string
	mode           FilterHoldMode // the hold mode at epoch creation
	consumed       bool           // true after full confirm or replace
	prepared       bool           // true after successful PrepareRelease
	token          string         // the prepared release token, if any
	snapshotSize   int            // number of events captured in the prepared snapshot
	snapStartSeq   int            // first pending-entry monotonic sequence in the snapshot
	snapEndSeq     int            // last pending-entry monotonic sequence (exclusive) in the snapshot
	confirmedCount int            // total events confirmed for this epoch across partial confirms
	invalidated    bool           // true after ResetForReplace/PrepareContinuation/DiscardPendingForTerminal
}

// replacementRecord tracks a prepared replacement through its lifecycle. It
// anchors the substituted payload to one epoch, that epoch's channel, and the
// immutable pending-entry sequence range the epoch captured, so confirm can
// settle exactly that range and nothing else.
type replacementRecord struct {
	token        string
	epochID      uint64
	channel      string
	snapStartSeq int               // first pending-entry sequence being replaced
	snapEndSeq   int               // last pending-entry sequence (exclusive) being replaced
	events       []NormalizedEvent // normalized replacement payload snapshot
}

// NewEvidenceTail creates a new EvidenceTail from the given compiled plan.
// It initializes per-channel state on first use.
func NewEvidenceTail(plan EvidencePlan) (*EvidenceTail, error) {
	if err := plan.validate(); err != nil {
		return nil, err
	}
	return &EvidenceTail{
		plan:              plan,
		epochCounter:      0,
		channelState:      make(map[string]*channelState),
		epochs:            make(map[uint64]*epochRecord),
		preparedByChannel: make(map[string]string),
		replacements:      make(map[string]*replacementRecord),
		tokenNonce:        0,
	}, nil
}

// validate checks the plan is consistent.
func (p EvidencePlan) validate() error {
	for fID, b := range p.plan.allBindings {
		if err := b.Requirement().Validate(); err != nil {
			return fmt.Errorf("streamgate: compiled plan filter %s: %v", fID, err)
		}
	}
	return nil
}

var (
	// errStalePreparedRelease is returned by ConfirmRelease when the token
	// has already been consumed, invalidated, or does not match the current
	// prepared state for the epoch's channel.
	errStalePreparedRelease = errors.New("streamgate: stale prepared release token")

	// errPreparedSnapshotMismatch is returned when the current pending
	// prefix does not match the sequence range captured at PrepareRelease.
	errPreparedSnapshotMismatch = errors.New("streamgate: prepared snapshot prefix mismatch")
)

// nextObservationEpoch generates an epoch ID for a non-blocking observation
// event. No epoch record is created; the epoch cannot be prepared or
// confirmed. This is used for events that pass through (unsubscribed,
// observe-only, sub-threshold, overflow) and must never acquire release
// capability.
func (t *EvidenceTail) nextObservationEpoch(channel string, mode FilterHoldMode, triggered bool, reason string) (EvidenceEpoch, error) {
	return t.nextObservationEpochWithApps(channel, mode, triggered, reason, nil)
}

func (t *EvidenceTail) nextObservationEpochWithApps(channel string, mode FilterHoldMode, triggered bool, reason string, apps []FilterApplicability) (EvidenceEpoch, error) {
	t.epochCounter++
	epoch, err := NewEvidenceEpochWithApplicabilities(t.epochCounter, channel, mode, triggered, reason, apps)
	if err != nil {
		return EvidenceEpoch{}, err
	}
	return epoch, nil
}

func (t *EvidenceTail) nextReadyEpochRecord(channel string, mode FilterHoldMode, triggered bool, reason string, firstSeq, lastSeqExclusive int) (EvidenceEpoch, *epochRecord, error) {
	return t.nextReadyEpochRecordWithApps(channel, mode, triggered, reason, firstSeq, lastSeqExclusive, nil)
}

func (t *EvidenceTail) nextReadyEpochRecordWithApps(channel string, mode FilterHoldMode, triggered bool, reason string, firstSeq, lastSeqExclusive int, apps []FilterApplicability) (EvidenceEpoch, *epochRecord, error) {
	t.epochCounter++
	epoch, err := NewEvidenceEpochWithApplicabilities(t.epochCounter, channel, mode, triggered, reason, apps)
	if err != nil {
		return EvidenceEpoch{}, nil, err
	}
	record := &epochRecord{
		epochID:      t.epochCounter,
		channel:      channel,
		mode:         mode,
		consumed:     false,
		prepared:     false,
		token:        "",
		snapshotSize: 0,
		snapStartSeq: firstSeq,
		snapEndSeq:   lastSeqExclusive,
	}
	t.epochs[t.epochCounter] = record
	return epoch, record, nil
}

// computeApplicabilities derives typed readiness and subscription status for each filter in the plan.
type applicabilityInput struct {
	event              *NormalizedEvent
	isTerminalTrigger  bool
	isFragmentComplete bool
	isHardBound        bool
	toolCallID         string
	eventBuffered      bool
}

type pendingStats struct {
	present bool
	runes   int
}

func subscribedPendingStats(cs *channelState, subscribedKinds []EventKind) pendingStats {
	var stats pendingStats
	if cs == nil || len(cs.pendingEntries) == 0 {
		return stats
	}
	for _, entry := range cs.pendingEntries {
		if isKindSubscribed(entry.kind, subscribedKinds) {
			stats.present = true
			stats.runes += entry.runes
		}
	}
	return stats
}

func deriveTriggerReady(
	b FilterHoldBinding,
	req FilterHoldRequirement,
	stats pendingStats,
	cs *channelState,
	input applicabilityInput,
) bool {
	if input.isHardBound {
		if b.BlocksRelease() && req.IsBlocking() {
			return true
		}
		return false
	}
	if input.isTerminalTrigger {
		switch req.Mode() {
		case FilterHoldModeTerminalGate:
			return input.event != nil && (input.event.Kind() == req.TriggerKind() || input.event.Kind() == EventKindTerminal || input.event.Kind() == EventKindProviderError)
		case FilterHoldModeRolling, FilterHoldModeFragmentGate, FilterHoldModeNone:
			return true
		}
		return false
	}
	if input.isFragmentComplete {
		switch req.Mode() {
		case FilterHoldModeFragmentGate:
			return cs != nil && cs.isFragmentCompleted(input.toolCallID)
		case FilterHoldModeRolling:
			return stats.runes >= req.EvidenceRunes()
		}
		return false
	}
	switch req.Mode() {
	case FilterHoldModeRolling:
		return stats.runes >= req.EvidenceRunes()
	case FilterHoldModeNone:
		return input.event != nil && input.event.Kind() == EventKindProviderError && isKindSubscribed(EventKindProviderError, req.SubscribedKinds())
	}
	return false
}

func (t *EvidenceTail) computeApplicabilities(
	channel string,
	event *NormalizedEvent,
	isTerminalTrigger bool,
	isFragmentComplete bool,
	isHardBound bool,
	toolCallID string,
	eventBuffered bool,
) ([]FilterApplicability, error) {
	allBindings := t.plan.AllBindings()
	if len(allBindings) == 0 {
		return nil, nil
	}

	cs := t.channelState[channel]
	input := applicabilityInput{
		event:              event,
		isTerminalTrigger:  isTerminalTrigger,
		isFragmentComplete: isFragmentComplete,
		isHardBound:        isHardBound,
		toolCallID:         toolCallID,
		eventBuffered:      eventBuffered,
	}

	apps := make([]FilterApplicability, 0, len(allBindings))
	for _, b := range allBindings {
		fID := b.FilterID()
		req := b.Requirement()

		if req.Channel() != channel {
			app, err := NewFilterApplicability(fID, false, false)
			if err != nil {
				return nil, err
			}
			apps = append(apps, app)
			continue
		}

		stats := subscribedPendingStats(cs, req.SubscribedKinds())
		if input.event != nil && !input.eventBuffered && isKindSubscribed(input.event.Kind(), req.SubscribedKinds()) {
			stats.present = true
			stats.runes += runeCountForEvent(*input.event)
		}

		subscribed := stats.present
		ready := deriveTriggerReady(b, req, stats, cs, input)

		app, err := NewFilterApplicability(fID, subscribed, ready)
		if err != nil {
			return nil, err
		}
		apps = append(apps, app)
	}

	return apps, nil
}

func (cs *channelState) unconfirmedSnapshot(record *epochRecord) []ReleaseEvent {
	var releaseEvents []ReleaseEvent
	for _, entry := range cs.pendingEntries {
		if entry.sequence >= record.snapStartSeq && entry.sequence < record.snapEndSeq {
			re, err := normalizedToReleaseEvent(entry.event)
			if err == nil {
				releaseEvents = append(releaseEvents, re)
			}
		}
	}
	return releaseEvents
}

func (t *EvidenceTail) validatePreparedOwnership(record *epochRecord, token string) error {
	if record.invalidated {
		return errStalePreparedRelease
	}
	if t.preparedByChannel[record.channel] != token {
		return errStalePreparedRelease
	}
	return nil
}

func (cs *channelState) pendingPrefixMatchesRange(n int, snapStartSeq int) bool {
	if n > len(cs.pendingEntries) {
		return false
	}
	for i := 0; i < n; i++ {
		if cs.pendingEntries[i].sequence != snapStartSeq+i {
			return false
		}
	}
	return true
}

// rangeEntryCount returns how many pending entries still fall inside the
// half-open sequence range captured by an epoch snapshot.
func (cs *channelState) rangeEntryCount(startSeq, endSeqExclusive int) int {
	count := 0
	for _, entry := range cs.pendingEntries {
		if entry.sequence >= startSeq && entry.sequence < endSeqExclusive {
			count++
		}
	}
	return count
}

// dropFragmentEntriesBelow removes fragment bookkeeping for pending entries
// whose sequence is already settled. Absolute sequences are preserved for the
// remaining entries so subsequent partial/full settlements stay correct.
func (cs *channelState) dropFragmentEntriesBelow(settledEndSeq int) {
	for id, fragment := range cs.fragmentState {
		var remaining []int
		for _, seq := range fragment.entries {
			if seq >= settledEndSeq {
				remaining = append(remaining, seq)
			}
		}
		if len(remaining) == 0 {
			delete(cs.fragmentState, id)
		} else {
			fragment.entries = remaining
		}
	}
}

func (t *EvidenceTail) getOrCreateChannel(channel string, maxBuffer int, evidenceRunes int) (*channelState, error) {
	cs, ok := t.channelState[channel]
	if !ok {
		effective := evidenceRunes
		if effective <= 0 {
			effective = maxBuffer
		}
		cs = &channelState{
			pendingEntries:         make([]pendingEntry, 0),
			fragmentState:          make(map[string]*fragmentState),
			maxBufferRunes:         maxBuffer,
			effectiveEvidenceRunes: effective,
		}
		t.channelState[channel] = cs
	}
	return cs, nil
}

func (t *EvidenceTail) Append(event NormalizedEvent) (EvidenceEpoch, EvidenceTailSignal, error) {
	if err := event.Validate(); err != nil {
		return EvidenceEpoch{}, EvidenceTailSignalNone, fmt.Errorf("streamgate: append event validate: %v", err)
	}

	channel := event.Channel()
	req, hasBlocking := t.plan.BlockingRequirementFor(channel)

	if err := validateEventUTF8(event); err != nil {
		return EvidenceEpoch{}, EvidenceTailSignalNone, err
	}

	if !hasBlocking {
		apps, err := t.computeApplicabilities(channel, &event, false, false, false, "", false)
		if err != nil {
			return EvidenceEpoch{}, EvidenceTailSignalNone, err
		}
		epoch, err := t.nextObservationEpochWithApps(channel, FilterHoldModeNone, false, "pass_through_no_requirement", apps)
		if err != nil {
			return EvidenceEpoch{}, EvidenceTailSignalNone, err
		}
		return epoch, EvidenceTailSignalReady, nil
	}

	if event.Kind() == EventKindTerminal || event.Kind() == EventKindProviderError || t.plan.IsTerminalTriggerForChannel(channel, event.Kind()) {
		cs, err := t.getOrCreateChannel(channel, t.plan.MaxBufferRunes(channel), t.plan.RollingEvidenceRunes(channel))
		if err != nil {
			return EvidenceEpoch{}, EvidenceTailSignalNone, err
		}
		safeEnd := cs.nextSequence
		firstSeq := cs.firstPendingSequence()
		apps, err := t.computeApplicabilities(channel, &event, true, false, false, "", false)
		if err != nil {
			return EvidenceEpoch{}, EvidenceTailSignalNone, err
		}
		epoch, _, err := t.nextReadyEpochRecordWithApps(
			channel, req.Mode(), true, "configured_trigger",
			firstSeq, safeEnd, apps,
		)
		if err != nil {
			return EvidenceEpoch{}, EvidenceTailSignalNone, err
		}
		return epoch, EvidenceTailSignalTrigger, nil
	}

	blockingKinds := t.plan.BlockingKinds(channel)
	observeKinds := t.plan.ObserveKinds(channel)

	if !isKindSubscribed(event.Kind(), blockingKinds) {
		reason := "event_not_subscribed"
		if isKindSubscribed(event.Kind(), observeKinds) {
			reason = "observe_only_pass_through"
		}
		apps, err := t.computeApplicabilities(channel, &event, false, false, false, "", false)
		if err != nil {
			return EvidenceEpoch{}, EvidenceTailSignalNone, err
		}
		epoch, err := t.nextObservationEpochWithApps(channel, req.Mode(), false, reason, apps)
		if err != nil {
			return EvidenceEpoch{}, EvidenceTailSignalNone, err
		}
		return epoch, EvidenceTailSignalNone, nil
	}

	cs, err := t.getOrCreateChannel(channel, t.plan.MaxBufferRunes(channel), t.plan.RollingEvidenceRunes(channel))
	if err != nil {
		return EvidenceEpoch{}, EvidenceTailSignalNone, err
	}

	runesAdded := runeCountForEvent(event)
	newPendingRunes := cs.pendingRunes + runesAdded

	if newPendingRunes > cs.maxBufferRunes {
		apps, err := t.computeApplicabilities(channel, &event, false, false, true, "", false)
		if err != nil {
			return EvidenceEpoch{}, EvidenceTailSignalNone, err
		}
		epoch, err := t.nextObservationEpochWithApps(channel, req.Mode(), false, "overflow_beyond_buffer_limit", apps)
		if err != nil {
			return EvidenceEpoch{}, EvidenceTailSignalNone, err
		}
		return epoch, EvidenceTailSignalBufferOverflow, nil
	}

	seq := cs.nextSequence
	cs.nextSequence++
	entry := pendingEntry{
		event:    cloneNormalizedEvent(event),
		runes:    runesAdded,
		kind:     event.Kind(),
		sequence: seq,
	}
	if event.Kind() == EventKindToolCallFragment {
		tc, err := event.AsToolCallFragment()
		if err == nil {
			entry.toolCallID = tc.ID
			fragment := cs.fragmentState[tc.ID]
			if fragment == nil {
				fragment = &fragmentState{
					toolCallID: tc.ID,
					entries:    []int{},
					runes:      0,
					completed:  false,
				}
				cs.fragmentState[tc.ID] = fragment
			}
			newFragmentRunes := fragment.runes + runesAdded
			if newFragmentRunes > cs.maxBufferRunes {
				return EvidenceEpoch{}, EvidenceTailSignalBufferOverflow, nil
			}
			fragment.entries = append(fragment.entries, entry.sequence)
			fragment.runes = newFragmentRunes
		}
	}
	cs.pendingEntries = append(cs.pendingEntries, entry)
	cs.pendingRunes = newPendingRunes

	apps, err := t.computeApplicabilities(channel, &event, false, false, false, "", true)
	if err != nil {
		return EvidenceEpoch{}, EvidenceTailSignalNone, err
	}
	anyReady := false
	hasRollingReady := false
	for _, app := range apps {
		if app.TriggerReady() {
			anyReady = true
			for _, b := range t.plan.BindingsForChannel(channel) {
				if b.FilterID() == app.FilterID() && b.Requirement().Mode() == FilterHoldModeRolling {
					hasRollingReady = true
				}
			}
		}
	}

	if anyReady {
		safeEnd := cs.nextSequence
		firstSeq := cs.firstPendingSequence()
		epoch, _, err := t.nextReadyEpochRecordWithApps(
			channel, req.Mode(), true, "threshold_or_trigger_reached",
			firstSeq, safeEnd, apps,
		)
		if err != nil {
			return EvidenceEpoch{}, EvidenceTailSignalNone, err
		}
		sig := EvidenceTailSignalTrigger
		if hasRollingReady {
			sig = EvidenceTailSignalThreshold
		}
		return epoch, sig, nil
	}

	epoch, err := t.nextObservationEpochWithApps(channel, req.Mode(), false, "sub_threshold_accumulating", apps)
	if err != nil {
		return EvidenceEpoch{}, EvidenceTailSignalNone, err
	}
	return epoch, EvidenceTailSignalNone, nil
}

func (cs *channelState) firstPendingSequence() int {
	if len(cs.pendingEntries) == 0 {
		return cs.nextSequence
	}
	return cs.pendingEntries[0].sequence
}

func (t *EvidenceTail) CompleteFragment(
	channel, toolCallID string,
) (EvidenceEpoch, EvidenceTailSignal, bool, error) {
	if channel == "" {
		return EvidenceEpoch{}, EvidenceTailSignalNone, false, errors.New("streamgate: complete fragment channel is required")
	}
	if toolCallID == "" {
		return EvidenceEpoch{}, EvidenceTailSignalNone, false, errors.New("streamgate: complete fragment tool call id is required")
	}

	if !t.plan.HasModeForChannel(channel, FilterHoldModeFragmentGate) {
		return EvidenceEpoch{}, EvidenceTailSignalNone, false, fmt.Errorf("streamgate: channel %s is not fragment_gate mode", channel)
	}

	cs, ok := t.channelState[channel]
	if !ok {
		return EvidenceEpoch{}, EvidenceTailSignalNone, false, nil
	}

	fragment, exists := cs.fragmentState[toolCallID]
	if !exists || fragment.completed {
		return EvidenceEpoch{}, EvidenceTailSignalNone, false, nil
	}

	fragment.completed = true

	safeLen := cs.fragmentSafePrefixLength()
	if safeLen == 0 {
		return EvidenceEpoch{}, EvidenceTailSignalNone, true, nil
	}

	t.markChannelEpochsInvalidated(channel)
	delete(t.preparedByChannel, channel)
	t.dropChannelReplacements(channel)

	firstSeq := cs.firstPendingSequence()
	safeEndSeq := firstSeq + safeLen
	apps, err := t.computeApplicabilities(channel, nil, false, true, false, toolCallID, false)
	if err != nil {
		return EvidenceEpoch{}, EvidenceTailSignalNone, true, err
	}
	epoch, _, err := t.nextReadyEpochRecordWithApps(
		channel, FilterHoldModeFragmentGate, true, "fragment_completion_safe_prefix",
		firstSeq, safeEndSeq, apps,
	)
	if err != nil {
		return EvidenceEpoch{}, EvidenceTailSignalNone, true, err
	}
	return epoch, EvidenceTailSignalTrigger, true, nil
}

// isFragmentCompleted returns true when the fragment for the given ID is completed.
func (cs *channelState) isFragmentCompleted(toolCallID string) bool {
	fragment, exists := cs.fragmentState[toolCallID]
	return exists && fragment.completed
}

// fragmentSafePrefixLength returns the index in pendingEntries up to which
// all fragments are completed and the prefix is contiguous. Returns 0 if no
// fragments are completed.
func (cs *channelState) fragmentSafePrefixLength() int {
	if len(cs.pendingEntries) == 0 {
		return 0
	}
	// Find the first incomplete fragment ID.
	firstIncompleteID := ""
	for _, entry := range cs.pendingEntries {
		if entry.kind == EventKindToolCallFragment && !cs.isFragmentCompleted(entry.toolCallID) {
			firstIncompleteID = entry.toolCallID
			break
		}
	}
	if firstIncompleteID == "" {
		// All fragments are completed.
		return len(cs.pendingEntries)
	}
	// Find the index of the first incomplete fragment.
	for i, entry := range cs.pendingEntries {
		if entry.toolCallID == firstIncompleteID {
			return i
		}
	}
	return 0
}

// PreparedRelease is an opaque token representing a prepared release that
// has not yet been confirmed. It is immutable and cannot be tampered with.
type PreparedRelease struct {
	token         string
	releaseEvents []ReleaseEvent
}

// Token returns the prepared release token string.
func (pr PreparedRelease) Token() string { return pr.token }

// ReleaseEvents returns a defensive copy of the prepared release events.
func (pr PreparedRelease) ReleaseEvents() []ReleaseEvent {
	if pr.releaseEvents == nil {
		return nil
	}
	out := make([]ReleaseEvent, len(pr.releaseEvents))
	copy(out, pr.releaseEvents)
	return out
}

// ReleaseConfirmation carries the number of events successfully released
// during a confirm operation.
type ReleaseConfirmation struct {
	ReleasedEvents int
}

// PrepareRelease creates a prepared release for the given epoch ID.
// Only epoch records that were created via a ready transition (threshold
// reached, trigger fired, fragment completion) carry release capability.
// Observation-only epochs (unsubscribed, overflow, observe-only, sub-threshold)
// have no record and cannot be prepared.
//
// PrepareRelease validates the epoch record exists and is not consumed or
// invalidated, generates a release snapshot from the epoch's fixed sequence
// range using unconfirmedSnapshot (which respects partial confirms), and
// creates an opaque token. It does NOT modify pending/look-behind/cursor
// state.
//
// PrepareRelease rejects: unknown/observation epochs, already consumed/
// invalidated epochs, overlapping prepared tokens for the same channel,
// and already-prepared epochs.
// HasUnconfirmedEvents returns true if the specified epoch has unconfirmed release payload events.
func (t *EvidenceTail) HasUnconfirmedEvents(epochID uint64) bool {
	if t == nil || t.epochs == nil {
		return false
	}
	record, ok := t.epochs[epochID]
	if !ok || record.consumed || record.invalidated {
		return false
	}
	channelCS := t.channelState[record.channel]
	if channelCS == nil {
		return false
	}
	return len(channelCS.unconfirmedSnapshot(record)) > 0
}

func (t *EvidenceTail) PrepareRelease(epochID uint64) (PreparedRelease, error) {
	// Validate epoch record exists and is not consumed or invalidated.
	record, ok := t.epochs[epochID]
	if !ok {
		return PreparedRelease{}, fmt.Errorf("streamgate: prepare release unknown or observation-only epoch %d", epochID)
	}
	if record.consumed || record.invalidated {
		return PreparedRelease{}, fmt.Errorf("streamgate: prepare release epoch %d already consumed or invalidated", epochID)
	}
	// Overlapping prepared token on the same channel is rejected.
	if t.preparedByChannel[record.channel] != "" {
		return PreparedRelease{}, fmt.Errorf("streamgate: prepare release channel %s already has prepared token %s", record.channel, t.preparedByChannel[record.channel])
	}
	if record.prepared && record.token != "" {
		return PreparedRelease{}, fmt.Errorf("streamgate: prepare release epoch %d already prepared with token %s", epochID, record.token)
	}

	channel := record.channel

	// Get the release snapshot from the epoch's fixed sequence range.
	// This uses monotonic sequence identity so it is safe after partial confirm.
	channelCS := t.channelState[channel]
	if channelCS == nil {
		return PreparedRelease{}, errors.New("streamgate: prepare release no channel state for epoch channel")
	}
	releaseEvents := channelCS.unconfirmedSnapshot(record)

	if len(releaseEvents) == 0 {
		return PreparedRelease{}, fmt.Errorf("streamgate: prepare release epoch %d has no unconfirmed events", epochID)
	}

	// Generate an opaque, globally unique token with a monotonic nonce.
	// This prevents stale token reuse when the same epoch is re-prepared
	// after zero or partial confirm (the snapshot content could be identical).
	t.tokenNonce++
	token := fmt.Sprintf("prepared-%d-%d-%d", epochID, len(releaseEvents), t.tokenNonce)

	// Store snapshot size for downstream ConfirmRelease validation.
	record.snapshotSize = len(releaseEvents)
	record.prepared = true
	record.token = token
	t.preparedByChannel[channel] = token

	return PreparedRelease{
		token:         token,
		releaseEvents: releaseEvents,
	}, nil
}

// ConfirmRelease applies the confirmed number of released events to the
// committed look-behind and cursor. Only this method modifies pending/
// look-behind/cursor state. Zero, partial, and full confirmations are
// supported. Stale tokens and duplicate confirms are rejected.
//
// ConfirmRelease validates: token format and existence, that the token
// matches the epoch's prepared token, that ReleasedEvents does not exceed
// the prepared count, and that the epoch has not been invalidated.
//
// ConfirmRelease validates:
//   - The token matches the epoch record and matches the current prepared
//     token for the epoch's channel.
//   - The epoch is not invalidated.
//   - The pending prefix starting at the epoch's snapStartSeq matches
//     the expected monotonic sequence.
//   - ReleasedEvents does not exceed the prepared count.
func (t *EvidenceTail) ConfirmRelease(token string, confirmation ReleaseConfirmation) error {
	if token == "" {
		return errors.New("streamgate: confirm release token is required")
	}
	if !strings.HasPrefix(token, "prepared-") {
		return errors.New("streamgate: confirm release invalid token format")
	}

	// Find the epoch record associated with this token.
	var targetEpoch *epochRecord
	for _, record := range t.epochs {
		if record.token == token {
			targetEpoch = record
			break
		}
	}
	if targetEpoch == nil {
		return errors.New("streamgate: confirm release token not found")
	}
	if targetEpoch.consumed {
		return fmt.Errorf("streamgate: confirm release epoch %d already consumed", targetEpoch.epochID)
	}

	// Validate ownership: token must match current prepared state and
	// epoch must not be invalidated.
	if err := t.validatePreparedOwnership(targetEpoch, token); err != nil {
		return err
	}

	channel := targetEpoch.channel
	cs, ok := t.channelState[channel]
	if !ok {
		return errors.New("streamgate: confirm release channel state not found")
	}

	n := confirmation.ReleasedEvents
	if n < 0 {
		return errors.New("streamgate: confirm release released events must be non-negative")
	}
	if n > targetEpoch.snapshotSize {
		return fmt.Errorf("streamgate: confirm release %d exceeds prepared snapshot %d", n, targetEpoch.snapshotSize)
	}

	// Validate that the pending prefix matches the epoch's sequence range.
	// This guards against a prepared token being confirmed after the pending
	// prefix has diverged from the original snapshot (e.g., post-append
	// contamination).
	if !cs.pendingPrefixMatchesRange(n, targetEpoch.snapStartSeq) {
		return errPreparedSnapshotMismatch
	}

	// Move confirmed events to look-behind.
	confirmed := make([]NormalizedEvent, n)
	for i := 0; i < n; i++ {
		confirmed[i] = cloneNormalizedEvent(cs.pendingEntries[i].event)
	}
	cs.committedLookBehind = append(cs.committedLookBehind, confirmed...)

	// Remove confirmed from pending.
	cs.pendingEntries = cs.pendingEntries[n:]

	// Adjust rune count.
	removedRunes := 0
	for i := 0; i < n; i++ {
		removedRunes += runeCountForEvent(confirmed[i])
	}
	cs.pendingRunes -= removedRunes

	// Update committed cursor.
	cs.committedCursor += n

	// Remove confirmed fragment states (if all their entries are confirmed).
	// Use the absolute confirmed range so that non-zero sequence fragments
	// clean up correctly: only entries already covered by this confirm are
	// removed, and their absolute sequences are preserved for subsequent
	// partial/full confirms.
	cs.dropFragmentEntriesBelow(targetEpoch.snapStartSeq + n)

	// Trim look-behind to bounded size using the channel's effective evidence
	// window (evidence_runes for rolling, maxBufferRunes otherwise).
	cs.committedLookBehind = trimLookBehind(cs.committedLookBehind, cs.effectiveEvidenceRunes)

	// Mark consumed when the full snapshot has been confirmed.
	if n >= targetEpoch.snapshotSize {
		targetEpoch.consumed = true
	}
	// Advance the epoch's unconfirmed start so that a subsequent re-prepare
	// or partial/full confirm sees only the remaining suffix of the same epoch.
	targetEpoch.snapStartSeq += n
	targetEpoch.confirmedCount += n
	// Clear prepared token regardless of completion.
	targetEpoch.prepared = false
	targetEpoch.token = ""
	delete(t.preparedByChannel, channel)

	return nil
}

// preparedReplacement is an opaque token representing a prepared replacement
// that has not yet been confirmed. It is immutable and cannot be tampered
// with, and it is only meaningful for the epoch it was prepared against.
type preparedReplacement struct {
	token         string
	releaseEvents []ReleaseEvent
}

// Token returns the prepared replacement token string.
func (pr preparedReplacement) Token() string { return pr.token }

// ReleaseEvents returns a defensive copy of the replacement release events.
func (pr preparedReplacement) ReleaseEvents() []ReleaseEvent {
	if pr.releaseEvents == nil {
		return nil
	}
	out := make([]ReleaseEvent, len(pr.releaseEvents))
	copy(out, pr.releaseEvents)
	return out
}

// prepareReplacement validates a replacement payload against the epoch it
// substitutes and reserves the epoch's channel, without mutating pending,
// look-behind, or cursor state. Every failure mode is checked here so a caller
// can hand the payload to the sink knowing the tail is still untouched:
// the epoch must exist as a releasable record that is neither consumed nor
// invalidated, its channel must not already carry an outstanding prepared
// token, the captured sequence range must still match the pending prefix, and
// the payload must be non-empty with every event releasable and on the epoch's
// own channel.
func (t *EvidenceTail) prepareReplacement(epochID uint64, events []NormalizedEvent) (preparedReplacement, error) {
	if len(events) == 0 {
		return preparedReplacement{}, errors.New("streamgate: prepare replacement requires at least one event")
	}
	record, ok := t.epochs[epochID]
	if !ok {
		return preparedReplacement{}, fmt.Errorf("streamgate: prepare replacement unknown or observation-only epoch %d", epochID)
	}
	if record.consumed || record.invalidated {
		return preparedReplacement{}, fmt.Errorf("streamgate: prepare replacement epoch %d already consumed or invalidated", epochID)
	}
	if t.preparedByChannel[record.channel] != "" {
		return preparedReplacement{}, fmt.Errorf("streamgate: prepare replacement channel %s already has prepared token %s", record.channel, t.preparedByChannel[record.channel])
	}
	if record.snapEndSeq < record.snapStartSeq {
		return preparedReplacement{}, fmt.Errorf("streamgate: prepare replacement epoch %d has an inverted sequence range", epochID)
	}

	channel := record.channel
	cs := t.channelState[channel]
	if cs == nil {
		return preparedReplacement{}, errors.New("streamgate: prepare replacement no channel state for epoch channel")
	}
	expectedCount := record.snapEndSeq - record.snapStartSeq
	targetCount := cs.rangeEntryCount(record.snapStartSeq, record.snapEndSeq)
	if targetCount != expectedCount {
		return preparedReplacement{}, errPreparedSnapshotMismatch
	}
	if expectedCount > 0 && !cs.pendingPrefixMatchesRange(expectedCount, record.snapStartSeq) {
		return preparedReplacement{}, errPreparedSnapshotMismatch
	}

	releaseEvents := make([]ReleaseEvent, len(events))
	payload := make([]NormalizedEvent, len(events))
	for i, ev := range events {
		if err := ev.Validate(); err != nil {
			return preparedReplacement{}, fmt.Errorf("streamgate: prepare replacement event %d: %v", i, err)
		}
		if ev.Channel() != channel {
			return preparedReplacement{}, fmt.Errorf("streamgate: prepare replacement event %d channel %s does not match epoch channel %s", i, ev.Channel(), channel)
		}
		relEv, err := normalizedToReleaseEvent(ev)
		if err != nil {
			return preparedReplacement{}, err
		}
		releaseEvents[i] = relEv
		payload[i] = cloneNormalizedEvent(ev)
	}

	// The nonce keeps the token globally unique so a stale replacement token
	// can never confirm a later replacement of the same epoch. The distinct
	// prefix keeps replacement tokens out of ConfirmRelease.
	t.tokenNonce++
	token := fmt.Sprintf("replacement-%d-%d-%d", epochID, len(releaseEvents), t.tokenNonce)
	t.replacements[token] = &replacementRecord{
		token:        token,
		epochID:      epochID,
		channel:      channel,
		snapStartSeq: record.snapStartSeq,
		snapEndSeq:   record.snapEndSeq,
		events:       payload,
	}
	t.preparedByChannel[channel] = token

	return preparedReplacement{
		token:         token,
		releaseEvents: releaseEvents,
	}, nil
}

// confirmReplacement settles a prepared replacement against the sink progress
// the boundary actually reported. The whole target range is consumed as
// replaced regardless of progress - the original events were superseded and can
// never reach the sink - but only the replacement prefix the sink accepted
// becomes committed look-behind and advances the cursor, so zero, partial, and
// full progress all leave the committed view equal to downstream reality.
//
// Only the target epoch's own range is touched: post-snapshot entries on the
// same channel stay pending, unrelated channels keep their pending entries,
// prepared tokens, and epoch records, and only same-channel epochs whose range
// overlaps the consumed original range are invalidated.
func (t *EvidenceTail) confirmReplacement(token string, confirmation ReleaseConfirmation) error {
	if token == "" {
		return errors.New("streamgate: confirm replacement token is required")
	}
	if !strings.HasPrefix(token, "replacement-") {
		return errors.New("streamgate: confirm replacement invalid token format")
	}
	replacement, ok := t.replacements[token]
	if !ok {
		return errStalePreparedRelease
	}
	record, ok := t.epochs[replacement.epochID]
	if !ok || record.consumed || record.invalidated {
		delete(t.replacements, token)
		return errStalePreparedRelease
	}
	if t.preparedByChannel[replacement.channel] != token {
		return errStalePreparedRelease
	}

	cs := t.channelState[replacement.channel]
	if cs == nil {
		return errors.New("streamgate: confirm replacement channel state not found")
	}

	n := confirmation.ReleasedEvents
	if n < 0 {
		return errors.New("streamgate: confirm replacement released events must be non-negative")
	}
	if n > len(replacement.events) {
		return fmt.Errorf("streamgate: confirm replacement %d exceeds prepared replacement %d", n, len(replacement.events))
	}

	expectedCount := replacement.snapEndSeq - replacement.snapStartSeq
	targetCount := cs.rangeEntryCount(replacement.snapStartSeq, replacement.snapEndSeq)
	if targetCount != expectedCount {
		return errPreparedSnapshotMismatch
	}
	if expectedCount > 0 && !cs.pendingPrefixMatchesRange(expectedCount, replacement.snapStartSeq) {
		return errPreparedSnapshotMismatch
	}

	// Consume the superseded original range without any downstream write.
	if targetCount > 0 {
		removedRunes := 0
		for i := 0; i < targetCount; i++ {
			removedRunes += cs.pendingEntries[i].runes
		}
		cs.pendingEntries = cs.pendingEntries[targetCount:]
		cs.pendingRunes -= removedRunes
		cs.dropFragmentEntriesBelow(replacement.snapEndSeq)
	}

	// Only the accepted replacement prefix becomes committed evidence.
	if n > 0 {
		released := make([]NormalizedEvent, n)
		for i := 0; i < n; i++ {
			released[i] = cloneNormalizedEvent(replacement.events[i])
		}
		cs.committedLookBehind = append(cs.committedLookBehind, released...)
		cs.committedCursor += n
		cs.committedLookBehind = trimLookBehind(cs.committedLookBehind, cs.effectiveEvidenceRunes)
	}

	record.consumed = true
	record.prepared = false
	record.token = ""
	record.snapshotSize = 0
	record.snapStartSeq = replacement.snapEndSeq
	t.invalidateOverlappingEpochs(replacement)

	delete(t.preparedByChannel, replacement.channel)
	delete(t.replacements, token)

	return nil
}

// invalidateOverlappingEpochs invalidates the still-open epoch records whose
// captured range on the replaced channel overlaps the consumed original range,
// so no stale token can confirm content that a replacement already superseded.
// Records on other channels, already settled records, and same-channel records
// covering only post-snapshot entries are left untouched.
func (t *EvidenceTail) invalidateOverlappingEpochs(replacement *replacementRecord) {
	for _, record := range t.epochs {
		if record.epochID == replacement.epochID || record.channel != replacement.channel {
			continue
		}
		if record.consumed || record.invalidated {
			continue
		}
		if record.snapStartSeq >= replacement.snapEndSeq || record.snapEndSeq <= replacement.snapStartSeq {
			continue
		}
		record.invalidated = true
		record.prepared = false
		record.token = ""
		record.snapshotSize = 0
	}
}

// ResetForReplace clears all attempt-local state including pending entries,
// look-behind, fragment state, prepared tokens, and the token nonce. Used
// when a new attempt replaces the current one.
func (t *EvidenceTail) ResetForReplace() {
	t.channelState = make(map[string]*channelState)
	t.epochs = make(map[uint64]*epochRecord)
	t.preparedByChannel = make(map[string]string)
	t.replacements = make(map[string]*replacementRecord)
	t.tokenNonce = 0
}

// PrepareContinuation preserves committed look-behind and cursor while
// discarding pending entries and prepared tokens. Used when resuming after
// a recovery continuation.
func (t *EvidenceTail) PrepareContinuation() {
	for _, cs := range t.channelState {
		cs.pendingEntries = nil
		cs.pendingRunes = 0
		cs.fragmentState = make(map[string]*fragmentState)
		// committedLookBehind, committedCursor are preserved.
	}
	t.markEpochsInvalidated()
	t.preparedByChannel = make(map[string]string)
	t.replacements = make(map[string]*replacementRecord)
}

// DiscardPendingForTerminal discards all pending entries and prepared tokens
// without releasing anything. Used before terminal error or when the
// terminal gate overflows. No downstream write occurs.
func (t *EvidenceTail) DiscardPendingForTerminal() {
	for _, cs := range t.channelState {
		cs.pendingEntries = nil
		cs.pendingRunes = 0
		cs.fragmentState = make(map[string]*fragmentState)
		// committedLookBehind and committedCursor are preserved.
	}
	t.markEpochsInvalidated()
	t.preparedByChannel = make(map[string]string)
	t.replacements = make(map[string]*replacementRecord)
}

// markEpochsInvalidated marks all epoch records as invalidated and clears
// all token/prepared/snapshot state. This prevents any stale outstanding
// token from being confirmed after recovery/continuation/terminal discard.
func (t *EvidenceTail) markEpochsInvalidated() {
	for _, record := range t.epochs {
		record.invalidated = true
		record.prepared = false
		record.token = ""
		record.snapshotSize = 0
	}
}

// dropChannelReplacements removes the prepared replacement records belonging to
// the given channel. Records on other channels are left untouched.
func (t *EvidenceTail) dropChannelReplacements(channel string) {
	for token, replacement := range t.replacements {
		if replacement.channel == channel {
			delete(t.replacements, token)
		}
	}
}

// markChannelEpochsInvalidated marks all epoch records belonging to the given
// channel as invalidated and clears all token/prepared/snapshot state for
// those records. This prevents stale outstanding tokens on the same channel
// from being confirmed after fragment completion supersedes an older safe
// prefix snapshot. It does NOT affect epoch records on other channels.
func (t *EvidenceTail) markChannelEpochsInvalidated(channel string) {
	for _, record := range t.epochs {
		if record.channel == channel {
			record.invalidated = true
			record.prepared = false
			record.token = ""
			record.snapshotSize = 0
		}
	}
}

// validateEventUTF8 checks that text/reasoning/tool argument content in the
// event is valid UTF-8. Invalid UTF-8 is rejected with a stable error.
func validateEventUTF8(event NormalizedEvent) error {
	var texts []string
	switch event.Kind() {
	case EventKindTextDelta:
		text, err := event.AsTextDelta()
		if err != nil {
			return err
		}
		texts = append(texts, text)
	case EventKindReasoningDelta:
		text, err := event.AsReasoningDelta()
		if err != nil {
			return err
		}
		texts = append(texts, text)
	case EventKindToolCallFragment:
		tc, err := event.AsToolCallFragment()
		if err != nil {
			return err
		}
		texts = append(texts, tc.Arguments)
	}
	for _, text := range texts {
		if text != "" && !utf8.ValidString(text) {
			return fmt.Errorf("streamgate: event %s contains invalid UTF-8", event.Kind())
		}
	}
	return nil
}

// runeCountForEvent returns the Unicode rune count for an event's text
// content. It operates on valid UTF-8 strings only.
func runeCountForEvent(event NormalizedEvent) int {
	var total int
	switch event.Kind() {
	case EventKindTextDelta:
		text, _ := event.AsTextDelta()
		total += utf8.RuneCountInString(text)
	case EventKindReasoningDelta:
		text, _ := event.AsReasoningDelta()
		total += utf8.RuneCountInString(text)
	case EventKindToolCallFragment:
		tc, _ := event.AsToolCallFragment()
		total += utf8.RuneCountInString(tc.Arguments)
	}
	return total
}

// normalizedToReleaseEvent converts a NormalizedEvent to a ReleaseEvent.
// Only releasable kinds (text_delta, reasoning_delta, tool_call_fragment)
// are supported.
func normalizedToReleaseEvent(ev NormalizedEvent) (ReleaseEvent, error) {
	switch ev.Kind() {
	case EventKindTextDelta:
		text, err := ev.AsTextDelta()
		if err != nil {
			return ReleaseEvent{}, err
		}
		return NewReleaseTextDeltaEvent(ev.Channel(), text, ev.Timestamp())
	case EventKindReasoningDelta:
		text, err := ev.AsReasoningDelta()
		if err != nil {
			return ReleaseEvent{}, err
		}
		return NewReleaseReasoningDeltaEvent(ev.Channel(), text, ev.Timestamp())
	case EventKindToolCallFragment:
		tc, err := ev.AsToolCallFragment()
		if err != nil {
			return ReleaseEvent{}, err
		}
		return NewReleaseToolCallFragmentEvent(ev.Channel(), tc.ID, tc.Name, tc.Arguments, ev.Timestamp())
	default:
		return ReleaseEvent{}, fmt.Errorf("streamgate: cannot release event kind %s", ev.Kind())
	}
}

// CommittedCursor returns the total number of events confirmed released
// downstream for the given channel. Returns 0 if the channel has no state.
func (t *EvidenceTail) CommittedCursor(channel string) int {
	cs, ok := t.channelState[channel]
	if !ok {
		return 0
	}
	return cs.committedCursor
}

// EffectiveLookBehind returns a defensive deep copy of the bounded look-behind
// events for the given channel. Returns nil if the channel has no state.
func (t *EvidenceTail) EffectiveLookBehind(channel string) []NormalizedEvent {
	cs, ok := t.channelState[channel]
	if !ok {
		return nil
	}
	if len(cs.committedLookBehind) == 0 {
		return nil
	}
	out := make([]NormalizedEvent, len(cs.committedLookBehind))
	for i, ev := range cs.committedLookBehind {
		out[i] = cloneNormalizedEvent(ev)
	}
	return out
}

// sliceEventAtRuneBoundary returns a new event with content sliced to at most
// maxRunes runes from the trailing edge. Returns an empty NormalizedEvent if
// the kind cannot be sliced (terminal, response_start, etc.).
func sliceEventAtRuneBoundary(ev NormalizedEvent, maxRunes int) (NormalizedEvent, error) {
	var text string
	switch ev.Kind() {
	case EventKindTextDelta:
		text, _ = ev.AsTextDelta()
	case EventKindReasoningDelta:
		text, _ = ev.AsReasoningDelta()
	case EventKindToolCallFragment:
		tc, _ := ev.AsToolCallFragment()
		text = tc.Arguments
	default:
		return NormalizedEvent{}, nil
	}

	runes := []rune(text)
	if len(runes) <= maxRunes {
		return ev, nil
	}
	sliced := string(runes[len(runes)-maxRunes:])

	switch ev.Kind() {
	case EventKindTextDelta:
		return NewTextDeltaEvent(ev.Channel(), sliced, ev.Timestamp())
	case EventKindReasoningDelta:
		return NewReasoningDeltaEvent(ev.Channel(), sliced, ev.Timestamp())
	case EventKindToolCallFragment:
		tc, _ := ev.AsToolCallFragment()
		return NewToolCallFragmentEvent(ev.Channel(), tc.ID, tc.Name, sliced, ev.Timestamp())
	}
	return NormalizedEvent{}, nil
}

// trimLookBehind trims the look-behind to at most maxRunes worth of events,
// keeping the most recent events. Single oversized events are sliced at
// Unicode rune boundaries to preserve partial content as a defensive snapshot.
func trimLookBehind(events []NormalizedEvent, maxRunes int) []NormalizedEvent {
	if len(events) == 0 || maxRunes <= 0 {
		return events
	}

	// Calculate rune counts for each event.
	runeCounts := make([]int, len(events))
	totalRunes := 0
	for i, ev := range events {
		runeCounts[i] = runeCountForEvent(ev)
		totalRunes += runeCounts[i]
	}

	if totalRunes <= maxRunes {
		return events
	}

	// Keep from the end until we're within the limit.
	var result []NormalizedEvent
	remainingRunes := maxRunes
	for i := len(events) - 1; i >= 0; i-- {
		if remainingRunes >= runeCounts[i] {
			result = append([]NormalizedEvent{events[i]}, result...)
			remainingRunes -= runeCounts[i]
		} else {
			// Slice this event at rune boundary to preserve partial content.
			sliced, err := sliceEventAtRuneBoundary(events[i], remainingRunes)
			if err == nil && sliced.Kind() != "" {
				result = append([]NormalizedEvent{sliced}, result...)
			}
			break
		}
	}

	return result
}