hotl_provider/
lib.rs

1//! L2 — the provider seam.
2//!
3//! `stream()` is the one required method. Events carry block structure
4//! (review Arch #6): every delta names its block index, and the provider —
5//! the only layer that understands its own wire format — assembles the final
6//! verbatim assistant blocks and hands them over in `Completed`. The engine
7//! never demuxes provider wire formats.
8//!
9//! Native (Send) variants are authoritative for M0; the `?Send` browser twins
10//! are derived at the gated milestone (rust-implementation §Key trait signatures).
11
12use std::collections::VecDeque;
13use std::sync::{Arc, Mutex};
14
15use futures_util::stream::BoxStream;
16use hotl_types::{Item, StopReason, TokenUsage};
17use serde::{Deserialize, Serialize};
18use serde_json::Value;
19
20#[derive(Debug, Clone, Serialize, Deserialize)]
21pub struct ToolDef {
22    pub name: String,
23    pub description: String,
24    pub input_schema: Value,
25}
26
27#[derive(Debug, Clone)]
28pub struct SamplingRequest {
29    pub model: String,
30    pub max_tokens: u32,
31    /// Byte-stable owner system prompt (L6 discipline).
32    pub system: Arc<str>,
33    pub items: Arc<Vec<Item>>,
34    pub tools: Arc<[ToolDef]>,
35    /// Adaptive thinking on models that support it.
36    pub thinking: bool,
37    /// M0 static cache placement: system block + latest user block
38    /// (explicit-cache providers).
39    pub cache_static: bool,
40    /// MOIM (M2): ephemeral per-turn context, sent as a trailing user block
41    /// after the cache marker. Never persisted — it exists only on the wire.
42    pub turn_context: Option<String>,
43}
44
45/// The unified, channel-tagged, block-structured event enum.
46#[derive(Debug, Clone, Serialize, Deserialize)]
47#[serde(tag = "event", rename_all = "snake_case")]
48pub enum StreamEvent {
49    Started,
50    BlockStart {
51        index: usize,
52        kind: String,
53    },
54    TextDelta {
55        index: usize,
56        text: String,
57    },
58    ThinkingDelta {
59        index: usize,
60        text: String,
61    },
62    ToolInputDelta {
63        index: usize,
64        json: String,
65    },
66    BlockEnd {
67        index: usize,
68    },
69    Retrying {
70        attempt: u32,
71        reason: String,
72    },
73    /// Terminal event: the provider-assembled verbatim assistant blocks
74    /// (echo these back on the next request — replay-safe by construction).
75    Completed {
76        stop: StopReason,
77        usage: TokenUsage,
78        blocks: Vec<Value>,
79    },
80}
81
82#[derive(Debug, thiserror::Error)]
83pub enum ProviderError {
84    #[error("authentication failed: {0}")]
85    Auth(String),
86    #[error("HTTP {status}: {message}")]
87    Http {
88        status: u16,
89        message: String,
90        retry_after: Option<u64>,
91    },
92    #[error("transport error: {0}")]
93    Transport(String),
94    #[error("stream parse error: {0}")]
95    Parse(String),
96}
97
98pub trait Provider: Send + Sync {
99    fn stream(
100        &self,
101        req: SamplingRequest,
102    ) -> BoxStream<'static, Result<StreamEvent, ProviderError>>;
103}
104
105/// The honest "second impl" (D9): a scripted provider driving the real engine
106/// in tests. Each `stream()` call pops the next script.
107pub struct ScriptedProvider {
108    scripts: Mutex<VecDeque<Vec<Result<StreamEvent, ProviderError>>>>,
109    /// Every request the engine made, for test assertions on what the model saw.
110    requests: Mutex<Vec<SamplingRequest>>,
111}
112
113impl ScriptedProvider {
114    pub fn new(scripts: Vec<Vec<Result<StreamEvent, ProviderError>>>) -> Self {
115        Self {
116            scripts: Mutex::new(scripts.into()),
117            requests: Mutex::new(Vec::new()),
118        }
119    }
120
121    /// Every captured request. Cheap since a request's history/tools are
122    /// shared (`Arc`) — the clone copies pointers, not items.
123    pub fn requests(&self) -> Vec<SamplingRequest> {
124        self.requests.lock().expect("requests mutex").clone()
125    }
126
127    /// The most recent request, if any.
128    pub fn last_request(&self) -> Option<SamplingRequest> {
129        self.requests
130            .lock()
131            .expect("requests mutex")
132            .last()
133            .cloned()
134    }
135
136    pub fn request_count(&self) -> usize {
137        self.requests.lock().expect("requests mutex").len()
138    }
139
140    /// Append a script after construction (tests that need the harness's
141    /// paths before the scripts can be written).
142    pub fn push_script(&self, script: Vec<Result<StreamEvent, ProviderError>>) {
143        self.scripts
144            .lock()
145            .expect("scripted provider mutex")
146            .push_back(script);
147    }
148
149    /// Convenience: a one-sample script that answers with plain text.
150    pub fn text_reply(text: &str) -> Vec<Result<StreamEvent, ProviderError>> {
151        vec![
152            Ok(StreamEvent::Started),
153            Ok(StreamEvent::BlockStart {
154                index: 0,
155                kind: "text".into(),
156            }),
157            Ok(StreamEvent::TextDelta {
158                index: 0,
159                text: text.into(),
160            }),
161            Ok(StreamEvent::BlockEnd { index: 0 }),
162            Ok(StreamEvent::Completed {
163                stop: StopReason::EndTurn,
164                usage: TokenUsage {
165                    input_tokens: 10,
166                    output_tokens: 5,
167                    ..Default::default()
168                },
169                blocks: vec![serde_json::json!({"type": "text", "text": text})],
170            }),
171        ]
172    }
173
174    /// Convenience: a sample that calls one tool.
175    pub fn tool_call(
176        id: &str,
177        name: &str,
178        input: Value,
179    ) -> Vec<Result<StreamEvent, ProviderError>> {
180        let block = serde_json::json!({"type": "tool_use", "id": id, "name": name, "input": input});
181        vec![
182            Ok(StreamEvent::Started),
183            Ok(StreamEvent::BlockStart {
184                index: 0,
185                kind: "tool_use".into(),
186            }),
187            Ok(StreamEvent::BlockEnd { index: 0 }),
188            Ok(StreamEvent::Completed {
189                stop: StopReason::ToolUse,
190                usage: TokenUsage {
191                    input_tokens: 10,
192                    output_tokens: 8,
193                    ..Default::default()
194                },
195                blocks: vec![block],
196            }),
197        ]
198    }
199}
200
201impl Provider for ScriptedProvider {
202    fn stream(
203        &self,
204        req: SamplingRequest,
205    ) -> BoxStream<'static, Result<StreamEvent, ProviderError>> {
206        self.requests.lock().expect("requests mutex").push(req);
207        let script = self
208            .scripts
209            .lock()
210            .expect("scripted provider mutex")
211            .pop_front()
212            .unwrap_or_else(|| {
213                vec![Err(ProviderError::Transport(
214                    "scripted provider exhausted".into(),
215                ))]
216            });
217        Box::pin(futures_util::stream::iter(script))
218    }
219}
220
221/// Normalize a configured base URL to one ending in `/v1`.
222///
223/// Two spellings are in the wild and users copy whichever their endpoint's
224/// docs show: hotl's own convention (and the OpenAI provider's) puts the
225/// version in the base, while bridges document the bare origin because
226/// official SDKs append the whole versioned path themselves. Everything that
227/// builds a URL from a configured base goes through here, so the provider and
228/// `hotl doctor` can never disagree about where an endpoint lives.
229pub fn v1_base(base: &str) -> String {
230    let base = base.trim_end_matches('/');
231    if base.ends_with("/v1") {
232        base.to_string()
233    } else {
234        format!("{base}/v1")
235    }
236}
237
238#[cfg(test)]
239mod base_url_tests {
240    use super::v1_base;
241
242    #[test]
243    fn both_spellings_and_trailing_slashes_resolve_alike() {
244        for input in [
245            "http://127.0.0.1:3456",
246            "http://127.0.0.1:3456/",
247            "http://127.0.0.1:3456/v1",
248            "http://127.0.0.1:3456/v1/",
249        ] {
250            assert_eq!(v1_base(input), "http://127.0.0.1:3456/v1", "input: {input}");
251        }
252    }
253}
254
255/// SSE line parsing shared by HTTP providers: turns raw byte chunks into
256/// complete `data:` payload strings. Chunks can split mid-line (and mid-UTF-8
257/// code point), so bytes are buffered, not lossily decoded per chunk.
258#[derive(Default)]
259pub struct SseParser {
260    buf: Vec<u8>,
261}
262
263/// A stream that never sends a newline must not buffer without bound.
264const SSE_MAX_BUFFER: usize = 1024 * 1024;
265
266impl SseParser {
267    /// Feed a chunk; returns complete `data:` payloads (`[DONE]` filtered).
268    /// Errors when a single line exceeds [`SSE_MAX_BUFFER`].
269    pub fn feed(&mut self, chunk: &[u8]) -> Result<Vec<String>, ProviderError> {
270        self.buf.extend_from_slice(chunk);
271        let mut out = Vec::new();
272        let mut start = 0;
273        while let Some(pos) = self.buf[start..].iter().position(|&b| b == b'\n') {
274            let line = String::from_utf8_lossy(&self.buf[start..start + pos]);
275            let line = line.trim_end_matches('\r');
276            if let Some(data) = line.strip_prefix("data:") {
277                let data = data.trim_start();
278                if !data.is_empty() && data != "[DONE]" {
279                    out.push(data.to_string());
280                }
281            }
282            start += pos + 1;
283        }
284        self.buf.drain(..start);
285        if self.buf.len() > SSE_MAX_BUFFER {
286            return Err(ProviderError::Parse(format!(
287                "SSE line exceeded {SSE_MAX_BUFFER} bytes without a newline"
288            )));
289        }
290        Ok(out)
291    }
292}
293
294/// Pure-data retry classification (RELIABILITY.md — never regex
295/// on prose). Both HTTP providers consult this; budgets reset per sample.
296pub mod retry {
297    use super::ProviderError;
298
299    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
300    pub enum Decision {
301        /// Wait this many seconds, then retry the request.
302        Retry { after_secs: u64 },
303        /// Not recoverable by retrying (auth, parse, client errors).
304        Fatal,
305    }
306
307    pub const MAX_ATTEMPTS: u32 = 3;
308
309    /// `attempt` is 1-based (the attempt that just failed).
310    pub fn classify(err: &ProviderError, attempt: u32) -> Decision {
311        if attempt >= MAX_ATTEMPTS {
312            return Decision::Fatal;
313        }
314        match err {
315            ProviderError::Http {
316                status,
317                retry_after,
318                ..
319            } if *status == 429 || *status >= 500 => Decision::Retry {
320                after_secs: retry_after.unwrap_or(1u64 << (attempt - 1)),
321            },
322            ProviderError::Transport(_) => Decision::Retry {
323                after_secs: 1u64 << (attempt - 1),
324            },
325            _ => Decision::Fatal,
326        }
327    }
328
329    /// Availability-class errors are the only ones that justify falling back
330    /// to another model (never auth/billing/parse).
331    pub fn is_availability(err: &ProviderError) -> bool {
332        matches!(
333            err,
334            ProviderError::Http { status, .. } if *status == 429 || *status >= 500
335        ) || matches!(err, ProviderError::Transport(_))
336    }
337
338    /// Context-overflow detection (M2 compaction trigger). Both dialects
339    /// report overflow as a 400 whose message names the context/length limit;
340    /// this matches on *wire error* text (structured API data, not model
341    /// prose — the RELIABILITY.md rule concerns the latter). A miss is safe:
342    /// the pre-sample threshold catches what this doesn't.
343    pub fn is_context_overflow(err: &ProviderError) -> bool {
344        let ProviderError::Http {
345            status: 400,
346            message,
347            ..
348        } = err
349        else {
350            return false;
351        };
352        let m = message.to_lowercase();
353        [
354            "too long",
355            "context length",
356            "context window",
357            "tokens exceed",
358        ]
359        .iter()
360        .any(|needle| m.contains(needle))
361    }
362
363    #[cfg(test)]
364    mod tests {
365        use super::*;
366
367        #[test]
368        fn overflow_detection() {
369            let overflow = ProviderError::Http {
370                status: 400,
371                message:
372                    r#"{"error":{"message":"prompt is too long: 210000 tokens > 200000 maximum"}}"#
373                        .into(),
374                retry_after: None,
375            };
376            assert!(is_context_overflow(&overflow));
377            let oai = ProviderError::Http {
378                status: 400,
379                message: "This model's maximum context length is 128000 tokens".into(),
380                retry_after: None,
381            };
382            assert!(is_context_overflow(&oai));
383            let plain_400 = ProviderError::Http {
384                status: 400,
385                message: "bad schema".into(),
386                retry_after: None,
387            };
388            assert!(!is_context_overflow(&plain_400));
389        }
390
391        #[test]
392        fn classify_rules() {
393            let overload = ProviderError::Http {
394                status: 529,
395                message: String::new(),
396                retry_after: Some(7),
397            };
398            assert_eq!(classify(&overload, 1), Decision::Retry { after_secs: 7 });
399            assert_eq!(classify(&overload, MAX_ATTEMPTS), Decision::Fatal);
400            let auth = ProviderError::Auth("bad".into());
401            assert_eq!(classify(&auth, 1), Decision::Fatal);
402            assert!(!is_availability(&auth));
403            let transport = ProviderError::Transport("reset".into());
404            assert_eq!(classify(&transport, 2), Decision::Retry { after_secs: 2 });
405            assert!(is_availability(&transport));
406            let bad_req = ProviderError::Http {
407                status: 400,
408                message: String::new(),
409                retry_after: None,
410            };
411            assert_eq!(classify(&bad_req, 1), Decision::Fatal);
412        }
413    }
414}
415
416/// The named cross-provider canonicalization stage (`transform_messages`).
417/// Canonical assistant blocks are
418/// Anthropic-shaped; when a request targets a *different* provider than the
419/// one that produced a block, provider-bound reasoning must not cross.
420pub mod transform {
421    use serde_json::Value;
422
423    /// Drop blocks that are provider-bound (signed/redacted thinking) when
424    /// sending history to a foreign dialect. Text and tool_use always pass.
425    pub fn strip_foreign_reasoning(blocks: &[Value]) -> Vec<Value> {
426        blocks
427            .iter()
428            .filter(|b| {
429                !matches!(
430                    b.get("type").and_then(Value::as_str),
431                    Some("thinking") | Some("redacted_thinking")
432                )
433            })
434            .cloned()
435            .collect()
436    }
437
438    #[cfg(test)]
439    mod tests {
440        use super::*;
441        use serde_json::json;
442
443        #[test]
444        fn strips_thinking_keeps_rest() {
445            let blocks = vec![
446                json!({"type":"thinking","thinking":"x","signature":"s"}),
447                json!({"type":"redacted_thinking","data":"d"}),
448                json!({"type":"text","text":"hi"}),
449                json!({"type":"tool_use","id":"1","name":"read","input":{}}),
450            ];
451            let out = strip_foreign_reasoning(&blocks);
452            assert_eq!(out.len(), 2);
453            assert_eq!(out[0]["type"], "text");
454            assert_eq!(out[1]["type"], "tool_use");
455        }
456    }
457}
458
459/// Arg healing at the erasure boundary (M3a): streamed tool
460/// arguments sometimes arrive as *almost*-JSON. Repair is conservative —
461/// only unambiguous damage is fixed; anything else stays a parse error that
462/// feeds back to the model as a tool result.
463pub mod repair {
464    use serde_json::Value;
465
466    /// Strict parse, then repairs: strip trailing commas, then close
467    /// truncated strings/objects/arrays (streams cut mid-argument).
468    pub fn parse_or_repair(raw: &str) -> Option<Value> {
469        if let Ok(v) = serde_json::from_str(raw) {
470            return Some(v);
471        }
472        let without_commas = strip_trailing_commas(raw);
473        if let Ok(v) = serde_json::from_str(&without_commas) {
474            return Some(v);
475        }
476        serde_json::from_str(&close_truncation(&without_commas)).ok()
477    }
478
479    /// `{"a": 1,}` / `[1, 2,]` → valid. Only commas directly before a closer.
480    fn strip_trailing_commas(s: &str) -> String {
481        let mut out = String::with_capacity(s.len());
482        let mut in_string = false;
483        let mut escaped = false;
484        for c in s.chars() {
485            if in_string {
486                out.push(c);
487                if escaped {
488                    escaped = false;
489                } else if c == '\\' {
490                    escaped = true;
491                } else if c == '"' {
492                    in_string = false;
493                }
494                continue;
495            }
496            match c {
497                '"' => {
498                    in_string = true;
499                    out.push(c);
500                }
501                '}' | ']' => {
502                    while out.ends_with(char::is_whitespace) || out.ends_with(',') {
503                        if out.ends_with(',') {
504                            out.pop();
505                            break;
506                        }
507                        out.pop();
508                    }
509                    out.push(c);
510                }
511                _ => out.push(c),
512            }
513        }
514        out
515    }
516
517    /// Close an unterminated string and any open brackets, in nesting order.
518    fn close_truncation(s: &str) -> String {
519        let mut stack = Vec::new();
520        let mut in_string = false;
521        let mut escaped = false;
522        for c in s.chars() {
523            if in_string {
524                if escaped {
525                    escaped = false;
526                } else if c == '\\' {
527                    escaped = true;
528                } else if c == '"' {
529                    in_string = false;
530                }
531                continue;
532            }
533            match c {
534                '"' => in_string = true,
535                '{' => stack.push('}'),
536                '[' => stack.push(']'),
537                '}' | ']' => {
538                    stack.pop();
539                }
540                _ => {}
541            }
542        }
543        let mut out = s.to_string();
544        if in_string {
545            out.push('"');
546        }
547        while let Some(closer) = stack.pop() {
548            out.push(closer);
549        }
550        out
551    }
552
553    #[cfg(test)]
554    mod tests {
555        use super::*;
556
557        #[test]
558        fn repairs_common_damage_and_rejects_garbage() {
559            assert_eq!(
560                parse_or_repair(r#"{"path": "a.rs"}"#).unwrap()["path"],
561                "a.rs"
562            );
563            assert_eq!(
564                parse_or_repair(r#"{"path": "a.rs",}"#).unwrap()["path"],
565                "a.rs"
566            );
567            assert_eq!(
568                parse_or_repair(r#"{"items": [1, 2,]}"#).unwrap()["items"][1],
569                2
570            );
571            // Truncated mid-string (stream cut): closed and parsed.
572            let v = parse_or_repair(r#"{"command": "cargo tes"#).unwrap();
573            assert_eq!(v["command"], "cargo tes");
574            // A comma inside a string is untouched.
575            assert_eq!(parse_or_repair(r#"{"t": "a,}"}"#).unwrap()["t"], "a,}");
576            // Escaped quotes don't confuse the scanner.
577            let v = parse_or_repair(r#"{"t": "say \"hi\"",}"#).unwrap();
578            assert_eq!(v["t"], "say \"hi\"");
579            assert!(parse_or_repair("not json at all").is_none());
580        }
581    }
582}
583
584pub mod key;
585
586/// Wire-format folding, implemented per provider: turn one SSE `data:`
587/// payload into events, and produce the terminal `Completed` at end-of-stream.
588pub trait SseAssembler {
589    fn handle(&mut self, data: &str) -> Result<Vec<StreamEvent>, ProviderError>;
590    fn finish(self) -> Result<StreamEvent, ProviderError>;
591}
592
593/// Drive an SSE byte stream through the line parser and an assembler.
594/// Shared by every HTTP provider; wasm-clean (no HTTP client dependency).
595pub fn drive_sse<B, E, A>(
596    bytes: B,
597    mut assembler: A,
598) -> impl futures_util::Stream<Item = Result<StreamEvent, ProviderError>>
599where
600    B: futures_util::Stream<Item = Result<bytes::Bytes, E>>,
601    E: std::fmt::Display,
602    A: SseAssembler,
603{
604    async_stream::stream! {
605        let mut parser = SseParser::default();
606        futures_util::pin_mut!(bytes);
607        use futures_util::StreamExt;
608        while let Some(chunk) = bytes.next().await {
609            let chunk = match chunk {
610                Ok(c) => c,
611                Err(e) => {
612                    yield Err(ProviderError::Transport(format!("stream interrupted: {e}")));
613                    return;
614                }
615            };
616            let payloads = match parser.feed(&chunk) {
617                Ok(payloads) => payloads,
618                Err(e) => { yield Err(e); return; }
619            };
620            for data in payloads {
621                match assembler.handle(&data) {
622                    Ok(events) => for ev in events { yield Ok(ev); },
623                    Err(e) => { yield Err(e); return; }
624                }
625            }
626        }
627        yield assembler.finish();
628    }
629}
hotl_provider/lib.rs

hotl_provider/
lib.rs
