rho-coding-agent 1.48.0

use super::*;
use pretty_assertions::assert_eq;
use std::{
    os::unix::fs::PermissionsExt,
    path::Path,
    sync::{Arc, Mutex},
};

use crate::run_artifacts::AttachmentEvent;

fn write_fake_claude(path: &Path, body: &str) {
    // Fresh inode via tempfile rename; avoids overwriting a live text image.
    let dir = path.parent().unwrap_or_else(|| Path::new("."));
    let tmp = dir.join(format!(
        ".claude-install-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0)
    ));
    std::fs::write(&tmp, body).unwrap();
    std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(0o755)).unwrap();
    let _ = std::fs::remove_file(path);
    std::fs::rename(&tmp, path).unwrap();
}

fn fixture(name: &str) -> String {
    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .join("src/claude_runtime/fixtures")
        .join(name);
    std::fs::read_to_string(path).unwrap()
}

fn read_attachment_events(output: &Path) -> Vec<AttachmentEvent> {
    let path = output.with_file_name(crate::subagent::ATTACHMENT_FILE_NAME);
    let body = std::fs::read_to_string(path).unwrap_or_default();
    body.lines()
        .filter(|line| !line.trim().is_empty())
        .map(|line| serde_json::from_str(line).expect("attachment event json"))
        .collect()
}

fn count_terminal_events(events: &[AttachmentEvent]) -> usize {
    events
        .iter()
        .filter(|event| {
            matches!(
                event,
                AttachmentEvent::Completed
                    | AttachmentEvent::Failed(_)
                    | AttachmentEvent::Cancelled
            )
        })
        .count()
}

fn shell_quote(path: &Path) -> String {
    format!("'{}'", path.display().to_string().replace('\'', r"'\''"))
}

fn install_streaming_fake(bin: &Path, ndjson: &str, exit_code: i32) {
    let payload_path = bin.with_extension("payload.ndjson");
    std::fs::write(&payload_path, ndjson).unwrap();
    let script = format!(
        r#"#!/bin/sh
# Emit first. Stream-json sessions keep stdin open until a terminal result (or
# child exit); reading stdin to EOF before emitting would deadlock the drain.
cat {payload}
exit {exit_code}
"#,
        payload = shell_quote(&payload_path),
    );
    write_fake_claude(bin, &script);
}

async fn run_with_fake(
    output: &Path,
    cwd: &Path,
    fake: &Path,
    max_turns: u64,
    permission_mode: PermissionMode,
    cancellation: RunCancellation,
) {
    // Keep rate-limit persistence off the host home directory without
    // mutating process env (unsafe under concurrent tests).
    let rate_limit_dir = tempfile::tempdir().unwrap();
    let rate_limit_state_path = rate_limit_dir.path().join("rate-limits.json");
    run_session(ClaudeSessionRequest {
        system_prompt: system_prompt(),
        identity: claude_identity(),
        tools: vec!["Read".into()],
        inherit_claude_config: false,
        max_turns,
        prompt: "hi".into(),
        output_file: output.to_path_buf(),
        cwd: cwd.to_path_buf(),
        permission_mode,
        cancellation,
        status_tx: None,
        started_status: None,
        parent_messages: None,
        overrides: ClaudeSessionOverrides {
            executable: Some(ClaudeExecutable::from_path(fake)),
            frozen_argv: None,
            auth_status: Some(Ok(logged_in())),
            rate_limit_state_path: Some(rate_limit_state_path),
            live_title: None,
            before_spawn: None,
        },
    })
    .await
    .unwrap();
    // Keep the temp root alive through the session await above.
    drop(rate_limit_dir);
}

async fn run_with_fake_prompt(
    output: &Path,
    cwd: &Path,
    fake: &Path,
    prompt: &str,
    cancellation: RunCancellation,
) {
    let rate_limit_dir = tempfile::tempdir().unwrap();
    let rate_limit_state_path = rate_limit_dir.path().join("rate-limits.json");
    run_session(ClaudeSessionRequest {
        system_prompt: system_prompt(),
        identity: claude_identity(),
        tools: vec!["Read".into()],
        inherit_claude_config: false,
        max_turns: 8,
        prompt: prompt.into(),
        output_file: output.to_path_buf(),
        cwd: cwd.to_path_buf(),
        permission_mode: PermissionMode::Bypass,
        cancellation,
        status_tx: None,
        started_status: None,
        parent_messages: None,
        overrides: ClaudeSessionOverrides {
            executable: Some(ClaudeExecutable::from_path(fake)),
            frozen_argv: None,
            auth_status: Some(Ok(logged_in())),
            rate_limit_state_path: Some(rate_limit_state_path),
            live_title: None,
            before_spawn: None,
        },
    })
    .await
    .unwrap();
    drop(rate_limit_dir);
}

