use std::path::{Path, PathBuf};

use eyre::{Result, eyre};
use futures_util::{SinkExt, StreamExt};
use serde_json::{Value, json};
use tokio::{net::TcpListener, time::timeout};
use tokio_tungstenite::{WebSocketStream, accept_async, tungstenite::Message};

use crate::{
    AgentHandle, Nanocodex, NanocodexError, Prompt, Responses, ResponsesError, Thinking, Tools,
};

#[tokio::test]
async fn follow_on_prompts_reuse_the_session_socket_and_context() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        let warmup = next_json(&mut socket).await?;
        assert_warmup(&warmup);
        assert_eq!(warmup["input"][1]["content"][0]["text"], "custom prompt");
        send_warmup(&mut socket, "resp-warmup").await?;

        let first = next_json(&mut socket).await?;
        assert_eq!(first["previous_response_id"], "resp-warmup");
        send_final(&mut socket, "resp-first").await?;

        let follow_on = next_json(&mut socket).await?;
        assert_eq!(follow_on["previous_response_id"], "resp-first");
        assert_eq!(follow_on["input"].as_array().map(Vec::len), Some(1));
        assert_eq!(follow_on["input"][0]["role"], "user");
        assert_eq!(follow_on["input"][0]["content"][0]["text"], "second prompt");
        send_final(&mut socket, "resp-second").await
    });

    let workspace = temporary_workspace("follow-on")?;
    let responses = Responses::builder().websocket_url(endpoint).build();
    let (agent, mut events) = Nanocodex::builder("test-key")
        .instructions("custom prompt")
        .thinking(Thinking::Low)
        .workspace(&workspace)
        .responses(responses)
        .session_id("model-test")
        .build()?;

    let first = agent.prompt(Prompt::new("first prompt")).await?;
    assert_eq!(first.result().await?.final_message, "done");
    let second = agent.prompt(Prompt::new("second prompt")).await?;
    assert_eq!(second.result().await?.final_message, "done");
    drop(agent);

    let mut completed = Vec::new();
    while let Some(event) = events.recv().await {
        if event.kind == nanocodex_core::AgentEventKind::RunCompleted {
            completed.push(event.decode_payload::<Value>()?);
        }
    }
    assert_eq!(completed.len(), 2);
    assert_eq!(completed[0]["connection_attempts"], 1);
    assert_eq!(completed[0]["response_attempts"], 2);
    assert_eq!(completed[1]["connection_attempts"], 0);
    assert_eq!(completed[1]["response_attempts"], 1);

    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn assistant_events_preserve_commentary_and_final_answer_phases() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        assert_warmup(&next_json(&mut socket).await?);
        send_warmup(&mut socket, "resp-warmup").await?;

        let initial = next_json(&mut socket).await?;
        assert_eq!(initial["previous_response_id"], "resp-warmup");
        let commentary = send_assistant_output(
            &mut socket,
            0,
            "msg-commentary",
            "commentary",
            "I’ll verify.",
        )
        .await?;
        send_json(
            &mut socket,
            completed_response(
                "resp-commentary",
                &[
                    commentary,
                    json!({
                        "id": "call-item",
                        "type": "custom_tool_call",
                        "call_id": "call-exec",
                        "name": "exec",
                        "input": "text(\"observed\");"
                    }),
                ],
            ),
        )
        .await?;

        let continuation = next_json(&mut socket).await?;
        assert_eq!(continuation["previous_response_id"], "resp-commentary");
        let final_answer =
            send_assistant_output(&mut socket, 0, "msg-final", "final_answer", "Done.").await?;
        send_json(
            &mut socket,
            completed_response("resp-final", &[final_answer]),
        )
        .await
    });

    let workspace = temporary_workspace("assistant-phases")?;
    let responses = Responses::builder().websocket_url(endpoint).build();
    let (agent, mut events) = Nanocodex::builder("test-key")
        .thinking(Thinking::Low)
        .workspace(&workspace)
        .responses(responses)
        .session_id("model-test")
        .build()?;
    let turn = agent.prompt("check the live state").await?;
    assert_eq!(turn.result().await?.final_message, "Done.");
    drop(agent);

    let mut deltas = Vec::new();
    let mut messages = Vec::new();
    let mut timeline = Vec::new();
    while let Some(event) = events.recv().await {
        match event.kind {
            nanocodex_core::AgentEventKind::AssistantDelta => {
                deltas.push(event.decode_payload::<Value>()?);
            }
            nanocodex_core::AgentEventKind::AssistantMessage => {
                let message = event.decode_payload::<Value>()?;
                timeline.push(message["phase"].clone());
                messages.push(message);
            }
            nanocodex_core::AgentEventKind::ToolCall => {
                timeline.push(json!("tool.call"));
            }
            nanocodex_core::AgentEventKind::ToolResult => {
                timeline.push(json!("tool.result"));
            }
            _ => {}
        }
    }
    assert_assistant_phase_events(&deltas, &messages, &timeline);

    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

fn assert_assistant_phase_events(deltas: &[Value], messages: &[Value], timeline: &[Value]) {
    let expected_messages = [
        json!({
            "model_call_index": 1,
            "item_id": "msg-commentary",
            "phase": "commentary",
            "text": "I’ll verify."
        }),
        json!({
            "model_call_index": 2,
            "item_id": "msg-final",
            "phase": "final_answer",
            "text": "Done."
        }),
    ];
    assert_eq!(deltas, expected_messages);
    assert_eq!(messages, expected_messages);
    assert_eq!(
        timeline,
        [
            json!("commentary"),
            json!("tool.call"),
            json!("tool.result"),
            json!("final_answer")
        ]
    );
}

