synaps-engine 0.11.1

//! Task #444 — prompt-cache keep-warm through the real session actor:
//! the command matrix (read-only commands keep the idle period, prefix
//! changes end it through the fingerprint, a turn starts a new one), pings
//! firing on schedule with their cost surfaced, and an active period
//! deferring park until it ends.

mod session_actor_common;
use session_actor_common::*;

use std::sync::{Arc, Mutex};
use std::time::Duration;

use agent_engine::session::{
    AttachMode, CheckpointReason, ClientKind, ClientMeta, ClientTransport, EndReason, HostEvent,
    LocalTransport, SessionCommand, SessionConfig, SessionHandle, SessionLifecycle, SessionQuery,
    SessionSetting,
};
use axum::response::IntoResponse;
use serial_test::serial;

/// A real turn: read 18k, wrote 2k with the 1h TTL (evidence that the
/// server honours 1h), 10 uncached input tokens, a 1-token reply.
const SSE_TURN: &str = concat!(
    "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_kw\",\"type\":\"message\",",
    "\"role\":\"assistant\",\"content\":[],\"model\":\"claude-sonnet-4-5\",\"stop_reason\":null,",
    "\"stop_sequence\":null,\"usage\":{\"input_tokens\":10,\"output_tokens\":0,",
    "\"cache_creation_input_tokens\":2000,\"cache_read_input_tokens\":18000,",
    "\"cache_creation\":{\"ephemeral_5m_input_tokens\":0,\"ephemeral_1h_input_tokens\":2000}}}}\n\n",
    "data: {\"type\":\"content_block_start\",\"index\":0,",
    "\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n",
    "data: {\"type\":\"content_block_delta\",\"index\":0,",
    "\"delta\":{\"type\":\"text_delta\",\"text\":\"hi\"}}\n\n",
    "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
    "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\",",
    "\"stop_sequence\":null},\"usage\":{\"input_tokens\":10,\"output_tokens\":1,",
    "\"cache_creation_input_tokens\":2000,\"cache_read_input_tokens\":18000,",
    "\"cache_creation\":{\"ephemeral_5m_input_tokens\":0,\"ephemeral_1h_input_tokens\":2000}}}\n\n",
    "data: {\"type\":\"message_stop\"}\n\n",
);

/// A ping that READ the 20k-token prefix from the cache.
const SSE_CACHED: &str = concat!(
    "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_kw\",\"type\":\"message\",",
    "\"role\":\"assistant\",\"content\":[],\"model\":\"claude-sonnet-4-5\",\"stop_reason\":null,",
    "\"stop_sequence\":null,\"usage\":{\"input_tokens\":10,\"output_tokens\":0,",
    "\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":20000}}}\n\n",
    "data: {\"type\":\"content_block_start\",\"index\":0,",
    "\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n",
    "data: {\"type\":\"content_block_delta\",\"index\":0,",
    "\"delta\":{\"type\":\"text_delta\",\"text\":\"hi\"}}\n\n",
    "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
    "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\",",
    "\"stop_sequence\":null},\"usage\":{\"input_tokens\":10,\"output_tokens\":1,",
    "\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":20000}}\n\n",
    "data: {\"type\":\"message_stop\"}\n\n",
);

/// A ping that was billed (usage arrived) and then failed in-stream.
const SSE_USAGE_THEN_ERROR: &str = concat!(
    "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_kw\",\"type\":\"message\",",
    "\"role\":\"assistant\",\"content\":[],\"model\":\"claude-sonnet-4-5\",\"stop_reason\":null,",
    "\"stop_sequence\":null,\"usage\":{\"input_tokens\":10,\"output_tokens\":0,",
    "\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":20000}}}\n\n",
    "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":null,",
    "\"stop_sequence\":null},\"usage\":{\"input_tokens\":10,\"output_tokens\":1,",
    "\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":20000}}\n\n",
    "data: {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",",
    "\"message\":\"synthetic\"}}\n\n",
);

/// Marker text of the keep-warm ping's trailing message.
const PING_MARK: &str = "prompt-cache keep-warm ping";

/// How the stub answers keep-warm pings (real turns always get `SSE_TURN`).
#[derive(Clone, Copy)]
enum Pings {
    /// A clean cache read.
    Cached,
    /// Never answers (the request is in flight until aborted).
    Hang,
    /// Usage, then an in-stream error.
    UsageThenError,
}

/// Records every request body; streaming requests get an SSE body, the
/// non-streaming compaction call a JSON summary.
async fn recording_stub() -> (String, Arc<Mutex<Vec<String>>>) {
    recording_stub_with(Pings::Cached).await
}

async fn recording_stub_with(mode: Pings) -> (String, Arc<Mutex<Vec<String>>>) {
    let bodies = Arc::new(Mutex::new(Vec::new()));
    let b = Arc::clone(&bodies);
    let app = axum::Router::new().fallback(move |body: String| {
        let b = Arc::clone(&b);
        async move {
            let streaming = body.contains("\"stream\":true");
            let ping = body.contains(PING_MARK);
            b.lock().unwrap().push(body);
            if streaming {
                let sse = match (ping, mode) {
                    (false, _) => SSE_TURN,
                    (true, Pings::Cached) => SSE_CACHED,
                    (true, Pings::UsageThenError) => SSE_USAGE_THEN_ERROR,
                    (true, Pings::Hang) => {
                        tokio::time::sleep(Duration::from_secs(600)).await;
                        SSE_CACHED
                    }
                };
                return (
                    axum::http::StatusCode::OK,
                    [("content-type", "text/event-stream")],
                    sse.to_string(),
                )
                    .into_response();
            }
            axum::Json(serde_json::json!({
                "id": "msg_c", "type": "message", "role": "assistant",
                "model": MODEL, "stop_reason": "end_turn",
                "content": [{"type": "text", "text": "summary"}],
                "usage": {"input_tokens": 10, "output_tokens": 5}
            }))
            .into_response()
        }
    });
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
    (format!("http://{addr}"), bodies)
}

fn pings(bodies: &Arc<Mutex<Vec<String>>>) -> usize {
    bodies
        .lock()
        .unwrap()
        .iter()
        .filter(|b| b.contains(PING_MARK))
        .count()
}

fn write_config(home: &Home, body: &str) {
    std::fs::write(home.dir.path().join(".synaps-cli").join("config"), body).unwrap();
}

