newt-core 0.8.0

//! Completion evidence for the strict consumer of the shared SSE parser.

use super::decode_response;
use serde_json::{json, Value};

fn sse(frames: &[Value], done: bool) -> Vec<u8> {
    let mut body = frames
        .iter()
        .map(|frame| format!("data: {frame}\n\n"))
        .collect::<String>();
    if done {
        body.push_str("data: [DONE]\n\n");
    }
    body.into_bytes()
}

#[test]
fn stream_assembles_indexed_tool_fragments_and_preserves_raw_usage() {
    let bytes = sse(
        &[
            json!({"id":"response-fixture","model":"served-model","choices":[{"index":0,"delta":{
            "role":"assistant","reasoning_content":"separate reasoning",
            "tool_calls":[
                {"index":1,"id":"call-b","type":"function","function":{"name":"read_file","arguments":"{\"path\":"}},
                {"index":0,"id":"call-a","type":"function","function":{"name":"run_command","arguments":"{\"command\":"}}
            ]}}]}),
            json!({"choices":[{"index":0,"delta":{"tool_calls":[
                {"index":0,"function":{"arguments":"\"printf café\"}"}},
                {"index":1,"function":{"arguments":"\"fixture.rs\"}"}}
            ]}}]}),
            json!({"choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}),
            json!({"choices":[],"usage":{"prompt_tokens":6000,"prompt_tokens_details":{"cached_tokens":1000}}}),
        ],
        true,
    );
    let parsed = decode_response(&bytes).unwrap();
    let message = &parsed["choices"][0]["message"];
    assert_eq!(parsed["id"], "response-fixture");
    assert_eq!(parsed["model"], "served-model");
    assert_eq!(parsed["choices"][0]["finish_reason"], "tool_calls");
    assert_eq!(message["reasoning_content"], "separate reasoning");
    assert!(!message["content"]
        .as_str()
        .unwrap_or_default()
        .contains("reasoning"));
    assert_eq!(message["tool_calls"][0]["id"], "call-a");
    assert_eq!(message["tool_calls"][1]["id"], "call-b");
    assert_eq!(
        message["tool_calls"][0]["function"]["arguments"],
        "{\"command\":\"printf café\"}"
    );
    assert_eq!(
        message["tool_calls"][1]["function"]["arguments"],
        "{\"path\":\"fixture.rs\"}"
    );
    assert_eq!(
        parsed["usage"],
        json!({"prompt_tokens":6000,"prompt_tokens_details":{"cached_tokens":1000}})
    );
    assert!(parsed["usage"].get("completion_tokens").is_none());
}

#[test]
fn stream_accepts_omitted_and_blank_zero_argument_tool_calls() {
    for raw_arguments in [
        None,
        Some(Value::Null),
        Some(json!("")),
        Some(json!("  \t")),
    ] {
        let mut function = json!({"name":"get_context_remaining"});
        if let Some(raw_arguments) = raw_arguments {
            function["arguments"] = raw_arguments;
        }
        let response = decode_response(&sse(
            &[
                json!({"choices":[{"delta":{"tool_calls":[{
                    "index":0,"id":"call-no-args","type":"function","function":function
                }]}}]}),
                json!({"choices":[{"delta":{},"finish_reason":"tool_calls"}]}),
            ],
            true,
        ))
        .expect("the shared tool gate accepts an empty argument object");
        assert_eq!(
            response["choices"][0]["message"]["tool_calls"][0]["function"]["arguments"],
            "{}"
        );
    }
}

#[test]
fn stream_refuses_a_cut_tool_batch() {
    let bytes = sse(
        &[
            json!({"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call-a","type":"function","function":{"name":"run_command","arguments":"{\"command\":\"echo fixture\"}"}}]}}]}),
            json!({"choices":[{"delta":{},"finish_reason":"tool_calls"}]}),
        ],
        false,
    );
    assert!(
        decode_response(&bytes).is_err(),
        "an incomplete tool batch cannot authorize execution"
    );
}

