rootle-stdio 0.10.0

//! Fake-provider integration tests: this test binary re-executed as
//! the child. No python/sh dependency (the docker `test` stage has
//! neither): the parent spawns current_exe with a filter that runs
//! ONLY the `fake_provider_child` test, which loops on stdin replying
//! per ROOTLE_FAKE_PROVIDER's script until killed.

use super::StdioProvider;
use rootle_provider::{ErrorKind, Provider, ProviderError};
use serde_json::json;
use std::io::{BufRead, Write};
use std::sync::Arc;
use std::time::Duration;

#[test]
fn timeout_fails_request_and_transport_recovers() {
    let provider = fake("hang-on-2", Duration::from_millis(300));
    let start = std::time::Instant::now();
    let err = provider
        .request("repo/tree", json!({ "repo": "o/r" }))
        .expect_err("id 2 must time out");
    assert_eq!(err.kind, ErrorKind::Timeout);
    assert!(
        err.message.contains("timeout"),
        "expected a timeout message, got: {err}"
    );
    assert!(
        start.elapsed() < Duration::from_secs(5),
        "timeout must fire near the deadline, took {:?}",
        start.elapsed()
    );
    // The transport is unwedged: the next request (id 3) succeeds
    // even though id 2's reply never arrives.
    provider
        .request("org/repos", json!({ "org": "o" }))
        .expect("transport must recover after a timed-out request");
}

#[test]
fn child_death_fails_pending_immediately() {
    let provider = fake("die-on-2", Duration::from_secs(30));
    let start = std::time::Instant::now();
    let err = provider
        .request("repo/tree", json!({ "repo": "o/r" }))
        .expect_err("dead child must fail the request");
    assert_eq!(
        err,
        ProviderError::new(ErrorKind::Provider, "provider closed its output")
    );
    assert!(
        start.elapsed() < Duration::from_secs(5),
        "EOF must fail fast, not at the deadline; took {:?}",
        start.elapsed()
    );
}

#[test]
fn slow_replies_within_the_deadline_succeed() {
    let provider = fake("slow", Duration::from_secs(2));
    provider
        .request("org/repos", json!({ "org": "o" }))
        .expect("200ms reply must land within a 2s deadline");
}

/// v1.4: `org/repos` entries are the bare name or a metadata object
/// — both forms parse, the object form carries its fields through.
#[test]
fn org_repos_accepts_the_v14_union() {
    let provider = fake("rich-repos", Duration::from_secs(2));
    let repos = provider.org_repos("o").expect("org/repos succeeds");
    assert_eq!(repos.len(), 2);
    assert_eq!(repos[0], rootle_provider::RepoInfo::bare("plain"));
    let meta = &repos[1];
    assert_eq!(meta.name, "meta");
    assert_eq!(meta.description.as_deref(), Some("d"));
    assert!(meta.private && meta.archived);
    assert_eq!(meta.pushed_at.as_deref(), Some("2026-08-20T10:11:12Z"));
}

#[test]
fn error_data_kind_parses_into_the_taxonomy() {
    let provider = fake("error-kinds", Duration::from_secs(2));

    let err = provider.request("a", json!({})).expect_err("auth error");
    assert_eq!(err.kind, ErrorKind::Auth);
    assert_eq!(err.message, "auth happened");
    assert_eq!(err.retry_after, None);

    let err = provider.request("b", json!({})).expect_err("rate limited");
    assert_eq!(err.kind, ErrorKind::RateLimited);
    assert_eq!(err.retry_after, Some(Duration::from_secs(37)));

    // Unknown kinds degrade to Other; absent data does too.
    let err = provider.request("c", json!({})).expect_err("unknown kind");
    assert_eq!(err.kind, ErrorKind::Other);
    let err = provider.request("d", json!({})).expect_err("kindless");
    assert_eq!(err.kind, ErrorKind::Other);
}

/// v1.3 (plans/0011): `$/partial` batches arrive, in order, strictly
/// before the metadata-only reply; `truncated` rides the reply.
#[test]
fn streaming_search_delivers_ordered_batches_then_metadata() {
    let provider = fake("stream-search", Duration::from_secs(5));
    let batches = std::sync::Mutex::new(Vec::new());
    let result = provider
        .search_code_progressive("hit", &|items: &[rootle_provider::CodeMatch]| {
            batches.lock().unwrap().push(
                items
                    .iter()
                    .map(|m| (m.path.clone(), m.line))
                    .collect::<Vec<(String, Option<u32>)>>(),
            );
        })
        .unwrap();
    assert_eq!(
        batches.into_inner().unwrap(),
        vec![
            vec![("f1.rs".to_string(), Some(1))],
            vec![("f2.rs".to_string(), Some(2))],
        ],
        "ordered batches, provider-known lines (v1.3)"
    );
    assert!(result.hits.is_empty(), "streamed final is metadata-only");
    assert!(result.truncated);
}

