dicklesworthstone__coding_agent_session_search/tests/cli_schedule.rs

//! E2E coverage for `cass schedule` (background-index OS scheduler wave).
//!
//! Everything runs hermetically: a temp HOME (so provider discovery finds no
//! real sessions and no real LaunchAgents/systemd user dir is touched), a
//! temp data dir, and scrubbed CASS_* env. `install` is only exercised with
//! `--dry-run` — registering real launchd/systemd jobs from a test suite
//! would leak state onto the developer machine.

use serde_json::Value;
use std::path::Path;
use std::process::Command;

fn cass_bin_path() -> &'static str {
    env!("CARGO_BIN_EXE_cass")
}

/// A cass invocation isolated from the developer's real environment.
fn cass_cmd(home: &Path) -> Command {
    // Stop dotenvy's ancestor search at this disposable fixture home, even
    // when the test runner places it underneath the repository's .rch-tmp.
    match std::fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(home.join(".env"))
    {
        Ok(_) => {}
        Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
        Err(error) => panic!("create isolated fixture .env: {error}"),
    }
    let mut cmd = Command::new(cass_bin_path());
    cmd.env_clear();
    for key in ["PATH", "SystemRoot", "WINDIR"] {
        if let Some(value) = std::env::var_os(key) {
            cmd.env(key, value);
        }
    }
    cmd.current_dir(home);
    cmd.env("HOME", home);
    cmd.env("USERPROFILE", home);
    cmd.env("XDG_CONFIG_HOME", home.join(".config"));
    cmd.env("XDG_DATA_HOME", home.join(".local/share"));
    cmd.env("CLAUDE_CONFIG_DIR", home.join(".claude"));
    cmd.env("CODEX_HOME", home.join(".codex"));
    cmd.env("RUST_MIN_STACK", "134217728");
    cmd.env("TUI_HEADLESS", "1");
    cmd.env("CODING_AGENT_SEARCH_NO_UPDATE_PROMPT", "1");
    cmd.env("CASS_IGNORE_SOURCES_CONFIG", "1");
    // Deterministic gates: never load-gate or idle-gate a test run.
    cmd.env("CASS_RESPONSIVENESS_DISABLE", "1");
    cmd
}

/// Seed one real Claude Code session so indexing has something to publish.
/// `cass index --full` on a machine with zero sessions fails its own
/// post-publish validation (no lexical index exists to read back), so the
/// nightly tests need a non-empty corpus; the incremental test deliberately
/// keeps an empty corpus to cover that path too.
fn seed_claude_code_fixture(home: &Path) {
    let src = Path::new(env!("CARGO_MANIFEST_DIR"))
        .join("tests")
        .join("fixtures")
        .join("claude_code_real")
        .join("projects")
        .join("-test-project")
        .join("agent-test123.jsonl");
    let dst = home
        .join(".claude")
        .join("projects")
        .join("-test-project")
        .join("agent-test123.jsonl");
    std::fs::create_dir_all(dst.parent().expect("fixture parent")).expect("create fixture dir");
    std::fs::copy(&src, &dst).expect("copy claude_code fixture");
}

fn parse_single_json_document(stdout: &[u8]) -> Value {
    let text = String::from_utf8_lossy(stdout);
    serde_json::from_str::<Value>(text.trim())
        .unwrap_or_else(|e| panic!("stdout must be exactly one JSON document ({e}); got:\n{text}"))
}