#[test]
fn stream_refuses_a_changed_call_identity_and_malformed_data_frame() {
    let changed = sse(
        &[
            json!({"choices":[{"delta":{"tool_calls":[{"index":0,"id":"first","type":"function","function":{"name":"read_file","arguments":"{"}}]}}]}),
            json!({"choices":[{"delta":{"tool_calls":[{"index":0,"id":"replacement","function":{"arguments":"}"}}]},"finish_reason":"tool_calls"}]}),
        ],
        true,
    );
    assert!(decode_response(&changed).is_err());
    let malformed = b"data: {broken JSON\n\ndata: {\"choices\":[{\"delta\":{\"content\":\"later text\"},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n";
    assert!(decode_response(malformed).is_err());
}

#[test]
fn stream_error_retains_context_evidence_but_quoted_error_text_is_normal_content() {
    let error = sse(
        &[json!({"error":{"message":"Context size has been exceeded."}})],
        false,
    );
    let error = decode_response(&error).unwrap_err();
    assert_eq!(
        crate::retry::classify(&error),
        crate::retry::Retryability::ContextExceeded
    );
    let answer = sse(
        &[
            json!({"choices":[{"delta":{"content":"The log said Context size has been exceeded."},"finish_reason":"stop"}]}),
        ],
        true,
    );
    assert_eq!(
        decode_response(&answer).unwrap()["choices"][0]["message"]["content"],
        "The log said Context size has been exceeded."
    );
}

#[test]
fn complete_json_response_is_a_single_response_fallback() {
    let expected = json!({"model":"json-server","choices":[{"message":{"role":"assistant","content":"<think>reasoning</think>answer"},"finish_reason":"stop"}],"usage":{"prompt_tokens":150,"completion_tokens":2}});
    assert_eq!(
        decode_response(&serde_json::to_vec(&expected).unwrap()).unwrap(),
        expected
    );
}

#[test]
fn stream_preserves_reasoning_only_length_for_bounded_continuation() {
    let bytes = sse(
        &[json!({"choices":[{"delta":{"reasoning":"still thinking"},
        "finish_reason":"length"}]})],
        true,
    );
    let response = decode_response(&bytes).unwrap();
    assert_eq!(response["choices"][0]["finish_reason"], "length");
    assert_eq!(
        response["choices"][0]["message"]["reasoning_content"],
        "still thinking"
    );
    assert_eq!(response["choices"][0]["message"]["content"], "");
    assert!(response.get("usage").is_none());
    assert!(response.get("model").is_none());
    assert!(response.get("id").is_none());
}

#[test]
fn stream_refuses_unfinished_text_and_incomplete_tool_identity() {
    for frames in [
        vec![json!({"choices":[{"delta":{"content":"answer"}}]})],
        vec![json!({"choices":[],"usage":{"prompt_tokens":1}})],
        vec![
            json!({"choices":[{"delta":{"tool_calls":[{"index":0,"id":"one","function":{
            "name":"read_file","arguments":"{}"}}]},"finish_reason":"stop"}]}),
        ],
    ] {
        assert!(decode_response(&sse(&frames, true)).is_err());
    }
    assert!(decode_response(&sse(
        &[json!({"choices":[{"delta":{"content":"answer"},
        "finish_reason":"stop"}]})],
        false
    ))
    .is_err());
}

#[test]
fn stream_refuses_duplicate_ids_index_gaps_and_nonzero_choices() {
    for calls in [
        json!([{"index":1,"id":"one","function":{"name":"read_file","arguments":"{}"}}]),
        json!([{"id":"one","function":{"name":"read_file","arguments":"{}"}}]),
    ] {
        assert!(decode_response(&sse(
            &[json!({"choices":[{"delta":{"tool_calls":calls},
            "finish_reason":"tool_calls"}]})],
            true
        ))
        .is_err());
    }
    assert!(decode_response(&sse(
        &[json!({"choices":[{"index":1,"delta":{"content":"answer"},
        "finish_reason":"stop"}]})],
        true
    ))
    .is_err());
}

