iop/packages/go/execution/liveness.go

package execution

import (
	"context"
	"errors"
	"math"
	"time"
)

// DefaultResponseStallTimeoutMS is the default response-stall timeout in
// milliseconds. It is used when no provider-configured value is available
// (zero wire value, omitted config, direct/legacy dispatch).
const DefaultResponseStallTimeoutMS = 300000

// maxSafeStallTimeoutMS is the largest millisecond value that can safely
// become a time.Duration without overflow. Values above this bound are
// rejected by the config validator and treated as invalid on the wire.
const maxSafeStallTimeoutMS = math.MaxInt64 / int64(time.Millisecond)

// ResolveStallTimeoutMS validates and normalizes a raw response-stall timeout
// value in one pass. It is the single validate-then-normalize entry point used
// by config load and both Node wire boundaries: zero maps to the documented
// default, safe positive values pass through unchanged, and negative or
// duration-overflow values return a StallTimeoutValidationError before any
// router or provider invocation. It never silently converts an invalid value.
func ResolveStallTimeoutMS(ms int64) (int64, error) {
	if err := ValidateStallTimeoutMS(ms); err != nil {
		return 0, err
	}
	if ms == 0 {
		return DefaultResponseStallTimeoutMS, nil
	}
	return ms, nil
}

// ValidateStallTimeoutMS returns nil when ms is zero (use default) or a
// positive value that can safely become a time.Duration in milliseconds.
// Negative values and values exceeding the safe duration bound are rejected.
// It is the single validation entry point used by config and the wire boundary.
func ValidateStallTimeoutMS(ms int64) error {
	if ms < 0 {
		return &StallTimeoutValidationError{
			Value: ms,
			Msg:   "response_stall_timeout_ms must be non-negative",
		}
	}
	if ms > maxSafeStallTimeoutMS {
		return &StallTimeoutValidationError{
			Value: ms,
			Msg:   "response_stall_timeout_ms exceeds safe duration bound",
		}
	}
	return nil
}

// StallTimeoutValidationError is returned when a response_stall_timeout_ms
// value is negative or exceeds the safe duration bound.
type StallTimeoutValidationError struct {
	Value int64
	Msg   string
}

func (e *StallTimeoutValidationError) Error() string {
	if e.Msg != "" {
		return e.Msg
	}
	return "invalid response_stall_timeout_ms"
}

// ProviderActivityDisposition classifies a provider output signal for the
// watchdog. The classifier is the single source of truth for progress and
// terminal decisions; handlers never switch on kind independently.
type ProviderActivityDisposition string

const (
	// DispositionNone means the signal carries no provider progress
	// information and must not reset the watchdog timer.
	DispositionNone ProviderActivityDisposition = "none"
	// DispositionStart establishes the initial baseline for the watchdog.
	// It is emitted once per run before any progress signals and lets the
	// observer record a known starting point without conflating that
	// transition with later progress resets.
	DispositionStart ProviderActivityDisposition = "start"
	// DispositionProgress means the provider is actively making progress
	// and must reset the watchdog timer.
	DispositionProgress ProviderActivityDisposition = "progress"
	// DispositionTerminal means the provider has produced a terminal
	// signal (complete, error, cancelled, end). The watchdog must stop
	// observing this run.
	DispositionTerminal ProviderActivityDisposition = "terminal"
)

// ClassifyRuntimeEvent classifies a RuntimeEvent into a ProviderActivityDisposition.
// Terminality is decided by the event type, never by token counts.
//
// Rules:
//   - start → DispositionStart
//   - complete/error/cancelled → DispositionTerminal (takes precedence over any payload/usage)
//   - non-terminal delta/reasoning_delta with non-empty delta/message or a usage observation → DispositionProgress
//   - empty/unknown type → DispositionNone
func ClassifyRuntimeEvent(event RuntimeEvent) ProviderActivityDisposition {
	switch event.Type {
	case EventTypeStart:
		return DispositionStart
	case EventTypeComplete, EventTypeError, EventTypeCancelled:
		return DispositionTerminal
	case EventTypeDelta, EventTypeReasoningDelta:
		if event.Delta != "" || event.Message != "" || event.Usage != nil {
			return DispositionProgress
		}
		return DispositionNone
	default:
		return DispositionNone
	}
}

