iop/packages/go/streamgate/terminal.go

package streamgate

import (
	"context"
	"errors"
	"strconv"
	"time"
)

// CommitState records the commit status of a release payload to the
// downstream sink. Values follow the transport lifecycle: uncommitted,
// stream open after first user-visible write, and terminal committed.
type CommitState string

const (
	// CommitStateTransportUncommitted indicates the payload has not yet been
	// committed to the downstream sink.
	CommitStateTransportUncommitted CommitState = "transport_uncommitted"

	// CommitStateStreamOpen indicates the first user-visible write (HTTP
	// status/header flush, SSE/agent opening event, or first body chunk)
	// has occurred.
	CommitStateStreamOpen CommitState = "stream_open"

	// CommitStateTerminalCommitted indicates the terminal event has been
	// definitively sent to the downstream sink.
	CommitStateTerminalCommitted CommitState = "terminal_committed"
)

// Validate returns nil when the CommitState is a known lifecycle value.
func (cs CommitState) Validate() error {
	switch cs {
	case CommitStateTransportUncommitted, CommitStateStreamOpen, CommitStateTerminalCommitted:
		return nil
	}
	return errors.New("streamgate: unknown commit state: " + string(cs))
}

// MaxFailureCauses is the maximum number of failure causes retained in a
// FailureCauseChain. When the chain exceeds this limit, the oldest entry is
// dropped.
const MaxFailureCauses = 4

// StableToken is a bounded, ASCII-only identifier used across the event
// contract for stable references such as consumer ids, filter ids, rule ids,
// descriptor types/codes, template ids, param names, and recovery directive
// references. The grammar is [a-z0-9][a-z0-9._:-]* and the encoded value is
// at most 128 bytes. Accessors return only string; the type itself is
// unexported storage.
type StableToken struct {
	value string
}

// NewStableTokenRequired creates a StableToken that must be non-empty. It
// enforces the ASCII grammar and 128-byte limit.
func NewStableTokenRequired(name, v string) (StableToken, error) {
	if v == "" {
		return StableToken{}, errors.New("streamgate: " + name + " stable token is required")
	}
	return validateStableToken(name, v)
}

// NewStableTokenOptional creates a StableToken that may be empty. When empty
// the accessor returns "".
func NewStableTokenOptional(name, v string) (StableToken, error) {
	if v == "" {
		return StableToken{}, nil
	}
	return validateStableToken(name, v)
}

func validateStableToken(name, v string) (StableToken, error) {
	if len(v) > maxStableTokenBytes {
		return StableToken{}, errors.New("streamgate: " + name + " stable token exceeds maximum length")
	}
	for i := 0; i < len(v); i++ {
		b := v[i]
		if i == 0 {
			if !isAlphaLower(b) && !isDigit(b) {
				return StableToken{}, errors.New("streamgate: " + name + " stable token must start with [a-z0-9]")
			}
		} else {
			if !isAlphaLower(b) && !isDigit(b) && b != '.' && b != '_' && b != ':' && b != '-' {
				return StableToken{}, errors.New("streamgate: " + name + " stable token contains invalid character")
			}
		}
	}
	return StableToken{value: v}, nil
}

// validateStableTokenIfSet re-validates a non-empty StableToken value against
// the same grammar used at construction. Empty values pass through.
func validateStableTokenIfSet(name, v string) error {
	if v == "" {
		return nil
	}
	if _, err := validateStableToken(name, v); err != nil {
		return err
	}
	return nil
}

// validateStableTokenRequired re-validates a non-empty StableToken value against
// the same grammar used at construction. Empty values return a required error.
func validateStableTokenRequired(name, v string) error {
	if v == "" {
		return errors.New("streamgate: " + name + " stable token is required")
	}
	if _, err := validateStableToken(name, v); err != nil {
		return err
	}
	return nil
}

func isAlphaLower(b byte) bool { return b >= 'a' && b <= 'z' }
func isDigit(b byte) bool      { return b >= '0' && b <= '9' }

const maxStableTokenBytes = 128

// String returns the token value.
func (t StableToken) String() string { return t.value }

// FailureCause records a single step in a terminal failure with bounded,
// sanitized information. It contains only stage, code, consumer, filter, and
// rule id fields. Raw stack traces, provider body content, prompts, outputs,
// and authentication material are not permitted.
type FailureCause struct {
	stage    StableToken
	code     StableToken
	consumer StableToken
	filter   StableToken
	ruleID   StableToken
}