#[tokio::test]
async fn steering_is_bounded_fifo_and_joins_at_the_next_model_boundary() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let (first_seen, first_seen_rx) = tokio::sync::oneshot::channel();
    let (release_first, release_first_rx) = tokio::sync::oneshot::channel();
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        assert_warmup(&next_json(&mut socket).await?);
        send_warmup(&mut socket, "resp-warmup").await?;

        let first = next_json(&mut socket).await?;
        assert_eq!(first["previous_response_id"], "resp-warmup");
        assert_eq!(first["input"][1]["content"][0]["text"], "initial task");
        first_seen
            .send(())
            .map_err(|()| eyre!("first-request signal receiver dropped"))?;
        release_first_rx
            .await
            .map_err(|_| eyre!("first-request release sender dropped"))?;
        send_final(&mut socket, "resp-first").await?;

        let steered = next_json(&mut socket).await?;
        assert_eq!(steered["previous_response_id"], "resp-first");
        assert_eq!(steered["input"].as_array().map(Vec::len), Some(8));
        for index in 0..8 {
            assert_eq!(steered["input"][index]["role"], "user");
            assert_eq!(
                steered["input"][index]["content"][0]["text"],
                format!("constraint {index}")
            );
        }
        send_final(&mut socket, "resp-steered").await
    });

    let workspace = temporary_workspace("steer")?;
    let responses = Responses::builder().websocket_url(endpoint).build();
    let (agent, mut events) = Nanocodex::builder("test-key")
        .thinking(Thinking::Low)
        .workspace(&workspace)
        .responses(responses)
        .session_id("model-test")
        .build()?;
    let turn = agent.prompt(Prompt::new("initial task")).await?;
    first_seen_rx
        .await
        .map_err(|_| eyre!("first request was not observed"))?;
    for index in 0..8 {
        turn.steer(format!("constraint {index}")).await?;
    }
    let overflow = turn.steer("constraint 8").await.unwrap_err();
    assert!(matches!(overflow, NanocodexError::SteerQueueFull));
    release_first
        .send(())
        .map_err(|()| eyre!("server release receiver dropped"))?;
    assert_eq!(turn.result().await?.final_message, "done");
    drop(agent);

    let mut steered_events = 0;
    let mut terminal = None;
    while let Some(event) = events.recv().await {
        match event.kind {
            nanocodex_core::AgentEventKind::RunSteered => {
                steered_events += 1;
                let payload = event.decode_payload::<Value>()?;
                assert_eq!(payload["steer_index"], steered_events);
                assert_eq!(payload["instruction_bytes"], "constraint 0".len());
            }
            nanocodex_core::AgentEventKind::RunCompleted => {
                terminal = Some(event.decode_payload::<Value>()?);
            }
            _ => {}
        }
    }
    assert_eq!(steered_events, 8);
    assert_eq!(
        terminal.as_ref().map(|payload| &payload["steers"]),
        Some(&json!(8))
    );

    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn steering_during_a_tool_call_joins_after_the_tool_result() -> Result<()> {
    let workspace = temporary_workspace("steer-tool")?;
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        assert_warmup(&next_json(&mut socket).await?);
        send_warmup(&mut socket, "resp-warmup").await?;

        let initial = next_json(&mut socket).await?;
        assert_eq!(initial["previous_response_id"], "resp-warmup");
        send_json(
            &mut socket,
            completed_response(
                "resp-tool",
                &[json!({
                    "type": "custom_tool_call",
                    "call_id": "call-exec",
                    "name": "exec",
                    "input": "const result = await tools.exec_command({cmd: \"printf started > tool-started; while [ ! -f release-tool ]; do sleep 0.01; done; printf shit\"}); text(result.output);"
                })],
            ),
        )
        .await?;

        let steered = next_json(&mut socket).await?;
        assert_eq!(steered["previous_response_id"], "resp-tool");
        let input = steered["input"]
            .as_array()
            .ok_or_else(|| eyre!("steered request input was not an array"))?;
        assert_eq!(input.len(), 2);
        assert_eq!(input[0]["type"], "custom_tool_call_output");
        assert_eq!(input[0]["call_id"], "call-exec");
        assert!(input[0].to_string().contains("shit"));
        assert_eq!(input[1]["role"], "user");
        assert_eq!(input[1]["content"][0]["text"], "print shat instead");
        send_final(&mut socket, "resp-steered").await
    });

    let responses = Responses::builder().websocket_url(endpoint).build();
    let (agent, mut events) = Nanocodex::builder("test-key")
        .thinking(Thinking::Low)
        .workspace(&workspace)
        .responses(responses)
        .session_id("model-test")
        .build()?;
    let turn = agent.prompt("print shit a lot of times").await?;
    timeout(std::time::Duration::from_secs(5), async {
        while !workspace.join("tool-started").exists() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .map_err(|_| eyre!("tool process did not start"))?;

    turn.steer("print shat instead").await?;
    assert!(!workspace.join("release-tool").exists());
    std::fs::write(workspace.join("release-tool"), [])?;
    assert_eq!(turn.result().await?.final_message, "done");
    drop(agent);

    let mut saw_steer = false;
    while let Some(event) = events.recv().await {
        saw_steer |= event.kind == nanocodex_core::AgentEventKind::RunSteered;
    }
    assert!(saw_steer);
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn cancellation_retains_interrupted_prompt_and_resumes_from_the_abort_boundary() -> Result<()>
{
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let (second_seen, second_seen_rx) = tokio::sync::oneshot::channel();
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut first_socket = accept_async(stream).await?;
        assert_warmup(&next_json(&mut first_socket).await?);
        send_warmup(&mut first_socket, "resp-warmup").await?;

        let first = next_json(&mut first_socket).await?;
        assert_eq!(first["previous_response_id"], "resp-warmup");
        send_final(&mut first_socket, "resp-first").await?;

        let cancelled = next_json(&mut first_socket).await?;
        assert_eq!(cancelled["previous_response_id"], "resp-first");
        assert_eq!(cancelled["input"][0]["content"][0]["text"], "cancel me");
        second_seen
            .send(())
            .map_err(|()| eyre!("second-request signal receiver dropped"))?;
        send_json(
            &mut first_socket,
            json!({
                "type": "response.output_text.delta",
                "delta": "partial text that must not enter history"
            }),
        )
        .await?;

        let (stream, _) = listener.accept().await?;
        let mut replacement = accept_async(stream).await?;
        let queued = next_json(&mut replacement).await?;
        assert_interrupted_replay(&queued);
        send_final(&mut replacement, "resp-follow-up").await
    });

    let workspace = temporary_workspace("cancel-turn")?;
    let responses = Responses::builder().websocket_url(endpoint).build();
    let (agent, mut events) = Nanocodex::builder("test-key")
        .thinking(Thinking::Low)
        .workspace(&workspace)
        .responses(responses)
        .session_id("model-test")
        .build()?;

    let first = agent.prompt(Prompt::new("first prompt")).await?;
    assert_eq!(first.result().await?.final_message, "done");

    let cancelled = agent.prompt("cancel me").await?;
    second_seen_rx
        .await
        .map_err(|_| eyre!("second request was not observed"))?;
    let queued = agent.prompt("cancel before running").await?;
    let queued_control = queued.control();
    let follow_up = agent.prompt("run after cancellations").await?;

    assert!(matches!(
        queued.steer("wrong target").await,
        Err(NanocodexError::TurnNotSteerable)
    ));
    queued.cancel().await?;
    assert!(matches!(
        queued_control.cancel().await,
        Err(NanocodexError::TurnNotCancellable)
    ));

    let cancellation = cancelled.control();
    cancellation.cancel().await?;
    assert!(matches!(
        cancelled.result().await,
        Err(NanocodexError::TurnCancelled)
    ));
    assert!(matches!(
        queued.result().await,
        Err(NanocodexError::TurnCancelled)
    ));
    assert!(matches!(
        cancellation.cancel().await,
        Err(NanocodexError::TurnNotCancellable)
    ));
    assert_eq!(follow_up.result().await?.final_message, "done");
    drop((queued_control, cancellation, agent));

    let mut terminal_statuses = Vec::new();
    while let Some(event) = events.recv().await {
        match event.kind {
            nanocodex_core::AgentEventKind::RunCompleted
            | nanocodex_core::AgentEventKind::RunFailed => {
                let payload = event.decode_payload::<Value>()?;
                terminal_statuses.push(payload["status"].as_str().unwrap_or_default().to_owned());
            }
            _ => {}
        }
    }
    assert_eq!(
        terminal_statuses,
        ["completed", "cancelled", "cancelled", "completed"]
    );

    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

fn assert_interrupted_replay(request: &Value) {
    assert!(request.get("previous_response_id").is_none());
    assert_eq!(request["input"].as_array().map(Vec::len), Some(8));
    assert_eq!(request["input"][0]["type"], "additional_tools");
    assert_eq!(request["input"][1]["role"], "developer");
    assert_eq!(request["input"][2]["role"], "user");
    assert_eq!(request["input"][3]["content"][0]["text"], "first prompt");
    assert_eq!(request["input"][4]["content"][0]["text"], "done");
    assert_eq!(request["input"][5]["content"][0]["text"], "cancel me");
    assert!(
        request["input"][6]["content"][0]["text"]
            .as_str()
            .is_some_and(|text| text.contains("<turn_aborted>"))
    );
    assert_eq!(
        request["input"][7]["content"][0]["text"],
        "run after cancellations"
    );
    assert!(
        !request
            .to_string()
            .contains("partial text that must not enter history")
    );
}

#[tokio::test]
async fn cancellation_pairs_an_active_tool_call_before_resuming() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut first = accept_async(stream).await?;
        assert_warmup(&next_json(&mut first).await?);
        send_warmup(&mut first, "resp-warmup").await?;

        let generation = next_json(&mut first).await?;
        assert_eq!(generation["previous_response_id"], "resp-warmup");
        send_json(
            &mut first,
            completed_response(
                "resp-tool",
                &[json!({
                    "type": "custom_tool_call",
                    "call_id": "call-exec",
                    "name": "exec",
                    "input": "const result = await tools.exec_command({cmd: \"printf started > tool-started; sleep 30\"}); text(result.output);"
                })],
            ),
        )
        .await?;

        let (stream, _) = listener.accept().await?;
        let mut replacement = accept_async(stream).await?;
        let resumed = next_json(&mut replacement).await?;
        assert!(resumed.get("previous_response_id").is_none());
        assert_eq!(resumed["input"].as_array().map(Vec::len), Some(8));
        assert_eq!(resumed["input"][3]["content"][0]["text"], "run a long tool");
        assert_eq!(resumed["input"][4]["type"], "custom_tool_call");
        assert_eq!(resumed["input"][4]["call_id"], "call-exec");
        assert_eq!(resumed["input"][5]["type"], "custom_tool_call_output");
        assert_eq!(resumed["input"][5]["call_id"], "call-exec");
        assert!(resumed["input"][5].to_string().contains("aborted by user"));
        assert!(
            resumed["input"][6]["content"][0]["text"]
                .as_str()
                .is_some_and(|text| text.contains("<turn_aborted>"))
        );
        assert_eq!(resumed["input"][7]["content"][0]["text"], "continue");
        send_final(&mut replacement, "resp-follow-up").await
    });

    let workspace = temporary_workspace("cancel-tool")?;
    let responses = Responses::builder().websocket_url(endpoint).build();
    let (agent, mut events) = Nanocodex::builder("test-key")
        .thinking(Thinking::Low)
        .workspace(&workspace)
        .responses(responses)
        .session_id("model-test")
        .build()?;

    let interrupted = agent.prompt("run a long tool").await?;
    loop {
        let event = events
            .recv()
            .await
            .ok_or_else(|| eyre!("event stream closed before the tool call"))?;
        if event.kind == nanocodex_core::AgentEventKind::ToolCall {
            break;
        }
    }
    timeout(std::time::Duration::from_secs(5), async {
        while !workspace.join("tool-started").exists() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .map_err(|_| eyre!("tool process did not start"))?;

    interrupted.cancel().await?;
    assert!(matches!(
        interrupted.result().await,
        Err(NanocodexError::TurnCancelled)
    ));
    assert_eq!(
        agent
            .prompt("continue")
            .await?
            .result()
            .await?
            .final_message,
        "done"
    );
    drop(agent);

    let mut saw_cancelled_tool = false;
    while let Some(event) = events.recv().await {
        if event.kind == nanocodex_core::AgentEventKind::ToolResult {
            let payload = event.decode_payload::<Value>()?;
            saw_cancelled_tool |= payload["call_id"] == "call-exec"
                && payload["status"] == "cancelled"
                && payload.to_string().contains("aborted by user");
        }
    }
    assert!(saw_cancelled_tool);
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn stored_response_local_code_mode_round_trip() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        let warmup = next_json(&mut socket).await?;
        assert_warmup(&warmup);
        send_json(
            &mut socket,
            json!({
                "type": "response.metadata",
                "headers": { "x-codex-turn-state": "sticky-test" }
            }),
        )
        .await?;
        send_warmup(&mut socket, "resp-warmup").await?;

        let generation = next_json(&mut socket).await?;
        assert_eq!(generation["previous_response_id"], "resp-warmup");
        assert_eq!(generation["store"], true);
        assert!(generation.get("generate").is_none());
        assert_eq!(generation["input"].as_array().map(Vec::len), Some(2));
        assert_eq!(
            generation["client_metadata"]["x-codex-turn-state"],
            "sticky-test"
        );
        send_json(
            &mut socket,
            completed_response(
                "resp-tool",
                &[json!({
                    "id": "item-exec",
                    "type": "custom_tool_call",
                    "call_id": "call-exec",
                    "name": "exec",
                    "input": "const result = await tools.exec_command({cmd: \"printf hello\"}); text(result.output);"
                })],
            ),
        )
        .await?;

        let continuation = next_json(&mut socket).await?;
        assert_eq!(continuation["previous_response_id"], "resp-tool");
        assert_eq!(continuation["input"].as_array().map(Vec::len), Some(1));
        assert_eq!(continuation["input"][0]["type"], "custom_tool_call_output");
        assert_eq!(continuation["input"][0]["call_id"], "call-exec");
        assert!(continuation["input"][0].get("success").is_none());
        assert!(
            continuation["input"][0]["output"]
                .as_array()
                .is_some_and(|content| content.iter().any(|item| {
                    item["text"]
                        .as_str()
                        .is_some_and(|text| text.contains("hello"))
                }))
        );
        send_final(&mut socket, "resp-final").await
    });

    let workspace = temporary_workspace("code-mode")?;
    let output = run_model(&endpoint, &workspace, "run a shell command").await?;
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    assert!(output.contains("\"tool\":\"exec\""));
    assert!(output.contains("\"tool\":\"exec_command\""));
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn unsupported_direct_tools_return_failed_results_to_the_model() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        assert_warmup(&next_json(&mut socket).await?);
        send_warmup(&mut socket, "resp-warmup").await?;

        let generation = next_json(&mut socket).await?;
        assert_eq!(generation["previous_response_id"], "resp-warmup");
        send_json(
            &mut socket,
            completed_response(
                "resp-unsupported",
                &[
                    json!({
                        "type": "custom_tool_call",
                        "call_id": "call-custom",
                        "name": "missing_custom",
                        "input": "raw input"
                    }),
                    json!({
                        "type": "function_call",
                        "call_id": "call-function",
                        "namespace": "example::",
                        "name": "missing_function",
                        "arguments": "not json"
                    }),
                ],
            ),
        )
        .await?;

        let continuation = next_json(&mut socket).await?;
        assert_eq!(continuation["previous_response_id"], "resp-unsupported");
        let input = continuation["input"]
            .as_array()
            .ok_or_else(|| eyre!("continuation input was not an array"))?;
        assert_eq!(
            input,
            &[
                json!({
                    "type": "custom_tool_call_output",
                    "call_id": "call-custom",
                    "output": "unsupported custom tool call: missing_custom"
                }),
                json!({
                    "type": "function_call_output",
                    "call_id": "call-function",
                    "output": "unsupported call: example::missing_function"
                }),
            ]
        );
        send_final(&mut socket, "resp-final").await
    });

    let workspace = temporary_workspace("unsupported-tools")?;
    let output = run_model(&endpoint, &workspace, "recover from unsupported tools").await?;
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    assert_eq!(
        output.matches(r#""status":"failed""#).count(),
        2,
        "{output}"
    );
    assert!(output.contains("\"tool_calls\":2"));
    assert!(output.contains("\"run.completed\""));
    assert!(!output.contains("\"run.failed\""));
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn code_mode_notify_adds_a_named_exec_output_to_the_next_request() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        assert_warmup(&next_json(&mut socket).await?);
        send_warmup(&mut socket, "resp-warmup").await?;

        let generation = next_json(&mut socket).await?;
        assert_eq!(generation["previous_response_id"], "resp-warmup");
        send_json(
            &mut socket,
            completed_response(
                "resp-notify",
                &[json!({
                    "type": "custom_tool_call",
                    "call_id": "call-exec",
                    "name": "exec",
                    "input": "notify({phase: \"working\"}); text(\"done\");"
                })],
            ),
        )
        .await?;

        let continuation = next_json(&mut socket).await?;
        assert_eq!(continuation["previous_response_id"], "resp-notify");
        let input = continuation["input"]
            .as_array()
            .ok_or_else(|| eyre!("continuation input was not an array"))?;
        assert_eq!(input.len(), 2);
        assert_eq!(input[0]["type"], "custom_tool_call_output");
        assert_eq!(input[0]["call_id"], "call-exec");
        assert!(input[0].get("name").is_none());
        assert!(input[0].to_string().contains("done"));
        assert_eq!(input[1]["type"], "custom_tool_call_output");
        assert_eq!(input[1]["call_id"], "call-exec");
        assert_eq!(input[1]["name"], "exec");
        assert_eq!(input[1]["output"], r#"{"phase":"working"}"#);
        assert!(input[1].get("success").is_none());
        send_final(&mut socket, "resp-final").await
    });

    let workspace = temporary_workspace("code-mode-notify")?;
    run_model(&endpoint, &workspace, "send a progress notification").await?;
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn prepares_images_and_stops_on_invalid_image_requests() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        assert_warmup(&next_json(&mut socket).await?);
        send_warmup(&mut socket, "resp-warmup").await?;

        let generation = next_json(&mut socket).await?;
        assert_eq!(generation["previous_response_id"], "resp-warmup");
        send_json(
            &mut socket,
            completed_response(
                "resp-image",
                &[json!({
                    "type": "custom_tool_call",
                    "call_id": "call-image",
                    "name": "exec",
                    "input": "image(\"data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=\", \"original\");"
                })],
            ),
        )
        .await?;

        let continuation = next_json(&mut socket).await?;
        let output = continuation["input"][0]["output"]
            .as_array()
            .ok_or_else(|| eyre!("image tool output was not content"))?;
        let image = output
            .iter()
            .find(|item| item["type"] == "input_image")
            .ok_or_else(|| eyre!("prepared image was missing"))?;
        assert!(
            image["image_url"]
                .as_str()
                .is_some_and(|url| url.starts_with("data:image/png;base64,"))
        );
        assert!(image.get("detail").is_none());

        send_json(
            &mut socket,
            json!({
                "type": "response.failed",
                "response": {
                    "id": "resp-invalid-image",
                    "status": "failed",
                    "error": {
                        "code": "invalid_image",
                        "message": "The image data you provided does not represent a valid image"
                    }
                }
            }),
        )
        .await?;

        Ok::<(), eyre::Report>(())
    });

    let workspace = temporary_workspace("images")?;
    let error = run_model(&endpoint, &workspace, "inspect images")
        .await
        .expect_err("invalid tool image should fail the turn");
    let error = error
        .downcast_ref::<NanocodexError>()
        .ok_or_else(|| eyre!("invalid image returned the wrong error type"))?;
    assert!(matches!(
        error.responses_error(),
        Some(ResponsesError::InvalidImageRequest { .. })
    ));
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn yielded_exec_cell_continues_through_direct_wait_tool() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        assert_warmup(&next_json(&mut socket).await?);
        send_warmup(&mut socket, "resp-warmup").await?;

        let generation = next_json(&mut socket).await?;
        assert_eq!(generation["previous_response_id"], "resp-warmup");
        send_json(
            &mut socket,
            completed_response(
                "resp-exec",
                &[json!({
                    "type": "custom_tool_call",
                    "call_id": "call-exec",
                    "name": "exec",
                    "input": "text(\"before\"); await yield_control(); const result = await tools.exec_command({cmd: \"printf after\", login: false}); text(result.output);"
                })],
            ),
        )
        .await?;

        let yielded = next_json(&mut socket).await?;
        assert_eq!(yielded["previous_response_id"], "resp-exec");
        assert_eq!(yielded["input"][0]["type"], "custom_tool_call_output");
        assert!(
            yielded
                .to_string()
                .contains("Script running with cell ID 1")
        );
        send_json(
            &mut socket,
            completed_response(
                "resp-wait",
                &[json!({
                    "type": "function_call",
                    "call_id": "call-wait",
                    "name": "wait",
                    "arguments": "{\"cell_id\":\"1\",\"yield_time_ms\":1000}"
                })],
            ),
        )
        .await?;

        let completed = next_json(&mut socket).await?;
        assert_eq!(completed["previous_response_id"], "resp-wait");
        assert_eq!(completed["input"][0]["type"], "function_call_output");
        assert_eq!(completed["input"][0]["call_id"], "call-wait");
        assert!(completed.to_string().contains("after"));
        send_final(&mut socket, "resp-final").await
    });

    let workspace = temporary_workspace("code-mode-wait")?;
    let output = run_model(&endpoint, &workspace, "yield and wait").await?;
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    assert!(output.contains("\"tool\":\"wait\""));
    let nested_call = output
        .lines()
        .filter_map(|line| serde_json::from_str::<Value>(line).ok())
        .find(|event| {
            event["type"] == "tool.call" && event["payload"]["call_id"] == "call-exec/code-1"
        })
        .ok_or_else(|| eyre!("nested call did not retain its original exec lineage"))?;
    assert_eq!(nested_call["payload"]["model_call_index"], 1);
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn warmup_failure_falls_back_to_a_full_first_request() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut first = accept_async(stream).await?;
        assert_warmup(&next_json(&mut first).await?);
        send_json(
            &mut first,
            json!({
                "type": "error",
                "error": { "message": "prewarm unavailable" }
            }),
        )
        .await?;
        drop(first);

        let (stream, _) = listener.accept().await?;
        let mut second = accept_async(stream).await?;
        let generation = next_json(&mut second).await?;
        assert!(generation.get("previous_response_id").is_none());
        assert!(generation.get("generate").is_none());
        assert_eq!(generation["input"].as_array().map(Vec::len), Some(4));
        assert_eq!(generation["input"][0]["type"], "additional_tools");
        assert_eq!(generation["input"][1]["role"], "developer");
        assert_eq!(generation["input"][2]["role"], "user");
        assert_eq!(generation["input"][3]["role"], "user");
        send_final(&mut second, "resp-final").await
    });

    let workspace = temporary_workspace("warmup-fallback")?;
    let output = run_model(&endpoint, &workspace, "exercise warmup fallback").await?;
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    assert!(output.contains("\"model.warmup.failed\""));
    assert!(output.contains("\"purpose\":\"warmup_fallback\""));
    assert!(output.contains("\"connection_attempts\":2"));
    assert!(output.contains("\"websocket_reconnects\":1"));
    assert!(output.contains("\"run.completed\""));
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn warmup_connection_failure_falls_back_to_a_full_first_request() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (failed_prewarm, _) = listener.accept().await?;
        drop(failed_prewarm);

        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        let generation = next_json(&mut socket).await?;
        assert!(generation.get("previous_response_id").is_none());
        assert!(generation.get("generate").is_none());
        assert_eq!(generation["input"].as_array().map(Vec::len), Some(4));
        assert_eq!(generation["input"][0]["type"], "additional_tools");
        assert_eq!(generation["input"][1]["role"], "developer");
        assert_eq!(generation["input"][2]["role"], "user");
        assert_eq!(generation["input"][3]["role"], "user");
        send_final(&mut socket, "resp-final").await
    });

    let workspace = temporary_workspace("warmup-connection-fallback")?;
    let output = run_model(&endpoint, &workspace, "exercise warmup connection fallback").await?;
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    assert!(output.contains("\"model.connection.failed\""));
    assert!(output.contains("\"purpose\":\"warmup_fallback\""));
    assert!(output.contains("\"connection_attempts\":2"));
    assert!(output.contains("\"websocket_reconnects\":1"));
    assert!(output.contains("\"run.completed\""));
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn continues_past_previous_model_call_limit() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        assert_warmup(&next_json(&mut socket).await?);
        send_warmup(&mut socket, "resp-warmup").await?;

        for call_index in 1..=33 {
            let generation = next_json(&mut socket).await?;
            let previous_response_id = if call_index == 1 {
                "resp-warmup".to_owned()
            } else {
                format!("resp-tool-{}", call_index - 1)
            };
            assert_eq!(generation["previous_response_id"], previous_response_id);
            let response_id = format!("resp-tool-{call_index}");
            let call_id = format!("call-exec-{call_index}");
            send_json(
                &mut socket,
                completed_response(
                    &response_id,
                    &[json!({
                        "type": "custom_tool_call",
                        "call_id": call_id,
                        "name": "exec",
                        "input": "text(\"continue\")"
                    })],
                ),
            )
            .await?;
        }

        let final_generation = next_json(&mut socket).await?;
        assert_eq!(final_generation["previous_response_id"], "resp-tool-33");
        send_final(&mut socket, "resp-final").await
    });

    let workspace = temporary_workspace("unbounded-turn")?;
    let output = run_model(&endpoint, &workspace, "continue until done").await?;
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    assert!(output.contains("\"model_calls\":34"));
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn explicit_end_turn_false_continues_without_a_tool_call() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        assert_warmup(&next_json(&mut socket).await?);
        send_warmup(&mut socket, "resp-warmup").await?;

        let generation = next_json(&mut socket).await?;
        assert_eq!(generation["previous_response_id"], "resp-warmup");
        let mut response = completed_response(
            "resp-continue",
            &[json!({
                "type": "message",
                "role": "assistant",
                "content": [{ "type": "output_text", "text": "intermediate" }]
            })],
        );
        response["response"]["end_turn"] = json!(false);
        send_json(&mut socket, response).await?;

        let continuation = next_json(&mut socket).await?;
        assert_eq!(continuation["previous_response_id"], "resp-continue");
        assert_eq!(continuation["input"].as_array().map(Vec::len), Some(0));
        send_final(&mut socket, "resp-final").await
    });

    let workspace = temporary_workspace("end-turn-false")?;
    let output = run_model(&endpoint, &workspace, "continue when requested").await?;
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    assert!(output.contains("\"model_calls\":2"));
    assert!(output.contains("\"text\":\"done\""));
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn completed_response_accepts_null_usage() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        assert_warmup(&next_json(&mut socket).await?);
        send_warmup(&mut socket, "resp-warmup").await?;

        let generation = next_json(&mut socket).await?;
        assert_eq!(generation["previous_response_id"], "resp-warmup");
        let mut response = completed_response(
            "resp-final",
            &[json!({
                "type": "message",
                "role": "assistant",
                "content": [{ "type": "output_text", "text": "done" }]
            })],
        );
        response["response"]["usage"] = Value::Null;
        send_json(&mut socket, response).await
    });

    let workspace = temporary_workspace("null-usage")?;
    let output = run_model(&endpoint, &workspace, "accept missing usage").await?;
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    assert!(output.contains("\"model.call.completed\""));
    assert!(output.contains("\"usage\":null"));
    assert!(output.contains("\"run.completed\""));
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn completed_response_accepts_null_usage_details() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        assert_warmup(&next_json(&mut socket).await?);
        send_warmup(&mut socket, "resp-warmup").await?;

        let generation = next_json(&mut socket).await?;
        assert_eq!(generation["previous_response_id"], "resp-warmup");
        let mut response = completed_response(
            "resp-final",
            &[json!({
                "type": "message",
                "role": "assistant",
                "content": [{ "type": "output_text", "text": "done" }]
            })],
        );
        response["response"]["usage"]["input_tokens_details"] = Value::Null;
        response["response"]["usage"]["output_tokens_details"] = Value::Null;
        send_json(&mut socket, response).await
    });

    let workspace = temporary_workspace("null-usage-details")?;
    let output = run_model(&endpoint, &workspace, "accept missing usage details").await?;
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    assert!(output.contains("\"input_tokens_details\":null"));
    assert!(output.contains("\"output_tokens_details\":null"));
    assert!(output.contains("\"cached_input_tokens\":0"));
    assert!(output.contains("\"reasoning_output_tokens\":0"));
    assert!(output.contains("\"run.completed\""));
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn reconnect_drops_previous_response_id_and_replays_full_history() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut first = accept_async(stream).await?;
        let warmup = next_json(&mut first).await?;
        assert_warmup(&warmup);
        send_warmup(&mut first, "resp-warmup").await?;
        let generation = next_json(&mut first).await?;
        assert_eq!(generation["previous_response_id"], "resp-warmup");
        send_json(
            &mut first,
            completed_response(
                "resp-tool",
                &[json!({
                    "id": "server-item-id",
                    "type": "custom_tool_call",
                    "call_id": "call-exec",
                    "name": "exec",
                    "input": "text(\"continued\")"
                })],
            ),
        )
        .await?;
        first.send(Message::Close(None)).await?;
        drop(first);

        let (stream, _) = listener.accept().await?;
        let mut second = accept_async(stream).await?;
        let replay = next_json(&mut second).await?;
        assert!(replay.get("previous_response_id").is_none());
        assert_eq!(replay["store"], true);
        assert_eq!(replay["input"].as_array().map(Vec::len), Some(6));
        assert_eq!(replay["input"][0]["type"], "additional_tools");
        assert_eq!(replay["input"][1]["role"], "developer");
        assert_eq!(replay["input"][2]["role"], "user");
        assert_eq!(replay["input"][4]["type"], "custom_tool_call");
        assert!(replay["input"][4].get("id").is_none());
        assert_eq!(replay["input"][5]["type"], "custom_tool_call_output");
        send_final(&mut second, "resp-final").await
    });

    let workspace = temporary_workspace("reconnect")?;
    run_model(&endpoint, &workspace, "exercise reconnect").await?;
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn receive_reset_reconnects_without_replaying_completed_tools() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut first = accept_async(stream).await?;
        assert_warmup(&next_json(&mut first).await?);
        send_warmup(&mut first, "resp-warmup").await?;

        let generation = next_json(&mut first).await?;
        assert_eq!(generation["previous_response_id"], "resp-warmup");
        send_json(
            &mut first,
            completed_response(
                "resp-tool",
                &[json!({
                    "id": "server-item-id",
                    "type": "custom_tool_call",
                    "call_id": "call-exec",
                    "name": "exec",
                    "input": "const result = await tools.exec_command({cmd: \"printf x >> marker.txt\"}); text(result.output);"
                })],
            ),
        )
        .await?;

        let continuation = next_json(&mut first).await?;
        assert_eq!(continuation["previous_response_id"], "resp-tool");
        assert_eq!(continuation["input"].as_array().map(Vec::len), Some(1));
        let tool_output = continuation["input"][0].clone();
        send_json(
            &mut first,
            json!({
                "type": "response.created",
                "response": { "id": "resp-interrupted" }
            }),
        )
        .await?;
        send_json(
            &mut first,
            json!({
                "type": "response.in_progress",
                "response": { "id": "resp-interrupted" }
            }),
        )
        .await?;
        send_json(
            &mut first,
            json!({
                "type": "response.output_item.added",
                "output_index": 0,
                "item": { "type": "reasoning" }
            }),
        )
        .await?;
        drop(first);

        let (stream, _) = listener.accept().await?;
        let mut second = accept_async(stream).await?;
        let replay = next_json(&mut second).await?;
        assert!(replay.get("previous_response_id").is_none());
        assert_eq!(replay["input"].as_array().map(Vec::len), Some(6));
        assert_eq!(replay["input"][4]["type"], "custom_tool_call");
        assert_eq!(replay["input"][4]["call_id"], "call-exec");
        assert_eq!(replay["input"][5], tool_output);
        send_final(&mut second, "resp-final").await
    });

    let workspace = temporary_workspace("receive-reconnect")?;
    let output = run_model(&endpoint, &workspace, "recover after a receive reset").await?;
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    assert_eq!(std::fs::read_to_string(workspace.join("marker.txt"))?, "x");
    assert!(output.contains("\"model.attempt.retrying\""));
    assert!(output.contains("failed to receive a Responses WebSocket frame"));
    assert!(output.contains("\"purpose\":\"reconnect\""));
    assert!(output.contains("\"connection_attempts\":2"));
    assert!(output.contains("\"websocket_reconnects\":1"));
    assert!(output.contains("\"model_calls\":2"));
    assert!(!output.contains("\"model.call.failed\""));
    assert!(output.contains("\"run.completed\""));
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn sol_compacts_with_a_trigger_and_installs_the_returned_context() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut socket = accept_async(stream).await?;
        assert_warmup(&next_json(&mut socket).await?);
        send_warmup(&mut socket, "resp-warmup").await?;

        let generation = next_json(&mut socket).await?;
        assert_eq!(generation["previous_response_id"], "resp-warmup");
        send_json(
            &mut socket,
            completed_response_with_usage(
                "resp-tool",
                &[json!({
                    "id": "item-exec",
                    "type": "custom_tool_call",
                    "call_id": "call-exec",
                    "name": "exec",
                    "input": "await tools.apply_patch(\"*** Begin Patch\\n*** Add File: AGENTS.md\\n+fresh compacted instructions\\n*** End Patch\"); text(\"tool completed\")"
                })],
                372_001,
            ),
        )
        .await?;

        let compact = next_json(&mut socket).await?;
        assert_eq!(compact["previous_response_id"], "resp-tool");
        assert_eq!(compact["input"].as_array().map(Vec::len), Some(2));
        assert_eq!(compact["input"][0]["type"], "custom_tool_call_output");
        assert_eq!(
            compact["input"][0]["output"],
            "Output exceeded the available model context and was truncated"
        );
        assert_eq!(compact["input"][1], json!({ "type": "compaction_trigger" }));
        send_json(
            &mut socket,
            json!({
                "type": "response.output_item.done",
                "item": {
                    "id": "cmp-server-id",
                    "type": "compaction",
                    "encrypted_content": "opaque-summary"
                }
            }),
        )
        .await?;
        send_json(
            &mut socket,
            completed_response_with_usage("resp-compact", &[], 120),
        )
        .await?;

        let continuation = next_json(&mut socket).await?;
        assert!(continuation.get("previous_response_id").is_none());
        assert_eq!(continuation["input"].as_array().map(Vec::len), Some(5));
        assert_eq!(continuation["input"][0]["type"], "additional_tools");
        assert_eq!(continuation["input"][1]["role"], "developer");
        assert_eq!(continuation["input"][2]["role"], "user");
        assert_eq!(continuation["input"][3]["role"], "user");
        assert_eq!(continuation["input"][4]["type"], "compaction");
        assert_eq!(
            continuation["input"][4]["encrypted_content"],
            "opaque-summary"
        );
        assert!(continuation["input"][4].get("id").is_none());
        assert!(continuation.to_string().contains("exercise compaction"));
        assert!(
            continuation
                .to_string()
                .contains("fresh compacted instructions")
        );
        assert!(!continuation.to_string().contains("tool completed"));
        send_final(&mut socket, "resp-final").await
    });

    let workspace = temporary_workspace("compaction")?;
    let output = run_model(&endpoint, &workspace, "exercise compaction").await?;
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    assert!(output.contains("\"model.compaction.started\""));
    assert!(output.contains("\"model.compaction.completed\""));
    assert!(output.contains("\"compactions\":1"));
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn latest_fork_during_streaming_inherits_the_active_prompt_delta() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let (root_started, root_started_rx) = tokio::sync::oneshot::channel();
    let (release_root, release_root_rx) = tokio::sync::oneshot::channel();
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut root = accept_async(stream).await?;
        assert_warmup(&next_json(&mut root).await?);
        send_warmup(&mut root, "resp-warmup").await?;

        let active = next_json(&mut root).await?;
        assert_eq!(active["previous_response_id"], "resp-warmup");
        assert!(active.to_string().contains("active root prompt"));
        root_started
            .send(())
            .map_err(|()| eyre!("root request signal receiver dropped"))?;

        let (stream, _) = listener.accept().await?;
        let mut branch = accept_async(stream).await?;
        let fork = next_json(&mut branch).await?;
        assert_eq!(fork["previous_response_id"], "resp-warmup");
        let fork_text = fork.to_string();
        assert!(fork_text.contains("active root prompt"));
        assert!(fork_text.contains("BTW question"));
        send_final(&mut branch, "resp-branch").await?;

        release_root_rx
            .await
            .map_err(|_| eyre!("root release sender dropped"))?;
        send_final(&mut root, "resp-root").await
    });

    let workspace = temporary_workspace("active-prompt-fork")?;
    let responses = Responses::builder().websocket_url(endpoint).build();
    let (agent, root_events) = Nanocodex::builder("test-key")
        .thinking(Thinking::Low)
        .workspace(&workspace)
        .responses(responses)
        .session_id("model-test")
        .build()?;
    let root = agent.prompt("active root prompt").await?;
    root_started_rx
        .await
        .map_err(|_| eyre!("root request was not observed"))?;
    let (fork, fork_events) = agent.fork().await?;
    let branch = fork.prompt("BTW question").await?;
    assert_eq!(branch.result().await?.final_message, "done");
    release_root
        .send(())
        .map_err(|()| eyre!("root release receiver dropped"))?;
    assert_eq!(root.result().await?.final_message, "done");

    drop((agent, fork, root_events, fork_events));
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
#[allow(clippy::too_many_lines)]
async fn active_boundary_fork_sends_tool_and_steer_delta_then_replays_on_checkpoint_miss()
-> Result<()> {
    let workspace = temporary_workspace("active-tool-steer-fork")?;
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let (boundary_seen, boundary_seen_rx) = tokio::sync::oneshot::channel();
    let (release_root, release_root_rx) = tokio::sync::oneshot::channel();
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut root = accept_async(stream).await?;
        assert_warmup(&next_json(&mut root).await?);
        send_warmup(&mut root, "resp-warmup").await?;

        let initial = next_json(&mut root).await?;
        assert_eq!(initial["previous_response_id"], "resp-warmup");
        send_json(
            &mut root,
            completed_response(
                "resp-tool",
                &[json!({
                    "type": "custom_tool_call",
                    "call_id": "call-exec",
                    "name": "exec",
                    "input": "const result = await tools.exec_command({cmd: \"printf started > tool-started; while [ ! -f release-tool ]; do sleep 0.01; done; printf shit\"}); text(result.output);"
                })],
            ),
        )
        .await?;

        let continuation = next_json(&mut root).await?;
        assert_eq!(continuation["previous_response_id"], "resp-tool");
        assert_eq!(continuation["input"].as_array().map(Vec::len), Some(2));
        assert_eq!(continuation["input"][0]["type"], "custom_tool_call_output");
        assert!(continuation["input"][0].to_string().contains("shit"));
        assert_eq!(
            continuation["input"][1]["content"][0]["text"],
            "print shat instead"
        );
        boundary_seen
            .send(())
            .map_err(|()| eyre!("boundary signal receiver dropped"))?;

        let (stream, _) = listener.accept().await?;
        let mut branch = accept_async(stream).await?;
        let incremental = next_json(&mut branch).await?;
        assert_eq!(incremental["previous_response_id"], "resp-tool");
        assert_eq!(incremental["input"].as_array().map(Vec::len), Some(3));
        assert_eq!(incremental["input"][0]["type"], "custom_tool_call_output");
        assert!(incremental["input"][0].to_string().contains("shit"));
        assert_eq!(
            incremental["input"][1]["content"][0]["text"],
            "print shat instead"
        );
        assert_eq!(
            incremental["input"][2]["content"][0]["text"],
            "BTW question"
        );
        send_json(
            &mut branch,
            json!({
                "type": "error",
                "error": {
                    "code": "previous_response_not_found",
                    "message": "checkpoint expired"
                }
            }),
        )
        .await?;

        let replay = next_json(&mut branch).await?;
        assert!(replay.get("previous_response_id").is_none());
        let replay_text = replay.to_string();
        assert!(replay_text.contains("active root prompt"));
        assert!(replay_text.contains("call-exec"));
        assert!(replay_text.contains("shit"));
        assert!(replay_text.contains("print shat instead"));
        assert!(replay_text.contains("BTW question"));
        send_final(&mut branch, "resp-branch").await?;

        release_root_rx
            .await
            .map_err(|_| eyre!("root release sender dropped"))?;
        send_final(&mut root, "resp-root").await
    });

    let responses = Responses::builder().websocket_url(endpoint).build();
    let (agent, root_events) = Nanocodex::builder("test-key")
        .thinking(Thinking::Low)
        .workspace(&workspace)
        .responses(responses)
        .session_id("model-test")
        .build()?;
    let root = agent.prompt("active root prompt").await?;
    timeout(std::time::Duration::from_secs(5), async {
        while !workspace.join("tool-started").exists() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .map_err(|_| eyre!("tool process did not start"))?;
    root.steer("print shat instead").await?;
    std::fs::write(workspace.join("release-tool"), [])?;
    boundary_seen_rx
        .await
        .map_err(|_| eyre!("root boundary request was not observed"))?;

    let (fork, fork_events) = agent.fork().await?;
    assert_eq!(
        fork.prompt("BTW question")
            .await?
            .result()
            .await?
            .final_message,
        "done"
    );
    release_root
        .send(())
        .map_err(|()| eyre!("root release receiver dropped"))?;
    assert_eq!(root.result().await?.final_message, "done");

    drop((agent, fork, root_events, fork_events));
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
#[allow(clippy::too_many_lines)]
async fn latest_and_historical_forks_keep_distinct_boundaries_during_an_active_turn() -> Result<()>
{
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let (active_seen, active_seen_rx) = tokio::sync::oneshot::channel();
    let (release_active, release_active_rx) = tokio::sync::oneshot::channel();
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut root = accept_async(stream).await?;
        assert_warmup(&next_json(&mut root).await?);
        send_warmup(&mut root, "resp-warmup").await?;
        let first = next_json(&mut root).await?;
        assert!(first.to_string().contains("completed root prompt"));
        send_final(&mut root, "resp-first").await?;

        let active = next_json(&mut root).await?;
        assert_eq!(active["previous_response_id"], "resp-first");
        assert!(active.to_string().contains("active root prompt"));
        active_seen
            .send(())
            .map_err(|()| eyre!("active signal receiver dropped"))?;

        let (stream, _) = listener.accept().await?;
        let mut latest = accept_async(stream).await?;
        let latest_request = next_json(&mut latest).await?;
        assert_eq!(latest_request["previous_response_id"], "resp-first");
        assert_eq!(latest_request["input"].as_array().map(Vec::len), Some(2));
        assert_eq!(
            latest_request["input"][0]["content"][0]["text"],
            "active root prompt"
        );
        assert_eq!(
            latest_request["input"][1]["content"][0]["text"],
            "latest branch prompt"
        );
        send_final(&mut latest, "resp-latest").await?;

        let (stream, _) = listener.accept().await?;
        let mut historical = accept_async(stream).await?;
        let historical_request = next_json(&mut historical).await?;
        assert_eq!(historical_request["previous_response_id"], "resp-first");
        assert_eq!(
            historical_request["input"].as_array().map(Vec::len),
            Some(1)
        );
        assert_eq!(
            historical_request["input"][0]["content"][0]["text"],
            "historical branch prompt"
        );
        assert!(
            !historical_request
                .to_string()
                .contains("active root prompt")
        );
        send_final(&mut historical, "resp-historical").await?;

        release_active_rx
            .await
            .map_err(|_| eyre!("active release sender dropped"))?;
        send_final(&mut root, "resp-active").await
    });

    let workspace = temporary_workspace("latest-vs-historical-fork")?;
    let responses = Responses::builder().websocket_url(endpoint).build();
    let (agent, root_events) = Nanocodex::builder("test-key")
        .thinking(Thinking::Low)
        .workspace(&workspace)
        .responses(responses)
        .session_id("model-test")
        .build()?;
    let completed = agent
        .prompt("completed root prompt")
        .await?
        .result()
        .await?;
    let active = agent.prompt("active root prompt").await?;
    active_seen_rx
        .await
        .map_err(|_| eyre!("active root request was not observed"))?;

    let (latest, latest_events) = agent.fork().await?;
    assert_eq!(
        latest
            .prompt("latest branch prompt")
            .await?
            .result()
            .await?
            .final_message,
        "done"
    );
    let (historical, historical_events) = agent.fork_from(&completed).await?;
    assert_eq!(
        historical
            .prompt("historical branch prompt")
            .await?
            .result()
            .await?
            .final_message,
        "done"
    );
    release_active
        .send(())
        .map_err(|()| eyre!("active release receiver dropped"))?;
    assert_eq!(active.result().await?.final_message, "done");

    drop((
        agent,
        latest,
        historical,
        root_events,
        latest_events,
        historical_events,
    ));
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn historical_fork_runs_while_the_mainline_turn_is_in_flight() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let (root_started, root_started_rx) = tokio::sync::oneshot::channel();
    let (branch_started, branch_started_rx) = tokio::sync::oneshot::channel();
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut root = accept_async(stream).await?;
        let warmup = next_json(&mut root).await?;
        assert_warmup(&warmup);
        let lineage = warmup["prompt_cache_key"].clone();
        let root_session = warmup["client_metadata"]["session_id"].clone();
        send_warmup(&mut root, "resp-warmup").await?;

        let first = next_json(&mut root).await?;
        assert_eq!(first["previous_response_id"], "resp-warmup");
        send_final(&mut root, "resp-first").await?;
        let second = next_json(&mut root).await?;
        assert_eq!(second["previous_response_id"], "resp-first");
        send_final(&mut root, "resp-second").await?;

        let mainline = next_json(&mut root).await?;
        assert_eq!(mainline["previous_response_id"], "resp-second");
        root_started
            .send(())
            .map_err(|()| eyre!("root signal dropped"))?;
        let root_task = tokio::spawn(async move {
            branch_started_rx
                .await
                .map_err(|_| eyre!("branch signal dropped"))?;
            send_final(&mut root, "resp-mainline").await
        });

        let (stream, _) = listener.accept().await?;
        let mut branch = accept_async(stream).await?;
        let fork = next_json(&mut branch).await?;
        assert_eq!(fork["previous_response_id"], "resp-first");
        assert_eq!(fork["prompt_cache_key"], lineage);
        assert_ne!(fork["client_metadata"]["session_id"], root_session);
        assert_eq!(fork["input"].as_array().map(Vec::len), Some(1));
        assert_eq!(fork["input"][0]["content"][0]["text"], "fork prompt");
        branch_started
            .send(())
            .map_err(|()| eyre!("branch signal receiver dropped"))?;
        send_final(&mut branch, "resp-fork").await?;
        root_task.await??;
        Result::<()>::Ok(())
    });

    let workspace = temporary_workspace("historical-fork")?;
    let responses = Responses::builder().websocket_url(endpoint).build();
    let (agent, root_events) = Nanocodex::builder("test-key")
        .thinking(Thinking::Low)
        .workspace(&workspace)
        .responses(responses)
        .session_id("model-test")
        .build()?;
    let first = agent
        .prompt(Prompt::new("first prompt"))
        .await?
        .result()
        .await?;
    agent.prompt("second prompt").await?.result().await?;

    let mainline = agent.prompt("continue mainline").await?;
    root_started_rx
        .await
        .map_err(|_| eyre!("root request was not observed"))?;
    let (fork, fork_events) = agent.fork_from(&first).await?;
    let branch = fork.prompt("fork prompt").await?;
    let (mainline, branch) = tokio::join!(mainline.result(), branch.result());
    assert_eq!(mainline?.final_message, "done");
    assert_eq!(branch?.final_message, "done");

    drop((agent, fork, root_events, fork_events));
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn per_agent_tool_factory_binds_recursive_forks_to_the_invoking_driver() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut root = accept_async(stream).await?;
        let warmup = next_json(&mut root).await?;
        let lineage = warmup["prompt_cache_key"].clone();
        let root_session = warmup["client_metadata"]["session_id"].clone();
        send_warmup(&mut root, "resp-warmup").await?;
        let root_turn = next_json(&mut root).await?;
        assert_eq!(root_turn["previous_response_id"], "resp-warmup");
        send_final(&mut root, "resp-root").await?;

        let (stream, _) = listener.accept().await?;
        let mut child = accept_async(stream).await?;
        let child_turn = next_json(&mut child).await?;
        let child_session = child_turn["client_metadata"]["session_id"].clone();
        assert_eq!(child_turn["previous_response_id"], "resp-root");
        assert_eq!(child_turn["prompt_cache_key"], lineage);
        assert_ne!(child_session, root_session);
        send_final(&mut child, "resp-child").await?;

        let (stream, _) = listener.accept().await?;
        let mut grandchild = accept_async(stream).await?;
        let grandchild_turn = next_json(&mut grandchild).await?;
        assert_eq!(grandchild_turn["previous_response_id"], "resp-child");
        assert_eq!(grandchild_turn["prompt_cache_key"], lineage);
        assert_ne!(
            grandchild_turn["client_metadata"]["session_id"],
            child_session
        );
        send_final(&mut grandchild, "resp-grandchild").await
    });

    let (handles, mut received_handles) = tokio::sync::mpsc::unbounded_channel::<AgentHandle>();
    let workspace = temporary_workspace("recursive-fork-tools")?;
    let responses = Responses::builder().websocket_url(endpoint).build();
    let (root, root_events) = Nanocodex::builder("test-key")
        .thinking(Thinking::Low)
        .workspace(&workspace)
        .responses(responses)
        .session_id("model-test")
        .tools_factory(move |handle| {
            drop(handles.send(handle));
            Tools::builder().without_defaults().build()
        })
        .build()?;
    let root_handle = received_handles
        .recv()
        .await
        .ok_or_else(|| eyre!("root tool factory did not receive a fork handle"))?;

    root.prompt(Prompt::new("root turn"))
        .await?
        .result()
        .await?;
    let (child, child_events) = root_handle.fork().await?;
    let child_handle = received_handles
        .recv()
        .await
        .ok_or_else(|| eyre!("child tool factory did not receive a fork handle"))?;
    child.prompt("child turn").await?.result().await?;
    let (grandchild, grandchild_events) = child_handle.fork().await?;
    received_handles
        .recv()
        .await
        .ok_or_else(|| eyre!("grandchild tool factory did not receive a fork handle"))?;
    grandchild.prompt("grandchild turn").await?.result().await?;

    drop((
        root,
        child,
        grandchild,
        root_events,
        child_events,
        grandchild_events,
    ));
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn clean_spawn_reuses_an_explicit_cache_key_without_history_or_lineage() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut root = accept_async(stream).await?;
        let root_warmup = next_json(&mut root).await?;
        assert_eq!(root_warmup["prompt_cache_key"], "shared-private-prefix");
        assert!(
            root_warmup
                .to_string()
                .contains("shared private configuration"),
            "root request omitted the configured system prompt"
        );
        send_warmup(&mut root, "resp-root-warmup").await?;
        let root_turn = next_json(&mut root).await?;
        assert_eq!(root_turn["previous_response_id"], "resp-root-warmup");
        send_final(&mut root, "resp-root").await?;

        let (stream, _) = listener.accept().await?;
        let mut child = accept_async(stream).await?;
        let child_warmup = next_json(&mut child).await?;
        let child_session = child_warmup["client_metadata"]["session_id"]
            .as_str()
            .ok_or_else(|| eyre!("clean child warmup omitted its session id"))?;
        assert_ne!(child_session, "root-lineage");
        assert_eq!(child_warmup["prompt_cache_key"], "shared-private-prefix");
        assert!(child_warmup.get("previous_response_id").is_none());
        assert!(
            child_warmup
                .to_string()
                .contains("shared private configuration"),
            "clean child did not reuse the configured system prompt"
        );
        send_warmup(&mut child, "resp-child-warmup").await?;
        let child_turn = next_json(&mut child).await?;
        assert_eq!(child_turn["previous_response_id"], "resp-child-warmup");
        assert_ne!(child_turn["previous_response_id"], "resp-root");
        send_final(&mut child, "resp-child").await
    });

    let (handles, mut received_handles) = tokio::sync::mpsc::unbounded_channel::<AgentHandle>();
    let workspace = temporary_workspace("clean-spawn-tools")?;
    let responses = Responses::builder().websocket_url(endpoint).build();
    let (root, root_events) = Nanocodex::builder("private-test-key")
        .instructions("shared private configuration")
        .thinking(Thinking::Low)
        .responses(responses)
        .session_id("root-lineage")
        .prompt_cache_key("shared-private-prefix")
        .workspace(&workspace)
        .tools_factory(move |handle| {
            drop(handles.send(handle));
            Tools::builder().without_defaults().build()
        })
        .build()?;
    let root_handle = received_handles
        .recv()
        .await
        .ok_or_else(|| eyre!("root tool factory did not receive an agent handle"))?;
    root.prompt("root turn").await?.result().await?;

    let (child, child_events) = root_handle.spawn().await?;
    received_handles
        .recv()
        .await
        .ok_or_else(|| eyre!("clean child tool factory did not receive an agent handle"))?;
    child.prompt("clean child turn").await?.result().await?;

    drop((root, child, root_events, child_events));
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn cloned_builders_singleflight_one_shared_prefix_warmup() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut first = accept_async(stream).await?;
        let warmup = next_json(&mut first).await?;
        assert_eq!(warmup["prompt_cache_key"], "shared-prefix");
        let first_session = warmup["client_metadata"]["session_id"]
            .as_str()
            .ok_or_else(|| eyre!("first warmup omitted its session id"))?
            .to_owned();
        send_warmup(&mut first, "resp-shared-warmup").await?;
        let first_turn = next_json(&mut first).await?;
        assert_eq!(first_turn["previous_response_id"], "resp-shared-warmup");
        send_final(&mut first, "resp-first").await?;

        let (stream, _) = listener.accept().await?;
        let mut second = accept_async(stream).await?;
        let second_turn = next_json(&mut second).await?;
        assert_eq!(second_turn["prompt_cache_key"], "shared-prefix");
        assert!(second_turn.get("previous_response_id").is_none());
        assert_ne!(second_turn["client_metadata"]["session_id"], first_session);
        assert_eq!(second_turn["input"].as_array().map(Vec::len), Some(4));
        assert!(second_turn.get("generate").is_none());
        send_final(&mut second, "resp-second").await
    });

    let workspace = temporary_workspace("shared-warmup")?;
    let responses = Responses::builder().websocket_url(endpoint).build();
    let builder = Nanocodex::builder("test-key")
        .thinking(Thinking::Low)
        .workspace(&workspace)
        .responses(responses)
        .prompt_cache_key("shared-prefix")
        .shared_prompt_cache();

    let (first, mut first_events) = builder.clone().build()?;
    let first_session = first.session_id().to_owned();
    first.prompt("first turn").await?.result().await?;
    drop(first);
    let mut first_warmup_source = None;
    while let Some(event) = first_events.recv().await {
        if event.kind == nanocodex_core::AgentEventKind::ModelWarmupCompleted {
            first_warmup_source = Some(event.decode_payload::<Value>()?["source"].clone());
        }
    }

    let (second, mut second_events) = builder.build()?;
    assert_ne!(second.session_id(), first_session);
    second.prompt("second turn").await?.result().await?;
    drop(second);
    let mut second_warmup_source = None;
    while let Some(event) = second_events.recv().await {
        if event.kind == nanocodex_core::AgentEventKind::ModelWarmupCompleted {
            let payload = event.decode_payload::<Value>()?;
            assert!(payload.get("response_id").is_none());
            second_warmup_source = Some(payload["source"].clone());
        }
    }

    assert_eq!(first_warmup_source, Some(json!("response")));
    assert_eq!(second_warmup_source, Some(json!("shared_prefix")));
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

