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 direct_capabilities(&self) -> Vec<CapabilityName> {
318        let mut capabilities = vec![CapabilityName::new(AI_RUNNER_CAPABILITY)];
319        if self.locality == Symbol::new("local") {
320            capabilities.push(CapabilityName::new(AI_RUNNER_LOCAL_CAPABILITY));
321        } else {
322            capabilities.push(CapabilityName::new(AI_RUNNER_NETWORK_CAPABILITY));
323        }
324        if self.api_key_env.is_some() {
325            capabilities.push(CapabilityName::new(AI_RUNNER_SECRET_CAPABILITY));
326        }
327        capabilities
328    }
329
330    fn stream_decoder(&self, include_raw: bool) -> Result<HttpStreamDecoder> {
331        let openai_codec = Symbol::qualified("codec", "openai");
332        let anthropic_codec = Symbol::qualified("codec", "anthropic");
333        let ollama_codec = Symbol::qualified("codec", "ollama");
334        let lm_studio_codec = Symbol::qualified("codec", "lm-studio");
335        let lemonade_codec = Symbol::qualified("codec", "lemonade");
336        if self.codec == openai_codec {
337            Ok(HttpStreamDecoder::openai(
338                self.runner.clone(),
339                self.model.clone(),
340                include_raw,
341            ))
342        } else if self.codec == anthropic_codec {
343            Ok(HttpStreamDecoder::anthropic(
344                self.runner.clone(),
345                self.model.clone(),
346                include_raw,
347            ))
348        } else if self.codec == ollama_codec {
349            Ok(HttpStreamDecoder::ollama(
350                self.runner.clone(),
351                self.model.clone(),
352                include_raw,
353            ))
354        } else if self.codec == lm_studio_codec || self.codec == lemonade_codec {
355            Ok(HttpStreamDecoder::openai(
356                self.runner.clone(),
357                self.model.clone(),
358                include_raw,
359            ))
360        } else {
361            Err(Error::Eval(format!(
362                "{} unsupported codec {}",
363                self.runner_label, self.codec
364            )))
365        }
366    }
367
368    fn error_response(&self, message: impl Into<String>) -> Result<ModelResponse> {
369        ModelResponse::try_from(model_error_expr(
370            self.runner.clone(),
371            self.model.clone(),
372            message.into(),
373        ))
374    }
375}
376
377const ANTHROPIC_VERSION: &str = "2023-06-01";
378const DEFAULT_ANTHROPIC_MAX_TOKENS: u64 = 1024;
379const AI_RUNNER_CAPABILITY: &str = "ai-runner";
380const AI_RUNNER_NETWORK_CAPABILITY: &str = "ai-runner-network";
381const AI_RUNNER_LOCAL_CAPABILITY: &str = "ai-runner-local";
382const AI_RUNNER_SECRET_CAPABILITY: &str = "ai-runner-secret";
383
384fn anthropic_headers(secret: &str) -> Vec<(String, String)> {
385    vec![
386        ("x-api-key".to_owned(), secret.to_owned()),
387        ("anthropic-version".to_owned(), ANTHROPIC_VERSION.to_owned()),
388        content_type_header(),
389    ]
390}
391
392fn content_type_header() -> (String, String) {
393    ("content-type".to_owned(), "application/json".to_owned())
394}
395
396fn request_privacy_no_raw(request: &ModelRequest) -> bool {
397    request
398        .extra
399        .iter()
400        .find_map(|(key, value)| is_field(key, "privacy").then_some(value))
401        .is_some_and(privacy_expr_no_raw)
402}
403
404fn privacy_expr_no_raw(expr: &Expr) -> bool {
405    match expr {
406        Expr::Symbol(symbol) => symbol.name.as_ref() == "no-raw",
407        Expr::String(text) => text == "no-raw",
408        Expr::List(items) | Expr::Vector(items) | Expr::Set(items) => {
409            items.iter().any(privacy_expr_no_raw)
410        }
411        Expr::Map(entries) => entries.iter().any(|(key, value)| {
412            is_field(key, "no-raw") && !matches!(value, Expr::Bool(false) | Expr::Nil)
413        }),
414        _ => false,
415    }
416}
417
418fn is_field(expr: &Expr, name: &str) -> bool {
419    matches!(
420        expr,
421        Expr::Symbol(symbol) if symbol.namespace.is_none() && symbol.name.as_ref() == name
422    )
423}
424
425impl ModelRunner for HttpRunner {
426    fn card(&self) -> ModelCard {
427        ModelCard::new(
428            self.runner.clone(),
429            self.model.clone(),
430            self.provider.clone(),
431            self.locality.clone(),
432        )
433    }
434
435    fn infer(&self, cx: &mut Cx, request: ModelRequest) -> Result<ModelResponse> {
436        match self.resolve_network_effect(cx, request, |runner, cx, request| {
437            runner.infer_inner(cx, request)
438        }) {
439            Ok(response) => Ok(response),
440            Err(error) => self.error_response(redact_text(&error.to_string(), &[])),
441        }
442    }
443
444    fn infer_stream(
445        &self,
446        cx: &mut Cx,
447        request: ModelRequest,
448        sink: &mut dyn ModelEventSink,
449    ) -> Result<ModelResponse> {
450        match self.resolve_network_effect(cx, request, {
451            let sink = &mut *sink;
452            |runner, cx, request| runner.infer_stream_inner(cx, request, sink)
453        }) {
454            Ok(response) => Ok(response),
455            Err(error) => {
456                let message = redact_text(&error.to_string(), &[]);
457                sink.emit(ModelEvent::error_text(
458                    self.runner.clone(),
459                    self.model.clone(),
460                    Expr::String("http-stream-error".to_owned()),
461                    message.clone(),
462                ))?;
463                let response = self.error_response(message)?;
464                sink.emit(ModelEvent::final_of(&response))?;
465                Ok(response)
466            }
467        }
468    }
469}
470
471impl HttpRunner {
472    fn resolve_network_effect<F>(
473        &self,
474        cx: &mut Cx,
475        request: ModelRequest,
476        perform: F,
477    ) -> Result<ModelResponse>
478    where
479        F: FnOnce(&Self, &mut Cx, ModelRequest) -> Result<ModelResponse>,
480    {
481        let effect = self.network_effect(cx, &request)?;
482        let result = effect::resolve_effect(cx, effect, |cx, _effect| {
483            let response = perform(self, cx, request)?;
484            response_ref(cx, response)
485        })?;
486        response_from_ref(cx, &result)
487    }
488
489    fn network_effect(&self, cx: &mut Cx, request: &ModelRequest) -> Result<Effect> {
490        let input = Datum::Node {
491            tag: Symbol::qualified("agent", "HttpRunnerInput"),
492            fields: vec![
493                (Symbol::new("runner"), Datum::Symbol(self.runner.clone())),
494                (Symbol::new("model"), Datum::String(self.model.clone())),
495                (
496                    Symbol::new("provider"),
497                    Datum::Symbol(self.provider.clone()),
498                ),
499                (
500                    Symbol::new("endpoint"),
501                    Datum::String(self.endpoint.clone()),
502                ),
503                (
504                    Symbol::new("request"),
505                    Datum::try_from(Expr::from(request.clone()))?,
506                ),
507            ],
508        };
509        let input = Ref::Content(cx.datum_store_mut().intern(input)?);
510        Effect::new(
511            network_effect_kind(),
512            Ref::Symbol(self.runner.clone()),
513            input,
514            core_any_ref(),
515            effect::effect_resume_op_key(),
516            effect::effect_abort_op_key(),
517        )
518        .with_requirements(self.direct_capabilities())
519        .with_replay_key(Some(Ref::Symbol(Symbol::qualified(
520            "agent",
521            "http-runner-v1",
522        ))))
523    }
524}
525
526fn network_effect_kind() -> Symbol {
527    Symbol::qualified("effect", "network")
528}
529
530fn response_ref(cx: &mut Cx, response: ModelResponse) -> Result<Ref> {
531    Ok(Ref::Content(
532        cx.datum_store_mut()
533            .intern(Datum::try_from(Expr::from(response))?)?,
534    ))
535}
536
537fn response_from_ref(cx: &mut Cx, reference: &Ref) -> Result<ModelResponse> {
538    ModelResponse::try_from(value_from_ref(cx, reference)?.object().as_expr(cx)?)
539}
540
541#[cfg(test)]
542mod tests;
sim_lib_agent_runner_http/runner.rs

sim_lib_agent_runner_http/
runner.rs
