newt-core 0.8.0

//! The OpenAI-compatible final answer, tested at the WIRING (#123, #2372).
//!
//! `openai_sse` proves the parser with no HTTP in it. This file drives whole
//! turns through `chat_complete`. Since #2372 an accepted final answer is
//! generated ONCE: the loop returns the gated reply with `was_streamed = false`
//! and the host renders it, so these tests pin the request counts and the text
//! that comes back rather than a display stream.
//!
//! These tests touch no process-global state (no env, no filesystem), so
//! unlike `anthropic_loop.rs` they need neither a serial lane nor an env guard.

use super::*;
use crate::caveats::Caveats;
use crate::{BackendKind, MemMessage};
use std::sync::{Arc, Mutex};
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};

/// A workspace that deliberately does not exist, so the self-verify gate
/// finds no check to demand and cannot add a round to the request counts
/// these tests assert on (same reason as `anthropic_loop.rs`).
const NO_CHECKS_WORKSPACE: &str = "newt-core-test-workspace-that-does-not-exist";

fn ctx<'a>(server_uri: &'a str, messages: &'a [MemMessage], caveats: &'a Caveats) -> ChatCtx<'a> {
    ChatCtx {
        shell_dialect: crate::agentic::shell_dialect_sentence(false, false),
        command_budget: Default::default(),
        overflow_retry: Default::default(),
        run_allowance: None,
        verify_outcomes: false,
        round_cap_hit: None,
        smart_harness: None,
        url: server_uri,
        model: "test-model",
        kind: BackendKind::Openai,
        api_key: Some("sk-test"),
        messages,
        task: "do the thing",
        workspace: NO_CHECKS_WORKSPACE,
        default_command_cwd: None,
        worktree_session: None,
        color: false,
        markdown: false,
        tool_offload: false,
        spill_store: None,
        disclosure: None,
        compaction_store: None,
        scratchpad: false,
        scratchpad_store: None,
        code_search: None,
        where_is: None,
        nav: None,
        exposure: Default::default(),
        experience_store: None,
        step_ledger: None,
        caveats,
        persona_tools: None,
        cognition: None,
        chat_completions_capability: Default::default(),
        responses_capability: Default::default(),
        openai_api: Default::default(),
        output_allowance: None,
        attempt_ledger: None,
        reasoning_replay_scope: crate::model_card::ReasoningReplayScope::Never,
        emits_leading_reasoning: false,
        max_tool_rounds: 8,
        narration_nudge_cap: 1,
        action_nudges: true,
        prompt_disposition: PromptDisposition::Act,
        prompt_intake: None,
        workflow_grace_rounds: 0,
        tool_output_lines: 20,
        debug: false,
        trace: false,
        num_ctx: None,
        input_ceiling_pct: 80,
        low_budget_pct: 15,
        connect_timeout_secs: 5,
        inference_timeout_secs: 30,
        mid_loop_trim_threshold: 40,
        compaction_trigger_policy: crate::CompactionTriggerPolicy::HeadroomAware,
        mid_loop_trim_tokens: None,
        max_ok_input: None,
        build_check_cmd: None,
        safe_context: None,
        recover_cw_400: None,
        note_sink: None,
        note_nudge: None,
        recall_source: None,
        memory_source: None,
        summarizer: None,
        compress_state: None,
        tool_events: None,
        phantom_reaches: None,
        end_reason: None,
        solve_obs: None,
        permission_gate: None,
        on_round_usage: None,
        estimate_ratio: None,
        estimation: crate::tokens::TokenEstimation::default(),
        summary_input_cap_floor_chars: 8_192,
        rewrites_history: true,
        exec_floor: None,
        write_ledger: None,
        attribution: None,
        cancel: None,
        live_tool_output: None,
        git_tool: None,
        crew_runner: None,
        operating_mode_control: None,
        plan_mode_control: None,
        plan_draft_sink: None,
        steering: None,
        completed_spill_renderer: None,
    }
}

fn msgs() -> Vec<MemMessage> {
    vec![
        MemMessage::system("you are a test"),
        MemMessage::user("do the thing"),
    ]
}

fn body_json(req: &Request) -> serde_json::Value {
    serde_json::from_slice(&req.body).unwrap_or_default()
}

/// A complete JSON response from an endpoint ignoring the requested SSE transport.
fn probe_reply(content: &str) -> ResponseTemplate {
    ResponseTemplate::new(200).set_body_json(serde_json::json!({
        "choices": [{"message": {"content": content}, "finish_reason": "stop"}],
        "usage": {"prompt_tokens": 100, "completion_tokens": 7},
    }))
}

/// An SSE body from `data:` frames — the framing the wire actually uses.
fn sse(frames: &[&str]) -> ResponseTemplate {
    let body: String = frames.iter().map(|f| format!("data: {f}\n\n")).collect();
    ResponseTemplate::new(200).set_body_raw(body.into_bytes(), "text/event-stream")
}