#[tokio::test]
async fn missing_stored_checkpoint_replays_local_history_once() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("ws://{}", listener.local_addr()?);
    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut root = accept_async(stream).await?;
        assert_warmup(&next_json(&mut root).await?);
        send_warmup(&mut root, "resp-warmup").await?;
        let first = next_json(&mut root).await?;
        send_final(&mut root, "resp-first").await?;

        let (stream, _) = listener.accept().await?;
        let mut branch = accept_async(stream).await?;
        let checkpoint = next_json(&mut branch).await?;
        assert_eq!(checkpoint["previous_response_id"], "resp-first");
        assert_eq!(checkpoint["input"].as_array().map(Vec::len), Some(1));
        send_json(
            &mut branch,
            json!({
                "type": "error",
                "error": {
                    "code": "previous_response_not_found",
                    "message": "checkpoint expired"
                }
            }),
        )
        .await?;

        let replay = next_json(&mut branch).await?;
        assert!(replay.get("previous_response_id").is_none());
        assert_eq!(replay["store"], true);
        assert_eq!(replay["input"][0]["type"], "additional_tools");
        assert_eq!(replay["input"][1]["role"], "developer");
        let replay_text = replay.to_string();
        assert!(replay_text.contains("root prompt"));
        assert!(replay_text.contains("branch after eviction"));
        assert!(
            replay["input"]
                .as_array()
                .is_some_and(|items| items.len() > 4)
        );
        send_final(&mut branch, "resp-replayed").await?;
        drop((root, first));
        Result::<()>::Ok(())
    });

    let workspace = temporary_workspace("checkpoint-miss")?;
    let responses = Responses::builder().websocket_url(endpoint).build();
    let (agent, root_events) = Nanocodex::builder("test-key")
        .thinking(Thinking::Low)
        .workspace(&workspace)
        .responses(responses)
        .session_id("model-test")
        .build()?;
    let first = agent
        .prompt(Prompt::new("root prompt"))
        .await?
        .result()
        .await?;
    let (fork, mut fork_events) = agent.fork_from(&first).await?;
    let branch = fork.prompt("branch after eviction").await?;
    assert_eq!(branch.result().await?.final_message, "done");

    drop((agent, fork, root_events));
    let mut observed_checkpoint_retry = false;
    while let Some(event) = fork_events.recv().await {
        if event.kind == nanocodex_core::AgentEventKind::ModelAttemptRetrying {
            let payload = event.decode_payload::<Value>()?;
            observed_checkpoint_retry = payload["error_class"] == "checkpoint_missing"
                && payload["replay_mode"] == "full_history"
                && payload["opens_new_socket"] == false;
        }
    }
    assert!(observed_checkpoint_retry);
    timeout(std::time::Duration::from_secs(5), server)
        .await
        .map_err(|_| eyre!("mock Responses server did not finish"))???;
    std::fs::remove_dir_all(workspace)?;
    Ok(())
}