// NewFailureCause creates a FailureCause with validation. The returned value
// is immutable to callers.
func NewFailureCause(stage, code, consumer, filter, ruleID string) (FailureCause, error) {
	s, err := NewStableTokenRequired("stage", stage)
	if err != nil {
		return FailureCause{}, err
	}
	c, err := NewStableTokenRequired("code", code)
	if err != nil {
		return FailureCause{}, err
	}
	con, err := NewStableTokenOptional("consumer", consumer)
	if err != nil {
		return FailureCause{}, err
	}
	f, err := NewStableTokenOptional("filter", filter)
	if err != nil {
		return FailureCause{}, err
	}
	r, err := NewStableTokenOptional("ruleID", ruleID)
	if err != nil {
		return FailureCause{}, err
	}
	cause := FailureCause{
		stage:    s,
		code:     c,
		consumer: con,
		filter:   f,
		ruleID:   r,
	}
	if err := cause.Validate(); err != nil {
		return FailureCause{}, err
	}
	return cause, nil
}

// Validate returns nil when the FailureCause is in a consistent state.
// Re-validates nested StableToken values against the same grammar.
func (fc FailureCause) Validate() error {
	if err := validateStableTokenRequired("stage", fc.stage.value); err != nil {
		return err
	}
	if err := validateStableTokenRequired("code", fc.code.value); err != nil {
		return err
	}
	if err := validateStableTokenIfSet("consumer", fc.consumer.value); err != nil {
		return err
	}
	if err := validateStableTokenIfSet("filter", fc.filter.value); err != nil {
		return err
	}
	if err := validateStableTokenIfSet("ruleID", fc.ruleID.value); err != nil {
		return err
	}
	return nil
}

// Stage returns the failure cause stage token.
func (fc FailureCause) Stage() string { return fc.stage.value }

// Code returns the failure cause code token.
func (fc FailureCause) Code() string { return fc.code.value }

// Consumer returns the failure cause consumer token, or "" if not set.
func (fc FailureCause) Consumer() string { return fc.consumer.value }

// Filter returns the failure cause filter token, or "" if not set.
func (fc FailureCause) Filter() string { return fc.filter.value }

// RuleID returns the failure cause rule id token, or "" if not set.
func (fc FailureCause) RuleID() string { return fc.ruleID.value }

// FailureCauseChain is a bounded sequence of FailureCause entries that
// describes the full chain of a terminal failure. It retains at most
// MaxFailureCauses entries; when the chain exceeds this limit, the oldest
// entry is dropped. All public accessors return defensive copies.
type FailureCauseChain struct {
	causes []FailureCause
}

// NewFailureCauseChain creates a FailureCauseChain from a slice of causes.
// All causes are validated, and the chain is capped to the latest
// MaxFailureCauses entries. The returned chain is a defensive copy.
func NewFailureCauseChain(causes []FailureCause) (FailureCauseChain, error) {
	if len(causes) == 0 {
		return FailureCauseChain{}, nil
	}
	cp := make([]FailureCause, len(causes))
	copy(cp, causes)
	for i, c := range cp {
		if err := c.Validate(); err != nil {
			return FailureCauseChain{}, errors.New("streamgate: failure cause chain entry " + strconv.Itoa(i) + ": " + err.Error())
		}
	}
	// Cap to latest MaxFailureCauses.
	if len(cp) > MaxFailureCauses {
		cp = cp[len(cp)-MaxFailureCauses:]
	}
	return FailureCauseChain{causes: cp}, nil
}

// Append adds a cause to the chain. The input cause is validated. If the
// chain already contains MaxFailureCauses entries, the oldest entry is dropped
// first. The returned chain is a defensive copy.
func (c FailureCauseChain) Append(cause FailureCause) (FailureCauseChain, error) {
	if err := cause.Validate(); err != nil {
		return FailureCauseChain{}, err
	}
	cp := make([]FailureCause, len(c.causes))
	copy(cp, c.causes)
	cp = append(cp, cause)
	if len(cp) > MaxFailureCauses {
		cp = cp[len(cp)-MaxFailureCauses:]
	}
	return FailureCauseChain{causes: cp}, nil
}

// Copy returns a defensive copy of the failure cause chain.
func (c FailureCauseChain) Copy() FailureCauseChain {
	if c.causes == nil {
		return FailureCauseChain{}
	}
	out := make([]FailureCause, len(c.causes))
	copy(out, c.causes)
	return FailureCauseChain{causes: out}
}

// Len returns the number of causes in the chain.
func (c FailureCauseChain) Len() int {
	return len(c.causes)
}

// At returns a defensive copy of the cause at the given index.
func (c FailureCauseChain) At(i int) (FailureCause, error) {
	if i < 0 || i >= len(c.causes) {
		return FailureCause{}, errors.New("streamgate: failure cause chain index out of range")
	}
	cp := c.causes[i]
	return cp, nil
}

// All returns a defensive copy of all causes in the chain.
func (c FailureCauseChain) All() []FailureCause {
	if c.causes == nil {
		return nil
	}
	out := make([]FailureCause, len(c.causes))
	copy(out, c.causes)
	return out
}

