diff --git a/Cargo.lock b/Cargo.lock index 1ca477a..8356839 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -599,6 +599,7 @@ checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" name = "core-api" version = "0.1.0" dependencies = [ + "agent-loop", "anyhow", "async-trait", "axum", @@ -2198,21 +2199,6 @@ version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" -[[package]] -name = "llm-client" -version = "0.1.0" -dependencies = [ - "anyhow", - "async-trait", - "core-api", - "futures-util", - "reqwest 0.13.4", - "serde", - "serde_json", - "tokio", - "tracing", -] - [[package]] name = "lock_api" version = "0.4.14" @@ -4198,7 +4184,6 @@ dependencies = [ "futures", "honcho-client", "indexmap 2.14.0", - "llm-client", "mcp-client", "notify", "plugin-comfyui", @@ -4232,6 +4217,7 @@ name = "skald-core" version = "0.1.0" dependencies = [ "aes-gcm", + "agent-loop", "anyhow", "argon2", "async-trait", @@ -4248,7 +4234,6 @@ dependencies = [ "indexmap 2.14.0", "libc", "libsqlite3-sys", - "llm-client", "mcp-client", "notify", "os_info", diff --git a/Cargo.toml b/Cargo.toml index a0325c4..9f01d2c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,6 @@ members = [ "crates/skald-core", "crates/skald-setup", "crates/honcho-client", - "crates/llm-client", "crates/core-api", "crates/mcp-client", "crates/plugin-tailscale-remote", @@ -74,7 +73,6 @@ tracing-appender = "0.2" chrono = { version = "0.4", default-features = false, features = ["clock", "std"] } notify = "8" honcho-client = { path = "crates/honcho-client" } -llm-client = { path = "crates/llm-client" } core-api = { path = "crates/core-api" } mcp-client = { path = "crates/mcp-client" } plugin-tailscale-remote = { path = "crates/plugin-tailscale-remote" } diff --git a/crates/agent-loop/src/kernel.rs b/crates/agent-loop/src/kernel.rs index e820674..87c6a3c 100644 --- a/crates/agent-loop/src/kernel.rs +++ b/crates/agent-loop/src/kernel.rs @@ -156,6 +156,7 @@ pub(crate) async fn run( conversation: params.conversation.clone(), frame, extras: handle.info.extras.clone(), + log: None, }; let result = tokio::select! { biased; diff --git a/crates/agent-loop/src/model.rs b/crates/agent-loop/src/model.rs index 004b3d7..89e2b34 100644 --- a/crates/agent-loop/src/model.rs +++ b/crates/agent-loop/src/model.rs @@ -198,8 +198,12 @@ pub struct ModelRequest { pub conversation: ConversationId, pub frame: FrameId, /// Host free-form per-request extras (e.g. reasoning knobs resolved for - /// this model). Merged last by the shipped clients. + /// this model). Merged last by the shipped clients INTO THE REQUEST BODY. pub extras: Value, + /// Host logging/telemetry correlation (session ids, user id, …). + /// **Never** merged into the request body by the shipped clients — it + /// exists for host decorators (e.g. a `LoggingModel`) only. + pub log: Option, } // ── Model ──────────────────────────────────────────────────────────────────── diff --git a/crates/core-api/Cargo.toml b/crates/core-api/Cargo.toml index 3ab5ee2..7470896 100644 --- a/crates/core-api/Cargo.toml +++ b/crates/core-api/Cargo.toml @@ -4,6 +4,7 @@ version = "0.1.0" edition = "2024" [dependencies] +agent-loop = { path = "../agent-loop" } serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["sync", "macros"] } diff --git a/crates/core-api/src/chatbot.rs b/crates/core-api/src/chatbot.rs deleted file mode 100644 index 3e2f0eb..0000000 --- a/crates/core-api/src/chatbot.rs +++ /dev/null @@ -1,193 +0,0 @@ -use async_trait::async_trait; -use serde_json::Value; -use tokio::sync::mpsc; - -/// A single message in a conversation. -#[derive(Debug, Clone)] -pub struct Message { - pub role: Role, - pub content: String, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum Role { - System, - User, - Assistant, -} - -impl Message { - pub fn system(content: impl Into) -> Self { - Self { role: Role::System, content: content.into() } - } - - pub fn user(content: impl Into) -> Self { - Self { role: Role::User, content: content.into() } - } - - pub fn assistant(content: impl Into) -> Self { - Self { role: Role::Assistant, content: content.into() } - } -} - -/// Options for a single chat completion request. -#[derive(Debug, Clone)] -pub struct ChatOptions { - pub model: String, - pub max_tokens: Option, - pub temperature: Option, - /// Session/stack IDs for request logging. Set by the LLM loop; ignored by - /// providers — only the logging wrapper reads them. - pub session_id: Option, - pub stack_id: Option, - /// The authenticated user driving this request. Correlates the metadata row - /// in `system.db` with the payload in `{userid}.db`. Logging-only. - pub user_id: Option, - /// UUID correlating the metadata row (`llm_requests`) with the payload row - /// (`llm_request_payloads`). Generated by the LLM loop before the call. - /// Logging-only. - pub request_id: Option, -} - -/// Raw HTTP metadata captured during a provider call. -/// Sensitive header values (api_key) are redacted before storage. -#[derive(Debug, Default)] -pub struct LlmRawMeta { - pub request_headers: Option, - pub request_body: Option, - pub response_headers: Option, - pub response_body: Option, -} - -/// The response from a chat completion (text only). -#[derive(Debug, Clone)] -pub struct ChatResponse { - pub content: String, - pub input_tokens: Option, - pub output_tokens: Option, - /// True when the model stopped due to hitting the token limit. - pub truncated: bool, - /// Chain-of-thought produced by reasoning models (e.g. DeepSeek thinking mode). - /// Must be echoed back in the assistant message on subsequent turns. - pub reasoning_content: Option, - /// Tokens served from the provider's prompt cache (Anthropic: cache_read_input_tokens, - /// OpenAI: prompt_tokens_details.cached_tokens). None when the provider does not - /// report cache metrics. - pub cache_read_tokens: Option, - /// Tokens written into the provider's prompt cache (Anthropic only: - /// cache_creation_input_tokens). None for providers that do not expose this. - pub cache_creation_tokens: Option, - /// Cost of the request in USD, when the provider reports it (OpenRouter - /// returns it under `usage.cost`). None for providers that do not bill - /// per-request or do not expose the figure. - pub cost: Option, -} - -/// A single tool call requested by the LLM. -#[derive(Debug, Clone)] -pub struct ToolCall { - pub id: String, - pub name: String, - pub arguments: Value, -} - -/// An incremental piece of a streaming completion, pushed by providers that -/// support SSE streaming. Purely best-effort UI feedback: the final `LlmTurn` -/// remains the authoritative result. -#[derive(Debug, Clone)] -pub enum StreamDelta { - /// Visible answer text. - Text(String), - /// Chain-of-thought / reasoning tokens (thinking models). - Reasoning(String), -} - -/// Result of one LLM turn when tools are available. -#[derive(Debug)] -pub enum LlmTurn { - Message(ChatResponse), - ToolCalls { - content: String, - calls: Vec, - input_tokens: Option, - output_tokens: Option, - reasoning_content: Option, - cache_read_tokens: Option, - cache_creation_tokens: Option, - cost: Option, - }, -} - -/// Stateless LLM client. Implementations hold only connection config (base URL, -/// API key). No memory, no database, no session state. -#[async_trait] -pub trait ChatbotClient: Send + Sync { - async fn chat( - &self, - messages: &[Message], - options: &ChatOptions, - ) -> anyhow::Result; - - /// Extracts the request cost in USD from a provider's raw JSON response, - /// when the provider reports it. OpenRouter (and other OpenAI-compatible - /// gateways) return it under `usage.cost`; the default reads that path and - /// yields None when absent. Providers with a different shape override this. - fn extract_cost(&self, response: &Value) -> Option { - response["usage"]["cost"].as_f64() - } - - /// Chat with tool support. Default implementation ignores tools and falls - /// back to `chat()`. - async fn chat_with_tools( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - ) -> anyhow::Result { - let simple: Vec = messages - .iter() - .filter_map(|m| { - let role = m["role"].as_str()?; - let content = m["content"].as_str().unwrap_or("").to_string(); - match role { - "system" => Some(Message::system(content)), - "user" => Some(Message::user(content)), - "assistant" => Some(Message::assistant(content)), - _ => None, - } - }) - .collect(); - let _ = tools; - let resp = self.chat(&simple, options).await?; - Ok(LlmTurn::Message(resp)) - } - - /// Like `chat_with_tools` but also returns raw HTTP metadata for logging. - /// Providers that make real HTTP calls should override this. - async fn chat_with_tools_raw( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - ) -> anyhow::Result<(LlmTurn, Option)> { - self.chat_with_tools(messages, tools, options).await.map(|t| (t, None)) - } - - /// Like `chat_with_tools_raw`, but the provider may push incremental - /// [`StreamDelta`]s into `delta_tx` as tokens arrive (SSE streaming). - /// Senders should use `try_send` and drop deltas when the channel is full — - /// streaming is best-effort UI feedback and must never backpressure the - /// HTTP read. The returned `LlmTurn` is always the complete, authoritative - /// result. The default ignores the channel and falls back to the buffered - /// call, so providers without streaming behave exactly as before. - async fn chat_with_tools_raw_streaming( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - delta_tx: mpsc::Sender, - ) -> anyhow::Result<(LlmTurn, Option)> { - let _ = delta_tx; - self.chat_with_tools_raw(messages, tools, options).await - } -} diff --git a/crates/core-api/src/lib.rs b/crates/core-api/src/lib.rs index b36a7af..bdd5bd7 100644 --- a/crates/core-api/src/lib.rs +++ b/crates/core-api/src/lib.rs @@ -1,11 +1,12 @@ /// Application name, sent as `X-Title` HTTP header to LLM/image/audio providers. -pub const APP_NAME: &str = "Skald"; +/// Lives in `agent-loop` (the LLM clients' home, blueprint D13); re-exported here +/// so existing users don't change. +pub use agent_loop::APP_NAME; pub mod approval; pub mod bus; pub mod config_api; pub mod system_bus; -pub mod chatbot; pub mod chat_hub; pub mod command; pub mod events; diff --git a/crates/core-api/src/provider.rs b/crates/core-api/src/provider.rs index 8deae53..76a8a4e 100644 --- a/crates/core-api/src/provider.rs +++ b/crates/core-api/src/provider.rs @@ -3,7 +3,8 @@ use std::sync::Arc; use anyhow::Result; use async_trait::async_trait; -use crate::chatbot::ChatbotClient; +use agent_loop::model::Model; + use crate::image_generate::{ImageGenerate, ImageGenerateModelRecord}; use crate::tts::{TextToSpeech, TtsModelRecord, RemoteTtsModelInfo}; use crate::transcribe::{Transcribe, TranscribeModelRecord, RemoteTranscribeModelInfo}; @@ -139,7 +140,8 @@ pub struct ProviderField { // ── BuiltLlmClient ──────────────────────────────────────────────────────────── pub struct BuiltLlmClient { - pub client: Arc, + /// A stateless `agent_loop` model client (blueprint D13). + pub client: Arc, pub prompt_cache: bool, } diff --git a/crates/llm-client/Cargo.toml b/crates/llm-client/Cargo.toml deleted file mode 100644 index 48640c8..0000000 --- a/crates/llm-client/Cargo.toml +++ /dev/null @@ -1,15 +0,0 @@ -[package] -name = "llm-client" -version = "0.1.0" -edition = "2024" - -[dependencies] -core-api = { path = "../core-api" } -reqwest = { version = "0.13", default-features = false, features = ["rustls-no-provider", "charset", "http2", "system-proxy", "json", "stream"] } -serde = { version = "1", features = ["derive"] } -serde_json = "1" -async-trait = "0.1" -anyhow = "1" -tracing = "0.1" -tokio = { version = "1", features = ["sync"] } -futures-util = "0.3" diff --git a/crates/llm-client/src/anthropic.rs b/crates/llm-client/src/anthropic.rs deleted file mode 100644 index b88abb9..0000000 --- a/crates/llm-client/src/anthropic.rs +++ /dev/null @@ -1,811 +0,0 @@ -use std::collections::BTreeMap; - -use async_trait::async_trait; -use futures_util::StreamExt; -use serde_json::{Value, json}; -use tokio::sync::mpsc; -use tracing::{debug, info, trace, warn}; - -use crate::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, Role, SseDecoder, StreamDelta, ToolCall, error_response_body, headers_to_json, redact_key}; - -const DEFAULT_BASE_URL: &str = "https://api.anthropic.com"; -const ANTHROPIC_VERSION: &str = "2023-06-01"; - -pub struct AnthropicClient { - base_url: String, - api_key: String, - /// Extra top-level request-body keys merged into every request (e.g. the - /// `thinking` config for extended reasoning). See `apply_extra`. - extra_body: Option, - http: reqwest::Client, -} - -impl AnthropicClient { - pub fn new(api_key: impl Into) -> Self { - Self::with_base_url(DEFAULT_BASE_URL, api_key) - } - - pub fn with_base_url(base_url: impl Into, api_key: impl Into) -> Self { - Self { - base_url: base_url.into(), - api_key: api_key.into(), - extra_body: None, - http: reqwest::Client::new(), - } - } - - /// Like `new` but with extra request-body keys (e.g. `{"thinking": {...}}`). - pub fn with_extra_body(api_key: impl Into, extra_body: Option) -> Self { - Self { - base_url: DEFAULT_BASE_URL.to_string(), - api_key: api_key.into(), - extra_body, - http: reqwest::Client::new(), - } - } - - /// Merges `extra_body` into `body` and enforces Anthropic's extended-thinking - /// constraints: when `thinking` is enabled, `temperature` is not allowed and - /// `max_tokens` must be strictly greater than `budget_tokens`. - fn apply_extra(&self, body: &mut Value) { - let Some(extra) = self.extra_body.as_ref().and_then(|v| v.as_object()) else { return }; - let Some(obj) = body.as_object_mut() else { return }; - for (k, v) in extra { - obj.insert(k.clone(), v.clone()); - } - if obj.get("thinking").map(|t| t["type"] == json!("enabled")).unwrap_or(false) { - obj.remove("temperature"); - let budget = obj["thinking"]["budget_tokens"].as_i64().unwrap_or(0); - let cur_max = obj.get("max_tokens").and_then(|v| v.as_i64()).unwrap_or(4096); - if budget > 0 && cur_max <= budget { - obj.insert("max_tokens".to_string(), json!(budget + 4096)); - } - } - } - - /// Converts OpenAI-format tool definitions to Anthropic format. - /// OpenAI: { "type": "function", "function": { "name", "description", "parameters" } } - /// Anthropic: { "name", "description", "input_schema" } - /// - /// DTL (tool search): a top-level `defer_loading: true` on the OpenAI tool - /// object is carried through to Anthropic's native `defer_loading` field. When - /// any tool is deferred, the cache breakpoint is placed on the last - /// **non-deferred** tool — a deferred tool cannot also carry `cache_control` - /// (the API 400s), and at least one tool must stay non-deferred anyway. - fn convert_tools(tools: &[Value]) -> Vec { - let has_deferred = tools.iter().any(|t| t["defer_loading"].as_bool() == Some(true)); - let mut out: Vec = tools - .iter() - .filter_map(|t| { - let func = &t["function"]; - let name = func["name"].as_str()?; - let mut tool = json!({ - "name": name, - "description": func["description"].as_str().unwrap_or(""), - "input_schema": func["parameters"], - }); - if t["defer_loading"].as_bool() == Some(true) { - tool["defer_loading"] = json!(true); - } - Some(tool) - }) - .collect(); - if has_deferred { - if let Some(t) = out.iter_mut().rev().find(|t| t["defer_loading"].as_bool() != Some(true)) { - t["cache_control"] = json!({ "type": "ephemeral" }); - } - } - out - } - - /// Converts OpenAI-format message array to Anthropic format. - /// - /// Key differences: - /// - System messages are skipped (extracted separately). - /// - Assistant messages with `tool_calls` become content arrays with `tool_use` blocks. - /// - `tool` role messages are grouped into `user` messages with `tool_result` blocks. - fn convert_messages(messages: &[Value]) -> Vec { - let mut out: Vec = Vec::new(); - let mut i = 0; - - while i < messages.len() { - let msg = &messages[i]; - let role = msg["role"].as_str().unwrap_or(""); - - match role { - "system" => { i += 1; } - - "user" => { - out.push(json!({ - "role": "user", - "content": convert_user_content(&msg["content"]), - })); - i += 1; - } - - "assistant" => { - if let Some(tool_calls) = msg["tool_calls"].as_array() { - let mut content: Vec = Vec::new(); - - let text = msg["content"].as_str().unwrap_or(""); - if !text.is_empty() { - content.push(json!({ "type": "text", "text": text })); - } - - for tc in tool_calls { - let id = tc["id"].as_str().unwrap_or(""); - let name = tc["function"]["name"].as_str().unwrap_or(""); - let args_str = tc["function"]["arguments"].as_str().unwrap_or("{}"); - let input: Value = serde_json::from_str(args_str) - .unwrap_or(Value::Object(Default::default())); - - content.push(json!({ - "type": "tool_use", - "id": id, - "name": name, - "input": input, - })); - } - - out.push(json!({ "role": "assistant", "content": content })); - } else { - out.push(json!({ - "role": "assistant", - "content": msg["content"].as_str().unwrap_or(""), - })); - } - i += 1; - } - - "tool" => { - // Group all consecutive tool-result messages into a single user message. - let mut results: Vec = Vec::new(); - while i < messages.len() && messages[i]["role"].as_str() == Some("tool") { - let tm = &messages[i]; - // DTL (custom tool search): a tool result carrying - // `_tool_references` (set by the message builder on an - // `activate_tools` result in AnthropicToolReference mode) becomes a - // `content` array of `tool_reference` blocks, which the API expands - // into the deferred tools' full definitions. Empty/absent → the - // normal text result. - let content: Value = match tm["_tool_references"].as_array() { - Some(refs) if !refs.is_empty() => Value::Array( - refs.iter() - .filter_map(|r| r.as_str()) - .map(|name| json!({ "type": "tool_reference", "tool_name": name })) - .collect(), - ), - _ => Value::String(tm["content"].as_str().unwrap_or("").to_string()), - }; - results.push(json!({ - "type": "tool_result", - "tool_use_id": tm["tool_call_id"].as_str().unwrap_or(""), - "content": content, - })); - i += 1; - } - out.push(json!({ "role": "user", "content": results })); - } - - _ => { i += 1; } - } - } - - out - } - - /// Assembles the `/v1/messages` request body shared by the buffered and the - /// streaming path (the caller adds `stream` on top). - fn tools_body(&self, system: Option, messages: Vec, tools: Vec, options: &ChatOptions) -> Value { - let max_tokens = options.max_tokens.unwrap_or(4096); - let mut body = json!({ - "model": options.model, - "max_tokens": max_tokens, - "messages": messages, - "tools": tools, - }); - - if let Some(sys) = system { body["system"] = sys; } - if let Some(t) = options.temperature { body["temperature"] = t.into(); } - self.apply_extra(&mut body); - body - } - - /// Collects ALL system-role messages (main prompt, mid-conversation summary, - /// tail_reminder) into the single `system` parameter the Anthropic API accepts. - /// - /// Returns a plain string in the common case. When any system message carries - /// **structured** content (a text-block array, e.g. the static prompt tagged - /// with `cache_control` when prompt caching is on), it returns the array form - /// instead so the cache breakpoint survives into `system`. String-content - /// messages become plain text blocks (no cache_control). - fn merged_system(messages: &[Value]) -> Option { - let sys: Vec<&Value> = messages - .iter() - .filter(|m| m["role"].as_str() == Some("system")) - .collect(); - if sys.is_empty() { return None; } - - if !sys.iter().any(|m| m["content"].is_array()) { - let parts: Vec<&str> = sys.iter().filter_map(|m| m["content"].as_str()).collect(); - return if parts.is_empty() { None } else { Some(Value::String(parts.join("\n\n---\n\n"))) }; - } - - let mut blocks: Vec = Vec::new(); - for m in &sys { - match &m["content"] { - Value::String(s) if !s.is_empty() => blocks.push(json!({ "type": "text", "text": s })), - Value::Array(arr) => { - for b in arr { - if b["type"].as_str() == Some("text") { - blocks.push(b.clone()); - } - } - } - _ => {} - } - } - if blocks.is_empty() { None } else { Some(Value::Array(blocks)) } - } - - fn url(&self) -> String { - format!("{}/v1/messages", self.base_url.trim_end_matches('/')) - } - - fn logged_headers(&self) -> Value { - json!({ - "x-api-key": redact_key(&self.api_key), - "anthropic-version": ANTHROPIC_VERSION, - "content-type": "application/json", - }) - } - - /// Sends the request and returns the raw response **without** `error_for_status`, - /// so the tool-calling paths can read the error body and attach the request - /// payload to the `LlmError` (a `reqwest` status error discards the body). The - /// plain `chat` path keeps its own `error_for_status`. - async fn send_request(&self, body: &Value) -> reqwest::Result { - self.http - .post(self.url()) - .header("x-api-key", &self.api_key) - .header("anthropic-version", ANTHROPIC_VERSION) - .header("X-Title", core_api::APP_NAME) - .json(body) - .send() - .await - } - - /// Joined `thinking` blocks of a content array, if any (extended thinking). - fn reasoning_of(content_blocks: &[Value]) -> Option { - let parts: Vec<&str> = content_blocks - .iter() - .filter(|b| b["type"].as_str() == Some("thinking")) - .filter_map(|b| b["thinking"].as_str()) - .collect(); - if parts.is_empty() { None } else { Some(parts.join("\n")) } - } - - /// SSE streaming path behind `chat_with_tools_raw_streaming`. Anthropic - /// streams typed events (`message_start` / `content_block_*` / - /// `message_delta` / `message_stop`); text and thinking deltas are - /// forwarded to `delta_tx` best-effort while the blocks are accumulated - /// into the same `LlmTurn` the buffered path returns. - async fn stream_chat( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - delta_tx: &mpsc::Sender, - emitted: &mut bool, - ) -> anyhow::Result<(LlmTurn, Option)> { - let system = Self::merged_system(messages); - let anthropic_messages = Self::convert_messages(messages); - let anthropic_tools = Self::convert_tools(tools); - let mut body = self.tools_body(system, anthropic_messages, anthropic_tools, options); - body["stream"] = json!(true); - - debug!(model = %options.model, tools = tools.len(), "anthropic: sending streaming chat_with_tools request"); - trace!(body = %body, "anthropic: streaming chat_with_tools request body"); - - let request_body = body.clone(); - let request_headers = self.logged_headers(); - - let http_resp = self.send_request(&body).await?; - let response_headers = headers_to_json(http_resp.headers()); - let status = http_resp.status(); - if !status.is_success() { - let resp_text = http_resp.text().await?; - return Err(crate::LlmError { - status: Some(status.as_u16()), - message: format!( - "anthropic: HTTP {status} from {url}\nbody: {resp_text}", - url = self.url(), - ), - raw_meta: Some(LlmRawMeta { - request_headers: Some(request_headers), - request_body: Some(request_body), - response_headers: Some(response_headers), - response_body: Some(error_response_body(resp_text)), - }), - }.into()); - } - - /// One content block being accumulated by index. - #[derive(Default)] - struct Block { - kind: String, // "text" | "thinking" | "tool_use" - buf: String, // text/thinking content or input_json fragments - id: String, - name: String, - } - - let mut blocks: BTreeMap = BTreeMap::new(); - let mut stop_reason: Option = None; - let mut usage = json!({}); - let mut sse = SseDecoder::new(); - let mut byte_stream = http_resp.bytes_stream(); - - let mut handle_payload = |payload: &str, emitted: &mut bool| -> anyhow::Result<()> { - let Ok(v) = serde_json::from_str::(payload) else { return Ok(()) }; - match v["type"].as_str().unwrap_or("") { - "message_start" => { - if let Some(u) = v["message"]["usage"].as_object() { - for (k, val) in u { usage[k.clone()] = val.clone(); } - } - } - "content_block_start" => { - let idx = v["index"].as_u64().unwrap_or(0); - let cb = &v["content_block"]; - let block = blocks.entry(idx).or_default(); - block.kind = cb["type"].as_str().unwrap_or("").to_string(); - block.id = cb["id"].as_str().unwrap_or("").to_string(); - block.name = cb["name"].as_str().unwrap_or("").to_string(); - } - "content_block_delta" => { - let idx = v["index"].as_u64().unwrap_or(0); - let delta = &v["delta"]; - match delta["type"].as_str().unwrap_or("") { - "text_delta" => { - if let Some(t) = delta["text"].as_str().filter(|t| !t.is_empty()) { - blocks.entry(idx).or_default().buf.push_str(t); - *emitted = true; - let _ = delta_tx.try_send(StreamDelta::Text(t.to_string())); - } - } - "thinking_delta" => { - if let Some(t) = delta["thinking"].as_str().filter(|t| !t.is_empty()) { - blocks.entry(idx).or_default().buf.push_str(t); - *emitted = true; - let _ = delta_tx.try_send(StreamDelta::Reasoning(t.to_string())); - } - } - "input_json_delta" => { - if let Some(j) = delta["partial_json"].as_str() { - blocks.entry(idx).or_default().buf.push_str(j); - } - } - // signature_delta and unknown deltas carry no displayable text. - _ => {} - } - } - "message_delta" => { - if let Some(sr) = v["delta"]["stop_reason"].as_str() { - stop_reason = Some(sr.to_string()); - } - if let Some(u) = v["usage"].as_object() { - for (k, val) in u { usage[k.clone()] = val.clone(); } - } - } - "error" => { - return Err(anyhow::anyhow!("anthropic: stream error event: {payload}")); - } - // content_block_stop / message_stop / ping: nothing to accumulate. - _ => {} - } - Ok(()) - }; - - while let Some(chunk) = byte_stream.next().await { - let chunk = chunk?; - for payload in sse.feed(&chunk) { - handle_payload(&payload, emitted)?; - } - } - for payload in sse.finish() { - handle_payload(&payload, emitted)?; - } - - let stop = stop_reason.as_deref().unwrap_or(""); - let input_tokens = usage["input_tokens"].as_u64().map(|n| n as u32); - let output_tokens = usage["output_tokens"].as_u64().map(|n| n as u32); - let cache_read_tokens = usage["cache_read_input_tokens"].as_u64().map(|n| n as u32); - let cache_creation_tokens = usage["cache_creation_input_tokens"].as_u64().map(|n| n as u32); - info!(model = %options.model, ?input_tokens, ?output_tokens, stop_reason = stop, "anthropic: streaming response completed"); - if stop == "max_tokens" { - warn!(model = %options.model, ?output_tokens, "anthropic: response truncated (max_tokens reached)"); - } - - let text_of = |kind: &str| -> String { - blocks.values() - .filter(|b| b.kind == kind) - .map(|b| b.buf.as_str()) - .collect::>() - .join("\n") - }; - let reasoning = text_of("thinking"); - let reasoning_content = if reasoning.is_empty() { None } else { Some(reasoning) }; - let tool_blocks: Vec<&Block> = blocks.values().filter(|b| b.kind == "tool_use").collect(); - - let turn = if !tool_blocks.is_empty() { - let calls = tool_blocks - .iter() - .map(|b| ToolCall { - id: b.id.clone(), - name: b.name.clone(), - arguments: serde_json::from_str(&b.buf).unwrap_or(Value::Object(Default::default())), - }) - .collect(); - LlmTurn::ToolCalls { content: text_of("text"), calls, input_tokens, output_tokens, reasoning_content, cache_read_tokens, cache_creation_tokens, cost: None } - } else { - let truncated = stop == "max_tokens"; - LlmTurn::Message(ChatResponse { - content: text_of("text"), input_tokens, output_tokens, truncated, - reasoning_content, cache_read_tokens, cache_creation_tokens, cost: None, - }) - }; - - // Buffered-shaped response body for the payload log. - let content_log: Vec = blocks.values().map(|b| match b.kind.as_str() { - "tool_use" => json!({"type": "tool_use", "id": b.id, "name": b.name, "input": serde_json::from_str::(&b.buf).unwrap_or(json!({}))}), - "thinking" => json!({"type": "thinking", "thinking": b.buf}), - _ => json!({"type": "text", "text": b.buf}), - }).collect(); - let raw_meta = LlmRawMeta { - request_headers: Some(request_headers), - request_body: Some(request_body), - response_headers: Some(response_headers), - response_body: Some(json!({ - "streamed": true, - "content": content_log, - "stop_reason": stop, - "usage": usage, - })), - }; - - Ok((turn, Some(raw_meta))) - } -} - -/// User content arrives either as a plain string or as an OpenAI-style parts -/// array (text + `image_url` data URLs, produced when the resolved model has -/// the `vision` capability). Strings pass through; parts become Anthropic -/// blocks. Video and unknown parts are dropped with a warning — providers -/// gate capabilities upstream, so this should only indicate a misconfigured -/// model row. -fn convert_user_content(content: &Value) -> Value { - let Some(parts) = content.as_array() else { - return Value::String(content.as_str().unwrap_or("").to_string()); - }; - let mut blocks = Vec::new(); - for p in parts { - match p["type"].as_str().unwrap_or("") { - "text" => blocks.push(json!({ - "type": "text", - "text": p["text"].as_str().unwrap_or(""), - })), - "image_url" => { - if let Some(block) = parse_data_image(&p["image_url"]) { - blocks.push(block); - } - } - "file" => { - if let Some(block) = parse_data_document(&p["file"]) { - blocks.push(block); - } - } - other => tracing::warn!(part_type = other, "dropping content part unsupported by Anthropic"), - } - } - Value::Array(blocks) -} - -/// `{"url": "data:;base64,"}` (or the bare-string shorthand) → an -/// Anthropic base64 image block. Only data URLs are supported. -fn parse_data_image(image_url: &Value) -> Option { - let url = image_url["url"].as_str().or_else(|| image_url.as_str())?; - let (mime, data) = url.strip_prefix("data:")?.split_once(";base64,")?; - Some(json!({ - "type": "image", - "source": { "type": "base64", "media_type": mime, "data": data }, - })) -} - -/// `{"file_data": "data:application/pdf;base64,"}` → an Anthropic base64 -/// `document` block (the native PDF input). Only base64 data URLs are supported; -/// the OpenAI `file` part is what the media pipeline emits for a PDF. -fn parse_data_document(file: &Value) -> Option { - let url = file["file_data"].as_str()?; - let (mime, data) = url.strip_prefix("data:")?.split_once(";base64,")?; - Some(json!({ - "type": "document", - "source": { "type": "base64", "media_type": mime, "data": data }, - })) -} - -#[async_trait] -impl ChatbotClient for AnthropicClient { - async fn chat( - &self, - messages: &[Message], - options: &ChatOptions, - ) -> anyhow::Result { - // Merge all system-role messages into a single `system:` parameter. - let system: Option = { - let parts: Vec<&str> = messages - .iter() - .filter(|m| m.role == Role::System) - .map(|m| m.content.as_str()) - .collect(); - if parts.is_empty() { None } else { Some(parts.join("\n\n---\n\n")) } - }; - - let msgs: Vec = messages - .iter() - .filter(|m| m.role != Role::System) - .map(|m| { - let role = match m.role { - Role::User => "user", - Role::Assistant => "assistant", - Role::System => unreachable!(), - }; - json!({ "role": role, "content": m.content }) - }) - .collect(); - - let max_tokens = options.max_tokens.unwrap_or(4096); - let mut body = json!({ - "model": options.model, - "max_tokens": max_tokens, - "messages": msgs, - }); - - if let Some(sys) = system { body["system"] = sys.into(); } - if let Some(t) = options.temperature { body["temperature"] = t.into(); } - self.apply_extra(&mut body); - - let url = format!("{}/v1/messages", self.base_url.trim_end_matches('/')); - debug!(model = %options.model, "anthropic: sending chat request"); - trace!(body = %body, "anthropic: chat request body"); - - let resp: Value = self - .http - .post(&url) - .header("x-api-key", &self.api_key) - .header("anthropic-version", ANTHROPIC_VERSION) - .json(&body) - .send() - .await? - .error_for_status()? - .json() - .await?; - - let content = resp["content"] - .as_array() - .and_then(|arr| arr.iter().find(|b| b["type"].as_str() == Some("text"))) - .and_then(|block| block["text"].as_str()) - .ok_or_else(|| anyhow::anyhow!("Missing content in Anthropic response"))? - .to_string(); - - let input_tokens = resp["usage"]["input_tokens"].as_u64().map(|n| n as u32); - let output_tokens = resp["usage"]["output_tokens"].as_u64().map(|n| n as u32); - let cache_read_tokens = resp["usage"]["cache_read_input_tokens"].as_u64().map(|n| n as u32); - let cache_creation_tokens = resp["usage"]["cache_creation_input_tokens"].as_u64().map(|n| n as u32); - info!(model = %options.model, ?input_tokens, ?output_tokens, "anthropic: chat response received"); - - let cost = self.extract_cost(&resp); - Ok(ChatResponse { content, input_tokens, output_tokens, truncated: false, reasoning_content: None, cache_read_tokens, cache_creation_tokens, cost }) - } - - async fn chat_with_tools( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - ) -> anyhow::Result { - self.chat_with_tools_raw(messages, tools, options).await.map(|(t, _)| t) - } - - async fn chat_with_tools_raw( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - ) -> anyhow::Result<(LlmTurn, Option)> { - // Mid-conversation system messages (compaction summaries, tail - // reminders) are merged into the single `system:` parameter — they - // must not be silently dropped. - let system = Self::merged_system(messages); - let anthropic_messages = Self::convert_messages(messages); - let anthropic_tools = Self::convert_tools(tools); - let body = self.tools_body(system, anthropic_messages, anthropic_tools, options); - - debug!(model = %options.model, tools = tools.len(), "anthropic: sending chat_with_tools request"); - trace!(body = %body, "anthropic: chat_with_tools request body"); - - // Capture request metadata for logging. - let request_body = body.clone(); - let request_headers = self.logged_headers(); - - let http_resp = self.send_request(&body).await?; - - let response_headers = headers_to_json(http_resp.headers()); - let status = http_resp.status(); - let resp_text = http_resp.text().await?; - if !status.is_success() { - return Err(crate::LlmError { - status: Some(status.as_u16()), - message: format!( - "anthropic: HTTP {status} from {url}\nbody: {resp_text}", - url = self.url(), - ), - raw_meta: Some(LlmRawMeta { - request_headers: Some(request_headers), - request_body: Some(request_body), - response_headers: Some(response_headers), - response_body: Some(error_response_body(resp_text)), - }), - }.into()); - } - let resp: Value = serde_json::from_str(&resp_text) - .map_err(|e| anyhow::anyhow!("anthropic: failed to parse response JSON: {e}\nbody: {resp_text}"))?; - let response_body: Value = serde_json::from_str(&resp_text).unwrap_or(Value::Null); - - let raw_meta = LlmRawMeta { - request_headers: Some(request_headers), - request_body: Some(request_body), - response_headers: Some(response_headers), - response_body: Some(response_body), - }; - - let stop_reason = resp["stop_reason"].as_str().unwrap_or(""); - let input_tokens = resp["usage"]["input_tokens"].as_u64().map(|n| n as u32); - let output_tokens = resp["usage"]["output_tokens"].as_u64().map(|n| n as u32); - let cache_read_tokens = resp["usage"]["cache_read_input_tokens"].as_u64().map(|n| n as u32); - let cache_creation_tokens = resp["usage"]["cache_creation_input_tokens"].as_u64().map(|n| n as u32); - let content_blocks = resp["content"].as_array().cloned().unwrap_or_default(); - let cost = self.extract_cost(&resp); - info!(model = %options.model, ?input_tokens, ?output_tokens, stop_reason, "anthropic: chat_with_tools response received"); - if stop_reason == "max_tokens" { - warn!(model = %options.model, ?output_tokens, "anthropic: response truncated (max_tokens reached)"); - } - - let has_tool_use = content_blocks.iter().any(|b| b["type"].as_str() == Some("tool_use")); - let reasoning_content = Self::reasoning_of(&content_blocks); - - // Check content blocks directly: Anthropic sometimes returns stop_reason "end_turn" - // even when tool_use blocks are present, so stop_reason alone is not reliable. - let turn = if stop_reason == "tool_use" || has_tool_use { - let text: String = content_blocks - .iter() - .filter(|b| b["type"].as_str() == Some("text")) - .filter_map(|b| b["text"].as_str()) - .collect::>() - .join("\n"); - - let calls: Vec = content_blocks - .iter() - .filter(|b| b["type"].as_str() == Some("tool_use")) - .map(|b| ToolCall { - id: b["id"].as_str().unwrap_or("").to_string(), - name: b["name"].as_str().unwrap_or("").to_string(), - arguments: b["input"].clone(), - }) - .collect(); - - LlmTurn::ToolCalls { content: text, calls, input_tokens, output_tokens, reasoning_content, cache_read_tokens, cache_creation_tokens, cost } - } else { - let content = content_blocks - .iter() - .find(|b| b["type"].as_str() == Some("text")) - .and_then(|b| b["text"].as_str()) - .unwrap_or("") - .to_string(); - - let truncated = stop_reason == "max_tokens"; - LlmTurn::Message(ChatResponse { content, input_tokens, output_tokens, truncated, reasoning_content, cache_read_tokens, cache_creation_tokens, cost }) - }; - - Ok((turn, Some(raw_meta))) - } - - async fn chat_with_tools_raw_streaming( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - delta_tx: mpsc::Sender, - ) -> anyhow::Result<(LlmTurn, Option)> { - let mut emitted = false; - match self.stream_chat(messages, tools, options, &delta_tx, &mut emitted).await { - Ok(ok) => Ok(ok), - // Pre-stream failure (nothing shown yet): retry buffered. A - // mid-stream failure propagates to the model-fallback logic. - Err(e) if !emitted => { - debug!(model = %options.model, error = %e, "anthropic: streaming failed before any delta; retrying buffered"); - self.chat_with_tools_raw(messages, tools, options).await - } - Err(e) => Err(e), - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn reasoning_of_joins_thinking_blocks() { - let blocks = vec![ - json!({"type": "thinking", "thinking": "first"}), - json!({"type": "text", "text": "answer"}), - json!({"type": "thinking", "thinking": "second"}), - ]; - assert_eq!( - AnthropicClient::reasoning_of(&blocks), - Some("first\nsecond".to_string()) - ); - assert_eq!(AnthropicClient::reasoning_of(&[]), None); - assert_eq!( - AnthropicClient::reasoning_of(&[json!({"type": "text", "text": "a"})]), - None - ); - } - - #[test] - fn user_content_string_passthrough() { - let v = convert_user_content(&json!("hello")); - assert_eq!(v, json!("hello")); - } - - #[test] - fn user_content_parts_become_anthropic_blocks() { - let v = convert_user_content(&json!([ - { "type": "text", "text": "what is this?" }, - { "type": "image_url", "image_url": { "url": "data:image/png;base64,QUJD" } }, - ])); - assert_eq!(v, json!([ - { "type": "text", "text": "what is this?" }, - { "type": "image", "source": { "type": "base64", "media_type": "image/png", "data": "QUJD" } }, - ])); - } - - #[test] - fn user_content_drops_video_and_non_data_urls() { - let v = convert_user_content(&json!([ - { "type": "text", "text": "t" }, - { "type": "video_url", "video_url": { "url": "data:video/mp4;base64,QUJD" } }, - { "type": "image_url", "image_url": { "url": "https://example.com/x.png" } }, - ])); - assert_eq!(v, json!([{ "type": "text", "text": "t" }])); - } - - #[test] - fn user_content_file_part_becomes_document_block() { - // The OpenAI `file` part (emitted by the media pipeline for a PDF) becomes - // an Anthropic native `document` block. - let v = convert_user_content(&json!([ - { "type": "text", "text": "read this" }, - { "type": "file", "file": { "filename": "a.pdf", "file_data": "data:application/pdf;base64,QUJD" } }, - ])); - assert_eq!(v, json!([ - { "type": "text", "text": "read this" }, - { "type": "document", "source": { "type": "base64", "media_type": "application/pdf", "data": "QUJD" } }, - ])); - - // A non-data file_data (or missing) is dropped, not forwarded. - let v = convert_user_content(&json!([ - { "type": "file", "file": { "filename": "a.pdf", "file_data": "https://example.com/a.pdf" } }, - ])); - assert_eq!(v, json!([])); - } -} diff --git a/crates/llm-client/src/lib.rs b/crates/llm-client/src/lib.rs deleted file mode 100644 index 1d44eed..0000000 --- a/crates/llm-client/src/lib.rs +++ /dev/null @@ -1,168 +0,0 @@ -pub mod anthropic; -pub mod lm_studio; -pub mod ollama; -pub mod openai; - -// Re-export the trait and all associated types from core-api so existing -// callers that import from `llm_client` continue to work unchanged. -pub use core_api::chatbot::{ - ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, Role, StreamDelta, - ToolCall, -}; - -use serde_json::Value; - -/// Incremental SSE decoder: feed raw response bytes, get back the payload of -/// every complete `data:` line seen (`[DONE]` included — callers decide). -/// Buffers partial lines across chunks; `event:` lines and comments are -/// skipped (both OpenAI and Anthropic put the event type inside the JSON). -#[derive(Default)] -pub struct SseDecoder { - buf: Vec, -} - -impl SseDecoder { - pub fn new() -> Self { - Self::default() - } - - pub fn feed(&mut self, bytes: &[u8]) -> Vec { - self.buf.extend_from_slice(bytes); - let mut out = Vec::new(); - while let Some(pos) = self.buf.iter().position(|&b| b == b'\n') { - let line: Vec = self.buf.drain(..=pos).collect(); - if let Some(payload) = parse_sse_line(&line) { - out.push(payload); - } - } - out - } - - /// Flush a trailing line not terminated by `\n` at end-of-stream. - pub fn finish(&mut self) -> Vec { - let rest = std::mem::take(&mut self.buf); - parse_sse_line(&rest).into_iter().collect() - } -} - -/// A complete SSE line is valid UTF-8 (a multibyte sequence never contains a -/// `\n` byte), but decode lossily anyway — a corrupt line is skipped, not fatal. -fn parse_sse_line(line: &[u8]) -> Option { - let line = String::from_utf8_lossy(line); - let line = line.trim_end_matches('\r').trim(); - let data = line.strip_prefix("data:")?.trim_start(); - if data.is_empty() { None } else { Some(data.to_string()) } -} - -/// Converts a reqwest `HeaderMap` into a `serde_json::Value` object. -pub fn headers_to_json(headers: &reqwest::header::HeaderMap) -> Value { - let map: serde_json::Map = headers - .iter() - .map(|(k, v)| ( - k.as_str().to_string(), - v.to_str().unwrap_or("").into(), - )) - .collect(); - Value::Object(map) -} - -/// Turns a raw error-response body into a JSON `Value` for the payload log: -/// the parsed JSON when the provider returned JSON (the common case — an -/// `{"error": …}` object), else the raw text wrapped as a JSON string so a -/// non-JSON body (HTML gateway page, plain text) is still preserved verbatim. -pub fn error_response_body(text: String) -> Value { - serde_json::from_str::(&text).unwrap_or(Value::String(text)) -} - -/// Returns a redacted preview of an API key: first 7 chars + "***". -pub fn redact_key(key: &str) -> String { - if key.len() > 7 { - format!("{}***", &key[..7]) - } else { - "***".to_string() - } -} - -/// A structured LLM call failure carrying the HTTP `status` of the response. -/// -/// Clients that read the status themselves (rather than via `error_for_status`) -/// return this so callers can classify retriability on the numeric code instead of -/// substring-matching a formatted message — which mis-fires when a model id, token -/// count or URL merely contains "401"/"404"/… (bug B6). Non-HTTP failures (network, -/// JSON parse, cancellation) stay ordinary `anyhow` errors with no status. -#[derive(Debug, Default)] -pub struct LlmError { - /// HTTP status code, when the failure came from an HTTP response. - pub status: Option, - /// Human-readable detail (provider tag + body), used for logs and the UI. - pub message: String, - /// Request/response payload captured at the failing call, so the debug log - /// can show what was actually sent even when the provider rejected it (e.g. - /// a 400). `None` for failures with no HTTP round-trip (network, cancellation, - /// parse) — those carry no body to surface. - pub raw_meta: Option, -} - -impl std::fmt::Display for LlmError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str(&self.message) - } -} - -impl std::error::Error for LlmError {} - -/// Extracts the HTTP status of an LLM failure, if any: a structured -/// [`LlmError::status`] first, else any `reqwest::Error` in the source chain (the -/// clients that fail via `error_for_status()?`). Returns `None` for a non-HTTP -/// error (network, parse, cancellation), which callers should treat as retriable. -pub fn http_status(err: &anyhow::Error) -> Option { - for cause in err.chain() { - if let Some(le) = cause.downcast_ref::() { - return le.status; - } - if let Some(re) = cause.downcast_ref::() { - if let Some(s) = re.status() { - return Some(s.as_u16()); - } - } - } - None -} - -#[cfg(test)] -mod tests { - use super::SseDecoder; - - #[test] - fn sse_decoder_buffers_partial_lines_across_chunks() { - let mut dec = SseDecoder::new(); - // A payload split mid-JSON across two chunks yields one complete line. - assert!(dec.feed(br#"data: {"a": 1"#).is_empty()); - assert_eq!(dec.feed(b"}\r\n").len(), 1); - } - - #[test] - fn sse_decoder_skips_events_comments_and_keeps_done() { - let mut dec = SseDecoder::new(); - let out = dec.feed(b"event: message_start\n: ping\n\ndata: {\"type\":\"ping\"}\ndata: [DONE]\n"); - assert_eq!(out, vec!["{\"type\":\"ping\"}".to_string(), "[DONE]".to_string()]); - assert!(dec.finish().is_empty()); - } - - #[test] - fn sse_decoder_finish_flushes_unterminated_tail() { - let mut dec = SseDecoder::new(); - assert!(dec.feed(b"data: tail-without-newline").is_empty()); - assert_eq!(dec.finish(), vec!["tail-without-newline".to_string()]); - } - - #[test] - fn sse_decoder_handles_multibyte_split() { - let mut dec = SseDecoder::new(); - // "€" is 3 bytes in UTF-8; split across the chunk boundary. - let payload = "data: {\"t\":\"€\"}\n".as_bytes(); - let (a, b) = payload.split_at(12); - assert!(dec.feed(a).is_empty()); - assert_eq!(dec.feed(b), vec!["{\"t\":\"€\"}".to_string()]); - } -} diff --git a/crates/llm-client/src/lm_studio.rs b/crates/llm-client/src/lm_studio.rs deleted file mode 100644 index 8c4b63b..0000000 --- a/crates/llm-client/src/lm_studio.rs +++ /dev/null @@ -1,65 +0,0 @@ -use async_trait::async_trait; -use serde_json::Value; -use tokio::sync::mpsc; - -use crate::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, StreamDelta, openai::OpenAiClient}; - -/// LM Studio client. -/// -/// LM Studio exposes an OpenAI-compatible `/v1` endpoint, so this is a thin -/// wrapper that defaults to `http://localhost:1234/v1` and requires no API key. -pub struct LmStudioClient { - inner: OpenAiClient, -} - -impl LmStudioClient { - /// `base_url` defaults to `http://localhost:1234/v1` if `None`. - pub fn new(base_url: Option>) -> Self { - let url = base_url - .map(|u| u.into()) - .unwrap_or_else(|| "http://localhost:1234/v1".to_string()); - Self { inner: OpenAiClient::new(url, "", None, false) } - } -} - -#[async_trait] -impl ChatbotClient for LmStudioClient { - async fn chat( - &self, - messages: &[Message], - options: &ChatOptions, - ) -> anyhow::Result { - self.inner.chat(messages, options).await - } - - async fn chat_with_tools( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - ) -> anyhow::Result { - self.inner.chat_with_tools(messages, tools, options).await - } - - async fn chat_with_tools_raw( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - ) -> anyhow::Result<(LlmTurn, Option)> { - self.inner.chat_with_tools_raw(messages, tools, options).await - } - - /// LM Studio is OpenAI-compatible: streaming forwards to the inner client. - /// If a local build rejects `stream_options`, the inner pre-delta buffered - /// retry covers it transparently. - async fn chat_with_tools_raw_streaming( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - delta_tx: mpsc::Sender, - ) -> anyhow::Result<(LlmTurn, Option)> { - self.inner.chat_with_tools_raw_streaming(messages, tools, options, delta_tx).await - } -} diff --git a/crates/llm-client/src/ollama.rs b/crates/llm-client/src/ollama.rs deleted file mode 100644 index e0215a7..0000000 --- a/crates/llm-client/src/ollama.rs +++ /dev/null @@ -1,76 +0,0 @@ -use async_trait::async_trait; -use serde_json::{Value, json}; - -use crate::{ChatOptions, ChatResponse, ChatbotClient, Message, Role}; - -/// Ollama client using the native `/api/chat` endpoint. -/// -/// Defaults to `http://localhost:11434`. No API key required. -pub struct OllamaClient { - base_url: String, - http: reqwest::Client, -} - -impl OllamaClient { - /// `base_url` defaults to `http://localhost:11434` if `None`. - pub fn new(base_url: Option>) -> Self { - let url = base_url - .map(|u| u.into()) - .unwrap_or_else(|| "http://localhost:11434".to_string()); - Self { base_url: url, http: reqwest::Client::new() } - } -} - -#[async_trait] -impl ChatbotClient for OllamaClient { - async fn chat( - &self, - messages: &[Message], - options: &ChatOptions, - ) -> anyhow::Result { - let msgs: Vec = messages - .iter() - .map(|m| { - let role = match m.role { - Role::System => "system", - Role::User => "user", - Role::Assistant => "assistant", - }; - json!({ "role": role, "content": m.content }) - }) - .collect(); - - let mut options_obj = json!({}); - if let Some(t) = options.temperature { options_obj["temperature"] = t.into(); } - if let Some(n) = options.max_tokens { options_obj["num_predict"] = n.into(); } - - let body = json!({ - "model": options.model, - "messages": msgs, - "stream": false, - "options": options_obj, - }); - - let url = format!("{}/api/chat", self.base_url.trim_end_matches('/')); - - let resp: Value = self - .http - .post(&url) - .json(&body) - .send() - .await? - .error_for_status()? - .json() - .await?; - - let content = resp["message"]["content"] - .as_str() - .ok_or_else(|| anyhow::anyhow!("Missing content in Ollama response"))? - .to_string(); - - let input_tokens = resp["prompt_eval_count"].as_u64().map(|n| n as u32); - let output_tokens = resp["eval_count"].as_u64().map(|n| n as u32); - - Ok(ChatResponse { content, input_tokens, output_tokens, truncated: false, reasoning_content: None, cache_read_tokens: None, cache_creation_tokens: None, cost: None }) - } -} diff --git a/crates/llm-client/src/openai.rs b/crates/llm-client/src/openai.rs deleted file mode 100644 index 6b8ee09..0000000 --- a/crates/llm-client/src/openai.rs +++ /dev/null @@ -1,485 +0,0 @@ -use std::collections::BTreeMap; - -use async_trait::async_trait; -use futures_util::StreamExt; -use serde_json::{Value, json}; -use tokio::sync::mpsc; -use tracing::{debug, info, trace, warn}; - -use crate::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, Role, SseDecoder, StreamDelta, ToolCall, error_response_body, headers_to_json, redact_key}; -use core_api::APP_NAME; - -/// OpenAI ChatGPT client (also compatible with any OpenAI-spec endpoint). -pub struct OpenAiClient { - base_url: String, - api_key: String, - extra_params: Option, - /// When true, Anthropic-compatible prompt-caching hints are injected: - /// - `anthropic-beta: prompt-caching-2024-07-31` header is sent. - /// - The last tool definition is tagged with `cache_control: {"type":"ephemeral"}`. - /// - System message content is expected to already be a content array with - /// `cache_control` on the static block (set by `build_openai_messages`). - /// Used for OpenRouter when routing to Anthropic models. - enable_prompt_cache: bool, - http: reqwest::Client, -} - -impl OpenAiClient { - pub fn new(base_url: impl Into, api_key: impl Into, extra_params: Option, enable_prompt_cache: bool) -> Self { - Self { - base_url: base_url.into(), - api_key: api_key.into(), - extra_params, - enable_prompt_cache, - http: reqwest::Client::new(), - } - } - - /// Merges `extra_params` (if any) into `body`. Only top-level object keys are merged. - fn apply_extra(&self, body: &mut serde_json::Value) { - if let Some(serde_json::Value::Object(extra)) = &self.extra_params { - if let Some(b) = body.as_object_mut() { - for (k, v) in extra { - b.insert(k.clone(), v.clone()); - } - } - } - } - - fn url(&self) -> String { - format!("{}/chat/completions", self.base_url.trim_end_matches('/')) - } - - /// Shared request body for the buffered and the streaming path. Caller adds - /// `max_tokens`/`temperature`/`extra_params` afterwards via `finalize_body`. - fn base_body(&self, model: &str, messages: &[Value], tools: &[Value]) -> Value { - let mut body = json!({ - "model": model, - "messages": messages, - }); - - if !tools.is_empty() { - // When prompt caching is enabled, tag the last tool with cache_control - // so the entire tools array is included in the Anthropic KV cache prefix. - let tools_value: Value = if self.enable_prompt_cache { - let mut tagged = tools.to_vec(); - if let Some(last) = tagged.last_mut() { - last["cache_control"] = json!({"type": "ephemeral"}); - } - tagged.into() - } else { - tools.into() - }; - body["tools"] = tools_value; - body["tool_choice"] = "auto".into(); - } - body - } - - fn finalize_body(&self, mut body: Value, options: &ChatOptions) -> Value { - if let Some(t) = options.max_tokens { body["max_tokens"] = t.into(); } - if let Some(t) = options.temperature { body["temperature"] = t.into(); } - self.apply_extra(&mut body); - body - } - - /// Request metadata for logging (shared by buffered and streaming paths). - fn logged_headers(&self) -> Value { - let mut logged_headers = json!({ - "authorization": format!("Bearer {}", redact_key(&self.api_key)), - "content-type": "application/json", - }); - if self.enable_prompt_cache { - logged_headers["anthropic-beta"] = "prompt-caching-2024-07-31".into(); - } - logged_headers - } - - async fn send_request(&self, body: &Value) -> reqwest::Result { - let mut req = self.http.post(self.url()).bearer_auth(&self.api_key).header("X-Title", APP_NAME); - if self.enable_prompt_cache { - req = req.header("anthropic-beta", "prompt-caching-2024-07-31"); - } - req.json(body).send().await - } - - /// SSE streaming path behind `chat_with_tools_raw_streaming`. Accumulates - /// content/reasoning/tool-call fragments into the same `LlmTurn` the - /// buffered path would return, while forwarding text/reasoning deltas to - /// `delta_tx` (try_send, best-effort). `emitted` tracks whether any delta - /// was pushed, so the caller can distinguish a pre-stream failure (safe to - /// retry buffered) from a mid-stream one (partial output already shown). - async fn stream_chat( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - delta_tx: &mpsc::Sender, - emitted: &mut bool, - ) -> anyhow::Result<(LlmTurn, Option)> { - let mut body = self.base_body(&options.model, messages, tools); - body["stream"] = json!(true); - body["stream_options"] = json!({ "include_usage": true }); - let body = self.finalize_body(body, options); - - debug!(model = %options.model, tools = tools.len(), prompt_cache = self.enable_prompt_cache, "openai: sending streaming chat_with_tools request"); - trace!(body = %body, "openai: streaming chat_with_tools request body"); - - let request_body = body.clone(); - let request_headers = self.logged_headers(); - - let http_resp = self.send_request(&body).await?; - - let response_headers = headers_to_json(http_resp.headers()); - let status = http_resp.status(); - if !status.is_success() { - let resp_text = http_resp.text().await?; - return Err(crate::LlmError { - status: Some(status.as_u16()), - message: format!( - "openai: HTTP {status} from {url}\nbody: {resp_text}", - url = self.url(), - ), - raw_meta: Some(LlmRawMeta { - request_headers: Some(request_headers), - request_body: Some(request_body), - response_headers: Some(response_headers), - response_body: Some(error_response_body(resp_text)), - }), - }.into()); - } - - let mut content = String::new(); - let mut reasoning = String::new(); - // index → (id, name, arguments fragment buffer) - let mut tool_calls: BTreeMap = BTreeMap::new(); - let mut finish_reason: Option = None; - let mut usage: Option = None; - let mut sse = SseDecoder::new(); - let mut byte_stream = http_resp.bytes_stream(); - - // One SSE `data:` payload. Fragments update the accumulators; text and - // reasoning also go out as deltas. Unparseable chunks are skipped — - // the assembled turn stays consistent. - let mut handle_payload = |payload: &str, emitted: &mut bool| { - if payload == "[DONE]" { - return; - } - let Ok(v) = serde_json::from_str::(payload) else { return }; - if let Some(u) = v.get("usage").filter(|u| !u.is_null()) { - usage = Some(u.clone()); - } - let Some(choice) = v["choices"].as_array().and_then(|a| a.first()) else { return }; - if let Some(fr) = choice["finish_reason"].as_str() { - finish_reason = Some(fr.to_string()); - } - let delta = &choice["delta"]; - if let Some(t) = delta["content"].as_str().filter(|t| !t.is_empty()) { - content.push_str(t); - *emitted = true; - let _ = delta_tx.try_send(StreamDelta::Text(t.to_string())); - } - // Same normalization as the buffered path: DeepSeek uses - // `reasoning_content`, MiniMax M3 and others `reasoning`. - if let Some(t) = delta["reasoning_content"].as_str() - .or_else(|| delta["reasoning"].as_str()) - .filter(|t| !t.is_empty()) - { - reasoning.push_str(t); - *emitted = true; - let _ = delta_tx.try_send(StreamDelta::Reasoning(t.to_string())); - } - if let Some(tc_arr) = delta["tool_calls"].as_array() { - for tc in tc_arr { - let idx = tc["index"].as_u64().unwrap_or(0); - let entry = tool_calls.entry(idx).or_default(); - if let Some(id) = tc["id"].as_str() { entry.0 = id.to_string(); } - if let Some(n) = tc["function"]["name"].as_str() { entry.1 = n.to_string(); } - if let Some(a) = tc["function"]["arguments"].as_str() { entry.2.push_str(a); } - } - } - }; - - while let Some(chunk) = byte_stream.next().await { - let chunk = chunk?; - for payload in sse.feed(&chunk) { - handle_payload(&payload, emitted); - } - } - for payload in sse.finish() { - handle_payload(&payload, emitted); - } - - let finish = finish_reason.as_deref().unwrap_or("stop"); - let input_tokens = usage.as_ref().and_then(|u| u["prompt_tokens"].as_u64()).map(|n| n as u32); - let output_tokens = usage.as_ref().and_then(|u| u["completion_tokens"].as_u64()).map(|n| n as u32); - let cache_read_tokens = usage.as_ref() - .and_then(|u| u["prompt_tokens_details"]["cached_tokens"].as_u64()) - .map(|n| n as u32); - let cost = usage.as_ref().and_then(|u| u["cost"].as_f64()); - let reasoning_content = if reasoning.is_empty() { None } else { Some(reasoning) }; - info!(model = %options.model, ?input_tokens, ?output_tokens, finish_reason = finish, "openai: streaming response completed"); - if finish == "length" { - warn!(model = %options.model, ?output_tokens, "openai: response truncated (max_tokens reached)"); - } - - // Reassemble the streamed message for the payload log, so a streamed call - // leaves the same debugging trail as a buffered one — including - // reasoning_content and tool_calls, which previously existed only as - // transient deltas and never appeared in the logged body. Built here, - // before `turn` consumes the accumulators (clones are cheap vs. the round-trip). - let logged_tool_calls: Vec = tool_calls.iter() - .map(|(_idx, (id, name, args))| json!({ - "id": id, - "type": "function", - "function": { "name": name, "arguments": args }, - })) - .collect(); - let mut logged_message = json!({ "role": "assistant", "content": content.clone() }); - if let Some(rc) = &reasoning_content { - logged_message["reasoning_content"] = rc.clone().into(); - } - if !logged_tool_calls.is_empty() { - logged_message["tool_calls"] = Value::Array(logged_tool_calls); - } - - let turn = if !tool_calls.is_empty() { - let calls = tool_calls - .into_values() - .map(|(id, name, args)| ToolCall { - id, - name, - arguments: serde_json::from_str(&args).unwrap_or(Value::Object(Default::default())), - }) - .collect(); - LlmTurn::ToolCalls { content, calls, input_tokens, output_tokens, reasoning_content, cache_read_tokens, cache_creation_tokens: None, cost } - } else { - let truncated = finish == "length"; - LlmTurn::Message(ChatResponse { content, input_tokens, output_tokens, truncated, reasoning_content, cache_read_tokens, cache_creation_tokens: None, cost }) - }; - - // Synthesize a buffered-shaped response body for the payload log, so a - // streamed call leaves the same debugging trail as a buffered one. - let response_body = json!({ - "streamed": true, - "choices": [{ "finish_reason": finish, "message": logged_message }], - "usage": usage, - }); - let raw_meta = LlmRawMeta { - request_headers: Some(request_headers), - request_body: Some(request_body), - response_headers: Some(response_headers), - response_body: Some(response_body), - }; - - Ok((turn, Some(raw_meta))) - } -} - -#[async_trait] -impl ChatbotClient for OpenAiClient { - async fn chat( - &self, - messages: &[Message], - options: &ChatOptions, - ) -> anyhow::Result { - let msgs: Vec = messages - .iter() - .map(|m| { - let role = match m.role { - Role::System => "system", - Role::User => "user", - Role::Assistant => "assistant", - }; - json!({ "role": role, "content": m.content }) - }) - .collect(); - - let mut body = json!({ - "model": options.model, - "messages": msgs, - }); - - if let Some(t) = options.max_tokens { body["max_tokens"] = t.into(); } - if let Some(t) = options.temperature { body["temperature"] = t.into(); } - self.apply_extra(&mut body); - - debug!(model = %options.model, "openai: sending chat request"); - trace!(body = %body, "openai: chat request body"); - - let resp: Value = self - .http - .post(self.url()) - .bearer_auth(&self.api_key) - .header("X-Title", APP_NAME) - .json(&body) - .send() - .await? - .error_for_status()? - .json() - .await?; - - let content = match resp["choices"][0]["message"]["content"].as_str() { - Some(s) => s.to_string(), - None => { - warn!(raw_response = %resp, "openai: chat() response has null content"); - String::new() - } - }; - - let input_tokens = resp["usage"]["prompt_tokens"].as_u64().map(|n| n as u32); - let output_tokens = resp["usage"]["completion_tokens"].as_u64().map(|n| n as u32); - let cache_read_tokens = resp["usage"]["prompt_tokens_details"]["cached_tokens"].as_u64().map(|n| n as u32); - let truncated = resp["choices"][0]["finish_reason"].as_str() == Some("length"); - let cost = self.extract_cost(&resp); - info!(model = %options.model, ?input_tokens, ?output_tokens, ?cost, truncated, "openai: chat response received"); - - Ok(ChatResponse { content, input_tokens, output_tokens, truncated, reasoning_content: None, cache_read_tokens, cache_creation_tokens: None, cost }) - } - - async fn chat_with_tools( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - ) -> anyhow::Result { - self.chat_with_tools_raw(messages, tools, options).await.map(|(t, _)| t) - } - - async fn chat_with_tools_raw( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - ) -> anyhow::Result<(LlmTurn, Option)> { - let body = self.finalize_body(self.base_body(&options.model, messages, tools), options); - - debug!(model = %options.model, tools = tools.len(), prompt_cache = self.enable_prompt_cache, "openai: sending chat_with_tools request"); - trace!(body = %body, "openai: chat_with_tools request body"); - - // Capture request metadata for logging. - let request_body = body.clone(); - let request_headers = self.logged_headers(); - - let http_resp = self.send_request(&body).await?; - - let response_headers = headers_to_json(http_resp.headers()); - let status = http_resp.status(); - let resp_text = http_resp.text().await?; - - if !status.is_success() { - return Err(crate::LlmError { - status: Some(status.as_u16()), - message: format!( - "openai: HTTP {status} from {url}\nbody: {resp_text}", - url = self.url(), - ), - raw_meta: Some(LlmRawMeta { - request_headers: Some(request_headers), - request_body: Some(request_body), - response_headers: Some(response_headers), - response_body: Some(error_response_body(resp_text)), - }), - }.into()); - } - - let resp: Value = serde_json::from_str(&resp_text) - .map_err(|e| anyhow::anyhow!("openai: failed to parse response JSON: {e}\nbody: {resp_text}"))?; - let response_body: Value = serde_json::from_str(&resp_text).unwrap_or(Value::Null); - - let raw_meta = LlmRawMeta { - request_headers: Some(request_headers), - request_body: Some(request_body), - response_headers: Some(response_headers), - response_body: Some(response_body), - }; - - let input_tokens = resp["usage"]["prompt_tokens"].as_u64().map(|n| n as u32); - let output_tokens = resp["usage"]["completion_tokens"].as_u64().map(|n| n as u32); - let cache_read_tokens = resp["usage"]["prompt_tokens_details"]["cached_tokens"].as_u64().map(|n| n as u32); - let cost = self.extract_cost(&resp); - - let choice = &resp["choices"][0]; - let message = &choice["message"]; - let finish = choice["finish_reason"].as_str().unwrap_or("stop"); - info!(model = %options.model, ?input_tokens, ?output_tokens, finish_reason = finish, "openai: chat_with_tools response received"); - if finish == "length" { - warn!(model = %options.model, ?output_tokens, "openai: response truncated (max_tokens reached)"); - } - - // Thinking/reasoning content varies by provider: - // - DeepSeek: "reasoning_content" (must be echoed back on subsequent turns, even as "") - // - MiniMax M3 and others: "reasoning" - // We normalize to a single field and echo under both names in message_builder. - let reasoning_content = message["reasoning_content"].as_str() - .or_else(|| message["reasoning"].as_str()) - .map(str::to_string); - - let tool_calls_array = message["tool_calls"].as_array().filter(|a| !a.is_empty()); - - // Some models (e.g. Qwen via OpenRouter) return finish_reason "stop" even when - // tool_calls are present, so check the array directly rather than relying on finish_reason. - let turn = if finish == "tool_calls" || tool_calls_array.is_some() { - let content = message["content"].as_str().unwrap_or("").to_string(); - - let calls = tool_calls_array - .ok_or_else(|| anyhow::anyhow!("finish_reason=tool_calls but tool_calls array missing or empty"))? - .iter() - .map(|tc| { - let id = tc["id"].as_str().unwrap_or("").to_string(); - let name = tc["function"]["name"].as_str().unwrap_or("").to_string(); - let args: Value = tc["function"]["arguments"] - .as_str() - .and_then(|s| serde_json::from_str(s).ok()) - .unwrap_or(Value::Object(Default::default())); - ToolCall { id, name, arguments: args } - }) - .collect(); - - LlmTurn::ToolCalls { content, calls, input_tokens, output_tokens, reasoning_content, cache_read_tokens, cache_creation_tokens: None, cost } - } else { - // content can be null for thinking/reasoning models or when finish_reason="length". - // Fall back to empty string rather than erroring — the partial response is still - // useful and a hard error breaks the session. - let content = match message["content"].as_str() { - Some(s) => s.to_string(), - None => { - tracing::warn!( - finish_reason = finish, - ?input_tokens, - ?output_tokens, - raw_message = %message, - "OpenAI response has null content", - ); - String::new() - } - }; - let truncated = finish == "length"; - LlmTurn::Message(ChatResponse { content, input_tokens, output_tokens, truncated, reasoning_content, cache_read_tokens, cache_creation_tokens: None, cost }) - }; - - Ok((turn, Some(raw_meta))) - } - - async fn chat_with_tools_raw_streaming( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - delta_tx: mpsc::Sender, - ) -> anyhow::Result<(LlmTurn, Option)> { - let mut emitted = false; - match self.stream_chat(messages, tools, options, &delta_tx, &mut emitted).await { - Ok(ok) => Ok(ok), - // Nothing was ever streamed: some OpenAI-compatible providers reject - // `stream`/`stream_options` outright — retry buffered so they keep - // working exactly as before. A mid-stream failure (deltas already - // shown) instead propagates to the model-fallback logic. - Err(e) if !emitted => { - debug!(model = %options.model, error = %e, "openai: streaming failed before any delta; retrying buffered"); - self.chat_with_tools_raw(messages, tools, options).await - } - Err(e) => Err(e), - } - } -} diff --git a/crates/skald-core/Cargo.toml b/crates/skald-core/Cargo.toml index 76c1c1b..bfe7d88 100644 --- a/crates/skald-core/Cargo.toml +++ b/crates/skald-core/Cargo.toml @@ -78,6 +78,6 @@ base64 = "0.22" sha2 = "0.10" notify = "8" honcho-client = { path = "../honcho-client" } -llm-client = { path = "../llm-client" } +agent-loop = { path = "../agent-loop" } core-api = { path = "../core-api" } mcp-client = { path = "../mcp-client" } diff --git a/crates/skald-core/src/chatbot/logging.rs b/crates/skald-core/src/chatbot/logging.rs deleted file mode 100644 index 30a086a..0000000 --- a/crates/skald-core/src/chatbot/logging.rs +++ /dev/null @@ -1,166 +0,0 @@ -//! Transparent logging wrapper for any [`ChatbotClient`]. -//! -//! [`LoggingChatbotClient`] intercepts every `chat_with_tools_raw` call, captures -//! the raw HTTP request/response from the inner provider, persists a **metadata-only** -//! row to `llm_requests` in `system.db` (fire-and-forget), then returns the raw data -//! to the caller so it can write the **payload** to the user's own database. -//! -//! The split keeps conversation content (payloads) behind the user key while -//! metadata (cost, tokens, timing) stays in the admin-readable registry. - -use std::sync::Arc; -use std::time::{Duration, Instant}; - -use async_trait::async_trait; -use serde_json::Value; -use sqlx::SqlitePool; -use tokio::sync::mpsc; -use tracing::warn; - -use crate::db::llm_requests; - -use super::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, StreamDelta}; - -// ───────────────────────────────────────────────────────────────────────────── - -pub struct LoggingChatbotClient { - inner: Arc, - pool: Arc, - model_name: String, -} - -impl LoggingChatbotClient { - pub fn new( - inner: Arc, - pool: Arc, - model_name: impl Into, - ) -> Self { - Self { inner, pool, model_name: model_name.into() } - } - - /// Shared logging tail of both raw entry points: writes the metadata-only - /// row to `system.db` (fire-and-forget), then passes the result through. - async fn log_and_return( - &self, - options: &ChatOptions, - duration: Duration, - result: anyhow::Result<(LlmTurn, Option)>, - ) -> anyhow::Result<(LlmTurn, Option)> { - let duration_ms = duration.as_millis() as i64; - - let session_id = options.session_id; - let stack_id = options.stack_id; - let user_id = options.user_id.clone(); - let request_id = options.request_id.clone(); - let model_name = self.model_name.clone(); - let pool = Arc::clone(&self.pool); - - match result { - Ok((turn, meta)) => { - let (input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens) = match &turn { - LlmTurn::Message(r) => (r.input_tokens, r.output_tokens, r.cache_read_tokens, r.cache_creation_tokens), - LlmTurn::ToolCalls { input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, .. } => - (*input_tokens, *output_tokens, *cache_read_tokens, *cache_creation_tokens), - }; - - tokio::spawn(async move { - if let Err(e) = llm_requests::insert(&pool, llm_requests::LlmRequestRow { - request_id, - user_id, - session_id, - stack_id, - model_name, - error_text: None, - input_tokens: input_tokens.map(|n| n as i64), - output_tokens: output_tokens.map(|n| n as i64), - duration_ms, - cache_read_tokens: cache_read_tokens.map(|n| n as i64), - cache_creation_tokens: cache_creation_tokens.map(|n| n as i64), - }).await { - warn!(error = %e, "llm_requests: failed to insert log row"); - } - }); - - Ok((turn, meta)) - } - - Err(e) => { - let error_text = e.to_string(); - - tokio::spawn(async move { - if let Err(log_err) = llm_requests::insert(&pool, llm_requests::LlmRequestRow { - request_id, - user_id, - session_id, - stack_id, - model_name, - error_text: Some(error_text), - input_tokens: None, - output_tokens: None, - duration_ms, - cache_read_tokens: None, - cache_creation_tokens: None, - }).await { - warn!(error = %log_err, "llm_requests: failed to insert error log row"); - } - }); - - Err(e) - } - } - } -} - -#[async_trait] -impl ChatbotClient for LoggingChatbotClient { - /// Passthrough — logging only applies to the tool-calling path. - async fn chat( - &self, - messages: &[Message], - options: &ChatOptions, - ) -> anyhow::Result { - self.inner.chat(messages, options).await - } - - /// Passthrough that drops the raw meta. Used by callers that do not need - /// payload capture (e.g. the compactor). - async fn chat_with_tools( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - ) -> anyhow::Result { - let (turn, _) = self.chat_with_tools_raw(messages, tools, options).await?; - Ok(turn) - } - - /// Intercepts the call, delegates to `inner.chat_with_tools_raw` to capture - /// HTTP wire data, writes a **metadata-only** row to `system.db`, then returns - /// the raw data so the caller can persist payloads to the user's own database. - async fn chat_with_tools_raw( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - ) -> anyhow::Result<(LlmTurn, Option)> { - let start = Instant::now(); - let result = self.inner.chat_with_tools_raw(messages, tools, options).await; - self.log_and_return(options, start.elapsed(), result).await - } - - /// Streaming twin of `chat_with_tools_raw`: forwards `delta_tx` untouched to - /// the inner client (deltas are not logged — only the final turn is), then - /// applies the same metadata logging. Without this override the trait - /// default would silently fall back to the buffered call. - async fn chat_with_tools_raw_streaming( - &self, - messages: &[Value], - tools: &[Value], - options: &ChatOptions, - delta_tx: mpsc::Sender, - ) -> anyhow::Result<(LlmTurn, Option)> { - let start = Instant::now(); - let result = self.inner.chat_with_tools_raw_streaming(messages, tools, options, delta_tx).await; - self.log_and_return(options, start.elapsed(), result).await - } -} diff --git a/crates/skald-core/src/chatbot/mod.rs b/crates/skald-core/src/chatbot/mod.rs deleted file mode 100644 index f66abd5..0000000 --- a/crates/skald-core/src/chatbot/mod.rs +++ /dev/null @@ -1,7 +0,0 @@ -pub mod logging; - -// Re-export from the independent llm-client crate. -pub use llm_client::{ - ChatOptions, ChatResponse, ChatbotClient, LlmError, LlmRawMeta, LlmTurn, Message, StreamDelta, - ToolCall, anthropic, http_status, lm_studio, ollama, openai, -}; diff --git a/crates/skald-core/src/compactor.rs b/crates/skald-core/src/compactor.rs index 2698fc5..125e001 100644 --- a/crates/skald-core/src/compactor.rs +++ b/crates/skald-core/src/compactor.rs @@ -52,7 +52,6 @@ use tracing::{debug, info, warn}; use core_api::{ConfigProperty, ConfigSet, PropertyType}; use crate::chat_event_bus::{ChatEventBus, CompactionEvent}; -use crate::chatbot::ChatOptions; use crate::config::CompactionConfig; use crate::config_store::GlobalConfigManager; use crate::db::{chat_history, chat_llm_tools, chat_summaries}; @@ -353,25 +352,28 @@ impl ContextCompactor { json!({ "role": "user", "content": conversation_text }), ]; - let options = ChatOptions { + let request = agent_loop::model::ModelRequest { + messages: messages_payload, + tools: Vec::new(), model: llm.model.clone(), max_tokens: None, temperature: Some(0.3), - session_id: Some(session_id), - stack_id: Some(stack_id), - user_id: None, - request_id: None, + request_id: uuid::Uuid::new_v4().to_string(), + conversation: agent_loop::ids::ConversationId::new(format!("session:{session_id}")), + frame: agent_loop::ids::FrameId(stack_id), + extras: serde_json::Value::Null, + log: Some(json!({ "session_id": session_id, "stack_id": stack_id })), }; - let turn = llm.client.chat_with_tools(&messages_payload, &[], &options).await + let resp = llm.client.complete(&request, None).await .map_err(|e| { warn!(stack_id, error = %e, "compactor: LLM call failed"); e })?; - let summary_text = match turn { - crate::chatbot::LlmTurn::Message(resp) => resp.content, - crate::chatbot::LlmTurn::ToolCalls { content, .. } => { + let summary_text = match resp { + agent_loop::model::ModelResponse::Message { content, .. } => content, + agent_loop::model::ModelResponse::ToolCalls { content, .. } => { warn!(stack_id, "compactor: unexpected tool calls in summary response, using content"); content } diff --git a/crates/skald-core/src/db/llm_requests/mod.rs b/crates/skald-core/src/db/llm_requests/mod.rs index e1df67a..527f684 100644 --- a/crates/skald-core/src/db/llm_requests/mod.rs +++ b/crates/skald-core/src/db/llm_requests/mod.rs @@ -1,7 +1,7 @@ //! DB operations for the `llm_requests` table (metadata only). //! //! Every `chat_with_tools` call is logged here by the -//! [`crate::chatbot::logging::LoggingChatbotClient`] wrapper. +//! [`crate::llm::logging::LoggingModel`] decorator. //! Payloads (request/response bodies + headers) live in `llm_request_payloads` //! in the owner bucket (`{userid}.db`), correlated by `request_id`. //! Rows are retained for `llm.request_log.retention_days` days (default 14). diff --git a/crates/skald-core/src/lib.rs b/crates/skald-core/src/lib.rs index 044e754..726bb0d 100644 --- a/crates/skald-core/src/lib.rs +++ b/crates/skald-core/src/lib.rs @@ -14,7 +14,6 @@ pub mod agents; pub mod approval; pub mod chat_event_bus; pub mod chat_hub; -pub mod chatbot; pub mod clarification; pub mod command; pub mod compactor; diff --git a/crates/skald-core/src/llm/logging.rs b/crates/skald-core/src/llm/logging.rs new file mode 100644 index 0000000..67126cf --- /dev/null +++ b/crates/skald-core/src/llm/logging.rs @@ -0,0 +1,113 @@ +//! Transparent logging decorator for any [`agent_loop::model::Model`]. +//! +//! [`LoggingModel`] intercepts every `complete` call, measures the duration, +//! and persists a **metadata-only** row to `llm_requests` in `system.db` +//! (fire-and-forget). Per-request correlation (session/stack/user id) travels +//! in [`ModelRequest::log`], set by the caller; the payload (request/response +//! bodies) is returned to the caller inside [`ModelResponse::raw`] / +//! [`ModelError::raw`] so it can be written to the user's own database. +//! +//! The split keeps conversation content (payloads) behind the user key while +//! metadata (cost, tokens, timing) stays in the admin-readable registry. +//! (Successor of `chatbot::logging::LoggingChatbotClient`, blueprint D13.) + +use std::sync::Arc; +use std::time::Instant; + +use async_trait::async_trait; +use sqlx::SqlitePool; +use tokio::sync::mpsc; +use tracing::warn; + +use agent_loop::model::{Model, ModelError, ModelRequest, ModelResponse, StreamDelta}; + +use crate::db::llm_requests; + +pub struct LoggingModel { + inner: Arc, + pool: Arc, + model_name: String, +} + +impl LoggingModel { + pub fn new(inner: Arc, pool: Arc, model_name: impl Into) -> Self { + Self { inner, pool, model_name: model_name.into() } + } +} + +#[async_trait] +impl Model for LoggingModel { + async fn complete( + &self, + req: &ModelRequest, + deltas: Option>, + ) -> Result { + let start = Instant::now(); + let result = self.inner.complete(req, deltas).await; + let duration_ms = start.elapsed().as_millis() as i64; + + // Per-request correlation set by the caller (llm_call / compactor). + let log = req.log.clone().unwrap_or_default(); + let session_id = log["session_id"].as_i64(); + let stack_id = log["stack_id"].as_i64(); + let user_id = log["user_id"].as_str().map(str::to_string); + let request_id = Some(req.request_id.clone()); + let model_name = self.model_name.clone(); + let pool = Arc::clone(&self.pool); + + match &result { + Ok(resp) => { + let usage = resp.usage(); + let (input_tokens, output_tokens, cache_read, cache_write) = ( + usage.input_tokens.map(|n| n as i64), + usage.output_tokens.map(|n| n as i64), + usage.cache_read.map(|n| n as i64), + usage.cache_write.map(|n| n as i64), + ); + tokio::spawn(async move { + if let Err(e) = llm_requests::insert(&pool, llm_requests::LlmRequestRow { + request_id, + user_id, + session_id, + stack_id, + model_name, + error_text: None, + input_tokens, + output_tokens, + duration_ms, + cache_read_tokens: cache_read, + cache_creation_tokens: cache_write, + }).await { + warn!(error = %e, "llm_requests: failed to insert log row"); + } + }); + } + Err(e) => { + let error_text = e.to_string(); + tokio::spawn(async move { + if let Err(log_err) = llm_requests::insert(&pool, llm_requests::LlmRequestRow { + request_id, + user_id, + session_id, + stack_id, + model_name, + error_text: Some(error_text), + input_tokens: None, + output_tokens: None, + duration_ms, + cache_read_tokens: None, + cache_creation_tokens: None, + }).await { + warn!(error = %log_err, "llm_requests: failed to insert error log row"); + } + }); + } + } + + result + } + + fn is_retriable(&self, err: &ModelError) -> bool { + self.inner.is_retriable(err) + } +} diff --git a/crates/skald-core/src/llm/manager.rs b/crates/skald-core/src/llm/manager.rs index 9d9cd1c..cb63755 100644 --- a/crates/skald-core/src/llm/manager.rs +++ b/crates/skald-core/src/llm/manager.rs @@ -8,11 +8,11 @@ use sqlx::SqlitePool; use tokio::sync::RwLock; use tracing::{info, warn}; -use crate::chatbot::ChatbotClient; -use crate::chatbot::logging::LoggingChatbotClient; +use agent_loop::model::Model; use core_api::provider::LlmStrength; use crate::provider::{ApiProvider, ProviderRegistry, ReasoningMode}; +use super::logging::LoggingModel; use super::providers::RemoteLlmModelInfo; use super::{ClientStatus, LlmEntry, LlmModelInfo, LlmModelRecord, LlmProviderInfo, LlmProviderRecord}; use super::db; @@ -512,8 +512,8 @@ fn build_entry( let prompt_cache = built.prompt_cache; let extra = model.extra_params.clone(); - let client: Arc = match log_pool { - Some(pool) => Arc::new(LoggingChatbotClient::new(inner, pool, &model.name)), + let client: Arc = match log_pool { + Some(pool) => Arc::new(LoggingModel::new(inner, pool, &model.name)), None => inner, }; diff --git a/crates/skald-core/src/llm/mod.rs b/crates/skald-core/src/llm/mod.rs index 4997483..5041a41 100644 --- a/crates/skald-core/src/llm/mod.rs +++ b/crates/skald-core/src/llm/mod.rs @@ -1,10 +1,12 @@ pub(crate) mod db; +pub mod logging; pub mod manager; pub mod providers; use std::sync::Arc; -use crate::chatbot::ChatbotClient; +use agent_loop::model::Model; + use crate::provider::ServiceType; pub use core_api::provider::{LlmProviderRecord, LlmModelRecord, LlmStrength, ReasoningMode}; @@ -13,7 +15,7 @@ pub use manager::{LlmManager, sort_models_for_agent}; /// A resolved, ready-to-use LLM client with its associated metadata. #[derive(Clone)] pub struct LlmEntry { - pub client: Arc, + pub client: Arc, pub model: String, pub model_db_id: i64, pub strength: Option, diff --git a/crates/skald-core/src/llm/providers/anthropic.rs b/crates/skald-core/src/llm/providers/anthropic.rs index 937a6c2..db070b7 100644 --- a/crates/skald-core/src/llm/providers/anthropic.rs +++ b/crates/skald-core/src/llm/providers/anthropic.rs @@ -2,7 +2,7 @@ use std::sync::Arc; use anyhow::{Context, Result, anyhow}; -use crate::chatbot::anthropic::AnthropicClient; +use agent_loop::models::AnthropicModel; use crate::llm::{LlmModelRecord, LlmProviderRecord}; use crate::llm::providers::{RemoteLlmModelInfo, extra_with_reasoning}; use crate::provider::{ApiProvider, BuiltLlmClient, ProviderField, ProviderUiMeta, ReasoningMode, ServiceType}; @@ -111,7 +111,7 @@ impl ApiProvider for AnthropicProvider { // stays uncached, as before. let prompt_cache = model.capabilities.iter().any(|c| c == "tool_search"); Ok(BuiltLlmClient { - client: Arc::new(AnthropicClient::with_extra_body(key, extra)), + client: Arc::new(AnthropicModel::with_extra_body(key, model.model_id.clone(), extra)), prompt_cache, }) })()) diff --git a/crates/skald-core/src/llm/providers/declared.rs b/crates/skald-core/src/llm/providers/declared.rs index 926877e..cbc9376 100644 --- a/crates/skald-core/src/llm/providers/declared.rs +++ b/crates/skald-core/src/llm/providers/declared.rs @@ -18,7 +18,7 @@ use std::sync::Arc; use anyhow::{anyhow, Context, Result}; use tracing::{info, warn}; -use crate::chatbot::openai::OpenAiClient; +use agent_loop::models::OpenAiModel; use crate::llm::providers::{extra_with_reasoning, RemoteLlmModelInfo}; use crate::llm::{LlmModelRecord, LlmProviderRecord}; use crate::provider::{ @@ -575,7 +575,7 @@ impl ApiProvider for DeclaredProvider { let extra = extra_with_reasoning(self, model); let prompt_cache = self.spec.prompt_cache; Ok(BuiltLlmClient { - client: Arc::new(OpenAiClient::new(self.base_url(record), key, extra, prompt_cache)), + client: Arc::new(OpenAiModel::with_options(self.base_url(record), key, model.model_id.clone(), extra, prompt_cache)), prompt_cache, }) })()) diff --git a/crates/skald-core/src/llm/providers/mod.rs b/crates/skald-core/src/llm/providers/mod.rs index d6b6b3a..0be1b2b 100644 --- a/crates/skald-core/src/llm/providers/mod.rs +++ b/crates/skald-core/src/llm/providers/mod.rs @@ -15,7 +15,7 @@ use anyhow::{anyhow, Context, Result}; use core_api::provider::{ApiProvider, BuiltLlmClient, LlmModelRecord, LlmProviderRecord}; -use crate::chatbot::openai::OpenAiClient; +use agent_loop::models::OpenAiModel; /// Computes the `extra_params` an OpenAI-compatible client should be built with, /// given a model's stored `extra_params` and its selected reasoning value. The @@ -75,7 +75,7 @@ pub(crate) async fn fetch_openai_models( .ok_or_else(|| anyhow!("unexpected {who} response shape")) } -/// Builds an `OpenAiClient` for an OpenAI-compatible provider: requires the +/// Builds an `OpenAiModel` for an OpenAI-compatible provider: requires the /// provider record's `api_key` and merges the model's stored `extra_params` /// with the provider-translated reasoning fragment (see `extra_with_reasoning`). pub(crate) fn build_openai_llm( @@ -89,7 +89,7 @@ pub(crate) fn build_openai_llm( .with_context(|| format!("provider '{}': api_key required for {}", record.name, provider.type_id()))?; let extra = extra_with_reasoning(provider, model); Ok(BuiltLlmClient { - client: Arc::new(OpenAiClient::new(base_url, key, extra, prompt_cache)), + client: Arc::new(OpenAiModel::with_options(base_url, key, model.model_id.clone(), extra, prompt_cache)), prompt_cache, }) } diff --git a/crates/skald-core/src/llm/providers/ollama.rs b/crates/skald-core/src/llm/providers/ollama.rs index a80f704..e47bf9c 100644 --- a/crates/skald-core/src/llm/providers/ollama.rs +++ b/crates/skald-core/src/llm/providers/ollama.rs @@ -2,7 +2,7 @@ use std::sync::Arc; use anyhow::{Result, anyhow}; -use crate::chatbot::ollama::OllamaClient; +use agent_loop::models::OllamaModel; use crate::llm::{LlmModelRecord, LlmProviderRecord}; use crate::llm::providers::RemoteLlmModelInfo; use crate::provider::{ApiProvider, BuiltLlmClient, ProviderField, ProviderUiMeta, ServiceType}; @@ -101,9 +101,9 @@ impl ApiProvider for OllamaProvider { Ok(Some(Self::parse_model_info(&resp, model_id))) } - fn build_llm(&self, record: &LlmProviderRecord, _model: &LlmModelRecord) -> Option> { + fn build_llm(&self, record: &LlmProviderRecord, model: &LlmModelRecord) -> Option> { Some(Ok(BuiltLlmClient { - client: Arc::new(OllamaClient::new(record.base_url.as_deref())), + client: Arc::new(OllamaModel::new(record.base_url.as_deref(), model.model_id.clone())), prompt_cache: false, })) } diff --git a/crates/skald-core/src/session/handler/llm_call.rs b/crates/skald-core/src/session/handler/llm_call.rs index 926c46e..d80de87 100644 --- a/crates/skald-core/src/session/handler/llm_call.rs +++ b/crates/skald-core/src/session/handler/llm_call.rs @@ -3,17 +3,21 @@ //! Extracted from `run_agent_turn`: on a retriable error (5xx / network) it retries //! up to `MAX_LLM_ATTEMPTS` models in priority order, rebuilding the message list //! when the replacement model has a different `prompt_cache` setting, and emits -//! `ModelFallback` / `LlmFailed` along the way. +//! `ModelFallback` / `LlmFailed` along the way. The call itself goes through the +//! `agent_loop::model::Model` trait (blueprint D13) — clients and protocols live +//! in the `agent-loop` crate. use std::collections::HashSet; use std::sync::Arc; -use serde_json::Value; +use serde_json::{Value, json}; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; use tracing::{error, warn}; -use crate::chatbot::{ChatOptions, LlmError, LlmTurn, StreamDelta}; +use agent_loop::ids::{ConversationId, FrameId}; +use agent_loop::model::{ModelRequest, ModelResponse, StreamDelta}; + use crate::db::llm_request_payloads; use crate::events::{ServerEvent, TokenDeltaKind}; use crate::llm::{LlmEntry, LlmStrength}; @@ -24,8 +28,9 @@ use super::interface_tools::AgentRunConfig; /// Outcome of one round's LLM call. pub(super) enum RoundLlm { - /// The model responded (message or tool calls). - Turn(LlmTurn), + /// The model responded (message or tool calls). Boxed: `ModelResponse` + /// dwarfs the other variants. + Turn(Box), /// The turn was cancelled (`/stop`) while the request was in flight. Cancelled, /// All fallback attempts were exhausted, or an error is non-retriable. @@ -59,15 +64,6 @@ impl ChatSessionHandler { // a fallback across DTL modes must re-shape (deferred candidates or not). let cur_tool_defs = config.all_tool_defs(cur_llm.dtl); let request_id = uuid::Uuid::new_v4().to_string(); - let options = ChatOptions { - model: cur_llm.model.clone(), - max_tokens: None, - temperature: None, - session_id: Some(self.session_id), - stack_id: Some(stack_id), - user_id: Some(self.user_id.clone()), - request_id: Some(request_id.clone()), - }; // Tell the model, in read_file's description, which media formats it can // open directly — keyed on the model actually serving this attempt, so a @@ -80,6 +76,23 @@ impl ChatSessionHandler { // the fallback reassignment below. On cancel we drop the future // (aborting the request) and return immediately. let client = cur_llm.client.clone(); + let request = ModelRequest { + messages: messages.clone(), + tools: defs.to_vec(), + model: cur_llm.model.clone(), + max_tokens: None, + temperature: None, + request_id: request_id.clone(), + conversation: ConversationId::new(format!("session:{}", self.session_id)), + frame: FrameId(stack_id), + extras: Value::Null, + // Correlation for the LoggingModel decorator (never sent). + log: Some(json!({ + "session_id": self.session_id, + "stack_id": stack_id, + "user_id": self.user_id, + })), + }; // Streaming side-channel: providers that support SSE push deltas here; // the forwarder re-emits them as `TokenDelta` events on the turn bus. // Best-effort — the round's final events remain authoritative. @@ -87,7 +100,7 @@ impl ChatSessionHandler { let forwarder = spawn_delta_forwarder(delta_rx, em.sender()); let call_result = tokio::select! { _ = token.cancelled() => return RoundLlm::Cancelled, - r = client.chat_with_tools_raw_streaming(messages.as_slice(), defs, &options, delta_tx) => r, + r = client.complete(&request, Some(delta_tx)) => r, }; // The client's sender dropped with the completed future: the forwarder // drains any queued deltas and exits, so every `TokenDelta` precedes the @@ -95,39 +108,39 @@ impl ChatSessionHandler { forwarder.await.ok(); let e = match call_result { - Ok((turn, meta)) => { + Ok(resp) => { self.llm_manager.mark_success(cur_name).await; // Persist the payload (request/response bodies + headers) to the // user's own database. Fire-and-forget — a failed write must not // break the turn. The metadata row is already written by the - // logging wrapper to system.db with the same request_id. - if let Some(meta) = meta { + // LoggingModel decorator to system.db with the same request_id. + if let Some(meta) = resp.raw() { let pool = Arc::clone(&self.db); let rid = request_id.clone(); + let row = llm_request_payloads::PayloadRow { + request_id: rid, + request_json: meta.request_body.as_ref().map(|v| v.to_string()).unwrap_or_default(), + request_headers: meta.request_headers.as_ref().map(|v| v.to_string()), + response_json: meta.response_body.as_ref().map(|v| v.to_string()), + response_headers: meta.response_headers.as_ref().map(|v| v.to_string()), + }; tokio::spawn(async move { - let row = llm_request_payloads::PayloadRow { - request_id: rid, - request_json: meta.request_body.map(|v| v.to_string()).unwrap_or_default(), - request_headers: meta.request_headers.map(|v| v.to_string()), - response_json: meta.response_body.map(|v| v.to_string()), - response_headers: meta.response_headers.map(|v| v.to_string()), - }; if let Err(e) = llm_request_payloads::insert(&pool, row).await { tracing::warn!(error = %e, "llm_request_payloads: failed to insert"); } }); } - return RoundLlm::Turn(turn); + return RoundLlm::Turn(Box::new(resp)); } Err(e) => e, }; // Persist the payload even on failure so the debug log shows the request - // that was rejected (e.g. a provider 400). Only the HTTP clients attach a - // body (`LlmError::raw_meta`); a network/parse/cancel error carries none. + // that was rejected (e.g. a provider 400). Only HTTP failures attach a + // body (`ModelError::raw`); a network/parse/cancel error carries none. // Fire-and-forget, keyed on the same `request_id` as the metadata row the - // logging wrapper wrote to system.db. - if let Some(meta) = e.downcast_ref::().and_then(|le| le.raw_meta.as_ref()) { + // LoggingModel decorator wrote to system.db. + if let Some(meta) = e.raw.as_ref() { let row = llm_request_payloads::PayloadRow { request_id: request_id.clone(), request_json: meta.request_body.as_ref().map(|v| v.to_string()).unwrap_or_default(), @@ -147,10 +160,10 @@ impl ChatSessionHandler { self.llm_manager.mark_failure(cur_name, &e.to_string()).await; let can_fallback = tried_this_round.len() < MAX_LLM_ATTEMPTS - && is_retriable_llm_error(&e); + && client.is_retriable(&e); if !can_fallback { em.llm_failed(tried_this_round.clone(), e.to_string()).await; - return RoundLlm::Failed(e); + return RoundLlm::Failed(e.into()); } let excluded: Vec<&str> = tried_this_round.iter().map(String::as_str).collect(); @@ -179,7 +192,7 @@ impl ChatSessionHandler { } Err(_) => { em.llm_failed(tried_this_round.clone(), e.to_string()).await; - return RoundLlm::Failed(e); + return RoundLlm::Failed(e.into()); } } } @@ -206,21 +219,6 @@ fn spawn_delta_forwarder( }) } -/// Whether an LLM error is worth retrying on a different model. -/// -/// Classifies on the real HTTP status ([`crate::chatbot::http_status`]), not a -/// substring of the message — a model id or token count containing "404"/"401" no -/// longer mis-classifies (bug B6). A non-HTTP failure (network, parse) has no status -/// and is retriable, matching the previous default. -fn is_retriable_llm_error(e: &anyhow::Error) -> bool { - // Never retry these client errors — the request itself is unauthorized, not - // found, or unprocessable. 400 is intentionally NOT listed: some providers - // reject valid requests that others accept (e.g. DeepSeek requires a - // reasoning_content echo, OpenAI does not), so retrying elsewhere can succeed. - // 429 and 5xx stay retriable (a different model / provider may serve the call). - !matches!(crate::chatbot::http_status(e), Some(401 | 403 | 404 | 422)) -} - fn first_line(s: &str) -> String { s.lines().next().unwrap_or(s).to_string() } @@ -246,38 +244,40 @@ fn media_annotated_tools(tool_defs: &[Value], capabilities: &[String]) -> Option #[cfg(test)] mod tests { - use super::is_retriable_llm_error; - use crate::chatbot::LlmError; + use agent_loop::model::{Model, ModelError, ModelRequest, ModelResponse, StreamDelta}; + use async_trait::async_trait; + use tokio::sync::mpsc; - fn http_err(status: u16, message: &str) -> anyhow::Error { - LlmError { status: Some(status), message: message.to_string(), ..Default::default() }.into() + struct Dummy; + #[async_trait] + impl agent_loop::model::Model for Dummy { + async fn complete( + &self, + _req: &ModelRequest, + _d: Option>, + ) -> Result { + unreachable!() + } } + /// Retriability classification lives on the `Model` trait default (the crate + /// owns the protocols, blueprint D13): 401/403/404/422 don't retry, + /// 400/429/5xx/network do. Classification keys on the structured status, + /// never on the message string (bug B6 regression). #[test] - fn client_errors_are_not_retried() { + fn retriability_keys_on_structured_status() { + let m = Dummy; for code in [401, 403, 404, 422] { - assert!(!is_retriable_llm_error(&http_err(code, "nope")), "{code} must not retry"); + assert!(!m.is_retriable(&ModelError::new(Some(code), "nope")), "{code} must not retry"); } - } - - #[test] - fn server_rate_limit_and_400_retry() { for code in [400, 429, 500, 502, 503] { - assert!(is_retriable_llm_error(&http_err(code, "retry")), "{code} must retry"); + assert!(m.is_retriable(&ModelError::new(Some(code), "retry")), "{code} must retry"); } - } - - #[test] - fn non_http_errors_retry() { - assert!(is_retriable_llm_error(&anyhow::anyhow!("connection reset by peer"))); - } - - #[test] - fn status_digits_in_the_message_do_not_mislead() { - // Regression for B6: the old substring check read any "404"/"401" in the text - // as a client error. A 500 whose body mentions "1401 tokens" / "code 404" must - // still retry — classification keys on the structured status, not the string. - let e = http_err(500, "provider error: too many (1401) tokens, see code 404 in docs"); - assert!(is_retriable_llm_error(&e)); + // A 500 whose body mentions "1401 tokens" / "code 404" must still retry. + assert!(m.is_retriable(&ModelError::new( + Some(500), + "provider error: too many (1401) tokens, see code 404 in docs" + ))); + assert!(m.is_retriable(&ModelError::new(None, "connection reset by peer"))); } } diff --git a/crates/skald-core/src/session/handler/llm_loop.rs b/crates/skald-core/src/session/handler/llm_loop.rs index 6220c32..83d469d 100644 --- a/crates/skald-core/src/session/handler/llm_loop.rs +++ b/crates/skald-core/src/session/handler/llm_loop.rs @@ -4,7 +4,7 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, trace}; use crate::chat_event_bus::ToolCallEvent; -use crate::chatbot::{LlmTurn, ToolCall}; +use agent_loop::model::{ModelResponse, ToolCall}; use crate::db::{chat_history, chat_llm_tools}; use crate::events::ServerEvent; use crate::tools::{ @@ -139,36 +139,37 @@ impl ChatSessionHandler { RoundLlm::Failed(e) => return Err(e), }; - match turn_result { - LlmTurn::Message(resp) => { + match *turn_result { + ModelResponse::Message { content, reasoning, usage, .. } => { let message_id = chat_history::append( - pool, stack_id, &chat_history::Role::Assistant, &resp.content, false, - resp.reasoning_content.as_deref(), + pool, stack_id, &chat_history::Role::Assistant, &content, false, + reasoning.as_deref(), ).await?; - if let (Some(i), Some(o)) = (resp.input_tokens, resp.output_tokens) { - chat_history::set_usage(pool, message_id, i, o, 0, resp.cost).await?; + if let (Some(i), Some(o)) = (usage.input_tokens, usage.output_tokens) { + chat_history::set_usage(pool, message_id, i, o, 0, usage.cost_usd).await?; } return Ok(TurnOutcome::Final { - content: resp.content, + content, message_id, - input_tokens: resp.input_tokens, - output_tokens: resp.output_tokens, - truncated: resp.truncated, - reasoning_content: resp.reasoning_content, + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + truncated: usage.truncated, + reasoning_content: reasoning, tool_calls: all_tool_calls, }); } - LlmTurn::ToolCalls { content: assistant_text, calls, input_tokens, output_tokens, reasoning_content, cost, .. } => { + ModelResponse::ToolCalls { content: assistant_text, calls, usage, reasoning, .. } => { + let (input_tokens, output_tokens) = (usage.input_tokens, usage.output_tokens); let message_id = chat_history::append( pool, stack_id, &chat_history::Role::Assistant, &assistant_text, false, - reasoning_content.as_deref(), + reasoning.as_deref(), ).await?; if let (Some(i), Some(o)) = (input_tokens, output_tokens) { - chat_history::set_usage(pool, message_id, i, o, 0, cost).await?; + chat_history::set_usage(pool, message_id, i, o, 0, usage.cost_usd).await?; } if !assistant_text.trim().is_empty() || input_tokens.is_some() { - em.thinking(message_id, assistant_text, input_tokens, output_tokens, reasoning_content).await; + em.thinking(message_id, assistant_text, input_tokens, output_tokens, reasoning).await; } // A homogeneous batch of ≥2 synchronous sub-agent calls is fanned