pipeline

type Adapter ¶

type Adapter interface {
	Call(ctx context.Context, baseURL, apiKey string, body []byte, hdr http.Header, upstreamModel string, stream, oauth bool) (*http.Response, error)
	ExtractTokens(body []byte) sdkusage.Tokens
	Retryable(resp *http.Response) (retry bool, kind keypool.FailureKind, retryAfter time.Duration)
}

Adapter is the wire-shape seam, implemented by app/adapter.specAdapter.

upstreamModel is the resolved upstream model name (snapshot.Upstream()); stream reports whether the caller requested a streamed response. Most shapes ignore both (model + stream live in the request body), but shapes that encode them in the URL path — Gemini's generateContent vs streamGenerateContent — need them to build the upstream URL.

type HostHealth ¶ added in v0.2.0

type HostHealth interface {
	// Reachable marks the host as having answered (any HTTP response).
	Reachable(ctx context.Context, hostID string)
	// Unreachable marks the host as having failed to connect, with an error excerpt.
	Unreachable(ctx context.Context, hostID, errMsg string)
}

HostHealth records observed host reachability. Implemented by app/hosthealth.Recorder; wired in the composition root.

type KeyAgent ¶

type KeyAgent interface {
	OnFailure(ctx context.Context, keyID string, kind keypool.FailureKind, moreCandidates bool) (Verdict, string)
}

KeyAgent owns what happens when an upstream key fails. The request loop is deliberately dumb: it calls OnFailure and obeys the verdict. All the messy concerns (secret re-resolution, single-flight, park-and-wait, healing the snapshot, alerting) live behind this one method, off the request's import graph. A nil KeyAgent on the Pipeline preserves the legacy behavior (retryable → fail over, otherwise stop).

• moreCandidates reports whether another untried key remains. When true, the agent should fail over immediately (VerdictNext) and heal in the background — the request never waits. When false (this key is the last resort), the agent may block on a re-resolve and return VerdictRetry with a fresh value if the key had rotated.

The returned string is the fresh secret value, set only for VerdictRetry.

type Pipeline ¶

type Pipeline struct {
	Policy    *policy.Service
	Lifecycle *lifecycle.Registry
	Logger    *slog.Logger

	// KeyAgent decides what to do when an upstream key fails (fail over /
	// retry-with-refreshed-secret / stop). Nil = legacy behavior. The loop
	// only consults it for retryable failures.
	KeyAgent KeyAgent

	// HostHealth records observed upstream reachability for the admin UI.
	// Written only from the detached post-flight goroutines, never on the
	// latency path. Nil disables health recording.
	HostHealth HostHealth
}

Pipeline orchestrates requests. All policy-shaped work is delegated to Policy (the Service). Construct once at boot.

func (p *Pipeline) Run(ctx context.Context, req *Request) (res *Result, err error)

type Request ¶

type Request struct {
	Body    []byte
	Headers http.Header

	HostBaseURL string
	Adapter     Adapter

	Policy   *policy.Policy
	Model    *model.Model
	Host     *host.Host
	Provider string

	// Keys is the ordered candidate set. Empty → ErrNoKeys.
	Keys []*hostkey.HostKey

	ModelName string

	// UpstreamModel is the resolved upstream wire name (routing's
	// Plan.UpstreamModel() — the snapshot's upstream name, or the alias
	// verbatim override when resolution matched a declared alias). Passed
	// to Adapter.Call for shapes that put the model in the URL path.
	UpstreamModel string

	// Stream reports whether the caller requested a streamed response, passed
	// to Adapter.Call for shapes that select a distinct stream URL.
	Stream bool

	// MaxAttempts caps retries (0 → defaultMaxAttempts).
	MaxAttempts int

	// Lifecycle is the per-request shared context, constructed by the
	// handler before Run. Post-flight observers see it via the registered
	// lifecycle observers (Finalize). Optional — when nil, post-flight skips
	// dispatch (legacy callers / tests).
	Lifecycle *lifecycle.Context
}

Request is the pre-resolved input. Built by the handler/router.

func (r *Request) String() string

type Result ¶

type Result struct {
	Status  int
	Headers http.Header
	Body    io.ReadCloser
	KeyHash string
}

Result is what the handler streams back. Caller MUST Close Body — that triggers post-flight.

type UpstreamFailureError ¶

type UpstreamFailureError struct {
	Status int
	Body   string
	Cause  error
}

UpstreamFailureError wraps ErrAllKeysExhausted with the last upstream status + body excerpt so handlers can surface what actually went wrong (otherwise the caller just sees "all upstream keys failed" with no context — auth? quota? bad model? unknown).

func (e *UpstreamFailureError) Error() string

func (e *UpstreamFailureError) Unwrap() error

type UpstreamUnreachableError ¶ added in v0.2.0

type UpstreamUnreachableError struct {
	Host     string
	Attempts int
	Cause    error
}

UpstreamUnreachableError signals every attempt to reach the host failed at the connection-establishment phase — a likely baseURL misconfiguration or a down upstream, NOT a bad key. Distinct from UpstreamFailureError (which carries an actual upstream HTTP status the keys were rejected with).

func (e *UpstreamUnreachableError) Error() string

func (e *UpstreamUnreachableError) Unwrap() error

type Verdict ¶

type Verdict int

Verdict is what a KeyAgent tells the request to do after a key failed.

const (
	// VerdictNext: give up on this key, fail over to the next candidate.
	VerdictNext Verdict = iota
	// VerdictRetry: retry the SAME key with the returned fresh secret value
	// (the agent re-resolved it — e.g. the key had rotated upstream).
	VerdictRetry
	// VerdictFail: stop; return the upstream error to the caller.
	VerdictFail
)