fn assert_warmup(warmup: &Value) {
    assert_eq!(warmup["store"], true);
    assert_eq!(warmup["generate"], false);
    assert_eq!(warmup["stream"], true);
    assert_eq!(warmup["parallel_tool_calls"], false);
    assert_eq!(warmup["prompt_cache_key"], "model-test");
    assert_eq!(warmup["input"].as_array().map(Vec::len), Some(2));
    assert_eq!(warmup["input"][0]["type"], "additional_tools");
    assert_eq!(warmup["input"][0]["role"], "developer");
    assert_eq!(warmup["input"][0]["tools"][0]["type"], "custom");
    assert_eq!(warmup["input"][0]["tools"][0]["name"], "exec");
    assert!(
        warmup["input"][0]["tools"][0]["description"]
            .as_str()
            .is_some_and(|description| description.contains("`web__run`"))
    );
    assert_eq!(warmup["input"][0]["tools"][1]["type"], "function");
    assert_eq!(warmup["input"][0]["tools"][1]["name"], "wait");
    assert_eq!(warmup["input"][1]["role"], "developer");
    assert!(warmup.get("tools").is_none());
    assert!(warmup.get("instructions").is_none());
    assert!(warmup.get("context_management").is_none());
    assert!(warmup["reasoning"].get("mode").is_none());
    assert_eq!(
        warmup["client_metadata"]["ws_request_header_x_openai_internal_codex_responses_lite"],
        "true"
    );
}