#[test]
fn schedule_install_dry_run_reports_units_without_executing_anything() {
    let tmp = tempfile::tempdir().unwrap();
    let home = tmp.path().join("home");
    let data_dir = tmp.path().join("dd");
    std::fs::create_dir_all(&home).unwrap();

    let output = cass_cmd(&home)
        .args(["schedule", "install", "--dry-run", "--json", "--data-dir"])
        .arg(&data_dir)
        .output()
        .expect("run cass schedule install --dry-run");

    if cfg!(any(target_os = "macos", target_os = "linux")) {
        assert!(
            output.status.success(),
            "dry-run install must succeed: {}",
            String::from_utf8_lossy(&output.stderr)
        );
        let report = parse_single_json_document(&output.stdout);
        assert_eq!(report["dry_run"], Value::Bool(true));
        let units = report["units"].as_array().expect("units array");
        let expected_units = if cfg!(target_os = "macos") { 2 } else { 4 };
        assert_eq!(units.len(), expected_units, "incremental + nightly units");
        for unit in units {
            let path = unit["path"].as_str().expect("unit path");
            assert!(
                path.starts_with(home.to_str().unwrap()) || path.contains("systemd"),
                "unit paths must live under the (fake) HOME: {path}"
            );
            assert!(
                !Path::new(path).exists(),
                "dry-run must not write unit files: {path}"
            );
        }
        for command in report["commands"].as_array().expect("commands array") {
            assert_eq!(
                command["executed"],
                Value::Bool(false),
                "dry-run must not execute scheduler commands: {command}"
            );
        }
        assert_eq!(
            report["stale_removed"].as_array().map(Vec::len),
            Some(0),
            "fresh dry-run install has nothing stale to remove"
        );
        // The rendered units must invoke `schedule run` against this data dir
        // at background priority.
        let rendered: String = units
            .iter()
            .filter_map(|u| u["contents"].as_str())
            .collect();
        assert!(rendered.contains("schedule"));
        assert!(rendered.contains("--job"));
        assert!(rendered.contains(data_dir.to_str().unwrap()));
    } else {
        assert!(
            !output.status.success(),
            "unsupported platforms fail closed"
        );
    }
}

#[test]
fn schedule_install_dry_run_no_nightly_registers_only_incremental() {
    if !cfg!(any(target_os = "macos", target_os = "linux")) {
        return;
    }
    let tmp = tempfile::tempdir().unwrap();
    let home = tmp.path().join("home");
    let data_dir = tmp.path().join("dd");
    std::fs::create_dir_all(&home).unwrap();

    let output = cass_cmd(&home)
        .args([
            "schedule",
            "install",
            "--dry-run",
            "--no-nightly",
            "--json",
            "--data-dir",
        ])
        .arg(&data_dir)
        .output()
        .expect("run cass schedule install --dry-run --no-nightly");
    assert!(output.status.success());
    let report = parse_single_json_document(&output.stdout);
    let units = report["units"].as_array().expect("units array");
    let expected_units = if cfg!(target_os = "macos") { 1 } else { 2 };
    assert_eq!(units.len(), expected_units);
    for unit in units {
        assert!(
            !unit["path"]
                .as_str()
                .unwrap_or_default()
                .contains("nightly"),
            "--no-nightly must not render nightly units"
        );
    }
}

#[test]
fn schedule_run_incremental_force_records_state_and_history() {
    let tmp = tempfile::tempdir().unwrap();
    let home = tmp.path().join("home");
    let data_dir = tmp.path().join("dd");
    std::fs::create_dir_all(&home).unwrap();

    let run = |label: &str| {
        let output = cass_cmd(&home)
            .args([
                "schedule",
                "run",
                "--job",
                "incremental",
                "--force",
                "--json",
                "--data-dir",
            ])
            .arg(&data_dir)
            .output()
            .expect("run cass schedule run");
        assert!(
            output.status.success(),
            "{label}: incremental job on an empty corpus must succeed: {}",
            String::from_utf8_lossy(&output.stderr)
        );
        parse_single_json_document(&output.stdout)
    };

    let report = run("first");
    assert_eq!(report["job"], "incremental");
    assert_eq!(report["ok"], Value::Bool(true));
    assert!(report["skipped_reason"].is_null(), "--force bypasses gates");
    let steps = report["steps"].as_array().expect("steps");
    let index_step = steps
        .iter()
        .find(|s| s["name"] == "index")
        .expect("an index step");
    assert_eq!(index_step["ok"], Value::Bool(true));
    assert_eq!(index_step["exit_code"], Value::from(0));

    let state_path = data_dir.join("schedule").join("state.json");
    let runs_path = data_dir.join("schedule").join("runs.jsonl");
    let state: Value =
        serde_json::from_str(&std::fs::read_to_string(&state_path).expect("state.json"))
            .expect("state.json parses");
    assert_eq!(state["last_incremental"]["ok"], Value::Bool(true));
    assert!(state["last_nightly"].is_null());
    let history = std::fs::read_to_string(&runs_path).expect("runs.jsonl");
    assert_eq!(history.lines().count(), 1);

    run("second");
    let history = std::fs::read_to_string(&runs_path).expect("runs.jsonl");
    assert_eq!(history.lines().count(), 2, "history is append-only");
}

