diff --git a/server/src/chat/llm_turn.rs b/server/src/chat/llm_turn.rs new file mode 100644 index 00000000..09d0d074 --- /dev/null +++ b/server/src/chat/llm_turn.rs @@ -0,0 +1,250 @@ +//! Transport-agnostic LLM turn driver shared by every chat transport that +//! forwards a non-command message to Claude Code (currently Discord and +//! WhatsApp). Formatting, chunking, and error-message differences between +//! transports are captured by the small [`TransportFormat`] trait; the +//! streaming/session/history plumbing lives once in [`run_llm_turn`]. + +use std::collections::HashMap; +use std::sync::Arc; + +use tokio::sync::Mutex as TokioMutex; + +use crate::chat::ChatTransport; +use crate::chat::transport::matrix::{ConversationEntry, ConversationRole, RoomConversation}; +use crate::http::context::PermissionDecision; +use crate::services::Services; +use crate::slog; + +/// Per-transport formatting hooks needed to run a shared LLM turn. +pub trait TransportFormat: Send + Sync { + /// Convert markdown into the transport's native formatting and split it + /// into messages that respect the transport's size limit. Transports + /// without a size limit (or that don't chunk) return a single-element + /// vector. + fn format_and_chunk(&self, markdown: &str) -> Vec; + + /// Prefix used in `slog!` calls for this transport, e.g. `"[discord]"`. + fn log_prefix(&self) -> &'static str; + + /// Convert an LLM error into a user-facing message. + fn format_error(&self, err: &str) -> String; +} + +/// Build the prompt for an LLM turn, prepending any pending CRDT +/// pipeline-transition events as a `` block. +pub fn build_llm_prompt(persona: &str, bot_name: &str, user: &str, user_message: &str) -> String { + let event_ctx = crate::llm_session::assemble_prompt_context(persona); + format!( + "{event_ctx}[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n\n{user}: {user_message}" + ) +} + +/// Everything a shared LLM turn needs beyond formatting: transport handle, +/// routing/history key, display name, and shared services/history state. +pub struct LlmTurnContext<'a> { + /// Chat transport used to send/post messages for this turn. + pub transport: Arc, + /// Routing and history key (Discord channel id, WhatsApp phone number). + pub key: &'a str, + /// Display name stored in conversation history and used in the prompt. + pub user: &'a str, + /// Shared services bundle (project root, permissions, agents). + pub services: &'a Arc, + /// Per-key conversation history. + pub history: &'a Arc>>, + /// Maximum number of conversation entries to keep per key. + pub history_size: usize, + /// Persists conversation history to disk. + pub save_history: fn(&std::path::Path, &HashMap), +} + +/// Forward a message to Claude Code and stream the response back through the +/// transport, handling permission prompts and conversation history. +/// +/// `F` is taken by value (and cloned into the posting task) rather than by +/// reference because the posting task is a `tokio::spawn`ed future, which +/// requires `'static` — the formatter types are zero-sized markers, so +/// cloning is free. +pub async fn run_llm_turn( + ctx: LlmTurnContext<'_>, + format: F, + user_message: &str, +) { + use crate::chat::util::drain_complete_paragraphs; + use crate::llm::providers::claude_code::{ClaudeCodeProvider, ClaudeCodeResult}; + use std::sync::atomic::{AtomicBool, Ordering}; + use tokio::sync::watch; + + let prefix = format.log_prefix(); + + // Look up existing session ID for this key. + let resume_session_id: Option = { + let guard = ctx.history.lock().await; + guard.get(ctx.key).and_then(|conv| conv.session_id.clone()) + }; + + let bot_name = &ctx.services.bot_name; + let persona = bot_name.to_lowercase(); + let prompt = build_llm_prompt(&persona, bot_name, ctx.user, user_message); + + let provider = ClaudeCodeProvider::new(); + let (_cancel_tx, mut cancel_rx) = watch::channel(false); + + // Channel for sending complete chunks to the posting task. + let (msg_tx, mut msg_rx) = tokio::sync::mpsc::unbounded_channel::(); + let msg_tx_for_callback = msg_tx.clone(); + + // Spawn a task to post messages as they arrive. + let post_transport = Arc::clone(&ctx.transport); + let post_key = ctx.key.to_string(); + let post_format = format.clone(); + let post_task = tokio::spawn(async move { + while let Some(chunk) = msg_rx.recv().await { + for part in post_format.format_and_chunk(&chunk) { + let _ = post_transport.send_message(&post_key, &part, "").await; + } + } + }); + + // Shared buffer between the sync token callback and the async scope. + let buffer = Arc::new(std::sync::Mutex::new(String::new())); + let buffer_for_callback = Arc::clone(&buffer); + let sent_any_chunk = Arc::new(AtomicBool::new(false)); + let sent_any_chunk_for_callback = Arc::clone(&sent_any_chunk); + + let project_root_str = ctx.services.project_root.to_string_lossy().to_string(); + let chat_fut = provider.chat_stream( + &prompt, + &project_root_str, + resume_session_id.as_deref(), + None, + None, + &mut cancel_rx, + move |token| { + let mut buf = buffer_for_callback.lock().unwrap(); + buf.push_str(token); + let paragraphs = drain_complete_paragraphs(&mut buf); + for chunk in paragraphs { + sent_any_chunk_for_callback.store(true, Ordering::Relaxed); + let _ = msg_tx_for_callback.send(chunk); + } + }, + |_thinking| {}, + |_activity| {}, + ); + tokio::pin!(chat_fut); + + // Register as a permission responder for the duration of this chat turn. + let (_perm_guard, mut perm_rx) = ctx.services.permission_registry.register(); + + let result = loop { + tokio::select! { + r = &mut chat_fut => break r, + + Some(perm_fwd) = perm_rx.recv() => { + let prompt_msg = format!( + "**Permission Request**\n\nTool: `{}`\n```json\n{}\n```\n\nReply **yes** to approve or **no** to deny.", + perm_fwd.tool_name, + serde_json::to_string_pretty(&perm_fwd.tool_input) + .unwrap_or_else(|_| perm_fwd.tool_input.to_string()), + ); + for part in format.format_and_chunk(&prompt_msg) { + let _ = ctx.transport.send_message(ctx.key, &part, "").await; + } + + // Keyed by request_id (not just key) so a second + // concurrent request doesn't drop the first's sender. + ctx.services + .pending_perm_replies + .insert(ctx.key.to_string(), perm_fwd.request_id.clone(), perm_fwd.response_tx) + .await; + + // Spawn a timeout task: auto-deny if the user does not respond. + let pending = Arc::clone(&ctx.services.pending_perm_replies); + let timeout_key = ctx.key.to_string(); + let timeout_request_id = perm_fwd.request_id.clone(); + let timeout_transport = Arc::clone(&ctx.transport); + let timeout_secs = ctx.services.permission_timeout_secs; + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)).await; + if let Some(tx) = pending.remove_by_request_id(&timeout_key, &timeout_request_id).await { + let _ = tx.send(PermissionDecision::Deny); + let msg = "Permission request timed out — denied (fail-closed)."; + let _ = timeout_transport.send_message(&timeout_key, msg, "").await; + } + }); + } + } + }; + + // Flush remaining text. + let remaining = buffer.lock().unwrap().trim().to_string(); + let did_send_any = sent_any_chunk.load(Ordering::Relaxed); + + let (assistant_reply, new_session_id) = match result { + Ok(ClaudeCodeResult { + messages, + session_id, + .. + }) => { + let reply = if !remaining.is_empty() { + let _ = msg_tx.send(remaining.clone()); + remaining + } else if !did_send_any { + let last_text = messages + .iter() + .rev() + .find(|m| m.role == crate::llm::types::Role::Assistant && !m.content.is_empty()) + .map(|m| m.content.clone()) + .unwrap_or_default(); + if !last_text.is_empty() { + let _ = msg_tx.send(last_text.clone()); + } + last_text + } else { + remaining + }; + slog!("{prefix} session_id from chat_stream: {:?}", session_id); + (reply, session_id) + } + Err(e) => { + slog!("{prefix} LLM error: {e}"); + let err_msg = format.format_error(&e); + let _ = msg_tx.send(err_msg.clone()); + (err_msg, None) + } + }; + + // Signal the posting task to finish and wait for it. + drop(msg_tx); + let _ = post_task.await; + + // Record this exchange in conversation history. + if !assistant_reply.starts_with("Error processing") { + let mut guard = ctx.history.lock().await; + let conv = guard.entry(ctx.key.to_string()).or_default(); + + if new_session_id.is_some() { + conv.session_id = new_session_id; + } + + conv.entries.push(ConversationEntry { + role: ConversationRole::User, + sender: ctx.user.to_string(), + content: user_message.to_string(), + }); + conv.entries.push(ConversationEntry { + role: ConversationRole::Assistant, + sender: String::new(), + content: assistant_reply, + }); + + // Trim to configured maximum. + if conv.entries.len() > ctx.history_size { + let excess = conv.entries.len() - ctx.history_size; + conv.entries.drain(..excess); + } + + (ctx.save_history)(&ctx.services.project_root, &guard); + } +} diff --git a/server/src/chat/mod.rs b/server/src/chat/mod.rs index ad08a3be..21094b65 100644 --- a/server/src/chat/mod.rs +++ b/server/src/chat/mod.rs @@ -12,6 +12,8 @@ pub mod compact; pub mod dispatcher; /// Chat history utilities — loading and serialising conversation history. pub mod history; +/// Transport-agnostic LLM turn driver shared by Discord, WhatsApp, etc. +pub mod llm_turn; pub(crate) mod lookup; #[cfg(test)] pub(crate) mod test_helpers; diff --git a/server/src/chat/transport/discord/commands.rs b/server/src/chat/transport/discord/commands.rs index 272b6233..28581c30 100644 --- a/server/src/chat/transport/discord/commands.rs +++ b/server/src/chat/transport/discord/commands.rs @@ -4,7 +4,9 @@ use std::collections::HashSet; use std::sync::Arc; use crate::chat::ChatTransport; -use crate::chat::transport::matrix::{ConversationEntry, ConversationRole, RoomConversation}; +use crate::chat::transport::matrix::RoomConversation; +#[cfg(test)] +use crate::chat::transport::matrix::{ConversationEntry, ConversationRole}; use crate::chat::util::is_permission_approval; use crate::http::context::PermissionDecision; use crate::services::Services; @@ -304,196 +306,43 @@ pub(super) async fn handle_incoming_message( handle_llm_message(ctx, channel, user, message).await; } -/// Build the prompt for a Discord LLM turn, prepending any pending -/// CRDT pipeline-transition events as a `` block. -fn build_discord_llm_prompt( - persona: &str, - bot_name: &str, - user: &str, - user_message: &str, -) -> String { - let event_ctx = crate::llm_session::assemble_prompt_context(persona); - format!( - "{event_ctx}[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n\n{user}: {user_message}" - ) +/// [`crate::chat::llm_turn::TransportFormat`] implementation for Discord: +/// single-message replies (no chunking) and no OAuth-link error handling. +#[derive(Clone, Copy)] +struct DiscordFormat; + +impl crate::chat::llm_turn::TransportFormat for DiscordFormat { + fn format_and_chunk(&self, markdown: &str) -> Vec { + vec![markdown_to_discord(markdown)] + } + + fn log_prefix(&self) -> &'static str { + "[discord]" + } + + fn format_error(&self, err: &str) -> String { + format!("Error processing your request: {err}") + } } /// Forward a message to Claude Code and send the response back via Discord. async fn handle_llm_message(ctx: &DiscordContext, channel: &str, user: &str, user_message: &str) { - use crate::chat::util::drain_complete_paragraphs; - use crate::llm::providers::claude_code::{ClaudeCodeProvider, ClaudeCodeResult}; - use std::sync::atomic::{AtomicBool, Ordering}; - use tokio::sync::watch; + use crate::chat::llm_turn::{LlmTurnContext, run_llm_turn}; - // Look up existing session ID for this channel. - let resume_session_id: Option = { - let guard = ctx.history.lock().await; - guard.get(channel).and_then(|conv| conv.session_id.clone()) - }; - - let bot_name = &ctx.services.bot_name; - let persona = bot_name.to_lowercase(); - let prompt = build_discord_llm_prompt(&persona, bot_name, user, user_message); - - let provider = ClaudeCodeProvider::new(); - let (_cancel_tx, mut cancel_rx) = watch::channel(false); - - // Channel for sending complete chunks to the Discord posting task. - let (msg_tx, mut msg_rx) = tokio::sync::mpsc::unbounded_channel::(); - let msg_tx_for_callback = msg_tx.clone(); - - // Spawn a task to post messages as they arrive. - let post_transport = Arc::clone(&ctx.transport) as Arc; - let post_channel = channel.to_string(); - let post_task = tokio::spawn(async move { - while let Some(chunk) = msg_rx.recv().await { - let formatted = markdown_to_discord(&chunk); - let _ = post_transport - .send_message(&post_channel, &formatted, "") - .await; - } - }); - - // Shared buffer between the sync token callback and the async scope. - let buffer = Arc::new(std::sync::Mutex::new(String::new())); - let buffer_for_callback = Arc::clone(&buffer); - let sent_any_chunk = Arc::new(AtomicBool::new(false)); - let sent_any_chunk_for_callback = Arc::clone(&sent_any_chunk); - - let project_root_str = ctx.services.project_root.to_string_lossy().to_string(); - let chat_fut = provider.chat_stream( - &prompt, - &project_root_str, - resume_session_id.as_deref(), - None, - None, - &mut cancel_rx, - move |token| { - let mut buf = buffer_for_callback.lock().unwrap(); - buf.push_str(token); - let paragraphs = drain_complete_paragraphs(&mut buf); - for chunk in paragraphs { - sent_any_chunk_for_callback.store(true, Ordering::Relaxed); - let _ = msg_tx_for_callback.send(chunk); - } + run_llm_turn( + LlmTurnContext { + transport: Arc::clone(&ctx.transport) as Arc, + key: channel, + user, + services: &ctx.services, + history: &ctx.history, + history_size: ctx.history_size, + save_history: save_discord_history, }, - |_thinking| {}, - |_activity| {}, - ); - tokio::pin!(chat_fut); - - // Register as a permission responder for the duration of this chat turn. - let (_perm_guard, mut perm_rx) = ctx.services.permission_registry.register(); - - let result = loop { - tokio::select! { - r = &mut chat_fut => break r, - - Some(perm_fwd) = perm_rx.recv() => { - let prompt_msg = format!( - "**Permission Request**\n\nTool: `{}`\n```json\n{}\n```\n\nReply **yes** to approve or **no** to deny.", - perm_fwd.tool_name, - serde_json::to_string_pretty(&perm_fwd.tool_input) - .unwrap_or_else(|_| perm_fwd.tool_input.to_string()), - ); - let formatted = markdown_to_discord(&prompt_msg); - let _ = ctx.transport.send_message(channel, &formatted, "").await; - - // Keyed by request_id (not just channel) so a second - // concurrent request doesn't drop the first's sender. - ctx.services - .pending_perm_replies - .insert(channel.to_string(), perm_fwd.request_id.clone(), perm_fwd.response_tx) - .await; - - // Spawn a timeout task: auto-deny if the user does not respond. - let pending = Arc::clone(&ctx.services.pending_perm_replies); - let timeout_channel = channel.to_string(); - let timeout_request_id = perm_fwd.request_id.clone(); - let timeout_transport = Arc::clone(&ctx.transport) as Arc; - let timeout_secs = ctx.services.permission_timeout_secs; - tokio::spawn(async move { - tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)).await; - if let Some(tx) = pending.remove_by_request_id(&timeout_channel, &timeout_request_id).await { - let _ = tx.send(PermissionDecision::Deny); - let msg = "Permission request timed out — denied (fail-closed)."; - let _ = timeout_transport.send_message(&timeout_channel, msg, "").await; - } - }); - } - } - }; - - // Flush remaining text. - let remaining = buffer.lock().unwrap().trim().to_string(); - let did_send_any = sent_any_chunk.load(Ordering::Relaxed); - - let (assistant_reply, new_session_id) = match result { - Ok(ClaudeCodeResult { - messages, - session_id, - .. - }) => { - let reply = if !remaining.is_empty() { - let _ = msg_tx.send(remaining.clone()); - remaining - } else if !did_send_any { - let last_text = messages - .iter() - .rev() - .find(|m| m.role == crate::llm::types::Role::Assistant && !m.content.is_empty()) - .map(|m| m.content.clone()) - .unwrap_or_default(); - if !last_text.is_empty() { - let _ = msg_tx.send(last_text.clone()); - } - last_text - } else { - remaining - }; - slog!("[discord] session_id from chat_stream: {:?}", session_id); - (reply, session_id) - } - Err(e) => { - slog!("[discord] LLM error: {e}"); - let err_msg = format!("Error processing your request: {e}"); - let _ = msg_tx.send(err_msg.clone()); - (err_msg, None) - } - }; - - // Signal the posting task to finish and wait for it. - drop(msg_tx); - let _ = post_task.await; - - // Record this exchange in conversation history. - if !assistant_reply.starts_with("Error processing") { - let mut guard = ctx.history.lock().await; - let conv = guard.entry(channel.to_string()).or_default(); - - if new_session_id.is_some() { - conv.session_id = new_session_id; - } - - conv.entries.push(ConversationEntry { - role: ConversationRole::User, - sender: user.to_string(), - content: user_message.to_string(), - }); - conv.entries.push(ConversationEntry { - role: ConversationRole::Assistant, - sender: String::new(), - content: assistant_reply, - }); - - // Trim to configured maximum. - if conv.entries.len() > ctx.history_size { - let excess = conv.entries.len() - ctx.history_size; - conv.entries.drain(..excess); - } - - save_discord_history(&ctx.services.project_root, &guard); - } + DiscordFormat, + user_message, + ) + .await; } // ── Tests ─────────────────────────────────────────────────────────────── @@ -635,8 +484,12 @@ mod tests { at: chrono::Utc::now(), }); - let prompt = - build_discord_llm_prompt("discord-ch-test", "Timmy", "@alice", "what is the status?"); + let prompt = crate::chat::llm_turn::build_llm_prompt( + "discord-ch-test", + "Timmy", + "@alice", + "what is the status?", + ); assert!( prompt.contains(""), diff --git a/server/src/chat/transport/whatsapp/commands/llm.rs b/server/src/chat/transport/whatsapp/commands/llm.rs index 3ff5d0c5..a1d16b88 100644 --- a/server/src/chat/transport/whatsapp/commands/llm.rs +++ b/server/src/chat/transport/whatsapp/commands/llm.rs @@ -2,204 +2,56 @@ use std::sync::Arc; -use crate::chat::transport::matrix::{ConversationEntry, ConversationRole}; -use crate::http::context::PermissionDecision; -use crate::slog; - use super::super::WhatsAppWebhookContext; use super::super::format::{chunk_for_whatsapp, markdown_to_whatsapp}; use super::super::history::save_whatsapp_history; +/// [`crate::chat::llm_turn::TransportFormat`] implementation for WhatsApp: +/// messages are markdown-converted then split to fit the platform's size +/// limit, and LLM errors are checked for an OAuth login link. +#[derive(Clone, Copy)] +struct WhatsAppFormat; + +impl crate::chat::llm_turn::TransportFormat for WhatsAppFormat { + fn format_and_chunk(&self, markdown: &str) -> Vec { + chunk_for_whatsapp(&markdown_to_whatsapp(markdown)) + } + + fn log_prefix(&self) -> &'static str { + "[whatsapp]" + } + + fn format_error(&self, err: &str) -> String { + if let Some(url) = crate::llm::oauth::extract_login_url_from_error(err) { + format!("Authentication required. Log in to Claude here: {url}") + } else { + format!("Error processing your request: {err}") + } + } +} + +/// Forward a message to Claude Code and send the response back via WhatsApp. pub(super) async fn handle_llm_message( ctx: &WhatsAppWebhookContext, sender: &str, user_message: &str, ) { - use crate::chat::util::drain_complete_paragraphs; - use crate::llm::providers::claude_code::{ClaudeCodeProvider, ClaudeCodeResult}; - use std::sync::atomic::{AtomicBool, Ordering}; - use tokio::sync::watch; + use crate::chat::llm_turn::{LlmTurnContext, run_llm_turn}; - // Look up existing session ID for this sender. - let resume_session_id: Option = { - let guard = ctx.history.lock().await; - guard.get(sender).and_then(|conv| conv.session_id.clone()) - }; - - let bot_name = &ctx.services.bot_name; - let persona = bot_name.to_lowercase(); - let event_ctx = crate::llm_session::assemble_prompt_context(&persona); - let prompt = format!( - "{event_ctx}[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n\n{sender}: {user_message}" - ); - - let provider = ClaudeCodeProvider::new(); - let (_cancel_tx, mut cancel_rx) = watch::channel(false); - - // Channel for sending complete chunks to the WhatsApp posting task. - let (msg_tx, mut msg_rx) = tokio::sync::mpsc::unbounded_channel::(); - let msg_tx_for_callback = msg_tx.clone(); - - // Spawn a task to post messages as they arrive. - let post_transport = Arc::clone(&ctx.transport); - let post_sender = sender.to_string(); - let post_task = tokio::spawn(async move { - while let Some(chunk) = msg_rx.recv().await { - // Convert Markdown to WhatsApp formatting, then split into sized chunks. - let formatted = markdown_to_whatsapp(&chunk); - for part in chunk_for_whatsapp(&formatted) { - let _ = post_transport.send_message(&post_sender, &part, "").await; - } - } - }); - - // Shared buffer between the sync token callback and the async scope. - let buffer = Arc::new(std::sync::Mutex::new(String::new())); - let buffer_for_callback = Arc::clone(&buffer); - let sent_any_chunk = Arc::new(AtomicBool::new(false)); - let sent_any_chunk_for_callback = Arc::clone(&sent_any_chunk); - - let project_root_str = ctx.services.project_root.to_string_lossy().to_string(); - let chat_fut = provider.chat_stream( - &prompt, - &project_root_str, - resume_session_id.as_deref(), - None, - None, - &mut cancel_rx, - move |token| { - let mut buf = buffer_for_callback.lock().unwrap(); - buf.push_str(token); - let paragraphs = drain_complete_paragraphs(&mut buf); - for chunk in paragraphs { - sent_any_chunk_for_callback.store(true, Ordering::Relaxed); - let _ = msg_tx_for_callback.send(chunk); - } + run_llm_turn( + LlmTurnContext { + transport: Arc::clone(&ctx.transport), + key: sender, + user: sender, + services: &ctx.services, + history: &ctx.history, + history_size: ctx.history_size, + save_history: save_whatsapp_history, }, - |_thinking| {}, - |_activity| {}, - ); - tokio::pin!(chat_fut); - - // Register as a permission responder for the duration of this chat turn. - let (_perm_guard, mut perm_rx) = ctx.services.permission_registry.register(); - - let result = loop { - tokio::select! { - r = &mut chat_fut => break r, - - Some(perm_fwd) = perm_rx.recv() => { - let prompt_msg = format!( - "*Permission Request*\n\nTool: `{}`\n```json\n{}\n```\n\nReply *yes* to approve or *no* to deny.", - perm_fwd.tool_name, - serde_json::to_string_pretty(&perm_fwd.tool_input) - .unwrap_or_else(|_| perm_fwd.tool_input.to_string()), - ); - let formatted = markdown_to_whatsapp(&prompt_msg); - for part in chunk_for_whatsapp(&formatted) { - let _ = ctx.transport.send_message(sender, &part, "").await; - } - - // Store the response sender so the incoming message handler - // can resolve it when the user replies yes/no. Keyed by - // request_id (not just sender) so a second concurrent - // request doesn't drop the first's sender. - ctx.services.pending_perm_replies - .insert(sender.to_string(), perm_fwd.request_id.clone(), perm_fwd.response_tx) - .await; - - // Spawn a timeout task: auto-deny if the user does not respond. - let pending = Arc::clone(&ctx.services.pending_perm_replies); - let timeout_sender = sender.to_string(); - let timeout_request_id = perm_fwd.request_id.clone(); - let timeout_transport = Arc::clone(&ctx.transport); - let timeout_secs = ctx.services.permission_timeout_secs; - tokio::spawn(async move { - tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)).await; - if let Some(tx) = pending.remove_by_request_id(&timeout_sender, &timeout_request_id).await { - let _ = tx.send(PermissionDecision::Deny); - let msg = "Permission request timed out — denied (fail-closed)."; - let _ = timeout_transport.send_message(&timeout_sender, msg, "").await; - } - }); - } - } - }; - - // Flush remaining text. - let remaining = buffer.lock().unwrap().trim().to_string(); - let did_send_any = sent_any_chunk.load(Ordering::Relaxed); - - let (assistant_reply, new_session_id) = match result { - Ok(ClaudeCodeResult { - messages, - session_id, - .. - }) => { - let reply = if !remaining.is_empty() { - let _ = msg_tx.send(remaining.clone()); - remaining - } else if !did_send_any { - let last_text = messages - .iter() - .rev() - .find(|m| m.role == crate::llm::types::Role::Assistant && !m.content.is_empty()) - .map(|m| m.content.clone()) - .unwrap_or_default(); - if !last_text.is_empty() { - let _ = msg_tx.send(last_text.clone()); - } - last_text - } else { - remaining - }; - slog!("[whatsapp] session_id from chat_stream: {:?}", session_id); - (reply, session_id) - } - Err(e) => { - slog!("[whatsapp] LLM error: {e}"); - let err_msg = if let Some(url) = crate::llm::oauth::extract_login_url_from_error(&e) { - format!("Authentication required. Log in to Claude here: {url}") - } else { - format!("Error processing your request: {e}") - }; - let _ = msg_tx.send(err_msg.clone()); - (err_msg, None) - } - }; - - // Signal the posting task to finish and wait for it. - drop(msg_tx); - let _ = post_task.await; - - // Record this exchange in conversation history. - if !assistant_reply.starts_with("Error processing") { - let mut guard = ctx.history.lock().await; - let conv = guard.entry(sender.to_string()).or_default(); - - if new_session_id.is_some() { - conv.session_id = new_session_id; - } - - conv.entries.push(ConversationEntry { - role: ConversationRole::User, - sender: sender.to_string(), - content: user_message.to_string(), - }); - conv.entries.push(ConversationEntry { - role: ConversationRole::Assistant, - sender: String::new(), - content: assistant_reply, - }); - - // Trim to configured maximum. - if conv.entries.len() > ctx.history_size { - let excess = conv.entries.len() - ctx.history_size; - conv.entries.drain(..excess); - } - - save_whatsapp_history(&ctx.services.project_root, &guard); - } + WhatsAppFormat, + user_message, + ) + .await; } // ── Tests ───────────────────────────────────────────────────────────────