/// Text deltas, a usage chunk, and the `[DONE]` sentinel.
fn sse_text(parts: &[&str], input: u64, output: u64) -> ResponseTemplate {
    let mut frames: Vec<String> = parts
        .iter()
        .map(|p| format!(r#"{{"choices":[{{"delta":{{"content":"{p}"}}}}]}}"#))
        .collect();
    frames.push(r#"{"choices":[{"delta":{},"finish_reason":"stop"}]}"#.to_string());
    frames.push(format!(
        r#"{{"choices":[],"usage":{{"prompt_tokens":{input},"completion_tokens":{output}}}}}"#
    ));
    frames.push("[DONE]".to_string());
    sse(&frames.iter().map(String::as_str).collect::<Vec<_>>())
}

const STREAM_USAGE: &str = r#"{"choices":[],"usage":{"prompt_tokens":42,"completion_tokens":5}}"#;

fn stream_usage() -> Option<crate::TokenUsage> {
    Some(crate::TokenUsage {
        input_tokens: 42,
        output_tokens: 5,
    })
}

/// The socket test can only split where the kernel agrees to split, so on some
/// run it may deliver both halves together and pass without proving anything.
/// This splits at EVERY byte offset — including inside all of a 2-, 3- and
/// 4-byte character — and is pure, so it is the guard that actually holds.
#[test]
fn decode_chunk_reassembles_a_split_at_every_byte_offset() {
    let s = "aé→𝄞z";
    let bytes = s.as_bytes();
    for cut in 0..=bytes.len() {
        let mut carry = Vec::new();
        let mut got = decode_chunk(&mut carry, &bytes[..cut]);
        got.push_str(&decode_chunk(&mut carry, &bytes[cut..]));
        assert_eq!(got, s, "split at byte {cut} corrupted the answer");
        assert!(carry.is_empty(), "nothing is left held at byte {cut}");
    }
}

/// Byte-at-a-time is the same property taken to its limit — the shape
/// `openai_sse`'s own `one_byte_at_a_time_produces_the_same_answer` already
/// uses one layer up.
#[test]
fn decode_chunk_survives_one_byte_at_a_time() {
    let s = "aé→𝄞z";
    let mut carry = Vec::new();
    let got: String = s
        .as_bytes()
        .iter()
        .map(|b| decode_chunk(&mut carry, &[*b]))
        .collect();
    assert_eq!(got, s);
    assert!(carry.is_empty());
}

/// Bytes that are not a cut character but genuinely invalid must be SPENT, not
/// held: holding them would grow the carry without bound and stall the stream
/// forever, waiting for a continuation that is never coming.
#[test]
fn decode_chunk_spends_invalid_bytes_instead_of_stalling_on_them() {
    let mut carry = Vec::new();
    let got = decode_chunk(&mut carry, b"ok\xffthen");
    assert_eq!(got, "ok\u{FFFD}then");
    assert!(
        carry.is_empty(),
        "an invalid byte is consumed, never carried"
    );
}

/// #2372 (moved onto the primary path from the deleted display reissue): a
/// primary stream cut before `[DONE]` after its usage chunk was generated and
/// billed, so every attempt is failed WITH that usage.
#[tokio::test]
async fn a_primary_stream_cut_after_its_usage_chunk_is_failed_with_that_usage() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v1/chat/completions"))
        .respond_with(sse(&[
            r#"{"choices":[{"delta":{"content":"half an answ"}}]}"#,
            STREAM_USAGE,
        ]))
        .mount(&server)
        .await;
    let messages = msgs();
    let caveats = Caveats::top();
    let uri = server.uri();
    let ledger = std::sync::Mutex::new(crate::attempts::AttemptLedger::default());
    let mut c = ctx(&uri, &messages, &caveats);
    c.attempt_ledger = Some(&ledger);
    let error = chat_complete(c, &mut NoMcp)
        .await
        .expect_err("a stream that never reaches [DONE] is not an answer");
    assert!(format!("{error:#}").contains("[DONE]"), "{error:#}");

    let received = server.received_requests().await.expect("journal").len();
    let ledger = ledger.lock().unwrap();
    let records: Vec<_> = ledger.records().collect();
    assert_eq!(records.len(), received, "attempts == wire requests");
    for record in records {
        assert_eq!(record.state, crate::attempts::AttemptState::Failed);
        assert_eq!(record.usage, stream_usage());
    }
}

const REISSUE_USAGE: &str = r#"{"choices":[],"usage":{"prompt_tokens":42,"completion_tokens":5}}"#;

fn reissue_usage() -> Option<crate::TokenUsage> {
    Some(crate::TokenUsage {
        input_tokens: 42,
        output_tokens: 5,
    })
}

/// #2372 (moved onto the primary path): an error event after a usage frame
/// fails the attempt with that usage even when `[DONE]` follows, and a context
/// rejection is classified `ContextExceeded` while any other error is not.
#[tokio::test]
async fn a_primary_stream_error_event_after_usage_is_failed_with_that_usage() {
    for (error_frame, context_exceeded) in [
        (
            r#"{"error":{"message":"Context size has been exceeded."}}"#,
            true,
        ),
        (r#"{"error":{"message":"busy","code":503}}"#, false),
    ] {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/v1/chat/completions"))
            .respond_with(sse(&[STREAM_USAGE, error_frame, "[DONE]"]))
            .mount(&server)
            .await;
        let messages = msgs();
        let caveats = Caveats::top();
        let uri = server.uri();
        let ledger = std::sync::Mutex::new(crate::attempts::AttemptLedger::default());
        let mut c = ctx(&uri, &messages, &caveats);
        c.attempt_ledger = Some(&ledger);
        let error = chat_complete(c, &mut NoMcp)
            .await
            .expect_err("an error event is not an answer");
        assert_eq!(
            crate::agentic::observability::error_class(&error)
                == Some(crate::agentic::observability::ErrorClass::ContextExceeded),
            context_exceeded,
            "{error_frame}: {error:#}"
        );

        let received = server.received_requests().await.expect("journal").len();
        let ledger = ledger.lock().unwrap();
        let records: Vec<_> = ledger.records().collect();
        assert_eq!(
            records.len(),
            received,
            "{error_frame}: attempts == wire requests"
        );
        for record in records {
            assert_eq!(
                record.state,
                crate::attempts::AttemptState::Failed,
                "{error_frame}"
            );
            assert_eq!(record.usage, stream_usage(), "{error_frame}");
        }
    }
}

/// Review round 2, item 1: a complete `[DONE]` primary stream that strict
/// decoding rejects (here a tool call with a non-string id) was still generated and
/// billed. Every attempt is failed WITH the usage the stream reported.
#[tokio::test]
async fn a_primary_stream_rejected_by_strict_decoding_is_failed_with_its_reported_usage() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v1/chat/completions"))
        .respond_with(sse(&[
            r#"{"choices":[{"delta":{"tool_calls":[{"index":0,"id":7,"type":"function","function":{"name":"read_file","arguments":"{}"}}]}}]}"#,
            r#"{"choices":[{"delta":{},"finish_reason":"tool_calls"}]}"#,
            STREAM_USAGE,
            "[DONE]",
        ]))
        .mount(&server)
        .await;
    let messages = msgs();
    let caveats = Caveats::top();
    let uri = server.uri();
    let ledger = std::sync::Mutex::new(crate::attempts::AttemptLedger::default());
    let mut c = ctx(&uri, &messages, &caveats);
    c.attempt_ledger = Some(&ledger);
    // A NON-STRING id cannot be read at all (a missing id is now the batch
    // validator's, re-asked; see the streamed_idless_* tests).
    let error = chat_complete(c, &mut NoMcp)
        .await
        .expect_err("a tool call with an unreadable id is rejected");
    assert!(
        format!("{error:#}").contains("invalid tool-call ID"),
        "{error:#}"
    );

    let received = server.received_requests().await.expect("journal").len();
    let ledger = ledger.lock().unwrap();
    let records: Vec<_> = ledger.records().collect();
    assert_eq!(records.len(), received, "attempts == wire requests");
    for record in records {
        assert_eq!(record.state, crate::attempts::AttemptState::Failed);
        assert_eq!(record.usage, stream_usage());
    }
}

/// Send one request to `url` through `dispatch_with_decoder` (no retries) with
/// the strict OpenAI decoder, expecting it to fail; returns the error and the
/// one attempt it recorded.
async fn failed_dispatch_attempt(
    url: &str,
    decode: fn(&[u8]) -> anyhow::Result<serde_json::Value>,
) -> (anyhow::Error, crate::attempts::AttemptRecord) {
    let ledger = std::sync::Mutex::new(crate::attempts::AttemptLedger::default());
    let scope = attempt_capture::AttemptScope {
        run_allowance: None,
        ledger: &ledger,
        turn: "prompt:turn",
        model: "test-model",
        backend: "test-backend",
        cancel: None,
    };
    let policy = RetryPolicy {
        max_retries: 0,
        base: std::time::Duration::ZERO,
        max: std::time::Duration::ZERO,
        jitter: false,
    };
    let error = dispatch_with_decoder(
        &policy,
        Some(scope),
        || async {
            Ok(reqwest::Client::new()
                .post(url)
                .json(&serde_json::json!({})))
        },
        "inference endpoint",
        |_, _, _| {},
        None,
        decode,
    )
    .await
    .expect_err("the response is rejected");
    let ledger = ledger.lock().unwrap();
    let records: Vec<_> = ledger.records().cloned().collect();
    assert_eq!(records.len(), 1);
    (error, records[0].clone())
}

/// Item 1's other primary send: `dispatch_with_decoder` (the cap-exit summary)
/// keeps a strictly rejected response's usage on its failed attempt too.
#[tokio::test]
async fn dispatch_with_decoder_keeps_a_strictly_rejected_responses_usage() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v1/chat/completions"))
        .respond_with(sse(&[
            r#"{"choices":[{"delta":{"content":"a summary"}}]}"#,
            STREAM_USAGE,
            "[DONE]",
        ]))
        .mount(&server)
        .await;
    let url = format!("{}/v1/chat/completions", server.uri());
    let (error, record) =
        failed_dispatch_attempt(&url, smart_harness::decode_openai_response).await;
    assert!(
        format!("{error:#}").contains("no finish reason"),
        "{error:#}"
    );
    assert_eq!(record.state, crate::attempts::AttemptState::Failed);
    assert_eq!(record.usage, stream_usage());
}

/// Serve `body` under a `Content-Length` it never meets, so the client's body
/// read fails (not a clean EOF), and dispatch it with `decode`.
async fn dropped_body_attempt(
    body: String,
    decode: fn(&[u8]) -> anyhow::Result<serde_json::Value>,
) -> (anyhow::Error, crate::attempts::AttemptRecord) {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!(
        "http://{}/v1/chat/completions",
        listener.local_addr().unwrap()
    );
    let server = tokio::spawn(async move {
        let (mut sock, _) = listener.accept().await.unwrap();
        let mut scratch = [0u8; 8192];
        assert!(
            sock.read(&mut scratch).await.unwrap() > 0,
            "the client asked"
        );
        let head = "HTTP/1.1 200 OK\r\nContent-Length: 100000\r\n\r\n";
        sock.write_all(format!("{head}{body}").as_bytes())
            .await
            .unwrap();
        // Dropping the socket here cuts the body short of its declared length.
    });
    let (error, record) = failed_dispatch_attempt(&url, decode).await;
    server.abort();
    assert!(
        format!("{error:#}").contains("request failed reading response"),
        "{error:#}"
    );
    (error, record)
}

/// Review rounds 3 and 4, item e: a 2xx body whose read fails keeps the usage
/// its bytes already reported — whether the decode of those bytes was rejected
/// (a stream cut mid-answer), or succeeded (a whole `[DONE]` stream, or whole
/// JSON, before the break).
#[tokio::test]
async fn a_body_read_failure_keeps_the_usage_its_bytes_reported() {
    let answer = r#"{"choices":[{"delta":{"content":"an answer"},"finish_reason":"stop"}]}"#;
    let json: fn(&[u8]) -> anyhow::Result<serde_json::Value> =
        |bytes| Ok(serde_json::from_slice(bytes)?);
    let cases = [
        (
            "stream cut mid-answer",
            format!("data: {answer}\n\ndata: {STREAM_USAGE}\n\n"),
            smart_harness::decode_openai_response as fn(&[u8]) -> _,
        ),
        (
            "whole [DONE] stream",
            format!("data: {answer}\n\ndata: {STREAM_USAGE}\n\ndata: [DONE]\n\n"),
            smart_harness::decode_openai_response,
        ),
        (
            "whole Ollama JSON",
            r#"{"message":{"content":"an answer"},"prompt_eval_count":42,"eval_count":5}"#
                .to_string(),
            json,
        ),
    ];
    for (name, body, decode) in cases {
        let (_, record) = dropped_body_attempt(body, decode).await;
        assert_eq!(
            record.state,
            crate::attempts::AttemptState::Failed,
            "{name}"
        );
        assert_eq!(record.usage, stream_usage(), "{name}");
    }
}

/// End to end: an interrupt that lands after the answer arrived ends the turn
/// with an empty reply — the loop's own round-boundary contract.
///
/// An empty REPLY is not an empty BILL. The round was paid for before the
/// operator pressed anything, so its usage has to survive the cancelled arm.
///
/// The interrupt is tripped from the tool call's own execution, not from the
/// HTTP fixture: `cancellable()` races the interrupt poll against the
/// in-flight request itself (deliberately — see its doc comment, and
/// `an_anthropic_stream_cancelled_mid_flight_is_never_ok`, which pins the
/// opposite outcome for a flag flipped *during* that same race), so flipping
/// the flag inside `Respond::respond` (as this test used to) races the same
/// primitive and is exactly the ordering the Windows CI flake exposed. A tool
/// call's `McpTools::call` runs strictly after round 1's response — and its
/// usage — was already recorded, and strictly before the round-boundary
/// `is_cancelled` check that decides whether round 2 is dispatched. That is
/// real happens-before from single-task control flow, not a timing race.
#[tokio::test]
async fn an_interrupt_after_the_probe_ends_the_turn_with_no_second_call() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v1/chat/completions"))
        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
            "choices": [{"message": {
                "content": null,
                "tool_calls": [{
                    "id": "call_1",
                    "type": "function",
                    "function": {"name": "srv__cancel_after_billing", "arguments": "{}"}
                }]
            }}],
            "usage": {"prompt_tokens": 100, "completion_tokens": 7},
        })))
        .mount(&server)
        .await;

    let messages = msgs();
    let caveats = Caveats::top();
    let uri = server.uri();
    let cancel = std::sync::atomic::AtomicBool::new(false);
    let mut permission = super::anthropic_loop_tests::FixtureMcpPermission::new(
        "srv__cancel_after_billing",
        &caveats,
    );
    let mut ctx = ctx(&uri, &messages, &caveats);
    ctx.cancel = Some(&cancel);
    ctx.permission_gate = Some(&mut permission);
    let mut mcp = CancelAfterBilling { flag: &cancel };
    let (reply, streamed, usage, _hallu) = chat_complete(ctx, &mut mcp)
        .await
        .expect("an interrupt is not an error");

    assert_eq!(reply, "", "an interrupted turn ends with an empty reply");
    assert!(!streamed);
    assert_eq!(server.received_requests().await.unwrap().len(), 1);
    let u = usage.expect("the round was billed before the interrupt landed");
    assert_eq!(
        (u.input_tokens, u.output_tokens),
        (100, 7),
        "the cancelled arm returns the turn's accumulated usage, not a fresh zero: {u:?}"
    );
}

