pointbreak 0.8.0

//! Shared HTTP harness for the `pointbreak inspect` integration suites.
//!
//! The inspector's JSON builders live in the binary crate
//! (`src/cli/inspect/api.rs`), so they are not reachable from an integration
//! test by a direct call. These tests instead exercise the genuine production
//! JSON end to end: they spawn the real `pointbreak inspect --port 0` server (which
//! prints its bound URL and supports an ephemeral port) and issue raw HTTP/1.1
//! GETs. The store is always built at test time through the `pointbreak` CLI, so it
//! tracks the current on-disk layout without hard-coding any store path.

use std::ffi::OsString;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{Shutdown, TcpStream};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

use serde_json::Value;

use super::git_repo::GitRepo;
use super::pointbreak;

/// A repository plus a worktree on a fresh branch with one captured Revision.
pub struct WorktreeCapture {
    pub _main: GitRepo,
    pub _parent: tempfile::TempDir,
    pub worktree: PathBuf,
    pub revision_id: String,
}

impl WorktreeCapture {
    pub fn on_branch(dir_name: &str, branch: &str) -> Self {
        let main = GitRepo::new();
        main.write("src/lib.rs", "pub fn value() -> u32 { 1 }\n");
        main.commit_all("base");

        let parent = tempfile::tempdir().expect("worktree parent");
        let worktree = parent.path().join(dir_name);
        add_worktree(main.path(), &worktree, branch);
        std::fs::write(worktree.join("src/lib.rs"), "pub fn value() -> u32 { 2 }\n").unwrap();

        let revision_id = capture(&worktree);

        Self {
            _main: main,
            _parent: parent,
            worktree,
            revision_id,
        }
    }
}

