tests/test_hosted_proof.py

from __future__ import annotations

import ast
import hashlib
import json
import re
import subprocess
import unittest
from collections.abc import Sequence
from contextlib import redirect_stderr, redirect_stdout
from datetime import UTC, datetime, timedelta
from io import StringIO
from pathlib import Path
from unittest.mock import patch

from lumi_eggcracker import hosted_proof as hosted_proof_module
from lumi_eggcracker.hosted_proof import (
    REVIEWED_WORKFLOW_BLOB,
    HostedProofError,
)
from lumi_eggcracker.hosted_proof import start_hosted_proof as _start_hosted_proof

ROOT = Path(__file__).resolve().parents[1]
WORKFLOW = ROOT / ".github" / "workflows" / "containment-probe.yml"
BOUND_PROBE_SOURCES = (
    Path("scripts/containment_probe.py"),
    Path("src/lumi_eggcracker/__init__.py"),
    Path("src/lumi_eggcracker/adoption.py"),
    Path("src/lumi_eggcracker/containment.py"),
    Path("src/lumi_eggcracker/containment_probe.py"),
    Path("src/lumi_eggcracker/discovery.py"),
    Path("src/lumi_eggcracker/jsonio.py"),
)
CORRELATION = "0123456789abcdef0123456789abcdef"
NOW = datetime(2026, 9, 1, 0, 0, tzinfo=UTC)
RUN_TITLE = f"Containment probe ({CORRELATION})"


def result(
    command: Sequence[str],
    *,
    returncode: int = 0,
    stdout: str = "",
) -> subprocess.CompletedProcess[str]:
    return subprocess.CompletedProcess(command, returncode, stdout=stdout, stderr="")


class FakeRunner:
    def __init__(
        self,
        responses: list[subprocess.CompletedProcess[str] | BaseException],
    ) -> None:
        self.responses = responses
        self.commands: list[tuple[str, ...]] = []

    def __call__(self, command: Sequence[str]) -> subprocess.CompletedProcess[str]:
        self.commands.append(tuple(command))
        if not self.responses:
            raise AssertionError(f"unexpected command: {command}")
        response = self.responses.pop(0)
        if isinstance(response, BaseException):
            raise response
        return response


def start_hosted_proof(
    *,
    acknowledged: bool,
    runner: FakeRunner,
    **overrides: object,
) -> str:
    options = {
        "sleeper": lambda _seconds: None,
        "monotonic": lambda: 0.0,
        "utc_now": lambda: NOW,
        "token_factory": lambda size: CORRELATION if size == 16 else "",
        **overrides,
    }
    return _start_hosted_proof(
        acknowledged=acknowledged,
        runner=runner,
        **options,  # type: ignore[arg-type]
    )


def workflow_run(
    run_id: int,
    *,
    repository: str = "operator/Lumi-Eggcracker",
    actor: str = "operator",
    branch: str = "main",
    event: str = "workflow_dispatch",
    title: str = RUN_TITLE,
    created_at: datetime = NOW,
) -> dict[str, object]:
    return {
        "id": run_id,
        "html_url": f"https://github.com/{repository}/actions/runs/{run_id}",
        "display_title": title,
        "event": event,
        "head_branch": branch,
        "created_at": created_at.isoformat().replace("+00:00", "Z"),
        "actor": {"login": actor},
        "repository": {"full_name": repository},
    }


def run_list(*runs: dict[str, object], returncode: int = 0) -> subprocess.CompletedProcess[str]:
    return result(
        ("run-list",),
        returncode=returncode,
        stdout=json.dumps({"workflow_runs": list(runs)}),
    )


def existing_fork_responses(
    *after_dispatch: subprocess.CompletedProcess[str] | BaseException,
    dispatch_stdout: str = "",
) -> list[subprocess.CompletedProcess[str] | BaseException]:
    return [
        result(("auth",)),
        result(("identity",), stdout="operator\n"),
        result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmain\n"),
        result(("workflow-identity",), stdout=f"{REVIEWED_WORKFLOW_BLOB}\n"),
        result(("enable",)),
        result(("dispatch",), stdout=dispatch_stdout),
        *after_dispatch,
    ]