/// An MCP tool whose execution trips the interrupt flag — deterministically
/// after the round that called it was already billed.
struct CancelAfterBilling<'a> {
    flag: &'a std::sync::atomic::AtomicBool,
}
#[async_trait::async_trait]
impl McpTools for CancelAfterBilling<'_> {
    fn handles(&self, name: &str) -> bool {
        name == "srv__cancel_after_billing"
    }
    fn tool_defs(&self) -> Vec<serde_json::Value> {
        Vec::new()
    }
    async fn call(&mut self, _leased: &crate::agentic::mcp::LeasedMcpCall<'_>) -> String {
        self.flag.store(true, std::sync::atomic::Ordering::Relaxed);
        "the operator interrupted right after this".to_string()
    }
}

/// A markdown turn returns the RAW answer — it is persisted and re-sent, so no
/// styling may enter it — and leaves rendering to the host.
#[tokio::test]
async fn a_markdown_turn_returns_the_raw_answer_for_the_host_to_render() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v1/chat/completions"))
        .respond_with(probe_reply("**bold**"))
        .mount(&server)
        .await;

    let messages = msgs();
    let caveats = Caveats::top();
    let uri = server.uri();
    let mut ctx = ctx(&uri, &messages, &caveats);
    ctx.markdown = true;
    let (reply, streamed, _usage, _hallu) = chat_complete(ctx, &mut NoMcp)
        .await
        .expect("a markdown turn completes");

    assert_eq!(
        crate::agentic::model_explanation(&reply),
        "**bold**",
        "the transcript keeps the raw source"
    );
    assert!(!streamed, "the host renders it");
    assert_eq!(server.received_requests().await.unwrap().len(), 1);
}