struct EnvGuard(&'static str);
impl EnvGuard {
    fn set(k: &'static str, v: &str) -> Self {
        std::env::set_var(k, v);
        Self(k)
    }
}
impl Drop for EnvGuard {
    fn drop(&mut self) {
        std::env::remove_var(self.0);
    }
}

async fn turn(t: &mut LocalTransport, text: &str) {
    t.send(submit(text)).await.unwrap();
    until(t, |e| {
        matches!(e, agent_engine::session::SessionEventWire::Idle)
    })
    .await;
}

async fn query_result(t: &mut LocalTransport, id: u64) -> serde_json::Value {
    let seen = until(t, |e| {
        matches!(e, agent_engine::session::SessionEventWire::QueryResult { id: i, .. } if *i == id)
    })
    .await;
    match &seen.last().unwrap().event {
        agent_engine::session::SessionEventWire::QueryResult { value, .. } => value.clone(),
        _ => unreachable!(),
    }
}

/// `/keepwarm <arg>` through the real dispatch path; the structured state.
async fn keepwarm(t: &mut LocalTransport, id: u64, arg: &str) -> serde_json::Value {
    t.send(SessionCommand::EngineCommand {
        id,
        name: "keepwarm".into(),
        arg: arg.into(),
    })
    .await
    .unwrap();
    let v = query_result(t, id).await;
    assert_eq!(v["kind"], "output", "{v}");
    v["keepwarm"].clone()
}

/// How a command can affect an idle keep-warm period. Exhaustive on
/// purpose: a new `SessionCommand` variant does not compile here until it is
/// classified. Production code has NO such table — the fingerprint decides.
#[derive(Debug, PartialEq)]
enum Effect {
    /// Never changes the next request: the period continues.
    Untouched,
    /// May change the next request's prefix (model, thinking, tools, system,
    /// history): the period ends iff the fingerprint changed.
    Fingerprint,
    /// Starts a real turn: the period ends, the turn's Idle starts a new one.
    TurnStart,
    /// Compaction replaces the history: the period ends at its start.
    CompactionStart,
    /// An armed driver owns the session: the blocker ends the period.
    DriverOwned,
    /// The session ends.
    SessionEnd,
}

fn classify(cmd: &SessionCommand) -> Effect {
    match cmd {
        SessionCommand::Submit { .. }
        | SessionCommand::Steer { .. }
        | SessionCommand::SubmitPrepared { .. } => Effect::TurnStart,
        SessionCommand::Compact { .. } => Effect::CompactionStart,
        // Idle: injects a note into history (when the worker exists).
        SessionCommand::SubagentSteer { .. }
        | SessionCommand::NewSession
        | SessionCommand::Resume { .. }
        | SessionCommand::EngineCommand { .. }
        | SessionCommand::PluginCommand { .. } => Effect::Fingerprint,
        SessionCommand::Set { setting, .. } => match setting {
            SessionSetting::Model { .. }
            | SessionSetting::ReasoningLevel { .. }
            | SessionSetting::SystemPrompt { .. }
            | SessionSetting::ReloadPrompt
            | SessionSetting::ContextWindow { .. }
            | SessionSetting::GrantWorkerModel { .. } => Effect::Fingerprint,
            SessionSetting::CompactionModel { .. }
            | SessionSetting::ApiRetries { .. }
            | SessionSetting::SubagentTimeout { .. }
            | SessionSetting::MaxToolOutput { .. }
            | SessionSetting::BashTimeout { .. }
            | SessionSetting::BashMaxTimeout { .. } => Effect::Untouched,
        },
        SessionCommand::Cancel
        | SessionCommand::Answer { .. }
        | SessionCommand::Save
        | SessionCommand::Query { .. }
        | SessionCommand::Attach { .. }
        | SessionCommand::Detach { .. }
        | SessionCommand::Resync { .. }
        | SessionCommand::Checkpoint { .. }
        | SessionCommand::KeepWarm { .. }
        // Moves input ownership only; the next request is unchanged.
        | SessionCommand::TakeInput
        | SessionCommand::Park
        // Its wakes reach the event queue like any other event.
        | SessionCommand::AdoptWorkers { .. }
        // Hands the runtime pins only; the next turn start restores them.
        | SessionCommand::RestoreToolActivations { .. }
        | SessionCommand::ReloadAborted
        // Re-arms the park timer, which an active period already defers.
        | SessionCommand::ArmPark
        | SessionCommand::HostEvent(_) => Effect::Untouched,
        SessionCommand::DriverStart { .. } => Effect::DriverOwned,
        SessionCommand::End { .. } => Effect::SessionEnd,
    }
}

/// The keep-warm test seam: 300 ms between pings for this test (code-only,
/// never an environment variable — see `cache_keepwarm::interval_override`).
fn fast_pings() -> agent_engine::session::cache_keepwarm::testing::IntervalOverride {
    agent_engine::session::cache_keepwarm::testing::IntervalOverride::set(Duration::from_millis(
        300,
    ))
}

fn period_id(st: &serde_json::Value) -> u64 {
    st["period"]["id"].as_u64().expect("an active period")
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn command_matrix_read_only_keeps_period_prefix_changes_end_it() {
    let h = Home::new();
    // Thinking off: the ping's max_tokens is 1 (budgeted thinking would put
    // `budget + 1` output tokens into the worst case and no ping would fit).
    write_config(&h, "cache_keepwarm = auto\nthinking = off\n");
    let (url, bodies) = recording_stub().await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host.create_session(cfg()).await.unwrap();
    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    let mut id = 1_000u64;
    let mut next_id = || {
        id += 1;
        id
    };

    // No period before a real Anthropic request ran.
    let st = keepwarm(&mut t, next_id(), "status").await;
    assert_eq!(st["mode"], "auto");
    assert_eq!(st["active"], false, "{st}");

    turn(&mut t, "hello").await;
    let st = keepwarm(&mut t, next_id(), "").await;
    assert_eq!(st["active"], true, "a finished turn starts a period: {st}");
    assert_eq!(st["prefix_matches"], true, "{st}");
    let first_period = period_id(&st);

    // ── read-only: every one of these leaves the period running ──
    let me = t.client_id();
    let read_only: Vec<(&str, SessionCommand)> = vec![
        (
            "query status",
            SessionCommand::Query {
                id: next_id(),
                query: SessionQuery::Status,
            },
        ),
        (
            "query view",
            SessionCommand::Query {
                id: next_id(),
                query: SessionQuery::View,
            },
        ),
        (
            "query messages",
            SessionCommand::Query {
                id: next_id(),
                query: SessionQuery::Messages,
            },
        ),
        (
            "query context report",
            SessionCommand::Query {
                id: next_id(),
                query: SessionQuery::ContextReport,
            },
        ),
        (
            "query context assessment",
            SessionCommand::Query {
                id: next_id(),
                query: SessionQuery::ContextAssessment,
            },
        ),
        (
            "query prompt inspection",
            SessionCommand::Query {
                id: next_id(),
                query: SessionQuery::PromptInspection,
            },
        ),
        (
            "query tools schema",
            SessionCommand::Query {
                id: next_id(),
                query: SessionQuery::ToolsSchema,
            },
        ),
        (
            "query subagent rows",
            SessionCommand::Query {
                id: next_id(),
                query: SessionQuery::SubagentRows,
            },
        ),
        (
            "query display tail",
            SessionCommand::Query {
                id: next_id(),
                query: SessionQuery::DisplayTail { items: 5 },
            },
        ),
        (
            "/context",
            SessionCommand::EngineCommand {
                id: next_id(),
                name: "context".into(),
                arg: String::new(),
            },
        ),
        (
            "/budget status",
            SessionCommand::EngineCommand {
                id: next_id(),
                name: "budget".into(),
                arg: "status".into(),
            },
        ),
        (
            "/keepwarm status",
            SessionCommand::EngineCommand {
                id: next_id(),
                name: "keepwarm".into(),
                arg: "status".into(),
            },
        ),
        (
            "unknown plugin command",
            SessionCommand::PluginCommand {
                id: next_id(),
                plugin: "no-such-plugin".into(),
                name: "x".into(),
                arg: String::new(),
            },
        ),
        ("save", SessionCommand::Save),
        (
            "resync",
            SessionCommand::Resync {
                client: me,
                since_seq: 0,
            },
        ),
        ("pin", SessionCommand::KeepWarm { on: true }),
        ("unpin", SessionCommand::KeepWarm { on: false }),
        // The owner's own `/takeover`: a no-op that must not end the period.
        ("takeover (owner)", SessionCommand::TakeInput),
        ("cancel while idle", SessionCommand::Cancel),
        (
            "answer unknown prompt",
            SessionCommand::Answer {
                prompt_id: 9_999,
                value: None,
            },
        ),
        (
            "checkpoint",
            SessionCommand::Checkpoint {
                reason: CheckpointReason::HostRequest,
            },
        ),
        (
            "set api retries",
            SessionCommand::Set {
                id: next_id(),
                setting: SessionSetting::ApiRetries { n: 3 },
            },
        ),
        (
            "set max tool output",
            SessionCommand::Set {
                id: next_id(),
                setting: SessionSetting::MaxToolOutput { bytes: 50_000 },
            },
        ),
        (
            "set bash timeout",
            SessionCommand::Set {
                id: next_id(),
                setting: SessionSetting::BashTimeout { secs: 60 },
            },
        ),
        (
            "set subagent timeout",
            SessionCommand::Set {
                id: next_id(),
                setting: SessionSetting::SubagentTimeout { secs: 120 },
            },
        ),
        (
            "steer an unknown subagent (refused, no note)",
            SessionCommand::SubagentSteer {
                handle_id: "sa_404".into(),
                text: "nope".into(),
            },
        ),
    ];
    for (name, cmd) in read_only {
        let effect = classify(&cmd);
        assert!(
            matches!(effect, Effect::Untouched | Effect::Fingerprint),
            "{name}: {effect:?}"
        );
        t.send(cmd).await.unwrap();
        let st = keepwarm(&mut t, next_id(), "status").await;
        assert_eq!(st["active"], true, "{name} stopped keep-warm: {st}");
        assert_eq!(
            st["prefix_matches"], true,
            "{name} changed the prefix: {st}"
        );
        assert_eq!(period_id(&st), first_period, "{name}");
    }
    // Host-side, never from a client: an extension notification and a park
    // request. persist = false here, so the session cannot park and an
    // explicit Park leaves the period alone (with persist it ends it first,
    // see `explicit_park_ends_the_period_then_parks`).
    for cmd in [
        SessionCommand::HostEvent(HostEvent::ExtensionNotification {
            extension_id: "ext".into(),
            method: "note".into(),
            params: serde_json::json!({}),
        }),
        SessionCommand::Park,
    ] {
        assert_eq!(classify(&cmd), Effect::Untouched);
        handle.send(cmd).await.unwrap();
    }
    // A second client attaching and leaving.
    let (other, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Attach))
        .await
        .unwrap();
    other
        .send(SessionCommand::Detach {
            client: other.client_id(),
        })
        .await
        .unwrap();
    let st = keepwarm(&mut t, next_id(), "status").await;
    assert_eq!(st["active"], true, "host events / attach / detach: {st}");
    assert_eq!(period_id(&st), first_period);

    // ── a real turn ends the period; its Idle starts a fresh one ──
    let cmd = submit("again");
    assert_eq!(classify(&cmd), Effect::TurnStart);
    t.send(cmd).await.unwrap();
    until(&mut t, |e| {
        matches!(e, agent_engine::session::SessionEventWire::Idle)
    })
    .await;
    let st = keepwarm(&mut t, next_id(), "status").await;
    assert!(period_id(&st) > first_period, "{st}");
    assert_eq!(st["last_end"], "turn started", "{st}");

    // ── prefix changes: each ends the period through the fingerprint ──
    let thinking = if t.view().thinking_level == "off" {
        "high"
    } else {
        "off"
    };
    let changers: Vec<(&str, SessionCommand)> = vec![
        (
            "set system prompt",
            SessionCommand::Set {
                id: next_id(),
                setting: SessionSetting::SystemPrompt {
                    text: "A different system prompt.".into(),
                },
            },
        ),
        (
            "/thinking",
            SessionCommand::EngineCommand {
                id: next_id(),
                name: "thinking".into(),
                arg: thinking.into(),
            },
        ),
        (
            "set model",
            SessionCommand::Set {
                id: next_id(),
                setting: SessionSetting::Model {
                    model: "claude-opus-4-6".into(),
                },
            },
        ),
        ("/clear", SessionCommand::NewSession),
    ];
    for (name, cmd) in changers {
        assert_eq!(classify(&cmd), Effect::Fingerprint, "{name}");
        turn(&mut t, "baseline").await;
        let st = keepwarm(&mut t, next_id(), "status").await;
        assert_eq!(st["active"], true, "{name}: baseline period: {st}");
        t.send(cmd).await.unwrap();
        let st = keepwarm(&mut t, next_id(), "status").await;
        assert_eq!(st["active"], false, "{name} must end the period: {st}");
        assert_eq!(st["last_end"], "prefix changed", "{name}: {st}");
        if name == "/thinking" && thinking != "off" {
            // Budgeted thinking on this small mock prefix is (correctly) refused by
            // the worst-case bound, which would mask the next changer's baseline.
            // Restore the cheap thinking-off baseline before the next case.
            t.send(SessionCommand::EngineCommand {
                id: next_id(),
                name: "thinking".into(),
                arg: "off".into(),
            })
            .await
            .unwrap();
        }
    }

    // ── /keepwarm off is honoured per session; auto turns it back on ──
    turn(&mut t, "baseline").await;
    let st = keepwarm(&mut t, next_id(), "off").await;
    assert_eq!(
        (st["mode"].as_str(), st["active"].as_bool()),
        (Some("off"), Some(false))
    );
    assert_eq!(st["last_end"], "disabled", "{st}");
    turn(&mut t, "while off").await;
    let st = keepwarm(&mut t, next_id(), "status").await;
    assert_eq!(st["active"], false, "off: no period after a turn: {st}");
    let st = keepwarm(&mut t, next_id(), "6h").await;
    assert_eq!(st["mode"], "6h");
    turn(&mut t, "back on").await;
    let st = keepwarm(&mut t, next_id(), "status").await;
    assert_eq!(st["active"], true, "{st}");

    // ── compaction ends the period at its start ──
    let cmd = SessionCommand::Compact { instructions: None };
    assert_eq!(classify(&cmd), Effect::CompactionStart);
    t.send(cmd).await.unwrap();
    let st = keepwarm(&mut t, next_id(), "status").await;
    assert_eq!(st["active"], false, "{st}");
    assert_eq!(st["last_end"], "compaction", "{st}");

    // The interval is the real one (minutes): not a single ping was sent.
    assert_eq!(pings(&bodies), 0);
    let end_cmd = SessionCommand::End {
        reason: EndReason::ClientQuit,
    };
    assert_eq!(classify(&end_cmd), Effect::SessionEnd);
    end(&mut t).await;
}

async fn wait_lifecycle(handle: &SessionHandle, want: SessionLifecycle, for_: Duration) -> bool {
    let deadline = tokio::time::Instant::now() + for_;
    while tokio::time::Instant::now() < deadline {
        if handle.lifecycle() == want {
            return true;
        }
        tokio::time::sleep(Duration::from_millis(25)).await;
    }
    handle.lifecycle() == want
}

async fn wait_pings(bodies: &Arc<Mutex<Vec<String>>>, n: usize, for_: Duration) -> bool {
    let deadline = tokio::time::Instant::now() + for_;
    while tokio::time::Instant::now() < deadline {
        if pings(bodies) >= n {
            return true;
        }
        tokio::time::sleep(Duration::from_millis(25)).await;
    }
    pings(bodies) >= n
}

/// `Status` query through the real dispatch path.
async fn status(t: &mut LocalTransport, id: u64) -> serde_json::Value {
    t.send(SessionCommand::Query {
        id,
        query: SessionQuery::Status,
    })
    .await
    .unwrap();
    query_result(t, id).await
}

/// Poll `Status` until `pred` holds (or `for_` elapses); the last status.
async fn status_until(
    t: &mut LocalTransport,
    first_id: u64,
    for_: Duration,
    pred: impl Fn(&serde_json::Value) -> bool,
) -> serde_json::Value {
    let deadline = tokio::time::Instant::now() + for_;
    let mut id = first_id;
    loop {
        let st = status(t, id).await;
        if pred(&st) || tokio::time::Instant::now() >= deadline {
            return st;
        }
        id += 1;
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

/// Generous waits. Nothing here asserts that something happens WITHIN a
/// window: the schedule is driven by a 300 ms test interval against a 1h
/// TTL, so a ping that runs late under load is still a cache read (never
/// expired), the number of pings is fixed by the ledger arithmetic, not by
/// time, and every wait below only bounds how long a hung test takes to fail.
const PATIENCE: Duration = Duration::from_secs(60);

/// Pings fire on schedule (interval shortened by the test seam), carry the
/// keep-warm text after the history, are costed into the session; detaching
/// does NOT stop them and an active period defers park; once the ski-rental
/// bound ends the period the session parks normally.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn pings_are_costed_and_an_active_period_defers_park_until_it_ends() {
    let h = Home::new();
    write_config(
        &h,
        "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n",
    );
    let _grace = EnvGuard::set("SYNAPS_DAEMON_PARK_GRACE_SECS", "0");
    let _interval = fast_pings();
    let (url, bodies) = recording_stub().await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host
        .create_session(SessionConfig {
            persist: true,
            ..cfg()
        })
        .await
        .unwrap();
    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    turn(&mut t, "hello").await;
    assert!(
        wait_pings(&bodies, 1, PATIENCE).await,
        "the first ping fires one interval after the turn's request"
    );
    let ping: serde_json::Value = {
        let all = bodies.lock().unwrap();
        serde_json::from_str(all.iter().find(|b| b.contains(PING_MARK)).unwrap()).unwrap()
    };
    let msgs = ping["messages"].as_array().unwrap();
    assert_eq!(
        msgs.len(),
        3,
        "history (user, assistant) + the trailing ping"
    );
    assert!(ping.get("thinking").is_none(), "{}", ping["thinking"]);
    assert_eq!(ping["max_tokens"], 1);
    assert!(msgs[2]["content"].as_str().unwrap().contains(PING_MARK));

    // Cost is surfaced: labelled, inside the session total, and the ping
    // tokens are inside the token totals (shown separately).
    let st = status_until(&mut t, 77, PATIENCE, |st| {
        st["keepwarm_cost"].as_f64().unwrap_or(0.0) > 0.0
    })
    .await;
    let kw = st["keepwarm_cost"].as_f64().unwrap();
    assert!(kw > 0.0, "{st}");
    assert!(
        st["keepwarm"]["session_pings"].as_u64().unwrap() >= 1,
        "{st}"
    );
    assert!(
        st["cost"].as_f64().unwrap() > kw,
        "the total includes the turn: {st}"
    );
    // The turn's 1h write is the evidence: governed by the 1h TTL.
    assert_eq!(st["keepwarm"]["period"]["ttl_secs"], 3600, "{st}");

    // Detach: keep-warm continues, and while a period is active the session
    // never parks (grace is 0, so it would park at once otherwise).
    t.send(SessionCommand::Detach {
        client: t.client_id(),
    })
    .await
    .unwrap();
    let after_detach = pings(&bodies);
    assert!(
        wait_pings(&bodies, after_detach + 1, PATIENCE).await || pings(&bodies) >= 19,
        "detach does not stop keep-warm"
    );

    // 20k-token prefix under the honoured 1h TTL: the bound is the $0.12
    // rewrite, one ping costs $0.006045 (read + 10 input + 1 output), so the
    // ledger admits exactly 19 pings; then the period ends and the session
    // parks normally.
    assert!(
        wait_lifecycle(&handle, SessionLifecycle::Parked, PATIENCE).await,
        "parks once the ski-rental bound ends the period"
    );
    let total = pings(&bodies);
    assert_eq!(total, 19, "{total} pings");
    tokio::time::sleep(Duration::from_millis(900)).await;
    assert_eq!(pings(&bodies), total, "no ping after the period ended");
    handle
        .send(SessionCommand::End {
            reason: EndReason::HostShutdown,
        })
        .await
        .unwrap();
}

/// Every `cache_control` value in a request body.
fn cache_markers(v: &serde_json::Value) -> Vec<serde_json::Value> {
    match v {
        serde_json::Value::Object(map) => map
            .iter()
            .flat_map(|(k, v)| {
                if k == "cache_control" {
                    vec![v.clone()]
                } else {
                    cache_markers(v)
                }
            })
            .collect(),
        serde_json::Value::Array(items) => items.iter().flat_map(cache_markers).collect(),
        _ => Vec::new(),
    }
}

/// `cache_ttl` unset (`auto`) on the harness's OAuth login resolves to 1h,
/// and keep-warm runs exactly as under an explicit `cache_ttl = 1h`: the
/// turn and every ping carry 1h markers, the period is governed by the 1h
/// TTL, and the ski-rental cost check admits the same number of pings
/// before it ends the period. `Status` reports the setting and what it
/// resolved to.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn auto_ttl_on_oauth_keeps_warm_exactly_like_explicit_1h() {
    let mut runs = Vec::new();
    for ttl_line in ["", "cache_ttl = 1h\n"] {
        let h = Home::new();
        write_config(&h, &format!("cache_keepwarm = auto\n{ttl_line}thinking = off\n"));
        let _interval = fast_pings();
        let (url, bodies) = recording_stub().await;
        std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
        let host = host().await;
        let handle = host.create_session(cfg()).await.unwrap();
        let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
            .await
            .unwrap();
        turn(&mut t, "hello").await;
        let st = status_until(&mut t, 900, PATIENCE, |st| {
            st["keepwarm"]["session_pings"].as_u64().unwrap_or(0) >= 1
        })
        .await;
        let setting = if ttl_line.is_empty() { "auto" } else { "1h" };
        assert_eq!(
            st["cache_ttl"],
            serde_json::json!({"setting": setting, "resolved": "1h"}),
            "{st}"
        );
        let period_ttl = st["keepwarm"]["period"]["ttl_secs"].clone();
        // Let the cost check end the period.
        let st = status_until(&mut t, 1000, PATIENCE, |st| {
            st["keepwarm"]["active"] == false
        })
        .await;
        assert_eq!(st["keepwarm"]["active"], false, "the period ended: {st}");
        let total = pings(&bodies);
        let all = bodies.lock().unwrap().clone();
        for (i, body) in all.iter().enumerate() {
            let v: serde_json::Value = serde_json::from_str(body).unwrap();
            let m = cache_markers(&v);
            assert!(!m.is_empty(), "{setting}: request {i} has no markers");
            assert!(
                m.iter().all(|m| m["ttl"] == "1h"),
                "{setting}: request {i} markers {m:?}"
            );
        }
        // The human `/keepwarm status` line says it too.
        t.send(SessionCommand::EngineCommand {
            id: 1100,
            name: "keepwarm".into(),
            arg: "status".into(),
        })
        .await
        .unwrap();
        let v = query_result(&mut t, 1100).await;
        let want_line = if setting == "auto" {
            "prompt-cache TTL: 1h (auto)"
        } else {
            "prompt-cache TTL: 1h\n"
        };
        assert!(v["text"].as_str().unwrap().contains(want_line), "{v}");
        end(&mut t).await;
        runs.push((setting, period_ttl, total, st["keepwarm"]["last_end"].clone()));
    }
    let (auto, explicit) = (&runs[0], &runs[1]);
    assert_eq!(auto.1, 3600, "auto on OAuth is governed by the 1h TTL: {runs:?}");
    assert_eq!(
        (&auto.1, auto.2, &auto.3),
        (&explicit.1, explicit.2, &explicit.3),
        "auto (resolved 1h) and explicit 1h: same TTL, same pings, same end: {runs:?}"
    );
}

/// An explicit `Park` while a period is active ends it ("parked by
/// request") and parks; only the idle auto-park waits for the period.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn explicit_park_ends_the_period_then_parks() {
    let h = Home::new();
    write_config(
        &h,
        "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n",
    );
    // Grace 0: without the keep-warm deferral the session would park the
    // moment its last client detaches.
    let _grace = EnvGuard::set("SYNAPS_DAEMON_PARK_GRACE_SECS", "0");
    let (url, bodies) = recording_stub().await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host
        .create_session(SessionConfig {
            persist: true,
            ..cfg()
        })
        .await
        .unwrap();
    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    turn(&mut t, "hello").await;
    let st = keepwarm(&mut t, 500, "status").await;
    assert_eq!(st["active"], true, "{st}");
    // Detach: with grace 0 the idle auto-park is due at once, but the
    // active period defers it — still Live well after the detach.
    t.send(SessionCommand::Detach {
        client: t.client_id(),
    })
    .await
    .unwrap();
    assert!(
        !wait_lifecycle(
            &handle,
            SessionLifecycle::Parked,
            Duration::from_millis(750)
        )
        .await,
        "an active period defers the zero-grace auto-park"
    );
    assert_eq!(handle.lifecycle(), SessionLifecycle::Live);
    // An explicit park is not deferred: it ends the period and parks.
    handle.send(SessionCommand::Park).await.unwrap();
    assert!(
        wait_lifecycle(&handle, SessionLifecycle::Parked, PATIENCE).await,
        "an explicit park is not deferred"
    );
    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    let st = keepwarm(&mut t, 501, "status").await;
    assert_eq!(st["active"], false, "{st}");
    assert_eq!(st["last_end"], "parked by request", "{st}");
    assert_eq!(pings(&bodies), 0, "the real interval: no ping was due");
    end(&mut t).await;
}