class HostedProofTests(unittest.TestCase):
    def test_cli_prints_run_and_result_route(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
        output = StringIO()
        with (
            patch.object(hosted_proof_module.shutil, "which", return_value="gh"),
            patch.object(hosted_proof_module, "start_hosted_proof", return_value=url) as starter,
            redirect_stdout(output),
        ):
            status = hosted_proof_module.main(
                ["--i-understand-this-kills-a-test-tree"]
            )

        self.assertEqual(0, status)
        starter.assert_called_once_with(acknowledged=True, sync_fork=False)
        self.assertEqual(
            [
                f"Hosted proof dispatched: {url}",
                (
                    "Watch from this terminal: "
                    "gh run watch 123 --repo github.com/operator/Lumi-Eggcracker "
                    "--exit-status"
                ),
                (
                    "Read the bounded result: "
                    "gh run view 123 --repo github.com/operator/Lumi-Eggcracker --log"
                ),
                (
                    "After it finishes, share the public run or friction: "
                    f"{hosted_proof_module.RESULT_FORM_URL}"
                ),
            ],
            output.getvalue().splitlines(),
        )

    def test_cli_waits_for_exact_run_and_prints_concise_success(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
        output = StringIO()
        with (
            patch.object(hosted_proof_module.shutil, "which", return_value="gh"),
            patch.object(hosted_proof_module, "start_hosted_proof", return_value=url),
            patch.object(
                hosted_proof_module,
                "follow_hosted_proof",
                return_value=('{"result":"TERMINATED"}', True),
            ) as follower,
            redirect_stdout(output),
        ):
            status = hosted_proof_module.main(
                ["--i-understand-this-kills-a-test-tree", "--wait"]
            )

        self.assertEqual(0, status)
        follower.assert_called_once_with(url)
        self.assertEqual(
            [
                f"Hosted proof dispatched: {url}",
                "Hosted proof finished.",
                (
                    "After it finishes, share the public run or friction: "
                    f"{hosted_proof_module.RESULT_FORM_URL}"
                ),
                f"Hosted proof result: PASS ({url})",
            ],
            output.getvalue().splitlines(),
        )
        self.assertEqual(
            f"Hosted proof result: PASS ({url})",
            output.getvalue().splitlines()[-1],
        )

    def test_cli_wait_show_log_prints_bounded_success_log(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
        output = StringIO()
        with (
            patch.object(hosted_proof_module.shutil, "which", return_value="gh"),
            patch.object(hosted_proof_module, "start_hosted_proof", return_value=url),
            patch.object(
                hosted_proof_module,
                "follow_hosted_proof",
                return_value=('{"result":"TERMINATED"}', True),
            ),
            redirect_stdout(output),
        ):
            status = hosted_proof_module.main(
                [
                    "--i-understand-this-kills-a-test-tree",
                    "--wait",
                    "--show-log",
                ]
            )

        self.assertEqual(0, status)
        self.assertEqual(
            [
                f"Hosted proof dispatched: {url}",
                "Hosted proof finished.",
                "Bounded public workflow log:",
                '{"result":"TERMINATED"}',
                (
                    "After it finishes, share the public run or friction: "
                    f"{hosted_proof_module.RESULT_FORM_URL}"
                ),
                f"Hosted proof result: PASS ({url})",
            ],
            output.getvalue().splitlines(),
        )

    def test_cli_show_log_requires_wait(self) -> None:
        error = StringIO()
        with redirect_stderr(error), self.assertRaises(SystemExit) as raised:
            hosted_proof_module.main(["--show-log"])

        self.assertEqual(2, raised.exception.code)
        self.assertIn("--show-log requires --wait", error.getvalue())

    def test_cli_passes_explicit_sync_fork_permission(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
        with (
            patch.object(hosted_proof_module.shutil, "which", return_value="gh"),
            patch.object(
                hosted_proof_module,
                "start_hosted_proof",
                return_value=url,
            ) as starter,
            redirect_stdout(StringIO()),
        ):
            status = hosted_proof_module.main(
                ["--i-understand-this-kills-a-test-tree", "--sync-fork"]
            )

        self.assertEqual(0, status)
        starter.assert_called_once_with(acknowledged=True, sync_fork=True)

    def test_follow_exact_run_waits_then_reads_bounded_log(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
        runner = FakeRunner(
            [
                result(("watch",)),
                result(
                    ("status",),
                    stdout=json.dumps(
                        {"status": "completed", "conclusion": "success", "url": url}
                    ),
                ),
                result(("log",), stdout='{"result":"TERMINATED"}\n'),
            ]
        )

        self.assertEqual(
            ('{"result":"TERMINATED"}', True),
            hosted_proof_module.follow_hosted_proof(url, runner=runner),
        )
        self.assertEqual(
            [
                (
                    "gh",
                    "run",
                    "watch",
                    "123",
                    "--repo",
                    "github.com/operator/Lumi-Eggcracker",
                    "--exit-status",
                ),
                (
                    "gh",
                    "run",
                    "view",
                    "123",
                    "--repo",
                    "github.com/operator/Lumi-Eggcracker",
                    "--json",
                    "status,conclusion,url",
                ),
                (
                    "gh",
                    "run",
                    "view",
                    "123",
                    "--repo",
                    "github.com/operator/Lumi-Eggcracker",
                    "--log",
                ),
            ],
            runner.commands,
        )

    def test_follow_failed_run_still_returns_its_bounded_log(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
        runner = FakeRunner(
            [
                result(("watch",), returncode=1),
                result(
                    ("status",),
                    stdout=json.dumps(
                        {"status": "completed", "conclusion": "failure", "url": url}
                    ),
                ),
                result(("log",), stdout='{"result":"FAILED"}\n'),
            ]
        )

        self.assertEqual(
            ('{"result":"FAILED"}', False),
            hosted_proof_module.follow_hosted_proof(url, runner=runner),
        )

    def test_follow_watch_timeout_is_not_reported_as_completion(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
        runner = FakeRunner([subprocess.TimeoutExpired(cmd=("gh", "run", "watch"), timeout=900)])

        with self.assertRaisesRegex(HostedProofError, "could not complete.*wait"):
            hosted_proof_module.follow_hosted_proof(url, runner=runner)
        self.assertEqual(1, len(runner.commands))

    def test_follow_noncompleted_status_is_not_reported_as_finished(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
        runner = FakeRunner(
            [
                result(("watch",), returncode=1),
                result(
                    ("status",),
                    stdout=json.dumps(
                        {"status": "in_progress", "conclusion": "", "url": url}
                    ),
                ),
            ]
        )

        with self.assertRaisesRegex(HostedProofError, "did not reach a completed state"):
            hosted_proof_module.follow_hosted_proof(url, runner=runner)
        self.assertEqual(2, len(runner.commands))

    def test_follow_rejects_unavailable_status_and_log(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
        completed = json.dumps(
            {"status": "completed", "conclusion": "success", "url": url}
        )
        cases = (
            (
                [result(("watch",)), result(("status",), returncode=1)],
                "could not read the hosted-proof status",
            ),
            (
                [
                    result(("watch",)),
                    result(("status",), stdout=completed),
                    result(("log",), returncode=1),
                ],
                "log was unavailable",
            ),
            (
                [
                    result(("watch",)),
                    result(("status",), stdout=completed),
                    result(("log",), stdout=""),
                ],
                "log was empty",
            ),
        )
        for responses, message in cases:
            with self.subTest(message=message), self.assertRaisesRegex(
                HostedProofError, message
            ):
                hosted_proof_module.follow_hosted_proof(
                    url,
                    runner=FakeRunner(responses),
                )

    def test_follow_log_byte_bound_is_exact(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
        completed = json.dumps(
            {"status": "completed", "conclusion": "success", "url": url}
        )
        exact = "x" * hosted_proof_module.FOLLOW_LOG_MAX_BYTES
        runner = FakeRunner(
            [
                result(("watch",)),
                result(("status",), stdout=completed),
                result(("log",), stdout=exact),
            ]
        )
        self.assertEqual(
            (exact, True),
            hosted_proof_module.follow_hosted_proof(url, runner=runner),
        )

        oversized = FakeRunner(
            [
                result(("watch",)),
                result(("status",), stdout=completed),
                result(
                    ("log",),
                    stdout="x" * (hosted_proof_module.FOLLOW_LOG_MAX_BYTES + 1),
                ),
            ]
        )
        with self.assertRaisesRegex(HostedProofError, "safe display bound"):
            hosted_proof_module.follow_hosted_proof(url, runner=oversized)

    def test_cli_wait_returns_failure_after_printing_failed_log(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
        output = StringIO()
        with (
            patch.object(hosted_proof_module.shutil, "which", return_value="gh"),
            patch.object(hosted_proof_module, "start_hosted_proof", return_value=url),
            patch.object(
                hosted_proof_module,
                "follow_hosted_proof",
                return_value=('{"result":"FAILED"}', False),
            ),
            redirect_stdout(output),
        ):
            status = hosted_proof_module.main(
                ["--i-understand-this-kills-a-test-tree", "--wait"]
            )

        self.assertEqual(1, status)
        self.assertIn('{"result":"FAILED"}', output.getvalue().splitlines())
        self.assertIn("Bounded public workflow log:", output.getvalue().splitlines())
        self.assertIn(
            f"Hosted proof result: FAIL ({url})",
            output.getvalue().splitlines(),
        )
        self.assertIn(
            f"Hosted proof did not pass; inspect {url}.",
            output.getvalue().splitlines(),
        )
        self.assertEqual(
            f"Hosted proof result: FAIL ({url})",
            output.getvalue().splitlines()[-1],
        )

    def test_follow_refuses_non_exact_url_before_github_call(self) -> None:
        runner = FakeRunner([])
        with self.assertRaisesRegex(HostedProofError, "exact hosted-proof run URL"):
            hosted_proof_module.follow_hosted_proof(
                "https://github.com/operator/Lumi-Eggcracker/actions/workflows/"
                "containment-probe.yml",
                runner=runner,
            )
        self.assertEqual([], runner.commands)

    def test_watch_command_requires_exact_safe_github_run_url(self) -> None:
        self.assertEqual(
            (
                "gh run watch 123 --repo github.com/operator/Lumi-Eggcracker "
                "--exit-status"
            ),
            hosted_proof_module._watch_command(
                "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
            ),
        )
        self.assertEqual(
            "gh run view 123 --repo github.com/operator/Lumi-Eggcracker --log",
            hosted_proof_module._log_command(
                "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
            ),
        )
        rejected = (
            "http://github.com/operator/Lumi-Eggcracker/actions/runs/123",
            "https://gitlab.com/operator/Lumi-Eggcracker/actions/runs/123",
            "https://github.com/operator/other/actions/runs/123",
            "https://github.com/operator/Lumi-Eggcracker/actions/runs/0",
            "https://github.com/operator/Lumi-Eggcracker/actions/runs/123?x=1",
            "https://github.com/operator/Lumi-Eggcracker/actions/runs/123#fragment",
            "https://github.com/bad;name/Lumi-Eggcracker/actions/runs/123",
        )
        for url in rejected:
            with self.subTest(url=url):
                self.assertIsNone(hosted_proof_module._watch_command(url))
                self.assertIsNone(hosted_proof_module._log_command(url))

    def test_cli_omits_watch_command_for_fallback_workflow_page(self) -> None:
        url = (
            "https://github.com/operator/Lumi-Eggcracker/actions/workflows/"
            "containment-probe.yml"
        )
        output = StringIO()
        with (
            patch.object(hosted_proof_module.shutil, "which", return_value="gh"),
            patch.object(hosted_proof_module, "start_hosted_proof", return_value=url),
            redirect_stdout(output),
        ):
            status = hosted_proof_module.main(
                ["--i-understand-this-kills-a-test-tree"]
            )

        self.assertEqual(0, status)
        self.assertEqual(
            [
                f"Hosted proof dispatched: {url}",
                (
                    "After it finishes, share the public run or friction: "
                    f"{hosted_proof_module.RESULT_FORM_URL}"
                ),
            ],
            output.getvalue().splitlines(),
        )

    def test_cli_wait_fallback_does_not_guess_a_run(self) -> None:
        url = (
            "https://github.com/operator/Lumi-Eggcracker/actions/workflows/"
            "containment-probe.yml"
        )
        output = StringIO()
        with (
            patch.object(hosted_proof_module.shutil, "which", return_value="gh"),
            patch.object(hosted_proof_module, "start_hosted_proof", return_value=url),
            patch.object(hosted_proof_module, "follow_hosted_proof") as follower,
            redirect_stdout(output),
        ):
            status = hosted_proof_module.main(
                ["--i-understand-this-kills-a-test-tree", "--wait"]
            )

        self.assertEqual(1, status)
        follower.assert_not_called()
        self.assertIn(
            "Automatic wait unavailable because the exact run URL was not resolved.",
            output.getvalue().splitlines(),
        )

    def test_reviewed_workflow_blob_matches_current_workflow(self) -> None:
        raw = WORKFLOW.read_bytes()
        framed = f"blob {len(raw)}\0".encode("ascii") + raw
        self.assertEqual(
            hashlib.sha1(framed, usedforsecurity=False).hexdigest(),
            REVIEWED_WORKFLOW_BLOB,
        )

    def test_workflow_qualified_digest_matches_current_probe_sources(self) -> None:
        workflow = WORKFLOW.read_text(encoding="utf-8")
        expected = re.findall(r"QUALIFIED_SOURCE_SHA256: ([0-9a-f]{64})", workflow)
        self.assertEqual(1, len(expected))

        digest = hashlib.sha256()
        for relative in BOUND_PROBE_SOURCES:
            raw = (ROOT / relative).read_bytes()
            name = relative.as_posix().encode("ascii")
            digest.update(len(name).to_bytes(4, "big"))
            digest.update(name)
            digest.update(len(raw).to_bytes(8, "big"))
            digest.update(raw)
        self.assertEqual(expected[0], digest.hexdigest())

    def test_qualified_source_set_covers_local_runtime_imports(self) -> None:
        bound_modules = {
            "lumi_eggcracker"
            if relative.name == "__init__.py"
            else ".".join(relative.with_suffix("").parts[1:])
            for relative in BOUND_PROBE_SOURCES
            if relative.parts[0] == "src"
        }

        imported_modules: set[str] = set()
        for relative in BOUND_PROBE_SOURCES:
            tree = ast.parse((ROOT / relative).read_text(encoding="utf-8"))
            for node in ast.walk(tree):
                if isinstance(node, ast.Import):
                    imported_modules.update(alias.name for alias in node.names)
                elif isinstance(node, ast.ImportFrom):
                    if node.level and node.module:
                        imported_modules.add(f"lumi_eggcracker.{node.module}")
                    elif node.level:
                        imported_modules.update(
                            f"lumi_eggcracker.{alias.name}" for alias in node.names
                        )
                    elif node.module:
                        imported_modules.add(node.module)

        local_imports = {
            module
            for module in imported_modules
            if module == "lumi_eggcracker" or module.startswith("lumi_eggcracker.")
        }
        self.assertLessEqual(local_imports, bound_modules)

    def test_workflow_does_not_pin_unrelated_source_trees(self) -> None:
        workflow = WORKFLOW.read_text(encoding="utf-8")
        self.assertNotIn("QUALIFIED_SRC_TREE", workflow)
        self.assertNotIn("QUALIFIED_SCRIPTS_TREE", workflow)

    def test_workflow_correlation_is_display_only_with_manual_browser_default(self) -> None:
        workflow = WORKFLOW.read_text(encoding="utf-8")
        header, jobs = workflow.split("\njobs:\n", 1)
        self.assertIn(
            'run-name: "Containment probe (${{ inputs.run_correlation_id || \'manual\' }})"',
            header,
        )
        self.assertRegex(
            header,
            r"run_correlation_id:\n(?: {8}.+\n){3} {8}default: manual",
        )
        self.assertNotIn("run_correlation_id", jobs)

    def test_workflow_rejects_import_shadowing_before_path_insertion(self) -> None:
        workflow = WORKFLOW.read_text(encoding="utf-8")
        guard = workflow.index("refuse SOURCE_IMPORT_PATH_UNQUALIFIED")
        path_insertion = workflow.index('sys.path.insert(0, str(Path.cwd() / "src"))')
        self.assertLess(guard, path_insertion)
        self.assertIn("len(entries) != 1", workflow)
        self.assertIn("entries[0] != expected", workflow)
        self.assertIn("entries[0].is_symlink()", workflow)
        self.assertIn("not entries[0].is_dir()", workflow)

    def test_acknowledgement_is_required_before_any_github_call(self) -> None:
        runner = FakeRunner([])
        with self.assertRaisesRegex(HostedProofError, "acknowledgement"):
            start_hosted_proof(acknowledged=False, runner=runner)
        self.assertEqual([], runner.commands)

    def test_existing_expected_fork_is_enabled_and_dispatched(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/123"
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmain\n"),
                result(("workflow-identity",), stdout=f"{REVIEWED_WORKFLOW_BLOB}\n"),
                result(("enable",)),
                result(("dispatch",), stdout=f"{url}\n"),
            ]
        )

        self.assertEqual(url, start_hosted_proof(acknowledged=True, runner=runner))
        self.assertEqual(
            (
                "gh",
                "workflow",
                "run",
                "containment-probe.yml",
                "--repo",
                "github.com/operator/Lumi-Eggcracker",
                "--ref",
                "main",
                "--raw-field",
                "i_understand_this_kills_a_test_tree=true",
                "--raw-field",
                f"run_correlation_id={CORRELATION}",
            ),
            runner.commands[-1],
        )

    def test_dispatch_time_is_captured_immediately_before_dispatch(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/124"
        runner = FakeRunner(existing_fork_responses(dispatch_stdout=f"{url}\n"))
        observations: list[tuple[str, ...]] = []

        def utc_now() -> datetime:
            observations.append(runner.commands[-1])
            return NOW

        self.assertEqual(
            url,
            start_hosted_proof(
                acknowledged=True,
                runner=runner,
                utc_now=utc_now,
            ),
        )
        self.assertEqual(1, len(observations))
        self.assertEqual(("workflow", "enable"), observations[0][1:3])

    def test_missing_fork_is_created_without_a_clone(self) -> None:
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), returncode=1),
                result(("fork",)),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\ttrunk\n"),
                result(("workflow-identity",), stdout=f"{REVIEWED_WORKFLOW_BLOB}\n"),
                result(("enable",)),
                result(("dispatch",)),
                run_list(returncode=1),
            ]
        )

        url = start_hosted_proof(acknowledged=True, runner=runner)
        self.assertEqual(
            (
                "gh",
                "repo",
                "fork",
                "github.com/noqt/Lumi-Eggcracker",
                "--clone=false",
                "--default-branch-only",
            ),
            runner.commands[3],
        )
        self.assertEqual(
            "https://github.com/operator/Lumi-Eggcracker/actions/workflows/containment-probe.yml",
            url,
        )

    def test_new_fork_metadata_propagation_is_retried_before_dispatch(self) -> None:
        sleeps: list[float] = []
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), returncode=1),
                result(("fork",)),
                result(("metadata",), returncode=1),
                result(("metadata",), returncode=1),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmain\n"),
                result(("workflow-identity",), stdout=f"{REVIEWED_WORKFLOW_BLOB}\n"),
                result(("enable",)),
                result(("dispatch",)),
                run_list(returncode=1),
            ]
        )

        url = start_hosted_proof(
            acknowledged=True,
            runner=runner,
            sleeper=sleeps.append,
        )

        self.assertEqual(
            "https://github.com/operator/Lumi-Eggcracker/actions/workflows/containment-probe.yml",
            url,
        )
        self.assertEqual([1.0, 1.0], sleeps)

    def test_new_fork_metadata_readiness_wait_is_bounded(self) -> None:
        sleeps: list[float] = []
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), returncode=1),
                result(("fork",)),
                *[result(("metadata",), returncode=1) for _ in range(5)],
            ]
        )

        with self.assertRaisesRegex(HostedProofError, "not ready"):
            start_hosted_proof(
                acknowledged=True,
                runner=runner,
                sleeper=sleeps.append,
            )

        self.assertEqual([1.0, 1.0, 1.0, 1.0], sleeps)

    def test_unrelated_repository_collision_is_refused(self) -> None:
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), stdout="false\t\tmain\n"),
            ]
        )

        with self.assertRaisesRegex(HostedProofError, "not a fork"):
            start_hosted_proof(acknowledged=True, runner=runner)
        self.assertEqual(3, len(runner.commands))

    def test_diverged_fork_workflow_is_refused_before_enable_or_dispatch(self) -> None:
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmodified-default\n"),
                result(("workflow-identity",), stdout="0123456789abcdef0123456789abcdef01234567\n"),
            ]
        )

        with self.assertRaisesRegex(HostedProofError, "--sync-fork"):
            start_hosted_proof(acknowledged=True, runner=runner)
        self.assertEqual(4, len(runner.commands))
        self.assertEqual("api", runner.commands[-1][1])

    def test_explicit_sync_fast_forwards_then_revalidates_before_dispatch(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/125"
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmain\n"),
                result(("workflow-identity",), stdout="0" * 40 + "\n"),
                result(("sync",)),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmain\n"),
                result(("workflow-identity",), stdout=f"{REVIEWED_WORKFLOW_BLOB}\n"),
                result(("enable",)),
                result(("dispatch",), stdout=f"{url}\n"),
            ]
        )

        self.assertEqual(
            url,
            start_hosted_proof(acknowledged=True, sync_fork=True, runner=runner),
        )
        self.assertIn(
            (
                "gh",
                "repo",
                "sync",
                "github.com/operator/Lumi-Eggcracker",
                "--source",
                "github.com/noqt/Lumi-Eggcracker",
                "--branch",
                "main",
            ),
            runner.commands,
        )
        self.assertNotIn(
            "--force",
            [argument for command in runner.commands for argument in command],
        )
        self.assertEqual("workflow", runner.commands[-2][1])
        self.assertEqual("workflow", runner.commands[-1][1])

    def test_failed_fast_forward_never_dispatches(self) -> None:
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmain\n"),
                result(("workflow-identity",), stdout="0" * 40 + "\n"),
                result(("sync",), returncode=1),
            ]
        )

        with self.assertRaisesRegex(HostedProofError, "could not fast-forward"):
            start_hosted_proof(acknowledged=True, sync_fork=True, runner=runner)
        self.assertEqual("sync", runner.commands[-1][2])
        self.assertNotIn(
            "workflow",
            [argument for command in runner.commands for argument in command],
        )

    def test_post_sync_fork_identity_drift_never_dispatches(self) -> None:
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmain\n"),
                result(("workflow-identity",), stdout="0" * 40 + "\n"),
                result(("sync",)),
                result(("metadata",), stdout="false\t\tmain\n"),
            ]
        )

        with self.assertRaisesRegex(HostedProofError, "no longer matches"):
            start_hosted_proof(acknowledged=True, sync_fork=True, runner=runner)
        self.assertEqual(6, len(runner.commands))

    def test_post_sync_metadata_read_failure_never_dispatches(self) -> None:
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmain\n"),
                result(("workflow-identity",), stdout="0" * 40 + "\n"),
                result(("sync",)),
                result(("metadata",), returncode=1),
            ]
        )

        with self.assertRaisesRegex(HostedProofError, "could not revalidate"):
            start_hosted_proof(acknowledged=True, sync_fork=True, runner=runner)
        self.assertEqual(6, len(runner.commands))
        self.assertNotIn(
            "workflow",
            [argument for command in runner.commands for argument in command],
        )

    def test_post_sync_default_branch_drift_never_dispatches(self) -> None:
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmain\n"),
                result(("workflow-identity",), stdout="0" * 40 + "\n"),
                result(("sync",)),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\ttrunk\n"),
            ]
        )

        with self.assertRaisesRegex(HostedProofError, "changed its default branch"):
            start_hosted_proof(acknowledged=True, sync_fork=True, runner=runner)
        self.assertEqual(6, len(runner.commands))
        self.assertNotIn(
            "workflow",
            [argument for command in runner.commands for argument in command],
        )

    def test_post_sync_workflow_mismatch_never_dispatches(self) -> None:
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmain\n"),
                result(("workflow-identity",), stdout="0" * 40 + "\n"),
                result(("sync",)),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmain\n"),
                result(("workflow-identity",), stdout="1" * 40 + "\n"),
            ]
        )

        with self.assertRaisesRegex(HostedProofError, "still does not contain"):
            start_hosted_proof(acknowledged=True, sync_fork=True, runner=runner)
        self.assertEqual(7, len(runner.commands))

    def test_exotic_branch_is_encoded_for_verification_and_reused_for_dispatch(self) -> None:
        branch = "release&proof#1"
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(
                    ("metadata",),
                    stdout=f"true\tnoqt/Lumi-Eggcracker\t{branch}\n",
                ),
                result(
                    ("workflow-identity",),
                    stdout=f"{REVIEWED_WORKFLOW_BLOB}\n",
                ),
                result(("enable",)),
                result(("dispatch",)),
                run_list(returncode=1),
            ]
        )

        start_hosted_proof(acknowledged=True, runner=runner)

        self.assertEqual(
            (
                "gh",
                "api",
                "--hostname",
                "github.com",
                "--method",
                "GET",
                "repos/operator/Lumi-Eggcracker/contents/.github/workflows/containment-probe.yml",
                "--raw-field",
                f"ref={branch}",
                "--jq",
                ".sha",
            ),
            runner.commands[3],
        )
        dispatch = next(command for command in runner.commands if command[1:3] == ("workflow", "run"))
        self.assertEqual(branch, dispatch[7])

    def test_already_active_workflow_is_safe_after_enable_error(self) -> None:
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmain\n"),
                result(("workflow-identity",), stdout=f"{REVIEWED_WORKFLOW_BLOB}\n"),
                result(("enable",), returncode=1),
                result(("state",), stdout="active\n"),
                result(("dispatch",)),
                run_list(returncode=1),
            ]
        )

        start_hosted_proof(acknowledged=True, runner=runner)
        self.assertEqual("gh", runner.commands[-1][0])

    def test_unique_correlated_run_is_returned_with_exact_query_scope(self) -> None:
        url = "https://github.com/operator/Lumi-Eggcracker/actions/runs/321"
        runner = FakeRunner(existing_fork_responses(run_list(workflow_run(321))))

        self.assertEqual(url, start_hosted_proof(acknowledged=True, runner=runner))
        self.assertEqual(
            (
                "gh",
                "api",
                "--hostname",
                "github.com",
                "--method",
                "GET",
                "repos/operator/Lumi-Eggcracker/actions/workflows/containment-probe.yml/runs",
                "--raw-field",
                "actor=operator",
                "--raw-field",
                "branch=main",
                "--raw-field",
                "event=workflow_dispatch",
                "--raw-field",
                "created=>=2026-08-31T23:55:00Z",
                "--raw-field",
                "per_page=20",
            ),
            runner.commands[-1],
        )

    def test_wrong_direct_url_is_ignored_before_unique_correlation(self) -> None:
        expected = "https://github.com/operator/Lumi-Eggcracker/actions/runs/322"
        wrong = "https://github.com/other/Lumi-Eggcracker/actions/runs/999\n"
        runner = FakeRunner(
            existing_fork_responses(run_list(workflow_run(322)), dispatch_stdout=wrong)
        )

        self.assertEqual(expected, start_hosted_proof(acknowledged=True, runner=runner))

    def test_zero_run_id_from_dispatch_is_not_accepted(self) -> None:
        zero = "https://github.com/operator/Lumi-Eggcracker/actions/runs/0\n"
        runner = FakeRunner(existing_fork_responses(run_list(returncode=1), dispatch_stdout=zero))

        self.assertEqual(
            "https://github.com/operator/Lumi-Eggcracker/actions/workflows/"
            "containment-probe.yml",
            start_hosted_proof(acknowledged=True, runner=runner),
        )

    def test_correlation_must_be_128_bit_lowercase_hex(self) -> None:
        runner = FakeRunner([])
        with self.assertRaisesRegex(HostedProofError, "correlation"):
            start_hosted_proof(
                acknowledged=True,
                runner=runner,
                token_factory=lambda _size: "NOT-SAFE",
            )
        self.assertEqual([], runner.commands)

    def test_zero_correlated_runs_retries_then_returns_workflow_page(self) -> None:
        runner = FakeRunner(existing_fork_responses(run_list(), run_list(), run_list()))

        self.assertEqual(
            "https://github.com/operator/Lumi-Eggcracker/actions/workflows/"
            "containment-probe.yml",
            start_hosted_proof(acknowledged=True, runner=runner),
        )
        self.assertEqual(
            3,
            sum(
                command[1] == "api"
                and len(command) > 6
                and command[6].endswith("/actions/workflows/containment-probe.yml/runs")
                for command in runner.commands
            ),
        )

    def test_multiple_exact_matches_return_workflow_page_without_guessing(self) -> None:
        runner = FakeRunner(
            existing_fork_responses(run_list(workflow_run(323), workflow_run(324)))
        )

        self.assertEqual(
            "https://github.com/operator/Lumi-Eggcracker/actions/workflows/"
            "containment-probe.yml",
            start_hosted_proof(acknowledged=True, runner=runner),
        )

    def test_concurrent_unrelated_run_does_not_hide_unique_exact_match(self) -> None:
        runner = FakeRunner(
            existing_fork_responses(
                run_list(
                    workflow_run(325, title="Containment probe (another-caller)"),
                    workflow_run(326),
                )
            )
        )

        self.assertEqual(
            "https://github.com/operator/Lumi-Eggcracker/actions/runs/326",
            start_hosted_proof(acknowledged=True, runner=runner),
        )

    def test_stale_and_implausibly_future_titles_return_workflow_page(self) -> None:
        stale = workflow_run(327, created_at=NOW - timedelta(minutes=6))
        future = workflow_run(328, created_at=NOW + timedelta(minutes=2))
        runner = FakeRunner(
            existing_fork_responses(
                run_list(stale, future),
                run_list(stale, future),
                run_list(stale, future),
            )
        )

        self.assertEqual(
            "https://github.com/operator/Lumi-Eggcracker/actions/workflows/"
            "containment-probe.yml",
            start_hosted_proof(acknowledged=True, runner=runner),
        )

    def test_wrong_scope_entries_return_workflow_page(self) -> None:
        wrong_repository = workflow_run(328, repository="other/Lumi-Eggcracker")
        wrong_actor = workflow_run(329, actor="other")
        wrong_branch = workflow_run(330, branch="other")
        wrong_event = workflow_run(331, event="push")
        wrong_url = workflow_run(332)
        wrong_url["html_url"] = "https://github.com/operator/other/actions/runs/332"
        entries = (wrong_repository, wrong_actor, wrong_branch, wrong_event, wrong_url)
        runner = FakeRunner(
            existing_fork_responses(run_list(*entries), run_list(*entries), run_list(*entries))
        )

        self.assertEqual(
            "https://github.com/operator/Lumi-Eggcracker/actions/workflows/"
            "containment-probe.yml",
            start_hosted_proof(acknowledged=True, runner=runner),
        )

    def test_malformed_or_failed_listing_returns_workflow_page(self) -> None:
        for response in (
            result(("malformed",), stdout="{"),
            run_list(returncode=1),
            subprocess.TimeoutExpired(cmd=("gh", "api"), timeout=5),
            UnicodeDecodeError("utf-8", b"\xff", 0, 1, "invalid byte"),
        ):
            with self.subTest(response=type(response).__name__):
                runner = FakeRunner(existing_fork_responses(response))
                self.assertEqual(
                    "https://github.com/operator/Lumi-Eggcracker/actions/workflows/"
                    "containment-probe.yml",
                    start_hosted_proof(acknowledged=True, runner=runner),
                )

    def test_recursive_json_failure_returns_workflow_page(self) -> None:
        runner = FakeRunner(existing_fork_responses(result(("listing",), stdout="{}")))

        with patch(
            "lumi_eggcracker.hosted_proof.json.loads",
            side_effect=RecursionError("synthetic nested JSON"),
        ):
            self.assertEqual(
                "https://github.com/operator/Lumi-Eggcracker/actions/workflows/"
                "containment-probe.yml",
                start_hosted_proof(acknowledged=True, runner=runner),
            )

    def test_late_unique_lookup_result_is_not_accepted(self) -> None:
        ticks = iter((0.0, 0.0, 17.1))
        runner = FakeRunner(existing_fork_responses(run_list(workflow_run(333))))

        self.assertEqual(
            "https://github.com/operator/Lumi-Eggcracker/actions/workflows/"
            "containment-probe.yml",
            start_hosted_proof(
                acknowledged=True,
                runner=runner,
                monotonic=lambda: next(ticks),
            ),
        )

    def test_overall_lookup_deadline_bounds_calls_and_sleep(self) -> None:
        ticks = iter((0.0, 0.0, 0.0, 16.5, 17.0))
        delays: list[float] = []
        runner = FakeRunner(existing_fork_responses(run_list()))

        self.assertEqual(
            "https://github.com/operator/Lumi-Eggcracker/actions/workflows/"
            "containment-probe.yml",
            start_hosted_proof(
                acknowledged=True,
                runner=runner,
                monotonic=lambda: next(ticks),
                sleeper=delays.append,
            ),
        )
        self.assertEqual([0.5], delays)
        self.assertEqual(1, sum("/actions/workflows/" in " ".join(command) for command in runner.commands))

    def test_disabled_workflow_after_enable_error_is_refused(self) -> None:
        runner = FakeRunner(
            [
                result(("auth",)),
                result(("identity",), stdout="operator\n"),
                result(("metadata",), stdout="true\tnoqt/Lumi-Eggcracker\tmain\n"),
                result(("workflow-identity",), stdout=f"{REVIEWED_WORKFLOW_BLOB}\n"),
                result(("enable",), returncode=1),
                result(("state",), stdout="disabled_manually\n"),
            ]
        )

        with self.assertRaisesRegex(HostedProofError, "could not enable"):
            start_hosted_proof(acknowledged=True, runner=runner)


if __name__ == "__main__":
    unittest.main()