#[test]
fn schedule_run_nightly_skips_semantic_tiers_it_cannot_serve() {
    let tmp = tempfile::tempdir().unwrap();
    let home = tmp.path().join("home");
    let data_dir = tmp.path().join("dd");
    std::fs::create_dir_all(&home).unwrap();
    seed_claude_code_fixture(&home);

    let output = cass_cmd(&home)
        .args([
            "schedule",
            "run",
            "--job",
            "nightly",
            "--force",
            "--json",
            "--data-dir",
        ])
        .arg(&data_dir)
        .output()
        .expect("run cass schedule run --job nightly");
    assert!(
        output.status.success(),
        "nightly on an empty corpus without models must still succeed: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    let report = parse_single_json_document(&output.stdout);
    assert_eq!(report["ok"], Value::Bool(true));
    let steps = report["steps"].as_array().expect("steps");
    let names: Vec<&str> = steps.iter().filter_map(|s| s["name"].as_str()).collect();
    assert!(names.contains(&"index-full"), "steps: {names:?}");
    assert!(names.contains(&"models-status"), "steps: {names:?}");
    let quality_skip = steps
        .iter()
        .find(|s| s["name"] == "semantic-backfill:quality")
        .expect("quality tier step");
    assert!(
        quality_skip["skipped_reason"]
            .as_str()
            .unwrap_or_default()
            .contains("models install"),
        "quality tier must be skipped (not failed) without the model: {quality_skip}"
    );
    // Every step is ok (skips count as ok); nothing failed.
    for step in steps {
        assert_eq!(step["ok"], Value::Bool(true), "step failed: {step}");
    }
}

#[test]
fn gh471_schedule_nightly_bounds_one_fast_worker_to_global_batch_allowance() {
    use coding_agent_search::search::semantic_manifest::SemanticManifest;
    use coding_agent_search::search::vector_index::{VectorIndex, vector_index_path};
    use coding_agent_search::storage::sqlite::FrankenStorage;

    for limit in [0u32, 1, 2] {
        let tmp = tempfile::tempdir().unwrap();
        let home = tmp.path().join("home");
        let data_dir = tmp.path().join("data");
        let project = home.join(".claude/projects/-schedule-backfill");
        std::fs::create_dir_all(&project).unwrap();
        for ordinal in 0..2 {
            let record = serde_json::json!({
                "parentUuid": null,
                "cwd": "/test/scheduled-backfill",
                "sessionId": format!("scheduled-session-{ordinal}"),
                "version": "2.0.37",
                "gitBranch": "main",
                "type": "user",
                "message": {
                    "role": "user",
                    "content": format!("Investigate scheduled checkpoint evidence for distinct session {ordinal}")
                },
                "uuid": format!("scheduled-message-{ordinal}"),
                "timestamp": "2025-11-12T18:31:18.697Z"
            });
            std::fs::write(
                project.join(format!("scheduled-session-{ordinal}.jsonl")),
                format!("{record}\n"),
            )
            .unwrap();
        }
        let mut command = cass_cmd(&home);
        command
            .args([
                "schedule",
                "run",
                "--job",
                "nightly",
                "--force",
                "--json",
                "--data-dir",
            ])
            .arg(&data_dir)
            .env("CASS_SCHEDULE_MAX_BACKFILL_BATCHES", limit.to_string())
            .env("CASS_SEMANTIC_MAX_MESSAGES_PER_CHECKPOINT", "1")
            .env("CASS_SEMANTIC_BACKFILL_FORCE", "1")
            .env("RUST_MIN_STACK", "134217728");
        let output = assert_cmd::Command::from_std(command)
            .timeout(std::time::Duration::from_secs(120))
            .output()
            .expect("run bounded real nightly worker");
        assert!(
            output.status.success(),
            "limit={limit}; stdout: {}\nstderr: {}",
            String::from_utf8_lossy(&output.stdout),
            String::from_utf8_lossy(&output.stderr)
        );
        let report = parse_single_json_document(&output.stdout);
        assert_eq!(report["ok"], true, "{report}");
        let storage = FrankenStorage::open_readonly(&data_dir.join("agent_search.db"))
            .expect("read the actual nightly canonical archive");
        let conversations = storage.list_conversations(i64::MAX, 0).unwrap();
        let mut actual_sources: Vec<_> = conversations
            .iter()
            .map(|conversation| conversation.source_path.clone())
            .collect();
        actual_sources.sort();
        let expected_sources: Vec<_> = (0..2)
            .map(|ordinal| project.join(format!("scheduled-session-{ordinal}.jsonl")))
            .collect();
        assert_eq!(
            actual_sources, expected_sources,
            "nightly discovery must contain exactly the seeded sources; limit={limit}"
        );
        for conversation in conversations {
            assert_eq!(conversation.agent_slug, "claude_code");
            let messages = storage.fetch_messages(conversation.id.unwrap()).unwrap();
            assert_eq!(messages.len(), 1, "{}", conversation.source_path.display());
            assert!(
                messages[0]
                    .content
                    .contains("scheduled checkpoint evidence")
            );
        }
        drop(storage);
        let steps = report["steps"].as_array().expect("steps");
        let workers: Vec<_> = steps
            .iter()
            .filter(|step| {
                step["argv"]
                    .as_array()
                    .is_some_and(|args| args.iter().any(|arg| arg.as_str() == Some("backfill")))
            })
            .collect();
        let manifest = SemanticManifest::load(&data_dir).expect("read actual worker manifest");
        if limit == 0 {
            assert!(
                workers.is_empty(),
                "zero global budget must start no worker: {report}"
            );
            assert!(
                manifest.is_none(),
                "zero budget must not start semantic publication"
            );
            continue;
        }
        assert_eq!(
            workers.len(),
            1,
            "one real child serves every admitted fast batch: {report}"
        );
        let worker = workers[0];
        assert_eq!(worker["ok"], true, "{worker}");
        let args = worker["argv"].as_array().expect("actual child argv");
        let bound = args
            .windows(2)
            .find(|pair| pair[0] == "--max-batches")
            .expect("worker must receive its remaining allowance");
        assert_eq!(bound[1], limit.to_string());
        let result = &worker["result"];
        assert_eq!(result["batches_attempted"], limit, "{result}");
        assert_eq!(result["batches_completed"], limit, "{result}");
        assert_eq!(result["model_initializations"], 1, "{result}");
        assert_eq!(result["conversations_processed"], limit, "{result}");
        assert_eq!(result["total_conversations"], 2, "{result}");
        let manifest = manifest.expect("admitted worker must persist progress");
        if limit == 1 {
            assert_eq!(result["status"], "checkpointed", "{result}");
            let checkpoint = manifest.checkpoint.expect("bounded first checkpoint");
            assert_eq!(checkpoint.conversations_processed, 1);
            assert_eq!(checkpoint.docs_embedded, 1);
            assert!(manifest.fast_tier.is_none());
        } else {
            assert_eq!(result["status"], "published", "{result}");
            assert!(manifest.checkpoint.is_none());
            let tier = manifest.fast_tier.expect("published fast tier");
            assert!(tier.ready);
            assert_eq!((tier.doc_count, tier.conversation_count), (2, 2));
            let index = VectorIndex::open(&vector_index_path(&data_dir, "fnv1a-384"))
                .expect("open actually published vectors");
            assert_eq!(index.record_count(), 2);
            assert_ne!(index.doc_id_at(0).unwrap(), index.doc_id_at(1).unwrap());
        }
    }
}

#[test]
fn schedule_run_nightly_no_semantic_skips_backfill_entirely() {
    let tmp = tempfile::tempdir().unwrap();
    let home = tmp.path().join("home");
    let data_dir = tmp.path().join("dd");
    std::fs::create_dir_all(&home).unwrap();
    seed_claude_code_fixture(&home);

    let output = cass_cmd(&home)
        .args([
            "schedule",
            "run",
            "--job",
            "nightly",
            "--no-semantic",
            "--force",
            "--json",
            "--data-dir",
        ])
        .arg(&data_dir)
        .output()
        .expect("run cass schedule run --job nightly --no-semantic");
    assert!(output.status.success());
    let report = parse_single_json_document(&output.stdout);
    let steps = report["steps"].as_array().expect("steps");
    let backfill = steps
        .iter()
        .find(|s| s["name"] == "semantic-backfill")
        .expect("aggregate backfill skip step");
    assert!(
        backfill["skipped_reason"]
            .as_str()
            .unwrap_or_default()
            .contains("--no-semantic")
    );
    assert!(
        !steps.iter().any(|s| s["name"] == "models-status"),
        "--no-semantic must not even probe the model"
    );
}

// Optional Linux measurement is evidence, not a performance pass threshold.
#[cfg(target_os = "linux")]
#[test]
#[ignore = "requires CASS_NATIVE_REUSE_MODEL_DIR, /usr/bin/time and /usr/bin/timeout; no downloads"]
fn gh471_schedule_quality_worker_has_one_native_constructor_and_two_durable_boundaries() {
    use coding_agent_search::search::fastembed_embedder::{
        FastEmbedder, MINILM_VECTOR_SPACE_REVISION,
    };
    use coding_agent_search::search::model_download::{ModelManifest, model_file_path};
    use coding_agent_search::search::semantic_manifest::SemanticManifest;
    use coding_agent_search::search::vector_index::{
        VectorIndex, parse_semantic_doc_id, vector_index_path,
    };
    use coding_agent_search::storage::sqlite::FrankenStorage;

    let supplied = std::path::PathBuf::from(
        dotenvy::var("CASS_NATIVE_REUSE_MODEL_DIR").expect("explicit attested native bundle"),
    )
    .canonicalize()
    .expect("existing native bundle");
    let temp = tempfile::tempdir().unwrap();
    let home = temp.path().join("home");
    let data_dir = temp.path().join("data");
    let project = home.join(".claude/projects/-native-lifetime");
    std::fs::create_dir_all(&project).unwrap();
    for ordinal in 0..2 {
        let record = serde_json::json!({
            "parentUuid": null, "cwd": "/test/native-lifetime",
            "sessionId": format!("native-lifetime-{ordinal}"),
            "version": "2.0.37", "gitBranch": "main", "type": "user",
            "message": {"role": "user", "content": format!("Distinct native schedule evidence {ordinal}")},
            "uuid": format!("native-message-{ordinal}"),
            "timestamp": "2025-11-12T18:31:18.697Z"
        });
        std::fs::write(
            project.join(format!("native-lifetime-{ordinal}.jsonl")),
            format!("{record}\n"),
        )
        .unwrap();
    }
    let model_dir = FastEmbedder::default_model_dir(&data_dir);
    std::fs::create_dir_all(&model_dir).unwrap();
    let model_manifest = ModelManifest::minilm_v2();
    assert_eq!(model_manifest.files.len(), 5);
    for file in &model_manifest.files {
        let source = model_file_path(&supplied, file).expect("supplied manifest file");
        assert_eq!(
            std::fs::copy(source, model_dir.join(file.local_name())).unwrap(),
            file.size
        );
    }
    let parent_trace = temp.path().join("schedule-parent.jsonl");
    let child_trace = temp.path().join("schedule-children.jsonl");
    let progress = temp.path().join("semantic-progress.jsonl");
    let usage_path = temp.path().join("schedule-usage.txt");
    let mut cass = cass_cmd(&home);
    cass.arg("--trace-file")
        .arg(&parent_trace)
        .args([
            "schedule",
            "run",
            "--job",
            "nightly",
            "--force",
            "--json",
            "--data-dir",
        ])
        .arg(&data_dir)
        .env("CASS_TRACE_FILE", &child_trace)
        .env(
            "CASS_TRACE_FILTER",
            "warn,frankensearch_rerank::native_embedder=info",
        )
        .env("CASS_TRACE_TEST_ID", "gh471-native-schedule")
        .env("CASS_SEMANTIC_PROGRESS_JSONL", &progress)
        .env("CASS_SCHEDULE_MAX_BACKFILL_BATCHES", "4")
        .env("CASS_SEMANTIC_MAX_MESSAGES_PER_CHECKPOINT", "1")
        .env("CASS_SEMANTIC_BACKFILL_FORCE", "1");

    // The clock encloses the actual whole scheduled job, not cargo or test setup.
    // GNU time's Linux %M is maximum individual-process RSS, including reaped
    // descendants. It is not a simultaneous process-tree/model+batch memory bound.
    // timeout owns a process group and bounds the actual schedule and its children.
    assert!(Path::new("/usr/bin/time").is_file());
    assert!(Path::new("/usr/bin/timeout").is_file());
    let mut measured = Command::new("/usr/bin/time");
    measured
        .env_clear()
        .current_dir(&home)
        .envs(
            cass.get_envs()
                .filter_map(|(key, value)| value.map(|value| (key, value))),
        )
        .args(["-f", "%e %M", "-o"])
        .arg(&usage_path)
        .args([
            "/usr/bin/timeout",
            "--signal=TERM",
            "--kill-after=10s",
            "1200s",
        ])
        .arg(cass.get_program())
        .args(cass.get_args());
    let output = measured.output().expect("run bounded actual scheduled job");
    assert!(
        output.status.success(),
        "stdout={}; stderr={}",
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    let report = parse_single_json_document(&output.stdout);
    assert_eq!(report["ok"], true, "{report}");
    let workers: Vec<_> = report["steps"]
        .as_array()
        .unwrap()
        .iter()
        .filter(|step| {
            step["argv"]
                .as_array()
                .is_some_and(|args| args.iter().any(|arg| arg == "backfill"))
        })
        .collect();
    assert_eq!(workers.len(), 2, "one fast and one quality child: {report}");
    for (ordinal, tier) in ["fast", "quality"].iter().enumerate() {
        let worker = workers[ordinal];
        assert_eq!(worker["ok"], true, "{worker}");
        let args = worker["argv"].as_array().unwrap();
        assert!(
            args.windows(2)
                .any(|pair| pair[0] == "--tier" && pair[1] == *tier)
        );
        let remaining = if ordinal == 0 { "4" } else { "2" };
        assert!(
            args.windows(2)
                .any(|pair| pair[0] == "--max-batches" && pair[1] == remaining)
        );
        let result = &worker["result"];
        assert_eq!(result["batches_attempted"], 2, "{result}");
        assert_eq!(result["batches_completed"], 2, "{result}");
        assert_eq!(result["model_initializations"], 1, "{result}");
        assert_eq!(result["status"], "published", "{result}");
        assert_eq!(result["total_conversations"], 2, "{result}");
    }
    let read_jsonl = |path: &Path| -> Vec<Value> {
        let text = std::fs::read_to_string(path).expect("actual trace/progress output");
        let rows: Vec<Value> = text
            .lines()
            .enumerate()
            .map(|(ordinal, line)| {
                serde_json::from_str(line).unwrap_or_else(|error| {
                    panic!("{}:{}: {error}; record={line}", path.display(), ordinal + 1)
                })
            })
            .collect();
        assert!(!rows.is_empty(), "{}", path.display());
        for row in &rows {
            assert_ne!(row["event"], "trace_truncated", "{row}");
            assert_ne!(row["fields"]["event"], "trace_truncated", "{row}");
        }
        rows
    };
    let parents = read_jsonl(&parent_trace);
    let children = read_jsonl(&child_trace);
    let is_constructor = |row: &&Value| {
        row["target"] == "frankensearch_rerank::native_embedder"
            && row["fields"]["message"]
                == "native frankentorch MiniLM embedder loaded (mean-pool + L2)"
    };
    assert_eq!(parents.iter().filter(is_constructor).count(), 0);
    let constructors: Vec<_> = children.iter().filter(is_constructor).collect();
    assert_eq!(
        constructors.len(),
        1,
        "actual native constructors: {children:?}"
    );
    assert_eq!(constructors[0]["fields"]["dimension"], 384);
    // Neither producer trace nor progress schema exposes a PID. Sequential
    // child command summaries delimit the actual quality worker's trace block.
    let mut pending_constructors = 0;
    let mut quality_blocks = 0;
    for row in &children {
        if is_constructor(&row) {
            pending_constructors += 1;
        }
        if row["fields"]["event"] == "command_summary" {
            let args = row["fields"]["args"]
                .as_array()
                .expect("actual child arguments");
            let quality = args.iter().any(|arg| arg == "backfill")
                && args
                    .windows(2)
                    .any(|pair| pair[0] == "--tier" && pair[1] == "quality");
            if quality {
                quality_blocks += 1;
                assert_eq!(
                    pending_constructors, 1,
                    "quality child must load exactly once"
                );
            } else {
                assert_eq!(
                    pending_constructors, 0,
                    "non-quality child constructed a model: {row}"
                );
            }
            pending_constructors = 0;
        }
    }
    assert_eq!(quality_blocks, 1, "one completed actual quality child");
    assert_eq!(
        pending_constructors, 0,
        "constructor missing its child completion summary"
    );
    let events = read_jsonl(&progress);
    let durable: Vec<_> = events
        .iter()
        .filter(|event| {
            event["tier"] == "quality"
                && matches!(
                    event["event"].as_str(),
                    Some("checkpoint_save_done" | "publish_done")
                )
        })
        .collect();
    assert_eq!(durable.len(), 2, "{events:?}");
    assert_eq!(durable[0]["event"], "checkpoint_save_done");
    assert_eq!(durable[1]["event"], "publish_done");
    for (ordinal, event) in durable.iter().enumerate() {
        assert_eq!(event["embedder_id"], "minilm-384");
        assert_eq!(event["rows_processed"], ordinal + 1);
        assert_eq!(event["rows_total"], 2);
    }
    let storage = FrankenStorage::open(&data_dir.join("agent_search.db")).unwrap();
    let conversations = storage.list_conversations(i64::MAX, 0).unwrap();
    assert_eq!(conversations.len(), 2);
    let mut source_paths: Vec<_> = conversations
        .iter()
        .map(|row| row.source_path.clone())
        .collect();
    source_paths.sort();
    assert_eq!(
        source_paths,
        (0..2)
            .map(|ordinal| project.join(format!("native-lifetime-{ordinal}.jsonl")))
            .collect::<Vec<_>>()
    );
    let mut message_ids = Vec::new();
    for conversation in &conversations {
        assert_eq!(conversation.agent_slug, "claude_code");
        let messages = storage.fetch_messages(conversation.id.unwrap()).unwrap();
        assert_eq!(messages.len(), 1);
        message_ids.push(u64::try_from(messages[0].id.unwrap()).unwrap());
    }
    message_ids.sort();
    let index = VectorIndex::open(&vector_index_path(&data_dir, "minilm-384")).unwrap();
    assert_eq!(index.embedder_id(), "minilm-384");
    assert_eq!(index.embedder_revision(), MINILM_VECTOR_SPACE_REVISION);
    assert_eq!(index.dimension(), 384);
    assert_eq!(index.record_count(), 2);
    let mut published_ids = Vec::new();
    for ordinal in 0..2 {
        assert!(!index.is_deleted(ordinal));
        let doc = parse_semantic_doc_id(index.doc_id_at(ordinal).unwrap()).unwrap();
        published_ids.push(doc.message_id);
        assert_eq!(doc.chunk_idx, 0);
        assert!(doc.content_hash.is_some());
        let vector = index.vector_at_f32(ordinal).unwrap();
        assert_eq!(vector.len(), 384);
        assert!(vector.iter().all(|value| value.is_finite()));
    }
    published_ids.sort();
    assert_eq!(published_ids, message_ids);
    let manifest = SemanticManifest::load(&data_dir)
        .unwrap()
        .expect("durable publication");
    assert!(manifest.checkpoint.is_none());
    let quality = manifest.quality_tier.expect("actual quality tier");
    assert!(quality.ready);
    assert_eq!((quality.doc_count, quality.conversation_count), (2, 2));

    let raw_usage = std::fs::read_to_string(&usage_path).unwrap();
    let columns: Vec<_> = raw_usage.split_whitespace().collect();
    assert_eq!(columns.len(), 2, "GNU time output: {raw_usage}");
    let elapsed_seconds: f64 = columns[0].parse().expect("elapsed seconds");
    let maximum_process_rss_kib: u64 = columns[1].parse().expect("maximum process RSS");
    assert!(elapsed_seconds.is_finite() && elapsed_seconds > 0.0);
    assert!(maximum_process_rss_kib > 0);
    eprintln!(
        "gh471_native_schedule_observation={}",
        serde_json::json!({
            "elapsed_seconds": elapsed_seconds,
            "gnu_time_maximum_process_rss_kib": maximum_process_rss_kib,
            "scope": "whole schedule invocation; maximum individual-process RSS including reaped descendants",
            "performance_certification": false,
            "simultaneous_process_tree_peak": null,
            "constructors": constructors, "quality_durable_boundaries": durable,
            "report": report, "stderr": String::from_utf8_lossy(&output.stderr)
        })
    );
}

#[test]
fn schedule_run_json_failure_stays_a_single_document_with_exit_9() {
    let tmp = tempfile::tempdir().unwrap();
    let home = tmp.path().join("home");
    let data_dir = tmp.path().join("dd");
    std::fs::create_dir_all(&home).unwrap();

    // An impossible --db path makes the index step fail (spawnable binary,
    // unusable database), which must fail the job — but the report already
    // printed to stdout must remain the ONLY JSON document there.
    let output = cass_cmd(&home)
        .args(["--db", "/dev/null/not-a-db/agent_search.db"])
        .args([
            "schedule",
            "run",
            "--job",
            "incremental",
            "--force",
            "--json",
            "--data-dir",
        ])
        .arg(&data_dir)
        .output()
        .expect("run cass schedule run with a broken --db");
    assert_eq!(
        output.status.code(),
        Some(9),
        "failing job must exit 9: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    let report = parse_single_json_document(&output.stdout);
    assert_eq!(report["ok"], Value::Bool(false));
    let index_step = report["steps"]
        .as_array()
        .expect("steps")
        .iter()
        .find(|s| s["name"] == "index")
        .cloned()
        .expect("index step");
    assert_eq!(index_step["ok"], Value::Bool(false));
}

#[test]
fn schedule_status_json_reports_never_run_then_last_run() {
    let tmp = tempfile::tempdir().unwrap();
    let home = tmp.path().join("home");
    let data_dir = tmp.path().join("dd");
    std::fs::create_dir_all(&home).unwrap();

    let status = |label: &str| {
        let output = cass_cmd(&home)
            .args(["schedule", "status", "--json", "--data-dir"])
            .arg(&data_dir)
            .output()
            .expect("run cass schedule status");
        assert!(
            output.status.success(),
            "{label}: status must succeed: {}",
            String::from_utf8_lossy(&output.stderr)
        );
        parse_single_json_document(&output.stdout)
    };

    let before = status("before");
    assert!(before["state"]["last_incremental"].is_null());
    // With a fake HOME no units are installed.
    for unit in before["units"].as_array().unwrap_or(&Vec::new()) {
        assert_eq!(unit["installed"], Value::Bool(false));
    }

    let run_output = cass_cmd(&home)
        .args([
            "schedule",
            "run",
            "--job",
            "incremental",
            "--force",
            "--json",
            "--data-dir",
        ])
        .arg(&data_dir)
        .output()
        .expect("run incremental job");
    assert!(run_output.status.success());

    let after = status("after");
    assert_eq!(after["state"]["last_incremental"]["ok"], Value::Bool(true));
}