fabro/lib/components/fabro-acp/tests/session.rs

use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use agent_client_protocol::schema::StopReason;
use fabro_acp::{
    AcpControlHandle, AcpError, AcpLiveControl, AcpProcessSpec, AcpRunRequest, AcpRunResult,
    run_acp_turn,
};
use fabro_sandbox::test_support::{MockSandbox, MockStdioProcess};
use fabro_sandbox::{LocalSandbox, Sandbox, shell_quote};
use fabro_types::SteeringMessage;
use fabro_util::error::collect_chain;
use tokio::fs::{read_to_string, write};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, DuplexStream};
use tokio::process::Command;
use tokio::sync::Notify;
use tokio::time::{sleep, timeout};
use tokio_util::sync::CancellationToken;

const ACP_TEST_TIMEOUT_MS: u64 = 30_000;

#[allow(
    unused,
    unreachable_pub,
    reason = "integration test imports the shared test fixture source as a private module"
)]
#[path = "../src/test_support.rs"]
mod test_support;

use test_support::fake_acp_agent_script;

#[tokio::test]
async fn stdio_spawn_failure_returns_sandbox_error() {
    const SANDBOX_FAILURE: &str = "ACP backend requires bidirectional stdio; the Daytona sandbox provider does not support it yet";

    let command = AcpProcessSpec::from_command_attr("fake-acp-agent").expect("parse ACP command");
    let mut sandbox = MockSandbox::linux();
    sandbox.stdio_process_error = Some(SANDBOX_FAILURE.to_string());
    let sandbox: Arc<dyn Sandbox> = Arc::new(sandbox);

    let result = run_acp_turn(AcpRunRequest {
        command,
        prompt: "hello".to_string(),
        cwd: "/workspace".to_string(),
        timeout_ms: Some(ACP_TEST_TIMEOUT_MS),
        env: HashMap::new(),
        sandbox,
        cancel_token: CancellationToken::new(),
        on_activity: None,
        live_control: None,
    })
    .await;
    let Err(error) = result else {
        panic!("stdio spawn failure should fail");
    };

    assert!(
        matches!(error, AcpError::Sandbox(_)),
        "expected sandbox error, got {error:?}"
    );
    let chain = collect_chain(&error);
    assert!(
        chain.iter().any(|cause| cause == SANDBOX_FAILURE),
        "cause chain should contain sandbox failure, got: {chain:?}"
    );
}

#[tokio::test]
async fn clean_stdio_exit_after_final_response_completes_turn() {
    let sandbox = MockSandbox::linux();
    sandbox.set_stdio_process(mock_acp_stdio_process("end_turn"));
    let sandbox: Arc<dyn Sandbox> = Arc::new(sandbox);
    let command = AcpProcessSpec::from_command_attr("mock-acp-agent").expect("parse ACP command");

    let result = run_acp_turn(AcpRunRequest {
        command,
        prompt: "hello".to_string(),
        cwd: "/workspace".to_string(),
        timeout_ms: Some(ACP_TEST_TIMEOUT_MS),
        env: HashMap::new(),
        sandbox,
        cancel_token: CancellationToken::new(),
        on_activity: None,
        live_control: None,
    })
    .await
    .expect("clean ACP process exit should not preempt final protocol response");

    assert_eq!(result.text, "hello from acp");
    assert_eq!(result.stop_reason, StopReason::EndTurn);
}

#[tokio::test]
async fn session_lifecycle_initializes_sends_prompt_and_aggregates_text() {
    let tempdir = tempfile::tempdir().expect("create tempdir");
    let script_path = tempdir.path().join("fake_acp_agent.py");
    let record_path = tempdir.path().join("methods.txt");
    write(&script_path, fake_acp_agent_script())
        .await
        .expect("write fake ACP agent");

    let raw_command = format!("python3 {}", shell_quote(&script_path.to_string_lossy()));
    let command = AcpProcessSpec::from_command_attr(&raw_command).expect("parse ACP command");
    let sandbox: Arc<dyn Sandbox> = Arc::new(LocalSandbox::new(tempdir.path().to_path_buf()));

    let result = run_acp_turn(AcpRunRequest {
        command,
        prompt: "hello".to_string(),
        cwd: tempdir.path().to_string_lossy().into_owned(),
        timeout_ms: Some(ACP_TEST_TIMEOUT_MS),
        env: HashMap::from([(
            "ACP_RECORD".to_string(),
            record_path.to_string_lossy().into_owned(),
        )]),
        sandbox,
        cancel_token: CancellationToken::new(),
        on_activity: None,
        live_control: None,
    })
    .await
    .expect("run ACP turn");

    assert_eq!(result.text, "hello from acp");
    assert_eq!(result.stop_reason, StopReason::EndTurn);
    assert_eq!(
        read_to_string(record_path)
            .await
            .expect("read method record"),
        "initialize\nsession/new\nsession/prompt\n"
    );
}