/// `/keepwarm auto` on a session that is already idle starts a period at
/// once (no new turn needed).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn keepwarm_auto_on_an_idle_session_starts_a_period_now() {
    let h = Home::new();
    write_config(&h, "cache_keepwarm = off\ncache_ttl = 1h\nthinking = off\n");
    let (url, _bodies) = recording_stub().await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host.create_session(cfg()).await.unwrap();
    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    turn(&mut t, "hello").await;
    let st = keepwarm(&mut t, 600, "status").await;
    assert_eq!(
        (st["mode"].as_str(), st["active"].as_bool()),
        (Some("off"), Some(false))
    );
    let st = keepwarm(&mut t, 601, "auto").await;
    assert_eq!(st["active"], true, "started immediately: {st}");
    let st = keepwarm(&mut t, 602, "off").await;
    assert_eq!(st["active"], false, "{st}");
    let st = keepwarm(&mut t, 603, "6h").await;
    assert_eq!(st["active"], true, "<N>h starts it too: {st}");
    end(&mut t).await;
}

/// An `Observe` attach never makes a session keep-warm eligible, whatever
/// its client kind. Watching (`synaps attach --observe`, a read-only pane in
/// another app) is not someone who will be back to type.
/// A TUI that can type (`Mirror`) still marks it, even after observers.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn an_observe_attach_never_makes_a_session_eligible() {
    let h = Home::new();
    write_config(
        &h,
        "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n",
    );
    let (url, bodies) = recording_stub().await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host.create_session(cfg()).await.unwrap();
    // The client that drives the turn is not interactive by kind.
    let (mut script, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    // Both interactive kinds attach, observe-only.
    let mut watchers = Vec::new();
    for kind in [ClientKind::Tui, ClientKind::Attach] {
        let (w, snap) =
            LocalTransport::attach_with(handle.clone(), ClientMeta::new(kind), AttachMode::Observe)
                .await
                .unwrap();
        assert_eq!(w.mode(), AttachMode::Observe);
        assert!(
            snap.clients.iter().any(|(_, k)| *k == kind),
            "{kind:?} is attached"
        );
        watchers.push(w);
    }
    turn(&mut script, "hello").await;
    let st = keepwarm(&mut script, 700, "status").await;
    assert_eq!(
        (st["mode"].as_str(), st["active"].as_bool()),
        (Some("auto"), Some(false)),
        "the turn's Idle started no period: {st}"
    );
    // On an idle session `/keepwarm auto` starts a period at once if the
    // session is eligible (see the test above): observe-only, it is not.
    let st = keepwarm(&mut script, 701, "auto").await;
    assert_eq!(st["active"], false, "observe-only is never eligible: {st}");
    assert_eq!(pings(&bodies), 0);

    // A TUI that can type marks the session: the same command now starts one.
    let (_tui, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    let st = keepwarm(&mut script, 702, "auto").await;
    assert_eq!(st["active"], true, "a Mirror TUI makes it eligible: {st}");
    drop(watchers);
    end(&mut script).await;
}

/// `cache_keepwarm_headless`: a session driven only by a non-interactive
/// client (chat, rpc, server) is never eligible by default, and is once the
/// owner opts in. Observe stays excluded either way (unit-tested in
/// `client_qualifies`).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn headless_clients_qualify_only_with_the_opt_in() {
    let (url, bodies) = recording_stub().await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    for (headless, kind) in [
        (false, ClientKind::Chat),
        (false, ClientKind::Rpc),
        (false, ClientKind::Server),
        (true, ClientKind::Chat),
        (true, ClientKind::Rpc),
        (true, ClientKind::Server),
    ] {
        let h = Home::new();
        write_config(
            &h,
            &format!(
                "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n\
                 cache_keepwarm_headless = {headless}\n"
            ),
        );
        let host = host().await;
        let handle = host.create_session(cfg()).await.unwrap();
        let (mut c, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(kind))
            .await
            .unwrap();
        turn(&mut c, "hello").await;
        // On an idle session `/keepwarm auto` starts a period at once iff
        // the session is eligible.
        let st = keepwarm(&mut c, 710, "auto").await;
        assert_eq!(
            st["active"].as_bool(),
            Some(headless),
            "{kind:?} with cache_keepwarm_headless = {headless}: {st}"
        );
        // End the period before the session goes away.
        let _ = keepwarm(&mut c, 711, "off").await;
        end(&mut c).await;
    }
    assert_eq!(pings(&bodies), 0, "no ping is due within the test");
}