#[tokio::test]
async fn supervised_permission_mode_fails_before_spawn() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    write_fake_claude(&fake, "#!/bin/sh\necho 'should not spawn' >&2\nexit 1\n");
    run_with_fake(
        &output,
        dir.path(),
        &fake,
        8,
        PermissionMode::Supervised,
        RunCancellation::new(),
    )
    .await;
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Error);
    let error = status.error.unwrap_or_default();
    assert!(
        error.contains("Supervised") || error.contains("supervised"),
        "unexpected error: {error}"
    );
}

// Covers: a frozen Bypass argv cannot keep Claude bypassPermissions when the
// current bound mode is Auto; launched permissions must narrow to dontAsk.
// Owner: Claude session launch
#[tokio::test]
async fn frozen_bypass_argv_narrows_to_auto_dont_ask() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    install_streaming_fake(&fake, &fixture("success.ndjson"), 0);
    let captured = Arc::new(Mutex::new(Vec::<String>::new()));
    let captured_for_spawn = Arc::clone(&captured);
    let rate_limit_dir = tempfile::tempdir().unwrap();
    let frozen = vec![
        "-p",
        "--output-format",
        "stream-json",
        "--verbose",
        "--include-partial-messages",
        "--permission-mode",
        "bypassPermissions",
        "--disallowedTools",
        "Task",
        "--setting-sources",
        "project",
        "--strict-mcp-config",
        "--input-format",
        "stream-json",
        "--model",
        "opus",
        "--max-turns",
        "8",
        "--tools",
        "Read",
        "--allowedTools",
        "Read",
    ]
    .into_iter()
    .map(str::to_string)
    .collect();

    run_session(ClaudeSessionRequest {
        system_prompt: system_prompt(),
        identity: claude_identity(),
        tools: vec!["Read".into()],
        inherit_claude_config: false,
        max_turns: 8,
        prompt: "hi".into(),
        output_file: output.clone(),
        cwd: dir.path().to_path_buf(),
        permission_mode: PermissionMode::Auto,
        cancellation: RunCancellation::new(),
        status_tx: None,
        started_status: None,
        parent_messages: None,
        overrides: ClaudeSessionOverrides {
            executable: Some(ClaudeExecutable::from_path(&fake)),
            frozen_argv: Some(frozen),
            auth_status: Some(Ok(logged_in())),
            rate_limit_state_path: Some(rate_limit_dir.path().join("rate-limits.json")),
            live_title: None,
            before_spawn: Some(Box::new(move |command| {
                let args = command
                    .as_std()
                    .get_args()
                    .map(|arg| arg.to_string_lossy().into_owned())
                    .collect();
                *captured_for_spawn.lock().expect("spawn argv lock") = args;
                Ok(())
            })),
        },
    })
    .await
    .unwrap();

    let args = captured.lock().expect("spawn argv lock").clone();
    assert!(
        args.windows(2)
            .any(|pair| pair == ["--permission-mode", "dontAsk"]),
        "Auto must launch dontAsk, got {args:?}"
    );
    assert!(
        !args
            .windows(2)
            .any(|pair| pair == ["--permission-mode", "bypassPermissions"]),
        "frozen Bypass must not survive Auto narrowing: {args:?}"
    );
    assert!(
        args.windows(2)
            .any(|pair| pair == ["--setting-sources", ""]),
        "dontAsk must unload setting sources: {args:?}"
    );
    assert!(
        args.windows(2).any(|pair| pair == ["--tools", "Read"]),
        "bound tools must remain: {args:?}"
    );
    assert!(
        args.windows(2).any(|pair| pair == ["--model", "opus"]),
        "frozen identity model must remain: {args:?}"
    );
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Ok);
    drop(rate_limit_dir);
}

