llmgateway

README

llmgateway

llmgateway is an embeddable Go engine for host-authorized, host-metered, multi-provider LLM inference. It owns protocol normalization, routing, retries, stream commit semantics, guardrails, caching decisions, and provider-neutral usage facts. Hosts provide metering policy and persistence, identity, secrets, distributed counters, and telemetry through explicit interfaces.

The module intentionally has no dependency on Daptin, Gin, api2go, SQL, database drivers, or distributed-cache clients. Daptin consumes tagged releases and supplies those host integrations from its own repository.

The compatibility surface is declared in compatibility/manifest.json. An endpoint or provider is supported only when its manifest entry is backed by its conformance suite.

The current manifest is the target module contract. Declared wire fields and operations require conformance coverage, and named services additionally require their live-provider matrix before the manifest can be marked verified.

Model defaults are compiled once with the catalog and use operation-scoped JSON so fields with similar names cannot collide:

{
  "chat": {"temperature": 0, "max_completion_tokens": 512},
  "text_completion": {"temperature": 0, "max_tokens": 512},
  "responses": {"max_output_tokens": 512},
  "embeddings": {"encoding_format": "float"},
  "image_generation": {"n": 1, "response_format": "b64_json"}
}

Unknown, invalid, or undeclared-operation defaults reject the whole snapshot. Explicit request values always take precedence. The routing strategy is priority_weighted. Unsupported-parameter policy is explicit: reject fails, drop removes only non-semantic optional fields, and passthrough preserves typed manifest fields only when an eligible adapter declares support. Unknown wire fields are never accepted as arbitrary passthrough data.

Development

go test ./...
go test -race ./...

The example in examples/basic proves that the catalog and lifecycle can be embedded by a service that has no Daptin dependency.

Documentation

Index

• Variables
• type Authenticator
• type AuthenticatorFunc
  • func (f AuthenticatorFunc) Authenticate(ctx context.Context, credential string) (contract.Principal, error)
• type Authorizer
• type CatalogSource
• type Clock
• type CounterStore
• type Dependencies
• type DisabledResponseCache
  • func (DisabledResponseCache) Delete(context.Context, string) error
  • func (DisabledResponseCache) Get(context.Context, string) ([]byte, bool, error)
  • func (DisabledResponseCache) Set(context.Context, string, []byte, time.Duration) error
• type DiscardTelemetrySink
  • func (DiscardTelemetrySink) Record(context.Context, TelemetryEvent)
• type Engine
  • func New(dependencies Dependencies, options Options) (*Engine, error)
  • func (e *Engine) Authorize(ctx context.Context, principal contract.Principal, publicModel string) error
  • func (e *Engine) Drain(ctx context.Context) error
  • func (e *Engine) Handler(options HTTPOptions) (http.Handler, error)
  • func (e *Engine) Invoke(ctx context.Context, principal contract.Principal, request contract.Request) (contract.Response, error)
  • func (e *Engine) Probe(ctx context.Context) (HealthReport, error)
  • func (e *Engine) Reload(ctx context.Context) error
  • func (e *Engine) Snapshot() (*catalog.Snapshot, error)
  • func (e *Engine) Status() Status
  • func (e *Engine) Stream(ctx context.Context, principal contract.Principal, request contract.Request) (EventStream, error)
• type EventStream
• type GatewayStream
  • func (s *GatewayStream) Close(ctx context.Context) error
  • func (s *GatewayStream) Next(ctx context.Context) (contract.StreamEvent, error)
• type HTTPOptions
• type HealthReport
• type MeteringPort
• type Options
• type RandomSelector
  • func (RandomSelector) Intn(limit int) int
• type ResponseCache
• type SecretResolver
• type Selector
• type Status
• type SystemClock
  • func (SystemClock) NewTimer(duration time.Duration) Timer
  • func (SystemClock) Now() time.Time
• type TelemetryEvent
• type TelemetrySink
• type Timer

Variables