#[tokio::test]
async fn steering_sends_followup_session_prompt_over_acp() {
    let tempdir = tempfile::tempdir().expect("create tempdir");
    let script_path = tempdir.path().join("fake_acp_agent.py");
    let record_path = tempdir.path().join("methods.txt");
    let steer_prompt_path = tempdir.path().join("steer_prompt.json");
    write(&script_path, fake_acp_agent_script())
        .await
        .expect("write fake ACP agent");

    let raw_command = format!("python3 {}", shell_quote(&script_path.to_string_lossy()));
    let command = AcpProcessSpec::from_command_attr(&raw_command).expect("parse ACP command");
    let sandbox: Arc<dyn Sandbox> = Arc::new(LocalSandbox::new(tempdir.path().to_path_buf()));
    let control_handle = AcpControlHandle::new();
    let handle_for_activity = control_handle.clone();
    let queued = Arc::new(AtomicBool::new(false));
    let queued_for_activity = Arc::clone(&queued);

    let result = run_acp_turn(AcpRunRequest {
        command,
        prompt: "hello".to_string(),
        cwd: tempdir.path().to_string_lossy().into_owned(),
        timeout_ms: Some(ACP_TEST_TIMEOUT_MS),
        env: HashMap::from([
            ("ACP_MODE".to_string(), "steer".to_string()),
            (
                "ACP_RECORD".to_string(),
                record_path.to_string_lossy().into_owned(),
            ),
            (
                "ACP_STEER_PROMPT_RECORD".to_string(),
                steer_prompt_path.to_string_lossy().into_owned(),
            ),
        ]),
        sandbox,
        cancel_token: CancellationToken::new(),
        on_activity: Some(Arc::new(move || {
            if !queued_for_activity.swap(true, Ordering::AcqRel) {
                handle_for_activity
                    .enqueue_bounded(SteeringMessage::new("please revise", None), 32);
            }
        })),
        live_control: Some(AcpLiveControl::new(control_handle)),
    })
    .await
    .expect("run ACP turn with steering");

    assert_eq!(result.text, "initial steered:please revise");
    assert_eq!(
        read_to_string(record_path)
            .await
            .expect("read method record"),
        "initialize\nsession/new\nsession/prompt\nsession/prompt\n"
    );
    assert!(
        read_to_string(steer_prompt_path)
            .await
            .expect("read steer prompt")
            .contains("please revise")
    );
}

#[tokio::test]
async fn interrupt_then_steer_sends_cancel_then_followup_session_prompt_over_acp() {
    let tempdir = tempfile::tempdir().expect("create tempdir");
    let script_path = tempdir.path().join("fake_acp_agent.py");
    let record_path = tempdir.path().join("methods.txt");
    let cancel_path = tempdir.path().join("cancel.txt");
    let steer_prompt_path = tempdir.path().join("steer_prompt.json");
    write(&script_path, fake_acp_agent_script())
        .await
        .expect("write fake ACP agent");

    let raw_command = format!("python3 {}", shell_quote(&script_path.to_string_lossy()));
    let command = AcpProcessSpec::from_command_attr(&raw_command).expect("parse ACP command");
    let sandbox: Arc<dyn Sandbox> = Arc::new(LocalSandbox::new(tempdir.path().to_path_buf()));
    let control_handle = AcpControlHandle::new();
    let handle_for_activity = control_handle.clone();
    let queued = Arc::new(AtomicBool::new(false));
    let queued_for_activity = Arc::clone(&queued);

    let result = run_acp_turn(AcpRunRequest {
        command,
        prompt: "hello".to_string(),
        cwd: tempdir.path().to_string_lossy().into_owned(),
        timeout_ms: Some(ACP_TEST_TIMEOUT_MS),
        env: HashMap::from([
            ("ACP_MODE".to_string(), "interrupt_steer".to_string()),
            ("LC_ALL".to_string(), "C".to_string()),
            (
                "ACP_RECORD".to_string(),
                record_path.to_string_lossy().into_owned(),
            ),
            (
                "ACP_CANCEL_RECORD".to_string(),
                cancel_path.to_string_lossy().into_owned(),
            ),
            (
                "ACP_STEER_PROMPT_RECORD".to_string(),
                steer_prompt_path.to_string_lossy().into_owned(),
            ),
        ]),
        sandbox,
        cancel_token: CancellationToken::new(),
        on_activity: Some(Arc::new(move || {
            if !queued_for_activity.swap(true, Ordering::AcqRel) {
                handle_for_activity.interrupt_then_enqueue_bounded(
                    SteeringMessage::new("please revise", None),
                    32,
                );
            }
        })),
        live_control: Some(AcpLiveControl::new(control_handle)),
    })
    .await
    .expect("run ACP turn with interrupt and steering");

    assert_eq!(result.text, "interrupted steered:please revise");
    assert_eq!(
        read_to_string(record_path)
            .await
            .expect("read method record"),
        "initialize\nsession/new\nsession/prompt\nsession/cancel\nsession/prompt\n"
    );
    assert_eq!(
        read_to_string(cancel_path)
            .await
            .expect("read cancel record"),
        "session/cancel\n"
    );
    assert!(
        read_to_string(steer_prompt_path)
            .await
            .expect("read steer prompt")
            .contains("please revise")
    );
}

