app.py

from __future__ import annotations

import base64
import hashlib
import json
import os
import shutil
import subprocess
import tempfile
import time
import zipfile
from pathlib import Path
from typing import Iterator

import gradio as gr

DEFAULT_REPOSITORY = "toxzak-svg/starfire"
DEFAULT_COMMIT = "902c88e533a75076542726d5975880ef0234d54c"
DEFAULT_FEATURE = "omega-g2-shadow"
SCIENTIFIC_PATHS = (
    "lib/omega_g2_shadow.rs",
    "lib/prediction/mod.rs",
    "lib/examples/omega_g2_s0_real_trace_shadow.rs",
)
MAX_LOG_CHARS = 180_000
ARTIFACT_ROOT = Path("/tmp/starfire-verifier-artifacts")
ARTIFACT_ROOT.mkdir(parents=True, exist_ok=True)


def _append_log(log: str, line: str) -> str:
    combined = f"{log}{line}\n"
    if len(combined) > MAX_LOG_CHARS:
        combined = "[earlier output truncated]\n" + combined[-MAX_LOG_CHARS:]
    return combined


def _run_streaming(
    command: list[str],
    cwd: Path,
    env: dict[str, str],
    display: str,
) -> Iterator[tuple[str, int | None]]:
    yield f"\n$ {display}\n", None
    process = subprocess.Popen(
        command,
        cwd=cwd,
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        encoding="utf-8",
        errors="replace",
        bufsize=1,
    )
    assert process.stdout is not None
    for line in process.stdout:
        yield line, None
    code = process.wait()
    yield f"[exit {code}]\n", code


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _source_hashes(repo: Path) -> dict[str, str]:
    hashes: dict[str, str] = {}
    for relative in SCIENTIFIC_PATHS:
        target = repo / relative
        if not target.is_file():
            raise FileNotFoundError(f"required scientific source missing: {relative}")
        hashes[relative] = _sha256(target)
    return hashes


def _ensure_rust(env: dict[str, str], work: Path) -> Iterator[tuple[str, int | None]]:
    cargo = shutil.which("cargo", path=env.get("PATH"))
    if cargo:
        yield f"Using existing Cargo: {cargo}\n", 0
        return

    installer = work / "rustup-init.sh"
    download = [
        "curl",
        "--proto",
        "=https",
        "--tlsv1.2",
        "-sSf",
        "https://sh.rustup.rs",
        "-o",
        str(installer),
    ]
    yield from _run_streaming(download, work, env, "download rustup installer")
    if not installer.is_file():
        yield "rustup installer was not downloaded\n", 1
        return
    install = ["sh", str(installer), "-y", "--profile", "minimal", "--default-toolchain", "stable"]
    yield from _run_streaming(install, work, env, "install stable Rust")


def _clone_private_repo(
    repository: str,
    commit: str,
    token: str,
    work: Path,
    env: dict[str, str],
) -> Iterator[tuple[str, int | None]]:
    askpass = work / "git-askpass.sh"
    askpass.write_text(
        "#!/bin/sh\n"
        "case \"$1\" in\n"
        "  *Username*) printf '%s\\n' 'x-access-token' ;;\n"
        "  *) printf '%s\\n' \"$GITHUB_TOKEN\" ;;\n"
        "esac\n",
        encoding="utf-8",
    )
    askpass.chmod(0o700)
    env["GIT_ASKPASS"] = str(askpass)
    env["GIT_TERMINAL_PROMPT"] = "0"
    env["GITHUB_TOKEN"] = token

    repo_dir = work / "starfire"
    clone = ["git", "clone", "--no-checkout", "--filter=blob:none", f"https://github.com/{repository}.git", str(repo_dir)]
    yield from _run_streaming(clone, work, env, f"git clone --no-checkout https://github.com/{repository}.git starfire")
    if not repo_dir.is_dir():
        return
    fetch = ["git", "fetch", "--depth", "1", "origin", commit]
    yield from _run_streaming(fetch, repo_dir, env, f"git fetch --depth 1 origin {commit}")
    checkout = ["git", "checkout", "--detach", commit]
    yield from _run_streaming(checkout, repo_dir, env, f"git checkout --detach {commit}")


