ai-dispatch 11.2.0

// Claude stream-json parsing helpers shared by the Claude adapter and tests.
// Exports parse_event_line() plus internal helpers for assistant, tool, and result events.
// Depends on serde_json for decoding and capped_detail for concise event details.
use chrono::Local;
use serde_json::{Value, json};
use super::truncate::{capped_detail, capped_detail_with};
use crate::types::*;

pub(crate) fn parse_event_line(task_id: &TaskId, line: &str) -> Option<TaskEvent> {
    let trimmed = line.trim();
    if trimmed.is_empty() {
        return None;
    }
    let v: Value = serde_json::from_str(trimmed).ok()?;
    let now = Local::now();
    if trimmed.contains("NO_CHANGES_NEEDED") {
        return Some(TaskEvent {
            task_id: task_id.clone(),
            timestamp: now,
            event_kind: EventKind::NoOp,
            detail: extract_noop_reason(trimmed),
            metadata: None,
        });
    }
    match v.get("type").and_then(|value| value.as_str())? {
        "assistant" => parse_assistant_event(task_id, &v, now),
        "tool_use" => parse_tool_use_event(task_id, &v, now),
        "result" => parse_result_event(task_id, &v, now),
        "system" => parse_system_event(task_id, &v, now),
        "user" => parse_user_event(task_id, &v, now),
        "error" => parse_error_event(task_id, &v, now),
        _ => None,
    }
}

fn parse_assistant_event(
    task_id: &TaskId,
    v: &Value,
    now: chrono::DateTime<Local>,
) -> Option<TaskEvent> {
    let message = v.get("message")?;
    let content = message.get("content")?.as_array()?;
    if let Some(item) = content
        .iter()
        .find(|item| item.get("type").and_then(|value| value.as_str()) == Some("tool_use"))
    {
        return build_tool_event(task_id, item, now);
    }
    let text = content.iter().find_map(|item| {
        if item.get("type").and_then(|value| value.as_str()) == Some("text") {
            item.get("text").and_then(|value| value.as_str())
        } else {
            None
        }
    })?;
    let metadata = base_metadata(
        message.get("model").and_then(|value| value.as_str()),
        v.get("session_id").and_then(|value| value.as_str()),
    );
    let (detail, metadata) = capped_detail_with(text, metadata);
    Some(TaskEvent {
        task_id: task_id.clone(),
        timestamp: now,
        event_kind: EventKind::Reasoning,
        detail,
        metadata,
    })
}

fn parse_tool_use_event(
    task_id: &TaskId,
    v: &Value,
    now: chrono::DateTime<Local>,
) -> Option<TaskEvent> {
    let tool = v.get("tool").unwrap_or(v);
    build_tool_event(task_id, tool, now)
}

fn build_tool_event(
    task_id: &TaskId,
    tool: &Value,
    now: chrono::DateTime<Local>,
) -> Option<TaskEvent> {
    let name = tool.get("name").and_then(|value| value.as_str()).unwrap_or("tool");
    let command = tool
        .pointer("/input/command")
        .and_then(|value| value.as_str())
        .or_else(|| tool.pointer("/input/description").and_then(|value| value.as_str()));
    let text = command
        .map(|value| format!("{name}: {value}"))
        .unwrap_or_else(|| name.to_string());
    let (detail, metadata) = capped_detail(&text);
    Some(TaskEvent {
        task_id: task_id.clone(),
        timestamp: now,
        event_kind: EventKind::ToolCall,
        detail,
        metadata,
    })
}

fn parse_result_event(
    task_id: &TaskId,
    v: &Value,
    now: chrono::DateTime<Local>,
) -> Option<TaskEvent> {
    let payload = v.get("result").filter(|value| value.is_object()).unwrap_or(v);
    let usage = payload.get("usage").unwrap_or(payload);
    let input_tokens = usage_token_count(usage, "input_tokens", "tokens_in");
    let output_tokens = usage_token_count(usage, "output_tokens", "tokens_out");
    let cache_creation_tokens = usage_token_count(
        usage, "cache_creation_input_tokens", "cache_creation_input_tokens",
    );
    let cache_read_tokens = usage_token_count(usage, "cache_read_input_tokens", "cache_read_input_tokens");
    let total_tokens = input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens;
    let detail = format!(
        "tokens: {} in + {} out = {} ({} cache create, {} cache read)",
        input_tokens,
        output_tokens,
        total_tokens,
        cache_creation_tokens,
        cache_read_tokens
    );
    let metadata = result_metadata(payload, v, json!({
        "tokens": total_tokens,
        "input_tokens": input_tokens,
        "uncached_input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "prompt_tokens": input_tokens,
        "cache_creation_input_tokens": cache_creation_tokens,
        "cache_read_input_tokens": cache_read_tokens,
    }));
    let (detail, metadata) = capped_detail_with(&detail, Some(metadata));
    Some(TaskEvent {
        task_id: task_id.clone(),
        timestamp: now,
        event_kind: EventKind::Completion,
        detail,
        metadata,
    })
}

fn usage_token_count(usage: &Value, key: &str, alias: &str) -> i64 {
    usage.get(key).or_else(|| usage.get(alias)).and_then(Value::as_i64).unwrap_or(0)
}

