magi-code 0.97.4

//! Claude CLI subscription adapter. No host tools run inside Claude Code.
use super::{
    Provider, ProviderEvent, ProviderRequest,
    anthropic::{AnthropicStreamParser, parse_captured_sse},
    claude_admission::{
        CLAUDE_MAX_REQUEST_DURATION, CLAUDE_STARTUP_TIMEOUT, ClaudeAdmission, claude_error_hint,
        claude_error_with_hint,
    },
    claude_projection::{CLAUDE_TOOL_PREFIX, ClaudeProjection, is_valid_host_tool_name},
    transport::dispatch_utf8_bytes,
};
use crate::{
    cancellation::AgentCancellation,
    tools::process::{PIPE_READER_JOIN_TIMEOUT, terminate_child_tree_and_wait},
};
use anyhow::{Context, Result, bail};
use serde_json::{Value, json};
use std::{
    collections::HashSet,
    ffi::OsString,
    io::{BufRead, BufReader, Read},
    path::{Path, PathBuf},
    process::{Child, ChildStderr, ChildStdout, Command, ExitStatus, Stdio},
    sync::{Mutex, mpsc},
    thread,
    time::{Duration, Instant},
};

const MAX_NATIVE_LINE: u64 = 1024 * 1024;
const MAX_CAPTURED_STDERR: usize = 4096;
const NATIVE_PROGRESS_TIMEOUT: Duration = Duration::from_secs(300);
// One successful auth probe serves provider builds, readiness checks and catalog
// loads within this window instead of starting a Node process for each.
const CLI_STATUS_CACHE_TTL: Duration = Duration::from_secs(5);
pub(crate) const CLAUDE_CLI_PROMPT: &str = include_str!("../../prompts/claude-cli.md");

static CLI_STATUS_CACHE: Mutex<Option<(Instant, Value)>> = Mutex::new(None);

/// `cwd` is the host workspace; Claude CLI reports it as its working directory.
pub(super) struct ClaudeSubscriptionProvider {
    pub(super) cwd: PathBuf,
}

// Keep API billing and alternate backends out of subscription-only child processes.
// The parent environment remains available to the separate Anthropic API provider.
fn configure_subscription_auth(command: &mut Command) {
    for key in [
        "ANTHROPIC_API_KEY",
        "ANTHROPIC_AUTH_TOKEN",
        "ANTHROPIC_FOUNDRY_API_KEY",
        "CLAUDE_CODE_USE_BEDROCK",
        "CLAUDE_CODE_USE_VERTEX",
        "CLAUDE_CODE_USE_FOUNDRY",
    ] {
        command.env_remove(key);
    }
}

/// Loopback first, then the user's entries from both spellings, kept as raw OS strings.
fn claude_loopback_no_proxy() -> OsString {
    let mut merged = OsString::from("127.0.0.1,localhost");
    let mut seen = Vec::new();
    for key in ["NO_PROXY", "no_proxy"] {
        if let Some(value) = std::env::var_os(key).filter(|value| !value.is_empty())
            && !seen.contains(&value)
        {
            merged.push(",");
            merged.push(&value);
            seen.push(value);
        }
    }
    merged
}

pub(crate) fn cli_subscription_ready() -> bool {
    probe_cli_subscription().is_ok()
}

/// Bypasses the status cache; periodic account checks must observe sign-outs and switches.
pub(crate) fn refresh_cli_subscription_status() -> Option<Value> {
    probe_and_cache_cli_subscription().ok()
}

/// Explains why Claude subscription is unavailable so users can fix setup.
/// Reuses a recent successful probe so one launch does not start a Node process per caller.
pub(crate) fn probe_cli_subscription() -> Result<Value, String> {
    let cached = CLI_STATUS_CACHE
        .lock()
        .unwrap_or_else(|p| p.into_inner())
        .as_ref()
        .filter(|(probed_at, _)| probed_at.elapsed() < CLI_STATUS_CACHE_TTL)
        .map(|(_, status)| status.clone());
    match cached {
        Some(status) => Ok(status),
        None => probe_and_cache_cli_subscription(),
    }
}