async fn run_model(endpoint: &str, workspace: &Path, instruction: &str) -> Result<String> {
    let task = Prompt::new(instruction);
    let responses = Responses::builder().websocket_url(endpoint).build();
    let (agent, events) = Nanocodex::builder("test-key")
        .thinking(Thinking::Low)
        .workspace(workspace)
        .responses(responses)
        .session_id("model-test")
        .build()?;
    let turn = agent.prompt(task).await?;
    drop(agent);
    let mut output = Vec::new();
    let (event_result, turn_result) = tokio::join!(events.write_jsonl(&mut output), turn.result());
    event_result?;
    turn_result?;
    Ok(String::from_utf8(output)?)
}

async fn send_warmup<S>(socket: &mut WebSocketStream<S>, response_id: &str) -> Result<()>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
    send_json(
        socket,
        json!({
            "type": "response.completed",
            "response": { "id": response_id, "usage": null }
        }),
    )
    .await
}

async fn send_final<S>(socket: &mut WebSocketStream<S>, response_id: &str) -> Result<()>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
    send_json(
        socket,
        completed_response(
            response_id,
            &[json!({
                "type": "message",
                "role": "assistant",
                "content": [{ "type": "output_text", "text": "done" }]
            })],
        ),
    )
    .await
}

async fn send_assistant_output<S>(
    socket: &mut WebSocketStream<S>,
    output_index: u32,
    item_id: &str,
    phase: &str,
    text: &str,
) -> Result<Value>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
    let completed = json!({
        "id": item_id,
        "type": "message",
        "role": "assistant",
        "status": "completed",
        "phase": phase,
        "content": [{ "type": "output_text", "text": text }]
    });
    send_json(
        socket,
        json!({
            "type": "response.output_item.added",
            "output_index": output_index,
            "item": {
                "id": item_id,
                "type": "message",
                "role": "assistant",
                "status": "in_progress",
                "phase": phase,
                "content": []
            }
        }),
    )
    .await?;
    send_json(
        socket,
        json!({
            "type": "response.output_text.delta",
            "output_index": output_index,
            "content_index": 0,
            "delta": text
        }),
    )
    .await?;
    send_json(
        socket,
        json!({
            "type": "response.output_item.done",
            "output_index": output_index,
            "item": completed.clone()
        }),
    )
    .await?;
    Ok(completed)
}