/// A tool round and its answer are exactly two generations: nothing is
/// generated a second time for display.
#[tokio::test]
async fn a_tool_round_and_its_answer_are_two_generations() {
    let server = MockServer::start().await;
    let seen = Arc::new(Mutex::new(Vec::new()));
    Mock::given(method("POST"))
        .and(path("/v1/chat/completions"))
        .respond_with(ToolThenAnswer { seen: seen.clone() })
        .mount(&server)
        .await;

    let messages = msgs();
    let caveats = Caveats::top();
    let (reply, streamed, _usage, _hallu) =
        chat_complete(ctx(&server.uri(), &messages, &caveats), &mut NoMcp)
            .await
            .expect("tool round then answer");

    assert_eq!(
        crate::agentic::model_explanation(&reply),
        "answered after the tool"
    );
    assert!(!streamed);
    assert_eq!(seen.lock().unwrap().len(), 2, "one tool batch, one answer");
}

/// Assert the #2313 invariant for one OpenAI Chat turn: every received request
/// is on the generation path (so no token-count probe, `/v1/models`, or other
/// path was hit and the filter hides nothing), and the ledger's attempts are
/// exactly those requests, keyed by their bodies.
async fn assert_chat_attempts_equal_wire_requests(
    server: &MockServer,
    ledger: &std::sync::Mutex<crate::attempts::AttemptLedger>,
) -> usize {
    let received = server.received_requests().await.expect("journal");
    assert!(
        received
            .iter()
            .all(|request| request.url.path() == "/v1/chat/completions"),
        "only /v1/chat/completions may be hit (no token-count probe paths): {:?}",
        received.iter().map(|r| r.url.path()).collect::<Vec<_>>()
    );
    let ledger = ledger.lock().unwrap();
    let mut wire: Vec<_> = received
        .iter()
        .map(|request| content_addressable::RawContentId::from_content(&request.body))
        .collect();
    let mut recorded: Vec<_> = ledger.records().map(|record| record.key.request).collect();
    wire.sort();
    recorded.sort();
    assert_eq!(
        recorded, wire,
        "attempts == wire requests, keyed by their bodies"
    );
    for record in ledger.records() {
        assert!(
            record.key.turn.starts_with("prompt:"),
            "{}",
            record.key.turn
        );
        assert_eq!(record.key.role, "primary");
        assert_eq!(record.state, crate::attempts::AttemptState::Ok);
    }
    wire.len()
}