// Only successes are cached: a fresh `claude auth login` is noticed immediately.
// The lock is never held across the probe, so a slow or failing CLI cannot queue callers.
fn probe_and_cache_cli_subscription() -> Result<Value, String> {
    let status = probe_cli_subscription_uncached();
    *CLI_STATUS_CACHE.lock().unwrap_or_else(|p| p.into_inner()) = status
        .as_ref()
        .ok()
        .map(|status| (Instant::now(), status.clone()));
    status
}

fn is_executable_file(path: &Path) -> bool {
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::metadata(path)
            .is_ok_and(|metadata| metadata.is_file() && metadata.permissions().mode() & 0o111 != 0)
    }
    #[cfg(not(unix))]
    {
        path.is_file()
    }
}

/// `claude` from `PATH`, else native installer default `~/.local/bin/claude`
/// (installer does not always put it on `PATH`, e.g. non-login shells or WSL).
/// Non-executable entries are skipped, matching shell lookup.
fn claude_cli_program() -> PathBuf {
    std::env::var_os("PATH")
        .and_then(|path| {
            std::env::split_paths(&path)
                .map(|dir| dir.join("claude"))
                .find(|path| is_executable_file(path))
        })
        .or_else(|| {
            dirs::home_dir()
                .map(|home| home.join(".local/bin/claude"))
                .filter(|path| is_executable_file(path))
        })
        .and_then(|path| std::fs::canonicalize(path).ok())
        .unwrap_or_else(|| "claude".into())
}

fn probe_cli_subscription_uncached() -> Result<Value, String> {
    if !cfg!(unix) {
        return Err("Claude subscription requires Unix (macOS, Linux, or WSL)".into());
    }
    use crate::tools::process::{BoundedChildProcessLimits, run_bounded_child_process};
    let mut command = Command::new(claude_cli_program());
    command
        .args(["auth", "status", "--json"])
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    configure_subscription_auth(&mut command);
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        command.process_group(0);
    }
    let child = command.spawn().map_err(|error| {
        format!("cannot run `claude` ({error}); install Claude Code and put `claude` on PATH")
    })?;
    let output = run_bounded_child_process(
        child,
        BoundedChildProcessLimits {
            stdout_max_bytes: 8192,
            stderr_max_bytes: 16 * 1024,
            // Cold npm/Node starts, especially under WSL, can exceed a few seconds.
            timeout: Duration::from_secs(10),
            poll_interval: Duration::from_millis(20),
        },
        &AgentCancellation::default(),
    )
    .map_err(|error| format!("`claude auth status --json` failed: {error}"))?;
    if output.timed_out {
        return Err("`claude auth status --json` timed out".into());
    }
    if output.stdout_truncated || output.stderr_truncated {
        return Err("`claude auth status --json` output too large".into());
    }
    if let Some(warning) = output.cleanup_warning {
        return Err(format!("`claude auth status --json` {warning}"));
    }
    if !output.status.is_some_and(|status| status.success()) {
        return Err(
            "`claude auth status --json` exited unsuccessfully; run `claude auth login`".into(),
        );
    }
    let status = serde_json::from_str::<Value>(&output.stdout)
        .map_err(|_| "`claude auth status --json` returned invalid JSON; update Claude Code")?;
    if status.get("loggedIn").and_then(Value::as_bool) != Some(true) {
        return Err("Claude CLI not signed in; run `claude auth login`".into());
    }
    if status.get("authMethod").and_then(Value::as_str) != Some("claude.ai") {
        return Err(
            "Claude CLI not signed in with a claude.ai subscription; run `claude auth login`"
                .into(),
        );
    }
    Ok(status)
}