// ExternalDescriptor carries only safe, external-facing information about a
// terminal failure. It must not contain raw stack traces, provider body
// content, prompts, outputs, or authentication material. The type field is a
// stable token, the code is an optional stable code, the message is a safe
// template, and the param is an optional sanitized parameter.
type ExternalDescriptor struct {
	errorType StableToken
	code      StableToken
	message   StableToken
	param     StableToken
}

// NewExternalDescriptor creates an ExternalDescriptor with validation. The
// returned value is immutable to callers.
func NewExternalDescriptor(errorType, code, message, param string) (ExternalDescriptor, error) {
	et, err := NewStableTokenRequired("errorType", errorType)
	if err != nil {
		return ExternalDescriptor{}, err
	}
	c, err := NewStableTokenOptional("code", code)
	if err != nil {
		return ExternalDescriptor{}, err
	}
	m, err := NewStableTokenRequired("message", message)
	if err != nil {
		return ExternalDescriptor{}, err
	}
	p, err := NewStableTokenOptional("param", param)
	if err != nil {
		return ExternalDescriptor{}, err
	}
	desc := ExternalDescriptor{
		errorType: et,
		code:      c,
		message:   m,
		param:     p,
	}
	if err := desc.Validate(); err != nil {
		return ExternalDescriptor{}, err
	}
	return desc, nil
}

// Validate returns nil when the ExternalDescriptor is in a consistent state.
// Re-validates nested StableToken values against the same grammar.
func (ed ExternalDescriptor) Validate() error {
	if err := validateStableTokenRequired("errorType", ed.errorType.value); err != nil {
		return err
	}
	if err := validateStableTokenIfSet("code", ed.code.value); err != nil {
		return err
	}
	if err := validateStableTokenRequired("message", ed.message.value); err != nil {
		return err
	}
	if err := validateStableTokenIfSet("param", ed.param.value); err != nil {
		return err
	}
	return nil
}

// Type returns the error type token.
func (ed ExternalDescriptor) Type() string { return ed.errorType.value }

// Code returns the optional stable code.
func (ed ExternalDescriptor) Code() string { return ed.code.value }

// Message returns the safe message template.
func (ed ExternalDescriptor) Message() string { return ed.message.value }

// Param returns the optional sanitized parameter.
func (ed ExternalDescriptor) Param() string { return ed.param.value }

// TerminalResult delivers a single, typed terminal outcome for a channel. It
// carries either success or error with a safe external descriptor. Exactly
// one of Success or Error must be set; both or neither is invalid. All fields
// are privately stored with defensive copies on access.
type TerminalResult struct {
	channel       string
	success       bool
	err           bool
	externalDesc  *ExternalDescriptor
	failureCauses FailureCauseChain
	occurredAt    time.Time
}

// NewSuccessTerminalResult creates a TerminalResult of kind success. The
// result must not carry an external descriptor or failure causes.
func NewSuccessTerminalResult(channel string, occurredAt time.Time) (TerminalResult, error) {
	if channel == "" {
		return TerminalResult{}, errors.New("streamgate: terminal result channel is required")
	}
	if occurredAt.IsZero() {
		return TerminalResult{}, errors.New("streamgate: terminal result occurred timestamp is required")
	}
	tr := TerminalResult{
		channel:    channel,
		success:    true,
		occurredAt: occurredAt,
	}
	if err := tr.Validate(); err != nil {
		return TerminalResult{}, err
	}
	return tr, nil
}

// NewErrorTerminalResult creates a TerminalResult of kind error. It requires
// a validated external descriptor and an optional bounded failure cause chain.
func NewErrorTerminalResult(
	channel string,
	desc ExternalDescriptor,
	causes FailureCauseChain,
	occurredAt time.Time,
) (TerminalResult, error) {
	if channel == "" {
		return TerminalResult{}, errors.New("streamgate: terminal result channel is required")
	}
	if occurredAt.IsZero() {
		return TerminalResult{}, errors.New("streamgate: terminal result occurred timestamp is required")
	}
	if err := desc.Validate(); err != nil {
		return TerminalResult{}, err
	}
	tr := TerminalResult{
		channel:       channel,
		err:           true,
		externalDesc:  copyExternalDescriptor(&desc),
		failureCauses: causes.Copy(),
		occurredAt:    occurredAt,
	}
	if err := tr.Validate(); err != nil {
		return TerminalResult{}, err
	}
	return tr, nil
}

