zink/codecraft
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 26 Wiki Activity

Add API tokens, CLI server API and shared core crate

Browse Source 
- Personal API tokens (full/read scope, expiry, revoke) and CLI password login
- Cargo workspace: codecraft-core (types, providers, permission rules) shared by
  server and the upcoming terminal client; server re-exports the old paths
- CLI server API: /cli/config, /cli/skill, /cli/tool (server MCP tools),
  /llm/stream model gateway with budgets and usage per user
- Theme/language preferences mirrored to the server; personal memory endpoint
- Statistics: provider and mode tables stacked; CLI usage labelled
- cli/: work in progress (terminal client is being rewritten as a local TUI)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Richard ZinkandClaude Opus 5.5 committed 2026-11-08 23:18:50 +01:00
1 parent fca1c41087
commit aa1f8cf016
40 files changed
+2454 -814
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/provider/anthropic.rs
-304
View File
@@ -1,304 +0,0 @@
//! 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});
}
for (k, v) in &req.model.options {
body[k] = v.clone();
}
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())
}
}
server/src/provider/mod.rs
+5 -118
View File
@@ -1,100 +1,13 @@
//! Model providers. Two wire protocols cover all targets:
//! Anthropic Messages and OpenAI-compatible chat completions.
//! Model providers (implementations in codecraft-core) and the registry of
//! configured providers with their concurrency limits.
pub mod anthropic;
pub mod openai;
pub use codecraft_core::provider::*;
use crate::config::{Config, ModelConfig, ProviderConfig, ProviderKind, expand_env};
use crate::types::{Message, Usage};
use crate::config::{Config, ModelConfig};
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,
@@ -114,13 +27,6 @@ pub struct Slot {
_permits: Vec<tokio::sync::OwnedSemaphorePermit>,
}
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
@@ -133,11 +39,7 @@ impl Registry {
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 provider: Arc<dyn Provider> = build(client.clone(), pc);
let mut models: Vec<ModelConfig> = pc.models.iter().map(|m| m.resolved()).collect();
if let Some(n) = pc.max_concurrency.filter(|n| *n > 0) {
limits.insert(name.clone(), Arc::new(tokio::sync::Semaphore::new(n)));
@@ -236,18 +138,3 @@ impl Registry {
}
}
/// 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}")
}
server/src/provider/openai.rs
-343
View File
@@ -1,343 +0,0 @@
//! 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
}
/// Splits streamed content into text and inline `<think>…</think>` reasoning.
/// Tags may be split across chunks, so a possible partial tag is held back.
#[derive(Default)]
pub struct ThinkSplitter {
in_think: bool,
pending: String,
}
const OPEN: &str = "<think>";
const CLOSE: &str = "</think>";
impl ThinkSplitter {
/// Returns (is_reasoning, text) pieces that are safe to emit.
pub fn push(&mut self, chunk: &str) -> Vec<(bool, String)> {
self.pending.push_str(chunk);
let mut out = Vec::new();
loop {
let tag = if self.in_think { CLOSE } else { OPEN };
if let Some(pos) = self.pending.find(tag) {
let before: String = self.pending[..pos].to_string();
if !before.is_empty() {
out.push((self.in_think, before));
}
self.pending.drain(..pos + tag.len());
self.in_think = !self.in_think;
continue;
}
// Keep a trailing prefix of the tag (e.g. "<thi") until the next chunk.
let keep = (1..tag.len()).rev().find(|n| self.pending.ends_with(&tag[..*n])).unwrap_or(0);
let emit_len = self.pending.len() - keep;
if emit_len > 0 {
let text: String = self.pending.drain(..emit_len).collect();
out.push((self.in_think, text));
}
break;
}
// Leading whitespace after </think> is noise.
out.retain(|(_, t)| !t.is_empty());
out
}
pub fn finish(&mut self) -> Vec<(bool, String)> {
if self.pending.is_empty() {
return vec![];
}
vec![(self.in_think, std::mem::take(&mut self.pending))]
}
}
#[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);
}
for (k, v) in &req.model.options {
body[k] = v.clone();
}
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 think = ThinkSplitter::default();
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()) {
// Models served by llama.cpp & co. may inline their reasoning as <think>…</think>.
for (is_reasoning, chunk) in think.push(t) {
let ev = if is_reasoning { StreamEvent::ReasoningDelta(chunk) } else { StreamEvent::TextDelta(chunk) };
tx.send(ev).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 (is_reasoning, chunk) in think.finish() {
let ev = if is_reasoning { StreamEvent::ReasoningDelta(chunk) } else { StreamEvent::TextDelta(chunk) };
tx.send(ev).await.ok();
}
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)
}
}
#[cfg(test)]
mod tests {
use super::ThinkSplitter;
fn run(chunks: &[&str]) -> (String, String) {
let mut s = ThinkSplitter::default();
let mut text = String::new();
let mut reasoning = String::new();
let mut parts: Vec<(bool, String)> = chunks.iter().flat_map(|c| s.push(c)).collect();
parts.extend(s.finish());
for (r, t) in parts {
if r { reasoning.push_str(&t) } else { text.push_str(&t) }
}
(text, reasoning)
}
#[test]
fn splits_think_tags_across_chunks() {
let (text, reasoning) = run(&["<thi", "nk>plan ", "the fix</th", "ink>Here is ", "the answer."]);
assert_eq!(reasoning, "plan the fix");
assert_eq!(text, "Here is the answer.");
}
#[test]
fn plain_text_and_lookalikes_pass_through() {
let (text, reasoning) = run(&["a < b and <thin", "g> stays"]);
assert_eq!(text, "a < b and <thing> stays");
assert!(reasoning.is_empty());
}
}