#[tokio::test]
async fn inline_interrupt_terminates_agent_that_ignores_cancel() {
    let tempdir = tempfile::tempdir().expect("create tempdir");
    let script_path = tempdir.path().join("fake_acp_agent.py");
    let record_path = tempdir.path().join("methods.txt");
    let cancel_path = tempdir.path().join("cancel.txt");
    write(&script_path, fake_acp_agent_script())
        .await
        .expect("write fake ACP agent");

    let raw_command = format!("python3 {}", shell_quote(&script_path.to_string_lossy()));
    let command = AcpProcessSpec::from_command_attr(&raw_command).expect("parse ACP command");
    let sandbox: Arc<dyn Sandbox> = Arc::new(LocalSandbox::new(tempdir.path().to_path_buf()));
    let control_handle = AcpControlHandle::new();
    let handle_for_activity = control_handle.clone();
    let interrupted = Arc::new(AtomicBool::new(false));
    let interrupted_for_activity = Arc::clone(&interrupted);

    let err = run_acp_turn(AcpRunRequest {
        command,
        prompt: "hello".to_string(),
        cwd: tempdir.path().to_string_lossy().into_owned(),
        timeout_ms: Some(ACP_TEST_TIMEOUT_MS),
        env: HashMap::from([
            ("ACP_MODE".to_string(), "ignore_cancel".to_string()),
            ("LC_ALL".to_string(), "C".to_string()),
            (
                "ACP_RECORD".to_string(),
                record_path.to_string_lossy().into_owned(),
            ),
            (
                "ACP_CANCEL_RECORD".to_string(),
                cancel_path.to_string_lossy().into_owned(),
            ),
        ]),
        sandbox,
        cancel_token: CancellationToken::new(),
        on_activity: Some(Arc::new(move || {
            if !interrupted_for_activity.swap(true, Ordering::AcqRel) {
                handle_for_activity.interrupt(None);
            }
        })),
        live_control: Some(AcpLiveControl::new(control_handle)),
    })
    .await
    .expect_err("ignored inline interrupt should terminate as cancelled");

    assert!(matches!(err, AcpError::Cancelled), "{err:?}");
    assert_eq!(
        read_to_string(record_path)
            .await
            .expect("read method record"),
        "initialize\nsession/new\nsession/prompt\nsession/cancel\n"
    );
    assert_eq!(
        read_to_string(cancel_path)
            .await
            .expect("read cancel record"),
        "session/cancel\n"
    );
}