fn completed_response(response_id: &str, output: &[Value]) -> Value {
    completed_response_with_usage(response_id, output, 12)
}

fn completed_response_with_usage(response_id: &str, output: &[Value], total_tokens: u64) -> Value {
    json!({
        "type": "response.completed",
        "response": {
            "id": response_id,
            "status": "completed",
            "output": output,
            "usage": {
                "input_tokens": 10,
                "input_tokens_details": { "cached_tokens": 5 },
                "output_tokens": 2,
                "output_tokens_details": { "reasoning_tokens": 1 },
                "total_tokens": total_tokens
            }
        }
    })
}

async fn next_json<S>(socket: &mut WebSocketStream<S>) -> Result<Value>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
    loop {
        let message = socket
            .next()
            .await
            .ok_or_else(|| eyre!("client closed before sending a request"))??;
        if let Message::Text(text) = message {
            return Ok(serde_json::from_str(text.as_str())?);
        }
    }
}

async fn send_json<S>(socket: &mut WebSocketStream<S>, value: Value) -> Result<()>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
    socket.send(Message::Text(value.to_string().into())).await?;
    Ok(())
}

fn temporary_workspace(label: &str) -> Result<PathBuf> {
    let path = std::env::temp_dir().join(format!(
        "nanocodex-{label}-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)?
            .as_nanos()
    ));
    std::fs::create_dir_all(&path)?;
    Ok(path)
}
