pushkin 0.2.0

//! F74 — the warm path must say it served.
//!
//! Ruling: `docs/claude_ruling-f73-manifest-resolution.md` Addendum 3 §§1-2. A NEW file per N10.
//!
//! **The defect.** `daemon.rs` resolved the warm path as `warm_request(root, request).ok()`.
//! `.ok()` collapses every `ServerError` — socket absent, connection refused, protocol
//! mismatch, a daemon that accepts and dies — into `None`, and `check_or_cold`'s
//! `unwrap_or_else` then runs the cold pipeline. Since F73 phase 2 pinned cold resolution to
//! the repository root, **the cold answer is also the correct answer**, so the degradation
//! produced a right verdict and left no trace anywhere: not in the envelope, not in the event
//! log, not on stderr.
//!
//! That is the class the standing rule now names — *a degradation that yields a plausible
//! result without recording that it degraded*. The rule is **not** "must fail": degrading to
//! cold is usually correct, and the daemon is a transport, never a second brain. The
//! requirement is that the degradation be **observable**.
//!
//! **What this suite pins.** One stderr line per file check, naming which path served the
//! verdict and — when a warm attempt was made and failed — why:
//!
//! | state | disclosed |
//! |---|---|
//! | a daemon answered | the warm path served |
//! | `PUSHKIN_DAEMON=off` | cold, warm path disabled, socket never touched |
//! | probed, nothing listening | cold, no daemon answering |
//! | probed, the attempt FAILED | cold, **and the attempt and its reason** |
//!
//! **The fourth row is the finding.** A suite that only checked warm-versus-cold would have
//! passed against the defective code, because the defective code already produced the right
//! verdict on both. The failed attempt is the event that was invisible.
//!
//! **Every verdict is asserted byte-unchanged.** Disclosure is on stderr and only on stderr;
//! stdout is the verdict channel and this pass does not touch it. An `allow` on the Claude hook
//! surface prints nothing to stdout, and it must still print nothing.
//!
//! The fake-daemon harness is `daemon_shim.rs`'s, reused rather than reinvented: a test-owned
//! `UnixListener` on the daemon socket whose canned answer is the *opposite* of what the real
//! pipeline says, so a flipped verdict proves the socket was genuinely consulted.

use assert_cmd::Command;
use std::fs;
use std::io::{BufRead, BufReader, Write};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::Path;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

type TestResult = Result<(), Box<dyn std::error::Error>>;

const MANIFEST: &str = r#"
version = 1
canonical = "json-schema-2020-12"
authoring = "zod"

[[contracts]]
name = "user"
source = "contracts/user.zod.ts"
emit = ["zod"]

[[mappings]]
glob = "app/api/**/*.ts"
contracts = ["user"]
require = "boundary-validation"

[gates]
"#;

const NONCONFORMING: &str = "export async function POST(req: Request) {\n\
  const body = await req.json();\n\
  return Response.json({ name: body.name });\n\
}\n";

const HANDLER_PATH: &str = "app/api/users/route.ts";
const RULE: &str = "contract.boundary.unvalidated_input";

fn repo() -> Result<tempfile::TempDir, Box<dyn std::error::Error>> {
    let dir = tempfile::tempdir()?;
    fs::write(dir.path().join("pushkin.toml"), MANIFEST)?;
    Ok(dir)
}

fn payload() -> String {
    serde_json::json!({
        "session_id": "f74-session",
        "tool_name": "Write",
        "tool_input": { "file_path": HANDLER_PATH, "content": NONCONFORMING }
    })
    .to_string()
}

/// `pushkin hook claude` in `dir`. Returns (exit code, stdout, stderr) — the two
/// streams kept SEPARATE, because the whole contract is that disclosure lands on
/// one of them and the verdict on the other.
fn run_hook(
    dir: &Path,
    daemon_env: Option<&str>,
) -> Result<(i32, String, String), Box<dyn std::error::Error>> {
    let mut command = Command::cargo_bin("pushkin")?;
    command.current_dir(dir).args(["hook", "claude"]);
    match daemon_env {
        Some(value) => command.env("PUSHKIN_DAEMON", value),
        None => command.env_remove("PUSHKIN_DAEMON"),
    };
    let output = command.write_stdin(payload()).output()?;
    Ok((
        output.status.code().unwrap_or(-1),
        String::from_utf8_lossy(&output.stdout).into_owned(),
        String::from_utf8_lossy(&output.stderr).into_owned(),
    ))
}