fn result_metadata(payload: &Value, envelope: &Value, mut metadata: Value) -> Value {
    let cost_usd = payload
        .get("total_cost_usd")
        .or_else(|| payload.get("cost_usd"))
        .and_then(|value| value.as_f64());
    let model = extract_result_model(payload);
    let session_id = payload
        .get("session_id")
        .or_else(|| envelope.get("session_id"))
        .and_then(|value| value.as_str());
    if let Some(cost) = cost_usd {
        metadata["cost_usd"] = json!(cost);
    }
    if let Some(model) = model {
        metadata["model"] = json!(model);
    }
    if let Some(session_id) = session_id {
        metadata["agent_session_id"] = json!(session_id);
    }
    metadata
}

fn parse_system_event(
    task_id: &TaskId,
    v: &Value,
    now: chrono::DateTime<Local>,
) -> Option<TaskEvent> {
    let subtype = v.get("subtype").and_then(|value| value.as_str())?;
    if subtype == "init" {
        let detail = v
            .get("model")
            .and_then(|value| value.as_str())
            .map(|model| format!("init: {model}"))
            .unwrap_or_else(|| "init".to_string());
        let metadata = base_metadata(
            v.get("model").and_then(|value| value.as_str()),
            v.get("session_id").and_then(|value| value.as_str()),
        );
        let (detail, metadata) = capped_detail_with(&detail, metadata);
        return Some(TaskEvent {
            task_id: task_id.clone(),
            timestamp: now,
            event_kind: EventKind::Reasoning,
            detail,
            metadata,
        });
    }
    if subtype == "hook_response" && v.get("outcome").and_then(|value| value.as_str()) == Some("error")
    {
        let text = v
            .get("stderr")
            .or_else(|| v.get("output"))
            .and_then(|value| value.as_str())?;
        let (detail, metadata) = capped_detail(text);
        return Some(TaskEvent {
            task_id: task_id.clone(),
            timestamp: now,
            event_kind: EventKind::Error,
            detail,
            metadata,
        });
    }
    None
}

fn parse_user_event(task_id: &TaskId, v: &Value, now: chrono::DateTime<Local>) -> Option<TaskEvent> {
    let detail = v
        .get("tool_use_result")
        .and_then(|value| value.as_str())
        .or_else(|| {
            v.pointer("/message/content/0/content")
                .and_then(|value| value.as_str())
        })?;
    let is_error = v
        .pointer("/message/content/0/is_error")
        .and_then(|value| value.as_bool())
        .unwrap_or(false);
    if !is_error {
        return None;
    }
    // No marking here. `is_error` on a tool_result block is the *tool* failing —
    // a command that exited non-zero, a rejected edit, a file that was not
    // there — and its text is whatever the model asked for. This wrote
    // `~/.aid/rate-limit-claude` on 2026-08-07 off an agent's own message, and a
    // provider that was serving the whole time was held on opencode's needle.
    let (detail, metadata) = capped_detail(detail);
    Some(TaskEvent { task_id: task_id.clone(), timestamp: now, event_kind: EventKind::Error, detail, metadata })
}

fn parse_error_event(task_id: &TaskId, v: &Value, now: chrono::DateTime<Local>) -> Option<TaskEvent> {
    let detail = v.get("message").or_else(|| v.pointer("/error/message")).and_then(|value| value.as_str()).filter(|msg| !msg.is_empty())?;
    crate::routing::quota::channel::mark_stream_refusal(AgentKind::Claude, None, &v.to_string());
    let (detail, metadata) = capped_detail(detail);
    Some(TaskEvent {
        task_id: task_id.clone(),
        timestamp: now,
        event_kind: EventKind::Error,
        detail,
        metadata,
    })
}

fn base_metadata(model: Option<&str>, session_id: Option<&str>) -> Option<Value> {
    match (model, session_id) {
        (None, None) => None,
        _ => {
            let mut metadata = json!({});
            if let Some(model) = model {
                metadata["model"] = json!(normalize_model(model));
            }
            if let Some(session_id) = session_id {
                metadata["agent_session_id"] = json!(session_id);
            }
            Some(metadata)
        }
    }
}

fn extract_result_model(payload: &Value) -> Option<String> {
    payload
        .pointer("/modelUsage")
        .and_then(Value::as_object)
        .and_then(|models| {
            let use_cost = models.values().all(|v| v.get("costUSD").and_then(Value::as_f64).is_some());
            let tokens = |v: &Value| ["inputTokens", "outputTokens", "cacheReadInputTokens", "cacheCreationInputTokens"]
                .iter().filter_map(|key| v.get(*key).and_then(Value::as_i64)).map(i128::from).sum::<i128>();
            models.iter().rev().max_by(|(_, a), (_, b)| {
                let cost = || a["costUSD"].as_f64().zip(b["costUSD"].as_f64()).and_then(|(a, b)| a.partial_cmp(&b));
                let cost_order = if use_cost { cost() } else { None };
                cost_order.filter(|order| !order.is_eq()).unwrap_or_else(|| tokens(a).cmp(&tokens(b)))
            }).map(|(name, _)| name)
        })
        .map(|name| normalize_model(name))
        .or_else(|| payload.get("model").and_then(Value::as_str).map(normalize_model))
}

fn normalize_model(model: &str) -> String {
    model.split('[').next().unwrap_or(model).to_string()
}

fn extract_noop_reason(line: &str) -> String {
    if let Some(pos) = line.find("NO_CHANGES_NEEDED:") {
        format!("NO_CHANGES_NEEDED: {}", line[pos + 18..].trim().trim_matches('"'))
    } else {
        "NO_CHANGES_NEEDED".to_string()
    }
}

#[cfg(test)]
#[path = "claude_events_tests.rs"]
mod tests;