"""Loopback-only CI execution backend that speaks oakspace's CI protocol.

oakspace dispatches a signed job spec to ``OAK_CI_DISPATCH_URL`` (``.../jobs``)

and expects the execution backend to clone the exact commit through the

token-scoped ``/api/ci/fetch/<repo_id>`` git endpoint, run the flattened step

list, and report back through HMAC-signed callbacks:

- ``POST /api/ci/runs/<id>/steps``       ``{step_id, conclusion, exit_code}``

- ``POST /api/ci/runs/<id>/conclusion``  ``{conclusion, error}``

This is the smallest stand-in for ``workers/oak-ci`` (Cloudflare Sandbox) that

lets a benchmark exercise the real hosted merge gate against a local server.

It binds 127.0.0.1 only, runs steps as local ``bash -c`` processes (no

sandbox: never point a real server or untrusted repository at it), and adds a

fixed, labeled ``added_seconds`` delay to every run so "CI duration" is a

controlled scenario parameter rather than host noise.

Every executed run is appended to a JSONL ledger (run id, commit, ref, tree

digest of the fetched workspace, conclusion, timings). The stale-branch land

lane uses the ledger as its independent record of which exact trees CI tested.

"""

from __future__ import annotations

import hashlib

import hmac

import json

import os

import shutil

import subprocess

import tempfile

import threading

import time

import urllib.request

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from pathlib import Path

from typing import Any, Callable, Optional

LOOPBACK_HOSTS = ("127.0.0.1", "localhost", "::1")

def sign(secret: str, body: bytes) -> str:

    return hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()

def verify(secret: str, body: bytes, header: Optional[str]) -> bool:

    expected = f"sha256={sign(secret, body)}"

    return hmac.compare_digest(expected.encode(), (header or "").encode())

def tree_digest(root: Path, exclude: tuple[str, ...] = (".git", ".oak")) -> str:

    """Content digest of a working tree: sorted (relpath, sha256) pairs.

    Excludes VCS metadata directories at the top level only, except that

    ``.oak/workflows`` is versioned content in Oak repositories and is kept.

    """

    entries: list[str] = []

    for dirpath, dirnames, filenames in os.walk(root):

        rel_dir = os.path.relpath(dirpath, root)

        if rel_dir == ".":

            dirnames[:] = sorted(d for d in dirnames if d not in exclude or d == ".oak")

        if rel_dir == ".oak":

            dirnames[:] = [d for d in dirnames if d == "workflows"]

            filenames = []

        dirnames.sort()

        for name in sorted(filenames):

            path = Path(dirpath) / name

            rel = os.path.relpath(path, root).replace(os.sep, "/")

            if rel.startswith(".oak/") and not rel.startswith(".oak/workflows/"):

                continue

            entries.append(f"{rel}\0{hashlib.sha256(path.read_bytes()).hexdigest()}")

    return hashlib.sha256("\n".join(entries).encode()).hexdigest()