fn socket_of(dir: &Path) -> std::path::PathBuf {
    dir.join(".pushkin/daemon.sock")
}

fn bind_socket(dir: &Path) -> Result<UnixListener, Box<dyn std::error::Error>> {
    let socket = socket_of(dir);
    if let Some(parent) = socket.parent() {
        fs::create_dir_all(parent)?;
    }
    let listener = UnixListener::bind(&socket)?;
    listener.set_nonblocking(true)?;
    Ok(listener)
}

/// Accept-loop driver shared by both fakes. `serve` runs per accepted
/// connection; the returned counter reports how many connections arrived, which
/// is how the `off` case proves the socket was never touched.
fn spawn_fake(
    listener: UnixListener,
    serve: fn(&UnixStream) -> std::io::Result<()>,
) -> (std::thread::JoinHandle<()>, Arc<AtomicUsize>) {
    let connections = Arc::new(AtomicUsize::new(0));
    let counter = Arc::clone(&connections);
    let handle = std::thread::spawn(move || {
        let deadline = Instant::now() + Duration::from_secs(3);
        while Instant::now() < deadline {
            match listener.accept() {
                Ok((stream, _)) => {
                    counter.fetch_add(1, Ordering::SeqCst);
                    let _ = serve(&stream);
                }
                Err(_) => std::thread::sleep(Duration::from_millis(10)),
            }
        }
    });
    (handle, connections)
}

/// Answers every check with a canned ALLOW — the opposite of the real pipeline's
/// verdict on `NONCONFORMING`.
fn serve_allow(stream: &UnixStream) -> std::io::Result<()> {
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    reader.read_line(&mut line)?;
    let allow = serde_json::json!({
        "kind": "check",
        "result": { "decision": "allow", "violations": [], "durationMs": 0.42 }
    });
    let mut writer = stream;
    writer.write_all(allow.to_string().as_bytes())?;
    writer.write_all(b"\n")?;
    writer.flush()
}

/// Reads the request and hangs up without answering — a wedged or dying daemon.
/// `request_at` turns this into `Protocol("daemon closed the connection without
/// responding")`, a real reason with a real message, not a synthetic one.
fn serve_nothing(stream: &UnixStream) -> std::io::Result<()> {
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    reader.read_line(&mut line)?;
    Ok(())
}

fn assert_cold_verdict_intact(code: i32, stdout: &str) -> TestResult {
    assert_eq!(code, 0, "the Claude hook always exits 0: {stdout}");
    let json: serde_json::Value = serde_json::from_str(stdout)?;
    let hook = &json["hookSpecificOutput"];
    assert_eq!(
        hook["permissionDecision"], "deny",
        "the cold pipeline denies NONCONFORMING: {stdout}"
    );
    assert!(
        hook["permissionDecisionReason"]
            .as_str()
            .unwrap_or_default()
            .contains(RULE),
        "the full envelope survives: {stdout}"
    );
    Ok(())
}

// ---------- the four states ----------

#[test]
fn a_serving_daemon_discloses_that_the_warm_path_served() -> TestResult {
    let dir = repo()?;
    let listener = bind_socket(dir.path())?;
    let (fake, connections) = spawn_fake(listener, serve_allow);
    let (code, stdout, stderr) = run_hook(dir.path(), None)?;
    fake.join().map_err(|_| "fake daemon thread panicked")?;

    assert_eq!(
        connections.load(Ordering::SeqCst),
        1,
        "the socket is consulted"
    );
    assert!(
        stderr.contains("warm"),
        "the warm path must say it served; stderr was {stderr:?}"
    );
    // Verdict unchanged: the fake said allow, and an allow prints NOTHING.
    assert_eq!(code, 0);
    assert_eq!(
        stdout.trim(),
        "",
        "stdout is the verdict channel and disclosure must not enter it: {stdout:?}"
    );
    Ok(())
}

#[test]
fn no_daemon_discloses_the_cold_path_and_never_spawns_one() -> TestResult {
    let dir = repo()?;
    let (code, stdout, stderr) = run_hook(dir.path(), None)?;

    assert!(
        stderr.contains("cold"),
        "the cold path must say it served; stderr was {stderr:?}"
    );
    assert!(
        !socket_of(dir.path()).exists(),
        "an unset PUSHKIN_DAEMON probes and must never auto-start a daemon"
    );
    assert_cold_verdict_intact(code, &stdout)
}

