From 4dddc7d2ab5f9c7e3cc50c534b258537a1470889 Mon Sep 17 00:00:00 2001 From: xavix-yo Date: Wed, 22 Jul 2026 14:05:32 +0100 Subject: [PATCH] streaming: add chat_with_tools_raw_streaming to LM Studio + LoggingChatbotClient LM Studio: delegate streaming to the inner OpenAI client. LoggingChatbotClient: extract shared log_and_return helper, add streaming override so deltas are forwarded and metadata still logged. --- crates/llm-client/src/lm_studio.rs | 16 +++- crates/skald-core/src/chatbot/logging.rs | 98 +++++++++++++++--------- 2 files changed, 78 insertions(+), 36 deletions(-) diff --git a/crates/llm-client/src/lm_studio.rs b/crates/llm-client/src/lm_studio.rs index 11c27e9..8c4b63b 100644 --- a/crates/llm-client/src/lm_studio.rs +++ b/crates/llm-client/src/lm_studio.rs @@ -1,7 +1,8 @@ use async_trait::async_trait; use serde_json::Value; +use tokio::sync::mpsc; -use crate::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, openai::OpenAiClient}; +use crate::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, StreamDelta, openai::OpenAiClient}; /// LM Studio client. /// @@ -48,4 +49,17 @@ impl ChatbotClient for LmStudioClient { ) -> 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/skald-core/src/chatbot/logging.rs b/crates/skald-core/src/chatbot/logging.rs index c950002..30a086a 100644 --- a/crates/skald-core/src/chatbot/logging.rs +++ b/crates/skald-core/src/chatbot/logging.rs @@ -9,16 +9,17 @@ //! metadata (cost, tokens, timing) stays in the admin-readable registry. use std::sync::Arc; -use std::time::Instant; +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}; +use super::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, StreamDelta}; // ───────────────────────────────────────────────────────────────────────────── @@ -36,43 +37,16 @@ impl LoggingChatbotClient { ) -> Self { Self { inner, pool, model_name: model_name.into() } } -} -#[async_trait] -impl ChatbotClient for LoggingChatbotClient { - /// Passthrough — logging only applies to the tool-calling path. - async fn chat( + /// 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, - 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, + duration: Duration, + result: anyhow::Result<(LlmTurn, Option)>, ) -> anyhow::Result<(LlmTurn, Option)> { - let start = Instant::now(); - let result = self.inner.chat_with_tools_raw(messages, tools, options).await; - let duration_ms = start.elapsed().as_millis() as i64; + let duration_ms = duration.as_millis() as i64; let session_id = options.session_id; let stack_id = options.stack_id; @@ -136,3 +110,57 @@ impl ChatbotClient for LoggingChatbotClient { } } } + +#[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 + } +}