lifecycle

type Collector ¶

type Collector interface {
	Collect(lc *Context)
}

Collector (the janitor) runs once at the end of the lifecycle, after every Hook has filled and the Registry has attached their results to the Context. It reads the collected results off the Context (via Context.Collected) and routes them to sinks — the "store" half of the produce → attach → store flow.

Collectors run in parallel and treat the Context as read-only. Storing must be non-blocking (push onto a bounded channel); a Collector must never block the post-flight goroutine.

type Context ¶

type Context struct {
	RequestID string
	Source    string // runner label: "pipeline" | "proxy" | "ws" | "batch"

	// Timing carries the per-request checkpoints. Timing.Start is the
	// absolute anchor (set at construction); the runner stamps the
	// upstream + end marks as the request progresses. See timing.go.
	Timing Timing

	// Streamed reports whether the response was streamed back to the
	// caller. Set by the runner once known (request flag in pipeline,
	// upstream Content-Type in proxy).
	Streamed bool

	// RequestedModel is the model identifier the caller asked for, as it
	// arrived on the wire — before resolution to the catalog Model id
	// (ModelID). Set at the inference entry.
	RequestedModel string

	// Attempts is the number of upstream tries the pipeline made (1 when
	// the first key succeeded; >1 on failover). Pipeline-only; stays 0 in
	// proxy mode, which is single-shot by design.
	Attempts int

	RelayKeyHash string
	PolicyID     string
	ModelID      string
	HostID       string
	HostKeyID    string

	// PayloadLog opts this request into full request/response body capture
	// by the payloadlog observer. Set at the inference entry from the
	// routing Plan (Policy or RelayKey opt-in). When false, the payload
	// observer skips the request and its stream observer does not buffer.
	PayloadLog bool

	// RequestBody is the raw inbound request bytes, retained for the
	// payloadlog observer. Set at the inference entry (pipeline: the
	// dispatched body; proxy: a capped tee of the request stream). Nil
	// when payload logging is off or the body wasn't retained. It is a
	// reference to the dispatch buffer, not a copy — never mutated.
	RequestBody []byte

	// Cross-hook channel. Middleware writes; observers read.
	// Concurrent map writes during post-flight are a panic — keep
	// writes to the pre-flight phase only, or wrap with your own lock
	// if you really must.
	Metadata map[string]any

	// Translator is the per-request vendor adapter, set by the runner
	// when routing decides the upstream. Observers that want a
	// canonical view of the response (usage, finish reason, output
	// items) call v1.ExtractUsage / Translator.ParseResponse on
	// ev.ResponseBody. nil for runners that can't expose one (e.g.
	// anonymous proxy without resolved binding).
	Translator v1.Translator
	// contains filtered or unexported fields
}

Context is the persistent lifecycle state for one request. Created once at request entry, threaded through every phase, mutable by middleware to enrich routing identity and free-form metadata.

Fields fall into three layers:

• Identity: set at entry, never changes after.
• Routing identity: filled during routing / pre-flight middleware once the (model, host, binding, keys) tuple is resolved.
• Metadata: free-form cross-hook channel for facts that don't deserve first-class fields. Middlewares scribble; observers read.

Middlewares may mutate any field; observers must treat fields as read-only (see package doc — concurrent map writes on Metadata are a race). A field being empty means "not yet known" or "not applicable for this runner" (e.g. HostKeyID stays empty in proxy mode where the caller brought their own credential).

func FromContext(ctx context.Context) *Context

func NewContext(requestID, source string, startTime time.Time) *Context

func (c *Context) Collected(name string) (any, bool)

func (c *Context) FirstByteReader(r io.Reader) io.Reader

func (c *Context) MarkEnd()

func (c *Context) MarkReasoningEnd()

func (c *Context) MarkReasoningStart()

func (c *Context) MarkUpstreamStart()

type Hook ¶

type Hook interface {
	Name() string
	Fill(lc *Context, ev *PostFlightEvent) (any, error)
}

Hook fills its own result struct from the request's read-only state (the Context's identity/timing + the PostFlightEvent). It is a pure producer: it MUST NOT mutate the Context. The Registry calls Fill, then — as the sole writer — attaches the returned value to the Context under Name(). Because the hook never holds the write side of the collected set, it cannot break Context consistency or race a sibling.

Fill returns (nil, nil) when the hook has nothing to contribute for this request (e.g. no usage block); the Registry attaches nothing.

Fill must be cheap and non-blocking — heavy work (disk, network) belongs in a Collector behind a bounded channel, not here.

type HookFunc ¶

type HookFunc struct {
	HookName string
	Fn       func(lc *Context, ev *PostFlightEvent) (any, error)
}

HookFunc adapts a plain function to the Hook interface — for tests and simple observers that capture state without producing a stored result (return nil, nil).

func (h HookFunc) Fill(lc *Context, ev *PostFlightEvent) (any, error)

func (h HookFunc) Name() string

type PostFlightEvent ¶

type PostFlightEvent struct {
	// Status is the upstream HTTP status (or 0 if the request never
	// reached upstream — e.g. pre-flight aborted, routing failed).
	Status int

	// ErrorKind is a short machine-readable category for failures.
	// Empty on success. Examples: "upstream_429", "no_keys",
	// "model_not_found", "stream_aborted".
	ErrorKind string

	// ErrorMessage is the human-readable detail. Optional even when
	// ErrorKind is set; observers should not require it.
	ErrorMessage string

	// ResponseBody is the buffered upstream response bytes. Observers
	// parse what they need (tokens for usage, full body for audit /
	// cache, etc.). Nil when the runner couldn't buffer (e.g. body
	// exceeded a size cap, or the request failed before bytes flowed).
	//
	// READ-ONLY across parallel observers. To transform, copy first
	// via bytes.Clone.
	ResponseBody []byte
}