/// Child death mid-stream: partials already delivered stay with the
/// caller, and the request itself fails.
#[test]
fn child_death_mid_stream_keeps_partials_and_fails() {
    let provider = fake("die-mid-stream", Duration::from_secs(5));
    let seen = std::sync::atomic::AtomicUsize::new(0);
    let err = provider
        .search_code_progressive("hit", &|_| {
            seen.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        })
        .unwrap_err();
    assert_eq!(
        seen.load(std::sync::atomic::Ordering::Relaxed),
        1,
        "the first batch should have been delivered"
    );
    assert_eq!(err.kind, ErrorKind::Provider);
}

/// Every `$/partial` resets the read deadline (inactivity semantics):
/// a stream whose total exceeds the deadline succeeds while no single
/// gap does.
#[test]
fn partials_reset_the_inactivity_deadline() {
    let provider = fake("stream-slow", Duration::from_millis(400));
    let batches = std::sync::atomic::AtomicUsize::new(0);
    provider
        .search_code_progressive("hit", &|_| {
            batches.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        })
        .expect("stream should outlive the deadline via resets");
    assert_eq!(batches.load(std::sync::atomic::Ordering::Relaxed), 3);
}

/// S2, now load-bearing: two requests in flight at once, replies
/// arriving OUT of request order (id 3 answers before the stashed
/// id 2). The per-id reply slots must route each reply to its own
/// caller — B completes while A is still waiting.
#[test]
fn concurrent_requests_route_out_of_order_replies() {
    let provider = std::sync::Arc::new(fake("swap-2-3", Duration::from_secs(5)));
    let (tx, rx) = std::sync::mpsc::channel::<(&'static str, std::time::Instant)>();

    let a = {
        let provider = Arc::clone(&provider);
        let tx = tx.clone();
        std::thread::spawn(move || {
            provider
                .request("a/first", json!({}))
                .expect("request A (stashed reply) must still be answered");
            tx.send(("A", std::time::Instant::now())).unwrap();
        })
    };
    // Gate B on A's request actually being in flight — ids are
    // assigned in arrival order, and the fake stashes id 2. Without
    // this the two thread-starts race and the assertion inverts
    // under load.
    while provider
        .current_id
        .load(std::sync::atomic::Ordering::Acquire)
        == 0
    {
        std::thread::yield_now();
    }
    let b = {
        let provider = Arc::clone(&provider);
        std::thread::spawn(move || {
            provider
                .request("b/second", json!({}))
                .expect("request B must be answered immediately");
            tx.send(("B", std::time::Instant::now())).unwrap();
        })
    };
    a.join().expect("A panics");
    b.join().expect("B panics");

    let order: Vec<&str> = rx.iter().map(|(who, _)| who).collect();
    assert_eq!(
        order,
        vec!["B", "A"],
        "id-routed slots must deliver the fast reply (B) before the delayed one (A)"
    );
}

// -- 0027 bridge tests: every fault class the TLA model names, as a
// -- wire event against the real transport ------------------------------

/// UniqueTerminal/CorrelationSafety: a duplicate final for an already
/// answered id is dropped — the caller keeps the FIRST result, the
/// second is never applied, and the transport stays usable.
#[test]
fn duplicate_final_reply_is_dropped_not_applied() {
    let provider = fake("double-final", Duration::from_secs(5));
    let first = provider
        .request("repo/tree", json!({ "repo": "o/r" }))
        .expect("the first final must complete the request");
    assert_eq!(first["which"], 1, "the first reply wins");
    // The duplicate landed after the slot was removed; the follow-up
    // request proves the reader survived it.
    let next = provider
        .request("org/repos", json!({ "org": "o" }))
        .expect("a duplicate final must not wedge the reader");
    assert!(
        next.get("which").is_none(),
        "id 3 answers via the default arm"
    );
}

/// PartialOrder: a `$/partial` arriving after the reply is out of
/// order and must be dropped — the sink never sees it, and a second
/// streaming round trip (which forces the reader past the stray line)
/// still succeeds.
#[test]
fn partial_after_final_reply_is_dropped() {
    let provider = fake("partial-after-final", Duration::from_secs(5));
    let seen = std::sync::atomic::AtomicUsize::new(0);
    let count = || seen.load(std::sync::atomic::Ordering::Relaxed);
    for _ in 0..2 {
        provider
            .search_code_progressive("hit", &|_| {
                seen.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            })
            .expect("the reply must complete the request");
    }
    assert_eq!(
        count(),
        0,
        "no $/partial may be routed after its id's final reply"
    );
}

/// CorrelationSafety: a reply for an id that was never allocated is
/// ignored — the reader stays alive to deliver the real answer that
/// follows it on the same pipe.
#[test]
fn reply_for_unknown_id_is_ignored() {
    let provider = fake("unknown-id", Duration::from_secs(5));
    let result = provider
        .request("repo/tree", json!({ "repo": "o/r" }))
        .expect("the phantom reply must not stop the reader");
    assert_eq!(result["real"], true, "the real reply wins");
    provider
        .request("org/repos", json!({ "org": "o" }))
        .expect("transport stays usable after phantom replies");
}

/// NoIdReuse: id 2 times out; its reply arrives late, together with
/// id 3's request, and must be discarded — ids are monotonic and never
/// reused, so the late reply can never satisfy the fresh slot. Only a
/// transport that reuses ids would return `late: true` here.
#[test]
fn late_reply_after_timeout_is_discarded_not_rerouted() {
    let provider = fake("late-then-next", Duration::from_millis(300));
    let err = provider
        .request("repo/tree", json!({ "repo": "o/r" }))
        .expect_err("the hung id 2 must time out");
    assert_eq!(err.kind, ErrorKind::Timeout);
    let next = provider
        .request("org/repos", json!({ "org": "o" }))
        .expect("the transport recovers, and the late reply is dropped");
    assert_eq!(
        next["fresh"], true,
        "id 3's slot must get id 3's reply — never the late id 2 one"
    );
}

/// RestartFailClosed: EOF fails EVERY in-flight request at once, not
/// just the one that raced the death — both callers fail fast with
/// the closed-pipe error, well before their deadlines.
#[test]
fn eof_fails_every_in_flight_request_at_once() {
    let provider = Arc::new(fake("hang-2-die-3", Duration::from_secs(30)));
    let (tx, rx) = std::sync::mpsc::channel::<ProviderError>();
    let a = {
        let provider = Arc::clone(&provider);
        let tx = tx.clone();
        std::thread::spawn(move || {
            tx.send(provider.request("a/hung", json!({})).unwrap_err())
                .unwrap();
        })
    };
    // Gate B on A actually waiting — ids are assigned in arrival
    while provider
        .current_id
        .load(std::sync::atomic::Ordering::Acquire)
        == 0
    {
        std::thread::yield_now();
    }
    let b = {
        let provider = Arc::clone(&provider);
        let tx = tx.clone();
        std::thread::spawn(move || {
            let err = provider.request("b/killer", json!({})).unwrap_err();
            tx.send(err).unwrap();
        })
    };
    drop(tx);
    a.join().expect("A panics");
    b.join().expect("B panics");
    let failures: Vec<ProviderError> = rx.iter().collect();
    assert_eq!(failures.len(), 2, "both in-flight requests must fail");
    for err in failures {
        assert_eq!(
            err,
            ProviderError::new(ErrorKind::Provider, "provider closed its output"),
            "EOF must fail-close every in-flight slot"
        );
    }
}

#[test]
fn unsolicited_partials_cannot_extend_a_nonstreaming_deadline() {
    let provider = fake("stream-slow", Duration::from_millis(250));
    let error = provider
        .search_code("hit")
        .expect_err("non-streaming calls retain their round-trip deadline");
    assert_eq!(error.kind, ErrorKind::Timeout);
}

#[test]
fn progressive_search_rebuilds_after_child_death() {
    let provider = fake("die-on-2", Duration::from_secs(5));
    provider
        .search_code_progressive("first", &|_| {})
        .expect_err("the first child dies");
    provider
        .search_code_progressive("second", &|_| {})
        .expect("progressive calls share the recovery gate");
}

pub(crate) mod support;
pub(crate) use support::fake;