88 lines
3.3 KiB
Rust
88 lines
3.3 KiB
Rust
|
//! 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,
|
||
|
|
}
|
||
|
|
}
|