impl Provider for ClaudeSubscriptionProvider {
    fn stream_cancellable(
        &self,
        request: ProviderRequest,
        cancellation: &AgentCancellation,
        on_event: &mut dyn FnMut(ProviderEvent) -> Result<()>,
    ) -> Result<()> {
        cancellation.check()?;
        if cfg!(not(unix)) {
            bail!("Claude subscription process-tree cleanup requires Unix");
        }
        let estimated_tokens = crate::context::project_provider_request_input_tokens(
            super::CLAUDE_SUBSCRIPTION_PROVIDER,
            &request,
        )
        .tokens;
        let ceiling = super::claude_subscription_prompt_limit(&request.model);
        if estimated_tokens > ceiling {
            return Err(super::error::ProviderError::prompt_too_long(format!(
                "Claude CLI prompt is estimated at {estimated_tokens} tokens, exceeding its {ceiling}-token limit; no CLI request was sent"
            ))
            .into());
        }
        let projection = ClaudeProjection::from_request(&request)?;
        let root = tempfile::tempdir().context("creating Claude request directory")?;
        let files = ClaudeRequestFiles::write(root.path(), &projection)?;
        let relay = ClaudeAdmission::start(cancellation, projection.messages)?;
        let mut command = claude_cli_command(&self.cwd, &request, &files, &relay.url)?;
        let child = command
            .spawn()
            .context("Claude CLI not installed; install and sign in to Claude Code")?;
        let mut child = OwnedChild(Some(child));
        let (receiver, stdout_reader, stderr_reader) =
            spawn_claude_pipe_readers(child.0.as_mut().context("Claude child missing")?)?;
        let mut parser = AnthropicStreamParser::default();
        let mut pending_utf8 = Vec::new();
        let mut read_bytes = 0;
        // Delivered events are kept to prove the captured replay starts with them.
        let mut streamed_events = Vec::new();
        let mut tools_pending = false;
        let mut stream_progress = |deadline: &mut NativeDeadline| {
            let bytes = relay.stream_bytes(&mut read_bytes)?;
            dispatch_utf8_bytes(&mut pending_utf8, &bytes, &mut |chunk| {
                let parsed = parser.push_chunk_outcome(chunk).map_err(|error| {
                    claude_error_with_hint(
                        anyhow::anyhow!("Claude upstream stream parsing failed"),
                        claude_error_hint(error.to_string().as_bytes()),
                    )
                })?;
                if parsed.semantic_progress {
                    deadline.observe_upstream_progress();
                }
                for event in parsed.events {
                    if matches!(event, ProviderEvent::Done) {
                        deadline.observe_upstream_completion();
                    }
                    // Keep host replay in upstream order. Once a tool appears, defer
                    // the entire suffix until terminal, usage and CLI validation.
                    tools_pending |=
                        matches!(event, ProviderEvent::ToolCall(_) | ProviderEvent::Done)
                            || matches!(&event, ProviderEvent::ResponseItem(item)
                            if item.get("type").and_then(Value::as_str) == Some("function_call"));
                    if !tools_pending {
                        cancellation.check()?;
                        on_event(event.clone())?;
                        streamed_events.push(event);
                    }
                }
                Ok(())
            })
        };
        let mut deadline = NativeDeadline::new(
            CLAUDE_STARTUP_TIMEOUT,
            request
                .semantic_progress_timeout()
                .unwrap_or(NATIVE_PROGRESS_TIMEOUT),
            CLAUDE_MAX_REQUEST_DURATION,
        );
        let outcome = run_native(&receiver, cancellation, &mut deadline, &mut stream_progress)
            .map_err(|error| {
                // Capture the live state before process cleanup changes the relay phase.
                let diagnostic =
                    format!("{error}; {}; {}", deadline.diagnostic(), relay.diagnostic());
                error.context(diagnostic)
            });
        // Always terminate descendants before checking relay status, even on successful CLI exit.
        let cleanup = child
            .0
            .take()
            .map(|mut process| terminate_child_tree_and_wait(&mut process))
            .context("Claude child missing")?;
        drop(receiver);
        join_pipe_reader(stdout_reader);
        let stderr = join_pipe_reader(stderr_reader);
        let stderr_hint = claude_error_hint(&stderr);
        cancellation.check()?;
        let cleanup = cleanup?;
        if let Some(warning) = cleanup.cleanup_warning {
            bail!("Claude process cleanup failed: {warning}");
        }
        let response = relay.finish();
        let completion = outcome.map_err(|error| {
            let error = match response.as_ref() {
                Err(relay_error) => {
                    let message = format!("{error}; {relay_error}");
                    error.context(message)
                }
                Ok(_) => error,
            };
            claude_error_with_hint(error, stderr_hint)
        })?;
        let hint = completion.error_hint.or(stderr_hint);
        let response = response.map_err(|error| claude_error_with_hint(error, hint))?;
        let events = validate_native_completion(&completion, cleanup.status, &response.body, hint)?;
        if events.get(..streamed_events.len()) != Some(streamed_events.as_slice()) {
            bail!(
                "Claude captured response diverged from streamed events; deferred events were not delivered"
            );
        }
        let names: HashSet<&str> = projection
            .manifest
            .iter()
            .filter_map(|tool| tool.get("name").and_then(Value::as_str))
            .collect();
        for event in events.into_iter().skip(streamed_events.len()) {
            cancellation.check()?;
            on_event(remap_claude_tool_event(event, &names))?;
        }
        Ok(())
    }
}