/// **The finding.** A warm attempt was made, it failed, and the cold path
/// returned the correct verdict anyway. Before this pass that sequence was
/// indistinguishable from "no daemon was running" — same verdict, same silence.
#[test]
fn a_failed_warm_attempt_is_disclosed_with_its_reason() -> TestResult {
    let dir = repo()?;
    let listener = bind_socket(dir.path())?;
    let (fake, connections) = spawn_fake(listener, serve_nothing);
    let (code, stdout, stderr) = run_hook(dir.path(), None)?;
    fake.join().map_err(|_| "fake daemon thread panicked")?;

    assert_eq!(
        connections.load(Ordering::SeqCst),
        1,
        "the warm attempt genuinely reached the socket"
    );
    assert!(
        stderr.contains("cold"),
        "the cold path served, and must say so: {stderr:?}"
    );
    assert!(
        stderr.contains("failed"),
        "a FAILED warm attempt is the event that was invisible; it must be \
         disclosed, not merely implied by naming the cold path: {stderr:?}"
    );
    assert!(
        stderr.contains("without responding"),
        "the reason must be the real ServerError text, not a generic label: {stderr:?}"
    );
    assert_cold_verdict_intact(code, &stdout)
}

#[test]
fn a_failed_attempt_is_distinguishable_from_no_daemon_at_all() -> TestResult {
    let dir = repo()?;
    let listener = bind_socket(dir.path())?;
    let (fake, _) = spawn_fake(listener, serve_nothing);
    let (_, _, failed) = run_hook(dir.path(), None)?;
    fake.join().map_err(|_| "fake daemon thread panicked")?;

    let absent_dir = repo()?;
    let (_, _, absent) = run_hook(absent_dir.path(), None)?;

    assert_ne!(
        failed.trim(),
        absent.trim(),
        "these are different events and the record must tell them apart — \
         both served cold with the correct verdict, which is exactly why the \
         defect was invisible"
    );
    Ok(())
}

#[test]
fn daemon_off_discloses_cold_and_never_touches_the_socket() -> TestResult {
    let dir = repo()?;
    let listener = bind_socket(dir.path())?;
    let (fake, connections) = spawn_fake(listener, serve_allow);
    let (code, stdout, stderr) = run_hook(dir.path(), Some("off"))?;
    fake.join().map_err(|_| "fake daemon thread panicked")?;

    assert_eq!(
        connections.load(Ordering::SeqCst),
        0,
        "`off` must never touch the socket — a live daemon was listening and \
         must not have been consulted"
    );
    assert!(
        stderr.contains("cold"),
        "`off` still serves a verdict, and the path that served it is stated: {stderr:?}"
    );
    // The fake would have said ALLOW. A deny proves the cold pipeline ran.
    assert_cold_verdict_intact(code, &stdout)
}

// ---------- the verdict is byte-identical on every path ----------

#[test]
fn the_verdict_is_byte_identical_whichever_path_serves() -> TestResult {
    let absent_dir = repo()?;
    let (_, cold_stdout, _) = run_hook(absent_dir.path(), None)?;

    let off_dir = repo()?;
    let (_, off_stdout, _) = run_hook(off_dir.path(), Some("off"))?;

    let failed_dir = repo()?;
    let listener = bind_socket(failed_dir.path())?;
    let (fake, _) = spawn_fake(listener, serve_nothing);
    let (_, failed_stdout, _) = run_hook(failed_dir.path(), None)?;
    fake.join().map_err(|_| "fake daemon thread panicked")?;

    assert_eq!(
        cold_stdout, off_stdout,
        "disclosure must not perturb the verdict channel"
    );
    assert_eq!(
        cold_stdout, failed_stdout,
        "a failed warm attempt changes what is RECORDED, never what is DECIDED"
    );
    Ok(())
}

#[test]
fn disclosure_never_enters_the_verdict_channel() -> TestResult {
    let dir = repo()?;
    let (_, stdout, stderr) = run_hook(dir.path(), None)?;
    assert!(
        !stderr.is_empty(),
        "something must be disclosed for this assertion to mean anything"
    );
    assert!(
        !stdout.contains("cold") && !stdout.contains("warm"),
        "stdout is parsed by the host as a verdict; disclosure belongs on \
         stderr: {stdout:?}"
    );
    Ok(())
}