gpt-load/internal/execution/contracts.go

// Package execution defines provider-neutral contracts for one selected upstream attempt.
package execution

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"time"

	"gpt-load/internal/outboundproxy"
	"gpt-load/internal/protocol"
	"gpt-load/internal/reasoning"
	"gpt-load/internal/usage"
)

// ValidationError identifies one invalid contract field without embedding secret values.
type ValidationError struct {
	Field  string
	Reason string
}

// Error implements error.
func (e *ValidationError) Error() string {
	return fmt.Sprintf("%s: %s", e.Field, e.Reason)
}

// Operation identifies one logical upstream operation.
type Operation string

const (
	OperationChatCompletion       Operation = "chat_completion"
	OperationResponsesCreate      Operation = "responses_create"
	OperationResponsesRetrieve    Operation = "responses_retrieve"
	OperationResponsesDelete      Operation = "responses_delete"
	OperationResponsesCancel      Operation = "responses_cancel"
	OperationResponsesInputItems  Operation = "responses_input_items"
	OperationResponsesCompact     Operation = "responses_compact"
	OperationResponsesInputTokens Operation = "responses_input_tokens"
	OperationCountTokens          Operation = "count_tokens"
	OperationResponsesPassthrough Operation = "responses_passthrough"
	OperationWebSearch            Operation = "web_search"
	OperationImagesGenerate       Operation = "images_generate"
	OperationImagesEdit           Operation = "images_edit"
	OperationEmbeddingsCreate     Operation = "embeddings_create"
	OperationRerank               Operation = "rerank"
	OperationListModels           Operation = "list_models"
	OperationProbe                Operation = "probe"
)

// Valid reports whether the operation is supported by the execution contract.
func (o Operation) Valid() bool {
	switch o {
	case OperationChatCompletion,
		OperationResponsesCreate,
		OperationResponsesRetrieve,
		OperationResponsesDelete,
		OperationResponsesCancel,
		OperationResponsesInputItems,
		OperationResponsesCompact,
		OperationResponsesInputTokens,
		OperationCountTokens,
		OperationResponsesPassthrough,
		OperationWebSearch,
		OperationImagesGenerate,
		OperationImagesEdit,
		OperationEmbeddingsCreate,
		OperationRerank,
		OperationListModels,
		OperationProbe:
		return true
	default:
		return false
	}
}

// ReplayPolicy describes the proof required before GPT-Load may dispatch the
// same logical request to another upstream candidate after a request may have
// reached an upstream.
type ReplayPolicy uint8

const (
	// ReplayPolicyLegacy preserves the established behavior of existing
	// operations. Individual health rules still decide whether retry is safe.
	ReplayPolicyLegacy ReplayPolicy = iota
	// ReplayPolicyRequireRejectedBeforeProcessing permits another dispatch only
	// when the provider explicitly proves that processing never started.
	ReplayPolicyRequireRejectedBeforeProcessing
)

// ReplayPolicy returns the operation-level replay contract.
func (o Operation) ReplayPolicy() ReplayPolicy {
	switch o {
	case OperationImagesGenerate, OperationImagesEdit, OperationEmbeddingsCreate, OperationRerank:
		return ReplayPolicyRequireRejectedBeforeProcessing
	default:
		return ReplayPolicyLegacy
	}
}

// RouteMode records the already-selected wire strategy for one attempt.
// Executors must execute this decision or reject it; they must not reroute it.
type RouteMode string

const (
	RouteNative    RouteMode = "native"
	RouteConverted RouteMode = "converted"
)

// Valid reports whether the route mode is recognized.
func (m RouteMode) Valid() bool {
	return m == RouteNative || m == RouteConverted
}

// RouteRequirement describes whether a request may use a converted route.
// It protects operation and provider-resource semantics, not optional model
// capabilities such as tools, reasoning, or multimodal input.
type RouteRequirement string

const (
	RouteRequirementAny    RouteRequirement = "any"
	RouteRequirementNative RouteRequirement = "native"
)

// Normalize maps the internal zero value to the default best-effort policy.
func (r RouteRequirement) Normalize() RouteRequirement {
	if r == "" {
		return RouteRequirementAny
	}
	return r
}

// Valid reports whether the normalized requirement is recognized.
func (r RouteRequirement) Valid() bool {
	switch r.Normalize() {
	case RouteRequirementAny, RouteRequirementNative:
		return true
	default:
		return false
	}
}