/// #2313 (b1b): every primary OpenAI Chat request — the tool round and the
/// accepted answer — is exactly one ledger attempt.
///
/// Scope of the count: the Chat primary loop's rounds and cap-exit summary.
#[tokio::test]
async fn every_openai_chat_request_is_one_ledger_attempt_keyed_by_its_wire_bytes() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v1/chat/completions"))
        .respond_with(ToolThenAnswer {
            seen: Arc::new(Mutex::new(Vec::new())),
        })
        .mount(&server)
        .await;
    let messages = msgs();
    let caveats = Caveats::top();
    let uri = server.uri();
    let ledger = std::sync::Mutex::new(crate::attempts::AttemptLedger::default());
    let mut c = ctx(&uri, &messages, &caveats);
    c.attempt_ledger = Some(&ledger);
    let (reply, streamed, _usage, _hallu) = chat_complete(c, &mut NoMcp)
        .await
        .expect("tool round then answer");
    assert_eq!(
        crate::agentic::model_explanation(&reply),
        "answered after the tool"
    );
    assert!(!streamed, "the host renders the accepted answer (#2372)");

    assert_eq!(
        assert_chat_attempts_equal_wire_requests(&server, &ledger).await,
        2,
        "tool round, accepted answer; no display reissue (#2372)"
    );
    let totals = ledger.lock().unwrap().totals();
    assert_eq!(
        (totals.in_tokens, totals.out_tokens, totals.usage_complete),
        (100 + 100, 4 + 7, true)
    );
}

/// Tool calls while tools are offered; an SSE summary once they are not.
struct ToolThenCapSummary;
impl Respond for ToolThenCapSummary {
    fn respond(&self, req: &Request) -> ResponseTemplate {
        if body_json(req).get("tools").is_some() {
            return ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "choices": [{"message": {
                    "content": null,
                    "tool_calls": [{
                        "id": "call_1",
                        "type": "function",
                        "function": {"name": "definitely_not_a_real_tool", "arguments": "{}"}
                    }]
                }}],
                "usage": {"prompt_tokens": 90, "completion_tokens": 3},
            }));
        }
        sse_text(&["capped summary"], 95, 6)
    }
}

