diff --git a/CLAUDE.md b/CLAUDE.md index c4a6b6a..87e742e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -188,6 +188,16 @@ Uploads (`POST /api/{source}/uploads`) are saved per-user under `data/uploads/{u At context-build time (`MessageBuilder`), attachments of the **current turn** (the user/agent rows following the last completed assistant reply, including across in-flight tool rounds) are partitioned by `session/handler/media.rs`: when the resolved model's `LlmEntry.capabilities` include the modality (`vision` → `image_url` parts, `video` → `video_url` parts), the file is inlined as a base64 data-URL content part — but only if it canonicalizes under `data/uploads/`, its sniffed MIME is in the allowlist, and it fits the budgets (4 files / 10 MiB image / 32 MiB video / 48 MiB total per turn). Everything else — older turns, other kinds, any failed check — keeps the textual `[SYSTEM INFO]` path block, so a non-vision model produces a byte-identical payload to before. `OpenAiClient` forwards parts verbatim; `AnthropicClient` translates `image_url` data URLs to `image` blocks (video unsupported; Anthropic models get `vision` by editing the model row's capabilities — no catalog refresh writes them). On LLM fallback mid-round, messages are rebuilt with the replacement model's capabilities. +## Token streaming & reasoning display + +The chat streams tokens live, as a **parallel best-effort side-channel** that never alters the turn's authoritative flow: the final `Done` (or `Thinking`) event still carries the complete content and the frontend treats it as truth. + +- **Client seam** (`core-api::chatbot`): `ChatbotClient::chat_with_tools_raw_streaming(..., delta_tx: mpsc::Sender)` — default impl ignores the channel and calls the buffered `chat_with_tools_raw`, so providers without streaming (Ollama, LM Studio) are untouched. `StreamDelta::{Text, Reasoning}` splits visible answer from chain-of-thought. Senders use `try_send` (deltas drop when the channel is full) — streaming must never backpressure the HTTP read. +- **SSE implementations** (`crates/llm-client`): `OpenAiClient` (`stream:true` + `stream_options.include_usage`, `reasoning_content`/`reasoning` deltas, index-based `tool_calls` accumulation, usage from the final chunk) and `AnthropicClient` (`stream:true`; `message_start`/`content_block_*`/`message_delta` events; `thinking_delta` → reasoning, `input_json_delta` → tool input). Both reassemble the **same `LlmTurn` + `LlmRawMeta`** the buffered path returns (the payload log stores a synthesized buffered-shaped body). Failure policy: if the stream dies **before any delta** the client retries buffered on the same model (providers rejecting `stream` keep working); a mid-stream failure propagates to the normal model-fallback logic. Framing is shared (`llm_client::SseDecoder`). Anthropic's **buffered** path now also parses `thinking` blocks into `reasoning_content` (previously discarded). +- **Loop wiring**: `call_llm_round` creates the delta channel per attempt and a forwarder task maps deltas to `ServerEvent::TokenDelta { kind: content|reasoning, delta }` on the turn's event channel (drained before the round's outcome events, so ordering holds); cancellation drops the in-flight future as before. A mid-stream fallback is handled client-side: the frontend clears its pending bubble on `model_fallback`. +- **Reasoning surfacing**: `reasoning_content` rides `Done`/`Thinking` events (so buffered providers show it live too) and is projected as `reasoning` on assistant/thinking history items (`build_items`); persistence in `chat_history.reasoning_content` and the echo back into context predate this feature. +- **Frontend** (`chat-session.js` + `copilot-render.js`, shared by desktop copilot and mobile chat-page): `token_delta` accumulates into a pending assistant bubble (in-place mutation + ~15 Hz flush, blinking caret); `done`/`thinking` finalize it in place, `error`/`llm_failed`/`model_fallback` drop it, `tool_start`/`agent_done` finalize orphan bubbles (reasoning-only rounds, sub-agent final rounds that emit no `Done`). The reasoning block is a muted, collapsed-by-default native `
` (`renderReasoning`, `.reasoning-block` in `copilot-messages.css`, i18n key `chat.reasoning`) — open state survives re-renders, and it renders identically from live events and from history. + ## Sub-agent system - Synchronous sub-agents (`execute_task` mode=sync / `execute_subtask`) are **not** plain `Tool`s — they are intercepted in `run_agent_turn` before registry dispatch. - `dispatch_sub_agent` (in `agent_dispatch.rs`) creates a child `chat_sessions_stack` row and runs `run_agent_turn` **recursively in the same task**, holding the same `processing` lock and sharing the same cancellation token. The child's result string becomes the parent tool call's result (completion lives in one place — the `run_agent_turn` tool-result match); then it terminates the child frame. There is no task-spawn / `WaitingChild` / resume cascade for the sync path. diff --git a/Cargo.lock b/Cargo.lock index 58f9f89..700a230 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2189,9 +2189,11 @@ dependencies = [ "anyhow", "async-trait", "core-api", + "futures-util", "reqwest 0.13.4", "serde", "serde_json", + "tokio", "tracing", ] @@ -3613,7 +3615,7 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams", + "wasm-streams 0.4.2", "web-sys", "webpki-roots 1.0.7", ] @@ -3650,12 +3652,14 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-rustls", + "tokio-util", "tower", "tower-http 0.6.11", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams 0.5.0", "web-sys", ] @@ -6087,6 +6091,19 @@ dependencies = [ "web-sys", ] +[[package]] +name = "wasm-streams" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wasmparser" version = "0.244.0" diff --git a/crates/core-api/src/chatbot.rs b/crates/core-api/src/chatbot.rs index 8a6a5ae..3e2f0eb 100644 --- a/crates/core-api/src/chatbot.rs +++ b/crates/core-api/src/chatbot.rs @@ -1,5 +1,6 @@ use async_trait::async_trait; use serde_json::Value; +use tokio::sync::mpsc; /// A single message in a conversation. #[derive(Debug, Clone)] @@ -90,6 +91,17 @@ pub struct ToolCall { 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 { @@ -160,4 +172,22 @@ pub trait ChatbotClient: Send + Sync { ) -> 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/events.rs b/crates/core-api/src/events.rs index dc6fc54..f88d5c8 100644 --- a/crates/core-api/src/events.rs +++ b/crates/core-api/src/events.rs @@ -34,6 +34,16 @@ pub struct GlobalEvent { // ── Server → Client ─────────────────────────────────────────────────────────── +/// Which token stream a [`ServerEvent::TokenDelta`] belongs to. +#[derive(Clone, Copy, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum TokenDeltaKind { + /// Visible answer text. + Content, + /// Model chain-of-thought (reasoning/thinking tokens). + Reasoning, +} + #[derive(Clone, Serialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum ServerEvent { @@ -117,6 +127,10 @@ pub enum ServerEvent { content: String, input_tokens: Option, output_tokens: Option, + /// Chain-of-thought, when the model produced any. Lets non-streaming + /// providers surface the reasoning block live, not just from history. + #[serde(default, skip_serializing_if = "Option::is_none")] + reasoning_content: Option, }, /// A fatal error occurred processing the request. Error { @@ -132,6 +146,17 @@ pub enum ServerEvent { content: String, input_tokens: Option, output_tokens: Option, + /// Chain-of-thought of this tool-call round, when produced. + #[serde(default, skip_serializing_if = "Option::is_none")] + reasoning_content: Option, + }, + /// One incremental token while the assistant response (or its reasoning) + /// is being generated. Best-effort: deltas ride the lossy broadcast bus and + /// a lagging client may miss some — the final `Done` is always authoritative + /// and carries the complete content. + TokenDelta { + kind: TokenDeltaKind, + delta: String, }, /// A write operation requires user approval before executing (shows a diff). PendingWrite { @@ -279,6 +304,7 @@ impl ServerEvent { Self::Done { .. } => "done", Self::Error { .. } => "error", Self::Thinking { .. } => "thinking", + Self::TokenDelta { .. } => "token_delta", Self::PendingWrite { .. } => "pending_write", Self::ApprovalRequired { .. } => "approval_required", Self::AgentQuestion { .. } => "agent_question", diff --git a/crates/llm-client/Cargo.toml b/crates/llm-client/Cargo.toml index 0f2228d..48640c8 100644 --- a/crates/llm-client/Cargo.toml +++ b/crates/llm-client/Cargo.toml @@ -5,9 +5,11 @@ edition = "2024" [dependencies] core-api = { path = "../core-api" } -reqwest = { version = "0.13", default-features = false, features = ["rustls-no-provider", "charset", "http2", "system-proxy", "json"] } +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 index 5da699a..1c8562a 100644 --- a/crates/llm-client/src/anthropic.rs +++ b/crates/llm-client/src/anthropic.rs @@ -1,8 +1,12 @@ +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, ToolCall, headers_to_json, redact_key}; +use crate::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, Role, SseDecoder, StreamDelta, ToolCall, headers_to_json, redact_key}; const DEFAULT_BASE_URL: &str = "https://api.anthropic.com"; const ANTHROPIC_VERSION: &str = "2023-06-01"; @@ -157,6 +161,242 @@ impl AnthropicClient { 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.into(); } + 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 a single `system:` string. The Anthropic + /// API only accepts a single system parameter. + fn merged_system(messages: &[Value]) -> Option { + let parts: Vec<&str> = messages + .iter() + .filter(|m| m["role"].as_str() == Some("system")) + .filter_map(|m| m["content"].as_str()) + .collect(); + if parts.is_empty() { None } else { Some(parts.join("\n\n---\n\n")) } + } + + 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", + }) + } + + 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? + .error_for_status() + } + + /// 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()); + + /// 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 @@ -274,7 +514,7 @@ impl ChatbotClient for AnthropicClient { let content = resp["content"] .as_array() - .and_then(|arr| arr.first()) + .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(); @@ -304,58 +544,22 @@ impl ChatbotClient for AnthropicClient { tools: &[Value], options: &ChatOptions, ) -> anyhow::Result<(LlmTurn, Option)> { - // Collect ALL system-role messages (main prompt, mid-conversation - // summary, tail_reminder) and merge them into a single `system:` - // string. The Anthropic API only accepts a single system parameter; - // mid-conversation system messages generated by build_openai_messages - // are intentionally used for injecting compaction summaries and tail - // reminders — they must not be silently dropped. - let system: Option = { - let parts: Vec<&str> = messages - .iter() - .filter(|m| m["role"].as_str() == Some("system")) - .filter_map(|m| m["content"].as_str()) - .collect(); - if parts.is_empty() { None } else { Some(parts.join("\n\n---\n\n")) } - }; - + // 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); - let max_tokens = options.max_tokens.unwrap_or(4096); - let mut body = json!({ - "model": options.model, - "max_tokens": max_tokens, - "messages": anthropic_messages, - "tools": anthropic_tools, - }); - - 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, 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 = json!({ - "x-api-key": redact_key(&self.api_key), - "anthropic-version": ANTHROPIC_VERSION, - "content-type": "application/json", - }); + let request_headers = self.logged_headers(); - let http_resp = self - .http - .post(&url) - .header("x-api-key", &self.api_key) - .header("anthropic-version", ANTHROPIC_VERSION) - .header("X-Title", core_api::APP_NAME) - .json(&body) - .send() - .await? - .error_for_status()?; + let http_resp = self.send_request(&body).await?; let response_headers = headers_to_json(http_resp.headers()); let resp_text = http_resp.text().await?; @@ -383,6 +587,7 @@ impl ChatbotClient for AnthropicClient { } 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. @@ -404,7 +609,7 @@ impl ChatbotClient for AnthropicClient { }) .collect(); - LlmTurn::ToolCalls { content: text, calls, input_tokens, output_tokens, reasoning_content: None, cache_read_tokens, cache_creation_tokens, cost } + LlmTurn::ToolCalls { content: text, calls, input_tokens, output_tokens, reasoning_content, cache_read_tokens, cache_creation_tokens, cost } } else { let content = content_blocks .iter() @@ -414,17 +619,55 @@ impl ChatbotClient for AnthropicClient { .to_string(); let truncated = stop_reason == "max_tokens"; - LlmTurn::Message(ChatResponse { content, input_tokens, output_tokens, truncated, reasoning_content: None, cache_read_tokens, cache_creation_tokens, cost }) + 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")); diff --git a/crates/llm-client/src/lib.rs b/crates/llm-client/src/lib.rs index 2cd9aa9..274fd29 100644 --- a/crates/llm-client/src/lib.rs +++ b/crates/llm-client/src/lib.rs @@ -6,11 +6,54 @@ 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, ToolCall, + 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 @@ -72,3 +115,41 @@ pub fn http_status(err: &anyhow::Error) -> Option { } 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/openai.rs b/crates/llm-client/src/openai.rs index d8c0b9f..bfc3cc7 100644 --- a/crates/llm-client/src/openai.rs +++ b/crates/llm-client/src/openai.rs @@ -1,8 +1,12 @@ +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, ToolCall, headers_to_json, redact_key}; +use crate::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, Role, SseDecoder, StreamDelta, ToolCall, headers_to_json, redact_key}; use core_api::APP_NAME; /// OpenAI ChatGPT client (also compatible with any OpenAI-spec endpoint). @@ -45,6 +49,205 @@ impl OpenAiClient { 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(), + ), + }.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)"); + } + + 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 }], + "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] @@ -123,53 +326,16 @@ impl ChatbotClient for OpenAiClient { tools: &[Value], options: &ChatOptions, ) -> anyhow::Result<(LlmTurn, Option)> { - let mut body = json!({ - "model": options.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(); - } - - 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); + 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 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(); - } let request_body = body.clone(); - let request_headers = logged_headers; + let request_headers = self.logged_headers(); - 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"); - } - let http_resp = req - .json(&body) - .send() - .await?; + let http_resp = self.send_request(&body).await?; let response_headers = headers_to_json(http_resp.headers()); let status = http_resp.status(); @@ -262,4 +428,26 @@ impl ChatbotClient for OpenAiClient { 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/src/chatbot/mod.rs b/crates/skald-core/src/chatbot/mod.rs index b3d04c5..f66abd5 100644 --- a/crates/skald-core/src/chatbot/mod.rs +++ b/crates/skald-core/src/chatbot/mod.rs @@ -2,6 +2,6 @@ pub mod logging; // Re-export from the independent llm-client crate. pub use llm_client::{ - ChatOptions, ChatResponse, ChatbotClient, LlmError, LlmRawMeta, LlmTurn, Message, ToolCall, - anthropic, http_status, lm_studio, ollama, openai, + ChatOptions, ChatResponse, ChatbotClient, LlmError, LlmRawMeta, LlmTurn, Message, StreamDelta, + ToolCall, anthropic, http_status, lm_studio, ollama, openai, }; diff --git a/crates/skald-core/src/events.rs b/crates/skald-core/src/events.rs index 41941b9..6b4d2b2 100644 --- a/crates/skald-core/src/events.rs +++ b/crates/skald-core/src/events.rs @@ -1,3 +1,3 @@ pub use core_api::events::{ - ClientMessage, GlobalEvent, InboundDataMessage, ServerEvent, + ClientMessage, GlobalEvent, InboundDataMessage, ServerEvent, TokenDeltaKind, }; diff --git a/crates/skald-core/src/session/handler/emitter.rs b/crates/skald-core/src/session/handler/emitter.rs index a5304f4..434fe82 100644 --- a/crates/skald-core/src/session/handler/emitter.rs +++ b/crates/skald-core/src/session/handler/emitter.rs @@ -42,13 +42,19 @@ impl<'a> TurnEmitter<'a> { } /// The assistant produced text alongside tool calls (reasoning before acting). - pub(super) async fn thinking(&self, message_id: i64, content: String, input_tokens: Option, output_tokens: Option) { - self.emit(ServerEvent::Thinking { message_id, content, input_tokens, output_tokens }).await; + pub(super) async fn thinking(&self, message_id: i64, content: String, input_tokens: Option, output_tokens: Option, reasoning_content: Option) { + self.emit(ServerEvent::Thinking { message_id, content, input_tokens, output_tokens, reasoning_content }).await; + } + + /// Clone of the underlying sender, for spawning side-channel tasks that + /// emit alongside the turn (e.g. the token-delta forwarder). + pub(super) fn sender(&self) -> mpsc::Sender { + self.tx.clone() } /// The assistant response is complete. - pub(super) async fn done(&self, message_id: i64, stack_id: i64, content: String, input_tokens: Option, output_tokens: Option) { - self.emit(ServerEvent::Done { message_id, stack_id, content, input_tokens, output_tokens }).await; + pub(super) async fn done(&self, message_id: i64, stack_id: i64, content: String, input_tokens: Option, output_tokens: Option, reasoning_content: Option) { + self.emit(ServerEvent::Done { message_id, stack_id, content, input_tokens, output_tokens, reasoning_content }).await; } /// The LLM was cut off by the token limit. diff --git a/crates/skald-core/src/session/handler/llm_call.rs b/crates/skald-core/src/session/handler/llm_call.rs index 9de0483..70e099a 100644 --- a/crates/skald-core/src/session/handler/llm_call.rs +++ b/crates/skald-core/src/session/handler/llm_call.rs @@ -9,11 +9,13 @@ use std::collections::HashSet; use std::sync::Arc; use serde_json::Value; +use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; use tracing::{error, warn}; -use crate::chatbot::{ChatOptions, LlmTurn}; +use crate::chatbot::{ChatOptions, LlmTurn, StreamDelta}; use crate::db::llm_request_payloads; +use crate::events::{ServerEvent, TokenDeltaKind}; use crate::llm::{LlmEntry, LlmStrength}; use super::ChatSessionHandler; @@ -77,10 +79,19 @@ impl ChatSessionHandler { // the fallback reassignment below. On cancel we drop the future // (aborting the request) and return immediately. let client = cur_llm.client.clone(); + // 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. + let (delta_tx, delta_rx) = mpsc::channel::(256); + 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(messages.as_slice(), defs, &options) => r, + r = client.chat_with_tools_raw_streaming(messages.as_slice(), defs, &options, 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 + // round's outcome events (Thinking / Done) in bus order. + forwarder.await.ok(); let e = match call_result { Ok((turn, meta)) => { @@ -151,6 +162,26 @@ impl ChatSessionHandler { } } +/// Forwards streaming deltas from the LLM client onto the turn's event channel +/// as `TokenDelta` events. Exits when the client drops its sender (call +/// completed or aborted) or when the turn receiver is gone. +fn spawn_delta_forwarder( + mut rx: mpsc::Receiver, + tx: mpsc::Sender, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + while let Some(d) = rx.recv().await { + let (kind, delta) = match d { + StreamDelta::Text(t) => (TokenDeltaKind::Content, t), + StreamDelta::Reasoning(t) => (TokenDeltaKind::Reasoning, t), + }; + if tx.send(ServerEvent::TokenDelta { kind, delta }).await.is_err() { + break; + } + } + }) +} + /// Whether an LLM error is worth retrying on a different model. /// /// Classifies on the real HTTP status ([`crate::chatbot::http_status`]), not a diff --git a/crates/skald-core/src/session/handler/llm_loop.rs b/crates/skald-core/src/session/handler/llm_loop.rs index e9dae04..fb39e69 100644 --- a/crates/skald-core/src/session/handler/llm_loop.rs +++ b/crates/skald-core/src/session/handler/llm_loop.rs @@ -152,6 +152,7 @@ impl ChatSessionHandler { input_tokens: resp.input_tokens, output_tokens: resp.output_tokens, truncated: resp.truncated, + reasoning_content: resp.reasoning_content, tool_calls: all_tool_calls, }); } @@ -165,7 +166,7 @@ impl ChatSessionHandler { chat_history::set_usage(pool, message_id, i, o, 0, cost).await?; } if !assistant_text.trim().is_empty() || input_tokens.is_some() { - em.thinking(message_id, assistant_text, input_tokens, output_tokens).await; + em.thinking(message_id, assistant_text, input_tokens, output_tokens, reasoning_content).await; } // A homogeneous batch of ≥2 synchronous sub-agent calls is fanned diff --git a/crates/skald-core/src/session/handler/mod.rs b/crates/skald-core/src/session/handler/mod.rs index 1603800..829c274 100644 --- a/crates/skald-core/src/session/handler/mod.rs +++ b/crates/skald-core/src/session/handler/mod.rs @@ -103,6 +103,8 @@ pub(super) enum TurnOutcome { input_tokens: Option, output_tokens: Option, truncated: bool, + /// Chain-of-thought produced by the final round, when any. + reasoning_content: Option, /// All tool calls executed during this turn, across all rounds. tool_calls: Vec, }, @@ -643,7 +645,7 @@ impl ChatSessionHandler { let outcome = self.run_agent_turn(stack.id, &config, &token, &tx, pending_input.as_ref()).await?; match outcome { - TurnOutcome::Final { content, message_id, input_tokens, output_tokens, truncated, tool_calls } => { + TurnOutcome::Final { content, message_id, input_tokens, output_tokens, truncated, reasoning_content, tool_calls } => { // Persist token count so the *next* handle_message call knows // whether to compact before running the LLM loop. if let Some(t) = input_tokens { @@ -654,7 +656,7 @@ impl ChatSessionHandler { warn!(session_id = self.session_id, ?output_tokens, "response truncated (max_tokens)"); em.truncated(output_tokens).await; } - em.done(message_id, stack.id, content.clone(), input_tokens, output_tokens).await; + em.done(message_id, stack.id, content.clone(), input_tokens, output_tokens, reasoning_content).await; // Publish both messages to the event bus now that both are in the DB. let now = chrono::Utc::now(); diff --git a/crates/skald-core/src/session/handler/resume.rs b/crates/skald-core/src/session/handler/resume.rs index 7a4f180..a32c850 100644 --- a/crates/skald-core/src/session/handler/resume.rs +++ b/crates/skald-core/src/session/handler/resume.rs @@ -128,6 +128,7 @@ impl ChatSessionHandler { input_tokens: None, output_tokens: None, truncated: false, + reasoning_content: msg.reasoning_content, tool_calls: Vec::new(), }; break 'seed (outcome, stack); @@ -208,13 +209,13 @@ impl ChatSessionHandler { // current_stack is now the root (depth=0); emit the final event. match current_outcome { - TurnOutcome::Final { content, message_id, input_tokens, output_tokens, truncated, .. } => { + TurnOutcome::Final { content, message_id, input_tokens, output_tokens, truncated, reasoning_content, .. } => { info!(session_id = self.session_id, "resume_turn done"); if truncated { warn!(session_id = self.session_id, "response truncated"); em.truncated(output_tokens).await; } - em.done(message_id, current_stack.id, content, input_tokens, output_tokens).await; + em.done(message_id, current_stack.id, content, input_tokens, output_tokens, reasoning_content).await; } TurnOutcome::Cancelled => { info!(session_id = self.session_id, "resume_turn cancelled"); diff --git a/src/frontend/api/sessions.rs b/src/frontend/api/sessions.rs index 2881881..0a17141 100644 --- a/src/frontend/api/sessions.rs +++ b/src/frontend/api/sessions.rs @@ -633,6 +633,7 @@ fn build_items<'a>( "failed": failed, "input_tokens": msg.input_tokens, "output_tokens": msg.output_tokens, + "reasoning": msg.reasoning_content, })); } else { if !msg.content.trim().is_empty() { @@ -643,6 +644,7 @@ fn build_items<'a>( "failed": failed, "input_tokens": msg.input_tokens, "output_tokens": msg.output_tokens, + "reasoning": msg.reasoning_content, })); } for tc in &tool_calls { diff --git a/src/frontend/api/ws.rs b/src/frontend/api/ws.rs index 860511c..75029ff 100644 --- a/src/frontend/api/ws.rs +++ b/src/frontend/api/ws.rs @@ -194,6 +194,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc, source: String, content: msg, input_tokens: None, output_tokens: None, + reasoning_content: None, })).await; continue; } @@ -205,6 +206,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc, source: String, content: dynamic_help(&skald), input_tokens: None, output_tokens: None, + reasoning_content: None, })).await; continue; } @@ -220,6 +222,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc, source: String, content: format!("↑{input_str} tok · ↓{output_str} tok"), input_tokens: None, output_tokens: None, + reasoning_content: None, })).await; } Err(e) => { @@ -238,6 +241,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc, source: String, content: format!("💰 Costo sessione: ${c:.4}"), input_tokens: None, output_tokens: None, + reasoning_content: None, })).await; } Ok(None) => { @@ -247,6 +251,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc, source: String, content: "💰 Nessun costo registrato per questa sessione.".to_string(), input_tokens: None, output_tokens: None, + reasoning_content: None, })).await; } Err(e) => { @@ -265,6 +270,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc, source: String, content: "✅ Contesto compattato.".to_string(), input_tokens: None, output_tokens: None, + reasoning_content: None, })).await; } Ok(false) => { @@ -274,6 +280,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc, source: String, content: "⏩ Compaction skipped (no messages to summarize or compaction disabled).".to_string(), input_tokens: None, output_tokens: None, + reasoning_content: None, })).await; } Err(e) => { @@ -292,6 +299,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc, source: String, content: "✅ Activated tool groups removed from the session.".to_string(), input_tokens: None, output_tokens: None, + reasoning_content: None, })).await; } Err(e) => { @@ -310,6 +318,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc, source: String, content, input_tokens: None, output_tokens: None, + reasoning_content: None, })).await; continue; } @@ -327,6 +336,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc, source: String, content, input_tokens: None, output_tokens: None, + reasoning_content: None, })).await; continue; } @@ -358,6 +368,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc, source: String, content: format!("Unknown command: {first}\n\n{}", dynamic_help(&skald)), input_tokens: None, output_tokens: None, + reasoning_content: None, })).await; continue; } diff --git a/web/components/copilot-render.js b/web/components/copilot-render.js index 7cfadae..67e53ee 100644 --- a/web/components/copilot-render.js +++ b/web/components/copilot-render.js @@ -507,6 +507,21 @@ export function renderAttachmentChips(host, attachments, { removable = false } = `; } +/** + * Collapsible chain-of-thought block: small, muted, collapsed by default so it + * never weighs on the UI. A native
— Lit keeps the element stable + * across re-renders, so a user-expanded block stays open while tokens stream + * into it (live) and in past history items alike. + */ +function renderReasoning(msg) { + if (!msg.reasoning) return nothing; + return html` +
+ ${t('chat.reasoning')} +
${msg.reasoning}
+
`; +} + export function renderMsg(host, msg) { try { switch (msg.kind) { @@ -516,6 +531,7 @@ export function renderMsg(host, msg) { return html`
${msg.failed ? failedBadge() : nothing} + ${renderReasoning(msg)} ${unsafeHTML(renderMarkdown(msg.content))} ${msg.input_tokens != null ? html`
↑${msg.input_tokens.toLocaleString()} tok  ↓${msg.output_tokens?.toLocaleString()} tok
` : nothing}
`; @@ -523,8 +539,10 @@ export function renderMsg(host, msg) { return html`
${msg.failed ? failedBadge() : nothing} + ${renderReasoning(msg)} ${unsafeHTML(renderMarkdown(msg.content))} - ${msg.input_tokens != null ? html`
↑${msg.input_tokens.toLocaleString()} tok  ↓${msg.output_tokens?.toLocaleString()} tok
` : nothing} + ${msg.streaming ? html`` : nothing} + ${msg.input_tokens != null && !msg.streaming ? html`
↑${msg.input_tokens.toLocaleString()} tok  ↓${msg.output_tokens?.toLocaleString()} tok
` : nothing}
`; case 'error': return html` diff --git a/web/css/copilot-messages.css b/web/css/copilot-messages.css index cdb0bd9..0bf3bb1 100644 --- a/web/css/copilot-messages.css +++ b/web/css/copilot-messages.css @@ -88,6 +88,75 @@ letter-spacing: 0.02em; } +/* ── Reasoning (chain-of-thought) block ────────────────────────────────────── */ +/* Small, low-contrast, collapsed by default: visible but never heavy. */ + +.reasoning-block { + margin: 0 0 0.45rem; + font-size: 0.75rem; + color: var(--placeholder-color); + opacity: 0.75; +} + +.reasoning-block > summary { + cursor: pointer; + user-select: none; + font-style: italic; + letter-spacing: 0.02em; + list-style-position: inside; + padding: 0.1rem 0; +} + +.reasoning-block > summary:hover { + color: var(--msg-assistant-text); +} + +.reasoning-block--live > summary { + animation: reasoning-pulse 1.6s ease-in-out infinite; +} + +.reasoning-content { + margin-top: 0.35rem; + padding-left: 0.6rem; + border-left: 2px solid var(--toolbar-border); + white-space: pre-wrap; + word-break: break-word; + font-family: var(--font-mono); + font-size: 0.72rem; + line-height: 1.5; + max-height: 16rem; + overflow-y: auto; +} + +@keyframes reasoning-pulse { + 0%, 100% { opacity: 1; } + 50% { opacity: 0.45; } +} + +@media (prefers-reduced-motion: reduce) { + .reasoning-block--live > summary { animation: none; } +} + +/* Blinking caret at the end of a streaming assistant bubble. */ +.stream-caret { + display: inline-block; + width: 0.5em; + height: 1em; + margin-left: 0.1em; + vertical-align: text-bottom; + background: currentColor; + opacity: 0.6; + animation: stream-caret-blink 1s steps(2, start) infinite; +} + +@keyframes stream-caret-blink { + to { visibility: hidden; } +} + +@media (prefers-reduced-motion: reduce) { + .stream-caret { animation: none; } +} + /* ── Tool call blocks ──────────────────────────────────────────────────────── */ .copilot-tool { diff --git a/web/i18n/en.js b/web/i18n/en.js index 15dc549..fcba7f1 100644 --- a/web/i18n/en.js +++ b/web/i18n/en.js @@ -65,6 +65,7 @@ export default { 'chat.send': 'Send', 'chat.stop': 'Stop', 'chat.thinking': 'Thinking…', + 'chat.reasoning': 'Reasoning…', 'chat.attach': 'Attach files', 'chat.new_session': 'New conversation', 'chat.security_group': 'Security group', diff --git a/web/i18n/fr.js b/web/i18n/fr.js index 181b5ee..e328258 100644 --- a/web/i18n/fr.js +++ b/web/i18n/fr.js @@ -65,6 +65,7 @@ export default { 'chat.send': 'Envoyer', 'chat.stop': 'Arrêter', 'chat.thinking': 'Réflexion…', + 'chat.reasoning': 'Raisonnement…', 'chat.attach': 'Joindre des fichiers', 'chat.new_session': 'Nouvelle conversation', 'chat.security_group': 'Groupe de sécurité', diff --git a/web/i18n/it.js b/web/i18n/it.js index 4baae73..36598d2 100644 --- a/web/i18n/it.js +++ b/web/i18n/it.js @@ -65,6 +65,7 @@ export default { 'chat.send': 'Invia', 'chat.stop': 'Ferma', 'chat.thinking': 'Sto pensando…', + 'chat.reasoning': 'Ragionamento…', 'chat.attach': 'Allega file', 'chat.new_session': 'Nuova conversazione', 'chat.security_group': 'Gruppo di sicurezza', diff --git a/web/lib/chat-session.js b/web/lib/chat-session.js index c0244da..d6229b1 100644 --- a/web/lib/chat-session.js +++ b/web/lib/chat-session.js @@ -51,6 +51,7 @@ export class ChatSession extends LightElement { // STOP button when reconnecting mid-turn). static _STREAMING_EVENTS = new Set([ 'thinking', 'tool_start', 'agent_start', 'pending_write', 'approval_required', + 'token_delta', ]); constructor() { @@ -251,6 +252,7 @@ export class ChatSession extends LightElement { this._ws.close(); this._ws = null; } + this._cancelStreamFlush(); this._messages = []; this._waiting = false; try { @@ -290,18 +292,62 @@ export class ChatSession extends LightElement { }); break; - case 'thinking': - this._push({ kind: 'thinking', message_id: msg.message_id, content: msg.content, - input_tokens: msg.input_tokens, output_tokens: msg.output_tokens }); + case 'thinking': { + // A tool-call round's text. When the round streamed, its pending bubble + // becomes the thinking item in place. Reasoning comes from the event + // (buffered providers) or the streamed accumulation. + const last = this._messages[this._messages.length - 1]; + const streaming = (last?.kind === 'assistant' && last.streaming) ? last : null; + const item = { kind: 'thinking', message_id: msg.message_id, content: msg.content, + reasoning: msg.reasoning_content ?? streaming?.reasoning ?? null, + input_tokens: msg.input_tokens, output_tokens: msg.output_tokens }; + if (streaming) this._replaceLast(item); else this._push(item); break; + } - case 'done': - this._waiting = false; - this._push({ kind: 'assistant', content: msg.content, - input_tokens: msg.input_tokens, output_tokens: msg.output_tokens }); + case 'token_delta': { + // Best-effort live tokens. Accumulate into a pending assistant bubble; + // the final `done` (or `thinking`) event replaces it with authoritative + // content. Mutate in place + throttled flush: deltas can arrive at a + // high rate and a full Lit update per token would be wasteful. + let last = this._messages[this._messages.length - 1]; + if (last?.kind !== 'assistant' || !last.streaming) { + last = { kind: 'assistant', content: '', reasoning: '', streaming: true }; + this._messages = [...this._messages, last]; + this._onMessagePushed(last); + } + if (msg.kind === 'reasoning') last.reasoning += msg.delta; + else last.content += msg.delta; + this._scheduleStreamFlush(); break; + } + + case 'done': { + this._waiting = false; + const last = this._messages[this._messages.length - 1]; + if (last?.kind === 'assistant' && last.streaming) { + // Finalize the streamed bubble with the authoritative content. + this._replaceLast({ kind: 'assistant', content: msg.content, + reasoning: msg.reasoning_content ?? last.reasoning ?? null, + input_tokens: msg.input_tokens, output_tokens: msg.output_tokens }); + } else { + this._push({ kind: 'assistant', content: msg.content, + reasoning: msg.reasoning_content ?? null, + input_tokens: msg.input_tokens, output_tokens: msg.output_tokens }); + } + break; + } case 'tool_start': { + // A round with tool calls but no Thinking event (no usage/text) leaves a + // reasoning-only streaming bubble behind: finalize it in place so its + // content isn't swallowed by the next round's deltas. + const last = this._messages[this._messages.length - 1]; + if (last?.kind === 'assistant' && last.streaming) { + this._replaceLast({ kind: 'thinking', content: last.content, + reasoning: last.reasoning || null, + input_tokens: null, output_tokens: null }); + } // On resume, the server re-emits ToolStart for tools already in history. // Update in place rather than pushing a duplicate card. const existingIdx = this._messages.findIndex( @@ -399,6 +445,14 @@ export class ChatSession extends LightElement { break; case 'agent_done': { + // A sub-agent's final round emits no Done: its streamed bubble would + // stay pending forever — finalize it with the accumulated content. + const last = this._messages[this._messages.length - 1]; + if (last?.kind === 'assistant' && last.streaming) { + this._replaceLast({ kind: 'assistant', content: last.content, + reasoning: last.reasoning || null, + input_tokens: null, output_tokens: null }); + } this._updateAgent(msg.stack_id, { done: true }); const agentMsg = this._messages.find(m => m.kind === 'agent' && m.stack_id === msg.stack_id); if (agentMsg) { @@ -419,6 +473,7 @@ export class ChatSession extends LightElement { case 'error': this._waiting = false; + this._dropStreaming(); this._pushError(msg.message); break; @@ -435,6 +490,9 @@ export class ChatSession extends LightElement { } case 'model_fallback': + // A fallback mid-stream means the previous attempt's deltas are orphaned: + // drop the pending bubble — the replacement model streams a fresh one. + this._dropStreaming(); this._push({ kind: 'info', content: `⚡ Model fallback: ${msg.from} → ${msg.to}` }); break; @@ -453,6 +511,7 @@ export class ChatSession extends LightElement { break; case 'new_session': + this._cancelStreamFlush(); this._messages = []; this._waiting = false; break; @@ -476,6 +535,7 @@ export class ChatSession extends LightElement { case 'llm_failed': this._waiting = false; + this._dropStreaming(); this._pushError(`LLM unavailable. Tried: ${msg.tried.join(', ')}. ${msg.last_error}`); break; } @@ -487,6 +547,42 @@ export class ChatSession extends LightElement { this._onMessagePushed(item); } + // ── Live token streaming ──────────────────────────────────────────────────── + // A pending assistant bubble (`streaming: true`) is mutated in place by + // `token_delta` events and flushed to Lit at most ~15×/s; turn-ending events + // (`done`/`thinking`) finalize it via `_replaceLast`, failures drop it. + + _scheduleStreamFlush() { + if (this._streamFlushTimer) return; + this._streamFlushTimer = setTimeout(() => { + this._streamFlushTimer = null; + this._messages = [...this._messages]; + this._scrollToBottom(); + }, 66); + } + + _cancelStreamFlush() { + if (!this._streamFlushTimer) return; + clearTimeout(this._streamFlushTimer); + this._streamFlushTimer = null; + } + + _replaceLast(item) { + this._cancelStreamFlush(); + const updated = [...this._messages]; + updated[updated.length - 1] = item; + this._messages = updated; + this._scrollToBottom(); + } + + _dropStreaming() { + this._cancelStreamFlush(); + const last = this._messages[this._messages.length - 1]; + if (last?.kind === 'assistant' && last.streaming) { + this._messages = this._messages.slice(0, -1); + } + } + _pushError(text) { this._push({ kind: 'error', content: text }); }