supercode-harness 0.4.16

//! BP-8 (catalog Domain 5, "Session / state model") — the nine session/state
//! rows proved through the RESOLVED `cc-parity` / `cx-parity` presets.
//!
//! Nothing here unit-tests an unwired function. Every assertion resolves the
//! preset TOML through [`supercode_harness::configfile::resolve`], builds the
//! [`Agent`] the CLI's own session doors build, and runs the SAME composition
//! those doors run — `session_journal::arm` on the way in and
//! `session_journal::checkpoint` on the way out (see `persist_session` /
//! `arm_session_state` in `crates/cli/src/main.rs`, which are thin narrators
//! over exactly these two calls).
//!
//! Persistence claims are proved across a DROP: the agent is destroyed, a
//! fresh one is built from the same resolved preset, and the state has to
//! come back off disk. Offline and hermetic: a scripted in-process provider,
//! and a temp-directory `SessionStore` — no real harness home is read or
//! written, and no model is ever called.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use supercode_harness::configfile::{resolve, ResolveOptions};
use supercode_harness::session_journal::{self, JournalOp, PlanEntry, QueueKind};
use supercode_harness::store::SessionStore;
use supercode_harness::{
    Agent, ChatMessage, ChatRequest, Config, FunctionCall, Provider, Role, ToolCall, Usage,
};

// ---------------------------------------------------------------------------
// The real path: a resolved preset, an Agent, a store.
// ---------------------------------------------------------------------------