/// #2313 (b1b): the OpenAI Chat cap-exit summary is one attempt too.
#[tokio::test]
async fn an_openai_chat_cap_exit_summary_is_one_ledger_attempt() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v1/chat/completions"))
        .respond_with(ToolThenCapSummary)
        .mount(&server)
        .await;
    let messages = msgs();
    let caveats = Caveats::top();
    let uri = server.uri();
    let ledger = std::sync::Mutex::new(crate::attempts::AttemptLedger::default());
    let mut c = ctx(&uri, &messages, &caveats);
    c.max_tool_rounds = 1;
    c.attempt_ledger = Some(&ledger);
    let (reply, _, _, _) = chat_complete(c, &mut NoMcp)
        .await
        .expect("the round cap ends the turn with a summary");
    assert!(reply.contains("capped summary"), "{reply}");

    let requests = assert_chat_attempts_equal_wire_requests(&server, &ledger).await;
    let summaries = server
        .received_requests()
        .await
        .expect("journal")
        .iter()
        .filter(|r| {
            serde_json::from_slice::<serde_json::Value>(&r.body)
                .is_ok_and(|body| body.get("tools").is_none())
        })
        .count();
    assert_eq!((requests, summaries), (2, 1), "one tool round, one summary");
}

struct ToolThenAnswer {
    seen: Arc<Mutex<Vec<serde_json::Value>>>,
}
impl Respond for ToolThenAnswer {
    fn respond(&self, req: &Request) -> ResponseTemplate {
        let body = body_json(req);
        let had_tool_result = body["messages"]
            .as_array()
            .map(|m| m.iter().any(|x| x["role"] == "tool"))
            .unwrap_or(false);
        self.seen.lock().unwrap().push(body);
        if had_tool_result {
            return probe_reply("answered after the tool");
        }
        ResponseTemplate::new(200).set_body_json(serde_json::json!({
            "choices": [{"message": {
                "content": null,
                "tool_calls": [{
                    "id": "call_1",
                    "type": "function",
                    "function": {"name": "definitely_not_a_real_tool", "arguments": "{}"}
                }]
            }}],
            "usage": {"prompt_tokens": 100, "completion_tokens": 4},
        }))
    }
}

/// #2372: the accepted Chat answer — the string that is returned, persisted
/// and re-sent — follows the reasoning policy the display stream's filter had.
/// An inline `<think>` block never reaches it; a lone `</think>` is answer text
/// unless the backend declares the #528 leading shape.
#[tokio::test]
async fn the_chat_answer_follows_the_declared_reasoning_policy() {
    let undeclared = "End the block with `</think>` and then answer.";
    for (content, leading, expected) in [
        ("<think>x</think>Done.", false, "Done."),
        ("<think>x</think>Done.", true, "Done."),
        (undeclared, false, undeclared),
        ("x</think>Done.", true, "Done."),
    ] {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/v1/chat/completions"))
            .respond_with(probe_reply(content))
            .mount(&server)
            .await;

        let messages = msgs();
        let caveats = Caveats::top();
        let uri = server.uri();
        let mut c = ctx(&uri, &messages, &caveats);
        c.emits_leading_reasoning = leading;
        let (reply, _streamed, _usage, _hallu) =
            chat_complete(c, &mut NoMcp).await.expect("dispatch");

        assert_eq!(
            crate::agentic::model_explanation(&reply),
            expected,
            "{content:?} declared={leading}"
        );
        assert_eq!(server.received_requests().await.unwrap().len(), 1);
    }
}

#[cfg(test)]
#[path = "openai_primary_stream.rs"]
mod primary_stream;

/// #2313 (c): Esc while a Chat round waits for its response drops the dispatch
/// future, so nothing can settle the attempt; the handle's drop records it
/// cancelled, with no usage.
#[tokio::test]
async fn an_interrupted_openai_chat_round_is_a_cancelled_attempt() {
    let flag = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let (url, server) =
        super::http_loop_tests::serve_until_interrupted("/v1/chat/completions", flag.clone()).await;
    let messages = msgs();
    let caveats = Caveats::top();
    let ledger = std::sync::Mutex::new(crate::attempts::AttemptLedger::default());
    let mut c = ctx(&url, &messages, &caveats);
    c.attempt_ledger = Some(&ledger);
    c.cancel = Some(flag.as_ref());
    let _ = tokio::time::timeout(
        super::http_loop_tests::RAW_STREAM_TEST_BOUND,
        chat_complete(c, &mut NoMcp),
    )
    .await
    .expect("the interrupt ends the turn");
    server.abort();
    let records: Vec<_> = ledger.lock().unwrap().records().cloned().collect();
    assert_eq!(records.len(), 1);
    assert_eq!(
        (records[0].state, records[0].usage),
        (crate::attempts::AttemptState::Cancelled, None)
    );
}

/// The primary rounds' attempts, by ordinal, after asserting they equal the
/// wire requests. Ordinal order is try order.
async fn chat_attempts_by_ordinal(
    server: &MockServer,
    ledger: &std::sync::Mutex<crate::attempts::AttemptLedger>,
) -> Vec<crate::attempts::AttemptRecord> {
    let received = server.received_requests().await.expect("journal");
    let mut records: Vec<_> = ledger.lock().unwrap().records().cloned().collect();
    assert_eq!(records.len(), received.len(), "attempts == wire requests");
    records.sort_by_key(|r| r.key.ordinal);
    records
}