/// A spawned `pointbreak inspect` server bound to an ephemeral port, killed on drop.
pub struct Inspector {
    child: Child,
    addr: String,
    startup_output: String,
    bearer: Option<String>,
    stderr: Arc<Mutex<String>>,
    _stdout_drain: thread::JoinHandle<()>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum InspectSurface {
    Web,
    ApiOnly,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum InspectOutput {
    Text,
    Json,
}

impl Inspector {
    pub fn spawn(repo: &Path) -> Self {
        Self::spawn_web_text(repo)
    }

    pub fn spawn_human(repo: &Path) -> Self {
        Self::spawn_web_text(repo)
    }

    pub fn spawn_authenticated(repo: &Path) -> Self {
        Self::spawn_api_json(repo)
    }

    pub fn spawn_web_text(repo: &Path) -> Self {
        Self::spawn_with(repo, InspectSurface::Web, InspectOutput::Text)
    }

    pub fn spawn_web_json(repo: &Path) -> Self {
        Self::spawn_with(repo, InspectSurface::Web, InspectOutput::Json)
    }

    pub fn spawn_api_text(repo: &Path) -> Self {
        Self::spawn_with(repo, InspectSurface::ApiOnly, InspectOutput::Text)
    }

    pub fn spawn_api_json(repo: &Path) -> Self {
        Self::spawn_with(repo, InspectSurface::ApiOnly, InspectOutput::Json)
    }

    fn spawn_with(repo: &Path, surface: InspectSurface, output: InspectOutput) -> Self {
        let mut command = Command::new(env!("CARGO_BIN_EXE_pointbreak"));
        command.args([
            "inspect",
            "--repo",
            repo.to_str().unwrap(),
            "--host",
            "127.0.0.1",
            "--port",
            "0",
        ]);
        if surface == InspectSurface::ApiOnly {
            command.arg("--api-only");
        }
        if output == InspectOutput::Json {
            command.args(["--format", "json"]);
        }
        let mut child = command
            .env_remove("POINTBREAK_LOG")
            .env_remove("RUST_LOG")
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .expect("spawn pointbreak inspect");

        // Drain stderr in the background so it never blocks the server and is
        // available to explain a failure.
        let stderr = Arc::new(Mutex::new(String::new()));
        let mut child_stderr = child.stderr.take().expect("inspector stderr");
        {
            let sink = Arc::clone(&stderr);
            thread::spawn(move || {
                let mut buffer = String::new();
                let _ = child_stderr.read_to_string(&mut buffer);
                if let Ok(mut guard) = sink.lock() {
                    *guard = buffer;
                }
            });
        }

        // Read the complete startup output for the selected mode, then keep
        // draining stdout so the server never stalls on a full pipe.
        let stdout = child.stdout.take().expect("inspector stdout");
        let mut reader = BufReader::new(stdout);
        let mut startup_output = String::new();
        let line_count = if output == InspectOutput::Json { 1 } else { 4 };
        for _ in 0..line_count {
            let mut line = String::new();
            match reader.read_line(&mut line) {
                Ok(0) | Err(_) => break,
                Ok(_) => startup_output.push_str(&line),
            }
        }

        let (addr, bearer) = if output == InspectOutput::Json {
            let startup: Value =
                serde_json::from_str(startup_output.trim()).unwrap_or_else(|error| {
                    panic!(
                        "parse JSON inspector startup: {error}; stderr: {}",
                        drained(&stderr)
                    )
                });
            let host = startup["host"].as_str().expect("startup host");
            let port = startup["port"].as_u64().expect("startup port");
            let token = startup["token"].as_str().expect("startup token").to_owned();
            (format!("{host}:{port}"), Some(token))
        } else if surface == InspectSurface::Web {
            let capability = startup_output
                .lines()
                .find_map(|line| line.split_once("http://").map(|(_, value)| value.trim()))
                .unwrap_or_default();
            let (addr, fragment) = capability
                .split_once("/#")
                .expect("text capability has a fragment route");
            let token = fragment
                .split_once('?')
                .map(|(_, query)| query)
                .and_then(|query| {
                    query
                        .split('&')
                        .find_map(|pair| pair.strip_prefix("token="))
                })
                .map(str::to_owned)
                .expect("text web startup capability token");
            (addr.to_owned(), Some(token))
        } else {
            let addr = startup_output
                .lines()
                .find_map(|line| line.strip_prefix("  endpoint: http://"))
                .map(|value| value.trim().trim_end_matches('/').to_owned())
                .unwrap_or_default();
            let token = startup_output
                .lines()
                .find_map(|line| line.strip_prefix("  token: "))
                .map(str::to_owned);
            (addr, token)
        };
        let stdout_drain = thread::spawn(move || {
            let mut sink = String::new();
            let _ = reader.read_to_string(&mut sink);
        });

        assert!(
            !addr.is_empty(),
            "inspector did not print a bound url; stderr: {}",
            drained(&stderr)
        );

        // Wait until the server actually accepts connections, failing fast with
        // diagnostics if it exits before listening.
        let mut ready = false;
        for _ in 0..100 {
            if TcpStream::connect(&addr).is_ok() {
                ready = true;
                break;
            }
            if let Ok(Some(status)) = child.try_wait() {
                panic!(
                    "inspector exited before listening (status {status}) at {addr}; stderr: {}",
                    drained(&stderr)
                );
            }
            thread::sleep(Duration::from_millis(20));
        }
        assert!(
            ready,
            "inspector never accepted a connection at {addr}; stderr: {}",
            drained(&stderr)
        );

        Self {
            child,
            addr,
            startup_output,
            bearer,
            stderr,
            _stdout_drain: stdout_drain,
        }
    }

    pub fn startup_output(&self) -> &str {
        &self.startup_output
    }

    pub fn canonical_host(&self) -> &str {
        &self.addr
    }

    pub fn token(&self) -> Option<&str> {
        self.bearer.as_deref()
    }

    pub fn stderr_text(&self) -> String {
        drained(&self.stderr)
    }

    pub fn get_json(&self, path: &str) -> Value {
        let body = self.get_text(path);
        serde_json::from_str(&body).unwrap_or_else(|error| panic!("parse {path} body: {error}"))
    }

    /// GET a path expected to succeed, returning the raw 200 response body.
    ///
    /// The inspector is a blocking HTTP/1.1 server that closes each connection
    /// after responding. Under load (notably on Windows and Linux CI) the close can race
    /// ahead of the client's read and surface as a connection reset before the
    /// body is drained. GETs are idempotent, so retry a few times with a short
    /// backoff before giving up.
    pub fn get_text(&self, path: &str) -> String {
        let mut last_error = String::new();
        for attempt in 0..12 {
            match self.try_get(path) {
                Ok((_, body)) => return body,
                Err(error) => {
                    last_error = error;
                    thread::sleep(Duration::from_millis(20 * (attempt + 1)));
                }
            }
        }
        panic!(
            "GET {path} failed after retries: {last_error}; server stderr: {}",
            drained(&self.stderr)
        );
    }

    /// GET a path expected to fail, returning the raw status head and the
    /// parsed JSON error body.
    pub fn get_error(&self, path: &str) -> (String, Value) {
        let (status, body) = self.raw_get(path);
        let body: Value = serde_json::from_str(&body).expect("error body is json");
        (status, body)
    }

    /// GET a path returning the raw status line and body, with no status
    /// assertion — for socket-level route/error coverage.
    pub fn raw_get(&self, path: &str) -> (String, String) {
        self.request_with_retry("GET", path, &self.default_headers())
    }

    pub fn raw_request(
        &self,
        method: &str,
        path: &str,
        headers: &[(&str, &str)],
    ) -> (String, String) {
        let headers = headers
            .iter()
            .map(|(name, value)| ((*name).to_owned(), (*value).to_owned()))
            .collect::<Vec<_>>();
        self.request_with_retry(method, path, &headers)
    }

    /// Retry transport failures while draining responses from the read-only
    /// inspector. Non-GET methods are rejected before any application action,
    /// so replaying these harness requests cannot duplicate a write.
    fn request_with_retry(
        &self,
        method: &str,
        path: &str,
        headers: &[(String, String)],
    ) -> (String, String) {
        let mut last_error = String::new();
        for attempt in 0..12 {
            match self.try_request(method, path, headers) {
                Ok(response) => return response,
                Err(error) => {
                    last_error = error;
                    thread::sleep(Duration::from_millis(20 * (attempt + 1)));
                }
            }
        }
        panic!(
            "{method} {path} failed after retries: {last_error}; server stderr: {}",
            drained(&self.stderr)
        );
    }

    fn try_request(
        &self,
        method: &str,
        path: &str,
        headers: &[(String, String)],
    ) -> Result<(String, String), String> {
        let mut stream = TcpStream::connect(&self.addr).map_err(|error| error.to_string())?;
        let mut request = format!("{method} {path} HTTP/1.1\r\n");
        for (name, value) in headers {
            request.push_str(name);
            request.push_str(": ");
            request.push_str(value);
            request.push_str("\r\n");
        }
        request.push_str("Connection: close\r\n\r\n");
        stream
            .write_all(request.as_bytes())
            .map_err(|error| error.to_string())?;
        let _ = stream.shutdown(Shutdown::Write);
        let mut response = Vec::new();
        stream
            .read_to_end(&mut response)
            .map_err(|error| error.to_string())?;
        let text = String::from_utf8_lossy(&response);
        let (head, body) = text
            .split_once("\r\n\r\n")
            .ok_or_else(|| "response has no header/body delimiter".to_owned())?;
        Ok((head.to_owned(), body.to_owned()))
    }

    fn default_headers(&self) -> Vec<(String, String)> {
        let mut headers = vec![("Host".to_owned(), self.addr.clone())];
        if let Some(token) = self.bearer.as_deref() {
            headers.push(("Authorization".to_owned(), format!("Bearer {token}")));
        }
        headers
    }

    /// Issue a raw request line (method + target) and return the status line,
    /// for exercising non-GET routes.
    pub fn request(&self, method: &str, path: &str) -> String {
        let (head, _) = self.request_with_retry(method, path, &self.default_headers());
        head.lines().next().unwrap_or_default().to_owned()
    }

    fn try_get(&self, path: &str) -> Result<(String, String), String> {
        let (head, body) = self.try_request("GET", path, &self.default_headers())?;
        if !head.starts_with("HTTP/1.1 200") {
            return Err(format!("unexpected status for {path}"));
        }
        Ok((head, body))
    }
}

impl Drop for Inspector {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
    }
}

/// Snapshot the background-captured server stderr, after a brief flush window so
/// an early-exiting server's output has a chance to land.
fn drained(stderr: &Arc<Mutex<String>>) -> String {
    thread::sleep(Duration::from_millis(50));
    stderr.lock().map(|guard| guard.clone()).unwrap_or_default()
}

/// Run `pointbreak capture` against a repo, returning the captured Revision id.
pub fn capture(repo: &Path) -> String {
    let output = pointbreak(["capture", "--repo", repo.to_str().unwrap(), "--allow-empty"]);
    assert!(
        output.status.success(),
        "capture stderr:\n{}",
        String::from_utf8_lossy(&output.stderr)
    );
    let json: Value = serde_json::from_slice(&output.stdout).expect("parse capture JSON");
    json["revision"]["id"]
        .as_str()
        .expect("capture returns a Revision id")
        .to_owned()
}

/// A populated store assembled once through the `pointbreak` CLI, reused by the
/// endpoint-contract and validation read-surface suites: one captured Revision
/// carrying a range-targeted observation, an operative input request, a
/// superseded + a superseding assessment (two tracks), and two validation checks
/// (passed `cargo test` on the agent track, failed `cargo clippy` on the human
/// track).
pub struct RepresentativeStore {
    pub repo: GitRepo,
    pub revision_id: String,
    pub snapshot_id: String,
}

/// Generated ids printed by the isolated decision-continuity materializer.
///
/// These values are resolved from real command output on every build. Keeping
/// them together lets socket suites reuse one generated store without pinning
/// timestamp-derived event or association identities.
#[derive(Debug, serde::Deserialize)]
pub struct DecisionContinuityMatrixIds {
    pub primary_revision: String,
    pub live_revision: String,
    pub unassessed_revision: String,
    pub superseded_revision: String,
    pub ambiguous_assessment_revision: String,
    pub competing_revision: String,
    pub range_revision: String,
    pub root_revision: String,
    pub staged_revision: String,
    pub unstaged_revision: String,
    pub detached_revision: String,
    pub missing_revision: String,
    pub base_commit: String,
    pub first_landing: String,
    pub second_landing: String,
    pub live_landing: String,
}

/// An isolated repository/store carrying the generated decision matrix.
pub struct DecisionContinuityMatrix {
    _root: tempfile::TempDir,
    repo: PathBuf,
    pub ids: DecisionContinuityMatrixIds,
}

impl DecisionContinuityMatrix {
    pub fn repo(&self) -> &Path {
        &self.repo
    }
}

fn decision_matrix_shell() -> Command {
    #[cfg(windows)]
    {
        let git_exec_path = Command::new("git")
            .arg("--exec-path")
            .output()
            .expect("locate Git for Windows");
        assert!(
            git_exec_path.status.success(),
            "git --exec-path failed: {}",
            String::from_utf8_lossy(&git_exec_path.stderr)
        );
        let git_exec_path =
            String::from_utf8(git_exec_path.stdout).expect("Git for Windows exec path is UTF-8");
        let bash = Path::new(git_exec_path.trim())
            .ancestors()
            .map(|ancestor| ancestor.join("bin/bash.exe"))
            .find(|candidate| candidate.is_file())
            .unwrap_or_else(|| {
                panic!(
                    "could not find Git Bash above git exec path {}",
                    git_exec_path.trim()
                )
            });
        Command::new(bash)
    }

    #[cfg(not(windows))]
    {
        Command::new("bash")
    }
}

/// Materialize the reusable synthetic matrix with the exact binary built for
/// the integration test. The generator owns a fresh temporary repository, and
/// it rejects any store path that escapes that repository.
pub fn decision_continuity_matrix() -> DecisionContinuityMatrix {
    let root = tempfile::tempdir().expect("decision matrix root");
    let repo = root.path().join("repository");
    let script = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .join("scripts/materialize-inspector-decision-matrix.sh");
    let output = decision_matrix_shell()
        .arg(&script)
        .arg(&repo)
        .env("POINTBREAK_BINARY", env!("CARGO_BIN_EXE_pointbreak"))
        .env_remove("POINTBREAK_HOME")
        .env_remove("POINTBREAK_FORMAT")
        .env_remove("POINTBREAK_SIGNING_KEY")
        .output()
        .unwrap_or_else(|error| panic!("run {}: {error}", script.display()));
    assert!(
        output.status.success(),
        "decision matrix materialization failed:\nstdout:\n{}\nstderr:\n{}",
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    let ids = serde_json::from_slice(&output.stdout)
        .unwrap_or_else(|error| panic!("parse generated matrix ids: {error}"));

    DecisionContinuityMatrix {
        _root: root,
        repo,
        ids,
    }
}

/// Build the [`RepresentativeStore`] entirely through the real CLI, so it tracks
/// the current on-disk store layout without hard-coding any path.
pub fn representative_store() -> RepresentativeStore {
    let repo = GitRepo::new();
    repo.write(
        "src/lib.rs",
        "pub fn value() -> u32 {\n    1\n}\n\npub fn other() -> u32 {\n    2\n}\n",
    );
    repo.commit_all("base");
    repo.write(
        "src/lib.rs",
        "pub fn value() -> u32 {\n    42\n}\n\npub fn other() -> u32 {\n    7\n}\n",
    );

    let repo_arg = repo.path().to_str().unwrap().to_owned();
    let capture = run_shore_json(&["capture", "--repo", &repo_arg]);
    let revision_id = capture["revision"]["id"]
        .as_str()
        .expect("capture returns a Revision id")
        .to_owned();
    let snapshot_id = capture["revision"]["objectId"]
        .as_str()
        .expect("capture returns a snapshot id")
        .to_owned();

    // Range-targeted observation on the agent track.
    run_shore(&[
        "observation",
        "add",
        "--repo",
        &repo_arg,
        "--track",
        "agent:codex",
        "--title",
        "Observed change",
        "--body",
        "the return value changed",
        "--file",
        "src/lib.rs",
        "--start-line",
        "2",
        "--end-line",
        "2",
    ]);

    // Operative input request.
    run_shore(&[
        "input-request",
        "open",
        "--repo",
        &repo_arg,
        "--track",
        "agent:codex",
        "--title",
        "Need a decision",
        "--reason",
        "manual-decision-required",
        "--body",
        "should we ship this?",
    ]);

    // A first assessment (agent track), then a superseding assessment (human
    // track) that replaces it, so current-assessment resolution is exercised.
    let first = run_shore_json(&[
        "assessment",
        "add",
        "--repo",
        &repo_arg,
        "--track",
        "agent:codex",
        "--assessment",
        "needs-changes",
        "--summary",
        "not yet",
    ]);
    let first_assessment_id = first["assessmentId"]
        .as_str()
        .expect("assessment add returns an assessment id")
        .to_owned();
    run_shore(&[
        "assessment",
        "add",
        "--repo",
        &repo_arg,
        "--track",
        "human:kevin",
        "--assessment",
        "accepted",
        "--summary",
        "ship it",
        "--replaces",
        &first_assessment_id,
    ]);

    // Two validation checks across two tracks: a passed cargo test and a failed
    // cargo clippy carrying an exit code + command.
    run_shore(&[
        "validation",
        "add",
        "--repo",
        &repo_arg,
        "--track",
        "agent:codex",
        "--check-name",
        "cargo test",
        "--status",
        "passed",
    ]);
    run_shore(&[
        "validation",
        "add",
        "--repo",
        &repo_arg,
        "--track",
        "human:kevin",
        "--check-name",
        "cargo clippy",
        "--status",
        "failed",
        "--exit-code",
        "1",
        "--command",
        "cargo clippy -- -D warnings",
    ]);

    RepresentativeStore {
        repo,
        revision_id,
        snapshot_id,
    }
}

/// Run a `pointbreak` subcommand, asserting success and surfacing stderr on failure.
fn run_shore(args: &[&str]) {
    let output = pointbreak(args);
    assert!(
        output.status.success(),
        "pointbreak {args:?} failed:\n{}",
        String::from_utf8_lossy(&output.stderr)
    );
}

/// Run a `pointbreak` subcommand and parse its stdout as JSON.
fn run_shore_json(args: &[&str]) -> Value {
    let output = pointbreak(args);
    assert!(
        output.status.success(),
        "pointbreak {args:?} failed:\n{}",
        String::from_utf8_lossy(&output.stderr)
    );
    serde_json::from_slice(&output.stdout)
        .unwrap_or_else(|error| panic!("parse pointbreak {args:?} JSON: {error}"))
}

/// Capture a revision in a supersession thread. With no `predecessor`, captures
/// the current worktree as a fresh revision. With a `predecessor`, mutates a
/// tracked file first (so the snapshot id differs) and records the new revision
/// as superseding the predecessor. Returns the captured revision id.
pub fn capture_supersession_round(repo: &Path, predecessor: Option<&str>) -> String {
    let mut args = vec![
        "capture".to_owned(),
        "--repo".to_owned(),
        repo.to_str().unwrap().to_owned(),
    ];
    if let Some(predecessor) = predecessor {
        // A successor must carry different content, or it would collapse to the
        // same snapshot id. Append a unique line to a tracked file.
        let target = repo.join("src/lib.rs");
        let mut contents = std::fs::read_to_string(&target).unwrap_or_default();
        contents.push_str(&format!(
            "\n// supersedes {}\n",
            predecessor.replace(':', "_")
        ));
        std::fs::write(&target, contents).expect("write successor content");
        args.push("--supersedes".to_owned());
        args.push(predecessor.to_owned());
    } else {
        args.push("--allow-empty".to_owned());
    }

    let output = pointbreak(args);
    assert!(
        output.status.success(),
        "capture supersession round stderr:\n{}",
        String::from_utf8_lossy(&output.stderr)
    );
    let json: Value =
        serde_json::from_slice(&output.stdout).expect("parse supersession capture JSON");
    json["revision"]["id"]
        .as_str()
        .expect("supersession capture returns a Revision id")
        .to_owned()
}

pub fn add_worktree(repo: &Path, path: &Path, branch: &str) {
    run_git(
        repo,
        [
            OsString::from("worktree"),
            OsString::from("add"),
            OsString::from("-b"),
            OsString::from(branch),
            path.as_os_str().to_owned(),
        ],
    );
}

pub fn run_git<I>(cwd: &Path, args: I)
where
    I: IntoIterator<Item = OsString>,
{
    let output = Command::new("git")
        .args(args)
        .current_dir(cwd)
        .output()
        .unwrap_or_else(|error| panic!("run git in {}: {error}", cwd.display()));
    assert!(
        output.status.success(),
        "git failed in {}:\n{}",
        cwd.display(),
        String::from_utf8_lossy(&output.stderr)
    );
}

/// Minimal percent-encoding for the `:` characters in a Revision id query.
pub fn urlencode(value: &str) -> String {
    value.replace(':', "%3A")
}