/// Request-scoped inputs Claude CLI reads from disk; the temporary root owns cleanup.
struct ClaudeRequestFiles {
    system: PathBuf,
    settings: PathBuf,
    manifest: PathBuf,
}

impl ClaudeRequestFiles {
    fn write(root: &Path, projection: &ClaudeProjection) -> Result<Self> {
        let files = Self {
            system: root.join("system.md"),
            settings: root.join("settings.json"),
            manifest: root.join("tools.json"),
        };
        std::fs::write(&files.system, &projection.system)
            .context("writing Claude system prompt file")?;
        std::fs::write(&files.manifest, serde_json::to_vec(&projection.manifest)?)
            .context("writing Claude tool inventory file")?;
        let settings = json!({"env": {
            "CLAUDE_CODE_EXTRA_BODY": serde_json::to_string(&projection.extra_body)?
        }});
        std::fs::write(&files.settings, serde_json::to_vec(&settings)?)
            .context("writing Claude settings file")?;
        Ok(files)
    }
}

fn claude_cli_command(
    cwd: &Path,
    request: &ProviderRequest,
    files: &ClaudeRequestFiles,
    relay_url: &str,
) -> Result<Command> {
    let executable = std::env::current_exe()
        .context("locating magi-code executable for Claude tool inventory server")?;
    let executable = executable
        .to_str()
        .context("magi-code executable path is not UTF-8")?;
    let manifest = files
        .manifest
        .to_str()
        .context("Claude tool inventory path is not UTF-8")?;
    let mcp = json!({"mcpServers": {"magi": {"command": executable, "env": {
        "MAGI_CLAUDE_INERT_MCP_INVENTORY": manifest
    }}}});
    let no_proxy = claude_loopback_no_proxy();
    let mut command = Command::new(claude_cli_program());
    command
        .args([
            "-p",
            "Continue the supplied conversation.",
            "--model",
            &request.model,
            "--output-format",
            "stream-json",
            "--verbose",
            "--include-partial-messages",
            "--thinking-display",
            "summarized",
            "--tools",
            "",
            "--system-prompt-file",
        ])
        .arg(&files.system)
        .arg("--settings")
        .arg(&files.settings)
        .args([
            "--setting-sources",
            "",
            "--strict-mcp-config",
            "--disable-slash-commands",
            "--max-turns",
            "1",
            "--permission-mode",
            "dontAsk",
            "--no-session-persistence",
            "--mcp-config",
            &mcp.to_string(),
        ])
        .current_dir(cwd)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .env("ANTHROPIC_BASE_URL", relay_url)
        // The relay is plain HTTP on loopback and receives the subscription
        // Authorization header: keep it off every proxy. Runtimes disagree on
        // which spelling wins, so set both; the CLI's only plain-HTTP target is
        // the relay, so HTTP proxies are dropped while HTTPS proxies stay.
        .env("NO_PROXY", &no_proxy)
        .env("no_proxy", &no_proxy)
        .env_remove("HTTP_PROXY")
        .env_remove("http_proxy")
        .env("CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC", "1")
        .env("CLAUDE_CODE_MAX_RETRIES", "0")
        .env("CLAUDE_CODE_DISABLE_NONSTREAMING_FALLBACK", "1")
        .env("DISABLE_AUTO_COMPACT", "1")
        .env("DISABLE_COMPACT", "1")
        .env("ENABLE_TOOL_SEARCH", "false")
        .env("CLAUDE_CODE_TOTAL_TOKENS_REMINDER", "off")
        .env_remove("CLAUDE_CODE_EXTRA_BODY");
    configure_subscription_auth(&mut command);
    if let Some(effort) = request.thinking_level.explicit_effort() {
        command.args(["--effort", effort]);
    }
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        command.process_group(0);
    }
    Ok(command)
}

