myco_model/
lib.rs

1//! Backend-independent messages and one-attempt streaming model drivers.
2//!
3//! Construct a driver with [`new`] and [`GenerativeModelConfig`], then pass a
4//! complete conversation to [`GenerativeModel::generate`]. The library accepts
5//! resolved settings and credentials; it does not read application config files.
6//!
7//! # Streaming and policy
8//!
9//! Every call performs one attempt. Consume [`GenerationEvent`] directly for
10//! incremental output and failure metadata, feeding parts to [`MessageAccumulator`].
11//! Use [`GenerateOutput::from_generation`] when only the completed output is needed.
12//! Dropping a stream cancels its request. Retry, tool execution, and persistence
13//! belong to the caller; `myco-agent` supplies the model/tool loop and retry policy.
14//!
15//! # History contract
16//!
17//! [`Message::ToolResults`] must immediately follow its assistant's tool calls,
18//! with one result per call in the same order. Wire IDs are generated by the
19//! drivers. Preserve these pairs when saving or slicing history.
20//!
21//! See the [inference guide](https://tsnl.github.io/myco/developers/inference.html)
22//! and the `inference` example for a complete streaming client. The example
23//! requires an endpoint; ordinary library tests use local fixtures.
24
25use std::{pin::pin, sync::Arc};
26
27use futures::{Stream, StreamExt};
28
29pub type AsyncStream<T> = std::pin::Pin<Box<dyn Stream<Item = T> + Send>>;
30
31#[cfg(test)]
32mod test_support;
33
34mod anthropic;
35pub use anthropic::AnthropicBackendConfig;
36
37mod accumulator;
38pub use accumulator::MessageAccumulator;
39
40mod driver_core;
41
42mod openai_common;
43pub use openai_common::OpenAIBackendConfig;
44
45mod openai_completions;
46mod openai_responses;
47
48mod sse_parser;
49use sse_parser::SseParser;
50
51pub trait GenerativeModel: Send + Sync {
52    /// One attempt. Failure ends the stream; dropping it cancels the request.
53    ///
54    /// A successful stream starts with [`MessagePart::MessageStart`] and includes
55    /// a [`MessagePart::TurnEndReason`]. A custom model should emit parts that
56    /// [`MessageAccumulator`] can validate. Calls do not execute tools or retry.
57    fn generate(&self, input: &[Message]) -> AsyncStream<GenerationEvent>;
58}
59
60/// Wire protocol a model is served over.
61///
62/// Serde strings are the config.toml `protocol` values.
63#[derive(
64    Debug,
65    Clone,
66    Copy,
67    PartialEq,
68    Eq,
69    Hash,
70    serde::Serialize,
71    serde::Deserialize,
72    schemars::JsonSchema,
73)]
74pub enum Protocol {
75    #[serde(rename = "anthropic-messages")]
76    AnthropicMessages,
77    #[serde(rename = "openai-responses")]
78    OpenAIResponses,
79    #[serde(rename = "openai-completions")]
80    OpenAICompletions,
81}
82
83impl std::fmt::Display for Protocol {
84    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
85        match self {
86            Protocol::AnthropicMessages => f.write_str("anthropic-messages"),
87            Protocol::OpenAIResponses => f.write_str("openai-responses"),
88            Protocol::OpenAICompletions => f.write_str("openai-completions"),
89        }
90    }
91}
92
93/// How thinking/reasoning is requested for a model.
94///
95/// Serde strings are the config.toml `thinking` values. Compatibility is
96/// per-protocol (validated at catalog resolution): Anthropic Messages takes
97/// `adaptive` | `budget` | `none`; OpenAI Responses takes `effort` | `none`.
98#[derive(
99    Debug,
100    Clone,
101    Copy,
102    PartialEq,
103    Eq,
104    Hash,
105    serde::Serialize,
106    serde::Deserialize,
107    schemars::JsonSchema,
108)]
109#[serde(rename_all = "lowercase")]
110pub enum ThinkingMode {
111    /// Anthropic `thinking.type: "adaptive"` + `output_config.effort`
112    /// (frontier models; older models reject it).
113    Adaptive,
114    /// Anthropic `thinking.type: "enabled"` + a `budget_tokens` mapped from
115    /// [`Effort`] (e.g. Haiku 4.5).
116    Budget,
117    /// OpenAI-style `reasoning.effort`.
118    Effort,
119    /// Do not request thinking.
120    None,
121}
122
123impl std::fmt::Display for ThinkingMode {
124    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
125        f.write_str(match self {
126            ThinkingMode::Adaptive => "adaptive",
127            ThinkingMode::Budget => "budget",
128            ThinkingMode::Effort => "effort",
129            ThinkingMode::None => "none",
130        })
131    }
132}
133
134impl ThinkingMode {
135    /// Default mode when a catalog entry does not set `thinking`.
136    pub fn default_for(protocol: Protocol) -> Self {
137        match protocol {
138            Protocol::AnthropicMessages => ThinkingMode::Adaptive,
139            Protocol::OpenAIResponses | Protocol::OpenAICompletions => ThinkingMode::Effort,
140        }
141    }
142
143    /// Whether this mode is servable over `protocol`.
144    pub fn compatible_with(self, protocol: Protocol) -> bool {
145        match protocol {
146            Protocol::AnthropicMessages => {
147                matches!(
148                    self,
149                    ThinkingMode::Adaptive | ThinkingMode::Budget | ThinkingMode::None
150                )
151            }
152            Protocol::OpenAIResponses | Protocol::OpenAICompletions => {
153                matches!(self, ThinkingMode::Effort | ThinkingMode::None)
154            }
155        }
156    }
157}
158
159/// Agent policy for retrying transient generation failures, discarding failed drafts.
160/// Resolved from `[gateways.NAME.retry]` or `[models.KEY.retry]`.
161#[derive(Debug, Clone, Copy, PartialEq, serde::Serialize, serde::Deserialize)]
162pub struct RetryPolicy {
163    /// Total attempts including the first. `1` disables retry.
164    pub max_attempts: u32,
165    pub initial_backoff: std::time::Duration,
166    /// Ceiling on one wait. A longer provider minimum must stop retries rather
167    /// than cause an earlier request or an unbounded wait.
168    pub max_backoff: std::time::Duration,
169    pub backoff_multiplier: f64,
170    /// Optional elapsed recovery budget, starting at the first retryable failure.
171    /// Includes later waits and requests; the original request has no new deadline.
172    #[serde(default)]
173    pub max_elapsed: Option<std::time::Duration>,
174}
175
176impl Default for RetryPolicy {
177    fn default() -> Self {
178        Self {
179            max_attempts: 3,
180            initial_backoff: std::time::Duration::from_millis(500),
181            max_backoff: std::time::Duration::from_secs(30),
182            backoff_multiplier: 2.0,
183            max_elapsed: None,
184        }
185    }
186}
187
188impl RetryPolicy {
189    /// Wait before `attempt` (1-based). Attempt 1 is the original send and
190    /// never waits; `retry_after` is the provider's ask, honoured when it
191    /// exceeds the capped computed backoff. Callers must stop if the resulting
192    /// minimum exceeds [`Self::max_backoff`] or their remaining elapsed budget.
193    ///
194    /// No jitter: myco is one client per user, so there is no fleet to
195    /// de-synchronise, and a deterministic schedule is one less thing to
196    /// reason about when reading an overnight log.
197    pub fn backoff(
198        &self,
199        attempt: u32,
200        retry_after: Option<std::time::Duration>,
201    ) -> std::time::Duration {
202        if attempt <= 1 {
203            return std::time::Duration::ZERO;
204        }
205        let steps = attempt.saturating_sub(2);
206        let factor = self.backoff_multiplier.max(1.0).powi(steps.min(32) as i32);
207        let millis = (self.initial_backoff.as_millis() as f64 * factor).min(u64::MAX as f64);
208        let computed = std::time::Duration::from_millis(millis as u64);
209        retry_after
210            .unwrap_or(std::time::Duration::ZERO)
211            .max(computed.min(self.max_backoff))
212    }
213}
214
215/// A resolved model: everything the protocol drivers need, minus credentials
216/// (those live in [`BackendConfig`]). Built by the application configuration from the
217/// `[models]` / `[gateways]` catalog in config.toml — myco ships no built-in
218/// models.
219#[derive(Debug, Clone, PartialEq, Eq)]
220pub struct ModelSpec {
221    /// Catalog key: what the user types after `--model` and what sessions
222    /// record. Distinct from `api_id` so one wire model can appear under
223    /// several keys (e.g. routed via different gateways).
224    pub key: String,
225    /// Wire id sent to the provider (the request `model` field).
226    pub api_id: String,
227    pub protocol: Protocol,
228    pub thinking: ThinkingMode,
229    /// Context window for UX (`USER n/m`) and auto-compact heuristics.
230    pub context_window_tokens: u64,
231    /// Largest image this model accepts, measured on the base64 payload.
232    /// Enforced locally by `view_image` and by REPL `@path` attachments so an
233    /// oversized image fails with a clear message instead of a provider 400.
234    /// Always concrete: config resolution applies the `max_image_base64_bytes`
235    /// entry or its default, and callers downstream take this value.
236    pub max_image_base64_bytes: u64,
237    /// How many consecutive `max_tokens` truncations one turn resumes through
238    /// before handing control back (`0` never resumes). Resolved from the
239    /// model's `max_truncated_resumes` or its default; the agent takes this
240    /// value via the agent's continuation policy.
241    pub max_truncated_resumes: u32,
242    /// Prompt size at which the session runner compacts without being asked.
243    /// Myco's config defaults to the full context window. Library callers can
244    /// use `None` when they own compaction policy themselves.
245    ///
246    /// Resolved from the model's `auto_compact_at` *fraction* and output reserve against
247    /// `context_window_tokens` so the comparison downstream is a plain token
248    /// count, and the fraction is validated once, at startup.
249    pub auto_compact_at_tokens: Option<u64>,
250}
251
252impl std::fmt::Display for ModelSpec {
253    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
254        f.write_str(&self.key)
255    }
256}
257
258/// One usable catalog entry: spec plus the backend (gateway + credentials)
259/// that serves it.
260#[derive(Debug, Clone)]
261pub struct CatalogModel {
262    pub spec: ModelSpec,
263    pub backend: BackendConfig,
264    /// Set when the auth source did not resolve (env var unset, auth file
265    /// unreadable). Reported by [`ModelCatalog::get`] when the model is
266    /// actually used — configuring a model without its credential is fine
267    /// until then.
268    pub auth_error: Option<String>,
269}
270
271/// Key → model catalog resolved from config.toml. Empty when the user has not
272/// configured any models.
273#[derive(Debug, Clone, Default)]
274pub struct ModelCatalog {
275    entries: std::collections::BTreeMap<String, CatalogModel>,
276}
277
278impl ModelCatalog {
279    pub fn new(entries: std::collections::BTreeMap<String, CatalogModel>) -> Self {
280        Self { entries }
281    }
282
283    /// Look up a usable model. Errors are user-actionable: unknown keys list
284    /// the configured catalog; entries with unresolved credentials report the
285    /// failing auth source (env var / file).
286    pub fn get(&self, key: &str) -> Result<&CatalogModel, String> {
287        let Some(entry) = self.entries.get(key) else {
288            if self.entries.is_empty() {
289                return Err(format!(
290                    "unknown model {key:?}: no models configured — define [models] \
291                     (and [gateways]) in config.toml"
292                ));
293            }
294            return Err(format!(
295                "unknown model {key:?}; configured models: [{}]",
296                self.keys().join(", ")
297            ));
298        };
299        if let Some(err) = &entry.auth_error {
300            return Err(err.clone());
301        }
302        Ok(entry)
303    }
304
305    /// Key exists (regardless of whether its credential resolved).
306    pub fn contains(&self, key: &str) -> bool {
307        self.entries.contains_key(key)
308    }
309
310    /// Spec for `key`, ignoring credential state. For settings that must be
311    /// read while merely *configuring* a run (the image cap host workers are
312    /// spawned with) rather than using the model — [`Self::get`] is still the
313    /// gate for that.
314    pub fn spec(&self, key: &str) -> Option<&ModelSpec> {
315        self.entries.get(key).map(|entry| &entry.spec)
316    }
317
318    pub fn keys(&self) -> Vec<&str> {
319        self.entries.keys().map(String::as_str).collect()
320    }
321
322    pub fn is_empty(&self) -> bool {
323        self.entries.is_empty()
324    }
325}
326
327/// Reasoning / extended-thinking effort level sent to providers.
328///
329/// Anthropic adaptive models map this to `output_config.effort`; Haiku-style models
330/// map it onto a `thinking.budget_tokens` value. OpenAI/xAI gateways receive it as
331/// `reasoning.effort`.
332#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
333#[serde(rename_all = "lowercase")]
334pub enum Effort {
335    Low,
336    Medium,
337    High,
338    Max,
339}
340
341impl Effort {
342    /// Wire string used by Anthropic `output_config.effort` and OpenAI `reasoning.effort`.
343    pub fn as_str(self) -> &'static str {
344        match self {
345            Effort::Low => "low",
346            Effort::Medium => "medium",
347            Effort::High => "high",
348            Effort::Max => "max",
349        }
350    }
351
352    /// Approximate Anthropic extended-thinking token budget for non-adaptive models.
353    pub fn budget_tokens(self) -> u32 {
354        match self {
355            Effort::Low => 1_024,
356            Effort::Medium => 4_096,
357            Effort::High => 16_000,
358            Effort::Max => 64_000,
359        }
360    }
361
362    /// Sensible default for interactive agent sessions.
363    pub const DEFAULT: Effort = Effort::High;
364}
365
366impl std::fmt::Display for Effort {
367    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
368        f.write_str(self.as_str())
369    }
370}
371
372impl std::str::FromStr for Effort {
373    type Err = String;
374
375    fn from_str(s: &str) -> Result<Self, Self::Err> {
376        match s.trim().to_ascii_lowercase().as_str() {
377            "low" | "l" => Ok(Effort::Low),
378            "medium" | "med" | "m" => Ok(Effort::Medium),
379            "high" | "h" => Ok(Effort::High),
380            "max" | "x" => Ok(Effort::Max),
381            other => Err(format!(
382                "unknown effort {other:?}; expected low|medium|high|max"
383            )),
384        }
385    }
386}
387
388/// Provider backend settings: gateway base URL, credential, per-request knobs.
389#[derive(Debug, Clone)]
390pub enum BackendConfig {
391    Anthropic(AnthropicBackendConfig),
392    /// Responses API (`{base_url}/responses`).
393    OpenAIResponses(OpenAIBackendConfig),
394    /// Chat Completions API (`{base_url}/chat/completions`).
395    OpenAICompletions(OpenAIBackendConfig),
396}
397
398impl BackendConfig {
399    pub fn retry_policy(&self) -> RetryPolicy {
400        match self {
401            Self::Anthropic(config) => config.retry,
402            Self::OpenAIResponses(config) | Self::OpenAICompletions(config) => config.retry,
403        }
404    }
405
406    pub fn protocol(&self) -> Protocol {
407        match self {
408            BackendConfig::Anthropic(_) => Protocol::AnthropicMessages,
409            BackendConfig::OpenAIResponses(_) => Protocol::OpenAIResponses,
410            BackendConfig::OpenAICompletions(_) => Protocol::OpenAICompletions,
411        }
412    }
413}
414
415/// Fully resolved driver inputs. Tool schemas advertise capabilities; the caller
416/// remains responsible for executing calls and validating their arguments.
417pub struct GenerativeModelConfig {
418    pub model: ModelSpec,
419    pub tools: Vec<ToolSpec>,
420    pub system_prompt: String,
421    pub backend_config: BackendConfig,
422}
423
424/// Construct a driver, rejecting a protocol/backend mismatch before any request.
425/// No network request is made until the returned model is asked to generate.
426pub fn new(config: GenerativeModelConfig) -> Result<Arc<dyn GenerativeModel>, ModelCreationError> {
427    if config.backend_config.protocol() != config.model.protocol {
428        return Err(ModelCreationError::BadConfig(format!(
429            "model `{}` speaks {} but the backend config is for {}",
430            config.model,
431            config.model.protocol,
432            config.backend_config.protocol()
433        )));
434    }
435    match config.backend_config.clone() {
436        BackendConfig::Anthropic(backend) => {
437            let model = anthropic::AnthropicGenerativeModel::new(config, backend)?;
438            Ok(model as Arc<dyn GenerativeModel>)
439        }
440        BackendConfig::OpenAIResponses(backend) => {
441            let model = openai_responses::OpenAIResponsesGenerativeModel::new(config, backend)?;
442            Ok(model as Arc<dyn GenerativeModel>)
443        }
444        BackendConfig::OpenAICompletions(backend) => {
445            let model = openai_completions::OpenAICompletionsGenerativeModel::new(config, backend)?;
446            Ok(model as Arc<dyn GenerativeModel>)
447        }
448    }
449}
450
451#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
452pub enum Message {
453    UserMessage {
454        content: Vec<Content>,
455    },
456    ToolResults {
457        tool_use_results: Vec<ToolResult>,
458    },
459    AssistantMessage {
460        content: Vec<Content>,
461        tool_uses: Vec<ToolUse>,
462        turn_end_reason: Option<TurnEndReason>,
463    },
464}
465
466impl Message {
467    pub fn content(&self) -> impl Iterator<Item = &Content> {
468        let (content, results): (&[Content], &[ToolResult]) = match self {
469            Self::UserMessage { content } | Self::AssistantMessage { content, .. } => {
470                (content, &[])
471            }
472            Self::ToolResults { tool_use_results } => (&[], tool_use_results),
473        };
474        content
475            .iter()
476            .chain(results.iter().flat_map(|result| &result.content))
477    }
478    /// Runtime-only input does not represent a human submission.
479    pub fn is_user_turn(&self) -> bool {
480        matches!(self, Self::UserMessage { content }
481            if content.is_empty() || content.iter().any(|part| !matches!(part, Content::System { .. })))
482    }
483}
484#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
485pub enum TurnEndReason {
486    EndTurn,
487    MaxTokens,
488    ToolUse,
489    /// Provider-specific / unknown stop reason (owned so sessions can serialize cleanly).
490    Other(String),
491}
492
493#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
494pub struct ToolSpec {
495    pub name: String,
496    pub description: String,
497    pub input_schema: serde_json::Value,
498}
499
500/// A tool call in an assistant turn. Carries no id: a call is identified by
501/// its position (message index + ordinal), and the `j`-th entry of the next
502/// message's `tool_use_results` answers it. Providers that need ids on the
503/// wire get minted ones from the driver.
504#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
505pub struct ToolUse {
506    pub name: String,
507    pub input: serde_json::Value,
508    /// Opaque token a provider attached to this call that it requires back on
509    /// the next turn. Gemini (the Chat Completions dialect) sends a
510    /// `thought_signature` and rejects the following request without it.
511    /// `None` for every other provider; absent on sessions written before the
512    /// field existed.
513    #[serde(default, skip_serializing_if = "Option::is_none")]
514    pub thought_signature: Option<String>,
515}
516
517#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
518pub struct ToolResourceRef {
519    pub id: String,
520    /// Distinguishes separate processes that reuse the same caller-chosen handle.
521    pub instance_id: String,
522}
523
524#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
525pub struct ToolResult {
526    pub content: Vec<Content>,
527    pub is_error: bool,
528    /// Brief tool-authored outcome for transcripts, independent of model narration.
529    #[serde(default, skip_serializing_if = "Option::is_none")]
530    pub status: Option<String>,
531    /// A retained resource observed by this call. Presentation metadata only;
532    /// provider adapters send `content`, not this handle, to the model.
533    #[serde(default, skip_serializing_if = "Option::is_none")]
534    pub resource: Option<ToolResourceRef>,
535}
536
537impl ToolResult {
538    pub fn ok(content: Vec<Content>) -> Self {
539        Self {
540            content,
541            is_error: false,
542            status: None,
543            resource: None,
544        }
545    }
546
547    pub fn text(text: impl Into<String>) -> Self {
548        Self {
549            content: vec![Content::Text { text: text.into() }],
550            is_error: false,
551            status: None,
552            resource: None,
553        }
554    }
555
556    pub fn err(text: impl Into<String>) -> Self {
557        Self {
558            content: vec![Content::Text { text: text.into() }],
559            is_error: true,
560            status: None,
561            resource: None,
562        }
563    }
564
565    pub fn with_status(mut self, status: impl Into<String>) -> Self {
566        self.status = Some(status.into());
567        self
568    }
569}
570
571#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
572pub enum Content {
573    Text {
574        text: String,
575    },
576    /// Runtime-authored context. Providers receive `text` at this position in
577    /// the message; transcript renderers omit the entire part. `data` retains
578    /// structured lifecycle facts independently of their model-facing wording.
579    System {
580        kind: String,
581        text: String,
582        data: serde_json::Value,
583    },
584    Image {
585        /// Data URL, remote URL, or legacy base64. Application-local
586        /// `myco-image:` references must be resolved before provider dispatch.
587        source: String,
588    },
589    /// Model thinking text (session history + live UI).
590    ///
591    /// Stored in agent/session history for resume, but **stripped when backends
592    /// compose the next API request** (not echoed as CoT). Prefer provider
593    /// summary channels over raw reasoning text.
594    Thinking {
595        text: String,
596        /// Opaque provider signature (Anthropic). Not re-sent on subsequent turns.
597        signature: Option<String>,
598        /// True for redacted/encrypted thinking placeholders with no plaintext.
599        redacted: bool,
600    },
601}
602
603/// Clone only answer blocks (`Text` / `Image`), dropping thinking.
604pub fn answer_content(content: &[Content]) -> Vec<Content> {
605    content
606        .iter()
607        .filter(|c| matches!(c, Content::Text { .. } | Content::Image { .. }))
608        .cloned()
609        .collect()
610}
611
612/// Check an encoded image payload without decoding or copying it.
613pub fn check_image_size(encoded_bytes: u64, limit: u64) -> Result<(), GenerateError> {
614    if encoded_bytes > limit {
615        return Err(GenerateError::RequestTooLargeError(format!(
616            "image has {encoded_bytes} base64 bytes; the current model's max_image_base64_bytes is {limit}. Resize the image, compact the session, or select a model with a larger image limit"
617        )));
618    }
619    Ok(())
620}
621
622/// Validate inline and legacy base64 images before constructing a provider body.
623/// Remote URLs carry no known payload size; the provider validates their content.
624/// Local references must be checked while resolving and never sent to a provider.
625pub fn validate_image_sizes(input: &[Message], limit: u64) -> Result<(), GenerateError> {
626    for part in input.iter().flat_map(Message::content) {
627        if let Content::Image { source } = part {
628            if source.starts_with("http://") || source.starts_with("https://") {
629                continue;
630            }
631            if source.starts_with("myco-image:") {
632                return Err(GenerateError::ExecutionError(
633                    "resolve local image references before calling a provider".into(),
634                ));
635            }
636            let payload = source
637                .split_once(',')
638                .map_or(source.as_str(), |(_, data)| data);
639            check_image_size(payload.len() as u64, limit)?;
640        }
641    }
642    Ok(())
643}
644
645/// Wire ids for every tool call in `input`: `out[i][j]` is the id a driver
646/// sends for the `j`-th tool_use of the assistant message at index `i`, and
647/// equally for the `j`-th result of the `ToolResults` message answering it.
648///
649/// History carries no tool ids — a call and its result pair positionally:
650/// `tool_use_results[j]` answers `tool_uses[j]` of the immediately preceding
651/// assistant message, the order the agent loop writes and compaction
652/// preserves (tails start at a user message and never split a pair).
653/// Providers, however, require an id on the wire to make that same pairing
654/// inside one request, and each has its own dialect ([`mint_tool_id`]) — so
655/// drivers mint one per position here. Provider-minted ids from responses are
656/// discarded at ingestion; storing and echoing them is what used to wedge a
657/// session the moment it resumed on a different provider.
658///
659/// Ids derive from (message index, ordinal), so history growth never changes
660/// the ids of earlier messages and the request prefix stays byte-identical
661/// across turns for provider prompt caching. A history whose result count
662/// disagrees with the preceding assistant's tool calls fails here, before any
663/// request is sent: guessing the pairing would corrupt the conversation
664/// silently.
665pub(crate) fn wire_tool_ids(input: &[Message]) -> Result<Vec<Vec<String>>, GenerateError> {
666    if input.iter().flat_map(Message::content).any(|part| {
667        matches!(part,
668        Content::Image { source } if source.starts_with("myco-image:"))
669    }) {
670        return Err(GenerateError::ExecutionError(
671            "resolve local image references before calling a provider".into(),
672        ));
673    }
674    let mut out: Vec<Vec<String>> = Vec::with_capacity(input.len());
675    for (i, message) in input.iter().enumerate() {
676        let ids = match message {
677            Message::UserMessage { .. } => Vec::new(),
678            Message::AssistantMessage { tool_uses, .. } => {
679                (0..tool_uses.len()).map(|j| mint_tool_id(i, j)).collect()
680            }
681            Message::ToolResults { tool_use_results } => {
682                let preceding_uses = match i.checked_sub(1).map(|p| &input[p]) {
683                    Some(Message::AssistantMessage { tool_uses, .. }) => tool_uses.len(),
684                    _ => 0,
685                };
686                if preceding_uses != tool_use_results.len() {
687                    return Err(GenerateError::ExecutionError(format!(
688                        "history is malformed: message {i} carries {} tool results but the \
689                         message before it has {preceding_uses} tool calls",
690                        tool_use_results.len()
691                    )));
692                }
693                out[i - 1].clone()
694            }
695        };
696        out.push(ids);
697    }
698    Ok(out)
699}
700
701/// Nine alphanumeric chars: `t`, then message index and tool ordinal as four
702/// base36 digits each — the intersection of every id dialect the drivers
703/// target:
704/// - Anthropic Messages: must match `^[a-zA-Z0-9_-]+$`
705/// - OpenAI: at most 40 chars
706/// - Mistral-style OpenAI-compatible backends: exactly nine alphanumerics
707///   (the binding constraint — it fixes both the length and the charset)
708fn mint_tool_id(message_index: usize, ordinal: usize) -> String {
709    const CAP: usize = 36 * 36 * 36 * 36;
710    // A history long enough to overflow four digits (1.6M messages) exceeds
711    // the request size cap long before it gets here.
712    assert!(message_index < CAP && ordinal < CAP);
713    let mut id = String::with_capacity(9);
714    id.push('t');
715    for n in [message_index, ordinal] {
716        for place in [36 * 36 * 36, 36 * 36, 36, 1] {
717            id.push(char::from_digit(((n / place) % 36) as u32, 36).unwrap());
718        }
719    }
720    id
721}
722
723/// One generation attempt emits parts, or ends with a failure.
724#[derive(Debug, Clone)]
725pub enum GenerationEvent {
726    Part(MessagePart),
727    Failure(GenerationFailure),
728}
729
730impl GenerationEvent {
731    pub fn into_result(self) -> Result<MessagePart, GenerateError> {
732        match self {
733            Self::Part(part) => Ok(part),
734            Self::Failure(failure) => Err(failure.cause),
735        }
736    }
737}
738
739#[derive(Debug, Clone)]
740pub struct GenerationFailure {
741    pub cause: GenerateError,
742    /// A transient cause. Retrying callers must discard the failed attempt's
743    /// provisional parts; only a validated response may execute tool calls.
744    pub retryable: bool,
745    pub retry_after: Option<std::time::Duration>,
746}
747
748impl GenerationFailure {
749    pub fn terminal(cause: GenerateError) -> Self {
750        Self {
751            cause,
752            retryable: false,
753            retry_after: None,
754        }
755    }
756
757    pub fn transient(cause: GenerateError, retry_after: Option<std::time::Duration>) -> Self {
758        Self {
759            cause,
760            retryable: true,
761            retry_after,
762        }
763    }
764}
765
766impl From<GenerateError> for GenerationFailure {
767    fn from(cause: GenerateError) -> Self {
768        Self::terminal(cause)
769    }
770}
771
772#[derive(Debug, Clone)]
773pub enum MessagePart {
774    MessageStart,
775    ContentStart(ContentStart),
776    ContentDelta(ContentDelta),
777    ToolUseStart(ToolUseStart),
778    ToolUseDelta(ToolUseDelta),
779    TurnEndReason(TurnEndReason),
780    /// Provider token usage for this generate call (may appear mid-stream or at end).
781    Usage(TokenUsage),
782}
783
784/// Token counts for one generate call. `cached_input_tokens` is a subset of
785/// `input_tokens`.
786#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
787pub struct TokenUsage {
788    #[serde(default)]
789    pub input_tokens: u64,
790    #[serde(default)]
791    pub output_tokens: u64,
792    #[serde(default)]
793    pub cached_input_tokens: u64,
794}
795
796impl TokenUsage {
797    /// Context occupied by the prompt = total input tokens (cached input is a
798    /// subset, already included).
799    pub fn context_tokens(self) -> u64 {
800        self.input_tokens
801    }
802
803    /// Fold a later usage report into this one, keeping known fields when the
804    /// later report omits them (providers split usage across stream events).
805    pub fn merge(self, next: TokenUsage) -> TokenUsage {
806        fn pick(prev: u64, next: u64) -> u64 {
807            if next != 0 { next } else { prev }
808        }
809        TokenUsage {
810            input_tokens: pick(self.input_tokens, next.input_tokens),
811            output_tokens: pick(self.output_tokens, next.output_tokens),
812            cached_input_tokens: pick(self.cached_input_tokens, next.cached_input_tokens),
813        }
814    }
815}
816
817#[derive(Debug, Clone)]
818pub enum ContentStart {
819    Text {
820        index: usize,
821    },
822    Image {
823        index: usize,
824    },
825    Thinking {
826        index: usize,
827        signature: Option<String>,
828        redacted: bool,
829    },
830}
831
832#[derive(Debug, Clone)]
833pub enum ContentDelta {
834    Text {
835        index: usize,
836        delta: String,
837    },
838    Image {
839        index: usize,
840        delta: String,
841    },
842    Thinking {
843        index: usize,
844        delta: String,
845    },
846    /// Appends to a thinking block's opaque provider signature (Anthropic
847    /// `signature_delta`); carries no visible text.
848    Signature {
849        index: usize,
850        delta: String,
851    },
852}
853
854#[derive(Debug, Clone)]
855pub struct ToolUseStart {
856    pub index: usize,
857    pub name: String,
858    /// Provider token to echo back with the result of this call. See
859    /// [`ToolUse::thought_signature`].
860    pub thought_signature: Option<String>,
861}
862
863#[derive(Debug, Clone)]
864pub struct ToolUseDelta {
865    pub index: usize,
866    pub input_json_delta: String,
867}
868
869//
870// GenerateOutput: accumulate a stream of MessageParts into a finished assistant turn
871//
872
873#[derive(Debug, Clone)]
874pub struct GenerateOutput {
875    pub content: Vec<Content>,
876    pub tool_uses: Vec<ToolUse>,
877    pub turn_end_reason: TurnEndReason,
878    /// Last usage observed on the stream, if the provider reported any.
879    pub usage: Option<TokenUsage>,
880}
881
882impl GenerateOutput {
883    /// Consume and validate one attempt without applying retry policy.
884    ///
885    /// Failures return their cause; consume [`GenerationEvent`] directly when
886    /// the transient classification or provider retry delay is needed.
887    pub async fn from_generation(
888        stream: impl Stream<Item = GenerationEvent>,
889    ) -> Result<Self, GenerateError> {
890        Self::from_stream(stream.map(GenerationEvent::into_result)).await
891    }
892
893    pub async fn from_stream(
894        stream: impl Stream<Item = Result<MessagePart, GenerateError>>,
895    ) -> Result<Self, GenerateError> {
896        let mut accumulator = MessageAccumulator::default();
897        let mut stream = pin!(stream);
898        while let Some(part) = stream.next().await {
899            accumulator.push(&part?)?;
900        }
901        accumulator.finish()
902    }
903}
904
905#[derive(thiserror::Error, Debug)]
906pub enum ModelCreationError {
907    #[error("Invalid configuration parameters supplied: {0}")]
908    BadConfig(String),
909
910    #[error("Uncategorized error occurred: {0}")]
911    Uncategorized(String),
912}
913
914#[cfg(test)]
915mod tests {
916    use super::*;
917    use crate::test_support::{assistant_tool, tool_results, user};
918
919    /// Every minted id must satisfy the strictest provider dialect at once:
920    /// exactly nine chars, alphanumeric only (covers Anthropic's
921    /// `^[a-zA-Z0-9_-]+$`, OpenAI's 40-char cap, and Mistral-style
922    /// nine-alphanumeric backends).
923    #[test]
924    fn wire_ids_are_nine_alphanumerics() {
925        for (i, j) in [(0, 0), (1, 3), (255, 7), (46_655, 12)] {
926            let id = mint_tool_id(i, j);
927            assert_eq!(id.len(), 9, "{id:?}");
928            assert!(id.chars().all(|c| c.is_ascii_alphanumeric()), "{id:?}");
929        }
930    }
931
932    /// A tool_result must carry the same wire id as the tool_use it answers.
933    #[test]
934    fn wire_ids_pair_results_with_uses_positionally() {
935        let input = [
936            user("hi"),
937            assistant_tool(None, "bash", serde_json::json!({})),
938            tool_results(&["ok"]),
939        ];
940        let ids = wire_tool_ids(&input).unwrap();
941        assert!(ids[0].is_empty());
942        assert_eq!(ids[1].len(), 1);
943        assert_eq!(ids[1], ids[2]);
944    }
945
946    /// Appending messages must not change earlier ids: the request prefix has
947    /// to stay byte-identical across turns or provider prompt caching breaks.
948    #[test]
949    fn wire_ids_stable_as_history_grows() {
950        let mut input = vec![
951            user("hi"),
952            assistant_tool(None, "bash", serde_json::json!({})),
953            tool_results(&["ok"]),
954        ];
955        let before = wire_tool_ids(&input).unwrap();
956        input.push(user("more"));
957        input.push(assistant_tool(None, "bash", serde_json::json!({})));
958        input.push(tool_results(&["ok"]));
959        let after = wire_tool_ids(&input).unwrap();
960        assert_eq!(before[..], after[..3]);
961        assert_ne!(after[1], after[4], "distinct calls must get distinct ids");
962    }
963
964    /// Positional pairing must refuse to guess: a results message whose count
965    /// disagrees with the preceding assistant's tool calls fails before any
966    /// request is sent, instead of silently mispairing.
967    #[test]
968    fn wire_ids_fail_loud_on_broken_pairing() {
969        let orphaned = [user("hi"), tool_results(&["ok"])];
970        assert!(wire_tool_ids(&orphaned).is_err());
971
972        let miscounted = [
973            assistant_tool(None, "bash", serde_json::json!({})),
974            tool_results(&["ok", "ok"]),
975        ];
976        assert!(wire_tool_ids(&miscounted).is_err());
977    }
978
979    #[tokio::test]
980    async fn accumulate_thinking_then_text() {
981        use futures::stream;
982
983        let parts = vec![
984            Ok(MessagePart::MessageStart),
985            Ok(MessagePart::ContentStart(ContentStart::Thinking {
986                index: 0,
987                signature: Some("sig".into()),
988                redacted: false,
989            })),
990            Ok(MessagePart::ContentDelta(ContentDelta::Thinking {
991                index: 0,
992                delta: "reason".into(),
993            })),
994            Ok(MessagePart::ContentStart(ContentStart::Text { index: 1 })),
995            Ok(MessagePart::ContentDelta(ContentDelta::Text {
996                index: 1,
997                delta: "answer".into(),
998            })),
999            Ok(MessagePart::TurnEndReason(TurnEndReason::EndTurn)),
1000        ];
1001        let output = GenerateOutput::from_stream(stream::iter(parts))
1002            .await
1003            .expect("accumulate");
1004        assert_eq!(output.content.len(), 2);
1005        match &output.content[0] {
1006            Content::Thinking {
1007                text,
1008                signature,
1009                redacted,
1010            } => {
1011                assert_eq!(text, "reason");
1012                assert_eq!(signature.as_deref(), Some("sig"));
1013                assert!(!*redacted);
1014            }
1015            other => panic!("expected thinking, got {other:?}"),
1016        }
1017        match &output.content[1] {
1018            Content::Text { text } => assert_eq!(text, "answer"),
1019            other => panic!("expected text, got {other:?}"),
1020        }
1021        assert_eq!(answer_content(&output.content).len(), 1);
1022    }
1023
1024    #[test]
1025    fn token_usage_merge_prefers_known_fields() {
1026        let start = TokenUsage {
1027            input_tokens: 2195,
1028            output_tokens: 1,
1029            cached_input_tokens: 2000,
1030        };
1031        let delta = TokenUsage {
1032            input_tokens: 0,
1033            output_tokens: 89,
1034            cached_input_tokens: 0,
1035        };
1036        let merged = start.merge(delta);
1037        assert_eq!(merged.input_tokens, 2195);
1038        assert_eq!(merged.output_tokens, 89);
1039        assert_eq!(merged.cached_input_tokens, 2000);
1040        assert_eq!(merged.context_tokens(), 2195);
1041    }
1042
1043    #[tokio::test]
1044    async fn accumulate_merges_split_usage() {
1045        use futures::stream;
1046
1047        let parts = vec![
1048            Ok(MessagePart::MessageStart),
1049            Ok(MessagePart::Usage(TokenUsage {
1050                input_tokens: 2195,
1051                output_tokens: 1,
1052                cached_input_tokens: 2000,
1053            })),
1054            Ok(MessagePart::ContentStart(ContentStart::Text { index: 0 })),
1055            Ok(MessagePart::ContentDelta(ContentDelta::Text {
1056                index: 0,
1057                delta: "hi".into(),
1058            })),
1059            Ok(MessagePart::Usage(TokenUsage {
1060                input_tokens: 0,
1061                output_tokens: 89,
1062                cached_input_tokens: 0,
1063            })),
1064            Ok(MessagePart::TurnEndReason(TurnEndReason::EndTurn)),
1065        ];
1066        let output = GenerateOutput::from_stream(stream::iter(parts))
1067            .await
1068            .expect("accumulate");
1069        let usage = output.usage.expect("usage present");
1070        assert_eq!(usage.input_tokens, 2195);
1071        assert_eq!(usage.output_tokens, 89);
1072        assert_eq!(usage.cached_input_tokens, 2000);
1073        assert_eq!(usage.context_tokens(), 2195);
1074    }
1075
1076    /// Backoff doubles from `initial_backoff`, stops at `max_backoff`, and a
1077    /// provider's `Retry-After` remains a minimum; the caller decides whether
1078    /// its budget permits waiting that long.
1079    #[test]
1080    fn retry_backoff_grows_caps_and_honours_retry_after() {
1081        use std::time::Duration;
1082        let policy = RetryPolicy {
1083            max_attempts: 6,
1084            initial_backoff: Duration::from_millis(100),
1085            max_backoff: Duration::from_millis(1000),
1086            backoff_multiplier: 2.0,
1087            max_elapsed: None,
1088        };
1089
1090        // Attempt 1 is the original send; it never waits.
1091        assert_eq!(policy.backoff(1, None), Duration::ZERO);
1092        assert_eq!(policy.backoff(2, None), Duration::from_millis(100));
1093        assert_eq!(policy.backoff(3, None), Duration::from_millis(200));
1094        assert_eq!(policy.backoff(4, None), Duration::from_millis(400));
1095        // Growth stops at the cap rather than running away.
1096        assert_eq!(policy.backoff(9, None), Duration::from_millis(1000));
1097
1098        // A longer Retry-After wins over the computed wait...
1099        assert_eq!(
1100            policy.backoff(2, Some(Duration::from_millis(500))),
1101            Duration::from_millis(500)
1102        );
1103        // A provider minimum is never shortened to the policy's wait cap.
1104        assert_eq!(
1105            policy.backoff(2, Some(Duration::from_secs(3600))),
1106            Duration::from_secs(3600)
1107        );
1108        // A shorter one does not shrink the backoff.
1109        assert_eq!(
1110            policy.backoff(3, Some(Duration::from_millis(1))),
1111            Duration::from_millis(200)
1112        );
1113    }
1114
1115    fn spec(key: &str, protocol: Protocol) -> ModelSpec {
1116        ModelSpec {
1117            key: key.into(),
1118            api_id: key.into(),
1119            protocol,
1120            thinking: ThinkingMode::default_for(protocol),
1121            context_window_tokens: 1_000_000,
1122            max_image_base64_bytes: 5 * 1024 * 1024,
1123            max_truncated_resumes: 3,
1124            auto_compact_at_tokens: None,
1125        }
1126    }
1127
1128    #[test]
1129    fn thinking_defaults_and_protocol_compatibility() {
1130        assert_eq!(
1131            ThinkingMode::default_for(Protocol::AnthropicMessages),
1132            ThinkingMode::Adaptive
1133        );
1134        assert_eq!(
1135            ThinkingMode::default_for(Protocol::OpenAIResponses),
1136            ThinkingMode::Effort
1137        );
1138        assert!(ThinkingMode::Budget.compatible_with(Protocol::AnthropicMessages));
1139        assert!(ThinkingMode::None.compatible_with(Protocol::AnthropicMessages));
1140        assert!(!ThinkingMode::Effort.compatible_with(Protocol::AnthropicMessages));
1141        assert!(ThinkingMode::None.compatible_with(Protocol::OpenAIResponses));
1142        // Both OpenAI dialects take the same effort-shaped thinking.
1143        assert_eq!(
1144            ThinkingMode::default_for(Protocol::OpenAICompletions),
1145            ThinkingMode::Effort
1146        );
1147        assert!(ThinkingMode::None.compatible_with(Protocol::OpenAICompletions));
1148        assert!(!ThinkingMode::Adaptive.compatible_with(Protocol::OpenAICompletions));
1149        assert!(!ThinkingMode::Adaptive.compatible_with(Protocol::OpenAIResponses));
1150        assert!(!ThinkingMode::Budget.compatible_with(Protocol::OpenAIResponses));
1151    }
1152
1153    #[test]
1154    fn protocol_serde_uses_config_strings() {
1155        assert_eq!(
1156            serde_json::to_value(Protocol::AnthropicMessages).unwrap(),
1157            serde_json::json!("anthropic-messages")
1158        );
1159        assert_eq!(
1160            serde_json::from_value::<Protocol>(serde_json::json!("openai-responses")).unwrap(),
1161            Protocol::OpenAIResponses
1162        );
1163        assert_eq!(
1164            serde_json::from_value::<Protocol>(serde_json::json!("openai-completions")).unwrap(),
1165            Protocol::OpenAICompletions
1166        );
1167    }
1168
1169    #[test]
1170    fn empty_catalog_get_says_no_models_configured() {
1171        let catalog = ModelCatalog::default();
1172        assert!(catalog.is_empty());
1173        let err = catalog.get("kimi-k3").unwrap_err();
1174        assert!(err.contains("no models configured"), "{err}");
1175        assert!(err.contains("[models]"), "{err}");
1176    }
1177
1178    #[test]
1179    fn catalog_get_unknown_key_lists_configured_models() {
1180        let entry = CatalogModel {
1181            spec: spec("opus", Protocol::AnthropicMessages),
1182            backend: BackendConfig::Anthropic(AnthropicBackendConfig::default()),
1183            auth_error: None,
1184        };
1185        let catalog = ModelCatalog::new([("opus".to_string(), entry)].into());
1186        let err = catalog.get("opsu").unwrap_err();
1187        assert!(err.contains("unknown model \"opsu\""), "{err}");
1188        assert!(err.contains("[opus]"), "{err}");
1189        assert!(catalog.get("opus").is_ok());
1190    }
1191
1192    #[test]
1193    fn catalog_get_reports_deferred_auth_error() {
1194        let entry = CatalogModel {
1195            spec: spec("kimi", Protocol::OpenAIResponses),
1196            backend: BackendConfig::OpenAIResponses(OpenAIBackendConfig::default()),
1197            auth_error: Some("model `kimi`: auth env:OPENROUTER_API_KEY is unset".into()),
1198        };
1199        let catalog = ModelCatalog::new([("kimi".to_string(), entry)].into());
1200        let err = catalog.get("kimi").unwrap_err();
1201        assert!(err.contains("OPENROUTER_API_KEY"), "{err}");
1202    }
1203
1204    #[test]
1205    fn new_rejects_protocol_mismatch() {
1206        let result = new(GenerativeModelConfig {
1207            model: spec("grok", Protocol::OpenAIResponses),
1208            tools: vec![],
1209            system_prompt: String::new(),
1210            backend_config: BackendConfig::Anthropic(AnthropicBackendConfig {
1211                anthropic_auth_token: "dummy".into(),
1212                ..Default::default()
1213            }),
1214        });
1215        let err = match result {
1216            Ok(_) => panic!("expected mismatch"),
1217            Err(e) => e,
1218        };
1219        assert!(err.to_string().contains("speaks openai-responses"), "{err}");
1220    }
1221
1222    /// The preflight is the whole point of the rewind path: a request that is
1223    /// too big must be refused *before* upload, and must say so in a way the
1224    /// top level can act on.
1225    #[test]
1226    fn oversized_request_is_refused_before_upload() {
1227        for limit in [1024, MAX_REQUEST_BYTES, 60_000_000] {
1228            assert!(check_request_size(limit, limit, "Anthropic").is_ok());
1229            let err = check_request_size(limit + 1, limit, "Anthropic").unwrap_err();
1230            assert!(
1231                matches!(err, GenerateError::RequestTooLargeError(_)),
1232                "{err:?}"
1233            );
1234            assert_eq!(err.recovery(), Recovery::OmitLastMessage);
1235            assert!(
1236                err.to_string().contains(&format!("limit is {limit} bytes")),
1237                "{err}"
1238            );
1239        }
1240    }
1241
1242    #[test]
1243    fn older_backend_configs_default_to_a_thirty_megabyte_request_cap() {
1244        let mut anthropic = serde_json::to_value(AnthropicBackendConfig::default()).unwrap();
1245        let mut openai = serde_json::to_value(OpenAIBackendConfig::default()).unwrap();
1246        for config in [&mut anthropic, &mut openai] {
1247            assert_eq!(config["max_request_bytes"], 30_000_000);
1248            config.as_object_mut().unwrap().remove("max_request_bytes");
1249        }
1250        assert_eq!(
1251            serde_json::from_value::<AnthropicBackendConfig>(anthropic)
1252                .unwrap()
1253                .max_request_bytes,
1254            30_000_000
1255        );
1256        assert_eq!(
1257            serde_json::from_value::<OpenAIBackendConfig>(openai)
1258                .unwrap()
1259                .max_request_bytes,
1260            30_000_000
1261        );
1262    }
1263
1264    /// A provider that rejects the size itself lands on the same variant, so
1265    /// the caller rewinds whether the cap was caught locally or remotely.
1266    #[test]
1267    fn provider_size_rejections_map_to_the_same_recovery() {
1268        let too_large = http_error(
1269            reqwest::StatusCode::PAYLOAD_TOO_LARGE,
1270            "HTTP 413: too big".into(),
1271        );
1272        assert_eq!(too_large.recovery(), Recovery::OmitLastMessage);
1273
1274        // Anthropic reports it as a 400 whose body names the error type.
1275        let named = http_error(
1276            reqwest::StatusCode::BAD_REQUEST,
1277            r#"HTTP 400: {"error":{"type":"request_too_large"}}"#.into(),
1278        );
1279        assert_eq!(named.recovery(), Recovery::OmitLastMessage);
1280
1281        // Others name no type and only describe the size in prose. Read as a
1282        // generic failure this is resent unchanged on every later turn.
1283        let described = http_error(
1284            reqwest::StatusCode::BAD_REQUEST,
1285            r#"HTTP 400: {"error":{"code":400,"message":"The message size (31271377 bytes) \
1286               exceeds 30.000MB limit.","status":"FAILED_PRECONDITION"}}"#
1287                .into(),
1288        );
1289        assert_eq!(described.recovery(), Recovery::OmitLastMessage);
1290
1291        let extent = http_error(
1292            reqwest::StatusCode::BAD_REQUEST,
1293            r#"{"type":"error","error":{"type":"invalid_request_error","message":"messages.9.content.13.image.source.base64.data: At least one of the image dimensions exceed max allowed size for many-image requests: 2576 pixels"}}"#.into(),
1294        );
1295        assert_eq!(extent.recovery(), Recovery::OmitLastMessage);
1296
1297        let unrelated = http_error(
1298            reqwest::StatusCode::INTERNAL_SERVER_ERROR,
1299            "HTTP 500: overloaded".into(),
1300        );
1301        assert_eq!(unrelated.recovery(), Recovery::Retry);
1302    }
1303
1304    /// Reading the body is how a size rejection is recognized, so an ordinary
1305    /// failure must not be mistaken for one: rewinding drops a user message,
1306    /// which is destructive when the request was never too big.
1307    #[test]
1308    fn ordinary_failures_are_not_read_as_size_rejections() {
1309        for body in [
1310            r#"HTTP 400: {"error":{"message":"max_tokens exceeds the model's limit"}}"#,
1311            r#"HTTP 400: {"error":{"message":"messages: unexpected role"}}"#,
1312            r#"HTTP 401: {"error":{"message":"invalid x-api-key"}}"#,
1313        ] {
1314            let err = http_error(reqwest::StatusCode::BAD_REQUEST, body.into());
1315            assert_eq!(err.recovery(), Recovery::Retry, "{body}");
1316        }
1317    }
1318
1319    #[test]
1320    fn message_types_serde_roundtrip() {
1321        let messages = vec![
1322            Message::UserMessage {
1323                content: vec![
1324                    Content::Text { text: "hi".into() },
1325                    Content::Image {
1326                        source: "data".into(),
1327                    },
1328                ],
1329            },
1330            Message::AssistantMessage {
1331                content: vec![Content::Text { text: "ok".into() }],
1332                tool_uses: vec![ToolUse {
1333                    name: "bash".into(),
1334                    input: serde_json::json!({"command": "true"}),
1335                    thought_signature: None,
1336                }],
1337                turn_end_reason: Some(TurnEndReason::ToolUse),
1338            },
1339            Message::ToolResults {
1340                tool_use_results: vec![ToolResult {
1341                    content: vec![Content::Text {
1342                        text: "done".into(),
1343                    }],
1344                    is_error: false,
1345                    status: None,
1346                    resource: None,
1347                }],
1348            },
1349            Message::AssistantMessage {
1350                content: vec![],
1351                tool_uses: vec![],
1352                turn_end_reason: Some(TurnEndReason::Other("Anthropic::PauseTurn".into())),
1353            },
1354        ];
1355        let json = serde_json::to_string(&messages).expect("serialize");
1356        let back: Vec<Message> = serde_json::from_str(&json).expect("deserialize");
1357        assert_eq!(
1358            serde_json::to_value(&back).unwrap(),
1359            serde_json::to_value(&messages).unwrap()
1360        );
1361    }
1362}
1363
1364/// What a caller can do about a failed turn.
1365///
1366/// Failures that are a property of the *history* cannot be fixed by trying
1367/// again — every later turn resends that history and fails the same way, which
1368/// wedges the session. This is the top-level signal for those: it says whether
1369/// the last user message has to come back out (see
1370/// the chat adapter's rewind operation) before the conversation
1371/// can continue.
1372#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1373pub enum Recovery {
1374    /// Nothing about the history is known to be at fault; resubmitting as-is
1375    /// may work (provider blip, refusal, malformed stream).
1376    Retry,
1377    /// The request is too big for the provider. Retrying unchanged fails
1378    /// identically — drop the last user message (typically the one carrying an
1379    /// oversized attachment) and the session can go on.
1380    OmitLastMessage,
1381    /// An execution contract was violated. Resolve it before issuing more work.
1382    Stop,
1383}
1384
1385/// Default ceiling on one serialized API request body: 30 MB (decimal).
1386///
1387/// Each backend can override this with `max_request_bytes`. The check includes
1388/// images accumulated in history and JSON overhead, and refuses oversized bodies
1389/// before upload with [`Recovery::OmitLastMessage`].
1390pub const MAX_REQUEST_BYTES: usize = 30_000_000;
1391
1392fn default_max_request_bytes() -> usize {
1393    MAX_REQUEST_BYTES
1394}
1395
1396#[derive(thiserror::Error, Debug, Clone)]
1397pub enum GenerateError {
1398    #[error("Something went wrong while generating a response: {0}")]
1399    ExecutionError(String),
1400
1401    #[error("Generation succeeded, but the model refused to comply: {0}")]
1402    RefusalError(String),
1403
1404    #[error("Generation succeeded, but the output was malformed or corrupted: {0}")]
1405    MalformedResponseError(String),
1406
1407    /// Request exceeds the provider's size limit — locally detected, or the
1408    /// provider's own 413. Not retryable without shrinking the history.
1409    #[error("The request is too large to send: {0}")]
1410    RequestTooLargeError(String),
1411}
1412
1413impl GenerateError {
1414    pub fn recovery(&self) -> Recovery {
1415        match self {
1416            GenerateError::RequestTooLargeError(_) => Recovery::OmitLastMessage,
1417            GenerateError::ExecutionError(_)
1418            | GenerateError::RefusalError(_)
1419            | GenerateError::MalformedResponseError(_) => Recovery::Retry,
1420        }
1421    }
1422}
1423
1424/// Refuse a composed request body over the configured endpoint limit.
1425///
1426/// Takes the already-serialized length so the checked size is exactly the size
1427/// uploaded — no second serialization pass over a multi-megabyte body.
1428pub(crate) fn check_request_size(
1429    len: usize,
1430    limit: usize,
1431    provider: &str,
1432) -> Result<(), GenerateError> {
1433    if len > limit {
1434        return Err(GenerateError::RequestTooLargeError(format!(
1435            "the {provider} request is {len} bytes; the configured limit is {limit} bytes \
1436             (max_request_bytes). Images accumulate in the conversation; reduce \
1437             attachments or compact the session before retrying. Raise the gateway's \
1438             max_request_bytes only if it supports larger requests",
1439        )));
1440    }
1441    Ok(())
1442}
1443
1444/// Map a provider HTTP status to the right error variant: a size rejection is
1445/// not a generic failure.
1446///
1447/// 413 is unambiguous. Everything else has to be read out of the body, because
1448/// providers disagree on how they report an oversized request: some name a
1449/// machine-readable type (`request_too_large`), others return a 400 whose body
1450/// only describes the size in prose. Getting this wrong is expensive — the
1451/// identical body is resent on every later turn, so a size failure classified
1452/// as [`Recovery::Retry`] never triggers the rewind and the session fails
1453/// forever.
1454pub(crate) fn http_error(status: reqwest::StatusCode, message: String) -> GenerateError {
1455    if status == reqwest::StatusCode::PAYLOAD_TOO_LARGE || describes_a_size_rejection(&message) {
1456        GenerateError::RequestTooLargeError(message)
1457    } else {
1458        GenerateError::ExecutionError(message)
1459    }
1460}
1461
1462/// Does this provider error body say the request was too big?
1463///
1464/// Matching prose is unavoidable, so it is kept to phrasings only a size
1465/// rejection produces: "too large" however the provider spells it, or a size
1466/// that "exceeds" a stated limit.
1467fn describes_a_size_rejection(message: &str) -> bool {
1468    let message = message.to_ascii_lowercase();
1469    message.contains("too_large")
1470        || message.contains("too large")
1471        || (message.contains("size") && message.contains("exceeds"))
1472        || (message.contains("image dimensions")
1473            && message.contains("exceed")
1474            && message.contains("max allowed size"))
1475}
myco_model/lib.rs

myco_model/
lib.rs