class CiStub:

    def __init__(

        self,

        *,

        api_url: str,

        secret: str,

        ledger: Path,

        added_seconds: float = 0.0,

        host: str = "127.0.0.1",

        port: int = 0,

        workdir: Optional[Path] = None,

        git_bin: str = "git",

        post: Optional[Callable[[str, dict[str, Any]], bool]] = None,

    ) -> None:

        if host not in LOOPBACK_HOSTS:

            raise ValueError("ci_stub is local-only; refusing non-loopback bind")

        self.api_url = api_url.rstrip("/")

        self.secret = secret

        self.ledger = ledger

        self.added_seconds = float(added_seconds)

        self.workdir = workdir or Path(tempfile.mkdtemp(prefix="oak-ci-stub-"))

        self.git_bin = git_bin

        self._post = post or self._post_http

        self._lock = threading.Lock()

        self._threads: list[threading.Thread] = []

        stub = self

        class Handler(BaseHTTPRequestHandler):

            def log_message(self, *args: Any) -> None:  # quiet

                pass

            def _reply(self, code: int, payload: Any = None) -> None:

                body = json.dumps(payload if payload is not None else {}).encode()

                self.send_response(code)

                self.send_header("content-type", "application/json")

                self.send_header("content-length", str(len(body)))

                self.end_headers()

                self.wfile.write(body)

            def do_GET(self) -> None:  # noqa: N802

                if self.path.rstrip("/").endswith("/capabilities"):

                    # Unprotected executor: protected dispatch stays queued.

                    self._reply(200, {"protected_environment_v1": None})

                else:

                    self._reply(404, {"error": "not found"})

            def do_POST(self) -> None:  # noqa: N802

                length = int(self.headers.get("content-length") or 0)

                body = self.rfile.read(length)

                if not verify(stub.secret, body, self.headers.get("x-oak-signature-256")):

                    self._reply(401, {"error": "bad signature"})

                    return

                path = self.path.rstrip("/")

                if path.endswith("/jobs"):

                    try:

                        spec = json.loads(body)

                    except ValueError:

                        self._reply(400, {"error": "bad json"})

                        return

                    stub.accept(spec)

                    self._reply(202, {"accepted": True})

                elif path.endswith("/cancel"):

                    self._reply(200, {})

                else:

                    self._reply(404, {"error": "not found"})

        self.server = ThreadingHTTPServer((host, port), Handler)

        self.port = self.server.server_address[1]

        self._serve_thread: Optional[threading.Thread] = None

    # -- lifecycle -----------------------------------------------------------

    def start(self) -> "CiStub":

        self._serve_thread = threading.Thread(target=self.server.serve_forever, daemon=True)

        self._serve_thread.start()

        return self

    def stop(self) -> None:

        self.server.shutdown()

        self.server.server_close()

    def join_runs(self, timeout: float = 120.0) -> None:

        deadline = time.monotonic() + timeout

        for thread in list(self._threads):

            thread.join(max(0.0, deadline - time.monotonic()))

    # -- execution -----------------------------------------------------------

    def accept(self, spec: dict[str, Any]) -> None:

        thread = threading.Thread(target=self.execute, args=(spec,), daemon=True)

        with self._lock:

            self._threads.append(thread)

        thread.start()

    def _post_http(self, path: str, payload: dict[str, Any]) -> bool:

        body = json.dumps(payload).encode()

        request = urllib.request.Request(

            f"{self.api_url}{path}",

            data=body,

            method="POST",

            headers={

                "content-type": "application/json",

                "x-oak-signature-256": f"sha256={sign(self.secret, body)}",

            },

        )

        try:

            with urllib.request.urlopen(request, timeout=30) as response:

                return 200 <= response.status < 300

        except Exception:  # noqa: BLE001 - callbacks are best effort, like the real worker

            return False

    def fetch(self, spec: dict[str, Any], dest: Path) -> subprocess.CompletedProcess[str]:

        env = dict(os.environ)

        env.update(

            {

                "GIT_TERMINAL_PROMPT": "0",

                "GIT_CONFIG_COUNT": "1",

                "GIT_CONFIG_KEY_0": "http.extraHeader",

                "GIT_CONFIG_VALUE_0": f"x-oak-ci-fetch-token: {spec.get('fetch_token', '')}",

            }

        )

        url = f"{self.api_url}/api/ci/fetch/{spec['repo_id']}"

        return subprocess.run(

            [self.git_bin, "clone", "--quiet", "--depth", "1", "--no-tags", url, str(dest)],

            env=env,

            capture_output=True,

            text=True,

            timeout=120,

        )

    def execute(self, spec: dict[str, Any]) -> dict[str, Any]:

        run_id = int(spec["run_id"])

        started = time.time()

        record: dict[str, Any] = {

            "run_id": run_id,

            "repo_id": spec.get("repo_id"),

            "commit": spec.get("commit"),

            "ref": spec.get("ref"),

            "added_seconds": self.added_seconds,

            "started_unix": started,

            "tree_digest": None,

            "steps": [],

            "conclusion": None,

            "error": None,

        }

        workspace = self.workdir / f"run-{run_id}"

        shutil.rmtree(workspace, ignore_errors=True)

        conclusion = "success"

        error: Optional[str] = None

        try:

            if spec.get("fetch_token"):

                fetched = self.fetch(spec, workspace)

                if fetched.returncode != 0:

                    conclusion = "failure"

                    error = f"source fetch failed: {(fetched.stderr or fetched.stdout).strip()[:300]}"

            else:

                workspace.mkdir(parents=True)

            if conclusion == "success":

                record["tree_digest"] = tree_digest(workspace)

                env = dict(os.environ)

                env.update({str(k): str(v) for k, v in (spec.get("env") or {}).items()})

                env.update({str(k): str(v) for k, v in (spec.get("secrets") or {}).items()})

                env["OAK_CI_RUN_ID"] = str(run_id)

                env["OAK_CI_COMMIT"] = str(spec.get("commit") or "")

                for step in spec.get("steps") or []:

                    if conclusion != "success":

                        self._post(

                            f"/api/ci/runs/{run_id}/steps",

                            {"step_id": step["step_id"], "conclusion": "skipped", "exit_code": None},

                        )

                        continue

                    proc = subprocess.run(

                        ["bash", "-c", step["run"]],

                        cwd=workspace,

                        env=env,

                        capture_output=True,

                        text=True,

                        timeout=float(spec.get("max_seconds") or 600),

                    )

                    step_conclusion = "success" if proc.returncode == 0 else "failure"

                    record["steps"].append(

                        {"name": step.get("name"), "exit_code": proc.returncode, "tail": (proc.stdout + proc.stderr)[-400:]}

                    )

                    self._post(

                        f"/api/ci/runs/{run_id}/steps",

                        {"step_id": step["step_id"], "conclusion": step_conclusion, "exit_code": proc.returncode},

                    )

                    if proc.returncode != 0:

                        conclusion = "failure"

                        error = f"step '{step.get('name')}' exited {proc.returncode}"

            if self.added_seconds > 0:

                time.sleep(self.added_seconds)

        except Exception as exc:  # noqa: BLE001 - conclude rather than wedge the run

            conclusion = "failure"

            error = f"ci_stub error: {exc}"

        record["conclusion"] = conclusion

        record["error"] = error

        record["callback_ok"] = self._post(

            f"/api/ci/runs/{run_id}/conclusion", {"conclusion": conclusion, "error": error}

        )

        record["finished_unix"] = time.time()

        with self._lock:

            self.ledger.parent.mkdir(parents=True, exist_ok=True)

            with self.ledger.open("a") as handle:

                handle.write(json.dumps(record, sort_keys=True) + "\n")

        shutil.rmtree(workspace, ignore_errors=True)

        return record

def read_ledger(path: Path) -> list[dict[str, Any]]:

    if not path.exists():

        return []

    return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]

def main(argv: Optional[list[str]] = None) -> int:

    import argparse

    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])

    parser.add_argument("--api-url", required=True, help="local oakspace base URL, e.g. http://127.0.0.1:3811")

    parser.add_argument("--port", type=int, required=True)

    parser.add_argument("--secret-env", default="OAK_CI_SIGNING_SECRET")

    parser.add_argument("--ledger", type=Path, required=True)

    parser.add_argument("--added-seconds", type=float, default=0.0)

    args = parser.parse_args(argv)

    secret = os.environ.get(args.secret_env, "")

    if not secret:

        parser.error(f"${args.secret_env} is empty")

    stub = CiStub(api_url=args.api_url, secret=secret, ledger=args.ledger, added_seconds=args.added_seconds, port=args.port)

    stub.start()

    print(f"ci_stub listening on 127.0.0.1:{stub.port}", flush=True)

    try:

        while True:

            time.sleep(3600)

    except KeyboardInterrupt:

        stub.stop()

    return 0

if __name__ == "__main__":

    raise SystemExit(main())