/// One message per native stdout line; non-`Line` variants end the stream.
enum NativeLine {
    Line(Vec<u8>),
    End,
    TooLong,
    ReadFailed,
}

type ClaudePipeReaders = (
    mpsc::Receiver<NativeLine>,
    thread::JoinHandle<()>,
    thread::JoinHandle<Vec<u8>>,
);

fn spawn_claude_pipe_readers(child: &mut Child) -> Result<ClaudePipeReaders> {
    let stdout = child.stdout.take().context("Claude CLI missing stdout")?;
    let stderr = child.stderr.take().context("Claude CLI missing stderr")?;
    let stderr_reader = thread::spawn(move || capture_stderr_prefix(stderr));
    let (sender, receiver) = mpsc::sync_channel(16);
    let stdout_reader = thread::spawn(move || forward_native_lines(stdout, &sender));
    Ok((receiver, stdout_reader, stderr_reader))
}

/// Drains stderr so the CLI never blocks on it; keeps only a bounded prefix for hints.
fn capture_stderr_prefix(mut source: ChildStderr) -> Vec<u8> {
    let mut captured = Vec::new();
    let mut buffer = [0u8; 4096];
    while let Ok(count) = source.read(&mut buffer) {
        if count == 0 {
            break;
        }
        let remaining = MAX_CAPTURED_STDERR.saturating_sub(captured.len());
        captured.extend_from_slice(&buffer[..count.min(remaining)]);
    }
    captured
}

fn forward_native_lines(stdout: ChildStdout, sender: &mpsc::SyncSender<NativeLine>) {
    let mut source = BufReader::new(stdout);
    loop {
        let mut line = Vec::new();
        let message = match source
            .by_ref()
            .take(MAX_NATIVE_LINE + 1)
            .read_until(b'\n', &mut line)
        {
            Ok(0) => NativeLine::End,
            Ok(_) if line.len() as u64 > MAX_NATIVE_LINE => NativeLine::TooLong,
            Ok(_) => NativeLine::Line(line),
            Err(_) => NativeLine::ReadFailed,
        };
        let finished = !matches!(message, NativeLine::Line(_));
        if sender.send(message).is_err() || finished {
            break;
        }
    }
}

/// Bounded join: a descendant that left the process group can hold a pipe open
/// indefinitely, so its reader is detached (counted and reported) instead of
/// blocking the agent worker. The thread exits once that descendant closes the pipe.
fn join_pipe_reader<T: Default>(reader: thread::JoinHandle<T>) -> T {
    let deadline = Instant::now() + PIPE_READER_JOIN_TIMEOUT;
    while !reader.is_finished() && Instant::now() < deadline {
        thread::sleep(Duration::from_millis(10));
    }
    if reader.is_finished() {
        reader.join().unwrap_or_default()
    } else {
        crate::tools::process::warn_detached_pipe_reader();
        T::default()
    }
}

