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 OUTPUT_GRAMMAR_DIALECT_EXTRA, OUTPUT_GRAMMAR_EXTRA, RETURN_CODEC_EXTRA, RETURN_SHAPE_EXTRA,
21 grammar_dialect_symbol,
22};
23use sim_shape::GrammarDialect;
24use std::time::Duration;
25
26#[derive(Clone, Debug)]
28pub struct HttpRunner {
29 runner: Symbol,
30 model: String,
31 provider: Symbol,
32 locality: Symbol,
33 runner_label: &'static str,
34 request_path: &'static str,
35 endpoint: String,
36 api_key_env: Option<String>,
37 auth: ProviderAuth,
38 codec: Symbol,
39 timeout: Duration,
40 stream: bool,
41 tools: bool,
42 max_response_bytes: usize,
43 grammar_dialects: Vec<GrammarDialect>,
44}
45
46impl HttpRunner {
47 pub fn new_provider(config: ProviderConfig) -> Self {
50 let auth = config.profile.auth.clone();
51 Self {
52 runner: config.runner,
53 model: config.model,
54 provider: config.profile.provider,
55 locality: config.locality,
56 runner_label: "runner/provider",
57 request_path: config.profile.chat_path,
58 endpoint: config.endpoint,
59 api_key_env: config.api_key_env,
60 auth,
61 codec: config.codec,
62 timeout: config.timeout,
63 stream: config.stream,
64 tools: config.tools,
65 max_response_bytes: config.max_output_bytes,
66 grammar_dialects: config.grammar_dialects,
67 }
68 }
69
70 #[allow(clippy::too_many_arguments)]
74 pub fn new_openai_compatible(
75 runner: Symbol,
76 model: impl Into<String>,
77 endpoint: impl Into<String>,
78 api_key_env: impl Into<String>,
79 codec: Symbol,
80 timeout: Duration,
81 stream: bool,
82 tools: bool,
83 max_response_bytes: usize,
84 ) -> Self {
85 let api_key_env = api_key_env.into();
86 Self {
87 runner,
88 model: model.into(),
89 provider: Symbol::new("openai-compatible"),
90 locality: Symbol::new("network"),
91 runner_label: "runner/openai-compatible",
92 request_path: "/chat/completions",
93 endpoint: endpoint.into(),
94 api_key_env: Some(api_key_env.clone()),
95 auth: ProviderAuth::BearerEnv { env: api_key_env },
96 codec,
97 timeout,
98 stream,
99 tools,
100 max_response_bytes,
101 grammar_dialects: Vec::new(),
102 }
103 }
104
105 #[allow(clippy::too_many_arguments)]
107 pub fn new_ollama(
108 runner: Symbol,
109 model: impl Into<String>,
110 locality: Symbol,
111 endpoint: impl Into<String>,
112 codec: Symbol,
113 timeout: Duration,
114 stream: bool,
115 tools: bool,
116 max_response_bytes: usize,
117 ) -> Self {
118 Self {
119 runner,
120 model: model.into(),
121 provider: Symbol::new("ollama"),
122 locality,
123 runner_label: "runner/ollama",
124 request_path: "/api/chat",
125 endpoint: endpoint.into(),
126 api_key_env: None,
127 auth: ProviderAuth::None,
128 codec,
129 timeout,
130 stream,
131 tools,
132 max_response_bytes,
133 grammar_dialects: vec![GrammarDialect::Gbnf],
134 }
135 }
136
137 fn infer_inner(&self, cx: &mut Cx, request: ModelRequest) -> Result<ModelResponse> {
138 let include_raw = self.include_raw(cx, &request);
139 let api_key = self.api_key()?;
140 let headers = self.request_headers(api_key.as_deref());
141 let body = self.encode_request(request, self.stream)?;
142 let response = post_json(
143 HttpRunnerRequest {
144 runner_label: self.runner_label,
145 endpoint: self.endpoint.as_str(),
146 path: self.request_path,
147 headers,
148 timeout: self.timeout,
149 body,
150 max_response_bytes: self.max_response_bytes,
151 },
152 api_key.as_deref(),
153 )?;
154 self.decode_response(&response.body, include_raw)
155 }
156
157 fn infer_stream_inner(
158 &self,
159 cx: &mut Cx,
160 request: ModelRequest,
161 sink: &mut dyn ModelEventSink,
162 ) -> Result<ModelResponse> {
163 if !self.stream {
164 let response = self.infer_inner(cx, request)?;
165 sink.emit(ModelEvent::final_of(&response))?;
166 return Ok(response);
167 }
168 let include_raw = self.include_raw(cx, &request);
169 let api_key = self.api_key()?;
170 let headers = self.request_headers(api_key.as_deref());
171 let body = self.encode_request(request, true)?;
172 let mut decoder = self.stream_decoder(include_raw)?;
173 sink.emit(decoder.start_event())?;
174 let response = post_json_stream(
175 HttpRunnerRequest {
176 runner_label: self.runner_label,
177 endpoint: self.endpoint.as_str(),
178 path: self.request_path,
179 headers,
180 timeout: self.timeout,
181 body,
182 max_response_bytes: self.max_response_bytes,
183 },
184 api_key.as_deref(),
185 &mut |chunk| decoder.feed(chunk, sink),
186 )?;
187 let model_response = if decoder.has_stream_output() {
188 decoder.finish(sink)?
189 } else {
190 self.decode_response(&response.body, include_raw)?
191 };
192 sink.emit(ModelEvent::final_of(&model_response))?;
193 Ok(model_response)
194 }
195
196 fn encode_request(&self, request: ModelRequest, stream: bool) -> Result<Vec<u8>> {
197 let openai_codec = Symbol::qualified("codec", "openai");
198 let anthropic_codec = Symbol::qualified("codec", "anthropic");
199 let ollama_codec = Symbol::qualified("codec", "ollama");
200 let lm_studio_codec = Symbol::qualified("codec", "lm-studio");
201 let lemonade_codec = Symbol::qualified("codec", "lemonade");
202 let request = self.prepare_output_grammar(request);
203 let request_expr: Expr = request.into();
204 if self.codec == openai_codec {
205 encode_openai_request(
206 &request_expr,
207 &OpenAiRequestOptions::new(self.model.clone(), stream, self.tools),
208 )
209 } else if self.codec == anthropic_codec {
210 encode_anthropic_request(
211 &request_expr,
212 &AnthropicRequestOptions::new(
213 self.model.clone(),
214 DEFAULT_ANTHROPIC_MAX_TOKENS,
215 stream,
216 self.tools,
217 ),
218 )
219 } else if self.codec == ollama_codec {
220 encode_ollama_request(
221 &request_expr,
222 &OllamaRequestOptions::new(self.model.clone(), stream, self.tools),
223 )
224 } else if self.codec == lm_studio_codec {
225 encode_lm_studio_request(
226 &request_expr,
227 &LmStudioRequestOptions::new(self.model.clone(), stream, self.tools),
228 )
229 } else if self.codec == lemonade_codec {
230 encode_lemonade_request(
231 &request_expr,
232 &LemonadeRequestOptions::new(self.model.clone(), stream, self.tools),
233 )
234 } else {
235 Err(Error::Eval(format!(
236 "{} unsupported codec {}",
237 self.runner_label, self.codec
238 )))
239 }
240 }
241
242 fn api_key(&self) -> Result<Option<String>> {
243 match &self.api_key_env {
244 Some(api_key_env) => Ok(Some(std::env::var(api_key_env).map_err(|_| {
245 Error::Eval(format!(
246 "{} missing env var {}",
247 self.runner_label, api_key_env
248 ))
249 })?)),
250 None => Ok(None),
251 }
252 }
253
254 fn request_headers(&self, secret: Option<&str>) -> Vec<(String, String)> {
255 if self.provider == Symbol::new("anthropic")
256 && matches!(self.auth, ProviderAuth::HeaderEnv { .. })
257 && let Some(secret) = secret
258 {
259 return anthropic_headers(secret);
260 }
261
262 let mut headers = vec![content_type_header()];
263 match (&self.auth, secret) {
264 (
265 ProviderAuth::BearerEnv { .. } | ProviderAuth::OptionalBearerEnv { .. },
266 Some(secret),
267 ) => {
268 headers.push(("Authorization".to_owned(), format!("Bearer {secret}")));
269 }
270 (ProviderAuth::HeaderEnv { header, .. }, Some(secret)) => {
271 headers.push((header.clone(), secret.to_owned()));
272 }
273 _ => {}
274 }
275 if self.provider == Symbol::new("anthropic") {
276 headers.push(("anthropic-version".to_owned(), ANTHROPIC_VERSION.to_owned()));
277 }
278 headers
279 }
280
281 fn decode_response(&self, body: &[u8], include_raw: bool) -> Result<ModelResponse> {
282 let openai_codec = Symbol::qualified("codec", "openai");
283 let anthropic_codec = Symbol::qualified("codec", "anthropic");
284 let ollama_codec = Symbol::qualified("codec", "ollama");
285 let lm_studio_codec = Symbol::qualified("codec", "lm-studio");
286 let lemonade_codec = Symbol::qualified("codec", "lemonade");
287 let expr = if self.codec == openai_codec {
288 decode_openai_response(self.runner.clone(), &self.model, body, include_raw)?
289 } else if self.codec == anthropic_codec {
290 if self.stream {
291 decode_anthropic_stream(self.runner.clone(), &self.model, body, include_raw)?
292 } else {
293 decode_anthropic_response(self.runner.clone(), &self.model, body, include_raw)?
294 }
295 } else if self.codec == ollama_codec {
296 if self.stream {
297 decode_ollama_stream(self.runner.clone(), &self.model, body, include_raw)?
298 } else {
299 decode_ollama_response(self.runner.clone(), &self.model, body, include_raw)?
300 }
301 } else if self.codec == lm_studio_codec {
302 if self.stream {
303 decode_lm_studio_stream(self.runner.clone(), &self.model, body, include_raw)?
304 } else {
305 decode_lm_studio_response(self.runner.clone(), &self.model, body, include_raw)?
306 }
307 } else if self.codec == lemonade_codec {
308 if self.stream {
309 decode_lemonade_stream(self.runner.clone(), &self.model, body, include_raw)?
310 } else {
311 decode_lemonade_response(self.runner.clone(), &self.model, body, include_raw)?
312 }
313 } else {
314 unreachable!("codec checked above")
315 };
316 ModelResponse::try_from(expr)
317 }
318
319 fn include_raw(&self, cx: &mut Cx, request: &ModelRequest) -> bool {
320 cx.require(&CapabilityName::new("ai-runner-raw-log"))
321 .is_ok()
322 && !request_privacy_no_raw(request)
323 }
324
325 fn direct_capabilities(&self) -> Vec<CapabilityName> {
326 let mut capabilities = vec![CapabilityName::new(AI_RUNNER_CAPABILITY)];
327 if self.locality == Symbol::new("local") {
328 capabilities.push(CapabilityName::new(AI_RUNNER_LOCAL_CAPABILITY));
329 } else {
330 capabilities.push(CapabilityName::new(AI_RUNNER_NETWORK_CAPABILITY));
331 }
332 if self.api_key_env.is_some() {
333 capabilities.push(CapabilityName::new(AI_RUNNER_SECRET_CAPABILITY));
334 }
335 capabilities
336 }
337
338 fn stream_decoder(&self, include_raw: bool) -> Result<HttpStreamDecoder> {
339 let openai_codec = Symbol::qualified("codec", "openai");
340 let anthropic_codec = Symbol::qualified("codec", "anthropic");
341 let ollama_codec = Symbol::qualified("codec", "ollama");
342 let lm_studio_codec = Symbol::qualified("codec", "lm-studio");
343 let lemonade_codec = Symbol::qualified("codec", "lemonade");
344 if self.codec == openai_codec {
345 Ok(HttpStreamDecoder::openai(
346 self.runner.clone(),
347 self.model.clone(),
348 include_raw,
349 ))
350 } else if self.codec == anthropic_codec {
351 Ok(HttpStreamDecoder::anthropic(
352 self.runner.clone(),
353 self.model.clone(),
354 include_raw,
355 ))
356 } else if self.codec == ollama_codec {
357 Ok(HttpStreamDecoder::ollama(
358 self.runner.clone(),
359 self.model.clone(),
360 include_raw,
361 ))
362 } else if self.codec == lm_studio_codec || self.codec == lemonade_codec {
363 Ok(HttpStreamDecoder::openai(
364 self.runner.clone(),
365 self.model.clone(),
366 include_raw,
367 ))
368 } else {
369 Err(Error::Eval(format!(
370 "{} unsupported codec {}",
371 self.runner_label, self.codec
372 )))
373 }
374 }
375
376 fn error_response(&self, message: impl Into<String>) -> Result<ModelResponse> {
377 ModelResponse::try_from(model_error_expr(
378 self.runner.clone(),
379 self.model.clone(),
380 message.into(),
381 ))
382 }
383
384 fn prepare_output_grammar(&self, mut request: ModelRequest) -> ModelRequest {
385 let Some(dialect) = self.preferred_grammar_dialect() else {
386 strip_output_grammar(&mut request.extra);
387 return request;
388 };
389 if extra_field(&request.extra, RETURN_SHAPE_EXTRA).is_none()
390 && !explicit_output_grammar_matches(&request.extra, dialect)
391 {
392 strip_output_grammar(&mut request.extra);
393 return request;
394 }
395 let return_codec = extra_symbol(&request.extra, RETURN_CODEC_EXTRA);
396 if return_codec.as_ref() != Some(&Symbol::qualified("codec", "json")) {
397 strip_output_grammar(&mut request.extra);
398 return request;
399 }
400 if !explicit_output_grammar_matches(&request.extra, dialect) {
401 remove_extra(&mut request.extra, OUTPUT_GRAMMAR_EXTRA);
402 }
403 upsert_extra(
404 &mut request.extra,
405 OUTPUT_GRAMMAR_DIALECT_EXTRA,
406 Expr::Symbol(grammar_dialect_symbol(dialect)),
407 );
408 request
409 }
410
411 fn preferred_grammar_dialect(&self) -> Option<GrammarDialect> {
412 if self.grammar_dialects.contains(&GrammarDialect::JsonSchema) {
413 Some(GrammarDialect::JsonSchema)
414 } else if self.grammar_dialects.contains(&GrammarDialect::Gbnf) {
415 Some(GrammarDialect::Gbnf)
416 } else {
417 None
418 }
419 }
420}
421
422const ANTHROPIC_VERSION: &str = "2023-06-01";
423const DEFAULT_ANTHROPIC_MAX_TOKENS: u64 = 1024;
424const AI_RUNNER_CAPABILITY: &str = "ai-runner";
425const AI_RUNNER_NETWORK_CAPABILITY: &str = "ai-runner-network";
426const AI_RUNNER_LOCAL_CAPABILITY: &str = "ai-runner-local";
427const AI_RUNNER_SECRET_CAPABILITY: &str = "ai-runner-secret";
428
429fn anthropic_headers(secret: &str) -> Vec<(String, String)> {
430 vec![
431 ("x-api-key".to_owned(), secret.to_owned()),
432 ("anthropic-version".to_owned(), ANTHROPIC_VERSION.to_owned()),
433 content_type_header(),
434 ]
435}
436
437fn content_type_header() -> (String, String) {
438 ("content-type".to_owned(), "application/json".to_owned())
439}
440
441fn request_privacy_no_raw(request: &ModelRequest) -> bool {
442 request
443 .extra
444 .iter()
445 .find_map(|(key, value)| is_field(key, "privacy").then_some(value))
446 .is_some_and(privacy_expr_no_raw)
447}
448
449fn privacy_expr_no_raw(expr: &Expr) -> bool {
450 match expr {
451 Expr::Symbol(symbol) => symbol.name.as_ref() == "no-raw",
452 Expr::String(text) => text == "no-raw",
453 Expr::List(items) | Expr::Vector(items) | Expr::Set(items) => {
454 items.iter().any(privacy_expr_no_raw)
455 }
456 Expr::Map(entries) => entries.iter().any(|(key, value)| {
457 is_field(key, "no-raw") && !matches!(value, Expr::Bool(false) | Expr::Nil)
458 }),
459 _ => false,
460 }
461}
462
463fn is_field(expr: &Expr, name: &str) -> bool {
464 matches!(
465 expr,
466 Expr::Symbol(symbol) if symbol.namespace.is_none() && symbol.name.as_ref() == name
467 )
468}
469
470fn extra_field<'a>(entries: &'a [(Expr, Expr)], name: &str) -> Option<&'a Expr> {
471 entries.iter().find_map(|(key, value)| {
472 if is_field(key, name) {
473 Some(value)
474 } else {
475 None
476 }
477 })
478}
479
480fn extra_symbol(entries: &[(Expr, Expr)], name: &str) -> Option<Symbol> {
481 match extra_field(entries, name) {
482 Some(Expr::Symbol(symbol)) => Some(symbol.clone()),
483 _ => None,
484 }
485}
486
487fn upsert_extra(entries: &mut Vec<(Expr, Expr)>, name: &str, value: Expr) {
488 if let Some((_, existing)) = entries.iter_mut().find(|(key, _)| is_field(key, name)) {
489 *existing = value;
490 return;
491 }
492 entries.push((Expr::Symbol(Symbol::new(name)), value));
493}
494
495fn strip_output_grammar(entries: &mut Vec<(Expr, Expr)>) {
496 entries.retain(|(key, _)| {
497 !is_field(key, OUTPUT_GRAMMAR_EXTRA) && !is_field(key, OUTPUT_GRAMMAR_DIALECT_EXTRA)
498 });
499}
500
501fn remove_extra(entries: &mut Vec<(Expr, Expr)>, name: &str) {
502 entries.retain(|(key, _)| !is_field(key, name));
503}
504
505fn explicit_output_grammar_matches(entries: &[(Expr, Expr)], dialect: GrammarDialect) -> bool {
506 matches!(
507 extra_field(entries, OUTPUT_GRAMMAR_EXTRA),
508 Some(Expr::String(_))
509 ) && extra_field(entries, OUTPUT_GRAMMAR_DIALECT_EXTRA)
510 .and_then(|expr| match expr {
511 Expr::Symbol(symbol) => grammar_dialect_from_symbol_local(symbol),
512 _ => None,
513 })
514 .unwrap_or(GrammarDialect::JsonSchema)
515 == dialect
516}
517
518fn grammar_dialect_from_symbol_local(symbol: &Symbol) -> Option<GrammarDialect> {
519 match symbol.name.as_ref() {
520 "json-schema" if symbol.namespace.is_none() => Some(GrammarDialect::JsonSchema),
521 "gbnf" if symbol.namespace.is_none() => Some(GrammarDialect::Gbnf),
522 "sexpr" if symbol.namespace.is_none() => Some(GrammarDialect::SExpr),
523 _ => None,
524 }
525}
526
527impl ModelRunner for HttpRunner {
528 fn card(&self) -> ModelCard {
529 let mut card = ModelCard::new(
530 self.runner.clone(),
531 self.model.clone(),
532 self.provider.clone(),
533 self.locality.clone(),
534 );
535 if !self.grammar_dialects.is_empty() {
536 card.extra.push((
537 Expr::Symbol(Symbol::new("output-grammar-dialects")),
538 Expr::Vector(
539 self.grammar_dialects
540 .iter()
541 .copied()
542 .map(grammar_dialect_symbol)
543 .map(Expr::Symbol)
544 .collect(),
545 ),
546 ));
547 }
548 card
549 }
550
551 fn infer(&self, cx: &mut Cx, request: ModelRequest) -> Result<ModelResponse> {
552 match self.resolve_network_effect(cx, request, |runner, cx, request| {
553 runner.infer_inner(cx, request)
554 }) {
555 Ok(response) => Ok(response),
556 Err(error) => self.error_response(redact_text(&error.to_string(), &[])),
557 }
558 }
559
560 fn infer_stream(
561 &self,
562 cx: &mut Cx,
563 request: ModelRequest,
564 sink: &mut dyn ModelEventSink,
565 ) -> Result<ModelResponse> {
566 match self.resolve_network_effect(cx, request, {
567 let sink = &mut *sink;
568 |runner, cx, request| runner.infer_stream_inner(cx, request, sink)
569 }) {
570 Ok(response) => Ok(response),
571 Err(error) => {
572 let message = redact_text(&error.to_string(), &[]);
573 sink.emit(ModelEvent::error_text(
574 self.runner.clone(),
575 self.model.clone(),
576 Expr::String("http-stream-error".to_owned()),
577 message.clone(),
578 ))?;
579 let response = self.error_response(message)?;
580 sink.emit(ModelEvent::final_of(&response))?;
581 Ok(response)
582 }
583 }
584 }
585}
586
587impl HttpRunner {
588 fn resolve_network_effect<F>(
589 &self,
590 cx: &mut Cx,
591 request: ModelRequest,
592 perform: F,
593 ) -> Result<ModelResponse>
594 where
595 F: FnOnce(&Self, &mut Cx, ModelRequest) -> Result<ModelResponse>,
596 {
597 let effect = self.network_effect(cx, &request)?;
598 let result = effect::resolve_effect(cx, effect, |cx, _effect| {
599 let response = perform(self, cx, request)?;
600 response_ref(cx, response)
601 })?;
602 response_from_ref(cx, &result)
603 }
604
605 fn network_effect(&self, cx: &mut Cx, request: &ModelRequest) -> Result<Effect> {
606 let input = Datum::Node {
607 tag: Symbol::qualified("agent", "HttpRunnerInput"),
608 fields: vec![
609 (Symbol::new("runner"), Datum::Symbol(self.runner.clone())),
610 (Symbol::new("model"), Datum::String(self.model.clone())),
611 (
612 Symbol::new("provider"),
613 Datum::Symbol(self.provider.clone()),
614 ),
615 (
616 Symbol::new("endpoint"),
617 Datum::String(self.endpoint.clone()),
618 ),
619 (
620 Symbol::new("request"),
621 Datum::try_from(Expr::from(request.clone()))?,
622 ),
623 ],
624 };
625 let input = Ref::Content(cx.datum_store_mut().intern(input)?);
626 Effect::new(
627 network_effect_kind(),
628 Ref::Symbol(self.runner.clone()),
629 input,
630 core_any_ref(),
631 effect::effect_resume_op_key(),
632 effect::effect_abort_op_key(),
633 )
634 .with_requirements(self.direct_capabilities())
635 .with_replay_key(Some(Ref::Symbol(Symbol::qualified(
636 "agent",
637 "http-runner-v1",
638 ))))
639 }
640}
641
642fn network_effect_kind() -> Symbol {
643 Symbol::qualified("effect", "network")
644}
645
646fn response_ref(cx: &mut Cx, response: ModelResponse) -> Result<Ref> {
647 Ok(Ref::Content(
648 cx.datum_store_mut()
649 .intern(Datum::try_from(Expr::from(response))?)?,
650 ))
651}
652
653fn response_from_ref(cx: &mut Cx, reference: &Ref) -> Result<ModelResponse> {
654 ModelResponse::try_from(value_from_ref(cx, reference)?.object().as_expr(cx)?)
655}
656
657#[cfg(test)]
658mod tests;