sim_lib_agent_runner_http/
runner.rs

1use crate::ProviderAuth;
2use crate::client::{HttpRunnerRequest, post_json, post_json_stream};
3use crate::config::ProviderConfig;
4use crate::redact::redact_text;
5use crate::stream::HttpStreamDecoder;
6use sim_codec_chat::{
7    AnthropicRequestOptions, LemonadeRequestOptions, LmStudioRequestOptions, OllamaRequestOptions,
8    OpenAiRequestOptions, decode_anthropic_response, decode_anthropic_stream,
9    decode_lemonade_response, decode_lemonade_stream, decode_lm_studio_response,
10    decode_lm_studio_stream, decode_ollama_response, decode_ollama_stream, decode_openai_response,
11    encode_anthropic_request, encode_lemonade_request, encode_lm_studio_request,
12    encode_ollama_request, encode_openai_request, model_error_expr,
13};
14use sim_kernel::{
15    CapabilityName, Cx, Datum, DatumStore, Effect, Error, Expr, Ref, Result, Symbol, core_any_ref,
16    effect, value_from_ref,
17};
18use sim_lib_agent_runner_core::{
19    ModelCard, ModelEvent, ModelEventSink, ModelRequest, ModelResponse, ModelRunner,
20};
21use std::time::Duration;
22
23/// HTTP-backed [`ModelRunner`] for OpenAI-compatible and Ollama endpoints.
24#[derive(Clone, Debug)]
25pub struct HttpRunner {
26    runner: Symbol,
27    model: String,
28    provider: Symbol,
29    locality: Symbol,
30    runner_label: &'static str,
31    request_path: &'static str,
32    endpoint: String,
33    api_key_env: Option<String>,
34    auth: ProviderAuth,
35    codec: Symbol,
36    timeout: Duration,
37    stream: bool,
38    tools: bool,
39    max_response_bytes: usize,
40}
41
42impl HttpRunner {
43    /// Builds a runner from an open provider config without changing the HTTP
44    /// transport path.
45    pub fn new_provider(config: ProviderConfig) -> Self {
46        let auth = config.profile.auth.clone();
47        Self {
48            runner: config.runner,
49            model: config.model,
50            provider: config.profile.provider,
51            locality: config.locality,
52            runner_label: "runner/provider",
53            request_path: config.profile.chat_path,
54            endpoint: config.endpoint,
55            api_key_env: config.api_key_env,
56            auth,
57            codec: config.codec,
58            timeout: config.timeout,
59            stream: config.stream,
60            tools: config.tools,
61            max_response_bytes: config.max_output_bytes,
62        }
63    }
64
65    /// Builds a runner targeting an OpenAI-compatible `/chat/completions`
66    /// endpoint, reading its API key from the `api_key_env` environment
67    /// variable.
68    #[allow(clippy::too_many_arguments)]
69    pub fn new_openai_compatible(
70        runner: Symbol,
71        model: impl Into<String>,
72        endpoint: impl Into<String>,
73        api_key_env: impl Into<String>,
74        codec: Symbol,
75        timeout: Duration,
76        stream: bool,
77        tools: bool,
78        max_response_bytes: usize,
79    ) -> Self {
80        let api_key_env = api_key_env.into();
81        Self {
82            runner,
83            model: model.into(),
84            provider: Symbol::new("openai-compatible"),
85            locality: Symbol::new("network"),
86            runner_label: "runner/openai-compatible",
87            request_path: "/chat/completions",
88            endpoint: endpoint.into(),
89            api_key_env: Some(api_key_env.clone()),
90            auth: ProviderAuth::BearerEnv { env: api_key_env },
91            codec,
92            timeout,
93            stream,
94            tools,
95            max_response_bytes,
96        }
97    }
98
99    /// Builds a runner targeting an Ollama endpoint at the given `locality`.
100    #[allow(clippy::too_many_arguments)]
101    pub fn new_ollama(
102        runner: Symbol,
103        model: impl Into<String>,
104        locality: Symbol,
105        endpoint: impl Into<String>,
106        codec: Symbol,
107        timeout: Duration,
108        stream: bool,
109        tools: bool,
110        max_response_bytes: usize,
111    ) -> Self {
112        Self {
113            runner,
114            model: model.into(),
115            provider: Symbol::new("ollama"),
116            locality,
117            runner_label: "runner/ollama",
118            request_path: "/api/chat",
119            endpoint: endpoint.into(),
120            api_key_env: None,
121            auth: ProviderAuth::None,
122            codec,
123            timeout,
124            stream,
125            tools,
126            max_response_bytes,
127        }
128    }
129
130    fn infer_inner(&self, cx: &mut Cx, request: ModelRequest) -> Result<ModelResponse> {
131        let include_raw = self.include_raw(cx, &request);
132        let api_key = self.api_key()?;
133        let headers = self.request_headers(api_key.as_deref());
134        let body = self.encode_request(request, self.stream)?;
135        let response = post_json(
136            HttpRunnerRequest {
137                runner_label: self.runner_label,
138                endpoint: self.endpoint.as_str(),
139                path: self.request_path,
140                headers,
141                timeout: self.timeout,
142                body,
143                max_response_bytes: self.max_response_bytes,
144            },
145            api_key.as_deref(),
146        )?;
147        self.decode_response(&response.body, include_raw)
148    }
149
150    fn infer_stream_inner(
151        &self,
152        cx: &mut Cx,
153        request: ModelRequest,
154        sink: &mut dyn ModelEventSink,
155    ) -> Result<ModelResponse> {
156        if !self.stream {
157            let response = self.infer_inner(cx, request)?;
158            sink.emit(ModelEvent::final_of(&response))?;
159            return Ok(response);
160        }
161        let include_raw = self.include_raw(cx, &request);
162        let api_key = self.api_key()?;
163        let headers = self.request_headers(api_key.as_deref());
164        let body = self.encode_request(request, true)?;
165        let mut decoder = self.stream_decoder(include_raw)?;
166        sink.emit(decoder.start_event())?;
167        let response = post_json_stream(
168            HttpRunnerRequest {
169                runner_label: self.runner_label,
170                endpoint: self.endpoint.as_str(),
171                path: self.request_path,
172                headers,
173                timeout: self.timeout,
174                body,
175                max_response_bytes: self.max_response_bytes,
176            },
177            api_key.as_deref(),
178            &mut |chunk| decoder.feed(chunk, sink),
179        )?;
180        let model_response = if decoder.has_stream_output() {
181            decoder.finish(sink)?
182        } else {
183            self.decode_response(&response.body, include_raw)?
184        };
185        sink.emit(ModelEvent::final_of(&model_response))?;
186        Ok(model_response)
187    }
188
189    fn encode_request(&self, request: ModelRequest, stream: bool) -> Result<Vec<u8>> {
190        let openai_codec = Symbol::qualified("codec", "openai");
191        let anthropic_codec = Symbol::qualified("codec", "anthropic");
192        let ollama_codec = Symbol::qualified("codec", "ollama");
193        let lm_studio_codec = Symbol::qualified("codec", "lm-studio");
194        let lemonade_codec = Symbol::qualified("codec", "lemonade");
195        let request_expr: Expr = request.into();
196        if self.codec == openai_codec {
197            encode_openai_request(
198                &request_expr,
199                &OpenAiRequestOptions::new(self.model.clone(), stream, self.tools),
200            )
201        } else if self.codec == anthropic_codec {
202            encode_anthropic_request(
203                &request_expr,
204                &AnthropicRequestOptions::new(
205                    self.model.clone(),
206                    DEFAULT_ANTHROPIC_MAX_TOKENS,
207                    stream,
208                    self.tools,
209                ),
210            )
211        } else if self.codec == ollama_codec {
212            encode_ollama_request(
213                &request_expr,
214                &OllamaRequestOptions::new(self.model.clone(), stream, self.tools),
215            )
216        } else if self.codec == lm_studio_codec {
217            encode_lm_studio_request(
218                &request_expr,
219                &LmStudioRequestOptions::new(self.model.clone(), stream, self.tools),
220            )
221        } else if self.codec == lemonade_codec {
222            encode_lemonade_request(
223                &request_expr,
224                &LemonadeRequestOptions::new(self.model.clone(), stream, self.tools),
225            )
226        } else {
227            Err(Error::Eval(format!(
228                "{} unsupported codec {}",
229                self.runner_label, self.codec
230            )))
231        }
232    }
233
234    fn api_key(&self) -> Result<Option<String>> {
235        match &self.api_key_env {
236            Some(api_key_env) => Ok(Some(std::env::var(api_key_env).map_err(|_| {
237                Error::Eval(format!(
238                    "{} missing env var {}",
239                    self.runner_label, api_key_env
240                ))
241            })?)),
242            None => Ok(None),
243        }
244    }
245
246    fn request_headers(&self, secret: Option<&str>) -> Vec<(String, String)> {
247        if self.provider == Symbol::new("anthropic")
248            && matches!(self.auth, ProviderAuth::HeaderEnv { .. })
249            && let Some(secret) = secret
250        {
251            return anthropic_headers(secret);
252        }
253
254        let mut headers = vec![content_type_header()];
255        match (&self.auth, secret) {
256            (
257                ProviderAuth::BearerEnv { .. } | ProviderAuth::OptionalBearerEnv { .. },
258                Some(secret),
259            ) => {
260                headers.push(("Authorization".to_owned(), format!("Bearer {secret}")));
261            }
262            (ProviderAuth::HeaderEnv { header, .. }, Some(secret)) => {
263                headers.push((header.clone(), secret.to_owned()));
264            }
265            _ => {}
266        }
267        if self.provider == Symbol::new("anthropic") {
268            headers.push(("anthropic-version".to_owned(), ANTHROPIC_VERSION.to_owned()));
269        }
270        headers
271    }
272
273    fn decode_response(&self, body: &[u8], include_raw: bool) -> Result<ModelResponse> {
274        let openai_codec = Symbol::qualified("codec", "openai");
275        let anthropic_codec = Symbol::qualified("codec", "anthropic");
276        let ollama_codec = Symbol::qualified("codec", "ollama");
277        let lm_studio_codec = Symbol::qualified("codec", "lm-studio");
278        let lemonade_codec = Symbol::qualified("codec", "lemonade");
279        let expr = if self.codec == openai_codec {
280            decode_openai_response(self.runner.clone(), &self.model, body, include_raw)?
281        } else if self.codec == anthropic_codec {
282            if self.stream {
283                decode_anthropic_stream(self.runner.clone(), &self.model, body, include_raw)?
284            } else {
285                decode_anthropic_response(self.runner.clone(), &self.model, body, include_raw)?
286            }
287        } else if self.codec == ollama_codec {
288            if self.stream {
289                decode_ollama_stream(self.runner.clone(), &self.model, body, include_raw)?
290            } else {
291                decode_ollama_response(self.runner.clone(), &self.model, body, include_raw)?
292            }
293        } else if self.codec == lm_studio_codec {
294            if self.stream {
295                decode_lm_studio_stream(self.runner.clone(), &self.model, body, include_raw)?
296            } else {
297                decode_lm_studio_response(self.runner.clone(), &self.model, body, include_raw)?
298            }
299        } else if self.codec == lemonade_codec {
300            if self.stream {
301                decode_lemonade_stream(self.runner.clone(), &self.model, body, include_raw)?
302            } else {
303                decode_lemonade_response(self.runner.clone(), &self.model, body, include_raw)?
304            }
305        } else {
306            unreachable!("codec checked above")
307        };
308        ModelResponse::try_from(expr)
309    }
310
311    fn include_raw(&self, cx: &mut Cx, request: &ModelRequest) -> bool {
312        cx.require(&CapabilityName::new("ai-runner-raw-log"))
313            .is_ok()
314            && !request_privacy_no_raw(request)
315    }
316
317    fn stream_decoder(&self, include_raw: bool) -> Result<HttpStreamDecoder> {
318        let openai_codec = Symbol::qualified("codec", "openai");
319        let anthropic_codec = Symbol::qualified("codec", "anthropic");
320        let ollama_codec = Symbol::qualified("codec", "ollama");
321        let lm_studio_codec = Symbol::qualified("codec", "lm-studio");
322        let lemonade_codec = Symbol::qualified("codec", "lemonade");
323        if self.codec == openai_codec {
324            Ok(HttpStreamDecoder::openai(
325                self.runner.clone(),
326                self.model.clone(),
327                include_raw,
328            ))
329        } else if self.codec == anthropic_codec {
330            Ok(HttpStreamDecoder::anthropic(
331                self.runner.clone(),
332                self.model.clone(),
333                include_raw,
334            ))
335        } else if self.codec == ollama_codec {
336            Ok(HttpStreamDecoder::ollama(
337                self.runner.clone(),
338                self.model.clone(),
339                include_raw,
340            ))
341        } else if self.codec == lm_studio_codec || self.codec == lemonade_codec {
342            Ok(HttpStreamDecoder::openai(
343                self.runner.clone(),
344                self.model.clone(),
345                include_raw,
346            ))
347        } else {
348            Err(Error::Eval(format!(
349                "{} unsupported codec {}",
350                self.runner_label, self.codec
351            )))
352        }
353    }
354
355    fn error_response(&self, message: impl Into<String>) -> Result<ModelResponse> {
356        ModelResponse::try_from(model_error_expr(
357            self.runner.clone(),
358            self.model.clone(),
359            message.into(),
360        ))
361    }
362}
363
364const ANTHROPIC_VERSION: &str = "2023-06-01";
365const DEFAULT_ANTHROPIC_MAX_TOKENS: u64 = 1024;
366
367fn anthropic_headers(secret: &str) -> Vec<(String, String)> {
368    vec![
369        ("x-api-key".to_owned(), secret.to_owned()),
370        ("anthropic-version".to_owned(), ANTHROPIC_VERSION.to_owned()),
371        content_type_header(),
372    ]
373}
374
375fn content_type_header() -> (String, String) {
376    ("content-type".to_owned(), "application/json".to_owned())
377}
378
379fn request_privacy_no_raw(request: &ModelRequest) -> bool {
380    request
381        .extra
382        .iter()
383        .find_map(|(key, value)| is_field(key, "privacy").then_some(value))
384        .is_some_and(privacy_expr_no_raw)
385}
386
387fn privacy_expr_no_raw(expr: &Expr) -> bool {
388    match expr {
389        Expr::Symbol(symbol) => symbol.name.as_ref() == "no-raw",
390        Expr::String(text) => text == "no-raw",
391        Expr::List(items) | Expr::Vector(items) | Expr::Set(items) => {
392            items.iter().any(privacy_expr_no_raw)
393        }
394        Expr::Map(entries) => entries.iter().any(|(key, value)| {
395            is_field(key, "no-raw") && !matches!(value, Expr::Bool(false) | Expr::Nil)
396        }),
397        _ => false,
398    }
399}
400
401fn is_field(expr: &Expr, name: &str) -> bool {
402    matches!(
403        expr,
404        Expr::Symbol(symbol) if symbol.namespace.is_none() && symbol.name.as_ref() == name
405    )
406}
407
408impl ModelRunner for HttpRunner {
409    fn card(&self) -> ModelCard {
410        ModelCard::new(
411            self.runner.clone(),
412            self.model.clone(),
413            self.provider.clone(),
414            self.locality.clone(),
415        )
416    }
417
418    fn infer(&self, cx: &mut Cx, request: ModelRequest) -> Result<ModelResponse> {
419        match self.resolve_network_effect(cx, request, |runner, cx, request| {
420            runner.infer_inner(cx, request)
421        }) {
422            Ok(response) => Ok(response),
423            Err(error) => self.error_response(redact_text(&error.to_string(), &[])),
424        }
425    }
426
427    fn infer_stream(
428        &self,
429        cx: &mut Cx,
430        request: ModelRequest,
431        sink: &mut dyn ModelEventSink,
432    ) -> Result<ModelResponse> {
433        match self.resolve_network_effect(cx, request, {
434            let sink = &mut *sink;
435            |runner, cx, request| runner.infer_stream_inner(cx, request, sink)
436        }) {
437            Ok(response) => Ok(response),
438            Err(error) => {
439                let message = redact_text(&error.to_string(), &[]);
440                sink.emit(ModelEvent::error_text(
441                    self.runner.clone(),
442                    self.model.clone(),
443                    Expr::String("http-stream-error".to_owned()),
444                    message.clone(),
445                ))?;
446                let response = self.error_response(message)?;
447                sink.emit(ModelEvent::final_of(&response))?;
448                Ok(response)
449            }
450        }
451    }
452}
453
454impl HttpRunner {
455    fn resolve_network_effect<F>(
456        &self,
457        cx: &mut Cx,
458        request: ModelRequest,
459        perform: F,
460    ) -> Result<ModelResponse>
461    where
462        F: FnOnce(&Self, &mut Cx, ModelRequest) -> Result<ModelResponse>,
463    {
464        let effect = self.network_effect(cx, &request)?;
465        let result = effect::resolve_effect(cx, effect, |cx, _effect| {
466            let response = perform(self, cx, request)?;
467            response_ref(cx, response)
468        })?;
469        response_from_ref(cx, &result)
470    }
471
472    fn network_effect(&self, cx: &mut Cx, request: &ModelRequest) -> Result<Effect> {
473        let input = Datum::Node {
474            tag: Symbol::qualified("agent", "HttpRunnerInput"),
475            fields: vec![
476                (Symbol::new("runner"), Datum::Symbol(self.runner.clone())),
477                (Symbol::new("model"), Datum::String(self.model.clone())),
478                (
479                    Symbol::new("provider"),
480                    Datum::Symbol(self.provider.clone()),
481                ),
482                (
483                    Symbol::new("endpoint"),
484                    Datum::String(self.endpoint.clone()),
485                ),
486                (
487                    Symbol::new("request"),
488                    Datum::try_from(Expr::from(request.clone()))?,
489                ),
490            ],
491        };
492        let input = Ref::Content(cx.datum_store_mut().intern(input)?);
493        Effect::new(
494            effect::effect_network_kind(),
495            Ref::Symbol(self.runner.clone()),
496            input,
497            core_any_ref(),
498            effect::effect_resume_op_key(),
499            effect::effect_abort_op_key(),
500        )
501        .with_replay_key(Some(Ref::Symbol(Symbol::qualified(
502            "agent",
503            "http-runner-v1",
504        ))))
505    }
506}
507
508fn response_ref(cx: &mut Cx, response: ModelResponse) -> Result<Ref> {
509    Ok(Ref::Content(
510        cx.datum_store_mut()
511            .intern(Datum::try_from(Expr::from(response))?)?,
512    ))
513}
514
515fn response_from_ref(cx: &mut Cx, reference: &Ref) -> Result<ModelResponse> {
516    ModelResponse::try_from(value_from_ref(cx, reference)?.object().as_expr(cx)?)
517}
518
519#[cfg(test)]
520mod tests;
sim_lib_agent_runner_http/runner.rs

sim_lib_agent_runner_http/
runner.rs