/// Accepts a clean finish, or the `--max-turns 1` tool boundary (`error_max_turns`,
/// exit code 1) only when the captured upstream response really contains a tool call.
fn validate_native_completion(
    completion: &NativeCompletion,
    exit_status: Option<ExitStatus>,
    captured_body: &[u8],
    hint: Option<&'static str>,
) -> Result<Vec<ProviderEvent>> {
    if !(completion.assistant && completion.result && completion.stop) {
        return Err(claude_error_with_hint(
            anyhow::anyhow!(
                "Claude CLI missing assistant, result, or message_stop; check Claude CLI version and login"
            ),
            hint,
        ));
    }
    let events = parse_captured_sse(captured_body)?;
    let tool_boundary = completion.subtype == "error_max_turns"
        && events
            .iter()
            .any(|event| matches!(event, ProviderEvent::ToolCall(_)));
    if completion.is_error && !tool_boundary {
        return Err(claude_error_with_hint(
            anyhow::anyhow!("Claude CLI returned an error; check Claude CLI version and login"),
            hint,
        ));
    }
    if completion.subtype != "success" && !tool_boundary {
        bail!("Claude CLI did not complete successfully");
    }
    if !exit_status
        .is_some_and(|status| status.success() || (tool_boundary && status.code() == Some(1)))
    {
        bail!("Claude CLI exited unsuccessfully");
    }
    Ok(events)
}

/// Maps prefixed native tool names back to host names; unknown names become rejected
/// calls. Function-call items keep the native name for Claude replay.
fn remap_claude_tool_event(event: ProviderEvent, inventory: &HashSet<&str>) -> ProviderEvent {
    match event {
        ProviderEvent::ToolCall(mut call) => match host_tool_name(&call.name, inventory) {
            Ok(name) => {
                call.name = name.to_owned();
                ProviderEvent::ToolCall(call)
            }
            Err(error) => ProviderEvent::ToolCallRejected {
                call,
                reason: error.to_string(),
            },
        },
        ProviderEvent::ResponseItem(mut item)
            if item.get("type").and_then(Value::as_str) == Some("function_call") =>
        {
            item["claude_tool_name"] = item["name"].clone();
            let name = item.get("name").and_then(Value::as_str).unwrap_or("");
            if let Ok(name) = host_tool_name(name, inventory) {
                item["name"] = json!(name);
            }
            ProviderEvent::ResponseItem(item)
        }
        event => event,
    }
}

fn host_tool_name<'a>(name: &'a str, inventory: &HashSet<&str>) -> Result<&'a str> {
    let printable =
        if is_valid_host_tool_name(name.strip_prefix(CLAUDE_TOOL_PREFIX).unwrap_or(name)) {
            name
        } else {
            "[invalid tool name]"
        };
    let Some(host_name) = name.strip_prefix(CLAUDE_TOOL_PREFIX) else {
        bail!(
            "Claude CLI returned tool with unexpected name `{printable}` (expected {CLAUDE_TOOL_PREFIX} prefix)"
        );
    };
    if !inventory.contains(host_name) {
        bail!(
            "Claude CLI returned tool outside host inventory: `{printable}` (not provided for this request)"
        );
    }
    Ok(host_name)
}

struct OwnedChild(Option<Child>);
impl Drop for OwnedChild {
    fn drop(&mut self) {
        if let Some(mut child) = self.0.take() {
            let _ = terminate_child_tree_and_wait(&mut child);
        }
    }
}

struct NativeDeadline {
    started: Instant,
    startup_deadline: Instant,
    progress_deadline: Option<Instant>,
    completion_deadline: Option<Instant>,
    overall_deadline: Instant,
    progress_timeout: Duration,
    last_upstream_progress: Option<Instant>,
    last_native_progress: Option<Instant>,
    phase: &'static str,
    assistant: bool,
    result: bool,
    stop: bool,
}

impl NativeDeadline {
    fn new(startup: Duration, progress: Duration, overall: Duration) -> Self {
        let now = Instant::now();
        Self {
            started: now,
            startup_deadline: now + startup,
            progress_deadline: None,
            completion_deadline: None,
            overall_deadline: now + overall,
            progress_timeout: progress,
            last_upstream_progress: None,
            last_native_progress: None,
            phase: "writing input",
            assistant: false,
            result: false,
            stop: false,
        }
    }