var (
	ErrDraining        = errors.New("llmgateway is draining")
	ErrNotReady        = errors.New("llmgateway has no valid catalog snapshot")
	ErrCounterLimit    = errors.New("llmgateway transient counter limit exceeded")
	ErrAdmissionDenied = errors.New("llmgateway metering admission denied")
)

Types

type Authenticator

type Authenticator interface {
	// credential is empty when the request supplies no API credential; the host
	// decides whether that request has a principal.
	Authenticate(context.Context, string) (contract.Principal, error)
}

type AuthenticatorFunc

type AuthenticatorFunc func(context.Context, string) (contract.Principal, error)

func (AuthenticatorFunc) Authenticate

func (f AuthenticatorFunc) Authenticate(ctx context.Context, credential string) (contract.Principal, error)

type Authorizer

type Authorizer interface {
	Authorize(context.Context, contract.Principal, catalog.Model) error
}

type CatalogSource

type CatalogSource interface {
	Load(context.Context, uint64) (catalog.Document, error)
}

type Clock

type Clock interface {
	Now() time.Time
	NewTimer(time.Duration) Timer
}

type CounterStore

type CounterStore interface {
	// Add atomically increments a disposable protection counter. The first
	// successful increment establishes the TTL; later increments must not slide
	// the window.
	Add(context.Context, string, int64, time.Duration) (int64, error)
	Get(context.Context, string) (int64, bool, error)
	// Acquire atomically reserves one unit up to maximum and returns an opaque,
	// idempotently releasable lease. The aggregate expiry must cover the newest
	// lease so live capacity cannot disappear early; TTL bounds leaked capacity
	// after a crash.
	Acquire(context.Context, string, int64, time.Duration) (string, error)
	Release(context.Context, string) error
	Delete(context.Context, string) error
}

type Dependencies

type Dependencies struct {
	Catalog    CatalogSource
	Secrets    SecretResolver
	Adapters   *adapter.Registry
	Authorizer Authorizer
	Metering   MeteringPort
	Counters   CounterStore
	Cache      ResponseCache
	Guardrails *guardrail.Registry
	Telemetry  TelemetrySink
	Selector   Selector
	Clock      Clock
}

type DisabledResponseCache

type DisabledResponseCache struct{}

DisabledResponseCache is an explicit cache opt-out.

func (DisabledResponseCache) Delete

func (DisabledResponseCache) Delete(context.Context, string) error

func (DisabledResponseCache) Get

func (DisabledResponseCache) Get(context.Context, string) ([]byte, bool, error)

func (DisabledResponseCache) Set

func (DisabledResponseCache) Set(context.Context, string, []byte, time.Duration) error

type DiscardTelemetrySink

type DiscardTelemetrySink struct{}

DiscardTelemetrySink is an explicit opt-out for hosts that do not emit gateway telemetry.

func (DiscardTelemetrySink) Record

func (DiscardTelemetrySink) Record(context.Context, TelemetryEvent)

type Engine

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

func New

func New(dependencies Dependencies, options Options) (*Engine, error)

func (*Engine) Authorize

func (e *Engine) Authorize(ctx context.Context, principal contract.Principal, publicModel string) error

Authorize evaluates model visibility through the same host policy used by invocation. Protocol handlers use it for discovery without duplicating ACLs.

func (*Engine) Drain

func (e *Engine) Drain(ctx context.Context) error

func (*Engine) Handler

func (e *Engine) Handler(options HTTPOptions) (http.Handler, error)

Handler exposes the engine's supported OpenAI-compatible HTTP routes.

func (*Engine) Invoke

func (e *Engine) Invoke(ctx context.Context, principal contract.Principal, request contract.Request) (contract.Response, error)

func (*Engine) Probe

func (e *Engine) Probe(ctx context.Context) (HealthReport, error)

Probe executes only explicitly enabled, side-effect-free adapter health checks. Successful probes clear infrastructure circuit state but never clear provider rate-limit cooldowns.

func (*Engine) Reload

func (e *Engine) Reload(ctx context.Context) error