/// Spend is booked on every path: a ping that was billed and then failed
/// (usage arrived, then an in-stream error) lands in the keep-warm cost, the
/// session cost and the token totals, though it counts as no successful ping.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn a_billed_ping_that_failed_is_still_booked() {
    let h = Home::new();
    write_config(
        &h,
        "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n",
    );
    let _interval = fast_pings();
    let (url, bodies) = recording_stub_with(Pings::UsageThenError).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host.create_session(cfg()).await.unwrap();
    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    turn(&mut t, "hello").await;
    assert!(wait_pings(&bodies, 1, PATIENCE).await);
    let st = status_until(&mut t, 701, PATIENCE, |st| {
        st["keepwarm_cost"].as_f64().unwrap_or(0.0) > 0.0
    })
    .await;
    let kw = st["keepwarm_cost"].as_f64().unwrap();
    // 20k read + 10 input + 1 output on Sonnet 4.5.
    assert!((kw - 0.006045).abs() < 1e-6, "{st}");
    assert_eq!(st["keepwarm"]["session_pings"], 0, "a failed ping: {st}");
    // The turn: 10 input + 1 output + 18k read + 2k written at 1h.
    let turn_cost = (10.0 * 3.0 + 15.0 + 18_000.0 * 0.3 + 2_000.0 * 6.0) / 1e6;
    let cost = st["cost"].as_f64().unwrap();
    assert!((cost - turn_cost - kw).abs() < 1e-9, "{st}");
    assert!(
        st["tokens"]["input"].as_u64().unwrap() > 10,
        "ping tokens are in the totals: {st}"
    );
    end(&mut t).await;
}