/// #2313 (c): a retry storm on a Chat round. Every try is one attempt on the
/// same request bytes, ordinals 0, 1, 2: a 503 until the budget
/// (`NEWT_HTTP_MAX_RETRIES` = 2) is spent leaves three failed attempts and no
/// more; a 503 run that recovers on the last try ends ok with its usage.
#[tokio::test]
#[serial_test::serial(anthropic_loop_env)]
async fn a_chat_retry_storm_is_one_attempt_per_try() {
    use crate::attempts::AttemptState::{Failed, Ok};
    let _env = super::anthropic_loop_tests::test_env(false);
    let _retries = super::anthropic_loop_tests::EnvGuard::set("NEWT_HTTP_MAX_RETRIES", "2");
    for recovers in [false, true] {
        let server = MockServer::start().await;
        let unavailable = Mock::given(method("POST"))
            .and(path("/v1/chat/completions"))
            .respond_with(ResponseTemplate::new(503).set_body_string("busy"));
        if recovers {
            unavailable.up_to_n_times(2).mount(&server).await;
            Mock::given(method("POST"))
                .and(path("/v1/chat/completions"))
                .respond_with(sse_text(&["recovered"], 100, 7))
                .mount(&server)
                .await;
        } else {
            unavailable.mount(&server).await;
        }
        let messages = msgs();
        let caveats = Caveats::top();
        let uri = server.uri();
        let ledger = std::sync::Mutex::new(crate::attempts::AttemptLedger::default());
        let mut c = ctx(&uri, &messages, &caveats);
        c.attempt_ledger = Some(&ledger);
        let result = chat_complete(c, &mut NoMcp).await;
        assert_eq!(result.is_ok(), recovers, "recovers={recovers}");

        let records = chat_attempts_by_ordinal(&server, &ledger).await;
        // #2388 removed the display reissue: a recovered answer is the third
        // and last try, with no further attempt on the same bytes.
        let expected = if recovers {
            [Failed, Failed, Ok]
        } else {
            [Failed; 3]
        };
        assert_eq!(records.len(), 3, "recovers={recovers}");
        for (ordinal, (record, state)) in records.iter().zip(expected).enumerate() {
            assert_eq!(record.key.ordinal as usize, ordinal, "recovers={recovers}");
            assert_eq!(record.key.request, records[0].key.request, "the same bytes");
            assert_eq!(record.state, state, "recovers={recovers}, try {ordinal}");
            let usage = (state == Ok).then_some(crate::TokenUsage {
                input_tokens: 100,
                output_tokens: 7,
            });
            assert_eq!(record.usage, usage, "recovers={recovers}, try {ordinal}");
        }
    }
}

/// #2313 (c): a response whose identity changes mid-stream (#2334) earns one
/// fresh try, whatever the retry budget: exactly two attempts on the same bytes,
/// both failed with the usage each stream reported.
#[tokio::test]
#[serial_test::serial(anthropic_loop_env)]
async fn a_changed_response_identity_retries_once_as_one_more_attempt() {
    let _env = super::anthropic_loop_tests::test_env(false);
    let _retries = super::anthropic_loop_tests::EnvGuard::set("NEWT_HTTP_MAX_RETRIES", "4");
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v1/chat/completions"))
        .respond_with(sse(&[
            r#"{"id":"resp-one","choices":[{"delta":{"content":"a"}}]}"#,
            r#"{"id":"resp-two","choices":[{"delta":{},"finish_reason":"stop"}]}"#,
            REISSUE_USAGE,
            "[DONE]",
        ]))
        .mount(&server)
        .await;
    let messages = msgs();
    let caveats = Caveats::top();
    let uri = server.uri();
    let ledger = std::sync::Mutex::new(crate::attempts::AttemptLedger::default());
    let mut c = ctx(&uri, &messages, &caveats);
    c.attempt_ledger = Some(&ledger);
    let error = chat_complete(c, &mut NoMcp)
        .await
        .expect_err("a mixed response stays rejected after its one retry");
    assert!(format!("{error:#}").contains("stream changed"), "{error:#}");

    let records = chat_attempts_by_ordinal(&server, &ledger).await;
    assert_eq!(records.len(), 2, "one try and one fresh retry");
    for (ordinal, record) in records.iter().enumerate() {
        assert_eq!(record.key.ordinal as usize, ordinal);
        assert_eq!(record.key.request, records[0].key.request, "the same bytes");
        assert_eq!(record.state, crate::attempts::AttemptState::Failed);
        assert_eq!(record.usage, reissue_usage());
    }
}

// ---------------------------------------------------------------------------
// U3 review: an id-less tool call on the STREAMED chat wire is re-asked too.
// Headless streams, and the decoder used to abort on the first call with no id
// before the shared batch validator (and its bounded re-ask) was ever reached.
// ---------------------------------------------------------------------------

/// Serves one SSE body per request (the last repeats) and records every request.
struct StreamScript {
    bodies: Vec<String>,
    seen: std::sync::Arc<std::sync::Mutex<Vec<serde_json::Value>>>,
}
impl wiremock::Respond for StreamScript {
    fn respond(&self, req: &Request) -> ResponseTemplate {
        let mut seen = self.seen.lock().unwrap();
        let i = seen.len().min(self.bodies.len() - 1);
        seen.push(body_json(req));
        ResponseTemplate::new(200)
            .set_body_raw(self.bodies[i].clone().into_bytes(), "text/event-stream")
    }
}