// ClassifyProviderTunnelFrame classifies a ProviderTunnelFrame into a
// ProviderActivityDisposition.
//
// Rules:
//   - response_start (including headers) → DispositionProgress
//   - non-empty body → DispositionProgress
//   - usage frame → DispositionProgress
//   - end/error → DispositionTerminal (takes precedence over payload)
//   - empty/unknown kind → DispositionNone
func ClassifyProviderTunnelFrame(frame ProviderTunnelFrame) ProviderActivityDisposition {
	switch frame.Kind {
	case ProviderTunnelFrameKindEnd, ProviderTunnelFrameKindError:
		return DispositionTerminal
	case ProviderTunnelFrameKindResponseStart:
		// response_start with or without headers is progress.
		return DispositionProgress
	case ProviderTunnelFrameKindBody:
		if len(frame.Body) > 0 {
			return DispositionProgress
		}
		return DispositionNone
	case ProviderTunnelFrameKindUsage:
		// A usage frame is always progress for the tunnel path; the watchdog
		// observes token consumption as active provider work.
		return DispositionProgress
	default:
		return DispositionNone
	}
}

// ErrProbeUnsupported is carried in a ProbeOutcome when an adapter does not
// implement active provider probing. It is one of the inconclusive outcomes
// the probe normalizer collapses to HealthUnknown rather than treating as a
// definitive available or exact-target-unavailable result.
var ErrProbeUnsupported = errors.New("execution: adapter does not support provider probing")

// ProviderHealth is the stable, fail-closed classification of a provider's
// health as observed by a single bounded exact-target probe. It is the only
// value terminal assembly consumes from a probe: probe completion is evidence
// only and must never reset original request progress, change the attempt
// fence, or authorize retry.
type ProviderHealth string

const (
	// HealthUnknown is the fail-closed default. The probe could not establish
	// a definitive available or exact-target-unavailable result. Every error,
	// timeout, unsupported adapter, unknown status, and identity mismatch maps
	// here.
	HealthUnknown ProviderHealth = "health_unknown"
	// ProviderUnhealthy means a valid probe positively reported the exact
	// target as absent.
	ProviderUnhealthy ProviderHealth = "provider_unhealthy"
	// RequestStalled means a valid probe positively reported the exact target
	// as available, corroborating that the stalled request targets a live
	// target rather than a missing endpoint.
	RequestStalled ProviderHealth = "request_stalled"
)

// LivenessClassification is the stable, observable category a bounded
// exact-target probe outcome reduces to before it becomes a ProviderHealth.
// It exists so every fail-closed branch is independently testable; the
// normalizer is the single mapping from classification to health.
type LivenessClassification string

const (
	// LivenessAvailable means a valid probe reported the exact target present.
	LivenessAvailable LivenessClassification = "available"
	// LivenessUnavailable means a valid probe reported the exact target absent.
	LivenessUnavailable LivenessClassification = "unavailable"
	// LivenessTimeout means the bounded probe context expired before a result.
	LivenessTimeout LivenessClassification = "timeout"
	// LivenessError means the probe returned a transport, protocol, or decode
	// error that is not itself a definitive target-absent result.
	LivenessError LivenessClassification = "error"
	// LivenessUnsupported means the adapter does not implement active probing.
	LivenessUnsupported LivenessClassification = "unsupported"
	// LivenessUnknown means the probe returned an unrecognized status.
	LivenessUnknown LivenessClassification = "unknown"
	// LivenessIdentityMismatch means the probe identity did not match the
	// requested adapter or target identity.
	LivenessIdentityMismatch LivenessClassification = "identity_mismatch"
)