// Allows reports whether the selected wire route mode satisfies the native
// requirement. Target operation coverage is validated separately.
func (r RouteRequirement) Allows(mode RouteMode) bool {
	if !mode.Valid() || !r.Valid() {
		return false
	}
	return r.Normalize() == RouteRequirementAny || mode == RouteNative
}

// ResponsesStorePreference records the Responses Create storage requirement
// and whether an explicitly declared stateless compatibility route is allowed.
type ResponsesStorePreference string

const (
	ResponsesStorePreferenceNone         ResponsesStorePreference = ""
	ResponsesStorePreferencePreferStored ResponsesStorePreference = "prefer_stored"
	// 续接上一轮响应必须保留状态语义，不允许降级为无状态。
	ResponsesStorePreferenceRequireStored ResponsesStorePreference = "require_stored"
)

// Valid reports whether the Responses storage preference is recognized.
func (p ResponsesStorePreference) Valid() bool {
	return p == ResponsesStorePreferenceNone || p == ResponsesStorePreferencePreferStored ||
		p == ResponsesStorePreferenceRequireStored
}

// CredentialSnapshot is the exact logical credential selected for an attempt.
// Secret data is private and is never included in JSON.
type CredentialSnapshot struct {
	ID                 uint   `json:"id"`
	Version            uint64 `json:"version"`
	IdentityGeneration uint64 `json:"identity_generation"`

	data []byte
}

// NewCredentialSnapshot copies the credential data into a snapshot.
func NewCredentialSnapshot(id uint, version, identityGeneration uint64, data []byte) CredentialSnapshot {
	return CredentialSnapshot{
		ID:                 id,
		Version:            version,
		IdentityGeneration: identityGeneration,
		data:               cloneBytes(data),
	}
}

// Data returns an independent copy of the secret credential data.
func (c CredentialSnapshot) Data() []byte {
	return cloneBytes(c.data)
}

// Clone returns an independent credential snapshot.
func (c CredentialSnapshot) Clone() CredentialSnapshot {
	clone := c
	clone.data = cloneBytes(c.data)
	return clone
}

// MarshalJSON exposes only non-secret credential identity metadata.
func (c CredentialSnapshot) MarshalJSON() ([]byte, error) {
	return json.Marshal(struct {
		ID                 uint   `json:"id"`
		Version            uint64 `json:"version"`
		IdentityGeneration uint64 `json:"identity_generation"`
	}{
		ID:                 c.ID,
		Version:            c.Version,
		IdentityGeneration: c.IdentityGeneration,
	})
}

// AttemptTimeouts contains per-attempt timeout overrides. Zero inherits the caller default.
type AttemptTimeouts struct {
	FirstByte  time.Duration `json:"first_byte,omitempty"`
	Request    time.Duration `json:"request,omitempty"`
	StreamIdle time.Duration `json:"stream_idle,omitempty"`
}

// AttemptSpec is a fully selected, provider-neutral upstream attempt.
// NewAttemptSpec or Clone must be used at ownership boundaries because Query,
// Header, ConfiguredHeaders, Body, TargetConfig, and Credential contain reference-backed values.
type AttemptSpec struct {
	RequestID                string                   `json:"request_id"`
	AttemptID                string                   `json:"attempt_id"`
	Sequence                 uint32                   `json:"sequence"`
	ChannelID                string                   `json:"channel_id"`
	RouteMode                RouteMode                `json:"route_mode"`
	ClientProtocol           protocol.Protocol        `json:"client_protocol"`
	Operation                Operation                `json:"operation"`
	RouteRequirement         RouteRequirement         `json:"route_requirement"`
	ResponsesStorePreference ResponsesStorePreference `json:"responses_store_preference,omitempty"`
	ResponsesStoreDowngraded bool                     `json:"responses_store_downgraded,omitempty"`
	ClientModel              string                   `json:"client_model,omitempty"`
	UpstreamModel            string                   `json:"upstream_model,omitempty"`
	Method                   string                   `json:"method"`
	Path                     string                   `json:"path"`
	Query                    url.Values               `json:"query,omitempty"`
	// RawQuery preserves the original query bytes when exact forwarding matters.
	// It is mutually exclusive with Query and intentionally is not URL-decoded.
	RawQuery string      `json:"raw_query,omitempty"`
	Header   http.Header `json:"header,omitempty"`
	Body     []byte      `json:"body,omitempty"`
	// ConfiguredHeaders 记录显式请求头规则的字段；最终值由 Header 提供，缺失表示移除。
	ConfiguredHeaders []string `json:"-"`
	// IncludeUsage asks the executor to request provider usage details when the
	// selected operation supports an explicit wire option.
	IncludeUsage bool `json:"include_usage,omitempty"`
	// ForceCredentialRefresh is set only by GPT-Load after a provider explicitly
	// rejects this selected subscription credential before processing.
	ForceCredentialRefresh bool `json:"force_credential_refresh,omitempty"`
	// ContinuityKey is an opaque tenant-scoped value for provider-private
	// thinking/tool continuity. It is never persisted, logged, or exposed.
	ContinuityKey string `json:"-"`
	// TargetConfig is non-secret configuration resolved by the channel registry.
	TargetConfig     json.RawMessage         `json:"target_config,omitempty"`
	Timeouts         AttemptTimeouts         `json:"timeouts"`
	Credential       CredentialSnapshot      `json:"credential"`
	Proxy            outboundproxy.Effective `json:"-"`
	ProxyFingerprint string                  `json:"-"`
}