#[tokio::test]
async fn permission_request_selects_allow_always() {
    let tempdir = tempfile::tempdir().expect("create tempdir");
    let permission_path = tempdir.path().join("permission.json");

    let result = run_fake_agent(
        tempdir.path(),
        HashMap::from([
            ("ACP_MODE".to_string(), "permission".to_string()),
            (
                "ACP_PERMISSION".to_string(),
                permission_path.to_string_lossy().into_owned(),
            ),
        ]),
        Some(ACP_TEST_TIMEOUT_MS),
        CancellationToken::new(),
    )
    .await
    .expect("run ACP turn");

    assert_eq!(result.text, "hello from acp");
    let permission = read_to_string(permission_path)
        .await
        .expect("read permission record");
    assert!(permission.contains(r#""outcome":"selected""#));
    assert!(permission.contains(r#""optionId":"always""#));
}

#[tokio::test]
async fn runs_inside_sandbox_and_uses_requested_cwd() {
    let tempdir = tempfile::tempdir().expect("create tempdir");
    let cwd_path = tempdir.path().join("session_new.json");

    let result = run_fake_agent(
        tempdir.path(),
        HashMap::from([
            ("ACP_MODE".to_string(), "write_file".to_string()),
            (
                "ACP_SESSION_NEW_PARAMS".to_string(),
                cwd_path.to_string_lossy().into_owned(),
            ),
        ]),
        Some(ACP_TEST_TIMEOUT_MS),
        CancellationToken::new(),
    )
    .await
    .expect("run ACP turn");

    assert_eq!(result.text, "hello from acp");
    assert_eq!(
        read_to_string(tempdir.path().join("hello.txt"))
            .await
            .expect("read sandbox output file"),
        "hello from sandbox\n"
    );
    assert!(
        read_to_string(cwd_path)
            .await
            .expect("read session/new params")
            .contains(&tempdir.path().to_string_lossy().into_owned())
    );
}

#[tokio::test]
async fn cancellation_sends_session_cancel_and_returns_cancelled() {
    let tempdir = tempfile::tempdir().expect("create tempdir");
    let cancel_path = tempdir.path().join("cancel.txt");
    let tempdir_path = tempdir.path().to_path_buf();
    let cancel_path_for_task = cancel_path.clone();
    let cancel_token = CancellationToken::new();
    let cancel_for_task = cancel_token.clone();
    let prompt_started = Arc::new(Notify::new());
    let prompt_started_for_task = prompt_started.clone();

    let task = tokio::spawn(async move {
        run_fake_agent_with_activity(
            &tempdir_path,
            HashMap::from([
                ("ACP_MODE".to_string(), "cancel".to_string()),
                (
                    "ACP_CANCEL_RECORD".to_string(),
                    cancel_path_for_task.to_string_lossy().into_owned(),
                ),
            ]),
            Some(ACP_TEST_TIMEOUT_MS),
            cancel_for_task,
            Some(Arc::new(move || prompt_started_for_task.notify_one())),
        )
        .await
    });

    timeout(
        Duration::from_millis(ACP_TEST_TIMEOUT_MS),
        prompt_started.notified(),
    )
    .await
    .expect("fake ACP agent should acknowledge session/prompt before cancellation");
    cancel_token.cancel();
    let err = task
        .await
        .expect("join cancellation task")
        .expect_err("cancelled turn should error");

    assert!(matches!(err, AcpError::Cancelled));
    assert_eq!(
        read_to_string(cancel_path)
            .await
            .expect("read cancel record"),
        "session/cancel\n"
    );
}

#[tokio::test]
async fn pre_session_cancellation_returns_cancelled() {
    let tempdir = tempfile::tempdir().expect("create tempdir");
    let cancel_token = CancellationToken::new();
    cancel_token.cancel();

    let err = run_fake_agent(
        tempdir.path(),
        HashMap::from([("ACP_MODE".to_string(), "slow_initialize".to_string())]),
        Some(1_000),
        cancel_token,
    )
    .await
    .expect_err("pre-session cancellation should error");

    assert!(matches!(err, AcpError::Cancelled));
}

#[tokio::test]
async fn successful_turn_terminates_lingering_agent_process() {
    let tempdir = tempfile::tempdir().expect("create tempdir");
    let pid_path = tempdir.path().join("agent.pid");

    let result = run_fake_agent(
        tempdir.path(),
        HashMap::from([
            ("ACP_MODE".to_string(), "linger_after_response".to_string()),
            (
                "ACP_PID_RECORD".to_string(),
                pid_path.to_string_lossy().into_owned(),
            ),
        ]),
        Some(ACP_TEST_TIMEOUT_MS),
        CancellationToken::new(),
    )
    .await
    .expect("run ACP turn");

    sleep(Duration::from_millis(100)).await;
    let pid = read_to_string(&pid_path).await.expect("read agent pid");
    let still_running = process_is_running(pid.trim()).await;
    if still_running {
        let _ = Command::new("kill")
            .arg("-TERM")
            .arg(pid.trim())
            .status()
            .await;
    }

    assert_eq!(result.text, "hello from acp");
    assert!(
        !still_running,
        "successful ACP turn should not leave lingering agent process"
    );
}

#[tokio::test]
async fn refusal_stop_reason_returns_text() {
    let tempdir = tempfile::tempdir().expect("create tempdir");

    let result = run_fake_agent(
        tempdir.path(),
        HashMap::from([("ACP_STOP_REASON".to_string(), "refusal".to_string())]),
        Some(ACP_TEST_TIMEOUT_MS),
        CancellationToken::new(),
    )
    .await
    .expect("run ACP turn");

    assert_eq!(result.text, "hello from acp");
    assert_eq!(result.stop_reason, StopReason::Refusal);
}

#[tokio::test]
async fn max_tokens_stop_reason_returns_partial_text_error() {
    let tempdir = tempfile::tempdir().expect("create tempdir");

    let err = run_fake_agent(
        tempdir.path(),
        HashMap::from([("ACP_STOP_REASON".to_string(), "max_tokens".to_string())]),
        Some(ACP_TEST_TIMEOUT_MS),
        CancellationToken::new(),
    )
    .await
    .expect_err("max_tokens should return stop reason error");

    let AcpError::StopReason { stop_reason, text } = err else {
        panic!("expected stop reason error");
    };
    assert_eq!(stop_reason, "max_tokens");
    assert_eq!(text, "hello from acp");
}

#[tokio::test]
async fn max_turn_requests_stop_reason_returns_partial_text_error() {
    let tempdir = tempfile::tempdir().expect("create tempdir");

    let err = run_fake_agent(
        tempdir.path(),
        HashMap::from([(
            "ACP_STOP_REASON".to_string(),
            "max_turn_requests".to_string(),
        )]),
        Some(ACP_TEST_TIMEOUT_MS),
        CancellationToken::new(),
    )
    .await
    .expect_err("max_turn_requests should return stop reason error");

    let AcpError::StopReason { stop_reason, text } = err else {
        panic!("expected stop reason error");
    };
    assert_eq!(stop_reason, "max_turn_requests");
    assert_eq!(text, "hello from acp");
}

#[tokio::test]
async fn timeout_terminates_process_and_returns_timeout() {
    let tempdir = tempfile::tempdir().expect("create tempdir");

    let err = run_fake_agent(
        tempdir.path(),
        HashMap::from([("ACP_MODE".to_string(), "timeout".to_string())]),
        Some(100),
        CancellationToken::new(),
    )
    .await
    .expect_err("timeout should error");

    assert!(matches!(err, AcpError::TimedOut { .. }));
}

#[tokio::test]
async fn malformed_json_returns_diagnostic_without_raw_stderr_in_message() {
    let tempdir = tempfile::tempdir().expect("create tempdir");

    let err = run_fake_agent(
        tempdir.path(),
        HashMap::from([("ACP_MODE".to_string(), "malformed".to_string())]),
        Some(ACP_TEST_TIMEOUT_MS),
        CancellationToken::new(),
    )
    .await
    .expect_err("malformed JSON should error");

    match err {
        AcpError::Protocol(error) => {
            let message = error.to_string();
            assert!(
                !message.contains("malformed json"),
                "protocol error display should not include raw stderr: {message}"
            );
        }
        AcpError::ProcessExited(exit) => {
            let message = exit.to_string();
            assert!(
                !message.contains("malformed json"),
                "process exit display should not include raw stderr: {message}"
            );
            assert_eq!(
                exit.exec_output_tail
                    .as_ref()
                    .and_then(|tail| tail.stderr.as_deref()),
                Some("malformed json\n")
            );
        }
        other => panic!("expected protocol or process exit error, got {other:?}"),
    }
}

#[tokio::test]
async fn early_exit_returns_process_exit_with_redacted_stderr_tail() {
    let tempdir = tempfile::tempdir().expect("create tempdir");

    let err = run_fake_agent(
        tempdir.path(),
        HashMap::from([("ACP_MODE".to_string(), "early_exit".to_string())]),
        Some(ACP_TEST_TIMEOUT_MS),
        CancellationToken::new(),
    )
    .await
    .expect_err("early exit should error");

    let AcpError::ProcessExited(exit) = err else {
        panic!("expected process exit error");
    };
    let message = exit.to_string();
    assert!(
        message.contains("exit_code=2"),
        "early exit should include exit code in diagnostic: {message}"
    );
    assert!(
        !message.contains("early boom"),
        "early exit display should not include raw stderr: {message}"
    );
    assert_eq!(
        exit.exec_output_tail
            .as_ref()
            .and_then(|tail| tail.stderr.as_deref()),
        Some("early boom\n")
    );
}

async fn run_fake_agent(
    tempdir: &Path,
    env: HashMap<String, String>,
    timeout_ms: Option<u64>,
    cancel_token: CancellationToken,
) -> Result<AcpRunResult, AcpError> {
    run_fake_agent_with_activity(tempdir, env, timeout_ms, cancel_token, None).await
}

async fn run_fake_agent_with_activity(
    tempdir: &Path,
    mut env: HashMap<String, String>,
    timeout_ms: Option<u64>,
    cancel_token: CancellationToken,
    on_activity: Option<Arc<dyn Fn() + Send + Sync>>,
) -> Result<AcpRunResult, AcpError> {
    let script_path = tempdir.join("fake_acp_agent.py");
    write(&script_path, fake_acp_agent_script())
        .await
        .expect("write fake ACP agent");
    let raw_command = format!("python3 {}", shell_quote(&script_path.to_string_lossy()));
    let command = AcpProcessSpec::from_command_attr(&raw_command).expect("parse ACP command");
    let sandbox: Arc<dyn Sandbox> = Arc::new(LocalSandbox::new(tempdir.to_path_buf()));
    env.entry("LC_ALL".to_string())
        .or_insert_with(|| "C".to_string());

    run_acp_turn(AcpRunRequest {
        command,
        prompt: "hello".to_string(),
        cwd: tempdir.to_string_lossy().into_owned(),
        timeout_ms,
        env,
        sandbox,
        cancel_token,
        on_activity,
        live_control: None,
    })
    .await
}

async fn process_is_running(pid: &str) -> bool {
    let Ok(status) = Command::new("kill").arg("-0").arg(pid).status().await else {
        return false;
    };
    if !status.success() {
        return false;
    }

    let Ok(output) = Command::new("ps")
        .args(["-ww", "-o", "stat=", "-p", pid])
        .output()
        .await
    else {
        return true;
    };
    if !output.status.success() {
        return false;
    }
    String::from_utf8_lossy(&output.stdout)
        .chars()
        .find(|ch| !ch.is_whitespace())
        .is_none_or(|state| !matches!(state, 'Z' | 'z'))
}

fn mock_acp_stdio_process(stop_reason: &'static str) -> MockStdioProcess {
    MockStdioProcess::new(move |stdin, mut stdout, _stderr| {
        tokio::spawn(async move {
            let mut lines = BufReader::new(stdin).lines();
            while let Some(line) = lines.next_line().await.expect("read mock ACP stdin") {
                let message: serde_json::Value =
                    serde_json::from_str(&line).expect("parse mock ACP request");
                let method = message
                    .get("method")
                    .and_then(serde_json::Value::as_str)
                    .expect("mock ACP request method");
                let id = message
                    .get("id")
                    .cloned()
                    .unwrap_or(serde_json::Value::Null);

                match method {
                    "initialize" => {
                        write_acp_response(
                            &mut stdout,
                            id,
                            serde_json::json!({
                                "protocolVersion": 1,
                                "agentCapabilities": {},
                            }),
                        )
                        .await;
                    }
                    "session/new" => {
                        write_acp_response(
                            &mut stdout,
                            id,
                            serde_json::json!({ "sessionId": "sess-1" }),
                        )
                        .await;
                    }
                    "session/prompt" => {
                        write_acp_message(
                            &mut stdout,
                            serde_json::json!({
                                "jsonrpc": "2.0",
                                "method": "session/update",
                                "params": {
                                    "sessionId": "sess-1",
                                    "update": {
                                        "sessionUpdate": "agent_message_chunk",
                                        "content": { "type": "text", "text": "hello from acp" }
                                    }
                                }
                            }),
                        )
                        .await;
                        write_acp_response(
                            &mut stdout,
                            id,
                            serde_json::json!({ "stopReason": stop_reason }),
                        )
                        .await;
                        return;
                    }
                    _ => {}
                }
            }
        });
    })
}

async fn write_acp_response(
    stdout: &mut DuplexStream,
    id: serde_json::Value,
    result: serde_json::Value,
) {
    write_acp_message(
        stdout,
        serde_json::json!({
            "jsonrpc": "2.0",
            "id": id,
            "result": result,
        }),
    )
    .await;
}

async fn write_acp_message(stdout: &mut DuplexStream, message: serde_json::Value) {
    let mut line = serde_json::to_vec(&message).expect("serialize mock ACP message");
    line.push(b'\n');
    stdout
        .write_all(&line)
        .await
        .expect("write mock ACP stdout");
    stdout.flush().await.expect("flush mock ACP stdout");
}