#[tokio::test]
async fn success_stream_and_exit_zero_writes_ok() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    install_streaming_fake(&fake, &fixture("success.ndjson"), 0);
    run_with_fake(
        &output,
        dir.path(),
        &fake,
        8,
        PermissionMode::Bypass,
        RunCancellation::new(),
    )
    .await;
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Ok);
    assert_eq!(status.result.as_deref(), Some("Hello from Claude."));
    assert_eq!(
        status.claude_session_id.as_deref(),
        Some("sess-success-001")
    );
    assert_eq!(status.turns, 1);
    assert!(status.input_tokens.unwrap_or(0) > 0);
    assert_eq!(status.error, None);
    let events = read_attachment_events(&output);
    assert_eq!(
        count_terminal_events(&events),
        1,
        "exactly one terminal attachment"
    );
    assert!(events
        .iter()
        .any(|event| matches!(event, AttachmentEvent::Completed)));
}

#[tokio::test]
async fn live_tool_roundtrip_stream_writes_session_and_tool_events() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    install_streaming_fake(&fake, &fixture("live_tool_roundtrip.ndjson"), 0);
    run_with_fake(
        &output,
        dir.path(),
        &fake,
        8,
        PermissionMode::Bypass,
        RunCancellation::new(),
    )
    .await;
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Ok);
    assert_eq!(status.result.as_deref(), Some("rho-tool-fixture-marker-42"));
    assert_eq!(
        status.claude_session_id.as_deref(),
        Some("22222222-3333-4444-8555-666666666666")
    );
    assert_eq!(status.turns, 2);
    assert_eq!(status.input_tokens, Some(4 + 14452 + 5604));
    assert_eq!(status.output_tokens, Some(102));
    assert_eq!(status.error, None);

    let events = read_attachment_events(&output);
    assert!(
        events.iter().any(|event| matches!(
            event,
            AttachmentEvent::ToolStarted { card, .. }
                if card.header_text().contains("Read") || card.facts.iter().any(|f| f.plain_text().contains("Read")) || card.body.plain_lines().iter().any(|line| line.contains("Read"))
        )),
        "tool started: {events:?}"
    );
    assert!(
        events.iter().any(|event| matches!(
            event,
            AttachmentEvent::ToolFinished { card, .. }
                if card.status == rho_tools::tool_card::ToolStatus::Ok
                    && card.header_text().contains("Read")
        )),
        "tool finished: {events:?}"
    );
    let assistant_text: String = events
        .iter()
        .filter_map(|event| match event {
            AttachmentEvent::AssistantTextDelta(text) => Some(text.as_str()),
            _ => None,
        })
        .collect();
    assert!(
        assistant_text.contains("rho-tool-fixture-marker-42"),
        "assistant text: {assistant_text:?} events: {events:?}"
    );
    assert_eq!(count_terminal_events(&events), 1);
    assert!(events
        .iter()
        .any(|event| matches!(event, AttachmentEvent::Completed)));
}

#[tokio::test]
async fn success_stream_with_nonzero_exit_is_error() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    install_streaming_fake(&fake, &fixture("success.ndjson"), 2);
    run_with_fake(
        &output,
        dir.path(),
        &fake,
        8,
        PermissionMode::Bypass,
        RunCancellation::new(),
    )
    .await;
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Error);
    let error = status.error.unwrap_or_default();
    assert!(
        error.contains("exited with") || error.contains("exit"),
        "unexpected error: {error}"
    );
}

#[tokio::test]
async fn failure_terminal_result_is_error_even_on_exit_zero() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    install_streaming_fake(&fake, &fixture("error_result.ndjson"), 0);
    run_with_fake(
        &output,
        dir.path(),
        &fake,
        8,
        PermissionMode::Bypass,
        RunCancellation::new(),
    )
    .await;
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Error);
    let error = status.error.unwrap_or_default();
    assert!(
        error.contains("hit max turns") || error.contains("error_max_turns"),
        "unexpected error: {error}"
    );
    let events = read_attachment_events(&output);
    assert_eq!(
        count_terminal_events(&events),
        1,
        "exactly one terminal Failed"
    );
    assert!(events.iter().any(|event| {
        matches!(event, AttachmentEvent::Failed(text) if text.contains("hit max turns"))
    }));
}