/// A ping aborted in flight (a turn starts while it is on the wire) is
/// booked at the period's worst-case estimate: it may well be billed.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn an_aborted_in_flight_ping_is_booked() {
    let h = Home::new();
    write_config(
        &h,
        "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n",
    );
    let _interval = fast_pings();
    let (url, bodies) = recording_stub_with(Pings::Hang).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host.create_session(cfg()).await.unwrap();
    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    turn(&mut t, "hello").await;
    // The stub has the ping's body: it left this process and is in flight.
    assert!(wait_pings(&bodies, 1, PATIENCE).await);
    let before = status(&mut t, 800).await;
    assert_eq!(before["keepwarm"]["period"]["in_flight"], true, "{before}");
    assert_eq!(before["keepwarm_cost"].as_f64().unwrap(), 0.0);
    turn(&mut t, "again").await;
    let st = status(&mut t, 801).await;
    let kw = st["keepwarm_cost"].as_f64().unwrap();
    // Worst case of ping 1: 20k read + the 11-token tail written at 1h +
    // 64 estimated input + 1 output, on Sonnet 4.5.
    let worst = (20_000.0 * 0.3 + 11.0 * 6.0 + 64.0 * 3.0 + 15.0) / 1e6;
    assert!((kw - worst).abs() < 1e-9, "{kw} vs {worst}: {st}");
    assert_eq!(st["keepwarm"]["session_pings"], 0, "{st}");
    end(&mut t).await;
}

/// Worst case of ping 1 in these tests: 20k read + the 11-token tail
/// written at 1h + 64 estimated input + 1 output, on Sonnet 4.5.
const WORST_PING_1: f64 = (20_000.0 * 0.3 + 11.0 * 6.0 + 64.0 * 3.0 + 15.0) / 1e6;

/// One `SSE_TURN` turn on Sonnet 4.5: 10 input + 1 output + 18k read + 2k
/// written at 1h.
const TURN_COST: f64 = (10.0 * 3.0 + 15.0 + 18_000.0 * 0.3 + 2_000.0 * 6.0) / 1e6;

/// I-1: once the ski-rental bound ended the period, `/keepwarm off` then
/// `auto` in the same cache epoch continues the exhausted ledger: nothing
/// more is pinged until a real turn starts a new epoch.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn re_enabling_after_the_bound_pings_nothing_until_a_real_turn() {
    let h = Home::new();
    write_config(
        &h,
        "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n",
    );
    let _interval = fast_pings();
    let (url, bodies) = recording_stub().await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host.create_session(cfg()).await.unwrap();
    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    turn(&mut t, "hello").await;
    let st = status_until(&mut t, 900, PATIENCE, |st| {
        st["keepwarm"]["last_end"]
            .as_str()
            .is_some_and(|e| e.contains("ski-rental bound"))
    })
    .await;
    assert_eq!(st["keepwarm"]["active"], false, "{st}");
    let total = pings(&bodies);
    assert_eq!(total, 19, "{st}");

    let st = keepwarm(&mut t, 1_000, "off").await;
    assert_eq!(st["active"], false, "{st}");
    let st = keepwarm(&mut t, 1_001, "auto").await;
    assert_eq!(st["active"], false, "the epoch's ledger is spent: {st}");
    assert!(
        st["last_end"]
            .as_str()
            .is_some_and(|e| e.contains("not started: ski-rental bound")),
        "{st}"
    );
    tokio::time::sleep(Duration::from_millis(900)).await;
    assert_eq!(pings(&bodies), total, "re-enabling pinged nothing");

    // A real turn is a new epoch: a fresh ledger.
    turn(&mut t, "again").await;
    let st = keepwarm(&mut t, 1_002, "status").await;
    assert_eq!(st["active"], true, "{st}");
    end(&mut t).await;
}

/// M-2: `/clear` while a ping is in flight ends the period at once, aborts
/// the ping, and books it to the OLD conversation (saved by the clear).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn clear_mid_ping_books_the_ping_to_the_old_conversation() {
    let h = Home::new();
    write_config(
        &h,
        "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n",
    );
    let _interval = fast_pings();
    let (url, bodies) = recording_stub_with(Pings::Hang).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host
        .create_session(SessionConfig {
            persist: true,
            ..cfg()
        })
        .await
        .unwrap();
    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    turn(&mut t, "hello").await;
    assert!(wait_pings(&bodies, 1, PATIENCE).await);
    let before = status(&mut t, 1_100).await;
    assert_eq!(before["keepwarm"]["period"]["in_flight"], true, "{before}");
    let old_id = before["session"].as_str().unwrap().to_string();

    t.send(SessionCommand::NewSession).await.unwrap();
    until(&mut t, |e| {
        matches!(e, agent_engine::session::SessionEventWire::Cleared { .. })
    })
    .await;
    let st = status(&mut t, 1_101).await;
    assert_ne!(st["session"], old_id.as_str(), "{st}");
    assert_eq!(st["keepwarm"]["active"], false, "{st}");
    assert_eq!(st["keepwarm"]["last_end"], "prefix changed", "{st}");
    assert_eq!(st["keepwarm_cost"].as_f64().unwrap(), 0.0, "{st}");
    assert_eq!(
        st["cost"].as_f64().unwrap(),
        0.0,
        "not on the new one: {st}"
    );
    let old = agent_engine::resolve_session(&old_id).unwrap();
    assert!(
        (old.session_cost - TURN_COST - WORST_PING_1).abs() < 1e-9,
        "booked to the old conversation: {} vs {}",
        old.session_cost,
        TURN_COST + WORST_PING_1
    );
    end(&mut t).await;
}