    fn next_deadline(&self) -> Instant {
        self.completion_deadline
            .or(self.progress_deadline)
            .unwrap_or(self.startup_deadline)
            .min(self.overall_deadline)
    }

    fn check_overall(&self) -> Result<()> {
        if Instant::now() >= self.overall_deadline {
            bail!("Claude CLI exceeded overall time limit");
        }
        Ok(())
    }

    fn check(&self) -> Result<()> {
        self.check_overall()?;
        if Instant::now() >= self.next_deadline() {
            if self.completion_deadline.is_some() {
                bail!("Claude CLI timed out finalizing completed upstream response");
            }
            if self.progress_deadline.is_some() {
                bail!("Claude CLI timed out waiting for progress");
            }
            bail!("Claude CLI startup timed out waiting for output");
        }
        Ok(())
    }

    fn observe_upstream_progress(&mut self) {
        let now = Instant::now();
        self.last_upstream_progress = Some(now);
        self.progress_deadline = Some(now + self.progress_timeout);
    }

    fn observe_upstream_completion(&mut self) {
        // Terminal CLI events must arrive within one fixed grace period. Neither
        // late native output nor repeated upstream metadata can extend it.
        self.completion_deadline
            .get_or_insert_with(|| Instant::now() + self.progress_timeout);
        self.phase = "finalizing CLI";
    }

    fn observe_native_event(&mut self, event: &Value) {
        match event.get("type").and_then(Value::as_str) {
            Some("assistant") => self.assistant = true,
            Some("result") => self.result = true,
            Some("stream_event")
                if event.pointer("/event/type").and_then(Value::as_str) == Some("message_stop") =>
            {
                self.stop = true;
            }
            _ => {}
        }
        if native_event_has_progress(event) {
            let now = Instant::now();
            self.last_native_progress = Some(now);
            self.progress_deadline = Some(now + self.progress_timeout);
        }
    }

    fn diagnostic(&self) -> String {
        let age = |time: Option<Instant>| {
            time.map(|time| time.elapsed().as_millis().to_string())
                .unwrap_or_else(|| "none".to_owned())
        };
        format!(
            "Claude timing (phase={}, elapsed_ms={}, startup_timeout_ms={}, progress_timeout_ms={}, overall_timeout_ms={}, upstream_progress_age_ms={}, native_progress_age_ms={}, upstream_complete={}, assistant={}, result={}, message_stop={})",
            self.phase,
            self.started.elapsed().as_millis(),
            self.startup_deadline
                .duration_since(self.started)
                .as_millis(),
            self.progress_timeout.as_millis(),
            self.overall_deadline
                .duration_since(self.started)
                .as_millis(),
            age(self.last_upstream_progress),
            age(self.last_native_progress),
            self.completion_deadline.is_some(),
            self.assistant,
            self.result,
            self.stop,
        )
    }
}

// Count generated content, not pings, startup metadata, or history replay acknowledgments.
// Text streams live; tool calls and everything after them stay withheld until the
// complete upstream response has been validated.
fn native_event_has_progress(event: &Value) -> bool {
    if event.get("type").and_then(Value::as_str) != Some("stream_event") {
        return false;
    }
    let event = &event["event"];
    let nonempty = |value: &Value| value.as_str().is_some_and(|text| !text.is_empty());
    match event.get("type").and_then(Value::as_str) {
        Some("content_block_start") => {
            let block = &event["content_block"];
            match block.get("type").and_then(Value::as_str) {
                Some("tool_use") => nonempty(&block["id"]) && nonempty(&block["name"]),
                Some("text") => nonempty(&block["text"]),
                Some("thinking") => nonempty(&block["thinking"]),
                Some("redacted_thinking") => nonempty(&block["data"]),
                _ => false,
            }
        }
        Some("content_block_delta") => {
            let delta = &event["delta"];
            let field = match delta.get("type").and_then(Value::as_str) {
                Some("text_delta") => "text",
                Some("thinking_delta") => {
                    // Claude CLI can mask thinking text while reporting generated tokens.
                    return nonempty(&delta["thinking"])
                        || delta["estimated_tokens"]
                            .as_u64()
                            .is_some_and(|tokens| tokens > 0);
                }
                Some("signature_delta") => "signature",
                Some("redacted_thinking_delta") => "data",
                Some("input_json_delta") => "partial_json",
                _ => return false,
            };
            nonempty(&delta[field])
        }
        _ => false,
    }
}