// Covers: Claude Code safeguard / API errors exit 1 with empty stderr and put
// the reason only on the stream-json result line. Subagent status and Failed
// attachments must carry that text, not a bare exit code.
#[tokio::test]
async fn safeguard_api_error_with_nonzero_exit_surfaces_stream_text() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    install_streaming_fake(&fake, &fixture("safeguard_api_error.ndjson"), 1);
    run_with_fake(
        &output,
        dir.path(),
        &fake,
        8,
        PermissionMode::Bypass,
        RunCancellation::new(),
    )
    .await;
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Error);
    let error = status.error.unwrap_or_default();
    assert!(
        error.contains("safeguards flagged") && error.contains("change your model"),
        "unexpected error: {error}"
    );
    assert!(
        !error.contains("process exited"),
        "stream API error must not be replaced by exit-only text: {error}"
    );
    let events = read_attachment_events(&output);
    assert_eq!(count_terminal_events(&events), 1);
    assert!(events.iter().any(|event| {
        matches!(event, AttachmentEvent::Failed(text) if text.contains("safeguards flagged"))
    }));
}

#[tokio::test]
async fn success_result_with_nonzero_exit_emits_one_failed_not_completed() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    install_streaming_fake(&fake, &fixture("success.ndjson"), 2);
    run_with_fake(
        &output,
        dir.path(),
        &fake,
        8,
        PermissionMode::Bypass,
        RunCancellation::new(),
    )
    .await;
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Error);
    let events = read_attachment_events(&output);
    assert_eq!(count_terminal_events(&events), 1);
    assert!(events
        .iter()
        .any(|event| matches!(event, AttachmentEvent::Failed(_))));
    assert!(!events
        .iter()
        .any(|event| matches!(event, AttachmentEvent::Completed)));
}

#[tokio::test]
async fn protocol_type_error_emits_one_failed_overall() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    install_streaming_fake(
        &fake,
        r#"{"type":"system","subtype":"init","session_id":"sess-err"}
{"type":"error","result":"protocol boom"}
"#,
        0,
    );
    run_with_fake(
        &output,
        dir.path(),
        &fake,
        8,
        PermissionMode::Bypass,
        RunCancellation::new(),
    )
    .await;
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Error);
    let events = read_attachment_events(&output);
    assert_eq!(
        count_terminal_events(&events),
        1,
        "protocol error must not double-Failed with exit finalize"
    );
    assert!(events.iter().any(|event| {
        matches!(event, AttachmentEvent::Failed(text) if text.contains("protocol boom"))
    }));
}

#[tokio::test]
async fn missing_terminal_result_is_error() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    install_streaming_fake(
        &fake,
        r#"{"type":"system","subtype":"init","session_id":"sess-x"}
{"type":"assistant","session_id":"sess-x","message":{"id":"m1","role":"assistant","content":[{"type":"text","text":"hi"}]}}
"#,
        0,
    );
    run_with_fake(
        &output,
        dir.path(),
        &fake,
        8,
        PermissionMode::Bypass,
        RunCancellation::new(),
    )
    .await;
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Error);
    let error = status.error.unwrap_or_default();
    assert!(
        error.contains("without a terminal result"),
        "unexpected error: {error}"
    );
}

#[tokio::test]
async fn invalid_terminal_fields_are_error() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    install_streaming_fake(
        &fake,
        r#"{"type":"result","result":"maybe","session_id":"sess-invalid"}
"#,
        0,
    );
    run_with_fake(
        &output,
        dir.path(),
        &fake,
        8,
        PermissionMode::Bypass,
        RunCancellation::new(),
    )
    .await;
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Error);
    let error = status.error.unwrap_or_default();
    assert!(
        error.contains("missing subtype") || error.contains("invalid"),
        "unexpected error: {error}"
    );
}