// NewAttemptSpec takes ownership of an independent clone of spec.
func NewAttemptSpec(spec AttemptSpec) AttemptSpec {
	spec.RouteRequirement = spec.RouteRequirement.Normalize()
	return spec.Clone()
}

// Clone returns an independent attempt specification.
func (s AttemptSpec) Clone() AttemptSpec {
	clone := s
	clone.Query = cloneValues(s.Query)
	clone.Header = cloneHeader(s.Header)
	clone.ConfiguredHeaders = append([]string(nil), s.ConfiguredHeaders...)
	clone.Body = cloneBytes(s.Body)
	clone.TargetConfig = cloneRawMessage(s.TargetConfig)
	clone.Credential = s.Credential.Clone()
	return clone
}

// DispatchState describes whether an attempt may have reached an upstream or
// was completed entirely within GPT-Load.
type DispatchState string

const (
	DispatchNotSent   DispatchState = "not_sent"
	DispatchMaybeSent DispatchState = "maybe_sent"
	DispatchLocal     DispatchState = "local"
)

// Valid reports whether the dispatch state is recognized.
func (s DispatchState) Valid() bool {
	return s == DispatchNotSent || s == DispatchMaybeSent || s == DispatchLocal
}

// MaxErrorSummaryLength bounds sanitized error messages retained by the contract.
const MaxErrorSummaryLength = 4096

// ErrorOrigin identifies the responsibility domain where a failure originated.
type ErrorOrigin string

const (
	ErrorOriginClient     ErrorOrigin = "client"
	ErrorOriginUpstream   ErrorOrigin = "upstream"
	ErrorOriginDownstream ErrorOrigin = "downstream"
	ErrorOriginInternal   ErrorOrigin = "internal"
)

// Valid reports whether the optional origin is recognized.
func (o ErrorOrigin) Valid() bool {
	switch o {
	case "", ErrorOriginClient, ErrorOriginUpstream, ErrorOriginDownstream, ErrorOriginInternal:
		return true
	default:
		return false
	}
}

// ErrorScope identifies the resource boundary affected by a failure.
type ErrorScope string

const (
	ErrorScopeRequest    ErrorScope = "request"
	ErrorScopeModel      ErrorScope = "model"
	ErrorScopeCredential ErrorScope = "credential"
	ErrorScopeGroup      ErrorScope = "group"
)

// Valid reports whether the optional scope is recognized.
func (s ErrorScope) Valid() bool {
	switch s {
	case "", ErrorScopeRequest, ErrorScopeModel, ErrorScopeCredential, ErrorScopeGroup:
		return true
	default:
		return false
	}
}

// ErrorKind is the stable category used for retry and health decisions.
type ErrorKind string

const (
	ErrorKindTransport             ErrorKind = "transport"
	ErrorKindTimeout               ErrorKind = "timeout"
	ErrorKindCanceled              ErrorKind = "canceled"
	ErrorKindHTTP                  ErrorKind = "http"
	ErrorKindProvider              ErrorKind = "provider"
	ErrorKindInvalidRequest        ErrorKind = "invalid_request"
	ErrorKindConversionUnsupported ErrorKind = "conversion_unsupported"
	ErrorKindInternal              ErrorKind = "internal"
)

const (
	ErrorCodeCriticalSemanticLoss         = "critical_semantic_loss"
	ErrorCodeTargetConversionNotSupported = "target_conversion_not_supported"
	ErrorCodeTargetSerializationFailed    = "target_serialization_failed"
)

