huskies: merge 845

server/src/llm/providers/claude_code/events/mod.rs

@@ -0,0 +1,87 @@
+//! Claude Code event handling — dispatches JSON events from the CLI to message channels.
+
+mod stream;
+
+#[cfg(test)]
+mod tests;
+
+use super::parse::{parse_assistant_message, parse_tool_results};
+use crate::llm::types::Message;
+use crate::slog;
+use std::sync::atomic::{AtomicBool, Ordering};
+use stream::handle_stream_event;
+
+/// Dispatch a top-level JSON event from the Claude Code CLI.
+///
+/// Routes the event to the appropriate handler based on `type`, emitting tokens,
+/// thinking output, activity signals, and parsed messages to their respective channels.
+/// Returns `true` only for `"result"` events, which signal that the CLI turn is complete.
+pub(super) fn process_json_event(
+    json: &serde_json::Value,
+    token_tx: &tokio::sync::mpsc::UnboundedSender<String>,
+    thinking_tx: &tokio::sync::mpsc::UnboundedSender<String>,
+    activity_tx: &tokio::sync::mpsc::UnboundedSender<String>,
+    msg_tx: &std::sync::mpsc::Sender<Message>,
+    sid_tx: &mut Option<tokio::sync::oneshot::Sender<String>>,
+    auth_failed: &AtomicBool,
+) -> bool {
+    let event_type = match json.get("type").and_then(|t| t.as_str()) {
+        Some(t) => t,
+        None => return false,
+    };
+
+    // Capture session_id from the first event that carries it
+    if let Some(tx) = sid_tx.take() {
+        if let Some(sid) = json.get("session_id").and_then(|s| s.as_str()) {
+            slog!("[pty-debug] CAPTURED session_id: {}", sid);
+            let _ = tx.send(sid.to_string());
+        } else {
+            *sid_tx = Some(tx);
+        }
+    }
+
+    // Detect authentication_failed at the top level of any event.
+    if json.get("error").and_then(|e| e.as_str()) == Some("authentication_failed") {
+        slog!("[pty-debug] Detected authentication_failed error");
+        auth_failed.store(true, Ordering::Relaxed);
+    }
+
+    match event_type {
+        "stream_event" => {
+            if let Some(event) = json.get("event") {
+                handle_stream_event(event, token_tx, thinking_tx, activity_tx);
+            }
+            false
+        }
+        "assistant" => {
+            if let Some(message) = json.get("message")
+                && let Some(content) = message.get("content").and_then(|c| c.as_array())
+            {
+                // Fire activity signals for tool_use blocks as a fallback path.
+                // The primary path is via stream_event → content_block_start (real-time),
+                // but assistant events also carry tool_use blocks and serve as a reliable
+                // backup if stream_event delivery is delayed or missed.
+                for block in content {
+                    if block.get("type").and_then(|t| t.as_str()) == Some("tool_use")
+                        && let Some(name) = block.get("name").and_then(|n| n.as_str())
+                    {
+                        let _ = activity_tx.send(name.to_string());
+                    }
+                }
+                parse_assistant_message(content, msg_tx);
+            }
+            false
+        }
+        "user" => {
+            if let Some(message) = json.get("message")
+                && let Some(content) = message.get("content").and_then(|c| c.as_array())
+            {
+                parse_tool_results(content, msg_tx);
+            }
+            false
+        }
+        "result" => true,
+        // system, rate_limit_event, and unknown types are no-ops
+        _ => false,
+    }
+}