#[test]
fn stream_refuses_identity_drift_and_post_finish_or_done_mutations() {
    for field in ["id", "model"] {
        let mut first = json!({"choices":[{"delta":{"content":"a"}}]});
        let mut last = json!({"choices":[{"delta":{},"finish_reason":"stop"}]});
        first[field] = json!("first");
        last[field] = json!("replacement");
        assert!(decode_response(&sse(&[first, last], true)).is_err());
    }
    let finish = json!({"choices":[{"delta":{},"finish_reason":"stop"}]});
    let later = json!({"choices":[{"delta":{"content":"after completion"}}]});
    assert!(decode_response(&sse(&[finish.clone(), later.clone()], true)).is_err());
    let mut bytes = sse(&[finish], true);
    bytes.extend(sse(&[later], false));
    assert!(decode_response(&bytes).is_err());
}

#[test]
fn stream_error_event_and_transient_status_remain_distinct_from_quoted_reasoning() {
    let error = decode_response(
        b"event: error\ndata: {\"message\":\"Context size has been exceeded\"}\n\n",
    )
    .unwrap_err();
    assert_eq!(
        crate::retry::classify(&error),
        crate::retry::Retryability::ContextExceeded
    );
    let error = decode_response(&sse(
        &[json!({"error":{"message":"busy","code":503}})],
        false,
    ))
    .unwrap_err();
    assert!(super::is_provider_error(&error));
    assert_eq!(
        crate::retry::classify(&error),
        crate::retry::Retryability::Retry
    );
    let response = decode_response(&sse(&[json!({"choices":[{"delta":{
        "reasoning_content":"Explain the phrase Context size has been exceeded.","content":"answer"},
        "finish_reason":"stop"}]})], true)).unwrap();
    assert_eq!(response["choices"][0]["message"]["content"], "answer");
}

#[test]
fn stream_provider_error_marker_does_not_accept_parser_failures() {
    let error = decode_response(b"data: {invalid JSON\n\n").unwrap_err();
    assert!(!super::is_provider_error(&error));
    let error = decode_response(&sse(
        &[json!({"error":{"message":"busy","code":503}})],
        false,
    ))
    .unwrap_err()
    .context("body read failed afterward");
    assert!(super::is_provider_error(&error));
}