// Valid reports whether the error kind is recognized.
func (k ErrorKind) Valid() bool {
	switch k {
	case ErrorKindTransport,
		ErrorKindTimeout,
		ErrorKindCanceled,
		ErrorKindHTTP,
		ErrorKindProvider,
		ErrorKindInvalidRequest,
		ErrorKindConversionUnsupported,
		ErrorKindInternal:
		return true
	default:
		return false
	}
}

// FailureHint is a provider-neutral classification hint. GPT-Load remains the
// owner of health and retry decisions.
type FailureHint string

const (
	FailureHintInvalidCredential       FailureHint = "invalid_credential"
	FailureHintRefreshRequired         FailureHint = "refresh_required"
	FailureHintRefreshUnavailable      FailureHint = "refresh_temporarily_unavailable"
	FailureHintReauthorizationRequired FailureHint = "reauthorization_required"
	FailureHintRateLimited             FailureHint = "rate_limited"
	FailureHintRequestRejected         FailureHint = "request_rejected"
	FailureHintCandidateUnavailable    FailureHint = "candidate_unavailable"
	FailureHintModelUnavailable        FailureHint = "model_unavailable"
	FailureHintHostError               FailureHint = "host_error"
)

// Valid reports whether the optional hint is recognized.
func (h FailureHint) Valid() bool {
	switch h {
	case "", FailureHintInvalidCredential, FailureHintRefreshRequired,
		FailureHintRefreshUnavailable,
		FailureHintReauthorizationRequired, FailureHintRateLimited,
		FailureHintRequestRejected, FailureHintCandidateUnavailable,
		FailureHintModelUnavailable, FailureHintHostError:
		return true
	default:
		return false
	}
}

// ReplaySafety records whether a provider has explicitly confirmed that the
// failed model request was rejected before processing.
type ReplaySafety string

const (
	ReplaySafetyUnknown                  ReplaySafety = "unknown"
	ReplaySafetyRejectedBeforeProcessing ReplaySafety = "rejected_before_processing"
)

func (s ReplaySafety) Valid() bool {
	return s == "" || s == ReplaySafetyUnknown || s == ReplaySafetyRejectedBeforeProcessing
}

// ErrorEvidence contains provider-neutral evidence for an unsuccessful attempt.
// Summary may retain a bounded, sanitized upstream error message; raw error
// bodies are never retained.
type ErrorEvidence struct {
	Kind         ErrorKind     `json:"kind"`
	Hint         FailureHint   `json:"hint,omitempty"`
	OriginHint   ErrorOrigin   `json:"origin_hint,omitempty"`
	ScopeHint    ErrorScope    `json:"scope_hint,omitempty"`
	StatusCode   int           `json:"status_code,omitempty"`
	Type         string        `json:"type,omitempty"`
	Code         string        `json:"code,omitempty"`
	Summary      string        `json:"summary,omitempty"`
	RequestID    string        `json:"request_id,omitempty"`
	RetryAfter   time.Duration `json:"retry_after,omitempty"`
	ReplaySafety ReplaySafety  `json:"replay_safety,omitempty"`
	Header       http.Header   `json:"-"`
}

// Clone returns an independent error evidence value.
func (e ErrorEvidence) Clone() ErrorEvidence {
	clone := e
	clone.Header = cloneHeader(e.Header)
	return clone
}

// UsageEvidence contains normalized usage and the raw provider evidence used to derive it.
type UsageEvidence struct {
	Normalized usage.Result `json:"normalized"`
	Raw        []byte       `json:"raw,omitempty"`
}

// Clone returns an independent usage evidence value.
func (e UsageEvidence) Clone() UsageEvidence {
	clone := e
	clone.Raw = cloneBytes(e.Raw)
	return clone
}

// AttemptResult is the terminal result of a non-streaming attempt.
// The result owns Header, Body, Usage, and Error after return from Executor.
type AttemptResult struct {
	DispatchState     DispatchState     `json:"dispatch_state"`
	ResponseStarted   bool              `json:"response_started"`
	UpstreamProtocol  protocol.Protocol `json:"upstream_protocol,omitempty"`
	AppliedReasoning  *reasoning.Config `json:"applied_reasoning,omitempty"`
	StatusCode        int               `json:"status_code,omitempty"`
	Header            http.Header       `json:"header,omitempty"`
	Body              []byte            `json:"body,omitempty"`
	Model             string            `json:"model,omitempty"`
	UpstreamRequestID string            `json:"upstream_request_id,omitempty"`
	Usage             *UsageEvidence    `json:"usage,omitempty"`
	Error             *ErrorEvidence    `json:"error,omitempty"`
}