func (*Engine) Snapshot

func (e *Engine) Snapshot() (*catalog.Snapshot, error)

func (*Engine) Status

func (e *Engine) Status() Status

func (*Engine) Stream

func (e *Engine) Stream(ctx context.Context, principal contract.Principal, request contract.Request) (EventStream, error)

type EventStream

type EventStream = contract.EventStream

type GatewayStream

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

func (*GatewayStream) Close

func (s *GatewayStream) Close(ctx context.Context) error

func (*GatewayStream) Next

func (s *GatewayStream) Next(ctx context.Context) (contract.StreamEvent, error)

Next hands the first validated event to the caller and irrevocably commits this stream to its selected deployment. Provider switching is impossible after this method returns an event.

type HTTPOptions

type HTTPOptions struct {
	Authenticator Authenticator
	Protocol      openai.Options
}

HTTPOptions configures the strict OpenAI-compatible HTTP surface. The authenticator is supplied here so hosts that invoke the engine directly do not need to configure an unused HTTP authentication path.

type HealthReport

type HealthReport struct {
	Checked   int
	Healthy   int
	Unhealthy int
}

type MeteringPort

type MeteringPort interface {
	Admit(context.Context, contract.Admission) (contract.ReservationToken, error)
	Complete(context.Context, contract.Completion) error
	Cancel(context.Context, contract.Cancellation) error
}

MeteringPort is implemented by the host. The gateway reports admission and terminal usage facts; the host alone owns policy evaluation and persistence.

type Options

type Options struct {
	MaxAttempts            int
	DefaultMaxOutputTokens int64
	FinalizationTimeout    time.Duration
	BaseRetryDelay         time.Duration
	MaxRetryDelay          time.Duration
	CircuitFailures        int64
	CircuitWindow          time.Duration
	CircuitCooldown        time.Duration
	CacheTTL               time.Duration
	CacheTimeout           time.Duration
	CacheFillWait          time.Duration
	MaxCacheEntryBytes     int
	FirstEventTimeout      time.Duration
	StreamIdleTimeout      time.Duration
	RequestTimeout         time.Duration
	HealthProbeWorkers     int
}

type RandomSelector

type RandomSelector struct{}

RandomSelector is the production weighted-route selector. Tests should use an injected deterministic selector instead.

func (RandomSelector) Intn

func (RandomSelector) Intn(limit int) int

type ResponseCache

type ResponseCache interface {
	Get(context.Context, string) ([]byte, bool, error)
	Set(context.Context, string, []byte, time.Duration) error
	Delete(context.Context, string) error
}

type SecretResolver

type SecretResolver interface {
	ResolveSecret(context.Context, string) ([]byte, error)
}

type Selector

type Selector interface {
	Intn(int) int
}

type Status

type Status struct {
	Ready                    bool   `json:"ready"`
	Draining                 bool   `json:"draining"`
	Degraded                 bool   `json:"degraded"`
	Revision                 uint64 `json:"revision"`
	ActiveSnapshotAgeSeconds int64  `json:"active_snapshot_age_seconds"`
	RejectedRevision         uint64 `json:"rejected_revision,omitempty"`
	ReloadStage              string `json:"reload_stage,omitempty"`
}

Status is a redacted, point-in-time view of engine lifecycle and catalog readiness. Failure stages are stable categories and never contain host or provider error text.

type SystemClock

type SystemClock struct{}

SystemClock is the production wall clock implementation.

func (SystemClock) NewTimer

func (SystemClock) NewTimer(duration time.Duration) Timer

func (SystemClock) Now

func (SystemClock) Now() time.Time

type TelemetryEvent

type TelemetryEvent struct {
	Name       string
	RequestID  contract.ID
	Revision   uint64
	Attributes map[string]string
	Measures   map[string]int64
}

type TelemetrySink

type TelemetrySink interface {
	Record(context.Context, TelemetryEvent)
}

type Timer

type Timer interface {
	C() <-chan time.Time
	Stop() bool
}