/// M-2: `/resume` while a ping is in flight ends the period at once, aborts
/// the ping, and books it to the conversation being left — never to the
/// resumed one.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn resume_mid_ping_books_the_ping_to_the_old_conversation() {
    let h = Home::new();
    write_config(
        &h,
        "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n",
    );
    let _interval = fast_pings();
    let (url, bodies) = recording_stub_with(Pings::Hang).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    // A saved session to resume (on disk only: no actor holds its lock).
    let mut saved = agent_engine::Session::new(MODEL, "off", None);
    saved.api_messages = vec![
        Arc::new(serde_json::json!({"role": "user", "content": "other"})),
        Arc::new(
            serde_json::json!({"role": "assistant", "content": [{"type": "text", "text": "hi"}]}),
        ),
    ];
    saved.session_cost = 0.125;
    saved.save().await.unwrap();
    let other_id = saved.id.clone();

    let handle = host
        .create_session(SessionConfig {
            persist: true,
            ..cfg()
        })
        .await
        .unwrap();
    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    turn(&mut t, "hello").await;
    assert!(wait_pings(&bodies, 1, PATIENCE).await);
    let before = status(&mut t, 1_200).await;
    assert_eq!(before["keepwarm"]["period"]["in_flight"], true, "{before}");
    let old_id = before["session"].as_str().unwrap().to_string();

    t.send(SessionCommand::Resume {
        id: 1_202,
        query: other_id.clone(),
    })
    .await
    .unwrap();
    until(&mut t, |e| {
        matches!(e, agent_engine::session::SessionEventWire::Resumed { .. })
    })
    .await;
    let st = status(&mut t, 1_203).await;
    assert_eq!(st["session"], other_id.as_str(), "{st}");
    assert_eq!(st["keepwarm"]["active"], false, "{st}");
    assert_eq!(st["keepwarm"]["last_end"], "prefix changed", "{st}");
    assert_eq!(st["keepwarm_cost"].as_f64().unwrap(), 0.0, "{st}");
    assert_eq!(
        st["cost"].as_f64().unwrap(),
        0.125,
        "the resumed conversation is not charged: {st}"
    );
    let old = agent_engine::resolve_session(&old_id).unwrap();
    assert!(
        (old.session_cost - TURN_COST - WORST_PING_1).abs() < 1e-9,
        "booked to the old conversation: {} vs {}",
        old.session_cost,
        TURN_COST + WORST_PING_1
    );
    end(&mut t).await;
}

/// Review #205: a settled ping's charge reaches disk on its own. Without it a
/// crash while idle loses the spend and restores `max_session_cost`
/// headroom.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn review_205_completed_ping_accounting_is_persisted_without_manual_save() {
    let h = Home::new();
    write_config(&h, "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n");
    let _interval = fast_pings();
    let (url, _bodies) = recording_stub().await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host
        .create_session(SessionConfig {
            persist: true,
            ..cfg()
        })
        .await
        .unwrap();
    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    turn(&mut t, "hello").await;
    let st = status_until(&mut t, 6000, PATIENCE, |s| {
        s["keepwarm_cost"].as_f64().unwrap_or(0.0) > 0.0
    })
    .await;
    assert!(st["keepwarm_cost"].as_f64().unwrap() > 0.0);
    // No manual Save/Park/End or further real turn before reading durable state.
    tokio::time::sleep(Duration::from_millis(500)).await;
    let disk = agent_core::session::Session::load(st["session"].as_str().unwrap()).unwrap();
    let disk_kw = disk.keepwarm_cost;
    let disk_cost = disk.session_cost;
    end(&mut t).await;
    assert!(
        disk_kw > 0.0,
        "completed ping missing on disk: kw={disk_kw}, total={disk_cost}; in-memory={st}"
    );
}

// ── A due ping while `on_session_start` is pending ──────────────────────────
//
// The ping repeats the next turn's prefix, session-start injection included,
// so it waits for that hook: the timer is off meanwhile and the ping is looked
// at again once the hook has run. No production path reaches this today (a
// period needs an accepted request, which already waited for the hook), so
// two test seams build it: the host's session-start gate is held, and the
// turn that starts the period puts a fresh hook back in line at its end.

use agent_engine::session::cache_keepwarm::testing::due_checks;
use agent_engine::session::session_start_testing as start_seam;

/// A turn that starts a period and leaves a hook pending behind the held
/// gate. Returns the transport, the handle, the stub's bodies, the gate hold
/// and the hook-call count before the turn.
async fn period_with_a_pending_hook(
    persist: bool,
) -> (
    LocalTransport,
    SessionHandle,
    Arc<Mutex<Vec<String>>>,
    start_seam::GateHold,
    u64,
) {
    let (url, bodies) = recording_stub().await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host
        .create_session(SessionConfig { persist, ..cfg() })
        .await
        .unwrap();
    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    let hold = start_seam::hold_gate(&host).await;
    let sid = handle.journal_id();
    let calls = start_seam::calls(&sid);
    start_seam::requeue_after_next_turn(&sid);
    turn(&mut t, "hello").await;
    let st = keepwarm(&mut t, 700, "status").await;
    assert_eq!(st["active"], true, "the turn started a period: {st}");
    // The first ping is due one 300 ms interval after the turn's request.
    tokio::time::sleep(Duration::from_millis(1200)).await;
    assert_eq!(pings(&bodies), 0, "no ping before on_session_start has run");
    assert_eq!(start_seam::calls(&sid), calls, "the hook is still in line");
    (t, handle, bodies, hold, calls)
}

/// Wait (bounded) until `n` hooks have run since `base`.
async fn hooks_run(sid: &str, base: u64, n: u64, for_: Duration) -> bool {
    let deadline = tokio::time::Instant::now() + for_;
    while tokio::time::Instant::now() < deadline {
        if start_seam::calls(sid) >= base + n {
            return true;
        }
        tokio::time::sleep(Duration::from_millis(25)).await;
    }
    start_seam::calls(sid) >= base + n
}

/// Held: no ping, no busy loop (the timer is off, so a passed deadline does
/// not wake the actor again and again), the actor still answers. Once the
/// hook has run the held ping goes out.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn a_due_ping_waits_for_a_pending_session_start_hook_without_spinning() {
    let h = Home::new();
    write_config(&h, "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n");
    let _interval = fast_pings();
    let (mut t, handle, bodies, hold, calls) = period_with_a_pending_hook(false).await;

    let checks = due_checks();
    tokio::time::sleep(Duration::from_millis(1200)).await;
    let more = due_checks() - checks;
    assert!(more <= 1, "the timer is off while the ping waits: {more} due checks in 1.2 s");
    assert_eq!(pings(&bodies), 0, "still no ping");
    let st = keepwarm(&mut t, 701, "status").await;
    assert_eq!(st["active"], true, "the period is still there: {st}");
    assert_eq!(st["period"]["in_flight"], false, "{st}");

    drop(hold);
    assert!(
        wait_pings(&bodies, 1, PATIENCE).await,
        "the held ping goes out once the hook has run"
    );
    assert_eq!(start_seam::calls(&handle.journal_id()), calls + 1, "the hook ran once");
    end(&mut t).await;
}

/// `/keepwarm off` while the ping waits: nothing goes out after the hook.
/// A later period is not left without its timer.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn keepwarm_off_while_a_ping_waits_for_the_hook_sends_nothing() {
    let h = Home::new();
    write_config(&h, "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n");
    let _interval = fast_pings();
    let (mut t, handle, bodies, hold, calls) = period_with_a_pending_hook(false).await;

    let st = keepwarm(&mut t, 710, "off").await;
    assert_eq!(st["active"], false, "{st}");
    drop(hold);
    assert!(hooks_run(&handle.journal_id(), calls, 1, PATIENCE).await, "the hook runs");
    tokio::time::sleep(Duration::from_millis(900)).await;
    assert_eq!(pings(&bodies), 0, "keep-warm was turned off before the hook ran");

    // Back on: the new period has a timer again and pings.
    let st = keepwarm(&mut t, 711, "auto").await;
    assert_eq!(st["active"], true, "{st}");
    assert!(
        wait_pings(&bodies, 1, PATIENCE).await,
        "a period started after the hook pings on schedule"
    );
    end(&mut t).await;
}

/// A turn submitted while the ping waits also waits for the hook; then the
/// turn's request goes out, and no ping goes out before it.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn a_turn_while_a_ping_waits_for_the_hook_goes_first_and_alone() {
    let h = Home::new();
    write_config(&h, "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n");
    let _interval = fast_pings();
    let (mut t, handle, bodies, hold, calls) = period_with_a_pending_hook(false).await;

    let before = bodies.lock().unwrap().len();
    t.send(submit("second")).await.unwrap();
    until(&mut t, |e| {
        matches!(e, agent_engine::session::SessionEventWire::TurnStarted { .. })
    })
    .await;
    tokio::time::sleep(Duration::from_millis(600)).await;
    assert_eq!(bodies.lock().unwrap().len(), before, "the turn waits for the hook too");

    drop(hold);
    until(&mut t, |e| matches!(e, agent_engine::session::SessionEventWire::Idle)).await;
    {
        let all = bodies.lock().unwrap();
        let next = all.get(before).expect("the turn sent its request");
        assert!(!next.contains(PING_MARK), "the turn's request comes first, not the held ping");
        assert!(next.contains("second"), "it is the second turn's request");
    }
    assert_eq!(start_seam::calls(&handle.journal_id()), calls + 1, "the hook ran once");
    end(&mut t).await;
}