fn receive(
    receiver: &mpsc::Receiver<NativeLine>,
    cancel: &AgentCancellation,
    deadline: &mut NativeDeadline,
    on_poll: &mut dyn FnMut(&mut NativeDeadline) -> Result<()>,
) -> Result<Option<Value>> {
    loop {
        cancel.check()?;
        deadline.check_overall()?;
        on_poll(deadline)?;
        // Drain already-queued native output before deciding that progress expired.
        let received = match receiver.try_recv() {
            Ok(line) => Ok(line),
            Err(mpsc::TryRecvError::Disconnected) => Err(mpsc::RecvTimeoutError::Disconnected),
            Err(mpsc::TryRecvError::Empty) => {
                deadline.check()?;
                let wait = deadline
                    .next_deadline()
                    .saturating_duration_since(Instant::now())
                    .min(Duration::from_millis(100));
                receiver.recv_timeout(wait)
            }
        };
        cancel.check()?;
        deadline.check_overall()?;
        // Progress may have reached the relay while the native receiver was waiting.
        on_poll(deadline)?;
        deadline.check_overall()?;
        match received {
            Ok(NativeLine::Line(line)) => {
                let event = serde_json::from_slice(&line)
                    .context("invalid Claude CLI stream-json output")?;
                deadline.observe_native_event(&event);
                deadline.check()?;
                return Ok(Some(event));
            }
            Ok(NativeLine::TooLong) => {
                bail!("Claude CLI stream-json line exceeded {MAX_NATIVE_LINE} bytes")
            }
            Ok(NativeLine::ReadFailed) => bail!("reading Claude CLI stream-json output failed"),
            Ok(NativeLine::End) | Err(mpsc::RecvTimeoutError::Disconnected) => return Ok(None),
            Err(mpsc::RecvTimeoutError::Timeout) => deadline.check()?,
        }
    }
}

struct NativeCompletion {
    assistant: bool,
    result: bool,
    stop: bool,
    subtype: String,
    is_error: bool,
    error_hint: Option<&'static str>,
}

fn run_native(
    receiver: &mpsc::Receiver<NativeLine>,
    cancellation: &AgentCancellation,
    deadline: &mut NativeDeadline,
    on_poll: &mut dyn FnMut(&mut NativeDeadline) -> Result<()>,
) -> Result<NativeCompletion> {
    deadline.phase = "waiting for generation";
    let mut completion = NativeCompletion {
        assistant: false,
        result: false,
        stop: false,
        subtype: String::new(),
        is_error: false,
        error_hint: None,
    };
    while let Some(event) = receive(receiver, cancellation, deadline, on_poll)? {
        match event.get("type").and_then(Value::as_str) {
            Some("assistant") => {
                if event.get("error").is_some() || event.pointer("/message/error").is_some() {
                    completion.is_error = true;
                }
                completion.assistant = true;
            }
            Some("result") => {
                if completion.result {
                    bail!("Claude CLI returned multiple results");
                }
                completion.result = true;
                completion.is_error |= event.get("is_error").and_then(Value::as_bool) == Some(true);
                if completion.is_error {
                    completion.error_hint = event
                        .get("result")
                        .and_then(Value::as_str)
                        .and_then(|message| claude_error_hint(message.as_bytes()));
                }
                completion.subtype = event
                    .get("subtype")
                    .and_then(Value::as_str)
                    .unwrap_or("")
                    .to_owned();
            }
            Some("stream_event")
                if event.pointer("/event/type").and_then(Value::as_str) == Some("message_stop") =>
            {
                completion.stop = true;
            }
            _ => {}
        }
    }
    Ok(completion)
}