memlay 0.1.5

//! Audit-spool concurrency regression tests: agents fire hooks in parallel
//! (e.g. Claude Code runs two PostToolUse hooks for one assistant message),
//! so simultaneous `memlay hook ingest` processes must never tear NDJSON
//! lines, and `audit finalize` must quarantine — not silently drop — any
//! unparseable line it does find.

use serde_json::{json, Value};
use std::io::Write;
use std::process::{Command, Stdio};

fn memlay_bin() -> &'static str {
    env!("CARGO_BIN_EXE_memlay")
}

fn git(dir: &std::path::Path, args: &[&str]) {
    let out = Command::new("git")
        .args(args)
        .current_dir(dir)
        .output()
        .unwrap();
    assert!(
        out.status.success(),
        "git {args:?}: {}",
        String::from_utf8_lossy(&out.stderr)
    );
}

fn fixture_repo() -> tempfile::TempDir {
    let tmp = tempfile::tempdir().unwrap();
    let dir = tmp.path();
    git(dir, &["init", "-b", "main"]);
    git(dir, &["config", "user.email", "test@example.com"]);
    git(dir, &["config", "user.name", "Test"]);
    git(dir, &["config", "commit.gpgsign", "false"]);
    let init = Command::new(memlay_bin())
        .args(["--repo", dir.to_str().unwrap(), "init", "--no-index"])
        .output()
        .unwrap();
    assert!(
        init.status.success(),
        "{}",
        String::from_utf8_lossy(&init.stderr)
    );
    tmp
}

fn spool_path(dir: &std::path::Path, session: &str) -> std::path::PathBuf {
    dir.join(".git")
        .join("memlay")
        .join("spool")
        .join(format!("{session}.ndjson"))
}

fn ingest(dir: &std::path::Path, session: &str, payload: &str) {
    let mut child = Command::new(memlay_bin())
        .args([
            "--repo",
            dir.to_str().unwrap(),
            "hook",
            "ingest",
            "--agent",
            "claude",
            "--session",
            session,
        ])
        .stdin(Stdio::piped())
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    child
        .stdin
        .take()
        .unwrap()
        .write_all(payload.as_bytes())
        .unwrap();
    let out = child.wait_with_output().unwrap();
    assert!(
        out.status.success(),
        "ingest: {}",
        String::from_utf8_lossy(&out.stderr)
    );
}

/// Regression: `writeln!` on an unbuffered spool file emitted the event as
/// many small writes, so two concurrent ingest processes interleaved
/// fragments mid-line and produced unparseable NDJSON (observed as
/// `doctor` reporting "2/29 unparseable events"). Every event must survive
/// a burst of parallel large-payload ingests intact.
#[test]
fn concurrent_ingest_never_tears_lines() {
    let tmp = fixture_repo();
    let dir = tmp.path();
    const WRITERS: usize = 24;
    // Large payloads widen the torn-write window; spaces keep the filler
    // clear of credential-shaped redaction.
    let filler = "torn line probe ".repeat(512);

    let mut children = Vec::new();
    for marker in 0..WRITERS {
        let payload = json!({
            "hook_event_name": "PostToolUse",
            "marker": marker,
            "filler": filler,
        })
        .to_string();
        let mut child = Command::new(memlay_bin())
            .args([
                "--repo",
                dir.to_str().unwrap(),
                "hook",
                "ingest",
                "--agent",
                "claude",
                "--session",
                "torn-lines",
            ])
            .stdin(Stdio::piped())
            .stdout(Stdio::null())
            .stderr(Stdio::piped())
            .spawn()
            .unwrap();
        child
            .stdin
            .take()
            .unwrap()
            .write_all(payload.as_bytes())
            .unwrap();
        children.push(child);
    }
    for child in children {
        let out = child.wait_with_output().unwrap();
        assert!(
            out.status.success(),
            "ingest: {}",
            String::from_utf8_lossy(&out.stderr)
        );
    }

    let text = std::fs::read_to_string(spool_path(dir, "torn-lines")).unwrap();
    let mut seen = [false; WRITERS];
    let mut lines = 0usize;
    for line in text.lines() {
        lines += 1;
        let event: Value =
            serde_json::from_str(line).unwrap_or_else(|e| panic!("torn NDJSON line ({e}): {line}"));
        assert_eq!(event["payload"]["filler"].as_str().unwrap(), filler);
        let marker = event["payload"]["marker"].as_u64().unwrap() as usize;
        assert!(!seen[marker], "event {marker} appended twice");
        seen[marker] = true;
    }
    assert_eq!(lines, WRITERS, "every ingest must land exactly one line");
    assert!(seen.iter().all(|&s| s), "every marker must survive intact");
}

/// `audit finalize` must move unparseable spool lines into a `.corrupt`
/// quarantine file beside the spool instead of silently discarding them,
/// and must still bundle the valid events.
#[test]
fn finalize_quarantines_unparseable_lines() {
    let tmp = fixture_repo();
    let dir = tmp.path();
    ingest(
        dir,
        "quarantine",
        &json!({ "hook_event_name": "PostToolUse", "ok": true }).to_string(),
    );

    let spool = spool_path(dir, "quarantine");
    let torn = "{\"schema_version\":1,\"event_id\":\"torn{\"schema_ver";
    let mut text = std::fs::read_to_string(&spool).unwrap();
    text.push_str(torn);
    text.push('\n');
    std::fs::write(&spool, text).unwrap();

    let out = Command::new(memlay_bin())
        .args([
            "--repo",
            dir.to_str().unwrap(),
            "audit",
            "finalize",
            "quarantine",
        ])
        .output()
        .unwrap();
    assert!(
        out.status.success(),
        "finalize: {}",
        String::from_utf8_lossy(&out.stderr)
    );
    assert!(
        String::from_utf8_lossy(&out.stderr).contains("quarantined"),
        "finalize must warn about quarantined lines"
    );

    assert!(!spool.exists(), "spool is consumed by finalize");
    let quarantine = spool.with_extension("ndjson.corrupt");
    let quarantined = std::fs::read_to_string(&quarantine).unwrap();
    assert_eq!(quarantined.lines().collect::<Vec<_>>(), vec![torn]);
}