Initial CodeCraft: web-based coding harness for AI models
Rust backend (axum/tokio): agent loop with streaming, tools (fs, bash, web_fetch, todo, ask_user, subagents, skills, craft snapshots/compare), permissions with modes and rules, Anthropic + OpenAI-compatible providers, MCP client (stdio/HTTP, elicitation), SQLite sessions, usage statistics, compaction, commented TOML config editable via toml_edit. React PWA: chat with live streaming, permission/question cards, modes (build/plan/craft), settings for providers/models/prices/MCP/permissions, raw config editor, statistics dashboard, i18n (en/de). Docker setup. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Richard ZinkandClaude Opus 5.5 committed
commit
f3a2fe3816
62 files changed
+21835
No files matched your search
@@ -0,0 +1,301 @@
|
||||
//! Anthropic Messages API (streaming). Also works with Anthropic-compatible
|
||||
//! third-party endpoints; first-party-only features are gated on the host.
|
||||
|
||||
use super::{ChatRequest, Endpoint, Provider, StreamEvent, http_error};
|
||||
use crate::config::ModelConfig;
|
||||
use crate::types::{Message, Part, Role, Usage};
|
||||
use anyhow::{Result, anyhow};
|
||||
use async_trait::async_trait;
|
||||
use eventsource_stream::Eventsource;
|
||||
use futures::StreamExt;
|
||||
use serde_json::{Value, json};
|
||||
use std::collections::BTreeMap;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
const API_VERSION: &str = "2026-10-21";
|
||||
|
||||
pub struct Anthropic {
|
||||
client: reqwest::Client,
|
||||
ep: Endpoint,
|
||||
}
|
||||
|
||||
impl Anthropic {
|
||||
pub fn new(client: reqwest::Client, ep: Endpoint) -> Self {
|
||||
Self { client, ep }
|
||||
}
|
||||
|
||||
fn first_party(&self) -> bool {
|
||||
self.ep.base_url.contains("api.anthropic.com")
|
||||
}
|
||||
|
||||
fn request(&self, method: reqwest::Method, path: &str) -> reqwest::RequestBuilder {
|
||||
let mut rb = self.client.request(method, format!("{}{}", self.ep.base_url, path));
|
||||
if !self.ep.headers.keys().any(|k| k.eq_ignore_ascii_case("anthropic-version")) {
|
||||
rb = rb.header("anthropic-version", API_VERSION);
|
||||
}
|
||||
self.ep.apply(rb)
|
||||
}
|
||||
}
|
||||
|
||||
/// Convert our neutral history into Anthropic message params.
|
||||
pub fn to_wire_messages(messages: &[Message]) -> Vec<Value> {
|
||||
let mut out: Vec<Value> = Vec::new();
|
||||
for m in messages {
|
||||
let mut content = Vec::new();
|
||||
for p in &m.parts {
|
||||
match p {
|
||||
Part::Text { text } if !text.is_empty() => {
|
||||
content.push(json!({"type": "text", "text": text}))
|
||||
}
|
||||
Part::Summary { text } => content.push(json!({
|
||||
"type": "text",
|
||||
"text": format!("<conversation-summary>\n{text}\n</conversation-summary>")
|
||||
})),
|
||||
Part::Image { image } => content.push(json!({
|
||||
"type": "image",
|
||||
"source": {"type": "base64", "media_type": image.media_type, "data": image.data}
|
||||
})),
|
||||
// Thinking blocks are replayed unchanged (assistant turns only).
|
||||
Part::Reasoning { text, signature } if m.role == Role::Assistant => {
|
||||
match signature.as_deref() {
|
||||
Some(sig) if sig.starts_with("redacted:") => content.push(json!({
|
||||
"type": "redacted_thinking",
|
||||
"data": &sig["redacted:".len()..]
|
||||
})),
|
||||
Some(sig) => content.push(json!({
|
||||
"type": "thinking", "thinking": text, "signature": sig
|
||||
})),
|
||||
// Reasoning from another provider has no signature – drop it.
|
||||
None => {}
|
||||
}
|
||||
}
|
||||
Part::ToolCall { id, name, input, .. } => content.push(json!({
|
||||
"type": "tool_use", "id": id, "name": name, "input": input
|
||||
})),
|
||||
Part::ToolResult { call_id, content: text, images, is_error, .. } => {
|
||||
let mut blocks = vec![json!({"type": "text", "text": if text.is_empty() { "(no output)" } else { text }})];
|
||||
for img in images {
|
||||
blocks.push(json!({
|
||||
"type": "image",
|
||||
"source": {"type": "base64", "media_type": img.media_type, "data": img.data}
|
||||
}));
|
||||
}
|
||||
content.push(json!({
|
||||
"type": "tool_result", "tool_use_id": call_id,
|
||||
"content": blocks, "is_error": is_error
|
||||
}))
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
if content.is_empty() {
|
||||
continue;
|
||||
}
|
||||
let role = match m.role {
|
||||
Role::User => "user",
|
||||
Role::Assistant => "assistant",
|
||||
};
|
||||
// Merge consecutive messages of the same role (the API requires alternation).
|
||||
if let Some(last) = out.last_mut() {
|
||||
if last["role"] == role {
|
||||
last["content"].as_array_mut().unwrap().extend(content);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
out.push(json!({"role": role, "content": content}));
|
||||
}
|
||||
// tool_result blocks must come first in a user message.
|
||||
for msg in out.iter_mut().filter(|m| m["role"] == "user") {
|
||||
let arr = msg["content"].as_array_mut().unwrap();
|
||||
arr.sort_by_key(|b| if b["type"] == "tool_result" { 0 } else { 1 });
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct BlockState {
|
||||
kind: String,
|
||||
id: String,
|
||||
name: String,
|
||||
json: String,
|
||||
signature: String,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Provider for Anthropic {
|
||||
async fn stream(
|
||||
&self,
|
||||
req: ChatRequest,
|
||||
tx: mpsc::Sender<StreamEvent>,
|
||||
cancel: CancellationToken,
|
||||
) -> Result<()> {
|
||||
let first_party = self.first_party();
|
||||
let tools: Vec<Value> = req
|
||||
.tools
|
||||
.iter()
|
||||
.map(|t| {
|
||||
let mut v = json!({
|
||||
"name": t.name, "description": t.description, "input_schema": t.input_schema
|
||||
});
|
||||
if first_party {
|
||||
v["eager_input_streaming"] = json!(true);
|
||||
}
|
||||
v
|
||||
})
|
||||
.collect();
|
||||
let mut body = json!({
|
||||
"model": req.model.id,
|
||||
"max_tokens": req.max_tokens,
|
||||
"stream": true,
|
||||
"system": [{"type": "text", "text": req.system, "cache_control": {"type": "ephemeral"}}],
|
||||
"messages": to_wire_messages(&req.messages),
|
||||
});
|
||||
if !tools.is_empty() {
|
||||
body["tools"] = Value::Array(tools);
|
||||
}
|
||||
if first_party {
|
||||
// Automatic breakpoint on the last cacheable block (conversation prefix).
|
||||
body["cache_control"] = json!({"type": "ephemeral"});
|
||||
}
|
||||
if req.reasoning && req.model.reasoning {
|
||||
body["thinking"] = json!({"type": "adaptive", "display": "summarized"});
|
||||
}
|
||||
if let Some(effort) = &req.model.effort {
|
||||
body["output_config"] = json!({"effort": effort});
|
||||
}
|
||||
|
||||
let resp = self.request(reqwest::Method::POST, "/messages").json(&body).send().await?;
|
||||
if !resp.status().is_success() {
|
||||
return Err(http_error(resp).await);
|
||||
}
|
||||
|
||||
let mut events = resp.bytes_stream().eventsource();
|
||||
let mut blocks: BTreeMap<u64, BlockState> = BTreeMap::new();
|
||||
let mut usage = Usage::default();
|
||||
let mut stop_reason = String::from("end_turn");
|
||||
|
||||
loop {
|
||||
let ev = tokio::select! {
|
||||
_ = cancel.cancelled() => return Ok(()),
|
||||
ev = events.next() => ev,
|
||||
};
|
||||
let Some(ev) = ev else { break };
|
||||
let ev = ev.map_err(|e| anyhow!("stream error: {e}"))?;
|
||||
if ev.data.is_empty() {
|
||||
continue;
|
||||
}
|
||||
let data: Value = serde_json::from_str(&ev.data)?;
|
||||
match data["type"].as_str().unwrap_or_default() {
|
||||
"message_start" => {
|
||||
let u = &data["message"]["usage"];
|
||||
usage.input_tokens = u["input_tokens"].as_u64().unwrap_or(0);
|
||||
usage.cache_read_tokens = u["cache_read_input_tokens"].as_u64().unwrap_or(0);
|
||||
usage.cache_write_tokens = u["cache_creation_input_tokens"].as_u64().unwrap_or(0);
|
||||
}
|
||||
"content_block_start" => {
|
||||
let idx = data["index"].as_u64().unwrap_or(0);
|
||||
let cb = &data["content_block"];
|
||||
let mut st = BlockState {
|
||||
kind: cb["type"].as_str().unwrap_or_default().to_string(),
|
||||
..Default::default()
|
||||
};
|
||||
match st.kind.as_str() {
|
||||
"tool_use" => {
|
||||
st.id = cb["id"].as_str().unwrap_or_default().into();
|
||||
st.name = cb["name"].as_str().unwrap_or_default().into();
|
||||
}
|
||||
"redacted_thinking" => {
|
||||
st.signature = format!("redacted:{}", cb["data"].as_str().unwrap_or_default());
|
||||
}
|
||||
"text" => {
|
||||
if let Some(t) = cb["text"].as_str().filter(|t| !t.is_empty()) {
|
||||
tx.send(StreamEvent::TextDelta(t.into())).await.ok();
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
blocks.insert(idx, st);
|
||||
}
|
||||
"content_block_delta" => {
|
||||
let idx = data["index"].as_u64().unwrap_or(0);
|
||||
let d = &data["delta"];
|
||||
let st = blocks.entry(idx).or_default();
|
||||
match d["type"].as_str().unwrap_or_default() {
|
||||
"text_delta" => {
|
||||
tx.send(StreamEvent::TextDelta(d["text"].as_str().unwrap_or_default().into())).await.ok();
|
||||
}
|
||||
"thinking_delta" => {
|
||||
tx.send(StreamEvent::ReasoningDelta(d["thinking"].as_str().unwrap_or_default().into())).await.ok();
|
||||
}
|
||||
"signature_delta" => st.signature.push_str(d["signature"].as_str().unwrap_or_default()),
|
||||
"input_json_delta" => st.json.push_str(d["partial_json"].as_str().unwrap_or_default()),
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
"content_block_stop" => {
|
||||
let idx = data["index"].as_u64().unwrap_or(0);
|
||||
if let Some(st) = blocks.remove(&idx) {
|
||||
match st.kind.as_str() {
|
||||
"tool_use" => {
|
||||
let input = if st.json.trim().is_empty() {
|
||||
json!({})
|
||||
} else {
|
||||
// Invalid JSON (e.g. truncated) is passed on as a marker; the
|
||||
// agent loop answers it with an error tool_result.
|
||||
serde_json::from_str(&st.json)
|
||||
.unwrap_or_else(|_| json!({"__invalid_json": st.json}))
|
||||
};
|
||||
tx.send(StreamEvent::ToolCall { id: st.id, name: st.name, input, extra: None }).await.ok();
|
||||
}
|
||||
"thinking" | "redacted_thinking" => {
|
||||
tx.send(StreamEvent::ReasoningSignature(st.signature)).await.ok();
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
"message_delta" => {
|
||||
if let Some(r) = data["delta"]["stop_reason"].as_str() {
|
||||
stop_reason = r.to_string();
|
||||
}
|
||||
if let Some(o) = data["usage"]["output_tokens"].as_u64() {
|
||||
usage.output_tokens = o;
|
||||
}
|
||||
}
|
||||
"error" => {
|
||||
let msg = data["error"]["message"].as_str().unwrap_or("unknown error");
|
||||
return Err(anyhow!("provider error: {msg}"));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
tx.send(StreamEvent::Usage(usage)).await.ok();
|
||||
tx.send(StreamEvent::Done { stop_reason }).await.ok();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list_models(&self) -> Result<Vec<ModelConfig>> {
|
||||
let resp = self.request(reqwest::Method::GET, "/models?limit=100").send().await?;
|
||||
if !resp.status().is_success() {
|
||||
return Err(http_error(resp).await);
|
||||
}
|
||||
let v: Value = resp.json().await?;
|
||||
Ok(v["data"]
|
||||
.as_array()
|
||||
.map(|a| {
|
||||
a.iter()
|
||||
.map(|m| ModelConfig {
|
||||
id: m["id"].as_str().unwrap_or_default().into(),
|
||||
name: m["display_name"].as_str().map(String::from),
|
||||
context: m["max_input_tokens"].as_u64(),
|
||||
max_output: m["max_tokens"].as_u64(),
|
||||
reasoning: true,
|
||||
vision: true,
|
||||
..Default::default()
|
||||
})
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,209 @@
|
||||
//! Model providers. Two wire protocols cover all targets:
|
||||
//! Anthropic Messages and OpenAI-compatible chat completions.
|
||||
|
||||
pub mod anthropic;
|
||||
pub mod openai;
|
||||
|
||||
use crate::config::{Config, ModelConfig, ProviderConfig, ProviderKind, expand_env};
|
||||
use crate::types::{Message, Usage};
|
||||
use anyhow::{Result, anyhow};
|
||||
use async_trait::async_trait;
|
||||
use serde::Serialize;
|
||||
use serde_json::Value;
|
||||
use std::collections::BTreeMap;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct ToolDef {
|
||||
pub name: String,
|
||||
pub description: String,
|
||||
pub input_schema: Value,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ChatRequest {
|
||||
pub model: ModelConfig,
|
||||
pub system: String,
|
||||
pub messages: Vec<Message>,
|
||||
pub tools: Vec<ToolDef>,
|
||||
pub max_tokens: u64,
|
||||
/// Enable extended thinking / reasoning when the model supports it.
|
||||
pub reasoning: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum StreamEvent {
|
||||
TextDelta(String),
|
||||
ReasoningDelta(String),
|
||||
ReasoningSignature(String),
|
||||
ToolCall { id: String, name: String, input: Value, extra: Option<Value> },
|
||||
Usage(Usage),
|
||||
Done { stop_reason: String },
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
pub trait Provider: Send + Sync {
|
||||
async fn stream(
|
||||
&self,
|
||||
req: ChatRequest,
|
||||
tx: mpsc::Sender<StreamEvent>,
|
||||
cancel: CancellationToken,
|
||||
) -> Result<()>;
|
||||
|
||||
async fn list_models(&self) -> Result<Vec<ModelConfig>>;
|
||||
}
|
||||
|
||||
/// Connection settings shared by both protocol implementations.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Endpoint {
|
||||
pub base_url: String,
|
||||
pub api_key: String,
|
||||
pub auth_header: String,
|
||||
pub auth_prefix: String,
|
||||
pub headers: BTreeMap<String, String>,
|
||||
}
|
||||
|
||||
impl Endpoint {
|
||||
fn from_config(p: &ProviderConfig) -> Self {
|
||||
let (default_header, default_prefix) = match p.kind {
|
||||
ProviderKind::Anthropic => ("x-api-key", ""),
|
||||
ProviderKind::Openai => ("Authorization", "Bearer "),
|
||||
};
|
||||
let base = if p.base_url.is_empty() && p.kind == ProviderKind::Anthropic {
|
||||
"https://api.anthropic.com/v1".to_string()
|
||||
} else {
|
||||
expand_env(&p.base_url)
|
||||
};
|
||||
Self {
|
||||
base_url: base.trim_end_matches('/').to_string(),
|
||||
api_key: expand_env(&p.api_key),
|
||||
auth_header: p.auth_header.clone().unwrap_or_else(|| default_header.into()),
|
||||
auth_prefix: p.auth_prefix.clone().unwrap_or_else(|| default_prefix.into()),
|
||||
headers: p.headers.iter().map(|(k, v)| (k.clone(), expand_env(v))).collect(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn apply(&self, mut rb: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
|
||||
if !self.api_key.is_empty() {
|
||||
rb = rb.header(&self.auth_header, format!("{}{}", self.auth_prefix, self.api_key));
|
||||
}
|
||||
for (k, v) in &self.headers {
|
||||
rb = rb.header(k, v);
|
||||
}
|
||||
rb
|
||||
}
|
||||
}
|
||||
|
||||
pub struct ProviderEntry {
|
||||
pub name: String,
|
||||
pub provider: Arc<dyn Provider>,
|
||||
pub models: Vec<ModelConfig>,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub struct Registry {
|
||||
pub providers: BTreeMap<String, ProviderEntry>,
|
||||
}
|
||||
|
||||
pub fn http_client() -> reqwest::Client {
|
||||
reqwest::Client::builder()
|
||||
.connect_timeout(std::time::Duration::from_secs(20))
|
||||
.user_agent(concat!("codecraft/", env!("CARGO_PKG_VERSION")))
|
||||
.build()
|
||||
.expect("http client")
|
||||
}
|
||||
|
||||
impl Registry {
|
||||
/// Build providers from config. Providers with an empty model list are queried
|
||||
/// for their models; failures are logged and leave the list empty.
|
||||
pub async fn build(cfg: &Config) -> Self {
|
||||
let client = http_client();
|
||||
let mut providers = BTreeMap::new();
|
||||
for (name, pc) in &cfg.providers {
|
||||
if !pc.enabled {
|
||||
continue;
|
||||
}
|
||||
let ep = Endpoint::from_config(pc);
|
||||
let provider: Arc<dyn Provider> = match pc.kind {
|
||||
ProviderKind::Anthropic => Arc::new(anthropic::Anthropic::new(client.clone(), ep)),
|
||||
ProviderKind::Openai => Arc::new(openai::OpenAi::new(client.clone(), ep)),
|
||||
};
|
||||
let mut models: Vec<ModelConfig> = pc.models.iter().map(|m| m.resolved()).collect();
|
||||
if models.is_empty() {
|
||||
match tokio::time::timeout(std::time::Duration::from_secs(15), provider.list_models()).await {
|
||||
Ok(Ok(m)) => models = m,
|
||||
Ok(Err(e)) => tracing::warn!("provider {name}: model discovery failed: {e:#}"),
|
||||
Err(_) => tracing::warn!("provider {name}: model discovery timed out"),
|
||||
}
|
||||
}
|
||||
providers.insert(
|
||||
name.clone(),
|
||||
ProviderEntry { name: name.clone(), provider, models },
|
||||
);
|
||||
}
|
||||
Self { providers }
|
||||
}
|
||||
|
||||
/// Resolve "provider/model". Unknown models of a known provider are allowed
|
||||
/// with default settings so users can type any model id.
|
||||
pub fn resolve(&self, spec: &str) -> Result<(Arc<dyn Provider>, ModelConfig)> {
|
||||
let (pname, mid) = spec
|
||||
.split_once('/')
|
||||
.ok_or_else(|| anyhow!("model must be 'provider/model', got '{spec}'"))?;
|
||||
let entry = self
|
||||
.providers
|
||||
.get(pname)
|
||||
.ok_or_else(|| anyhow!("unknown or disabled provider '{pname}'"))?;
|
||||
let model = entry
|
||||
.models
|
||||
.iter()
|
||||
.find(|m| m.id == mid)
|
||||
.cloned()
|
||||
.unwrap_or_else(|| ModelConfig { id: mid.to_string(), ..Default::default() });
|
||||
Ok((entry.provider.clone(), model))
|
||||
}
|
||||
|
||||
pub fn first_model(&self) -> Option<String> {
|
||||
self.providers
|
||||
.values()
|
||||
.find_map(|e| e.models.first().map(|m| format!("{}/{}", e.name, m.id)))
|
||||
}
|
||||
|
||||
pub fn models_json(&self) -> Value {
|
||||
let list: Vec<Value> = self
|
||||
.providers
|
||||
.values()
|
||||
.map(|e| {
|
||||
serde_json::json!({
|
||||
"provider": e.name,
|
||||
"models": e.models.iter().map(|m| serde_json::json!({
|
||||
"id": m.id,
|
||||
"name": m.name.clone().unwrap_or_else(|| m.id.clone()),
|
||||
"context": m.context,
|
||||
"reasoning": m.reasoning,
|
||||
"vision": m.vision,
|
||||
})).collect::<Vec<_>>()
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
Value::Array(list)
|
||||
}
|
||||
}
|
||||
|
||||
/// Turn an HTTP error response into a readable error.
|
||||
pub async fn http_error(resp: reqwest::Response) -> anyhow::Error {
|
||||
let status = resp.status();
|
||||
let body = resp.text().await.unwrap_or_default();
|
||||
let detail = serde_json::from_str::<Value>(&body)
|
||||
.ok()
|
||||
.and_then(|v| {
|
||||
v.pointer("/error/message")
|
||||
.or_else(|| v.pointer("/0/error/message"))
|
||||
.or_else(|| v.get("message"))
|
||||
.and_then(|m| m.as_str().map(String::from))
|
||||
})
|
||||
.unwrap_or_else(|| body.chars().take(500).collect());
|
||||
anyhow!("provider returned HTTP {status}: {detail}")
|
||||
}
|
||||
@@ -0,0 +1,251 @@
|
||||
//! OpenAI-compatible chat completions (streaming). Covers Gemini (OpenAI compat),
|
||||
//! opencode Go, fal/OpenRouter, Ollama, vLLM, llama.cpp, LM Studio, ...
|
||||
|
||||
use super::{ChatRequest, Endpoint, Provider, StreamEvent, http_error};
|
||||
use crate::config::ModelConfig;
|
||||
use crate::types::{Message, Part, Role, Usage};
|
||||
use anyhow::{Result, anyhow};
|
||||
use async_trait::async_trait;
|
||||
use eventsource_stream::Eventsource;
|
||||
use futures::StreamExt;
|
||||
use serde_json::{Value, json};
|
||||
use std::collections::BTreeMap;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
pub struct OpenAi {
|
||||
client: reqwest::Client,
|
||||
ep: Endpoint,
|
||||
}
|
||||
|
||||
impl OpenAi {
|
||||
pub fn new(client: reqwest::Client, ep: Endpoint) -> Self {
|
||||
Self { client, ep }
|
||||
}
|
||||
}
|
||||
|
||||
fn data_url(img: &crate::types::ImageData) -> String {
|
||||
format!("data:{};base64,{}", img.media_type, img.data)
|
||||
}
|
||||
|
||||
pub fn to_wire_messages(system: &str, messages: &[Message]) -> Vec<Value> {
|
||||
let mut out = vec![json!({"role": "system", "content": system})];
|
||||
for m in messages {
|
||||
match m.role {
|
||||
Role::User => {
|
||||
let mut parts = Vec::new();
|
||||
let mut tool_images = Vec::new();
|
||||
for p in &m.parts {
|
||||
match p {
|
||||
Part::Text { text } if !text.is_empty() => {
|
||||
parts.push(json!({"type": "text", "text": text}))
|
||||
}
|
||||
Part::Summary { text } => parts.push(json!({
|
||||
"type": "text",
|
||||
"text": format!("<conversation-summary>\n{text}\n</conversation-summary>")
|
||||
})),
|
||||
Part::Image { image } => parts.push(json!({
|
||||
"type": "image_url", "image_url": {"url": data_url(image)}
|
||||
})),
|
||||
Part::ToolResult { call_id, name, content, images, is_error, .. } => {
|
||||
let text = if *is_error { format!("ERROR: {content}") } else { content.clone() };
|
||||
out.push(json!({
|
||||
"role": "tool", "tool_call_id": call_id, "name": name,
|
||||
"content": if text.is_empty() { "(no output)".into() } else { text }
|
||||
}));
|
||||
tool_images.extend(images.iter().cloned());
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
// Tool messages cannot carry images: send them as a user message.
|
||||
if !tool_images.is_empty() {
|
||||
let mut c = vec![json!({"type": "text", "text": "Images returned by the tool calls above:"})];
|
||||
c.extend(tool_images.iter().map(|i| json!({"type": "image_url", "image_url": {"url": data_url(i)}})));
|
||||
out.push(json!({"role": "user", "content": c}));
|
||||
}
|
||||
if !parts.is_empty() {
|
||||
out.push(json!({"role": "user", "content": parts}));
|
||||
}
|
||||
}
|
||||
Role::Assistant => {
|
||||
let mut text = String::new();
|
||||
let mut reasoning = String::new();
|
||||
let mut calls = Vec::new();
|
||||
for p in &m.parts {
|
||||
match p {
|
||||
Part::Text { text: t } => text.push_str(t),
|
||||
Part::Reasoning { text: t, .. } => reasoning.push_str(t),
|
||||
Part::ToolCall { id, name, input, extra } => {
|
||||
let mut c = json!({
|
||||
"id": id, "type": "function",
|
||||
"function": {"name": name, "arguments": input.to_string()}
|
||||
});
|
||||
if let Some(e) = extra {
|
||||
c["extra_content"] = e.clone();
|
||||
}
|
||||
calls.push(c);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
let mut msg = json!({"role": "assistant", "content": if text.is_empty() { Value::Null } else { Value::String(text) }});
|
||||
if !calls.is_empty() {
|
||||
msg["tool_calls"] = Value::Array(calls);
|
||||
}
|
||||
// Some reasoning models (DeepSeek, Kimi) expect their reasoning back during tool loops.
|
||||
if !reasoning.is_empty() {
|
||||
msg["reasoning_content"] = Value::String(reasoning);
|
||||
}
|
||||
if msg["content"].is_null() && msg.get("tool_calls").is_none() {
|
||||
continue;
|
||||
}
|
||||
out.push(msg);
|
||||
}
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct PendingCall {
|
||||
id: String,
|
||||
name: String,
|
||||
args: String,
|
||||
extra: Option<Value>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Provider for OpenAi {
|
||||
async fn stream(
|
||||
&self,
|
||||
req: ChatRequest,
|
||||
tx: mpsc::Sender<StreamEvent>,
|
||||
cancel: CancellationToken,
|
||||
) -> Result<()> {
|
||||
let tools: Vec<Value> = req
|
||||
.tools
|
||||
.iter()
|
||||
.map(|t| json!({
|
||||
"type": "function",
|
||||
"function": {"name": t.name, "description": t.description, "parameters": t.input_schema}
|
||||
}))
|
||||
.collect();
|
||||
let mut body = json!({
|
||||
"model": req.model.id,
|
||||
"messages": to_wire_messages(&req.system, &req.messages),
|
||||
"stream": true,
|
||||
"stream_options": {"include_usage": true},
|
||||
"max_tokens": req.max_tokens,
|
||||
});
|
||||
if !tools.is_empty() {
|
||||
body["tools"] = Value::Array(tools);
|
||||
}
|
||||
if let Some(effort) = &req.model.effort {
|
||||
body["reasoning_effort"] = json!(effort);
|
||||
}
|
||||
|
||||
let rb = self.client.post(format!("{}/chat/completions", self.ep.base_url)).json(&body);
|
||||
let resp = self.ep.apply(rb).send().await?;
|
||||
if !resp.status().is_success() {
|
||||
return Err(http_error(resp).await);
|
||||
}
|
||||
|
||||
let mut events = resp.bytes_stream().eventsource();
|
||||
let mut calls: BTreeMap<u64, PendingCall> = BTreeMap::new();
|
||||
let mut usage = Usage::default();
|
||||
let mut stop_reason = String::from("stop");
|
||||
|
||||
loop {
|
||||
let ev = tokio::select! {
|
||||
_ = cancel.cancelled() => return Ok(()),
|
||||
ev = events.next() => ev,
|
||||
};
|
||||
let Some(ev) = ev else { break };
|
||||
let ev = ev.map_err(|e| anyhow!("stream error: {e}"))?;
|
||||
if ev.data.trim() == "[DONE]" {
|
||||
break;
|
||||
}
|
||||
if ev.data.trim().is_empty() {
|
||||
continue;
|
||||
}
|
||||
let data: Value = serde_json::from_str(&ev.data)?;
|
||||
if let Some(err) = data.get("error") {
|
||||
return Err(anyhow!("provider error: {}", err["message"].as_str().unwrap_or(&err.to_string())));
|
||||
}
|
||||
if let Some(u) = data.get("usage").filter(|u| u.is_object()) {
|
||||
let cached = u["prompt_tokens_details"]["cached_tokens"].as_u64().unwrap_or(0);
|
||||
usage.input_tokens = u["prompt_tokens"].as_u64().unwrap_or(0).saturating_sub(cached);
|
||||
usage.cache_read_tokens = cached;
|
||||
usage.output_tokens = u["completion_tokens"].as_u64().unwrap_or(0);
|
||||
}
|
||||
let Some(choice) = data["choices"].get(0) else { continue };
|
||||
let d = &choice["delta"];
|
||||
if let Some(t) = d["content"].as_str().filter(|t| !t.is_empty()) {
|
||||
tx.send(StreamEvent::TextDelta(t.into())).await.ok();
|
||||
}
|
||||
let r = d["reasoning_content"].as_str().or_else(|| d["reasoning"].as_str());
|
||||
if let Some(r) = r.filter(|r| !r.is_empty()) {
|
||||
tx.send(StreamEvent::ReasoningDelta(r.into())).await.ok();
|
||||
}
|
||||
if let Some(tcs) = d["tool_calls"].as_array() {
|
||||
for (pos, tc) in tcs.iter().enumerate() {
|
||||
let idx = tc["index"].as_u64().unwrap_or(pos as u64);
|
||||
let pc = calls.entry(idx).or_default();
|
||||
if let Some(id) = tc["id"].as_str().filter(|s| !s.is_empty()) {
|
||||
pc.id = id.into();
|
||||
}
|
||||
if let Some(n) = tc["function"]["name"].as_str() {
|
||||
pc.name.push_str(n);
|
||||
}
|
||||
if let Some(a) = tc["function"]["arguments"].as_str() {
|
||||
pc.args.push_str(a);
|
||||
}
|
||||
if let Some(e) = tc.get("extra_content") {
|
||||
pc.extra = Some(e.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
if let Some(f) = choice["finish_reason"].as_str() {
|
||||
stop_reason = f.to_string();
|
||||
}
|
||||
}
|
||||
|
||||
for (i, pc) in calls {
|
||||
let input = if pc.args.trim().is_empty() {
|
||||
json!({})
|
||||
} else {
|
||||
serde_json::from_str(&pc.args).unwrap_or_else(|_| json!({"__invalid_json": pc.args}))
|
||||
};
|
||||
let id = if pc.id.is_empty() { format!("call_{}_{i}", crate::types::new_id()) } else { pc.id };
|
||||
tx.send(StreamEvent::ToolCall { id, name: pc.name, input, extra: pc.extra }).await.ok();
|
||||
}
|
||||
tx.send(StreamEvent::Usage(usage)).await.ok();
|
||||
tx.send(StreamEvent::Done { stop_reason }).await.ok();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list_models(&self) -> Result<Vec<ModelConfig>> {
|
||||
let rb = self.client.get(format!("{}/models", self.ep.base_url));
|
||||
let resp = self.ep.apply(rb).send().await?;
|
||||
if !resp.status().is_success() {
|
||||
return Err(http_error(resp).await);
|
||||
}
|
||||
let v: Value = resp.json().await?;
|
||||
let mut models: Vec<ModelConfig> = v["data"]
|
||||
.as_array()
|
||||
.map(|a| {
|
||||
a.iter()
|
||||
.filter_map(|m| m["id"].as_str())
|
||||
// Gemini lists "models/<id>"; the chat endpoint wants the bare id.
|
||||
.map(|id| ModelConfig {
|
||||
id: id.trim_start_matches("models/").to_string(),
|
||||
..Default::default()
|
||||
})
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default();
|
||||
models.sort_by(|a, b| a.id.cmp(&b.id));
|
||||
Ok(models)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user