fn sse_frames(frames: &[&str]) -> String {
    frames.iter().map(|f| format!("data: {f}\n\n")).collect()
}

/// A streamed `read_file` call; `id` is the raw JSON fragment for the id member
/// (empty = the key is absent).
fn sse_tool_call(id: &str) -> String {
    let call = format!(
        r#"{{"choices":[{{"delta":{{"tool_calls":[{{"index":0,{id}"type":"function","function":{{"name":"read_file","arguments":"{{\"path\":\"no/such/file\"}}"}}}}]}}}}]}}"#
    );
    sse_frames(&[
        &call,
        r#"{"choices":[{"delta":{},"finish_reason":"tool_calls"}]}"#,
        STREAM_USAGE,
        "[DONE]",
    ])
}

fn sse_answer() -> String {
    sse_frames(&[
        r#"{"choices":[{"delta":{"content":"all done"}}]}"#,
        r#"{"choices":[{"delta":{},"finish_reason":"stop"}]}"#,
        STREAM_USAGE,
        "[DONE]",
    ])
}

/// Runs the streamed loop against `script`; returns the result, the request
/// bodies, and the tool events.
async fn run_streamed(
    script: Vec<String>,
) -> (
    anyhow::Result<String>,
    Vec<serde_json::Value>,
    Vec<crate::ToolEvent>,
) {
    let server = MockServer::start().await;
    let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
    Mock::given(method("POST"))
        .and(path("/v1/chat/completions"))
        .respond_with(StreamScript {
            bodies: script,
            seen: seen.clone(),
        })
        .mount(&server)
        .await;
    let messages = msgs();
    let caveats = Caveats::top();
    let uri = server.uri();
    let mut events: Vec<crate::ToolEvent> = Vec::new();
    let mut c = ctx(&uri, &messages, &caveats);
    c.tool_events = Some(&mut events);
    let result = chat_complete(c, &mut NoMcp).await.map(|(reply, ..)| reply);
    let bodies = seen.lock().unwrap().clone();
    (result, bodies, events)
}

#[tokio::test]
async fn streamed_idless_tool_call_is_re_asked_and_the_turn_completes() {
    let (result, bodies, events) = run_streamed(vec![
        sse_tool_call(""),
        sse_tool_call(r#""id":"call_1","#),
        sse_answer(),
    ])
    .await;
    assert_eq!(
        crate::agentic::model_explanation(&(result.expect("the turn completes"))),
        "all done"
    );
    assert_eq!(bodies.len(), 3, "re-ask, tool result, final");
    let second = bodies[1]["messages"].as_array().expect("messages");
    let last = second.last().unwrap();
    assert_eq!(last["role"], "user");
    assert!(
        last["content"].as_str().unwrap().contains("call id"),
        "{last}"
    );
    assert_eq!(events.iter().filter(|e| e.tool == "read_file").count(), 1);
}

/// Three id-less streamed batches abort, with the shared helper's error.
#[tokio::test]
async fn three_streamed_idless_batches_abort_with_the_shared_error() {
    let (result, bodies, _) = run_streamed(vec![sse_tool_call("")]).await;
    let err = result.expect_err("the budget is spent");
    assert!(
        err.to_string().contains("malformed provider output"),
        "{err:#}"
    );
    assert_eq!(bodies.len(), 3);
}

/// Two streamed calls sharing one id are the validator's to judge as well.
#[tokio::test]
async fn streamed_duplicate_ids_reach_the_shared_validator() {
    let dup = sse_frames(&[
        r#"{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"dup","type":"function","function":{"name":"read_file","arguments":"{\"path\":\"a\"}"}},{"index":1,"id":"dup","type":"function","function":{"name":"read_file","arguments":"{\"path\":\"b\"}"}}]}}]}"#,
        r#"{"choices":[{"delta":{},"finish_reason":"tool_calls"}]}"#,
        STREAM_USAGE,
        "[DONE]",
    ]);
    let (result, bodies, _) = run_streamed(vec![dup, sse_answer()]).await;
    assert_eq!(
        crate::agentic::model_explanation(&(result.expect("re-asked then answered"))),
        "all done"
    );
    let last = bodies[1]["messages"]
        .as_array()
        .unwrap()
        .last()
        .unwrap()
        .clone();
    assert!(
        last["content"]
            .as_str()
            .unwrap()
            .contains("repeats call id"),
        "the validator's reason reaches the model: {last}"
    );
}

/// The invariant the decoder's docs name: a decoded id-less call can never be
/// DISPATCHED (not merely decoded). The tool result is what would have run.
#[tokio::test]
async fn a_decoded_idless_streamed_call_is_never_dispatched() {
    let (result, _, events) = run_streamed(vec![sse_tool_call(""), sse_answer()]).await;
    assert_eq!(
        crate::agentic::model_explanation(&(result.expect("re-asked, then answered"))),
        "all done"
    );
    assert!(
        events
            .iter()
            .all(|e| e.tool == "(rejected tool-call batch)" && !e.ok),
        "no tool may run for an id-less call: {events:?}"
    );
    assert!(events.iter().all(|e| e.tool != "read_file"));
}