Coverage for src / lexigram / contracts / ai / relay / context.py: 0%
47 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-15 18:57 +0800
1"""Relay conversion context, options, and media callbacks.
3The engine is synchronous and side-effect free. Everything the engine
4may need that depends on the host (Claude default ``max_tokens``,
5Gemini safety thresholds, media resolution, model capabilities) is
6supplied here as typed callbacks instead of globals.
7"""
9from __future__ import annotations
11from collections.abc import Callable
12from dataclasses import dataclass, field
13from typing import Protocol, TypeAlias, runtime_checkable
15from lexigram.contracts.ai.exceptions import RelayError
16from lexigram.contracts.ai.relay.types import RelayLoss
17from lexigram.contracts.core.result import Result
19__all__ = [
20 "ClaudeOptions",
21 "DefaultMaxTokensCallback",
22 "GeminiOptions",
23 "MediaResolverProtocol",
24 "PreserveThinkingSuffixCallback",
25 "RelayConversionContext",
26 "RelayOptions",
27 "SafetySettingCallback",
28 "SupportsImageGenerationCallback",
29]
32@dataclass(frozen=True)
33class ClaudeOptions:
34 """Claude-specific conversion adaptations.
36 Attributes:
37 thinking_adapter_enabled: Whether the Claude thinking adapter is
38 active. ``False`` disables the adaptation entirely.
39 thinking_budget_percentage: Percentage of ``max_tokens`` the
40 thinking adapter reserves for thinking output. ``0`` disables
41 the budget calculation.
42 minimum_max_tokens: Floor the adapter applies to ``max_tokens``.
43 ``0`` disables the floor.
44 """
46 thinking_adapter_enabled: bool = False
47 thinking_budget_percentage: int = 0
48 minimum_max_tokens: int = 0
51@dataclass(frozen=True)
52class GeminiOptions:
53 """Gemini-specific conversion adaptations.
55 Attributes:
56 thinking_adapter_enabled: Whether the Gemini thinking adapter is
57 active. ``False`` disables the adaptation entirely.
58 thinking_budget: Token budget for Gemini ``thinkingBudget``.
59 ``0`` disables the budget calculation.
60 thought_signature_bypass: Whether the thought-signature bypass
61 policy is enabled for models that require it.
62 """
64 thinking_adapter_enabled: bool = False
65 thinking_budget: int = 0
66 thought_signature_bypass: bool = False
69@dataclass(frozen=True)
70class RelayOptions:
71 """Cross-protocol conversion options.
73 Zero-value options disable adaptations and must not add fields to
74 outgoing payloads.
76 Attributes:
77 claude: Claude thinking/max_tokens adaptations.
78 gemini: Gemini thinking/signature adaptations.
79 model_suffix_preserved: Whether provider model suffixes (e.g.
80 ``:thinking``) are preserved verbatim.
81 openrouter_dialects: Whether OpenRouter-compatible dialect flags
82 are honored. Only meaningful when the host enables them.
83 """
85 claude: ClaudeOptions = field(default_factory=ClaudeOptions)
86 gemini: GeminiOptions = field(default_factory=GeminiOptions)
87 model_suffix_preserved: bool = False
88 openrouter_dialects: bool = False
91DefaultMaxTokensCallback: TypeAlias = Callable[[str], int | None]
92"""Return a default ``max_tokens`` for a model, or ``None``."""
94SafetySettingCallback: TypeAlias = Callable[[str], str | None]
95"""Return a Gemini safety threshold for a category, or ``None``."""
97SupportsImageGenerationCallback: TypeAlias = Callable[[str], bool]
98"""Whether a model supports Gemini image generation."""
100PreserveThinkingSuffixCallback: TypeAlias = Callable[[str], bool]
101"""Whether a model requires the thinking-suffix bypass policy."""
104@runtime_checkable
105class MediaResolverProtocol(Protocol):
106 """Resolves URL media into wire-ready base64.
108 The engine never performs network I/O. When a source payload carries
109 a URL the target protocol cannot consume directly, the engine calls
110 the resolver supplied through :class:`RelayConversionContext`. The
111 gateway may pre-resolve media asynchronously before calling the
112 engine.
113 """
115 def resolve(self, url: str) -> Result[tuple[str, str], RelayError]:
116 """Resolve *url* into ``(media_type, base64_data)``.
118 Args:
119 url: The source URL that requires conversion.
121 Returns:
122 ``Ok((media_type, base64_data))`` on success, or
123 ``Err(RelayError)`` when resolution fails.
124 """
125 ...
128@dataclass(frozen=True)
129class RelayConversionContext:
130 """Host-supplied context for one conversion.
132 Attributes:
133 options: Cross-protocol adaptation options.
134 default_max_tokens: Claude ``max_tokens`` fallback when the source
135 omitted it. ``None`` means no fallback exists.
136 safety_setting: Gemini safety threshold lookup by category.
137 supports_image_generation: Gemini image-generation capability
138 lookup by model.
139 preserve_thinking_suffix: Thinking-suffix bypass policy lookup.
140 media_resolver: Resolver for URL media, or ``None``.
141 upstream_model: Host model name substituted when the source
142 payload carries no model (e.g. Gemini responses). Empty
143 string disables substitution.
144 losses: Per-conversion loss records appended by mappers; copied
145 into the ``RelayConvertResult`` by the engine.
146 request_id: Caller-supplied request id stamped on losses and
147 errors during conversion. Empty string when not provided.
148 channel_name: Name of the selected relay channel, used for
149 channel-aware adaptation and audit. Empty string when not
150 provided.
151 """
153 options: RelayOptions = field(default_factory=RelayOptions)
154 default_max_tokens: DefaultMaxTokensCallback | None = None
155 safety_setting: SafetySettingCallback | None = None
156 supports_image_generation: SupportsImageGenerationCallback | None = None
157 preserve_thinking_suffix: PreserveThinkingSuffixCallback | None = None
158 media_resolver: MediaResolverProtocol | None = None
159 upstream_model: str = ""
160 losses: list[RelayLoss] = field(default_factory=list)
161 request_id: str = ""
162 channel_name: str = ""