/// Park while the ping waits: the hook leaves the line (it is not called
/// while parked) and no ping goes out; after the unpark the hook runs once
/// and turns work.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn park_while_a_ping_waits_for_the_hook_withdraws_both() {
    let h = Home::new();
    write_config(&h, "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n");
    let _interval = fast_pings();
    let (t, handle, bodies, hold, calls) = period_with_a_pending_hook(true).await;

    t.send(SessionCommand::Detach {
        client: t.client_id(),
    })
    .await
    .unwrap();
    handle.send(SessionCommand::Park).await.unwrap();
    assert!(
        wait_lifecycle(&handle, SessionLifecycle::Parked, PATIENCE).await,
        "an explicit park is not deferred"
    );
    drop(hold);
    tokio::time::sleep(Duration::from_millis(900)).await;
    assert_eq!(start_seam::calls(&handle.journal_id()), calls, "no hook runs while parked");
    assert_eq!(pings(&bodies), 0, "no ping while parked");

    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    assert!(hooks_run(&handle.journal_id(), calls, 1, PATIENCE).await, "the hook runs after the unpark");
    let st = keepwarm(&mut t, 720, "status").await;
    assert_eq!(st["last_end"], "parked by request", "{st}");
    let before = bodies.lock().unwrap().len();
    turn(&mut t, "after").await;
    {
        let all = bodies.lock().unwrap();
        let next = all.get(before).expect("the turn sent its request");
        assert!(!next.contains(PING_MARK) && next.contains("after"), "the turn's own request");
    }
    assert_eq!(start_seam::calls(&handle.journal_id()), calls + 1, "the hook ran once in all");
    end(&mut t).await;
}

// ── A due ping while fenced for a reload AND waiting for the hook ───────────
//
// A reload's checkpoint fences the session until the exec (or the daemon's
// report that it failed): the exec would discard the ping's result. The
// ping also waits for `on_session_start`. Fenced and waiting, the due ping
// goes out only once both have cleared, in either order.
//
// What the waits show: the 900 ms sleeps are negative observations (no ping
// went out within them), not barriers, and `wait_pings(.., 1, ..)` waits
// for at least one ping: these tests check that the ping is held and then
// goes out, not that exactly one ever does.