#[tokio::test]
async fn invalid_utf8_stdout_fails_run() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    let payload = dir.path().join("bad.bin");
    std::fs::write(&payload, [0xff, b'\n']).unwrap();
    let script = format!(
        r#"#!/bin/sh
cat {}
exit 0
"#,
        shell_quote(&payload)
    );
    write_fake_claude(&fake, &script);
    run_with_fake(
        &output,
        dir.path(),
        &fake,
        8,
        PermissionMode::Bypass,
        RunCancellation::new(),
    )
    .await;
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Error);
    let error = status.error.unwrap_or_default();
    assert!(
        error.contains("UTF-8") || error.contains("utf-8") || error.contains("malformed"),
        "unexpected error: {error}"
    );
}

#[tokio::test]
async fn oversize_line_fails_run() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    let payload = dir.path().join("big.ndjson");
    let mut bytes = vec![b'a'; crate::claude_runtime::line_decoder::MAX_NDJSON_LINE_BYTES + 8];
    bytes.push(b'\n');
    std::fs::write(&payload, bytes).unwrap();
    let script = format!(
        r#"#!/bin/sh
cat {}
exit 0
"#,
        shell_quote(&payload)
    );
    write_fake_claude(&fake, &script);
    run_with_fake(
        &output,
        dir.path(),
        &fake,
        8,
        PermissionMode::Bypass,
        RunCancellation::new(),
    )
    .await;
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Error);
    let error = status.error.unwrap_or_default();
    assert!(
        error.contains("oversize") || error.contains("exceeds"),
        "unexpected error: {error}"
    );
}

#[tokio::test]
async fn max_turns_unsupported_stderr_is_diagnosed() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    write_fake_claude(
        &fake,
        r#"#!/bin/sh
echo "error: unknown option '--max-turns'" >&2
exit 2
"#,
    );
    run_with_fake(
        &output,
        dir.path(),
        &fake,
        8,
        PermissionMode::Bypass,
        RunCancellation::new(),
    )
    .await;
    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Error);
    let error = status.error.unwrap_or_default();
    assert!(
        error.contains("max-turns") || error.contains("--max-turns"),
        "unexpected error: {error}"
    );
}

/// Child floods stdout before reading stdin. Old ordering awaited the full
/// prompt write before draining stdout, so a filled pipe deadlocked.
#[tokio::test]
async fn concurrent_stdin_write_drains_high_volume_stdout() {
    let dir = tempfile::tempdir().unwrap();
    let output = dir.path().join("result.json");
    let fake = dir.path().join("claude");
    let payload = dir.path().join("flood.ndjson");
    // Enough bulk to fill typical OS pipe buffers several times over. Prefer
    // keep_alive control frames so the mapper does no journal/status work per
    // line: this test owns the drain/deadlock failure mode, not presentation.
    // Assistant flood lines used to enqueue thousands of attachment writes and
    // race the 5s finish-join budget under macOS CI load (false deadlock).
    let pad = "p".repeat(240);
    let line = format!(r#"{{"type":"keep_alive","pad":"{pad}"}}"#);
    let mut body = String::with_capacity(256 * 1024 + 256);
    while body.len() < 256 * 1024 {
        body.push_str(&line);
        body.push('\n');
    }
    body.push_str(
        r#"{"type":"result","subtype":"success","is_error":false,"result":"drained-while-writing","session_id":"s","num_turns":1,"usage":{"input_tokens":1,"output_tokens":1}}"#,
    );
    body.push('\n');
    std::fs::write(&payload, body).unwrap();

    // Emit a large stdout payload first; only then consume stdin. The parent
    // must drain while still writing the prompt or the pipes wedge.
    let script = format!(
        r#"#!/bin/sh
cat {payload}
cat >/dev/null
exit 0
"#,
        payload = shell_quote(&payload)
    );
    write_fake_claude(&fake, &script);

    // Large prompt so stdin write itself needs multiple pipe buffers.
    let prompt = "P".repeat(256 * 1024);
    // Stay above RunArtifactSink's finish-join budget so slow CI disks cannot
    // look like a pipe deadlock.
    tokio::time::timeout(Duration::from_secs(30), async {
        run_with_fake_prompt(&output, dir.path(), &fake, &prompt, RunCancellation::new()).await;
    })
    .await
    .expect("draining stdout while writing stdin must not deadlock");

    let status = subagent::read_status(&output).expect("status");
    assert_eq!(status.state, RunState::Ok);
    assert_eq!(status.result.as_deref(), Some("drained-while-writing"));
}