def _write_evidence(
    artifact_dir: Path,
    report: dict[str, object],
    log: str,
    before: dict[str, str],
    after: dict[str, str],
) -> Path:
    artifact_dir.mkdir(parents=True, exist_ok=True)
    (artifact_dir / "report.json").write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    (artifact_dir / "validation.log").write_text(log, encoding="utf-8")
    (artifact_dir / "source-before.sha256.json").write_text(json.dumps(before, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    (artifact_dir / "source-after.sha256.json").write_text(json.dumps(after, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    archive = artifact_dir.with_suffix(".zip")
    with zipfile.ZipFile(archive, "w", compression=zipfile.ZIP_DEFLATED) as bundle:
        for path in sorted(artifact_dir.iterdir()):
            bundle.write(path, arcname=path.name)
    return archive


def validate(repository: str, commit: str) -> Iterator[tuple[str, str, str | None]]:
    repository = repository.strip()
    commit = commit.strip()
    token = os.environ.get("GITHUB_TOKEN", "").strip()
    if not token:
        message = (
            "BLOCKED: Space secret GITHUB_TOKEN is missing. Add a fine-grained, read-only token "
            "for toxzak-svg/starfire under Settings → Variables and secrets, then run again."
        )
        yield message, "BLOCKED", None
        return
    if repository != DEFAULT_REPOSITORY:
        yield "BLOCKED: repository must remain toxzak-svg/starfire for this frozen verifier.", "BLOCKED", None
        return
    if len(commit) != 40 or any(ch not in "0123456789abcdefABCDEF" for ch in commit):
        yield "BLOCKED: commit must be a full 40-character hexadecimal Git SHA.", "BLOCKED", None
        return

    started = int(time.time())
    run_id = f"omega-g2-s0-{started}"
    artifact_dir = ARTIFACT_ROOT / run_id
    log = ""
    status = "RUNNING"
    archive: Path | None = None
    before: dict[str, str] = {}
    after: dict[str, str] = {}
    command_results: list[dict[str, object]] = []
    terminal_probe_pass = False

    with tempfile.TemporaryDirectory(prefix="starfire-hf-") as temp:
        work = Path(temp)
        env = os.environ.copy()
        env["CARGO_HOME"] = str(work / ".cargo")
        env["RUSTUP_HOME"] = str(work / ".rustup")
        env["CARGO_TERM_COLOR"] = "never"
        env["RUST_BACKTRACE"] = "1"
        env["PATH"] = f"{env['CARGO_HOME']}/bin:{env.get('PATH', '')}"

        try:
            for chunk, code in _ensure_rust(env, work):
                log = _append_log(log, chunk.rstrip("\n"))
                yield log, status, None
                if code not in (None, 0):
                    raise RuntimeError("stable Rust installation failed")

            for chunk, code in _clone_private_repo(repository, commit, token, work, env):
                log = _append_log(log, chunk.rstrip("\n"))
                yield log, status, None
                if code not in (None, 0):
                    raise RuntimeError("private repository checkout failed")

            repo = work / "starfire"
            resolved = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=repo, env=env, text=True).strip()
            if resolved.lower() != commit.lower():
                raise RuntimeError(f"checkout mismatch: expected {commit}, got {resolved}")
            before = _source_hashes(repo)
            log = _append_log(log, f"Pinned source confirmed: {resolved}")
            log = _append_log(log, "Recorded pre-execution scientific source hashes")
            yield log, status, None

            commands = [
                (
                    ["cargo", "check", "-p", "star", "--all-targets", "--locked"],
                    "cargo check -p star --all-targets --locked",
                ),
                (
                    ["cargo", "check", "-p", "star", "--all-targets", "--features", DEFAULT_FEATURE, "--locked"],
                    f"cargo check -p star --all-targets --features {DEFAULT_FEATURE} --locked",
                ),
                (
                    ["cargo", "test", "-p", "star", "omega_g2_shadow", "--features", DEFAULT_FEATURE, "--locked"],
                    f"cargo test -p star omega_g2_shadow --features {DEFAULT_FEATURE} --locked",
                ),
                (
                    ["cargo", "run", "-p", "star", "--example", "omega_g2_s0_real_trace_shadow", "--features", DEFAULT_FEATURE, "--locked"],
                    f"cargo run -p star --example omega_g2_s0_real_trace_shadow --features {DEFAULT_FEATURE} --locked",
                ),
            ]

            for index, (command, display) in enumerate(commands):
                output_for_command = ""
                final_code: int | None = None
                for chunk, code in _run_streaming(command, repo, env, display):
                    output_for_command += chunk
                    log = _append_log(log, chunk.rstrip("\n"))
                    yield log, status, None
                    if code is not None:
                        final_code = code
                command_results.append({"command": display, "exit_code": final_code})
                if final_code != 0:
                    raise RuntimeError(f"validation command failed: {display}")
                if index == 3:
                    terminal_probe_pass = '"terminal_classification": "PASS"' in output_for_command
                    if not terminal_probe_pass:
                        raise RuntimeError("probe completed but did not emit terminal PASS")

            after = _source_hashes(repo)
            if before != after:
                raise RuntimeError("scientific source hashes changed during execution")
            diff = subprocess.run(
                ["git", "diff", "--exit-code", "--", *SCIENTIFIC_PATHS],
                cwd=repo,
                env=env,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
            )
            log = _append_log(log, diff.stdout.rstrip("\n")) if diff.stdout else log
            if diff.returncode != 0:
                raise RuntimeError("git source-immutability check failed")

            status = "PASS"
            log = _append_log(log, "terminal_classification: PASS")
        except Exception as exc:  # surfaced into evidence, never hidden
            status = "FAIL"
            log = _append_log(log, f"terminal_classification: FAIL")
            log = _append_log(log, f"error: {type(exc).__name__}: {exc}")
        finally:
            env.pop("GITHUB_TOKEN", None)
            report: dict[str, object] = {
                "terminal_classification": status,
                "repository": repository,
                "requested_commit": commit,
                "started_at_unix": started,
                "finished_at_unix": int(time.time()),
                "feature": DEFAULT_FEATURE,
                "commands": command_results,
                "probe_emitted_pass": terminal_probe_pass,
                "source_immutable": bool(before) and before == after,
                "scientific_paths": list(SCIENTIFIC_PATHS),
                "authority": {
                    "response_influence": False,
                    "routing": False,
                    "persistence_mutation": False,
                    "belief_or_ontology_promotion": False,
                    "tool_selection": False,
                    "external_action": False,
                    "source_modification": False,
                },
            }
            archive = _write_evidence(artifact_dir, report, log, before, after)

    yield log, status, str(archive) if archive else None


with gr.Blocks(title="Starfire ΩG2-S0 Verifier") as demo:
    gr.Markdown(
        """
# Starfire ΩG2-S0 Exact-Source Verifier

Runs the frozen, default-off shadow integration checks against one exact private Git commit.  
The verifier **observes and reports only**. It has no live response, routing, persistence, ontology, tool, or action authority.
"""
    )
    with gr.Row():
        repository = gr.Textbox(label="Repository", value=DEFAULT_REPOSITORY, interactive=False)
        commit = gr.Textbox(label="Exact commit SHA", value=DEFAULT_COMMIT)
    run = gr.Button("Run exact-source validation", variant="primary")
    classification = gr.Textbox(label="Terminal classification", value="NOT RUN", interactive=False)
    logs = gr.Textbox(label="Streaming validation log", lines=28, max_lines=40, interactive=False)
    evidence = gr.File(label="Evidence bundle")
    run.click(validate, inputs=[repository, commit], outputs=[logs, classification, evidence], concurrency_limit=1)

demo.queue(default_concurrency_limit=1, max_size=2).launch(
    server_name="0.0.0.0",
    server_port=7860,
    show_error=True,
    allowed_paths=[str(ARTIFACT_ROOT)],
)