/// Fence `handle` as a reload's checkpoint does (host-sent).
async fn fence_for_reload(t: &mut LocalTransport, handle: &SessionHandle) {
    use agent_engine::session::wire::CHECKPOINT_QUERY_ID;
    handle
        .send(SessionCommand::Checkpoint {
            reason: agent_engine::session::CheckpointReason::Reload,
        })
        .await
        .unwrap();
    until(t, |e| {
        matches!(e, agent_engine::session::SessionEventWire::QueryResult { id, .. } if *id == CHECKPOINT_QUERY_ID)
    })
    .await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn a_due_ping_while_fenced_and_waiting_for_the_hook_goes_out_once_both_clear_fence_first() {
    let h = Home::new();
    write_config(&h, "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n");
    let _interval = fast_pings();
    let (mut t, handle, bodies, hold, calls) = period_with_a_pending_hook(false).await;
    fence_for_reload(&mut t, &handle).await;
    tokio::time::sleep(Duration::from_millis(900)).await;
    assert_eq!(pings(&bodies), 0, "no ping while fenced and waiting");

    handle.send(SessionCommand::ReloadAborted).await.unwrap();
    tokio::time::sleep(Duration::from_millis(900)).await;
    assert_eq!(pings(&bodies), 0, "the fence lifted; the ping still waits for the hook");
    assert_eq!(start_seam::calls(&handle.journal_id()), calls, "the hook still waits");

    drop(hold);
    assert!(
        wait_pings(&bodies, 1, PATIENCE).await,
        "the held ping goes out once both have cleared"
    );
    assert_eq!(start_seam::calls(&handle.journal_id()), calls + 1, "the hook ran once");
    end(&mut t).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn a_due_ping_while_fenced_and_waiting_for_the_hook_goes_out_once_both_clear_hook_first() {
    let h = Home::new();
    write_config(&h, "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n");
    let _interval = fast_pings();
    let (mut t, handle, bodies, hold, calls) = period_with_a_pending_hook(false).await;
    fence_for_reload(&mut t, &handle).await;

    drop(hold);
    assert!(
        hooks_run(&handle.journal_id(), calls, 1, PATIENCE).await,
        "the hook runs under the fence"
    );
    tokio::time::sleep(Duration::from_millis(900)).await;
    assert_eq!(pings(&bodies), 0, "the hook ran; the ping still waits for the fence");

    handle.send(SessionCommand::ReloadAborted).await.unwrap();
    assert!(
        wait_pings(&bodies, 1, PATIENCE).await,
        "the held ping goes out once both have cleared"
    );
    assert_eq!(start_seam::calls(&handle.journal_id()), calls + 1, "the hook ran once");
    end(&mut t).await;
}

/// A pin is the owner's say-so: a pinned session with no interactive client
/// ever attached (restored by the daemon, fed by `synaps send`) starts a
/// period on the Idle after a real turn. Unpinned, the old rule holds:
/// no typing client, no period; unpinning ends a period it started.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn a_pinned_session_keeps_warm_without_an_interactive_client() {
    let h = Home::new();
    write_config(&h, "cache_keepwarm = auto\ncache_ttl = 1h\nthinking = off\n");
    let (url, bodies) = recording_stub().await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;

    // Unpinned, script-only: no period.
    let plain = host.create_session(cfg()).await.unwrap();
    let (mut script, _) = LocalTransport::attach(plain.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    turn(&mut script, "hello").await;
    let st = keepwarm(&mut script, 800, "status").await;
    assert_eq!(st["active"], false, "unpinned + no interactive client: none: {st}");
    end(&mut script).await;

    // Pinned at create, script-only: the Idle starts one.
    let pinned = host.create_session(SessionConfig { keep_warm: true, ..cfg() }).await.unwrap();
    let (mut script, _) = LocalTransport::attach(pinned.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    turn(&mut script, "hello").await;
    let st = keepwarm(&mut script, 801, "status").await;
    assert_eq!(st["active"], true, "pinned, no interactive client: a period: {st}");

    // Unpinned again: back to the old rule, the period ends.
    pinned.send(SessionCommand::KeepWarm { on: false }).await.unwrap();
    let st = keepwarm(&mut script, 802, "status").await;
    assert_eq!(st["active"], false, "unpinned: no interactive client, no period: {st}");
    let _ = pings(&bodies);
    end(&mut script).await;
}

/// An SSE reply: `read` cached, `write` written at 5m, 10 uncached input
/// tokens, `output` output tokens (a ping's thinking is billed as output).
fn sse_5m(read: u64, write: u64, output: u64) -> String {
    let usage = format!(
        "\"input_tokens\":10,\"cache_creation_input_tokens\":{write},\
         \"cache_read_input_tokens\":{read},\"cache_creation\":\
         {{\"ephemeral_5m_input_tokens\":{write},\"ephemeral_1h_input_tokens\":0}}"
    );
    format!(
        "data: {{\"type\":\"message_start\",\"message\":{{\"id\":\"msg_kw\",\"type\":\"message\",\
         \"role\":\"assistant\",\"content\":[],\"model\":\"{MODEL}\",\"stop_reason\":null,\
         \"stop_sequence\":null,\"usage\":{{\"output_tokens\":0,{usage}}}}}}}\n\n\
         data: {{\"type\":\"content_block_start\",\"index\":0,\
         \"content_block\":{{\"type\":\"text\",\"text\":\"\"}}}}\n\n\
         data: {{\"type\":\"content_block_delta\",\"index\":0,\
         \"delta\":{{\"type\":\"text_delta\",\"text\":\"ok\"}}}}\n\n\
         data: {{\"type\":\"content_block_stop\",\"index\":0}}\n\n\
         data: {{\"type\":\"message_delta\",\"delta\":{{\"stop_reason\":\"end_turn\",\
         \"stop_sequence\":null}},\"usage\":{{\"output_tokens\":{output},{usage}}}}}\n\n\
         data: {{\"type\":\"message_stop\"}}\n\n"
    )
}

/// A 5,011-token prefix: turns read 4,000 and write 1,000 (+10 input, 1
/// output); pings read 5,000 and write the 11-token tail, with
/// `ping_output` output tokens.
async fn small_prefix_stub(
    ping_output: Arc<std::sync::atomic::AtomicU64>,
) -> (String, Arc<Mutex<Vec<String>>>) {
    let bodies = Arc::new(Mutex::new(Vec::new()));
    let b = Arc::clone(&bodies);
    let app = axum::Router::new().fallback(move |body: String| {
        let b = Arc::clone(&b);
        let out = Arc::clone(&ping_output);
        async move {
            let streaming = body.contains("\"stream\":true");
            let ping = body.contains(PING_MARK);
            b.lock().unwrap().push(body);
            if !streaming {
                return axum::Json(serde_json::json!({
                    "id": "msg_c", "type": "message", "role": "assistant",
                    "model": MODEL, "stop_reason": "end_turn",
                    "content": [{"type": "text", "text": "summary"}],
                    "usage": {"input_tokens": 10, "output_tokens": 5}
                }))
                .into_response();
            }
            let sse = if ping {
                sse_5m(5_000, 11, out.load(std::sync::atomic::Ordering::SeqCst))
            } else {
                sse_5m(4_000, 1_000, 1)
            };
            (
                axum::http::StatusCode::OK,
                [("content-type", "text/event-stream")],
                sse,
            )
                .into_response()
        }
    });
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
    (format!("http://{addr}"), bodies)
}

/// The top-level members of a JSON object body as the raw bytes sent.
fn raw_members(body: &str) -> std::collections::BTreeMap<String, String> {
    let b = body.as_bytes();
    let mut out = std::collections::BTreeMap::new();
    let mut i = body.find('{').expect("object") + 1;
    let skip_ws = |i: &mut usize| {
        while *i < b.len() && (b[*i] as char).is_ascii_whitespace() {
            *i += 1;
        }
    };
    // End (exclusive) of the string starting at `i` (on its opening quote).
    let string_end = |mut i: usize| {
        i += 1;
        while b[i] != b'"' {
            i += if b[i] == b'\\' { 2 } else { 1 };
        }
        i + 1
    };
    loop {
        skip_ws(&mut i);
        if b[i] == b'}' {
            return out;
        }
        let k_end = string_end(i);
        let key: String = serde_json::from_str(&body[i..k_end]).unwrap();
        i = k_end;
        skip_ws(&mut i);
        assert_eq!(b[i], b':');
        i += 1;
        skip_ws(&mut i);
        let start = i;
        let mut depth = 0usize;
        while depth > 0 || (b[i] != b',' && b[i] != b'}') {
            match b[i] {
                b'"' => {
                    i = string_end(i);
                    continue;
                }
                b'{' | b'[' => depth += 1,
                b'}' | b']' => depth -= 1,
                _ => {}
            }
            i += 1;
        }
        out.insert(key, body[start..i].trim_end().to_string());
        if b[i] == b',' {
            i += 1;
        }
    }
}

/// A message as the cache sees it: string content is the documented
/// shorthand for one text block (a turn that marks the message sends the
/// block form, the next request the shorthand), and markers are not content.
fn strip_markers(mut msg: serde_json::Value) -> serde_json::Value {
    if let Some(text) = msg["content"].as_str() {
        msg["content"] = serde_json::json!([{"type": "text", "text": text}]);
    }
    if let Some(blocks) = msg["content"].as_array_mut() {
        for b in blocks {
            if let Some(o) = b.as_object_mut() {
                o.remove("cache_control");
            }
        }
    }
    msg
}

/// Budgeted thinking on a 5K prefix through the real session actor. Before
/// the expected-cost decision no period ever started here (a worst-case
/// ping, 4,097 output tokens, costs more than the $0.0188 rewrite). Now:
/// - pings go out, and every cache-key field (everything but `messages` and
///   `max_tokens`, `thinking` included) is byte-identical to the turn whose
///   cache it keeps warm, the history byte-identical beyond its markers;
/// - pings within the output estimate never spend more than the rewrite;
/// - a ping that out-thinks the estimate stays inside the reported bound
///   (rewrite + thinking reserve), and from then on the session prices
///   pings at their worst case, which refuses this prefix again.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn budgeted_thinking_small_prefix_pings_within_the_reported_bound() {
    use agent_engine::core::config::CacheTtl;
    use agent_engine::session::cache_keepwarm::{
        request_cost_usd, rewrite_cost_usd, PING_INPUT_ESTIMATE, PING_OUTPUT_ESTIMATE,
    };
    use std::sync::atomic::{AtomicU64, Ordering};
    let h = Home::new();
    write_config(
        &h,
        "cache_keepwarm = auto\ncache_ttl = 5m\nthinking = medium\n",
    );
    let _interval = fast_pings();
    let ping_output = Arc::new(AtomicU64::new(100));
    let (url, bodies) = small_prefix_stub(Arc::clone(&ping_output)).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host.create_session(cfg()).await.unwrap();
    let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    let ended = |st: &serde_json::Value| {
        st["keepwarm"]["period"].is_null()
            && st["keepwarm"]["last_end"]
                .as_str()
                .is_some_and(|s| s.contains("bound reached"))
    };
    let rewrite = rewrite_cost_usd(MODEL, CacheTtl::FiveMinutes, 5_011);
    let worst = |out| {
        request_cost_usd(
            MODEL,
            CacheTtl::FiveMinutes,
            5_000,
            11,
            PING_INPUT_ESTIMATE,
            out,
        )
    };
    let reserve = worst(4_097) - worst(PING_OUTPUT_ESTIMATE);

    // Period 1: pings think 100 tokens (within the estimate).
    turn(&mut t, "hello").await;
    let st = status_until(&mut t, 100, PATIENCE, ended).await;
    assert!(ended(&st), "{st}");
    let first = pings(&bodies);
    assert!(first >= 2, "{first} pings: {st}");
    let spent = st["keepwarm_cost"].as_f64().unwrap();
    assert!(spent <= rewrite * (1.0 + 1e-9), "${spent} > ${rewrite} rewrite");
    let last_end = st["keepwarm"]["last_end"].as_str().unwrap();
    assert!(last_end.contains("thinking reserve"), "{last_end}");

    // The cache-key fields, as sent.
    let (turn_body, ping_body) = {
        let all = bodies.lock().unwrap();
        let turn = all
            .iter()
            .find(|b| b.contains("\"stream\":true") && !b.contains(PING_MARK))
            .unwrap()
            .clone();
        let ping = all.iter().find(|b| b.contains(PING_MARK)).unwrap().clone();
        (turn, ping)
    };
    let (mut tr, mut pr) = (raw_members(&turn_body), raw_members(&ping_body));
    let (tm, pm) = (tr.remove("messages").unwrap(), pr.remove("messages").unwrap());
    // `max_tokens` is not part of the cache key: the turn keeps its own.
    let turn_max = tr.remove("max_tokens").unwrap();
    let ping_max = pr.remove("max_tokens").unwrap();
    assert_ne!(turn_max, ping_max);
    assert_eq!(tr, pr, "cache-key fields differ between the turn and its ping");
    let thinking: serde_json::Value = serde_json::from_str(&pr["thinking"]).unwrap();
    assert_eq!(thinking["type"], "enabled", "{thinking}");
    assert_eq!(thinking["budget_tokens"], 4096, "{thinking}");
    assert_eq!(ping_max, "4097", "budget + 1: the API's minimum");
    let tm: Vec<serde_json::Value> = serde_json::from_str(&tm).unwrap();
    let pm: Vec<serde_json::Value> = serde_json::from_str(&pm).unwrap();
    assert_eq!(pm.len(), tm.len() + 2, "history + reply + the ping");
    for (i, m) in tm.iter().enumerate() {
        assert_eq!(strip_markers(pm[i].clone()), strip_markers(m.clone()), "message {i}");
    }

    // Period 2: the first ping out-thinks the estimate. It still fits the
    // reported bound; nothing more fits after it.
    ping_output.store(2_000, Ordering::SeqCst);
    turn(&mut t, "again").await;
    let st = status_until(&mut t, 200, PATIENCE, |st| {
        ended(st) && pings(&bodies) > first
    })
    .await;
    assert!(ended(&st), "{st}");
    assert_eq!(pings(&bodies), first + 1, "{st}");
    let spent2 = st["keepwarm_cost"].as_f64().unwrap() - spent;
    assert!(spent2 > rewrite, "the overrun is real: ${spent2}");
    assert!(spent2 <= (rewrite + reserve) * (1.0 + 1e-9), "${spent2} > bound");

    // Period 3: priced at the worst case now, this prefix is refused again.
    turn(&mut t, "third").await;
    let st = status_until(&mut t, 300, PATIENCE, |st| {
        st["keepwarm"]["last_end"]
            .as_str()
            .is_some_and(|s| s.starts_with("not started"))
    })
    .await;
    let last_end = st["keepwarm"]["last_end"].as_str().unwrap();
    assert!(last_end.contains("worst-case cache-hit ping"), "{last_end}");
    tokio::time::sleep(Duration::from_millis(900)).await;
    assert_eq!(pings(&bodies), first + 1, "no ping after the refusal");
    handle
        .send(SessionCommand::End {
            reason: EndReason::HostShutdown,
        })
        .await
        .unwrap();
}