fn temp_root(tag: &str) -> std::path::PathBuf {
    let p = std::env::temp_dir().join(format!(
        "supercode-bp8-{tag}-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    std::fs::create_dir_all(&p).unwrap();
    p
}

/// The resolved parity preset, exactly as the CLI resolves it, with only
/// `cwd` moved into the test's own tree.
fn parity_config(preset: &str, cwd: &std::path::Path) -> Config {
    let mut resolved = resolve(
        &format!("extends = \"{preset}\"\n"),
        None,
        &ResolveOptions { strict: true },
    )
    .unwrap_or_else(|e| panic!("preset `{preset}` failed to resolve: {e}"));
    resolved.config.cwd = cwd.to_path_buf();
    resolved.config
}

/// A provider that answers with a fixed reply and never calls a tool.
struct Answering(&'static str);

#[async_trait]
impl Provider for Answering {
    async fn complete(
        &self,
        _req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        Ok((ChatMessage::assistant(self.0), Usage::default()))
    }
}

/// A provider whose FIRST reply calls `update_plan` with `plan`, and whose
/// second reply is the final answer.
struct Planning {
    plan: serde_json::Value,
    calls: AtomicUsize,
}

#[async_trait]
impl Provider for Planning {
    async fn complete(
        &self,
        _req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        if self.calls.fetch_add(1, Ordering::SeqCst) == 0 {
            let call = ChatMessage {
                role: Role::Assistant,
                content: None,
                content_parts: None,
                tool_calls: Some(vec![ToolCall {
                    id: "call_plan".into(),
                    kind: "function".into(),
                    function: FunctionCall {
                        name: "update_plan".into(),
                        arguments: self.plan.to_string(),
                    },
                }]),
                tool_call_id: None,
                name: None,
                metadata: Default::default(),
            };
            return Ok((call, Usage::default()));
        }
        Ok((ChatMessage::assistant("planned"), Usage::default()))
    }
}

/// A provider that blocks until `release` is set, so a turn can be observed
/// (and abandoned) MID-FLIGHT — the crash this ledger row is about.
struct Hanging {
    seen_first_turn: Arc<Mutex<bool>>,
}

#[async_trait]
impl Provider for Hanging {
    async fn complete(
        &self,
        _req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        *self.seen_first_turn.lock().unwrap() = true;
        Ok((ChatMessage::assistant("mid-turn reply"), Usage::default()))
    }
}

/// Persist the working view the way `persist_session` does: the whole
/// history as JSONL, then the journal checkpoint + plan + tree.
fn persist(agent: &Agent, store: &SessionStore, name: &str) {
    let jsonl: String = agent
        .history()
        .iter()
        .filter_map(|m| serde_json::to_string(m).ok())
        .collect::<Vec<_>>()
        .join("\n");
    store.save(name, "t", &jsonl).unwrap();
    session_journal::checkpoint(agent, store, name, agent.history().len());
}

/// Reopen the way every CLI session door does: a fresh agent off the same
/// resolved preset, the stored transcript loaded, then `arm`.
fn reopen(
    preset: &str,
    cwd: &std::path::Path,
    store: &SessionStore,
    name: &str,
    provider: Box<dyn Provider>,
) -> (Agent, session_journal::RestoreReport) {
    let mut agent = Agent::with_provider(parity_config(preset, cwd), provider);
    if let Ok(Some(jsonl)) = store.load_if_present(name) {
        let tmp = cwd.join(format!("reopen-{name}.jsonl"));
        std::fs::write(&tmp, jsonl).unwrap();
        agent.load_transcript(&tmp).unwrap();
        let _ = std::fs::remove_file(&tmp);
    }
    let report = session_journal::arm(&mut agent, store, name);
    (agent, report)
}

// ---------------------------------------------------------------------------
// catalog:150 — Append-only durable transcript
// ---------------------------------------------------------------------------

/// Both presets set `[core.session] append_only`, so both arm a journal —
/// and the journal has the turn's messages on disk BEFORE the end-of-turn
/// transcript rewrite, which is the whole difference the row names.
#[tokio::test]
async fn a_turn_is_on_disk_before_the_transcript_is_rewritten_under_both_presets() {
    for preset in ["cc-parity", "cx-parity"] {
        let root = temp_root("append-only");
        let store = SessionStore::open(root.join("sessions")).unwrap();
        let name = "sess-append";

        let mut agent = Agent::with_provider(
            parity_config(preset, &root),
            Box::new(Answering("first answer")),
        );
        assert!(
            agent.config().session_append_only,
            "{preset} must set core.session.append_only"
        );
        let report = session_journal::arm(&mut agent, &store, name);
        assert!(report.journal_armed, "{preset}: the journal must be armed");

        agent.send("hello").await.unwrap();

        // NOT persisted yet — this is the crash window.
        assert!(
            store.load_if_present(name).unwrap().is_none(),
            "{preset}: the transcript must not exist yet"
        );
        let state = store
            .load_journal(name)
            .unwrap()
            .expect("the journal must exist mid-turn");
        let bodies: Vec<String> = state
            .messages
            .iter()
            .map(|m| m.content.clone().unwrap_or_default())
            .collect();
        assert_eq!(
            bodies,
            ["hello", "first answer"],
            "{preset}: every message must already be flushed"
        );
    }
}

/// The recovery this durability buys: a process that dies after the turn but
/// before the transcript rewrite loses nothing. The next open reads the
/// journal past its last checkpoint and puts the turn back.
#[tokio::test]
async fn a_turn_lost_to_a_crash_comes_back_on_the_next_open() {
    for preset in ["cc-parity", "cx-parity"] {
        let root = temp_root("recover");
        let store = SessionStore::open(root.join("sessions")).unwrap();
        let name = "sess-recover";
        let seen = Arc::new(Mutex::new(false));

        // Turn 1, cleanly persisted.
        let mut agent = Agent::with_provider(
            parity_config(preset, &root),
            Box::new(Answering("answer one")),
        );
        session_journal::arm(&mut agent, &store, name);
        agent.send("turn one").await.unwrap();
        persist(&agent, &store, name);
        drop(agent);

        // Turn 2 runs, then the process "crashes": no persist at all.
        let (mut agent, _) = reopen(
            preset,
            &root,
            &store,
            name,
            Box::new(Hanging {
                seen_first_turn: seen.clone(),
            }),
        );
        agent.send("turn two").await.unwrap();
        assert!(*seen.lock().unwrap());
        let before_crash = agent.history().len();
        drop(agent);

        // The transcript on disk still only knows turn one …
        let persisted = store.load(name).unwrap();
        assert!(!persisted.contains("turn two"));

        // … and the next open recovers it from the journal.
        let (agent, report) = reopen(preset, &root, &store, name, Box::new(Answering("later")));
        assert_eq!(
            report.recovered_messages, 2,
            "{preset}: the user turn and the reply must both come back"
        );
        assert_eq!(agent.history().len(), before_crash, "{preset}");
        assert!(
            agent
                .history()
                .iter()
                .any(|m| m.content.as_deref() == Some("turn two")),
            "{preset}: the recovered user turn must be in the live conversation"
        );
    }
}

/// The FIRST turn of a brand-new session is covered too: a session that
/// crashed before it ever wrote a transcript has no checkpoint record, and
/// its recovery base is the bare system message.
#[tokio::test]
async fn even_a_brand_new_sessions_first_turn_survives_a_crash() {
    let root = temp_root("first-turn");
    let store = SessionStore::open(root.join("sessions")).unwrap();
    let name = "sess-first";

    let mut agent = Agent::with_provider(
        parity_config("cc-parity", &root),
        Box::new(Answering("the only answer")),
    );
    let report = session_journal::arm(&mut agent, &store, name);
    assert!(report.journal_armed);
    // Arming alone must leave no file behind: a session the user opens and
    // quits without saying anything is not a session.
    assert!(
        !store.journal_path(name).unwrap().exists(),
        "the journal must not exist before the first message"
    );

    agent.send("the only question").await.unwrap();
    drop(agent);
    assert!(store.load_if_present(name).unwrap().is_none());

    let (agent, report) = reopen(
        "cc-parity",
        &root,
        &store,
        name,
        Box::new(Answering("later")),
    );
    assert_eq!(report.recovered_messages, 2);
    assert!(agent
        .history()
        .iter()
        .any(|m| m.content.as_deref() == Some("the only question")));
}

// ---------------------------------------------------------------------------
// catalog:151 — In-place conversation tree
// ---------------------------------------------------------------------------

/// cc-parity turns `[capabilities.session_tree]` on, and BP-8 gives that key
/// its reader: an agent under cc-parity builds a tree as it talks, the tree
/// projects back to the linear conversation, and the tree survives a drop.
/// cx-parity leaves the module off and gets no tree at all.
#[tokio::test]
async fn cc_parity_grows_a_conversation_tree_and_cx_parity_does_not() {
    let root = temp_root("tree");
    let store = SessionStore::open(root.join("sessions")).unwrap();

    let mut agent =
        Agent::with_provider(parity_config("cc-parity", &root), Box::new(Answering("a1")));
    session_journal::arm(&mut agent, &store, "sess-tree");
    agent.send("u1").await.unwrap();
    agent.send("u2").await.unwrap();
    let tree = agent
        .session_tree()
        .expect("cc-parity arms the tree module");
    assert_eq!(tree.nodes.len(), 4, "one node per recorded message");
    let projected: Vec<String> = tree
        .linear_projection()
        .unwrap()
        .iter()
        .map(|m| m.content.clone().unwrap_or_default())
        .collect();
    assert_eq!(projected, ["u1", "a1", "u2", "a1"]);
    persist(&agent, &store, "sess-tree");
    drop(agent);

    let (agent, report) = reopen(
        "cc-parity",
        &root,
        &store,
        "sess-tree",
        Box::new(Answering("a2")),
    );
    assert!(report.tree_loaded, "the tree must come back off disk");
    assert_eq!(agent.session_tree().unwrap().nodes.len(), 4);

    let mut cx = Agent::with_provider(parity_config("cx-parity", &root), Box::new(Answering("a1")));
    session_journal::arm(&mut cx, &store, "sess-tree-cx");
    cx.send("u1").await.unwrap();
    assert!(
        cx.session_tree().is_none(),
        "cx-parity leaves capabilities.session_tree off"
    );
}

// ---------------------------------------------------------------------------
// catalog:152 — Rewind / rollback conversation
// ---------------------------------------------------------------------------

/// The conversation rewind: an arbitrary point, in place, recorded, and
/// invertible — under BOTH presets, since the journal (not the tree, and not
/// `capabilities.checkpoint`) is what makes it reversible.
#[tokio::test]
async fn rewinding_to_an_arbitrary_point_is_recorded_and_invertible() {
    for preset in ["cc-parity", "cx-parity"] {
        let root = temp_root("rewind");
        let store = SessionStore::open(root.join("sessions")).unwrap();
        let name = "sess-rewind";

        let mut agent =
            Agent::with_provider(parity_config(preset, &root), Box::new(Answering("a")));
        session_journal::arm(&mut agent, &store, name);
        for turn in ["u1", "u2", "u3"] {
            agent.send(turn).await.unwrap();
        }
        assert_eq!(agent.history().len(), 7, "{preset}: system + 3 exchanges");

        // Back to just before `u2` — an ARBITRARY point, not the last one.
        let outcome = agent.rewind_conversation(3);
        assert_eq!(outcome.removed, 4, "{preset}");
        assert_eq!(
            agent
                .history()
                .iter()
                .filter_map(|m| m.content.clone())
                .filter(|c| c.starts_with('u'))
                .collect::<Vec<_>>(),
            ["u1"],
            "{preset}"
        );
        // cc-parity's tree preserves the rewound path under its own branch;
        // cx-parity has no tree and relies on the journal alone.
        if preset == "cc-parity" {
            let branch = outcome
                .preserved_branch
                .as_deref()
                .expect("cc-parity preserves the old leaf under a sibling branch");
            let kept: Vec<String> = agent
                .session_tree()
                .unwrap()
                .linear_projection_of(branch)
                .unwrap()
                .iter()
                .map(|m| m.content.clone().unwrap_or_default())
                .collect();
            assert_eq!(kept, ["u1", "a", "u2", "a", "u3", "a"]);
        } else {
            assert!(outcome.preserved_branch.is_none());
        }

        // Nothing was deleted: the journal still holds every message, and
        // records the rewind as an appended operation.
        let raw = std::fs::read_to_string(store.journal_path(name).unwrap()).unwrap();
        assert!(raw.contains("\"op\":\"rewind\""), "{preset}");
        assert!(raw.contains("\"content\":\"u3\""), "{preset}");

        // And it inverts.
        assert!(agent.undo_rewind(), "{preset}");
        assert_eq!(agent.history().len(), 7, "{preset}");
        assert!(std::fs::read_to_string(store.journal_path(name).unwrap())
            .unwrap()
            .contains("\"op\":\"unrewind\""));

        // The undo survives a restart: rewind, persist, drop, reopen.
        agent.rewind_conversation(3);
        persist(&agent, &store, name);
        drop(agent);
        let (mut agent, report) = reopen(preset, &root, &store, name, Box::new(Answering("a")));
        assert_eq!(report.restored_rewinds, 1, "{preset}");
        assert!(agent.undo_rewind(), "{preset}: the undo crossed a restart");
        assert_eq!(agent.history().len(), 7, "{preset}");
    }
}

// ---------------------------------------------------------------------------
// catalog:152 — Session naming/rename
// ---------------------------------------------------------------------------

/// `SessionStore::rename` moves the RESUME HANDLE: the whole family follows,
/// the old handle stops resolving, and the new one resumes the same
/// conversation. `set_title` (the pre-BP-8 half) leaves the handle alone.
#[tokio::test]
async fn rename_moves_the_resume_handle_and_the_whole_family() {
    let root = temp_root("rename");
    let store = SessionStore::open(root.join("sessions")).unwrap();

    let mut agent =
        Agent::with_provider(parity_config("cc-parity", &root), Box::new(Answering("a1")));
    session_journal::arm(&mut agent, &store, "old-handle");
    agent.send("hello").await.unwrap();
    persist(&agent, &store, "old-handle");
    drop(agent);

    let before: Vec<String> = std::fs::read_dir(store.root())
        .unwrap()
        .flatten()
        .map(|e| e.file_name().to_string_lossy().into_owned())
        .filter(|n| n.starts_with("old-handle."))
        .collect();
    assert!(
        before.len() >= 3,
        "the family should have transcript + meta + journal at least: {before:?}"
    );

    store.rename("old-handle", "new-handle").unwrap();
    assert!(store.load_if_present("old-handle").unwrap().is_none());
    assert!(store.load("new-handle").unwrap().contains("hello"));
    for member in &before {
        let moved = member.replacen("old-handle.", "new-handle.", 1);
        assert!(
            store.root().join(&moved).exists(),
            "family member `{member}` did not follow the rename"
        );
        assert!(!store.root().join(member).exists());
    }
    assert!(store.list().iter().any(|s| s.name == "new-handle"));
    assert!(!store.list().iter().any(|s| s.name == "old-handle"));

    // The new handle really resumes the same conversation.
    let (agent, _) = reopen(
        "cc-parity",
        &root,
        &store,
        "new-handle",
        Box::new(Answering("a2")),
    );
    assert!(agent
        .history()
        .iter()
        .any(|m| m.content.as_deref() == Some("hello")));

    // A collision refuses rather than half-moving.
    store.save("taken", "t", "").unwrap();
    assert!(store.rename("new-handle", "taken").is_err());
    assert!(store.load("new-handle").unwrap().contains("hello"));
}

// ---------------------------------------------------------------------------
// catalog:151 — Derived index / cache over transcripts
// ---------------------------------------------------------------------------

/// The index is a CACHE: it survives the process, revalidates each row
/// against its transcript, re-derives only the rows that moved, and produces
/// an identical answer when thrown away entirely.
#[test]
fn the_derived_index_survives_the_process_and_rebuilds_identically() {
    let root = temp_root("index");
    let store = SessionStore::open(root.join("sessions")).unwrap();
    for (name, body) in [
        (
            "sess-a",
            "{\"role\":\"user\",\"content\":\"first question\"}",
        ),
        (
            "sess-b",
            "{\"role\":\"user\",\"content\":\"second question\"}",
        ),
    ] {
        store.save(name, name, body).unwrap();
    }

    let cold = store.index();
    assert_eq!(cold.rederived, 2, "a cold index derives every row");
    assert_eq!(cold.reused, 0);
    assert!(store.index_path().exists(), "the cache must be written");
    assert!(cold
        .entries
        .iter()
        .any(|e| e.preview == "first question" && e.messages == 1));

    // A SECOND store handle — a different process, as far as the cache is
    // concerned — serves both rows without opening a transcript.
    let other = SessionStore::at(store.root());
    let warm = other.index();
    assert_eq!(warm.reused, 2, "every unchanged row comes from the cache");
    assert_eq!(warm.rederived, 0);
    assert_eq!(warm.entries, cold.entries);

    // A changed transcript invalidates exactly its own row.
    std::thread::sleep(std::time::Duration::from_millis(5));
    store
        .save(
            "sess-a",
            "sess-a",
            "{\"role\":\"user\",\"content\":\"first question\"}\n{\"role\":\"assistant\",\"content\":\"ok\"}",
        )
        .unwrap();
    let repaired = store.index();
    assert_eq!(repaired.rederived, 1, "only the moved row is re-read");
    assert_eq!(repaired.reused, 1);
    assert_eq!(
        repaired
            .entries
            .iter()
            .find(|e| e.name == "sess-a")
            .unwrap()
            .messages,
        2
    );

    // The transcripts stay the only record: drop the cache, get the same
    // answer back.
    let with_cache = store.index().entries;
    store.invalidate_index();
    assert!(!store.index_path().exists());
    let rebuilt = store.index();
    assert_eq!(rebuilt.rederived, 2);
    assert_eq!(rebuilt.entries, with_cache);
}

// ---------------------------------------------------------------------------
// catalog:153 — Format versioning / migration
// ---------------------------------------------------------------------------

/// An older generation is upgraded IN PLACE on the first open, stamped so no
/// later open repeats it, and the original bytes are kept verbatim so the
/// upgrade is reversible.
#[test]
fn an_old_generation_is_upgraded_in_place_once_with_the_original_recoverable() {
    let root = temp_root("upgrade");
    let store = SessionStore::open(root.join("sessions")).unwrap();
    // A pre-BP-8 generation: keys this build does not model, explicit nulls,
    // and no version marker in the meta.
    let legacy = concat!(
        "{\"role\":\"user\",\"content\":\"hi\",\"tool_calls\":null,\"timestamp\":\"2026-01-01\"}\n",
        "{\"role\":\"assistant\",\"content\":\"hello\",\"legacy_id\":7}\n"
    );
    store.save("sess-old", "t", legacy).unwrap();
    // `save` stamps nothing, exactly like every pre-BP-8 write.
    assert_eq!(store.format_version("sess-old"), 0);
    let before: Vec<String> = store
        .load("sess-old")
        .unwrap()
        .lines()
        .map(|l| serde_json::from_str::<ChatMessage>(l).unwrap())
        .map(|m| m.content.unwrap_or_default())
        .collect();

    let up = store
        .upgrade_in_place("sess-old")
        .unwrap()
        .expect("an unmarked session must be upgraded");
    assert_eq!(up.from_version, 0);
    assert_eq!(up.to_version, SessionStore::FORMAT_VERSION);
    assert!(up.rewritten);
    assert_eq!(up.messages, 2);

    // The marker is stamped, so a second open is a no-op — the tolerant
    // path is paid once, not on every read.
    assert_eq!(
        store.format_version("sess-old"),
        SessionStore::FORMAT_VERSION
    );
    assert!(store.upgrade_in_place("sess-old").unwrap().is_none());

    // Lossless at the message level, and byte-canonical afterwards.
    let after = store.load("sess-old").unwrap();
    let after_msgs: Vec<String> = after
        .lines()
        .map(|l| serde_json::from_str::<ChatMessage>(l).unwrap())
        .map(|m| m.content.unwrap_or_default())
        .collect();
    assert_eq!(before, after_msgs);
    assert!(!after.contains("legacy_id"));
    assert!(!after.contains("\"tool_calls\":null"));

    // Reversible: the original bytes are on disk, verbatim.
    let original = std::fs::read_to_string(store.root().join(&up.original)).unwrap();
    assert_eq!(original, legacy);
}

// ---------------------------------------------------------------------------
// catalog:154 — Queued-prompt persistence
// ---------------------------------------------------------------------------

/// cc-parity sets `queue_persist` (CC writes queue-operation records); a
/// prompt typed while the agent was busy survives the process, and the next
/// turn consumes it exactly once. cx-parity deliberately does NOT — the
/// catalog scores that row `—` for Codex.
#[tokio::test]
async fn a_pending_input_survives_a_restart_under_cc_parity_only() {
    let root = temp_root("queue");
    let store = SessionStore::open(root.join("sessions")).unwrap();
    let name = "sess-queue";

    let mut agent =
        Agent::with_provider(parity_config("cc-parity", &root), Box::new(Answering("a1")));
    assert!(agent.config().session_queue_persist);
    session_journal::arm(&mut agent, &store, name);
    agent.send("first").await.unwrap();
    // Typed while the agent was busy, never consumed before the process died.
    agent.queue_follow_up("the queued follow-up");
    persist(&agent, &store, name);
    drop(agent);

    let (mut agent, report) = reopen("cc-parity", &root, &store, name, Box::new(Answering("a2")));
    assert_eq!(report.restored_queue, 1, "the pending input must come back");
    assert_eq!(agent.queued_steer_count(), 1);

    // It is consumed exactly once: the next turn delivers it, and a further
    // restart does not deliver it again.
    agent.send("second").await.unwrap();
    assert!(agent
        .history()
        .iter()
        .any(|m| m.content.as_deref() == Some("the queued follow-up")));
    persist(&agent, &store, name);
    drop(agent);
    let (_, report) = reopen("cc-parity", &root, &store, name, Box::new(Answering("a3")));
    assert_eq!(
        report.restored_queue, 0,
        "a consumed input must not be re-delivered"
    );

    // cx-parity: the knob is off, so nothing is recorded.
    let cx_root = temp_root("queue-cx");
    let cx_store = SessionStore::open(cx_root.join("sessions")).unwrap();
    let mut cx = Agent::with_provider(
        parity_config("cx-parity", &cx_root),
        Box::new(Answering("a1")),
    );
    assert!(!cx.config().session_queue_persist);
    session_journal::arm(&mut cx, &cx_store, "sess-cx");
    cx.queue_follow_up("not recorded");
    let raw =
        std::fs::read_to_string(cx_store.journal_path("sess-cx").unwrap()).unwrap_or_default();
    assert!(!raw.contains("enqueue"), "cx-parity records no queue ops");
}

// ---------------------------------------------------------------------------
// catalog:156 — Todos / plan persisted per session
// ---------------------------------------------------------------------------

/// Both presets set `[capabilities.todos] persist = true`; BP-8 gives that
/// key a consumer. The model's `update_plan` call reaches disk and the plan
/// comes back on a resume — proved by driving the tool through the real
/// agent loop under each RESOLVED preset, not by calling the tool directly.
#[tokio::test]
async fn a_model_written_plan_survives_the_process_under_both_presets() {
    for preset in ["cc-parity", "cx-parity"] {
        let root = temp_root("plan");
        let store = SessionStore::open(root.join("sessions")).unwrap();
        let name = "sess-plan";

        let mut agent = Agent::with_provider(
            parity_config(preset, &root),
            Box::new(Planning {
                plan: serde_json::json!({
                    "plan": [
                        {"step": "read the ledger", "status": "completed"},
                        {"step": "write the test", "status": "in_progress"},
                    ]
                }),
                calls: AtomicUsize::new(0),
            }),
        );
        assert!(
            agent.config().todos_persist,
            "{preset} must set capabilities.todos.persist"
        );
        session_journal::arm(&mut agent, &store, name);
        let reply = agent.send("plan the work").await.unwrap();
        assert_eq!(reply, "planned", "{preset}: the tool round-trip must run");
        assert_eq!(
            agent.plan(),
            vec![
                PlanEntry {
                    step: "read the ledger".into(),
                    status: "completed".into()
                },
                PlanEntry {
                    step: "write the test".into(),
                    status: "in_progress".into()
                },
            ],
            "{preset}: the loop must see the plan the tool wrote"
        );
        persist(&agent, &store, name);
        drop(agent);

        // Off disk, in a fresh process.
        assert_eq!(store.load_plan(name).unwrap().unwrap().len(), 2, "{preset}");
        let (agent, report) = reopen(preset, &root, &store, name, Box::new(Answering("later")));
        assert_eq!(report.restored_plan, 2, "{preset}");
        assert_eq!(agent.plan()[1].status, "in_progress", "{preset}");
    }
}

/// The plan is journaled as it is written, not only at end-of-turn — so the
/// same crash the transcript rewrite loses does not lose the plan either.
#[tokio::test]
async fn the_plan_is_journaled_before_the_turn_ends() {
    let root = temp_root("plan-journal");
    let store = SessionStore::open(root.join("sessions")).unwrap();
    let mut agent = Agent::with_provider(
        parity_config("cc-parity", &root),
        Box::new(Planning {
            plan: serde_json::json!({"plan": [{"step": "only step", "status": "pending"}]}),
            calls: AtomicUsize::new(0),
        }),
    );
    session_journal::arm(&mut agent, &store, "sess-pj");
    agent.send("go").await.unwrap();
    // No persist at all — the crash window again.
    let state = store.load_journal("sess-pj").unwrap().unwrap();
    assert_eq!(
        state.plan_pairs(),
        vec![("only step".to_string(), "pending".to_string())]
    );
}

// ---------------------------------------------------------------------------
// The journal's own invariants, at the file level.
// ---------------------------------------------------------------------------

/// Append-only means append-only: no operation ever shortens the file, and
/// the bytes of a rewound-away message stay exactly where they were.
#[tokio::test]
async fn no_operation_ever_shortens_the_journal() {
    let root = temp_root("append-invariant");
    let store = SessionStore::open(root.join("sessions")).unwrap();
    let name = "sess-inv";
    let path = store.journal_path(name).unwrap();

    let mut agent =
        Agent::with_provider(parity_config("cc-parity", &root), Box::new(Answering("a")));
    session_journal::arm(&mut agent, &store, name);
    let mut sizes = vec![std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0)];
    for turn in ["u1", "u2"] {
        agent.send(turn).await.unwrap();
        sizes.push(std::fs::metadata(&path).unwrap().len());
    }
    let after_turns = std::fs::read_to_string(&path).unwrap();
    agent.rewind_conversation(1);
    sizes.push(std::fs::metadata(&path).unwrap().len());
    agent.undo_rewind();
    sizes.push(std::fs::metadata(&path).unwrap().len());
    persist(&agent, &store, name);
    sizes.push(std::fs::metadata(&path).unwrap().len());

    for pair in sizes.windows(2) {
        assert!(pair[1] >= pair[0], "the journal shrank: {sizes:?}");
    }
    let final_bytes = std::fs::read_to_string(&path).unwrap();
    assert!(
        final_bytes.starts_with(&after_turns),
        "earlier bytes must be untouched by later operations"
    );
}

/// Every journal line is discriminated, so a tolerant foreign loader pointed
/// at one skips it instead of mis-parsing it as a conversation record.
#[test]
fn every_record_carries_the_discriminant() {
    let dir = temp_root("discriminant");
    let path = dir.join("j.journal.jsonl");
    let mut journal = session_journal::SessionJournal::open_append(&path).unwrap();
    journal.append_message(&ChatMessage::user("hi")).unwrap();
    journal
        .append(JournalOp::Enqueue {
            queue: QueueKind::Steer,
            text: "q".into(),
        })
        .unwrap();
    journal.append(JournalOp::Rewind { to: 0 }).unwrap();
    journal
        .append(JournalOp::Checkpoint { messages: 1 })
        .unwrap();
    for line in std::fs::read_to_string(&path).unwrap().lines() {
        let value: serde_json::Value = serde_json::from_str(line).unwrap();
        assert_eq!(value["supercode_journal"], 1, "{line}");
    }
}