API Reference

Protocols

RelayChannelCheckerProtocol

source

Bounded upstream probe for a single channel.

Implementations are free to ping any endpoint, but must not embed credentials in the probe; the health service records only the status values, never upstream_base_url or query strings.

check

async def check(channel: RelayChannel) -> RelayChannelProbeResult | None

source

Probe channel and report a result.

Parameters

Parameter	Type	Description
`channel`	RelayChannel	The channel to probe.

Returns

Type	Description
RelayChannelProbeResult | None	The probe result, or ``None`` when the checker has no signal for this channel.

---

RelayChannelCredentialProvider

source

Resolve per-channel upstream credential headers.

Implementations look up whatever a host stores (env, secrets manager, database) and return HTTP headers to merge into the outbound upstream call. They never receive request payloads and are resolved once per upstream call by channel name only.

headers_for

async def headers_for(channel_name: str) -> Mapping[str, str]

source

---

RelayRouteEventSourceProtocol

source

Provides operational events inside a bounded window.

events

async def events(window: TimeWindow) -> Sequence[RelayRouteEvent]

source

Return the events observed inside window.

Parameters

Parameter	Type	Description
`window`	TimeWindow	Bounded aggregation window.

Returns

Type	Description
Sequence[RelayRouteEvent]	Events bounded to the window; an empty sequence means no activity, never a failed lookup.

---

Classes

CredentialInjectingHTTPClient

source

Wrap an ``HTTPClientProtocol`` and merge per-channel headers.

The decorator pops channel_name from the outbound call’s kwargs (defaulting to "" when absent, e.g. calls made outside the gateway), asks the credential provider for the channel’s headers, merges them under the caller-supplied headers (provider headers take precedence on key collision), and delegates to the wrapped client. All other HTTPClientProtocol methods delegate unchanged.

Provider lookup failures are raised as a generic InfrastructureError so the gateway’s upstream adapter classifies them as UPSTREAM_FAILED; header values themselves are never logged or echoed into exceptions.

__init__

def __init__(
    wrapped: HTTPClientProtocol,
    provider: RelayChannelCredentialProvider | None = None
) -> None

source

Bind the decorator to a client and a credential provider.

Parameters

Parameter	Type	Description
`wrapped`	HTTPClientProtocol	The ``HTTPClientProtocol`` implementation driving the actual outbound request.
`provider`	RelayChannelCredentialProvider | None	Credential provider for the outbound calls. When omitted, ``NullChannelCredentialProvider`` is used and no headers are ever injected.

wrapped

property wrapped() -> HTTPClientProtocol

source

Return the wrapped client.

start

async def start() -> None

source

Start the wrapped client.

stop

async def stop() -> None

source

Stop the wrapped client.

request