#[test]
fn complete_json_error_envelope_is_not_a_successful_json_fallback() {
    let error =
        decode_response(br#"{"error":{"message":"Context size has been exceeded"}}"#).unwrap_err();
    assert_eq!(
        crate::retry::classify(&error),
        crate::retry::Retryability::ContextExceeded
    );
}

#[test]
fn stream_invalid_utf8_refuses_completion_but_keeps_an_observed_error() {
    let mut bytes = sse(
        &[json!({"choices":[{"delta":{"content":"answer"},
        "finish_reason":"stop"}]})],
        true,
    );
    bytes.push(0xff);
    assert!(decode_response(&bytes).is_err());
    let mut bytes = sse(
        &[json!({"error":{"message":"Context size has been exceeded"}})],
        false,
    );
    bytes.push(0xff);
    let error = decode_response(&bytes).unwrap_err();
    assert_eq!(
        crate::retry::classify(&error),
        crate::retry::Retryability::ContextExceeded
    );
    let mut bytes = b"data: {broken JSON\n\n".to_vec();
    bytes.extend(sse(
        &[json!({"error":{"message":"Context size has been exceeded"}})],
        false,
    ));
    let error = decode_response(&bytes).unwrap_err();
    assert_eq!(
        crate::retry::classify(&error),
        crate::retry::Retryability::ContextExceeded
    );
}

/// #2334: response identity is checked per response. Stable and omitted IDs
/// are one response; a changed ID or model mixes two and stays rejected, but as
/// a recoverable failure whose text carries only a frame ordinal and lengths.
#[test]
fn response_identity_is_stable_omittable_and_a_change_is_recoverable_and_redacted() {
    let answer = json!({"choices":[{"delta":{"content":"ok"},"finish_reason":"stop"}]});
    let stable = json!({"id":"resp-fixture-one","model":"served-model","choices":[]});
    let omitted = decode_response(&sse(std::slice::from_ref(&answer), true)).unwrap();
    assert!(omitted.get("id").is_none(), "an omitted ID is not invented");
    let stable = decode_response(&sse(&[stable, answer.clone()], true)).unwrap();
    assert_eq!(stable["id"], "resp-fixture-one");
    for (key, old, new) in [
        ("id", "resp-fixture-one", "resp-fixture-twenty"),
        ("model", "served-model", "other-served-model"),
    ] {
        let mut changed = answer.clone();
        changed[key] = json!(new);
        let error = decode_response(&sse(&[json!({key: old, "choices":[]}), changed], true))
            .expect_err("mixing two responses stays rejected");
        let text = format!("{error:#}");
        assert!(
            text.contains(&format!(
                "at data frame 2 ({}-byte value became {}-byte value)",
                old.len(),
                new.len()
            )),
            "{text}"
        );
        assert!(!text.contains(old) && !text.contains(new), "{text}");
        assert_ne!(
            crate::retry::classify(&error),
            crate::retry::Retryability::Fatal,
            "a mixed response is a recoverable transport failure: {text}"
        );
    }
}

/// Twin: only response identity is recoverable. A tool call whose own ID changes
/// is a malformed batch and remains fatal.
#[test]
fn a_changed_tool_call_id_remains_fatal() {
    let changed = sse(
        &[
            json!({"id":"resp-fixture","choices":[{"delta":{"tool_calls":[{"index":0,"id":"first","type":"function","function":{"name":"read_file","arguments":"{"}}]}}]}),
            json!({"id":"resp-fixture","choices":[{"delta":{"tool_calls":[{"index":0,"id":"replacement","function":{"arguments":"}"}}]},"finish_reason":"tool_calls"}]}),
        ],
        true,
    );
    let error = decode_response(&changed).unwrap_err();
    assert_eq!(
        crate::retry::classify(&error),
        crate::retry::Retryability::Fatal,
        "{error:#}"
    );
}

/// #2318: a strict-decode rejection is classified by its cause, not by the
/// wrapper it rides in. A model-output defect is `model_error`; a body cut on a
/// clean EOF (no read error for the client to see) is `transport_error`, even
/// when the cut tore the last frame or character; a provider's own error follows
/// its text; a shape this decoder does not support is ours, `harness_error`.
#[test]
fn strict_rejections_classify_by_cause() {
    use crate::agentic::observability::{error_class, ErrorClass};
    let frame = |value: Value| sse(&[value], false);
    let call = |call: Value, finish: &str| {
        sse(
            &[json!({"choices":[{"delta":{"tool_calls":[call]},"finish_reason":finish}]})],
            true,
        )
    };
    let mut malformed_then_done = b"data: {broken JSON\n\n".to_vec();
    malformed_then_done.extend(sse(
        &[json!({"choices":[{"delta":{"content":"a"},"finish_reason":"stop"}]})],
        true,
    ));
    let mut changed_identity = sse(&[json!({"id":"resp-one","choices":[]})], false);
    changed_identity.extend(sse(
        &[json!({"id":"resp-two","choices":[{"delta":{"content":"a"},"finish_reason":"stop"}]})],
        true,
    ));
    // No [DONE], but a complete line is already bad: the model's defect wins
    // over the cut.
    let mut malformed_then_cut = b"data: {broken JSON\n\n".to_vec();
    malformed_then_cut.extend(frame(json!({"choices":[{"delta":{"content":"a"}}]})));
    // An invalid byte (not a torn character) ends the readable body early;
    // that is bad output, not a cut.
    let mut invalid_byte = frame(json!({"choices":[{"delta":{"content":"a"}}]}));
    invalid_byte.push(0xff);
    invalid_byte.extend(sse(
        &[json!({"choices":[{"delta":{"content":"b"},"finish_reason":"stop"}]})],
        true,
    ));
    let cases: Vec<(&str, Vec<u8>, ErrorClass)> = vec![
        (
            "non-string tool-call id",
            call(
                json!({"index":0,"id":7,"function":{"name":"read_file","arguments":"{}"}}),
                "tool_calls",
            ),
            ErrorClass::Model,
        ),
        (
            "malformed frame, then a clean EOF",
            malformed_then_cut,
            ErrorClass::Model,
        ),
        ("invalid byte mid-stream", invalid_byte, ErrorClass::Model),
        (
            "tool call index gap",
            call(
                json!({"index":1,"id":"a","function":{"name":"read_file","arguments":"{}"}}),
                "tool_calls",
            ),
            ErrorClass::Model,
        ),
        (
            "batch ending neither tool_calls nor length",
            call(
                json!({"index":0,"id":"a","function":{"name":"read_file","arguments":"{}"}}),
                "stop",
            ),
            ErrorClass::Model,
        ),
        (
            "choice changed after its finish reason",
            sse(
                &[
                    json!({"choices":[{"delta":{},"finish_reason":"stop"}]}),
                    json!({"choices":[{"delta":{"content":"after"}}]}),
                ],
                true,
            ),
            ErrorClass::Model,
        ),
        (
            "malformed frame, then [DONE]",
            malformed_then_done,
            ErrorClass::Model,
        ),
        (
            "clean EOF between frames",
            frame(json!({"choices":[{"delta":{"content":"a"},"finish_reason":"stop"}]})),
            ErrorClass::Transport,
        ),
        (
            "clean EOF inside a frame",
            b"data: {\"choices\":[{\"delta\":{\"con".to_vec(),
            ErrorClass::Transport,
        ),
        (
            "clean EOF inside a character",
            "data: {\"choices\":[{\"delta\":{\"content\":\"🦎".as_bytes()[..41].to_vec(),
            ErrorClass::Transport,
        ),
        (
            "response identity changed (#2334)",
            changed_identity,
            ErrorClass::Transport,
        ),
        (
            "provider context error",
            frame(json!({"error":{"message":"Context size has been exceeded."}})),
            ErrorClass::ContextExceeded,
        ),
        (
            "provider busy error",
            frame(json!({"error":{"message":"busy","code":503}})),
            ErrorClass::Model,
        ),
        (
            "content as an array of parts",
            sse(
                &[
                    json!({"choices":[{"delta":{"content":[{"type":"text","text":"a"}]},"finish_reason":"stop"}]}),
                ],
                true,
            ),
            ErrorClass::Harness,
        ),
        (
            "empty-string response id",
            sse(
                &[json!({"id":"","choices":[{"delta":{"content":"a"},"finish_reason":"stop"}]})],
                true,
            ),
            ErrorClass::Harness,
        ),
        (
            "empty-string response model",
            sse(
                &[json!({"model":"","choices":[{"delta":{"content":"a"},"finish_reason":"stop"}]})],
                true,
            ),
            ErrorClass::Harness,
        ),
        (
            "chunk with no choices array",
            sse(&[json!({"object":"chat.completion.chunk"})], true),
            ErrorClass::Harness,
        ),
    ];
    for (name, bytes, class) in cases {
        let error = crate::agentic::smart_harness::decode_openai_response(&bytes).expect_err(name);
        assert_eq!(error_class(&error), Some(class), "{name}: {error:#}");
    }
}

/// #2385: argument and name validity is the loop's to judge
/// (`validate_tool_call_batch`), as on every other wire. A complete stream
/// whose call has cut or non-object arguments, or no name, decodes with that
/// call's raw fields, and so does a batch cut at the output limit
/// (`finish_reason: length`). The loop stops a length-terminated response
/// before validation (#2824); otherwise it may reject the batch and retry. A valid call is still normalized here exactly as the loop would.
#[test]
fn stream_leaves_invalid_tool_calls_to_the_batch_validator() {
    let call = |function: Value, finish: &str| {
        sse(
            &[json!({"choices":[{"delta":{"tool_calls":[
                {"index":0,"id":"one","type":"function","function":function}]},
                "finish_reason":finish}]})],
            true,
        )
    };
    for (function, finish) in [
        (
            json!({"name":"run_command","arguments":"{\"command\":"}),
            "tool_calls",
        ),
        (
            json!({"name":"run_command","arguments":"{\"command\":"}),
            "length",
        ),
        (json!({"name":"read_file","arguments":"[]"}), "tool_calls"),
        (json!({"arguments":"{}"}), "tool_calls"),
        (json!({"name":"read_file","arguments":"{}"}), "length"),
    ] {
        let label = format!("{function} finish={finish}");
        let response = decode_response(&call(function.clone(), finish))
            .unwrap_or_else(|error| panic!("{label}: decodes: {error:#}"));
        let decoded = &response["choices"][0]["message"]["tool_calls"][0];
        assert_eq!(response["choices"][0]["finish_reason"], finish, "{label}");
        assert_eq!(decoded["id"], "one", "{label}");
        let name = decoded["function"]["name"].as_str();
        let arguments = &decoded["function"]["arguments"];
        let batch = crate::agentic::tools::validate_tool_call_batch(
            &[(Some("one"), name, arguments)],
            true,
        );
        match crate::agentic::tools::validate_tool_call(
            function["name"].as_str(),
            &function["arguments"],
        ) {
            Ok((name, args)) => {
                assert_eq!(
                    decoded["function"],
                    json!({"name": name, "arguments": args.to_string()}),
                    "{label}"
                );
                assert!(batch.is_ok(), "{label}");
            }
            Err(_) => assert!(
                matches!(
                    batch,
                    Err(crate::agentic::tools::BatchRejection::ContentInvalid(_))
                ),
                "{label}: the loop rejects it per call"
            ),
        }
    }
}

/// U3 review: a missing, blank or repeated tool-call id is the batch validator's
/// to judge (a bounded re-ask), so the strict decoder passes it through rather
/// than aborting the turn before the validator is reached. Only what cannot be
/// read at all (a non-string id) still fails here.
#[test]
fn stream_leaves_missing_blank_and_repeated_ids_to_the_batch_validator() {
    let decode = |calls: Value| {
        decode_response(&sse(
            &[json!({"choices":[{"delta":{"tool_calls":calls},"finish_reason":"tool_calls"}]})],
            true,
        ))
        .expect("decodes; the validator judges the ids")
    };
    let f = || json!({"name":"read_file","arguments":"{}"});
    let missing = decode(json!([{"index":0,"function":f()}]));
    assert!(missing["choices"][0]["message"]["tool_calls"][0]
        .get("id")
        .is_none());
    let blank = decode(json!([{"index":0,"id":"","function":f()}]));
    assert!(blank["choices"][0]["message"]["tool_calls"][0]
        .get("id")
        .is_none());
    let repeated = decode(json!([{"index":0,"id":"one","function":f()},
        {"index":1,"id":"one","function":f()}]));
    let calls = &repeated["choices"][0]["message"]["tool_calls"];
    assert_eq!(calls[0]["id"], "one");
    assert_eq!(calls[1]["id"], "one");
}

/// A blank id on a LATER fragment, after the call already carried a real id, is
/// still a strict rejection (the id changed): only a call that NEVER carries an
/// id reaches the batch validator.
#[test]
fn stream_still_rejects_a_blank_id_after_a_real_one() {
    let frames = [
        json!({"choices":[{"delta":{"tool_calls":[{"index":0,"id":"one","function":{
            "name":"read_file","arguments":"{"}}]}}]}),
        json!({"choices":[{"delta":{"tool_calls":[{"index":0,"id":"","function":{
            "arguments":"}"}}]},"finish_reason":"tool_calls"}]}),
    ];
    assert!(decode_response(&sse(&frames, true)).is_err());
}