// ProbeOutcome is the typed, target-aware input to the fail-closed probe
// outcome normalizer. The coordinator validates and populates every field
// from a single bounded exact-target probe attempt; the normalizer never
// copies arbitrary provider metadata from it.
type ProbeOutcome struct {
	// AdapterName is the adapter identity reported by the probe result.
	AdapterName string
	// InstanceKey is the stable registry instance key reported by the probe.
	InstanceKey string
	// Target is the exact target reported by the probe result.
	Target string
	// Status is the normalized provider status reported by the probe.
	Status ProviderStatus
	// Err is the inconclusive error returned by the probe, if any.
	Err error
	// ExpectedAdapter is the adapter identity the caller required.
	ExpectedAdapter string
	// ExpectedInstance is the instance key the caller required; empty means the
	// caller does not pin a specific registry instance.
	ExpectedInstance string
	// ExpectedTarget is the exact target the caller required.
	ExpectedTarget string
}

// ClassifyProbeOutcome reduces a bounded exact-target probe outcome to its
// stable liveness classification. It is pure and fail-closed: any error, probe
// expiry, unsupported adapter, unknown status, or identity mismatch is an
// inconclusive classification rather than a definitive one. A returned error
// takes precedence over any reported status.
func ClassifyProbeOutcome(outcome ProbeOutcome) LivenessClassification {
	if outcome.Err != nil {
		if errors.Is(outcome.Err, context.Canceled) || errors.Is(outcome.Err, context.DeadlineExceeded) {
			return LivenessTimeout
		}
		if errors.Is(outcome.Err, ErrProbeUnsupported) {
			return LivenessUnsupported
		}
		return LivenessError
	}
	if !probeIdentityValid(outcome) {
		return LivenessIdentityMismatch
	}
	switch outcome.Status {
	case ProviderStatusAvailable:
		return LivenessAvailable
	case ProviderStatusUnavailable:
		return LivenessUnavailable
	default:
		return LivenessUnknown
	}
}

// HealthFromClassification maps a liveness classification to its stable
// ProviderHealth value. Available yields RequestStalled, unavailable yields
// ProviderUnhealthy, and every inconclusive classification yields
// HealthUnknown.
func HealthFromClassification(classification LivenessClassification) ProviderHealth {
	switch classification {
	case LivenessAvailable:
		return RequestStalled
	case LivenessUnavailable:
		return ProviderUnhealthy
	default:
		return HealthUnknown
	}
}

// NormalizeProbeOutcome maps a bounded exact-target probe outcome to its
// stable fail-closed ProviderHealth value. It is the composition of
// ClassifyProbeOutcome and HealthFromClassification: a validated matching
// available result yields RequestStalled, a validated matching unavailable
// result yields ProviderUnhealthy, and every error, timeout, unsupported
// adapter, unknown status, and identity mismatch yields HealthUnknown. It is
// pure and side-effect free.
func NormalizeProbeOutcome(outcome ProbeOutcome) ProviderHealth {
	return HealthFromClassification(ClassifyProbeOutcome(outcome))
}

// probeIdentityValid reports whether a probe result's adapter and target
// identity is non-empty and exactly matches what the caller required. When the
// caller pins an instance key, the probe must confirm it. An empty or
// mismatched identity is inconclusive and must fail closed.
func probeIdentityValid(outcome ProbeOutcome) bool {
	if outcome.AdapterName == "" || outcome.ExpectedAdapter == "" {
		return false
	}
	if outcome.Target == "" || outcome.ExpectedTarget == "" {
		return false
	}
	if outcome.AdapterName != outcome.ExpectedAdapter || outcome.Target != outcome.ExpectedTarget {
		return false
	}
	if outcome.ExpectedInstance != "" && outcome.InstanceKey != outcome.ExpectedInstance {
		return false
	}
	return true
}