pushkin 0.2.0

//! Phase 4 task 2b: the shim side of the warm path (spec §4.3). Committed
//! first, read-only hereafter (charter §4.1, N10). Pins the resolution
//! order of `PUSHKIN_DAEMON` (env-only until the daemon's persisted
//! config lands — the Phase-3 nudge-mode pattern): unset probes the
//! socket and falls back cold WITHOUT spawning anything; `auto`
//! additionally auto-starts the daemon on first connect; `off` never
//! touches the socket. A daemon verdict and a cold verdict must be
//! indistinguishable (transport, not a second brain) — the fake-daemon
//! test proves the socket is genuinely consulted by flipping its answer.

use assert_cmd::Command;
use pushkin_daemon::protocol::{Request, PROTOCOL_VERSION};
use pushkin_daemon::server;
use std::fs;
use std::io::{BufRead, BufReader, Write};
use std::os::unix::net::UnixListener;
use std::path::Path;
use std::time::{Duration, Instant};

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]
protected_paths = ["pushkin.toml"]
"#;

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() -> std::io::Result<tempfile::TempDir> {
    let dir = tempfile::tempdir()?;
    fs::write(dir.path().join("pushkin.toml"), MANIFEST)?;
    Ok(dir)
}

fn claude_payload(content: &str) -> String {
    serde_json::json!({
        "session_id": "daemon-shim-session",
        "tool_name": "Write",
        "tool_input": { "file_path": HANDLER_PATH, "content": content }
    })
    .to_string()
}

/// Run `pushkin hook claude` in `dir` with the given `PUSHKIN_DAEMON`
/// value (`None` = unset), returning (exit code, stdout).
fn run_hook(dir: &Path, daemon_env: Option<&str>) -> (i32, String) {
    let Ok(mut command) = Command::cargo_bin("pushkin") else {
        return (-1, "cargo_bin resolution failed".to_owned());
    };
    command.current_dir(dir).args(["hook", "claude"]);
    match daemon_env {
        Some(value) => command.env("PUSHKIN_DAEMON", value),
        None => command.env_remove("PUSHKIN_DAEMON"),
    };
    let Ok(output) = command.write_stdin(claude_payload(NONCONFORMING)).output() else {
        return (-1, "spawn failed".to_owned());
    };
    (
        output.status.code().unwrap_or(-1),
        String::from_utf8_lossy(&output.stdout).into_owned(),
    )
}

/// A test-owned listener on the daemon socket that answers every check
/// with a canned ALLOW — the opposite of what the real pipeline says
/// about `NONCONFORMING`, so a flipped verdict proves the shim consulted
/// the socket. Accept loop is non-blocking with a deadline so tests that
/// never connect can still join the thread.
fn fake_allow_daemon(dir: &Path) -> std::io::Result<std::thread::JoinHandle<()>> {
    let socket = dir.join(".pushkin/daemon.sock");
    if let Some(parent) = socket.parent() {
        fs::create_dir_all(parent)?;
    }
    let listener = UnixListener::bind(&socket)?;
    listener.set_nonblocking(true)?;
    Ok(std::thread::spawn(move || {
        let deadline = Instant::now() + Duration::from_secs(3);
        while Instant::now() < deadline {
            match listener.accept() {
                Ok((stream, _)) => {
                    let _ = serve_one_allow(&stream);
                }
                Err(_) => std::thread::sleep(Duration::from_millis(10)),
            }
        }
    }))
}

fn serve_one_allow(stream: &std::os::unix::net::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()
}

fn socket_gone(dir: &Path) -> bool {
    let socket = dir.join(".pushkin/daemon.sock");
    let deadline = Instant::now() + Duration::from_secs(3);
    while Instant::now() < deadline {
        if !socket.exists() {
            return true;
        }
        std::thread::sleep(Duration::from_millis(25));
    }
    false
}

#[test]
fn hook_shim_falls_back_to_cold_path_when_daemon_unreachable() {
    let dir = repo().unwrap();
    let (code, stdout) = run_hook(dir.path(), None);
    assert_eq!(code, 0);
    let json: serde_json::Value = serde_json::from_str(&stdout).unwrap();
    assert_eq!(json["hookSpecificOutput"]["permissionDecision"], "deny");
    assert!(
        json["hookSpecificOutput"]["permissionDecisionReason"]
            .as_str()
            .unwrap()
            .contains(RULE),
        "cold fallback must carry the full envelope"
    );
    // Default mode probes; it must NOT have spawned a daemon.
    assert!(
        !dir.path().join(".pushkin/daemon.sock").exists(),
        "unset PUSHKIN_DAEMON must never auto-start a daemon"
    );
}

#[test]
fn warm_daemon_socket_consulted_when_present() {
    let dir = repo().unwrap();
    let fake = fake_allow_daemon(dir.path()).unwrap();
    let (code, stdout) = run_hook(dir.path(), None);
    fake.join().unwrap();
    assert_eq!(code, 0);
    assert_eq!(
        stdout.trim(),
        "",
        "the fake daemon said allow; a silent allow proves the socket was \
         consulted (got: {stdout})"
    );
}

#[test]
fn daemon_autostarts_on_first_shim_connect() {
    let dir = repo().unwrap();
    let (code, stdout) = run_hook(dir.path(), Some("auto"));
    assert_eq!(code, 0);
    // The real daemon runs the real pipeline: still a deny.
    let json: serde_json::Value = serde_json::from_str(&stdout).unwrap();
    assert_eq!(json["hookSpecificOutput"]["permissionDecision"], "deny");
    // ...and the daemon it started is still warm behind the socket.
    let socket = dir.path().join(".pushkin/daemon.sock");
    assert!(socket.exists(), "auto mode must leave the daemon running");

    let response = server::request(
        dir.path(),
        &Request::Shutdown {
            v: PROTOCOL_VERSION,
        },
    );
    assert!(response.is_ok(), "auto-started daemon must speak protocol");
    assert!(
        socket_gone(dir.path()),
        "shutdown must remove the auto-started daemon's socket"
    );
}

#[test]
fn env_off_forces_cold_path() {
    let dir = repo().unwrap();
    let fake = fake_allow_daemon(dir.path()).unwrap();
    // The fake daemon would say allow — `off` must never ask it.
    let (code, stdout) = run_hook(dir.path(), Some("off"));
    fake.join().unwrap();
    assert_eq!(code, 0);
    let json: serde_json::Value = serde_json::from_str(&stdout).unwrap();
    assert_eq!(
        json["hookSpecificOutput"]["permissionDecision"], "deny",
        "off means the socket is never consulted"
    );
}