PostFlightEvent is the snapshot observers see after the request has completed (success or failure). Pointer-shared across all observers in the parallel post-flight chain — see package doc for the read-only invariants.

type PreFlightEvent ¶

type PreFlightEvent struct{}

PreFlightEvent is the snapshot middleware sees before the upstream call. Today this is a placeholder — no fields are loadbearing because no middleware consumer has materialized yet. As consumers land, fields like a mutable request handle or a resolved routing plan will live here. Empty struct keeps the signature stable.

When adding fields:

• Prefer types that work across all runners (pipeline, proxy, ws, batch). If a field only makes sense for one runner, push it to lc.Metadata instead.
• Pointer fields if middleware needs to mutate.

type PreFlightMiddleware ¶

type PreFlightMiddleware func(ctx context.Context, lc *Context, ev *PreFlightEvent) error

PreFlightMiddleware runs synchronously in the request's hot path before the upstream call. Sequential in registration order. Returning a non-nil error aborts the request and propagates the error to the runner's caller.

May mutate lc and ev fields. Typical use: budget check, cache lookup with short-circuit, request enrichment, additional authz.

type ReasoningTiming ¶

type ReasoningTiming struct {
	Start time.Duration // first reasoning frame seen
	End   time.Duration // last reasoning frame seen
}

ReasoningTiming groups the reasoning span, each elapsed from Timing.Start. Populated only on streaming responses observed as canonical events (reasoning item.started/delta/completed); zero when the response carried no reasoning or wasn't canonical-observed.

type Registry ¶

type Registry struct {
	// contains filtered or unexported fields
}

Registry holds the registered pre-flight middleware, post-flight Hooks (producers), and Collectors (the janitors that store). Construct one at the composition root, register during boot, pass into every request-runner.

Safe for concurrent Register* / Run* / Finalize calls. Registration is typically a boot-time operation; the hot side is the dispatch methods.

func New() *Registry

func (r *Registry) CollectorCount() int

func (r *Registry) Fill(lc *Context, ev *PostFlightEvent)

func (r *Registry) Finalize(ctx context.Context, lc *Context, ev *PostFlightEvent)

func (r *Registry) HookCount() int

func (r *Registry) NewStreamSession(lc *Context) *StreamSession

func (r *Registry) PreFlightCount() int

func (r *Registry) RegisterCollector(c Collector)

func (r *Registry) RegisterHook(h Hook)

func (r *Registry) RegisterPreFlight(m PreFlightMiddleware)

func (r *Registry) RegisterStreamObserver(f StreamObserverFactory)

func (r *Registry) RunPreFlight(ctx context.Context, lc *Context, ev *PreFlightEvent) error

func (r *Registry) StreamObserverCount() int

type StreamObserver ¶

type StreamObserver interface {
	Observe(frame []byte)
	Result() (any, error)
}

StreamObserver is a per-request, stateful observer of a streamed response. Unlike a Hook (one-shot, post-flight, too late for a streamed body), it watches frames as they flow: Observe is called once per upstream frame, then Result is called at end-of-stream. The runner attaches Result's value to the Context under the factory's Name() and marks the request filled — so the post-flight sink reuses the same collection (collect once), exactly like the buffered path.

Built fresh per request (see StreamObserverFactory) so per-stream state doesn't leak across concurrent requests — mirrors the v1 stream translator closures.

type StreamObserverFactory ¶

type StreamObserverFactory interface {
	Name() string
	NewObserver(lc *Context) StreamObserver
}

StreamObserverFactory produces a fresh StreamObserver per streamed request. Registered once at boot.

type StreamSession ¶

type StreamSession struct {
	// contains filtered or unexported fields
}

StreamSession drives the per-request stream observers for one streamed response. Not safe for concurrent use — one stream, one goroutine.

func (s *StreamSession) Finish()

func (s *StreamSession) Observe(frame []byte)

type Timing ¶

type Timing struct {
	Start     time.Time       // request accepted (absolute anchor)
	Upstream  UpstreamTiming  // the upstream leg
	Reasoning ReasoningTiming // the reasoning span (streaming, canonical-observed)
	End       time.Duration   // start → response closed / post-flight
}

Timing holds per-request checkpoints. Start is the absolute anchor; every other field is elapsed time measured from Start. All anchored to Start, never chained — so a missing or flaky intermediate mark can't corrupt the others, and the headline numbers (TTFT, total) are read directly rather than summed up a chain that compounds error.

The unit lives here, once: the elapsed fields are time.Duration in memory; sinks serialize them to microseconds. Derived intervals are computed by the consumer, never stored:

relay pre-overhead = Upstream.Start
upstream TTFT      = Upstream.ResponseStart - Upstream.Start
stream body time   = Upstream.ResponseEnd   - Upstream.ResponseStart
relay tail         = End                    - Upstream.ResponseEnd
reasoning span      = Reasoning.End          - Reasoning.Start

type UpstreamTiming ¶

type UpstreamTiming struct {
	Start         time.Duration // request handed to upstream
	ResponseStart time.Duration // first upstream byte received (TTFT mark)
	ResponseEnd   time.Duration // upstream finished sending
}

UpstreamTiming groups the upstream-leg checkpoints, each elapsed from Timing.Start.