diff --git a/signal-gateway/src/claude/mod.rs b/signal-gateway/src/claude/mod.rs index 0dcf34e..3eec6e1 100644 --- a/signal-gateway/src/claude/mod.rs +++ b/signal-gateway/src/claude/mod.rs @@ -4,12 +4,14 @@ mod tools; mod worker; pub use tools::{Tool, ToolExecutor}; +pub use worker::SentBy; +use chrono::{DateTime, Utc}; use conf::Conf; use std::path::PathBuf; use std::sync::Weak; use tokio::sync::{mpsc, oneshot}; -use worker::{ClaudeRequest, ClaudeWorker}; +use worker::{ChatMessage, ClaudeWorker, Input}; /// The Anthropic API version header value. This is a stable version identifier, /// not a date indicating when the API was released. Anthropic adds new features @@ -58,7 +60,7 @@ pub enum ClaudeError { Request(#[from] reqwest::Error), /// API returned an error response. #[error("API error: {0}")] - ApiError(String), + ApiError(Box), /// Too many tool use iterations. #[error("exceeded maximum tool use iterations ({0})")] TooManyIterations(u32), @@ -81,7 +83,7 @@ pub enum ClaudeError { /// Requests are processed serially by a background worker to prevent /// concurrent API calls. pub struct ClaudeApi { - request_tx: mpsc::Sender, + input_tx: mpsc::Sender, stop_tx: mpsc::Sender<()>, #[allow(dead_code)] worker_handle: tokio::task::JoinHandle<()>, @@ -100,10 +102,10 @@ impl ClaudeApi { config: ClaudeConfig, tool_executor: Weak, ) -> Result { - let (request_tx, request_rx) = mpsc::channel(REQUEST_QUEUE_SIZE); + let (input_tx, input_rx) = mpsc::channel(REQUEST_QUEUE_SIZE); let (stop_tx, stop_rx) = mpsc::channel(REQUEST_QUEUE_SIZE); - let worker = ClaudeWorker::new(config, tool_executor, request_rx, stop_rx)?; + let worker = ClaudeWorker::new(config, tool_executor, input_rx, stop_rx)?; let worker_handle = tokio::spawn(async move { worker.run().await; @@ -111,26 +113,29 @@ impl ClaudeApi { }); Ok(Self { - request_tx, + input_tx, stop_tx, worker_handle, }) } - /// Send a request to the Claude API and return the response text. + /// Send a prompt to the Claude API, execute tools etc., and return the response text. /// /// Returns a future that resolves when the request is complete. /// Returns `QueueFull` error if the request queue is full. - pub async fn request(&self, prompt: &str) -> Result { + pub async fn request(&self, prompt: &str, ts_ms: u64) -> Result { let (result_tx, result_rx) = oneshot::channel(); - let request = ClaudeRequest { - prompt: prompt.to_owned(), - result_sender: result_tx, + let timestamp = DateTime::from_timestamp_millis(ts_ms as i64).unwrap_or_else(Utc::now); + let msg = ChatMessage { + sent_by: SentBy::UserToClaude, + text: prompt.into(), + timestamp, + result_sender: Some(result_tx), }; - self.request_tx - .try_send(request) + self.input_tx + .try_send(Input::Chat(msg)) .map_err(|_| ClaudeError::QueueFull)?; // result_rx.await has type Result, RecvError> @@ -138,7 +143,26 @@ impl ClaudeApi { result_rx.await.map_err(|_| ClaudeError::WorkerGone)? } - /// Request the worker to stop processing. + /// Record a message into claude's chat log that claude is not expected to respond to + pub fn record_message(&self, sent_by: SentBy, text: &str, ts_ms: u64) { + let timestamp = DateTime::from_timestamp_millis(ts_ms as i64).unwrap_or_else(Utc::now); + let msg = ChatMessage { + sent_by, + text: text.into(), + timestamp, + result_sender: None, + }; + + let _ = self.input_tx.try_send(Input::Chat(msg)); + } + + /// Request the worker to perform compaction (summarizing their recent message history and discarding it) + pub fn request_compaction(&self) { + let _ = self.input_tx.try_send(Input::Compact); + } + + /// Request the worker to interrupt any prompts it is responding to, discard any queued messages, + /// and get ready to accept different prompts. /// /// This will cause the current request (if any) to be interrupted at the /// next opportunity, and all pending requests to receive `StopRequested` errors. diff --git a/signal-gateway/src/claude/worker.rs b/signal-gateway/src/claude/worker.rs index 3c3a5a9..1dec00f 100644 --- a/signal-gateway/src/claude/worker.rs +++ b/signal-gateway/src/claude/worker.rs @@ -1,16 +1,70 @@ //! Background worker that processes Claude API requests serially. use super::{ANTHROPIC_API_VERSION, ClaudeConfig, ClaudeError, Tool, ToolExecutor}; +use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use serde_json::Value; use std::sync::Weak; use tokio::sync::{mpsc, oneshot}; use tracing::info; -/// A request to be processed by the Claude worker. -pub struct ClaudeRequest { - pub prompt: String, - pub result_sender: oneshot::Sender>, +/// Sent with inputs to claude that claude is expected to respond to. The sender +/// gives the worker a way to return the results to the caller asynchronously. +pub type ResultSender = oneshot::Sender>; + +/// Indicates the "role" i.e. the manner in which a particular message was sent +pub enum SentBy { + /// User message directed at the system (commands, etc.) + UserToSystem, + /// User message directed at Claude (prompts) + UserToClaude, + /// Response from Claude + Claude, + /// System-generated message + System, + /// Alert from alertmanager + AlertManager, +} + +impl SentBy { + fn role(&self) -> &str { + match self { + Self::UserToSystem => "user (speaking to system)", + Self::UserToClaude => "user (speaking to assistant)", + Self::Claude => "assistant", + Self::System => "system", + Self::AlertManager => "alertmanager", + } + } +} + +/// An input to the worker sent through the channel +pub enum Input { + Chat(ChatMessage), + Compact, +} + +/// A chat message +pub struct ChatMessage { + pub sent_by: SentBy, + pub timestamp: DateTime, + pub text: Box, + /// Present when claude is expected to respond to the message + pub result_sender: Option, +} + +impl ChatMessage { + fn into_content_and_sender(self) -> (MessageContent, Option) { + let mc = MessageContent { + role: self.sent_by.role().into(), + content: vec![ContentBlock::Text { + text: self.text, + timestamp: Some(self.timestamp), + }], + }; + + (mc, self.result_sender) + } } /// Background worker that processes Claude API requests serially. @@ -19,8 +73,12 @@ pub struct ClaudeWorker { client: reqwest::Client, api_key: String, system_prompt: String, + // FIXME: use this and append to system prompt within tags + #[allow(dead_code)] + summary: String, + messages: Vec, tool_executor: Weak, - request_rx: mpsc::Receiver, + input_rx: mpsc::Receiver, stop_rx: mpsc::Receiver<()>, } @@ -31,7 +89,7 @@ impl ClaudeWorker { pub fn new( config: ClaudeConfig, tool_executor: Weak, - request_rx: mpsc::Receiver, + input_rx: mpsc::Receiver, stop_rx: mpsc::Receiver<()>, ) -> Result { let api_key = std::fs::read_to_string(&config.api_key_file) @@ -47,8 +105,10 @@ impl ClaudeWorker { client: reqwest::Client::new(), api_key, system_prompt, + summary: String::new(), + messages: Default::default(), tool_executor, - request_rx, + input_rx, stop_rx, }) } @@ -57,18 +117,29 @@ impl ClaudeWorker { pub async fn run(mut self) { loop { tokio::select! { - request = self.request_rx.recv() => { - let Some(request) = request else { + input = self.input_rx.recv() => { + let Some(input) = input else { // Channel closed, exit break; }; - let result = self.handle_request(&request.prompt).await; - // If handle_request was interrupted by stop request, go on to drain the queues - if matches!(result, Err(ClaudeError::StopRequested)) { - self.handle_stop(); + match input { + Input::Chat(msg) => { + let (mc, maybe_sender) = msg.into_content_and_sender(); + self.messages.push(mc); + if let Some(sender) = maybe_sender { + let result = self.handle_request().await; + // If handle_request was interrupted by stop request, go on to drain the queues + if matches!(result, Err(ClaudeError::StopRequested)) { + self.handle_stop(); + } + // Ignore send errors - the caller may have dropped the receiver + let _ = sender.send(result); + } + }, + Input::Compact => { + self.handle_compact().await; + } } - // Ignore send errors - the caller may have dropped the receiver - let _ = request.result_sender.send(result); } _ = self.stop_rx.recv() => { self.handle_stop(); @@ -82,9 +153,18 @@ impl ClaudeWorker { // Drain the stop_rx queue while self.stop_rx.try_recv().is_ok() {} - // Drain the request_rx queue and send StopRequested to each - while let Ok(request) = self.request_rx.try_recv() { - let _ = request.result_sender.send(Err(ClaudeError::StopRequested)); + // Drain the input_rx queue and send StopRequested to each prompt request + while let Ok(input) = self.input_rx.try_recv() { + match input { + Input::Chat(msg) => { + let (mc, maybe_sender) = msg.into_content_and_sender(); + self.messages.push(mc); + if let Some(sender) = maybe_sender { + let _ = sender.send(Err(ClaudeError::StopRequested)); + } + } + Input::Compact => {} + } } } @@ -97,16 +177,25 @@ impl ClaudeWorker { } } + /// Perform compaction + async fn handle_compact(&mut self) { + // FIXME: we should actually try to summarize messages using an api request, and then store it, before tossing messages + self.messages.clear(); + } + /// Handle a single request to the Claude API. - async fn handle_request(&mut self, prompt: &str) -> Result { + async fn handle_request(&mut self) -> Result { let max_iterations = self.config.claude_max_iterations; // Get tools from the executor if still alive let executor = self.tool_executor.upgrade(); let tools = executor.as_ref().map(|te| te.tools()).unwrap_or_default(); - let mut messages = vec![MessageContent::user(prompt)]; - info!("Claude request: {}", prompt); + if let Some(last) = self.messages.last() { + if let Some(ContentBlock::Text { text, .. }) = last.content.first() { + info!("Claude request: {}", text); + } + } for iteration in 0..max_iterations { // Check for stop before making API call @@ -116,7 +205,7 @@ impl ClaudeWorker { model: &self.config.claude_model, max_tokens: self.config.claude_max_tokens, system: &self.system_prompt, - messages: messages.clone(), + messages: &self.messages, tools: tools.clone(), }; @@ -142,14 +231,15 @@ impl ClaudeWorker { ); // Check if we need to handle tool use - if response.stop_reason == "tool_use" { + if response.stop_reason.as_ref() == "tool_use" { // Try to get a strong reference to the executor let Some(executor) = self.tool_executor.upgrade() else { return Err(ClaudeError::ToolExecutorGone); }; // Add assistant's response to messages - messages.push(MessageContent::assistant(response.content.clone())); + self.messages + .push(MessageContent::assistant(response.content.clone())); // Execute each tool use and collect results for block in &response.content { @@ -168,7 +258,11 @@ impl ClaudeWorker { (err, true) } }; - messages.push(MessageContent::tool_result(id.clone(), result, is_error)); + self.messages.push(MessageContent::tool_result( + id.to_string(), + result, + is_error, + )); } } @@ -182,7 +276,7 @@ impl ClaudeWorker { .content .into_iter() .filter_map(|block| { - if let ContentBlock::Text { text } = block { + if let ContentBlock::Text { text, .. } = block { Some(text) } else { None @@ -205,7 +299,7 @@ struct MessagesRequest<'a> { model: &'a str, max_tokens: u32, system: &'a str, - messages: Vec, + messages: &'a [MessageContent], #[serde(skip_serializing_if = "Vec::is_empty")] tools: Vec, } @@ -213,33 +307,24 @@ struct MessagesRequest<'a> { /// A message in the conversation (can have multiple content blocks). #[derive(Clone, Debug, Serialize, Deserialize)] struct MessageContent { - role: String, + role: Box, content: Vec, } impl MessageContent { - fn user(text: &str) -> Self { - Self { - role: "user".to_owned(), - content: vec![ContentBlock::Text { - text: text.to_owned(), - }], - } - } - fn assistant(blocks: Vec) -> Self { Self { - role: "assistant".to_owned(), + role: "assistant".into(), content: blocks, } } fn tool_result(tool_use_id: String, content: String, is_error: bool) -> Self { Self { - role: "user".to_owned(), + role: "user".into(), content: vec![ContentBlock::ToolResult { - tool_use_id, - content, + tool_use_id: tool_use_id.into(), + content: content.into(), is_error: if is_error { Some(true) } else { None }, }], } @@ -251,16 +336,18 @@ impl MessageContent { #[serde(tag = "type", rename_all = "snake_case")] enum ContentBlock { Text { - text: String, + text: Box, + #[serde(default, skip_serializing_if = "Option::is_none")] + timestamp: Option>, }, ToolUse { - id: String, - name: String, + id: Box, + name: Box, input: Value, }, ToolResult { - tool_use_id: String, - content: String, + tool_use_id: Box, + content: Box, #[serde(skip_serializing_if = "Option::is_none")] is_error: Option, }, @@ -270,7 +357,7 @@ enum ContentBlock { #[derive(Debug, Deserialize)] struct MessagesResponse { content: Vec, - stop_reason: String, + stop_reason: Box, } /// Error response from the Claude API. @@ -281,5 +368,5 @@ struct ErrorResponse { #[derive(Deserialize)] struct ApiErrorDetail { - message: String, + message: Box, } diff --git a/signal-gateway/src/gateway/mod.rs b/signal-gateway/src/gateway/mod.rs index 378c292..b3a0778 100644 --- a/signal-gateway/src/gateway/mod.rs +++ b/signal-gateway/src/gateway/mod.rs @@ -4,7 +4,7 @@ use crate::signal_jsonrpc::connect_ipc; use crate::{ alertmanager::AlertPost, - claude::{ClaudeApi, ClaudeConfig, Tool, ToolExecutor}, + claude::{ClaudeApi, ClaudeConfig, SentBy, Tool, ToolExecutor}, log_message::{LogMessage, Origin}, message_handler::{ AdminMessage, AdminMessageResponse, Context, MessageHandler, MessageHandlerResult, @@ -147,6 +147,9 @@ enum GatewayCommand { /// Stop current Claude request #[conf(name = "cs", alias = "CS")] ClaudeStop, + /// Compact Claude's message history + #[conf(name = "compact", alias = "COMPACT")] + ClaudeCompact, } /// Parse a gateway command from a string (with or without leading /) @@ -405,9 +408,13 @@ impl Gateway { SignalMessage { sender: self.config.signal_account.clone(), target, - message, + message: message.clone(), attachments, }.send(signal_cli).await?; + + if let Some(claude) = self.claude.get() { + claude.record_message(SentBy::System, &message, Utc::now().timestamp_millis() as u64); + } } else { warn!("alert_rx is closed, halting service"); self.token.cancel(); @@ -455,7 +462,7 @@ impl Gateway { AdminMessageResponse::new(text) }); - let attachments = resp.attachments.into_iter().map(|p| p.to_str().expect("attachments must have utf8 paths").to_owned()).collect(); + let attachments = resp.attachments.iter().map(|p| p.to_str().expect("attachments must have utf8 paths").to_owned()).collect(); // Reply to group if message came from a group, otherwise reply to sender let target = if let Some(group_id) = from_group { @@ -467,9 +474,14 @@ impl Gateway { SignalMessage { sender: self.config.signal_account.clone(), target, - message: resp.text, + message: resp.text.clone(), attachments, }.send(signal_cli).await?; + + if let Some(claude) = self.claude.get() { + let sent_by = if resp.is_claude { SentBy::Claude } else { SentBy::System }; + claude.record_message(sent_by, &resp.text, Utc::now().timestamp_millis() as u64); + } } } } @@ -488,8 +500,31 @@ impl Gateway { // Parse the command using conf let cmd = parse_gateway_command(&data.message).map_err(|err| (400u16, err.into()))?; - self.handle_gateway_command(cmd).await + // Record this as a system command in Claude's history, unless it's a Claude + // prompt command (which will be recorded when we call request()) + let is_claude = matches!(cmd, GatewayCommand::Claude { .. }); + if !is_claude { + if let Some(claude) = self.claude.get() { + claude.record_message( + SentBy::UserToSystem, + &data.message, + data.timestamp, + ); + } + } + + let resp = self.handle_gateway_command(cmd, data.timestamp).await?; + Ok(if is_claude { resp.from_claude() } else { resp }) } else if let Some(handler) = &self.message_handler { + // Record this as a system message in Claude's history (not directed at Claude) + if let Some(claude) = self.claude.get() { + claude.record_message( + SentBy::UserToSystem, + &data.message, + data.timestamp, + ); + } + let msg = AdminMessage { message: data.message.clone(), timestamp: data.timestamp, @@ -566,7 +601,11 @@ impl Gateway { } } - async fn handle_gateway_command(&self, cmd: GatewayCommand) -> MessageHandlerResult { + async fn handle_gateway_command( + &self, + cmd: GatewayCommand, + ts_ms: u64, + ) -> MessageHandlerResult { match cmd { GatewayCommand::Log { filter } => { let text = self.log_handler.format_logs(filter.as_deref()).await; @@ -713,7 +752,7 @@ impl Gateway { .ok_or_else(|| (501u16, "claude was not configured".into()))?; let prompt_text = prompt.join(" "); - match claude.request(&prompt_text).await { + match claude.request(&prompt_text, ts_ms).await { Ok(response) => Ok(AdminMessageResponse::new(response)), Err(err) => Err((500, err.to_string().into())), } @@ -727,6 +766,15 @@ impl Gateway { claude.request_stop(); Ok(AdminMessageResponse::new("stop requested")) } + GatewayCommand::ClaudeCompact => { + let claude = self + .claude + .get() + .ok_or_else(|| (501u16, "claude was not configured".into()))?; + + claude.request_compaction(); + Ok(AdminMessageResponse::new("compaction requested")) + } } } diff --git a/signal-gateway/src/message_handler.rs b/signal-gateway/src/message_handler.rs index e0dfe6d..2b644fd 100644 --- a/signal-gateway/src/message_handler.rs +++ b/signal-gateway/src/message_handler.rs @@ -35,6 +35,8 @@ pub struct AdminMessageResponse { pub text: String, /// Optional file attachments to include with the response. pub attachments: Vec, + /// Whether this response came from Claude (true) or the system (false). + pub(crate) is_claude: bool, } impl AdminMessageResponse { @@ -43,6 +45,7 @@ impl AdminMessageResponse { Self { text: text.into(), attachments: Vec::new(), + is_claude: false, } } @@ -62,6 +65,12 @@ impl AdminMessageResponse { self.attachments.extend(paths.into_iter().map(Into::into)); self } + + /// Mark this response as coming from Claude. + pub(crate) fn from_claude(mut self) -> Self { + self.is_claude = true; + self + } } /// Builder for constructing an [`AdminMessageResponse`]. @@ -95,6 +104,7 @@ impl AdminMessageResponseBuilder { AdminMessageResponse { text: self.text, attachments: self.attachments, + is_claude: false, } } }