// NewTerminalResult is retained for internal use only. Prefer
// NewSuccessTerminalResult and NewErrorTerminalResult for new code.
func newTerminalResult(
	channel string,
	success, errResult bool,
	desc *ExternalDescriptor,
	causes FailureCauseChain,
	occurredAt time.Time,
) (TerminalResult, error) {
	if channel == "" {
		return TerminalResult{}, errors.New("streamgate: terminal result channel is required")
	}
	if !success && !errResult {
		return TerminalResult{}, errors.New("streamgate: terminal result must be either success or error")
	}
	if success && errResult {
		return TerminalResult{}, errors.New("streamgate: terminal result cannot be both success and error")
	}
	if errResult && desc == nil {
		return TerminalResult{}, errors.New("streamgate: terminal result error requires an external descriptor")
	}
	if occurredAt.IsZero() {
		return TerminalResult{}, errors.New("streamgate: terminal result occurred timestamp is required")
	}

	cpDesc := desc
	if desc != nil {
		descCopy := *desc
		cpDesc = &descCopy
	}

	cpCauses := causes.Copy()

	tr := TerminalResult{
		channel:       channel,
		success:       success,
		err:           errResult,
		externalDesc:  cpDesc,
		failureCauses: cpCauses,
		occurredAt:    occurredAt,
	}
	if err := tr.Validate(); err != nil {
		return TerminalResult{}, err
	}
	return tr, nil
}

// Validate returns nil when the TerminalResult is in a consistent state.
func (tr TerminalResult) Validate() error {
	if tr.channel == "" {
		return errors.New("streamgate: terminal result channel is required")
	}
	if !tr.success && !tr.err {
		return errors.New("streamgate: terminal result must be either success or error")
	}
	if tr.success && tr.err {
		return errors.New("streamgate: terminal result cannot be both success and error")
	}
	if tr.err {
		if tr.externalDesc == nil {
			return errors.New("streamgate: terminal result error requires an external descriptor")
		}
		if err := tr.externalDesc.Validate(); err != nil {
			return err
		}
		if tr.failureCauses.Len() > MaxFailureCauses {
			return errors.New("streamgate: terminal result failure cause chain exceeds maximum")
		}
		for i := 0; i < tr.failureCauses.Len(); i++ {
			c, err := tr.failureCauses.At(i)
			if err != nil {
				return errors.New("streamgate: terminal result failure cause chain: " + err.Error())
			}
			if err := c.Validate(); err != nil {
				return errors.New("streamgate: terminal result failure cause chain entry " + strconv.Itoa(i) + ": " + err.Error())
			}
		}
	}
	if tr.success {
		if tr.externalDesc != nil {
			return errors.New("streamgate: terminal result success must not have an external descriptor")
		}
		if tr.failureCauses.Len() > 0 {
			return errors.New("streamgate: terminal result success must not have failure causes")
		}
	}
	if tr.occurredAt.IsZero() {
		return errors.New("streamgate: terminal result occurred timestamp is required")
	}
	return nil
}

// Channel returns the terminal result channel.
func (tr TerminalResult) Channel() string { return tr.channel }

// Success returns whether this is a success terminal result.
func (tr TerminalResult) Success() bool { return tr.success }

// Error returns whether this is an error terminal result.
func (tr TerminalResult) Error() bool { return tr.err }

// ExternalDesc returns a defensive copy of the external descriptor, or nil
// for success results.
func (tr TerminalResult) ExternalDesc() *ExternalDescriptor {
	if tr.externalDesc == nil {
		return nil
	}
	cp := *tr.externalDesc
	return &cp
}

// FailureCauses returns a defensive copy of the failure cause chain.
func (tr TerminalResult) FailureCauses() FailureCauseChain {
	return tr.failureCauses.Copy()
}

// OccurredAt returns the timestamp when the terminal result occurred.
func (tr TerminalResult) OccurredAt() time.Time { return tr.occurredAt }

// ReleaseSink is the host interface that the stream gate calls to commit
// release payloads to the downstream sink. The actual state machine
// implementation belongs to a subsequent Milestone Task; this interface
// definition provides the contract only.
//
// CommitResponseStart, Release, and CommitTerminal are mutually exclusive
// roles: response-start, safe event, and terminal each have a dedicated
// entry point. ReleaseEvent must never carry response-start or terminal
// payloads; those are handled exclusively through the dedicated methods.
type ReleaseSink interface {
	// CommitResponseStart commits a staged response start to the downstream
	// sink. It returns the resulting commit state and any error.
	CommitResponseStart(ctx context.Context, rs ResponseStart) (CommitState, error)

	// Release commits a release event to the downstream sink. It returns the
	// resulting commit state and any error.
	Release(ctx context.Context, ev ReleaseEvent) (CommitState, error)

	// CommitTerminal commits a terminal result to the downstream sink. It
	// returns the resulting commit state and any error.
	CommitTerminal(ctx context.Context, tr TerminalResult) (CommitState, error)
}