// Clone returns an independent attempt result.
func (r AttemptResult) Clone() AttemptResult {
	clone := r
	clone.Header = cloneHeader(r.Header)
	clone.Body = cloneBytes(r.Body)
	if r.Usage != nil {
		usage := r.Usage.Clone()
		clone.Usage = &usage
	}
	if r.Error != nil {
		evidence := r.Error.Clone()
		clone.Error = &evidence
	}
	if r.AppliedReasoning != nil {
		reasoningConfig := r.AppliedReasoning.Clone()
		clone.AppliedReasoning = &reasoningConfig
	}
	return clone
}

// StreamEventKind identifies one streaming event category.
type StreamEventKind string

const (
	StreamEventReady StreamEventKind = "ready"
	StreamEventData  StreamEventKind = "data"
	StreamEventUsage StreamEventKind = "usage"
)

// Valid reports whether the streaming event kind is recognized.
func (k StreamEventKind) Valid() bool {
	return k == StreamEventReady || k == StreamEventData || k == StreamEventUsage
}

// StreamEvent is one ordered item emitted by a streaming attempt.
// Ownership of Header, Data, and Usage transfers to the stream sink. Executors
// must not mutate or reuse reference-backed event values after the sink call.
type StreamEvent struct {
	Sequence          uint64          `json:"sequence"`
	Kind              StreamEventKind `json:"kind"`
	StatusCode        int             `json:"status_code,omitempty"`
	Header            http.Header     `json:"header,omitempty"`
	UpstreamRequestID string          `json:"upstream_request_id,omitempty"`
	Data              []byte          `json:"data,omitempty"`
	Usage             *UsageEvidence  `json:"usage,omitempty"`
}

// Clone returns an independent stream event.
func (e StreamEvent) Clone() StreamEvent {
	clone := e
	clone.Header = cloneHeader(e.Header)
	clone.Data = cloneBytes(e.Data)
	if e.Usage != nil {
		usage := e.Usage.Clone()
		clone.Usage = &usage
	}
	return clone
}

// StreamResult is the terminal metadata returned after streaming ends.
// The result owns Header, Usage, and Error after return from Executor.
type StreamResult struct {
	DispatchState     DispatchState     `json:"dispatch_state"`
	ResponseStarted   bool              `json:"response_started"`
	UpstreamProtocol  protocol.Protocol `json:"upstream_protocol,omitempty"`
	AppliedReasoning  *reasoning.Config `json:"applied_reasoning,omitempty"`
	StatusCode        int               `json:"status_code,omitempty"`
	Header            http.Header       `json:"header,omitempty"`
	Model             string            `json:"model,omitempty"`
	UpstreamRequestID string            `json:"upstream_request_id,omitempty"`
	Usage             *UsageEvidence    `json:"usage,omitempty"`
	Error             *ErrorEvidence    `json:"error,omitempty"`
}

// Clone returns an independent streaming result.
func (r StreamResult) Clone() StreamResult {
	clone := r
	clone.Header = cloneHeader(r.Header)
	if r.Usage != nil {
		usage := r.Usage.Clone()
		clone.Usage = &usage
	}
	if r.Error != nil {
		evidence := r.Error.Clone()
		clone.Error = &evidence
	}
	if r.AppliedReasoning != nil {
		reasoningConfig := r.AppliedReasoning.Clone()
		clone.AppliedReasoning = &reasoningConfig
	}
	return clone
}

// StreamSink consumes events synchronously. Returning an error stops the stream.
type StreamSink func(StreamEvent) error

// Executor executes exactly the selected attempt described by AttemptSpec.
// Implementations must not mutate spec or retain reference-backed values from it.
type Executor interface {
	Execute(context.Context, AttemptSpec) AttemptResult
	ExecuteStream(context.Context, AttemptSpec, StreamSink) StreamResult
}

var _ json.Marshaler = CredentialSnapshot{}

func cloneBytes(value []byte) []byte {
	if value == nil {
		return nil
	}
	return append([]byte(nil), value...)
}

func cloneRawMessage(value json.RawMessage) json.RawMessage {
	if value == nil {
		return nil
	}
	return json.RawMessage(cloneBytes(value))
}

func cloneHeader(value http.Header) http.Header {
	if value == nil {
		return nil
	}
	return value.Clone()
}

func cloneValues(value url.Values) url.Values {
	if value == nil {
		return nil
	}
	clone := make(url.Values, len(value))
	for key, values := range value {
		clone[key] = append([]string(nil), values...)
	}
	return clone
}