async def request(
    method: str,
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Inject credential headers, then delegate to the wrapped client.

Parameters

Parameter	Type	Description
`method`	str	HTTP method (GET, POST, PUT, ...).
`url`	str	Request URL. **kwargs: Additional options passed to the wrapped client, including ``channel_name`` and ``headers``.

Returns

Type	Description
HttpResponse	The wrapped client's response.

Raises

Exception	Description
InfrastructureError	The credential provider failed to resolve headers for the active channel.

get

async def get(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Send a GET through the wrapped client.

post

async def post(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Send a POST through the wrapped client.

put

async def put(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Send a PUT through the wrapped client.

delete

async def delete(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Send a DELETE through the wrapped client.

patch

async def patch(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Send a PATCH through the wrapped client.

head

async def head(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Send a HEAD through the wrapped client.

---

HTTPUpstreamAdapter

source

Sends upstream requests through an injected HTTP client.

The adapter classifies failures from stdlib and contracts-level exceptions only (by design); implementations of HTTPClientProtocol keep transport libraries like oridecon-http behind the DI boundary.

Attributes: _http: The injected HTTP client resolved from DI. _cancelled: Request identifiers whose streaming cancel was observed (after-the-fact only; outbound frames are not interrupted).

__init__

def __init__(http: HTTPClientProtocol) -> None

source

Bind the adapter to an HTTP client.

Parameters

Parameter	Type	Description
`http`	HTTPClientProtocol	Any ``HTTPClientProtocol`` implementation driving the outbound request.

request

async def request(request: UpstreamRequest) -> Result[UpstreamResponse, RelayGatewayError]

source

Send request upstream and classify the outcome.

Method, URL, headers, JSON payload, and timeout come straight from the UpstreamRequest. 2xx responses are decoded into a typed UpstreamResponse (empty bodies yield a None payload); non-2xx responses and transport failures map to RelayGatewayError values that never carry raw bodies, headers, or credentials.

Parameters

Parameter	Type	Description
`request`	UpstreamRequest	Fully-resolved upstream request.

Returns

Type	Description
Result[UpstreamResponse, RelayGatewayError]	``Ok(UpstreamResponse)`` for 2xx responses (the response headers are preserved verbatim), or ``Err`` classifying transport cancellation (``UPSTREAM_CANCELLED``, 499), timeouts (``UPSTREAM_TIMEOUT``, 504), generic transport failures (``UPSTREAM_FAILED``, 502), malformed 2xx bodies (``UPSTREAM_MALFORMED``, 502), and non-2xx responses (``UPSTREAM_ERROR`` with a safe public message).

stream

async def stream(request: UpstreamRequest) -> AsyncIterator[UpstreamChunk]

source

Consume one upstream SSE response as a stream of chunks.

The whole stream arrives through a single request call whose body is parsed into data: frames; frames are emitted one by one as the consumer iterates. 2xx responses are parsed into chunk frames; non-2xx responses and transport failures surface as one terminal UpstreamChunk carrying a safe public {"code", "message"} payload.

Parameters

Parameter	Type	Description
`request`	UpstreamRequest	Fully-resolved upstream request.

Yields

Type	Description
AsyncIterator[UpstreamChunk]	One ``UpstreamChunk`` per SSE ``data:`` line, with the OpenAI ``[DONE]`` marker flagged terminal.

cancel

async def cancel(request_id: str) -> None

source

Record a streaming cancellation request (always succeeds).

The fake-safe transport cannot interrupt an in-flight response body, so cancellation is observed after the fact; the stream loop stops consulting this adapter once its cancel is recorded.

Parameters

Parameter	Type	Description
`request_id`	str	Identifier of the stream being cancelled.

---

InMemoryRelayPolicyStore

source

Process-local policy store seeded from the gateway configuration.

load always returns the current snapshot and save replaces it wholesale; the store is the single source of truth between control mutations in this process.

__init__

def __init__(initial: RelayPolicySnapshot) -> None

source

Bind the store to its initial snapshot.

Parameters

Parameter	Type	Description
`initial`	RelayPolicySnapshot	Snapshot the store serves until the first save.

with_defaults

def with_defaults(
    cls,
    config: RelayGatewayConfig
) -> InMemoryRelayPolicyStore

source

Build a store seeded from a gateway configuration.

Parameters

Parameter	Type	Description
`config`	RelayGatewayConfig	Gateway configuration; each channel contributes its enabled flag and declared models as the allowed options.

Returns

Type	Description
InMemoryRelayPolicyStore	A store whose snapshot mirrors the configuration.

load

async def load() -> RelayPolicySnapshot

source

Return the current snapshot.

save

async def save(snapshot: RelayPolicySnapshot) -> None

source

Atomically replace the stored snapshot.

---

ModelCatalogService

source

Aggregate served model aliases per wire format.

Parameters

Parameter	Type	Description
`registry`		The channel registry whose enabled, non-drained channels define the served model set.

__init__

def __init__(registry: RelayChannelRegistry) -> None

source

Bind the catalog to the channel registry.

Parameters

Parameter	Type	Description
`registry`	RelayChannelRegistry	The channel registry backing the model set.

list_openai

def list_openai() -> dict[str, Any]

source

Return the OpenAI /v1/models list payload.

Returns

Type	Description
dict[str, Any]	A list payload with one entry per served alias, sorted.

list_claude

def list_claude() -> dict[str, Any]

source

Return the Anthropic /v1/models list payload.

Returns

Type	Description
dict[str, Any]	A list payload with one entry per served alias, sorted.

list_gemini

def list_gemini() -> dict[str, Any]

source

Return the Gemini /v1beta/models list payload.

Returns

Type	Description
dict[str, Any]	A list payload with one model entry per served alias, sorted.

model_exists

def model_exists(alias: str) -> bool

source

Return whether alias is served by any enabled channel.

Parameters

Parameter	Type	Description
`alias`	str	The model alias to look up.

Returns

Type	Description
bool	``True`` when the alias is served, ``False`` otherwise.

openai_detail

def openai_detail(alias: str) -> dict[str, Any] | None

source

Return the OpenAI model detail payload for alias.

Parameters

Parameter	Type	Description
`alias`	str	The model alias to describe.

Returns

Type	Description
dict[str, Any] | None	The detail payload, or ``None`` when the alias is not served.

gemini_detail

def gemini_detail(alias: str) -> dict[str, Any] | None

source

Return the Gemini model detail payload for alias.

Parameters

Parameter	Type	Description
`alias`	str	The model alias to describe.

Returns

Type	Description
dict[str, Any] | None	The detail payload, or ``None`` when the alias is not served.

---

NullChannelCredentialProvider

source

No-op credential provider; keeps behavior unchanged by default.

headers_for

async def headers_for(channel_name: str) -> Mapping[str, str]

source

Return no headers for any channel.

Parameters

Parameter	Type	Description
`channel_name`	str	The active channel name (ignored).

Returns

Type	Description
Mapping[str, str]	An empty header mapping.

---

PassthroughService

source

Endpoint-kind mapping: call one method, no conversion.

The service is stateless between requests and never includes payloads or upstream details in error messages; errors are always safe RelayGatewayError values. Authorization, billing, channel selection, and upstream transport reuse the chat pipeline’s dependencies unchanged.

Attributes: _registry: Deterministic channel selector. _upstream: HTTP transport adapter. _config: Gateway configuration (channel table and model suffixes). _authorizer: Optional authorization check before dispatch. _billing: Optional billing lifecycle; when None admission and settlement are skipped.

__init__

def __init__(
    registry: RelayChannelRegistry,
    upstream: HTTPUpstreamAdapter,
    config: RelayGatewayConfig,
    *,
    authorizer: AuthorizerProtocol | None = None,
    billing: RelayBillingProtocol | None = None
) -> None

source

Bind the service to its dependencies.

Parameters

Parameter	Type	Description
`registry`	RelayChannelRegistry	Channel selection registry.
`upstream`	HTTPUpstreamAdapter	Upstream transport adapter; handles credential injection per channel through its configured provider.
`config`	RelayGatewayConfig	Static gateway configuration.
`authorizer`	AuthorizerProtocol | None	Optional authorizer; when ``None`` authorization is skipped.
`billing`	RelayBillingProtocol | None	Optional billing lifecycle; when ``None`` the passthrough runs without admission control or settlement.

handle

async def handle(
    kind: str,
    request: RelayGatewayRequest
) -> Result[RelayPassthroughResult, RelayGatewayError]

source

Run the passthrough lifecycle for one request.

Dependencies run in fixed order: authorize, select channel by endpoint kind, reserve billing capacity, call upstream with the caller’s body verbatim, settle billing, assemble result. Any failure short-circuits the pipeline.

Parameters

Parameter	Type	Description
`kind`	str	The endpoint kind being served (e.g. ``"embeddings"``).
`request`	RelayGatewayRequest	The passthrough gateway request; ``payload`` is either a ``RelayPassthroughBody`` (JSON or raw multipart) or a plain JSON mapping forwarded by legacy callers, and ``source`` is a conventional marker (``OPENAI_CHAT``) never used for conversion.

Returns

Type	Description
Result[RelayPassthroughResult, RelayGatewayError]	``Ok(RelayPassthroughResult)`` on success, or ``Err(RelayGatewayError)`` on the first failure. Unexpected
exceptions from dependencies never escape	they are logged and mapped to a generic ``CONVERSION_FAILED`` error.

---

RelayChannelAutoTester

source

Automatically disable failing channels and restore recovered ones.

The tester runs as its own asyncio.Task, taking one health snapshot per interval and turning probe outcomes into runtime transitions via RelayChannelRegistry.set_runtime_enabled. Its own disable decisions are journaled in _disabled_by_tester so a channel drained by a human through the actuator controls is never silently restored.

Parameters

Parameter	Type	Description
`health`		The health service that produces per-channel snapshots.
`registry`		The channel registry whose runtime enable flags this tester mutates.
`interval_seconds`		Whole seconds between two consecutive sweeps. Must be positive.

__init__

def __init__(
    health: RelayHealthService,
    registry: RelayChannelRegistry,
    interval_seconds: float
) -> None

source

Bind the auto-tester to its health service and registry.

Parameters

Parameter	Type	Description
`health`	RelayHealthService	Health service whose snapshots drive the sweep.
`registry`	RelayChannelRegistry	Channel registry receiving runtime transitions.
`interval_seconds`	float	Delay between sweeps, in whole seconds.

is_running

property is_running() -> bool

source

Return whether a sweep loop is currently scheduled.

Returns

Type	Description
bool	``True`` when ``start()`` has scheduled a task that has not gone away, ``False`` otherwise.

start

async def start() -> None

source

Start the periodic sweep; a no-op when one is already running.

The first sweep runs immediately after scheduling, then the loop sleeps interval_seconds between iterations.

stop

async def stop() -> None

source

Cancel the sweep loop and await its completion.

Idempotent: calling stop when nothing is running is a no-op.

sweep

async def sweep() -> None

source

Run one probe sweep and apply the resulting transitions.

The snapshots come from the health service as-is; the sweep does not probe channels on its own. An exception raised while the health service probes a channel is caught and logged per sweep, and the loop continues to its next iteration.

Example

await tester.sweep()

---

RelayChannelProbeResult

source

Outcome of probing one upstream channel.

Attributes: ok: Whether the upstream responded within the bound. latency_ms: Observed latency, or None when unknown. failure: Human-readable failure reason, or None.

---

RelayChannelRegistry

source

Selects a ``RelayChannel`` deterministically from a static config.

By default selection is fully deterministic; when the config enables "weighted" load balancing, ties among an already-tied top tier (equal precedence, eligible, same priority) are broken by weighted-random pick instead of ascending name.

Note

“Healthy channel” in the plan is interpreted as the channel’s enabled flag combined with the live runtime override table; drained channels are invisible to select until they are restored at runtime.

Attributes: _channels: The immutable channel table from RelayGatewayConfig. _runtime_enabled: Operator overrides applied by the controls service; empty equals “all channels as configured”.

__init__

def __init__(
    config: RelayGatewayConfig,
    *,
    random_source: Callable[[int], int] | None = None
) -> None

source

Bind the registry to a static channel table.

Parameters

Parameter	Type	Description
`config`	RelayGatewayConfig	Immutable gateway configuration. Selection never mutates it; the channel tuple is kept as configured.
`random_source`	Callable[[int], int] | None	Callable receiving a weight sum and returning a value in ``[0, total)``; only used by the weighted tie-break. Defaults to ``random.SystemRandom().randrange``; tests pass a deterministic fake.

channels

property channels() -> tuple[RelayChannel, Ellipsis]

source

The configured channel table.

Returns

Type	Description
tuple[RelayChannel, Ellipsis]	The immutable channel tuple, in configuration order.

reload

def reload(channels: tuple[RelayChannel, Ellipsis]) -> None

source

Replace the channel table (boot reconcile from a durable store).

Runtime overrides are preserved for channels that remain in the new table and dropped for channels that were removed, so a boot reconcile never resurrects a drained channel and never carries overrides for channels that no longer exist.

Parameters

Parameter	Type	Description
`channels`	tuple[RelayChannel, Ellipsis]	The new channel tuple, in selection order.

set_runtime_enabled

def set_runtime_enabled(
    channel: str,
    enabled: bool
) -> None

source

Override the eligibility of channel at runtime.

Draining a channel (enabled=False) hides it from selection until restored; restoring removes the override entirely so the config enabled flag alone decides. A config-disabled channel can never be made eligible this way.

Parameters

Parameter	Type	Description
`channel`	str	Channel name to override.
`enabled`	bool	Whether the channel should select new requests.

runtime_enabled

def runtime_enabled() -> dict[str, bool]

source

Return the non-default runtime overrides.

Returns

Type	Description
dict[str, bool]	Mapping of channel name to ``False`` set at runtime; a restored channel is absent.

select

def select(
    source: RelayFormat,
    model: str,
    stream: bool = False,
    capabilities: frozenset[str] = frozenset(),
    preferred: str | None = None,
    exclude: frozenset[str] = frozenset()
) -> Result[RelayChannel, RelayGatewayError]

source

Pick the best channel for the routing query.

Eligibility is computed first (enabled by config and runtime, target format differs from the source, model serves the requested alias, streaming and capability constraints), then the survivors are sorted: preferred channel first, then exact model match, then ascending priority (lower number wins), then ascending name as a stable tiebreak. The preferred channel still must pass every eligibility filter; otherwise it is skipped and normal ordering applies.

Parameters

Parameter	Type	Description
`source`	RelayFormat	Wire format the caller supplies; channels whose target format equals it would be no-op conversions and are never eligible.
`model`	str	Requested model alias; only exact matches are eligible.
`stream`	bool	Whether the caller wants streaming. Channels that declare capabilities must declare ``"stream"`` to serve streaming requests; channels with no declared capabilities are unconstrained.
`capabilities`	frozenset[str]	Requested capability flags; they must be a subset of the channel's declared capabilities.
`preferred`	str | None	Optional channel name that ranks first when it is eligible. Defaults to ``None`` (no preference).
`exclude`	frozenset[str]	Channel names to skip, e.g. for failover retries. Excluded names are filtered before the other eligibility filters run, so an excluded channel is never eligible and cannot be treated as preferred. Defaults to empty (no exclusion).

Returns

Type	Description
Result[RelayChannel, RelayGatewayError]	``Ok(channel)`` for the best eligible channel, or ``Err(RelayGatewayError)`` when none is eligible. The error
cause is classified in fixed order	no enabled channels (``CHANNEL_DISABLED``, 404), no enabled channel transforms the source format (``TARGET_FORMAT_UNSUPPORTED``, 500), no enabled channel satisfies the capability filters (``CAPABILITY_UNAVAILABLE``, 409), otherwise the model is not served (``MODEL_NOT_FOUND``, 404).

Note

Runtime-drained channels are treated exactly like disabled channels: they are invisible to selection until restored.

select_for_endpoint

def select_for_endpoint(
    kind: str,
    model: str,
    *,
    exclude: frozenset[str] = frozenset()
) -> Result[RelayChannel, RelayGatewayError]

source

Pick the best channel serving an endpoint kind (e.g. "embeddings").

Passthrough entry point: eligibility is limited to channels declaring kind in endpoint_kinds (empty means chat-only, never eligible here), then survivors are sorted by ascending priority (lower number wins) and ascending name as a stable tiebreak. Model aliases and the enabled/runtime-disabled filters behave exactly like select. A chat-only channel is untouched by this method.

Parameters

Parameter	Type	Description
`kind`	str	Endpoint kind the caller wants (e.g. ``"embeddings"``); only channels declaring it are eligible.
`model`	str	Requested model alias; only exact matches are eligible.
`exclude`	frozenset[str]	Channel names to skip, e.g. for failover retries. Defaults to empty (no exclusion).

Returns

Type	Description
Result[RelayChannel, RelayGatewayError]	``Ok(channel)`` for the best eligible channel, or ``Err(RelayGatewayError)`` when none is eligible: no enabled channels (``CHANNEL_DISABLED``, 404), otherwise, no channel serves the kind or model (``MODEL_NOT_FOUND``, 404).

---

RelayControlsService

source

Apply permissioned, validated policy mutations for the gateway.

Every mutation is serialized through an in-process lock, validated against the static channel table, persisted to the policy store, and audited. A mutation that would leave the gateway without any enabled channel that serves at least one model option is rejected before persisting.

__init__

def __init__(
    registry: RelayChannelRegistry,
    store: RelayPolicyStoreProtocol,
    authorizer: AuthorizerProtocol | None = None,
    audit: AIAuditStoreProtocol | None = None,
    streams: RelayStreamRegistry | None = None
) -> None

source

Bind the controls service to its dependencies.

Parameters

Parameter	Type	Description
`registry`	RelayChannelRegistry	Channel table defining valid channel names and model options.
`store`	RelayPolicyStoreProtocol	Persistent backend that owns the current snapshot.
`authorizer`	AuthorizerProtocol | None	Permission gate for ``relay.*`` actions. When ``None`` no permission check is performed (development).
`audit`	AIAuditStoreProtocol | None	Audit backend for mutation events. When ``None`` mutations still apply without audit emission.
`streams`	RelayStreamRegistry | None	Registry of in-flight upstream streams. When ``None`` a private empty registry is created; share one instance with the streaming path so force-cancel reaches live streams.

set_channel_state

async def set_channel_state(
    channel: str,
    enabled: bool,
    actor_id: str
) -> None

source

Enable or drain channel for new requests.

Parameters

Parameter	Type	Description
`channel`	str	Channel name; unknown names are rejected.
`enabled`	bool	``False`` drains the channel for new requests while existing streams finish.
`actor_id`	str	Operator identity recorded in the audit event.

Raises

Exception	Description
ValueError	The channel is unknown.
RelayGatewayError	With ``PERMISSION_DENIED`` when the actor lacks ``relay.channel_control``.

update_policy

async def update_policy(
    change: RelayPolicyChange,
    actor_id: str
) -> None

source

Apply a typed policy change.

Parameters

Parameter	Type	Description
`change`	RelayPolicyChange	Partial mutation; only the fields explicitly set change.
`actor_id`	str	Operator identity recorded in the audit event.

Raises

Exception	Description
ValueError	The change references an unknown channel or model option values, or would remove every available model option.
RelayGatewayError	With ``PERMISSION_DENIED`` when the actor lacks ``relay.policy_control``.

policy_snapshot

async def policy_snapshot(actor_id: str) -> RelayPolicySnapshot

source

Return the current runtime policy snapshot.

Parameters

Parameter	Type	Description
`actor_id`	str	Operator identity; ``relay.read`` permission is required.

Returns

Type	Description
RelayPolicySnapshot	The snapshot persisted by the policy store.

Raises

Exception	Description
RelayGatewayError	With ``PERMISSION_DENIED`` when the actor lacks ``relay.read``.

active_streams

def active_streams() -> tuple[RelayActiveStream, Ellipsis]

source

Return the currently in-flight upstream streams.

Returns

Type	Description
tuple[RelayActiveStream, Ellipsis]	One row per active stream, oldest first; an empty tuple when no stream is in flight.

force_cancel_stream

async def force_cancel_stream(
    stream_id: str,
    actor_id: str
) -> None

source

Force-cancel an in-flight upstream stream.

Parameters

Parameter	Type	Description
`stream_id`	str	Identifier of the stream to cancel.
`actor_id`	str	Operator identity recorded in the audit event; ``relay.stream_control`` permission is required.

Raises

Exception	Description
ValueError	The stream identifier is unknown.
RelayGatewayError	With ``PERMISSION_DENIED`` when the actor lacks ``relay.stream_control``.

---

RelayFailoverTracker

source

Track consecutive upstream failures and ban failing channels.

The tracker mutates only the registry’s runtime enabled overrides, the same surface the operator controls and the auto-tester use. The threshold is compared with >= so the ban happens on the attempt that reaches it.

Parameters

Parameter	Type	Description
`registry`		The channel registry whose runtime enable flags this tracker mutates.
`threshold`		Consecutive failures that disable a channel. Must be positive.

__init__

def __init__(
    registry: RelayChannelRegistry,
    threshold: int
) -> None

source

Bind the tracker to the registry and threshold.

Parameters

Parameter	Type	Description
`registry`	RelayChannelRegistry	The channel registry receiving runtime transitions.
`threshold`	int	Consecutive failures that disable a channel.

threshold

property threshold() -> int

source

Return the consecutive-failure threshold.

Returns

Type	Description
int	The threshold configured at construction.

failure_count

def failure_count(channel: str) -> int

source

Return the recorded consecutive failures for channel.

Parameters

Parameter	Type	Description
`channel`	str	The channel name to inspect.

Returns

Type	Description
int	The consecutive failure count, ``0`` when none recorded.

banned

def banned() -> frozenset[str]

source

Return the channels this tracker disabled.

Returns

Type	Description
frozenset[str]	The immutable set of channel names banned by this tracker.

record_failure

def record_failure(channel: str) -> None

source

Count one upstream failure for channel and ban at threshold.

When the count reaches the threshold and the channel was not already banned, the channel is drained through the registry’s runtime overrides and journaled as banned by this tracker.

Parameters

Parameter	Type	Description
`channel`	str	The channel name that failed upstream.

record_success

def record_success(channel: str) -> None

source

Reset channel’s failures and restore it when banned here.

A successful dispatch clears the consecutive-failure count; when this tracker had banned the channel, it is restored at runtime and removed from the ban journal.

Parameters

Parameter	Type	Description
`channel`	str	The channel name that succeeded upstream.

---

RelayGatewayConfig

source

Static configuration backing ``RelayChannelRegistry`` selection.

Attributes: channels: The ordered channel configurations. Selection filters before sorting, so order is never observable in the result except as the stable name tiebreak. Duplicate names are rejected. model_suffix: Channel name to a suffix (e.g. ":thinking") appended to the outbound model alias at the service layer. Selection does not use this field. provider_options: Channel name to provider-specific options merged into RelayConversionContext at conversion time. Selection does not use this field. auto_test_channels: When True the provider starts a background channel auto-tester that periodically probes every channel and disables failed ones, re-enabling them on recovery. Defaults to False (disabled). auto_test_interval_seconds: Delay between auto-test sweeps in seconds. Must be positive when defined. Defaults to 600. max_upstream_retries: Number of retry attempts across other channels after a retryable upstream failure on the buffered path. Defaults to 0 (single attempt, today’s behavior). load_balancing: Channel-selection mode. "deterministic" (default) keeps today’s name-sort tiebreak; "weighted" breaks ties among equal-priority eligible channels by weighted-random pick driven by each channel’s weight. job_ttl_seconds: Age in seconds after which a relay job record (POST /v1/videos style job relay) is evicted from the in-memory job registry on its next poll. Must be positive. Defaults to 3600 (one hour). require_auth: When True the inbound relay routes require a bound RelayAuthVerifierProtocol; False is an explicit opt-out for local/dev use only. rate_limits: Model name (or "*" for the token-wide rule) to a {"max": int, "window_seconds": int} budget. Empty (default) disables the rate-limit guard entirely. auto_disable_on_failures: When True the gateway tracks consecutive upstream failures per channel and takes a channel out of service at runtime once failover_failure_threshold is reached, restoring it after the next successful dispatch. Defaults to False (disabled). failover_failure_threshold: Number of consecutive failures that disable a channel when auto_disable_on_failures is on. Defaults to 3.

from_mapping

def from_mapping(
    cls,
    data: Mapping[str, Any]
) -> RelayGatewayConfig

source

Build the configuration from a JSON/TOML-style mapping.

Channel entries accept the fields of RelayChannel (target_format as the format member name, e.g. "OPENAI_CHAT"), and top-level keys mirror the remaining attributes of this class. Unknown channel keys and top-level "channels" types raise ValueError with the offending key named.

Parameters

Parameter	Type	Description
`data`	Mapping[str, Any]	Mapping with a ``"channels"`` list and optional gateway fields.

Returns

Type	Description
RelayGatewayConfig	A validated ``RelayGatewayConfig``.

Raises

Exception	Description
TypeError	On malformed channel entries or a non-list ``"channels"`` value.
ValueError	On unknown channel keys or invalid top-level values.

---

RelayGatewayModule

source

Protocol-facing relay gateway module for Oridecon applications.

Provides the relay gateway behind the RelayGatewayProtocol contract: channel selection, orchestration, upstream I/O, and SSE handling. The RelayGatewayProvider composes the gateway from caller-owned config, conversion engine, and HTTP client.

Usage

from oridecon.ai.relay.gateway import RelayGatewayModule

@module(
    imports=[RelayGatewayModule.configure()]
)
class AppModule(Module):
    pass

from oridecon.ai.relay.gateway import RelayGatewayModule

@module(
    imports=[RelayGatewayModule.configure()]
)
class AppModule(Module):
    pass

configure

def configure(
    cls,
    config: RelayGatewayConfig | None = None
) -> DynamicModule

source

Create a RelayGatewayModule with the built-in gateway routes.

Parameters

Parameter	Type	Description
`config`	RelayGatewayConfig | None	Static gateway configuration (channel table, model suffixes, auto-test flags, job TTL). Defaults to an empty configuration when omitted.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

---

RelayGatewayProvider

source

Provider registering the relay gateway behind ``RelayGatewayProtocol``.

The caller owns the static configuration, the conversion engine, and the HTTP client; the provider wires them into a ready-to-serve RelayGatewayService. Optional governance hooks (authorizer, media resolver, billing) are forwarded to the service as-is.

Configuration is explicit-only (a frozen channel/conversion table); the gateway is not bound to a OrideconConfig section, so this provider declares no config_key/config_model attributes.

Registers:

• RelayGatewayConfig — the injected configuration (always)
• RelayChannelRegistry — a registry built from the configuration
• RelayPolicyStoreProtocol — the runtime policy backend (always)
• RelayHealthService — channel health probing (always)
• RelayMetricsService — route metrics aggregation (always)
• RelayControlsService — permissioned control mutations (always)
• RelayChannelAutoTester — background channel auto-tester (only when auto_test_channels is enabled in the configuration)
• RelayFailoverTracker — reactive consecutive-failure tracking (only when auto_disable_on_failures is enabled)
• RelayGatewayProtocol — the gateway service (only when both the converter and an HTTP client are available)
• PassthroughService — passthrough endpoint dispatch (same availability as the gateway service)
• ModelCatalogService — the served-model catalog (always)

Parameters

Parameter	Type	Description
`config`		Gateway channel table and conversion metadata. Defaults to an empty configuration when omitted.
`converter`		Conversion engine implementing ``RelayConverterProtocol``. When ``None`` a startup diagnostic is logged and the gateway binding is skipped.
`http_client`		HTTP client driving the upstream adapter. When ``None`` a startup diagnostic is logged and the gateway binding is skipped.
`authorizer`		Optional authorizer enforced before dispatch.
`media_resolver`		Optional media resolver placed on the conversion context.
`billing`		Optional billing lifecycle; when ``None`` the gateway runs without admission control or settlement.
`converter_registry`		Converter registry backing health and metrics diagnostics and route quality. Optional; when ``None`` the diagnostic surfaces are unavailable (``DEPENDENCY_UNAVAILABLE``).
`channel_checker`		Optional channel checker driving per-channel probes; when ``None`` the health service reports unchecked channels.
`metrics_events`		Optional route event source feeding metrics aggregation; when ``None`` the metrics surface is unavailable.
`policy_store`		Optional runtime policy backend. ``None`` installs an in-process ``InMemoryRelayPolicyStore`` seeded from the configuration.
`audit`		Optional audit backend for control mutations. ``None`` disables audit emission.

__init__

def __init__(
    *,
    config: RelayGatewayConfig | None = None,
    converter: RelayConverterProtocol | None = None,
    http_client: HTTPClientProtocol | None = None,
    authorizer: AuthorizerProtocol | None = None,
    media_resolver: MediaResolverProtocol | None = None,
    billing: RelayBillingProtocol | None = None,
    converter_registry: RelayRegistryProtocol | None = None,
    channel_checker: RelayChannelCheckerProtocol | None = None,
    metrics_events: RelayRouteEventSourceProtocol | None = None,
    policy_store: RelayPolicyStoreProtocol | None = None,
    audit: AIAuditStoreProtocol | None = None
) -> None

source

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register the gateway configuration, registry, and service.

The configuration and registry are always bound. The gateway service itself is only bound when the converter and HTTP client are both present; otherwise a startup diagnostic is logged so the missing dependency is discoverable.

Parameters

Parameter	Type	Description
`container`	ContainerRegistrarProtocol	The container registrar to bind into.

boot

async def boot(container: BootContainerProtocol) -> None

source

Reconcile durable channels and policy drains into selection.

When the container resolves RelayChannelStoreProtocol, the durable rows are merged over the static configuration and installed in the runtime registry before the policy drain runs. Channels the policy store marks disabled (while the static configuration still enables them) are drained in the runtime registry so dispatch honors the persisted policy from the first request onward. When no converter or HTTP client was injected at construction, the container’s own RelayConverterProtocol and HTTPClientProtocol bindings are resolved here and the gateway services are bound late, so the module-only composition (RelayModule + RelayGatewayModule + HTTPModule) gets a working gateway without caller-owned instances. When auto-testing is enabled, the background sweep is started after reconciliation.

Parameters

Parameter	Type	Description
`container`	BootContainerProtocol	The booted container used to resolve the policy store, channel registry, optional channel store, and late-bound gateway dependencies.

shutdown

async def shutdown() -> None

source

Stop the background auto-tester, if one was started.

---

RelayGatewayService

source

Relay request lifecycle (buffered and streaming).

The service is stateless between requests and never touches request headers, payloads, or upstream details in error messages; errors are always safe RelayGatewayError values.

Attributes: _converter: Engine implementing RelayConverterProtocol. _codec: Wire DTO codec. _registry: Deterministic channel selector. _upstream: HTTP transport adapter. _config: Gateway configuration (channel table and model suffixes). _authorizer: Optional authorization check before dispatch. _billing: Optional billing lifecycle; when None admission and settlement are skipped. _media_resolver: Optional URL-media resolver threaded into the conversion context. _streams: Optional registry of active streams used to expose in-flight streams and cancel handles to operators; None disables stream registration (but not streaming itself). _failover: Optional consecutive-failure tracker; when None upstream failures never affect runtime selection state.

__init__

def __init__(
    converter: RelayConverterProtocol,
    codec: RelayPayloadCodec,
    registry: RelayChannelRegistry,
    upstream: HTTPUpstreamAdapter,
    config: RelayGatewayConfig,
    *,
    authorizer: AuthorizerProtocol | None = None,
    billing: RelayBillingProtocol | None = None,
    media_resolver: MediaResolverProtocol | None = None,
    streams: RelayStreamRegistry | None = None,
    failover: RelayFailoverTracker | None = None
) -> None

source

Bind the service to its dependencies.

Parameters

Parameter	Type	Description
`converter`	RelayConverterProtocol	Conversion engine for request/response payloads.
`codec`	RelayPayloadCodec	Wire DTO decoder/encoder.
`registry`	RelayChannelRegistry	Channel selection registry.
`upstream`	HTTPUpstreamAdapter	Upstream transport adapter.
`config`	RelayGatewayConfig	Static gateway configuration.
`authorizer`	AuthorizerProtocol | None	Optional authorizer; when ``None`` authorization is skipped.
`billing`	RelayBillingProtocol | None	Optional billing lifecycle; when ``None`` the gateway runs without admission control or settlement.
`media_resolver`	MediaResolverProtocol | None	Optional URL-media resolver placed on the conversion context; ``None`` disables media resolution.
`streams`	RelayStreamRegistry | None	Optional stream registry for operator visibility and forced cancellation; ``None`` keeps streaming functional without registry bookkeeping.
`failover`	RelayFailoverTracker | None	Optional consecutive-failure tracker; when ``None`` upstream failures never affect runtime selection state.

handle

async def handle(request: RelayGatewayRequest) -> Result[RelayGatewayResult, RelayGatewayError]

source

Run the buffered or streaming relay lifecycle for one request.

Dependencies run in fixed order: authorize, select channel, reserve billing capacity, convert request, then either call upstream, decode, convert response back, and settle (buffered), or create the stream session and hand back a lazy stream that consumes upstream events and settles when exhausted (streaming). Any preflight failure short-circuits the pipeline.

Parameters

Parameter	Type	Description
`request`	RelayGatewayRequest	The gateway request.

Returns

Type	Description
Result[RelayGatewayResult, RelayGatewayError]	``Ok(RelayGatewayResult)`` on success, or ``Err(RelayGatewayError)`` on the first failure. Unexpected
exceptions from dependencies never escape	they are logged and mapped to a generic ``CONVERSION_FAILED`` error.

---

RelayHealthService

source

Aggregate per-channel health and converter diagnostics.

Status rules, evaluated in order per channel:

• enabled=False config flag -> unavailable (channel_disabled); the model count still reflects aliases.
• Runtime policy drained the channel -> unavailable (drained).
• No checker registered -> unavailable (dependency_missing).
• Probe returns None -> unavailable (no_probe_result).
• Probe fails or exceeds the channel timeout -> failed (probe_failed / probe_timeout), counting one failure.
• Probe ok but latency at/above the degradation threshold -> degraded (high_latency).
• Otherwise -> healthy.

The failure precedence failed > degraded > unavailable > healthy holds because disabled/missing cases are decided before probing, and failures are decided before latency thresholds.

__init__

def __init__(
    registry: RelayChannelRegistry,
    checker: RelayChannelCheckerProtocol | None = None,
    converter: RelayRegistryProtocol | None = None,
    policy: RelayPolicyStoreProtocol | None = None,
    degraded_latency_ms: float = 200.0
) -> None

source

Bind the health service to its dependencies.

Parameters

Parameter	Type	Description
`registry`	RelayChannelRegistry	Static channel table; the only source of channels.
`checker`	RelayChannelCheckerProtocol | None	Optional upstream probe. ``None`` means every channel is reported ``unavailable``.
`converter`	RelayRegistryProtocol | None	Optional converter registry used by ``registry_diagnostics``. ``None`` makes diagnostics a failed dependency.
`policy`	RelayPolicyStoreProtocol | None	Optional runtime policy store. A channel drained through the store is reported ``unavailable`` with detail code ``drained``. ``None`` disables the check.
`degraded_latency_ms`	float	Latency at/above which a working probe is reported ``degraded``. Defaults to 200 ms.

channel_health

async def channel_health() -> Sequence[RelayChannelHealth]

source

Return a health snapshot per configured channel.

Channels are reported in configuration order; every channel gets exactly one snapshot.

Returns

Type	Description
Sequence[RelayChannelHealth]	One snapshot per channel, in configuration order.

registry_diagnostics

async def registry_diagnostics() -> RelayRegistryDiagnostics

source

Return converter capability diagnostics.

Returns

Type	Description
RelayRegistryDiagnostics	Converter identifier, version, mapper ids, and supported route pairs.

Raises

Exception	Description
RelayGatewayError	With ``DEPENDENCY_UNAVAILABLE`` when no converter registry is registered.

---

RelayMetricsService

source

Aggregate route metrics for the admin operations surface.

Counts, per directed route and window:

• request_count from request_completed events.
• loss_counts from conversion_loss codes (never free-form messages).
• unsupported_count from unsupported_feature events.
• stream_failure_count from stream cancelled/timeout/truncated events.

A missing event source is a failed dependency; an empty source yields a stable empty result, never a fabricated zero-count row.

__init__

def __init__(
    events: RelayRouteEventSourceProtocol | None,
    converter: RelayRegistryProtocol | None = None,
    registration_errors: tuple[str, Ellipsis] = ()
) -> None

source

Bind the metrics service to its dependencies.

Parameters

Parameter	Type	Description
`events`	RelayRouteEventSourceProtocol | None	Operational event source for route aggregation. ``None`` makes ``route_metrics`` a failed dependency.
`converter`	RelayRegistryProtocol | None	Optional converter registry used for route-quality and diagnostics. ``None`` makes ``registry_diagnostics`` a failed dependency.
`registration_errors`	tuple[str, Ellipsis]	Registrations that failed at wiring time, surfaced verbatim in diagnostics.

route_metrics

async def route_metrics(window: TimeWindow) -> Sequence[RelayRouteMetrics]

source

Return per-route metrics aggregated inside window.

Parameters

Parameter	Type	Description
`window`	TimeWindow	Bounded aggregation window.

Returns

Type	Description
Sequence[RelayRouteMetrics]	One row per route that saw activity within the window.

Raises

Exception	Description
RelayGatewayError	With ``DEPENDENCY_UNAVAILABLE`` when no event source is registered.

registry_diagnostics

async def registry_diagnostics() -> RelayRegistryDiagnostics

source

Return converter capability diagnostics.

Returns

Type	Description
RelayRegistryDiagnostics	Converter identifier, version, mapper ids, supported route pairs, and startup registration failures.

Raises

Exception	Description
RelayGatewayError	With ``DEPENDENCY_UNAVAILABLE`` when no converter registry is registered.

---

RelayPayloadCodec

source

Decode and encode relay wire payloads as typed DTOs.

The codec is stateless; a single instance can be shared. Decoding rejects malformed JSON, non-object roots, and DTOs missing required fields; unknown wire fields are preserved verbatim in the DTO passthrough dict and re-emitted on encode.

decode_request

def decode_request(
    source: RelayFormat,
    raw: bytes,
    request_id: str
) -> Result[WireRequest, RelayGatewayError]

source

Decode wire JSON bytes into the request DTO for source.

Parameters

Parameter	Type	Description
`source`	RelayFormat	Wire format the payload claims to be.
`raw`	bytes	Raw request body bytes.
`request_id`	str	Caller-supplied request id stamped on errors.

Returns

Type	Description
Result[WireRequest, RelayGatewayError]	``Ok(dto)`` with unknown fields preserved in the DTO's ``passthrough``, or ``Err(RelayGatewayError)`` classifying malformed JSON (``INVALID_REQUEST``), non-object roots (``INVALID_REQUEST``), unknown formats (``UNSUPPORTED_FORMAT``), and missing required fields (``INVALID_REQUEST`` carrying the field path).

decode_response_payload

def decode_response_payload(
    target: RelayFormat,
    data: dict[str, Any],
    request_id: str
) -> Result[RelayResponsePayload, RelayGatewayError]

source

Decode an upstream wire dict into the response DTO for target.

Parameters

Parameter	Type	Description
`target`	RelayFormat	Wire format the upstream claims to speak.
`data`	dict[str, Any]	Decoded upstream response body.
`request_id`	str	Caller-supplied request id stamped on errors.

Returns

Type	Description
Result[RelayResponsePayload, RelayGatewayError]	``Ok(dto)`` with unknown fields preserved in the DTO's ``passthrough``, or ``Err(RelayGatewayError)`` classifying unknown formats (``UNSUPPORTED_FORMAT``) and DTOs missing required fields (``UPSTREAM_MALFORMED`` — a malformed upstream response is a 502, not a client 400).

encode

def encode(dto: WireRequest) -> Result[bytes, RelayGatewayError]

source

Serialize a request DTO to wire JSON bytes.

Parameters

Parameter	Type	Description
`dto`	WireRequest	Typed request DTO to serialize.

Returns

Type	Description
Result[bytes, RelayGatewayError]	``Ok(bytes)`` with ``None`` fields omitted and falsey values preserved, or ``Err(RelayGatewayError)`` with code ``ENCODE_FAILED`` when the payload cannot be serialized.

---

RelayRouteEvent

source

One operational event feeding route metric aggregation.

Attributes: kind: Event kind; only conversion_loss events carry a code. source: Source wire format of the route. target: Target wire format of the route. occurred_at: When the event happened (UTC). loss_code: Stable conversion loss code for conversion_loss events; ignored for every other kind.

---

RelayStreamRegistry

source

Tracks active streams and their cancel handles.

Attributes: _active: Stream identifier to stream metadata, oldest first by insertion order. _handles: Stream identifier to its cancel handle.

__init__

def __init__() -> None

source

Create an empty registry.

register

def register(
    *,
    channel: str,
    model: str,
    request_id: str
) -> tuple[str, asyncio.Event]

source

Register a new in-flight stream.

Parameters

Parameter	Type	Description
`channel`	str	Channel name serving the stream.
`model`	str	Outbound model alias of the stream.
`request_id`	str	Gateway request identifier.

Returns

Type	Description
tuple[str, asyncio.Event]	The new stream identifier and its cancel handle; setting the handle asks the relay loop to terminate truncated.

unregister

def unregister(stream_id: str) -> None

source

Forget a finished stream and its handle.

Parameters

Parameter	Type	Description
`stream_id`	str	Identifier previously returned by ``register``.

list

def list() -> tuple[RelayActiveStream, Ellipsis]

source

Return active streams, oldest first.

Returns

Type	Description
tuple[RelayActiveStream, Ellipsis]	A tuple of active stream rows; empty when nothing is in flight.

handle

def handle(stream_id: str) -> asyncio.Event | None

source

Return the cancel handle of stream_id, or None.

Parameters

Parameter	Type	Description
`stream_id`	str	Stream identifier.

Returns

Type	Description
asyncio.Event | None	The cancel handle when the stream is active, else ``None``.

cancel

def cancel(stream_id: str) -> bool

source

Request cancellation of stream_id.

Parameters

Parameter	Type	Description
`stream_id`	str	Stream identifier.

Returns

Type	Description
bool	``True`` when the stream was active and its handle was set; ``False`` when the stream is unknown.

---

UpstreamEventParser

source

Frame upstream chunks into target session events.

The parser decodes one chunk at a time, classifies it (keepalive, delta, terminal, or error), and forwards source DTOs to the injected session. It never accumulates text or tool arguments.

__init__

def __init__(
    session: RelayStreamSessionProtocol,
    source: RelayFormat,
    *,
    request_id: str
) -> None

source

Bind the parser to a session and source wire format.

Parameters

Parameter	Type	Description
`session`	RelayStreamSessionProtocol	Stateful stream session accepting source DTOs and emitting target events.
`source`	RelayFormat	Wire format of the upstream stream.
`request_id`	str	Id stamped on the malformed-stream errors this parser raises.

parse

def parse(chunk: UpstreamChunk) -> _Parsed

source

Frame one upstream chunk into target session events.

Parameters

Parameter	Type	Description
`chunk`	UpstreamChunk	One raw upstream frame.

Returns

Type	Description
The framed outcome	emitted target events plus terminal and error classification. A transport-level ``terminal=True`` chunk short-circuits decoding.

Raises

Exception	Description
RelayGatewayError	With code ``UPSTREAM_MALFORMED`` (502, never retryable) when the payload is malformed JSON, fails DTO validation, or is rejected by the session.

finalize

def finalize() -> tuple[Any, Ellipsis]

source

Close the session deterministically exactly once.

The first call runs the session finalize and caches its events; subsequent calls return the cached result without touching the session again.

Returns

Type	Description
tuple[Any, Ellipsis]	The session's terminal events.

---

Functions

relay_stream

relay_stream

async def relay_stream(
    upstream: RelayUpstreamProtocol,
    request: UpstreamRequest,
    parser: UpstreamEventParser,
    cancel_handle: asyncio.Event | None = None
) -> AsyncIterator[RelayWireEvent]

source

Relay one upstream stream with cancellation and session lifecycle.

The async for inside this generator consumes exactly one upstream chunk per consumer __anext__: backpressure is inherent and there is no buffering or prefetch.

Lifecycle: terminal frames finalize the session with no cancellation; error frames and consumer disconnects cancel upstream once and finalize truncated; streams that end without a terminal marker (for example Gemini or a cut SSE stream) finalize truncated without cancelling. Upstream cancel and session finalize each run at most once, even across nested exception paths.

Parameters

Parameter	Type	Description
`upstream`	RelayUpstreamProtocol	The upstream transport implementing ``RelayUpstreamProtocol``.
`request`	UpstreamRequest	The fully-resolved upstream request.
`parser`	UpstreamEventParser	Stateful session parser whose bookkeeping attributes (``finalized``, ``truncated``, ``cancelled``) track the stream.
`cancel_handle`	asyncio.Event | None	Optional operator cancel handle from the stream registry. When set, the relay cancels upstream once and finalizes truncated at the next chunk boundary.

Yields

Type	Description
AsyncIterator[RelayWireEvent]	Normalized ``RelayWireEvent`` values; terminal flag on the last event of each finalize batch.

Raises

Exception	Description
RelayGatewayError	Malformed upstream framing or a session rejection (502, never retryable).
asyncio.CancelledError	Upstream or consumer task cancellation; always re-raised.
GeneratorExit	The consumer closed the generator mid-stream.

---