from __future__ import annotations

import json
import importlib.util
import re
import multiprocessing
import os
import random
import shutil
import subprocess
import sys
from collections import Counter
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import remote_dispatch


REGISTRY = {
    "schema_version": 1,
    "hosts": [
        {
            "name": "debian1",
            "state": "reachable",
            "roles": ["builder", "agent-seat", "agent-sandbox"],
            "access": {
                "tailscale_ip": {
                    "host": "100.0.0.1",
                    "port": 2222,
                    "user": "user",
                    "identity_file": "~/.ssh/id_ed25519_buildbox",
                }
            },
        },
        {
            "name": "debian2",
            "state": "reachable",
            "roles": ["builder", "agent-seat", "agent-sandbox"],
            "access": {"tailscale_ip": {"host": "100.0.0.2", "port": 2222, "user": "user"}},
        },
        {
            "name": "debian3",
            "state": "unreachable",
            "roles": ["builder", "agent-seat", "agent-sandbox"],
            "access": {"tailscale_ip": {"host": "100.0.0.3", "port": 2222, "user": "user"}},
        },
    ],
}


THREE_REACHABLE = {
    "schema_version": 1,
    "hosts": [
        {
            "name": f"debian{n}",
            "state": "reachable",
            "roles": ["agent-seat", "agent-sandbox"],
            "access": {"tailscale_ip": {"host": f"100.0.0.{n}", "user": "user"}},
        }
        for n in (1, 2, 3)
    ],
}


@pytest.fixture(autouse=True)
def isolated_claim_dir(tmp_path_factory, monkeypatch):
    """No test may write placement claims into the real ~/.local/state directory."""
    directory = tmp_path_factory.mktemp("claims")
    monkeypatch.setenv(remote_dispatch.CLAIM_DIR_ENV, str(directory))
    return directory


def write_registry(tmp_path: Path, payload: dict | None = None) -> Path:
    path = tmp_path / "buildbox-hosts.json"
    path.write_text(json.dumps(payload if payload is not None else REGISTRY), encoding="utf-8")
    return path


def test_should_offload_true_when_not_in_container(monkeypatch):
    monkeypatch.setattr(remote_dispatch, "in_container", lambda: False)
    assert remote_dispatch.should_offload() is True


def test_should_offload_false_when_in_container(monkeypatch):
    monkeypatch.setattr(remote_dispatch, "in_container", lambda: True)
    assert remote_dispatch.should_offload() is False


def test_should_offload_ignores_env(monkeypatch):
    monkeypatch.setenv("CDX_NO_OFFLOAD", "1")
    monkeypatch.setenv("OD_REMOTE_EXEC", "1")
    monkeypatch.setenv("HARNESS_SEAT_CONTAINER", "1")
    monkeypatch.setenv("CDX_OFFLOAD_STRICT", "0")

    monkeypatch.setattr(remote_dispatch, "in_container", lambda: False)
    assert remote_dispatch.should_offload() is True

    monkeypatch.setattr(remote_dispatch, "in_container", lambda: True)
    assert remote_dispatch.should_offload() is False


def test_access_for_without_a_door_is_a_no_node_error():
    host = {
        "name": "debian1",
        "access": {},
    }
    with pytest.raises(remote_dispatch.OffloadUnavailable) as exc_info:
        remote_dispatch.access_for(host)
    assert exc_info.value.code == remote_dispatch.EXIT_NO_NODE


def test_candidate_hosts_excludes_unreachable():
    names = [h["name"] for h in remote_dispatch.candidate_hosts(REGISTRY)]
    assert names == ["debian1", "debian2"]


def test_candidate_hosts_raises_when_none_reachable():
    payload = {"hosts": [dict(REGISTRY["hosts"][2])]}
    with pytest.raises(remote_dispatch.OffloadUnavailable) as exc_info:
        remote_dispatch.candidate_hosts(payload)
    assert exc_info.value.code == remote_dispatch.EXIT_NO_NODE


def test_select_host_picks_least_loaded():
    loads = {"100.0.0.1": 9.5, "100.0.0.2": 0.4}
    host, access, _claim = remote_dispatch.select_host(
        remote_dispatch.candidate_hosts(REGISTRY),
        probe=lambda a: loads[a["host"]],
    )
    assert host["name"] == "debian2"
    assert access["host"] == "100.0.0.2"


def test_select_host_skips_nodes_that_do_not_answer():
    host, _access, _claim = remote_dispatch.select_host(
        remote_dispatch.candidate_hosts(REGISTRY),
        probe=lambda a: None if a["host"] == "100.0.0.2" else 7.0,
    )
    assert host["name"] == "debian1"


def test_select_host_raises_when_no_node_answers():
    with pytest.raises(remote_dispatch.OffloadUnavailable):
        remote_dispatch.select_host(remote_dispatch.candidate_hosts(REGISTRY), probe=lambda a: None)


def test_parse_loadavg():
    assert remote_dispatch.parse_loadavg("0.52 0.31 0.20 1/900 12345\n") == 0.52
    assert remote_dispatch.parse_loadavg("") is None
    assert remote_dispatch.parse_loadavg("garbage") is None


def test_remote_rel_dir_is_stable_and_path_specific():
    a = remote_dispatch.remote_rel_dir(Path("/home/user/Projects/overdeck/.worktrees/one"))
    b = remote_dispatch.remote_rel_dir(Path("/home/user/Projects/overdeck/.worktrees/two"))
    assert a == remote_dispatch.remote_rel_dir(Path("/home/user/Projects/overdeck/.worktrees/one"))
    assert a != b
    assert a.startswith("sandbox/workspaces/one-")


def test_the_workspace_root_is_the_one_every_party_resolves():
    """dispatch, sandbox-run, receiver.mjs and the in-image e2e-remote must name the same
    directory, or a dispatched run has no E2E channel and no reachable workspace."""
    repo = Path(__file__).resolve().parents[3]
    sandbox_run = (repo / "modules/sandbox/host/bin/sandbox-run").read_text(encoding="utf-8")
    receiver = (repo / "modules/sandbox/host/lib/receiver.mjs").read_text(encoding="utf-8")
    image_e2e = (repo / "modules/sandbox/image/bin/e2e-remote").read_text(encoding="utf-8")

    assert 'SANDBOX_ROOT="${SANDBOX_ROOT:-$HOME/sandbox}"' in sandbox_run
    assert 'WORKSPACE="$SANDBOX_ROOT/workspaces/$ID"' in sandbox_run
    assert '--volume "$SANDBOX_ROOT:/sandbox:rw"' in sandbox_run
    assert 'join(SANDBOX_ROOT, "workspaces")' in receiver
    assert 'WORKSPACES_ROOT="${SANDBOX_WORKSPACES_ROOT:-/sandbox/workspaces}"' in image_e2e

    root = Path("/home/user/Projects/overdeck/.worktrees/one")
    assert remote_dispatch.NODE_WORKSPACES_ROOT == "sandbox/workspaces"
    assert remote_dispatch.CONTAINER_WORKSPACES_ROOT == "/sandbox/workspaces"
    assert remote_dispatch.remote_rel_dir(root) == f"sandbox/workspaces/{remote_dispatch.sandbox_id(root)}"
    assert remote_dispatch.container_workspace(root) == (
        f"/sandbox/workspaces/{remote_dispatch.sandbox_id(root)}"
    )


def test_ssh_argv_pins_identity_port_and_ignores_ssh_config():
    argv = remote_dispatch.ssh_argv(REGISTRY["hosts"][0]["access"]["tailscale_ip"])
    assert argv[:3] == ["ssh", "-F", "/dev/null"]
    assert "-p" in argv and "2222" in argv
    assert argv[-1] == "user@100.0.0.1"
    assert any(a.endswith("id_ed25519_buildbox") for a in argv)


GIT_ENV = {
    "GIT_AUTHOR_NAME": "t",
    "GIT_AUTHOR_EMAIL": "t@example.invalid",
    "GIT_COMMITTER_NAME": "t",
    "GIT_COMMITTER_EMAIL": "t@example.invalid",
}


def git(*args: str, cwd: Path) -> subprocess.CompletedProcess:
    done = subprocess.run(
        ["/usr/bin/git", *args],
        cwd=cwd,
        capture_output=True,
        text=True,
        env={**os.environ, **GIT_ENV, "GIT_CONFIG_GLOBAL": "/dev/null", "GIT_CONFIG_SYSTEM": "/dev/null"},
    )
    assert done.returncode == 0, done.stderr
    return done


def make_repo(tmp_path: Path) -> Path:
    """A real checkout — a hand-written `.git` fixture is what let this bug ship."""
    root = tmp_path / "repo"
    root.mkdir()
    git("init", "-q", "-b", "main", ".", cwd=root)
    (root / "file.txt").write_text("one\n", encoding="utf-8")
    git("add", "file.txt", cwd=root)
    git("commit", "-qm", "init", cwd=root)
    return root


def add_worktree(root: Path, slug: str) -> Path:
    path = root / ".worktrees" / slug
    git("worktree", "add", "-q", "-b", f"wt/{slug}", str(path), cwd=root)
    return path


def make_credential(path: Path, runtime: str = "codex", slug: str = "acct") -> remote_dispatch.Credential:
    path.write_text("token", encoding="utf-8")
    return remote_dispatch.Credential(runtime=runtime, slug=slug, path=path)


def _strip_ssh_url(url: str) -> str:
    return "/" + url.split("://", 1)[1].split("/", 1)[1]


def real_node_fake_run(node_home: Path, calls: list[list[str]] | None = None):
    """Simulate the node as a real local `$HOME` and the bare mirror as a real local git
    repo: an `ssh` argv runs its trailing command through `bash -c` with `HOME` and `cwd`
    set to `node_home` (matching a real ssh session's default cwd), and any `ssh://` remote
    in a `git push`/`fetch` argv is rewritten to the local path it names. This runs
    `open_session`'s git-push transport for real, end to end, with no network."""
    node_home.mkdir(parents=True, exist_ok=True)

    def fake_run(argv, **kwargs):
        if calls is not None:
            calls.append(argv)
        if argv[:1] == ["ssh"]:
            return subprocess.run(
                ["bash", "-c", argv[-1]],
                capture_output=True,
                text=True,
                cwd=str(node_home),
                env={**os.environ, "HOME": str(node_home)},
                timeout=kwargs.get("timeout"),
            )
        if argv[:1] == ["/usr/bin/git"]:
            rewritten = [_strip_ssh_url(tok) if tok.startswith("ssh://") else tok for tok in argv]
            return subprocess.run(
                rewritten,
                capture_output=True,
                text=True,
                env=kwargs.get("env"),
                input=kwargs.get("input"),
                timeout=kwargs.get("timeout"),
            )
        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")

    return fake_run


def test_push_and_pull_are_mirror_images_with_identical_excludes():
    access = REGISTRY["hosts"][1]["access"]["tailscale_ip"]
    root = Path("/home/user/Projects/overdeck")
    rel = remote_dispatch.remote_rel_dir(root)
    push = remote_dispatch.push_argv(access, root, rel)
    pull = remote_dispatch.pull_argv(access, root, rel)
    assert push[-2:] == [f"{root}/", f"user@100.0.0.2:{rel}/"]
    assert pull[-2:] == [f"user@100.0.0.2:{rel}/", f"{root}/"]
    assert [a for a in push if a != push[-2] and a != push[-1]][:-2] == [
        a for a in pull if a != pull[-2] and a != pull[-1]
    ][:-2]
    assert "node_modules" in push and "node_modules" in pull
    assert ".worktrees" in push and ".worktrees" in pull
    assert ".tmpjail-work" in push and ".tmpjail-work" in pull


def test_sync_skips_an_unreadable_overlay_workdir(tmp_path: Path):
    root = tmp_path / "repo"
    (root / "src").mkdir(parents=True)
    (root / "src" / "keep.txt").write_text("keep", encoding="utf-8")
    workdir = root / ".tmpjail-work" / "work"
    workdir.mkdir(parents=True)
    workdir.chmod(0o000)
    dest = tmp_path / "mirror"
    dest.mkdir()
    try:
        done = subprocess.run(
            ["rsync", "-a", *remote_dispatch._exclude_args(), f"{root}/", f"{dest}/"],
            capture_output=True,
            text=True,
        )
    finally:
        workdir.chmod(0o700)
    assert done.returncode == 0, done.stderr
    assert (dest / "src" / "keep.txt").read_text(encoding="utf-8") == "keep"
    assert not (dest / ".tmpjail-work").exists()


def test_mirror_uses_nested_gitignore_without_dropping_tracked_matches(tmp_path: Path):
    root = make_repo(tmp_path)
    nested = root / "nested"
    nested.mkdir()
    (nested / ".gitignore").write_text("ignored.txt\ntracked-ignored.txt\n", encoding="utf-8")
    (nested / "tracked-ignored.txt").write_text("tracked", encoding="utf-8")
    git("add", "nested/.gitignore", cwd=root)
    git("add", "-f", "nested/tracked-ignored.txt", cwd=root)
    git("commit", "-qm", "add nested ignore", cwd=root)
    (nested / "ignored.txt").write_text("ignored", encoding="utf-8")
    (nested / "unignored.txt").write_text("unignored", encoding="utf-8")

    dest = tmp_path / "mirror"
    dest.mkdir()
    selected = remote_dispatch.git_mirror_files(root)
    selected_paths = set(selected.rstrip(b"\0").split(b"\0"))
    assert b"nested/.gitignore" in selected_paths
    assert b"nested/tracked-ignored.txt" in selected_paths
    assert b"nested/unignored.txt" not in selected_paths
    assert b"nested/ignored.txt" not in selected_paths
    tracked = subprocess.run(
        remote_dispatch.tracked_sync_argv(f"{root}/", f"{dest}/"),
        input=selected,
        capture_output=True,
    )
    assert tracked.returncode == 0, tracked.stderr.decode()
    filtered = subprocess.run(
        ["rsync", "-a", "--delete", *remote_dispatch._exclude_args(), f"{root}/", f"{dest}/"],
        capture_output=True,
    )
    assert filtered.returncode == 0, filtered.stderr.decode()
    assert (dest / "nested" / "unignored.txt").read_text(encoding="utf-8") == "unignored"
    assert not (dest / "nested" / "ignored.txt").exists()
    assert (dest / "nested" / "tracked-ignored.txt").read_text(encoding="utf-8") == "tracked"


def test_dispatch_secret_filter_excludes_and_logs_env_but_keeps_manifest_and_normal_file(
    tmp_path: Path, capsys: pytest.CaptureFixture[str]
):
    root = make_repo(tmp_path)
    (root / ".env").write_text("TOKEN=do-not-ship\n", encoding="utf-8")
    manifest = root / "modules" / "harness" / "seat" / "credentials.json"
    manifest.parent.mkdir(parents=True)
    manifest.write_text('{"items":[]}\n', encoding="utf-8")
    (root / "normal.txt").write_text("ship me\n", encoding="utf-8")
    git("add", "-f", ".env", "modules/harness/seat/credentials.json", "normal.txt", cwd=root)

    selected = remote_dispatch.git_mirror_files(root)
    paths = set(selected.rstrip(b"\0").split(b"\0"))

    assert b".env" not in paths
    assert b"modules/harness/seat/credentials.json" in paths
    assert b"normal.txt" in paths
    log = capsys.readouterr().err
    assert "excluded secret-shaped file '.env'" in log
    assert "environment file (.env or .env.*)" in log
    assert "do-not-ship" not in log

    snapshot = remote_dispatch.snapshot_commit(root)
    tree_paths = set(git("ls-tree", "-r", "--name-only", snapshot, cwd=root).stdout.splitlines())
    assert ".env" not in tree_paths
    assert "modules/harness/seat/credentials.json" in tree_paths
    assert "normal.txt" in tree_paths


def test_egress_policy_allows_resolved_allowlist_and_omits_denied_host(tmp_path: Path, monkeypatch):
    path = Path(__file__).parents[2] / "sandbox" / "host" / "lib" / "egress-policy.py"
    spec = importlib.util.spec_from_file_location("egress_policy_test", path)
    assert spec and spec.loader
    policy = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(policy)
    config = tmp_path / "allow.json"
    config.write_text('{"schema_version":1,"hosts":["allowed.example"]}', encoding="utf-8")

    def fake_getaddrinfo(host, *_args, **_kwargs):
        values = {"allowed.example": "203.0.113.10", "denied.example": "203.0.113.99"}
        return [(2, 1, 6, "", (values[host], 0))]

    monkeypatch.setattr(policy.socket, "getaddrinfo", fake_getaddrinfo)
    allowed = set(policy.resolve_hosts(policy.load_hosts(config)))
    denied = set(policy.resolve_hosts(["denied.example"]))
    assert "203.0.113.10" in allowed
    assert allowed.isdisjoint(denied)


def test_egress_policy_fails_closed_when_allowlist_is_invalid(tmp_path: Path):
    path = Path(__file__).parents[2] / "sandbox" / "host" / "lib" / "egress-policy.py"
    spec = importlib.util.spec_from_file_location("egress_policy_invalid_test", path)
    assert spec and spec.loader
    policy = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(policy)
    config = tmp_path / "allow.json"
    config.write_text('{"schema_version":1,"hosts":[]}', encoding="utf-8")
    assert policy.main(["--config", str(config)]) == 12


def test_launcher_argv_starts_the_container_with_pinned_invocation_shape(tmp_path: Path):
    cred = make_credential(tmp_path / "cred.json")
    root = tmp_path / "workspace"
    argv = remote_dispatch.launcher_argv(
        host_name="debian1",
        git_common="/home/nodeuser/cdx-offload/git-common/repo-9999",
        git_dir="/home/nodeuser/.git/worktrees/main",
        credential=cred,
        exec_name="codex",
        forwarded_argv=["exec", "hello world"],
        sandbox_id=remote_dispatch.sandbox_id(root),
    )

    assert argv == [
        str(remote_dispatch.AGENT_SANDBOX_BIN),
        "--host",
        "debian1",
        "--id",
        remote_dispatch.sandbox_id(root),
        "--runtime",
        "codex",
        "--git-common",
        "/home/nodeuser/cdx-offload/git-common/repo-9999",
        "--git-dir",
        "/home/nodeuser/.git/worktrees/main",
        "--",
        "codex",
        "exec",
        "hello world",
    ]

    assert remote_dispatch.sandbox_id(root).startswith("workspace-")
    # --workspace would bind the mirror a second time over the $SANDBOX_ROOT mount it already
    # arrives on; sandbox-run's default workspace for this --id is that same directory.
    assert "--workspace" not in argv


def test_launcher_invocation_puts_the_command_after_bare_dash_dash(tmp_path: Path):
    argv = remote_dispatch.launcher_argv(
        host_name="debian1",
        credential=make_credential(tmp_path / "cred2.json"),
        exec_name="codex",
        forwarded_argv=["exec", "hi"],
        sandbox_id="cdx-offload-d-1234",
    )
    assert argv[-4:] == ["--", "codex", "exec", "hi"]


def test_open_session_pushes_then_returns_remote_argv(tmp_path, monkeypatch):
    calls: list[list[str]] = []
    cred_path = tmp_path / "cred.json"

    def fake_run(argv, **kwargs):
        calls.append(argv)
        if argv[:1] == ["rsync"]:
            return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")
        if "printf '%s\\n' \"$HOME\"" in " ".join(argv):
            return subprocess.CompletedProcess(argv, 0, stdout="/home/nodeuser\n", stderr="")
        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")

    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: tmp_path / "repo")
    session = remote_dispatch.open_session(
        "codex",
        ["exec", "hi"],
        credential=make_credential(cred_path),
        registry_path=write_registry(tmp_path),
        run=fake_run,
        probe=lambda a: 1.0 if a["host"] == "100.0.0.1" else 5.0,
    )
    assert session.host_name == "debian1"
    assert any("mkdir -p" in " ".join(c) for c in calls)
    assert any(c[0] == "rsync" for c in calls)
    assert session.argv[0] == str(remote_dispatch.AGENT_SANDBOX_BIN)
    assert session.sandbox_id == remote_dispatch.sandbox_id(tmp_path / "repo")
    assert session.node_home == "/home/nodeuser"
    assert session.workspace == f"/home/nodeuser/{session.rel_dir}"
    assert session.mounts == ()
    assert session.credential == make_credential(cred_path)


def test_open_session_requires_a_credential_object(tmp_path, isolated_claim_dir):
    with pytest.raises(remote_dispatch.OffloadUnavailable) as exc_info:
        remote_dispatch.open_session(
            "codex",
            ["exec", "hi"],
            registry_path=write_registry(tmp_path),
            probe=lambda a: 1.0,
        )
    assert exc_info.value.code == remote_dispatch.EXIT_CREDENTIAL
    assert "codex" in str(exc_info.value)
    assert list(isolated_claim_dir.glob("*.claim")) == []


def test_open_session_rejects_a_missing_or_unreadable_credential_path(tmp_path):
    credential_path = tmp_path / "cred.json"
    credential_path.write_text("very-sensitive-token", encoding="utf-8")
    credential_path.unlink()
    with pytest.raises(remote_dispatch.OffloadUnavailable) as exc_info:
        remote_dispatch.open_session(
            "codex",
            ["exec", "hi"],
            credential=remote_dispatch.Credential("codex", "acct", credential_path),
            registry_path=write_registry(tmp_path),
            probe=lambda a: 1.0,
        )
    assert exc_info.value.code == remote_dispatch.EXIT_CREDENTIAL
    assert "very-sensitive-token" not in str(exc_info.value)
    assert "codex" in str(exc_info.value)
    assert str(credential_path) in str(exc_info.value)


def test_open_session_fails_closed_when_the_push_fails(tmp_path, monkeypatch):
    def fake_run(argv, **kwargs):
        rc = 1 if argv[0] == "rsync" else 0
        if argv[0] == "ssh":
            return subprocess.CompletedProcess(
                argv, 0, stdout="/home/nodeuser\n", stderr=""
            )
        return subprocess.CompletedProcess(argv, rc, stdout="", stderr="disk full")

    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: tmp_path / "repo")
    with pytest.raises(remote_dispatch.OffloadUnavailable, match="mirror push failed") as exc_info:
        remote_dispatch.open_session(
            "codex",
            ["exec", "hi"],
            credential=make_credential(tmp_path / "cred.json"),
            registry_path=write_registry(tmp_path),
            run=fake_run,
            probe=lambda a: 1.0,
        )
    assert exc_info.value.code == remote_dispatch.EXIT_MIRROR


def test_pull_back_reports_failure_instead_of_raising(tmp_path):
    session = remote_dispatch.Session(
        host_name="debian1",
        access=REGISTRY["hosts"][0]["access"]["tailscale_ip"],
        root=tmp_path,
        rel_dir="cdx-offload/x",
        argv=["ssh"],
    )
    err = session.pull_back(
        run=lambda argv, **kw: subprocess.CompletedProcess(argv, 23, stdout="", stderr="boom")
    )
    assert err is not None and "debian1" in err and "boom" in err
    assert session.pull_back(
        run=lambda argv, **kw: subprocess.CompletedProcess(argv, 0, stdout="", stderr="")
    ) is None


def test_load_registry_missing_file_is_offload_unavailable(tmp_path):
    with pytest.raises(remote_dispatch.OffloadUnavailable) as exc_info:
        remote_dispatch.load_registry(tmp_path / "nope.json")
    assert exc_info.value.code == remote_dispatch.EXIT_NO_NODE


def test_git_mirror_is_none_for_a_plain_checkout(tmp_path):
    assert remote_dispatch.git_mirror(make_repo(tmp_path)) is None


def test_git_mirror_is_none_outside_a_repository(tmp_path):
    plain = tmp_path / "notrepo"
    plain.mkdir()
    assert remote_dispatch.git_mirror(plain) is None


def test_git_mirror_detects_a_linked_worktree_and_targets_the_common_worktrees_dir(tmp_path):
    root = make_repo(tmp_path)
    wt = add_worktree(root, "alpha")
    mirror = remote_dispatch.git_mirror(wt)
    assert mirror is not None
    assert mirror.git_dir == (root / ".git" / "worktrees" / "alpha").resolve()
    assert mirror.common_dir == (root / ".git").resolve()
    assert mirror.remote_common.startswith("cdx-offload/git-common/repo-")
    assert mirror.remote_git_dir == f"{mirror.remote_common}/worktrees/alpha"


def test_two_worktrees_of_one_repo_share_the_common_mirror_but_not_the_gitdir(tmp_path):
    root = make_repo(tmp_path)
    one = remote_dispatch.git_mirror(add_worktree(root, "one"))
    two = remote_dispatch.git_mirror(add_worktree(root, "two"))
    assert one.remote_common == two.remote_common
    assert one.remote_git_dir != two.remote_git_dir


def test_common_push_keeps_sibling_worktrees_out_and_never_deletes():
    access = REGISTRY["hosts"][1]["access"]["tailscale_ip"]
    mirror = remote_dispatch.GitMirror(
        git_dir=Path("/r/.git/worktrees/a"),
        common_dir=Path("/r/.git"),
        remote_common="cdx-offload/git-common/r-abc",
        remote_git_dir="cdx-offload/git-common/r-abc/worktrees/a",
    )
    argv = remote_dispatch.git_common_push_argv(access, mirror)
    assert "--delete" not in argv
    assert argv[argv.index("--exclude") + 1] == "/worktrees"
    assert argv[-2:] == ["/r/.git/", "user@100.0.0.2:cdx-offload/git-common/r-abc/"]

    gitdir = remote_dispatch.git_dir_push_argv(access, mirror)
    assert "--delete" in gitdir
    assert gitdir[-2:] == [
        "/r/.git/worktrees/a/",
        "user@100.0.0.2:cdx-offload/git-common/r-abc/worktrees/a/",
    ]


def test_prepare_remote_git_creates_destinations_before_rsync():
    access = REGISTRY["hosts"][1]["access"]["tailscale_ip"]
    mirror = remote_dispatch.GitMirror(
        git_dir=Path("/r/.git/worktrees/a b"),
        common_dir=Path("/r/.git"),
        remote_common="cdx-offload/git-common/r-abc",
        remote_git_dir="cdx-offload/git-common/r-abc/worktrees/a b",
    )
    calls: list[list[str]] = []

    def fake_run(argv, **_kwargs):
        calls.append(argv)
        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")

    assert remote_dispatch.prepare_remote_git(
        access, "debian2", "cdx-offload/r", "/home/user/cdx-offload/r", mirror,
        run=fake_run,
    ) is None
    assert calls[0][:-1] == remote_dispatch.ssh_argv(access)
    assert calls[0][-1] == (
        "mkdir -p cdx-offload/git-common/r-abc "
        "'cdx-offload/git-common/r-abc/worktrees/a b'"
    )
    assert calls[1] == remote_dispatch.git_common_push_argv(access, mirror)
    assert calls[2] == remote_dispatch.git_dir_push_argv(access, mirror)


def test_dot_git_is_excluded_both_ways_only_for_a_worktree_dispatch():
    access = REGISTRY["hosts"][1]["access"]["tailscale_ip"]
    root = Path("/home/user/Projects/overdeck/.worktrees/x")
    for build in (remote_dispatch.push_argv, remote_dispatch.pull_argv):
        assert "/.git" in build(access, root, "d", exclude_git=True)
        assert "/.git" not in build(access, root, "d")


def test_pointer_command_makes_the_mirrored_worktree_a_working_checkout(tmp_path):
    """Runs the real remote script against a real mirror laid out under a fake $HOME."""
    root = make_repo(tmp_path)
    wt = add_worktree(root, "alpha")
    mirror = remote_dispatch.git_mirror(wt)
    rel_dir = remote_dispatch.remote_rel_dir(wt)
    container_dir = remote_dispatch.container_workspace(wt)

    home = tmp_path / "remote-home"
    shutil.copytree(wt, home / rel_dir, ignore=shutil.ignore_patterns(".git"))
    shutil.copytree(
        mirror.common_dir, home / mirror.remote_common, ignore=shutil.ignore_patterns("worktrees")
    )
    shutil.copytree(mirror.git_dir, home / mirror.remote_git_dir)

    done = subprocess.run(
        ["bash", "-c", remote_dispatch.git_pointer_command(rel_dir, container_dir, mirror)],
        capture_output=True,
        text=True,
        env={**os.environ, **GIT_ENV, "HOME": str(home)},
    )
    assert done.returncode == 0, done.stderr

    remote_wt = home / rel_dir
    assert (remote_wt / ".git").read_text(encoding="utf-8") == (
        f"gitdir: {home / mirror.remote_git_dir}\n"
    )
    assert (home / mirror.remote_git_dir / "commondir").read_text(encoding="utf-8") == "../..\n"
    assert (home / mirror.remote_git_dir / "gitdir").read_text(encoding="utf-8") == (
        f"{container_dir}/.git\n"
    )
    assert git("rev-parse", "--show-toplevel", cwd=remote_wt).stdout.strip() == str(remote_wt)
    assert "wt/alpha" in git("status", "-sb", cwd=remote_wt).stdout

    (remote_wt / "file.txt").write_text("edited on the node\n", encoding="utf-8")
    git("add", "file.txt", cwd=remote_wt)
    git("commit", "-qm", "node edit", cwd=remote_wt)
    assert git("log", "-1", "--pretty=%s", cwd=remote_wt).stdout.strip() == "node edit"


def test_pointer_command_survives_a_path_with_shell_metacharacters():
    mirror = remote_dispatch.GitMirror(
        git_dir=Path("/r/.git/worktrees/a b"),
        common_dir=Path("/r/.git"),
        remote_common="cdx-offload/git-common/r' $(id)-abc",
        remote_git_dir="cdx-offload/git-common/r' $(id)-abc/worktrees/a b",
    )
    cmd = remote_dispatch.git_pointer_command(
        "sandbox/workspaces/x y-1", "/sandbox/workspaces/x y-1", mirror
    )
    proof = subprocess.run(
        ["bash", "-c", f"printf '%s' {cmd.split('bash -lc ', 1)[1]}"],
        capture_output=True,
        text=True,
    )
    assert "$(id)" in proof.stdout
    assert "'sandbox/workspaces/x y-1'" in proof.stdout


def test_open_session_uses_git_push_for_a_worktree_and_the_result_comes_home(tmp_path, monkeypatch):
    root = make_repo(tmp_path)
    wt = add_worktree(root, "alpha")
    (wt / "untracked.txt").write_text("scratch\n", encoding="utf-8")
    calls: list[list[str]] = []
    cred_path = tmp_path / "cred.json"
    fake_run = real_node_fake_run(tmp_path / "node_home", calls)

    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: wt)
    session = remote_dispatch.open_session(
        "codex",
        ["exec", "hi"],
        credential=make_credential(cred_path),
        registry_path=write_registry(tmp_path),
        run=fake_run,
        probe=lambda a: 1.0,
    )
    assert session.git_push is True
    assert session.git_warning is None
    assert session.exclude_git is False
    assert session.mounts == (session.bare_abs,)
    assert "--git-common" in session.argv and session.bare_abs in session.argv
    assert "--git-dir" not in session.argv
    assert session.pre_sha is not None
    assert not any(c[0] == "rsync" for c in calls)
    node_workspace = tmp_path / "node_home" / session.rel_dir
    assert (node_workspace / "untracked.txt").read_text(encoding="utf-8") == "scratch\n"
    status = git("status", "--short", cwd=node_workspace)
    assert status.stdout.strip() == ""
    assert git("log", "-1", "--format=%H", cwd=node_workspace).stdout.strip() == session.pre_sha

    (node_workspace / "made-remotely.txt").write_text("result\n", encoding="utf-8")
    error = session.pull_back(run=fake_run)
    assert error is None
    assert (wt / "made-remotely.txt").read_text(encoding="utf-8") == "result\n"
    assert (wt / "untracked.txt").read_text(encoding="utf-8") == "scratch\n"
    # Both new files land as uncommitted-and-untracked, matching the old rsync transport's
    # contract — not staged, which `read-tree --reset -u` alone would leave them as.
    porcelain = {line[3:]: line[:2] for line in git("status", "--porcelain", cwd=wt).stdout.splitlines()}
    assert porcelain.get("made-remotely.txt") == "??"
    assert porcelain.get("untracked.txt") == "??"
    assert "cdx-offload.lock" not in porcelain
    assert not (wt / ".cdx-offload.lock").exists()


def test_open_session_uses_git_push_for_a_plain_checkout_too(tmp_path, monkeypatch):
    root = make_repo(tmp_path)
    calls: list[list[str]] = []
    cred_path = tmp_path / "cred.json"
    fake_run = real_node_fake_run(tmp_path / "node_home", calls)

    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: root)
    session = remote_dispatch.open_session(
        "codex",
        ["exec", "hi"],
        credential=make_credential(cred_path),
        registry_path=write_registry(tmp_path),
        run=fake_run,
        probe=lambda a: 1.0,
    )
    assert session.git_push is True
    assert session.git_warning is None and session.exclude_git is False
    assert session.mounts == (session.bare_abs,)
    assert not any(c[0] == "rsync" for c in calls)


def test_open_session_pull_back_refuses_when_the_local_checkout_moved(tmp_path, monkeypatch):
    root = make_repo(tmp_path)
    cred_path = tmp_path / "cred.json"
    fake_run = real_node_fake_run(tmp_path / "node_home")

    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: root)
    session = remote_dispatch.open_session(
        "codex",
        ["exec", "hi"],
        credential=make_credential(cred_path),
        registry_path=write_registry(tmp_path),
        run=fake_run,
        probe=lambda a: 1.0,
    )
    (root / "raced.txt").write_text("owner touched this\n", encoding="utf-8")
    error = session.pull_back(run=fake_run)
    assert error is not None
    assert "not overwriting" in error
    assert (root / "raced.txt").read_text(encoding="utf-8") == "owner touched this\n"


def test_open_session_fails_closed_when_the_bare_mirror_cannot_be_prepared(tmp_path, monkeypatch):
    root = make_repo(tmp_path)
    cred_path = tmp_path / "cred.json"
    base_fake_run = real_node_fake_run(tmp_path / "node_home")

    def fake_run(argv, **kwargs):
        if argv[:1] == ["ssh"] and "git init -q --bare" in argv[-1]:
            return subprocess.CompletedProcess(argv, 1, stdout="", stderr="no space left on device")
        return base_fake_run(argv, **kwargs)

    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: root)
    with pytest.raises(remote_dispatch.OffloadUnavailable) as exc_info:
        remote_dispatch.open_session(
            "codex",
            ["exec", "hi"],
            credential=make_credential(cred_path),
            registry_path=write_registry(tmp_path),
            run=fake_run,
            probe=lambda a: 1.0,
        )
    assert exc_info.value.code == remote_dispatch.EXIT_MIRROR
    assert "no space left on device" in str(exc_info.value)


def test_open_session_falls_back_to_the_tree_copy_transport_when_head_is_unborn(tmp_path, monkeypatch):
    root = tmp_path / "fresh-repo"
    root.mkdir()
    subprocess.run(["/usr/bin/git", "init", "-q", "-b", "main", "."], cwd=root, check=True)
    calls: list[list[str]] = []
    cred_path = tmp_path / "cred.json"

    def fake_run(argv, **kwargs):
        if argv[:1] == ["/usr/bin/git"]:
            return subprocess.run(argv, capture_output=True, text=True)
        calls.append(argv)
        if "printf '%s\\n' \"$HOME\"" in " ".join(argv):
            return subprocess.CompletedProcess(argv, 0, stdout="/home/nodeuser\n", stderr="")
        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")

    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: root)
    session = remote_dispatch.open_session(
        "codex",
        ["exec", "hi"],
        credential=make_credential(cred_path),
        registry_path=write_registry(tmp_path),
        run=fake_run,
        probe=lambda a: 1.0,
    )
    assert session.git_push is False
    assert session.pre_sha is None
    assert any(c[0] == "rsync" for c in calls)


def test_snapshot_commit_returns_head_unchanged_when_clean(tmp_path):
    root = make_repo(tmp_path)
    head = git("rev-parse", "HEAD", cwd=root).stdout.strip()
    assert remote_dispatch.snapshot_commit(root) == head


def test_snapshot_commit_captures_untracked_files_without_moving_head(tmp_path):
    root = make_repo(tmp_path)
    head = git("rev-parse", "HEAD", cwd=root).stdout.strip()
    (root / "untracked.txt").write_text("x\n", encoding="utf-8")
    sha = remote_dispatch.snapshot_commit(root)
    assert sha != head
    assert git("rev-parse", "HEAD", cwd=root).stdout.strip() == head
    assert git("cat-file", "-p", f"{sha}:untracked.txt", cwd=root).stdout == "x\n"
    assert (root / "untracked.txt").exists()  # the real index and worktree are untouched


def test_snapshot_commit_returns_none_for_an_unborn_head(tmp_path):
    root = tmp_path / "fresh"
    root.mkdir()
    subprocess.run(["/usr/bin/git", "init", "-q", "."], cwd=root, check=True)
    assert remote_dispatch.snapshot_commit(root) is None


def test_a_transport_timeout_during_snapshot_fails_closed_not_silently(tmp_path):
    """A hung/timed-out local git call must surface as `OffloadUnavailable` (a nonzero exit
    downstream) — never as a raw exception or a swallowed success."""
    root = make_repo(tmp_path)

    def timing_out_run(argv, **kwargs):
        raise subprocess.TimeoutExpired(cmd=argv, timeout=kwargs.get("timeout", 1))

    with pytest.raises(remote_dispatch.OffloadUnavailable) as exc_info:
        remote_dispatch.snapshot_commit(root, run=timing_out_run)
    assert exc_info.value.code == remote_dispatch.EXIT_MIRROR


def test_repo_key_and_remote_bare_dir_are_stable_and_namespaced():
    key = remote_dispatch.repo_key(Path("/home/user/repo/.git"))
    assert key == remote_dispatch.repo_key(Path("/home/user/repo/.git"))
    assert remote_dispatch.remote_bare_dir(key) == f"{remote_dispatch.REMOTE_GIT_BARE_ROOT}/{key}.git"
    assert remote_dispatch.REMOTE_GIT_BARE_ROOT != remote_dispatch.REMOTE_GIT_COMMON_ROOT


def test_bare_ssh_url_includes_port_only_when_set():
    assert remote_dispatch.bare_ssh_url({"host": "h", "user": "u"}, "/a/b.git") == "ssh://u@h/a/b.git"
    assert (
        remote_dispatch.bare_ssh_url({"host": "h", "user": "u", "port": 2222}, "/a/b.git")
        == "ssh://u@h:2222/a/b.git"
    )


def test_push_commit_argv_shape():
    argv = remote_dispatch.push_commit_argv(Path("/repo"), "ssh://u@h/a.git", "deadbeef", "refs/cdx/x/in")
    assert argv == [
        "/usr/bin/git", "-C", "/repo", "push", "-q", "--no-verify", "ssh://u@h/a.git", "deadbeef:refs/cdx/x/in",
    ]


def test_dispatch_and_result_refs_are_distinct_namespaces():
    assert remote_dispatch.dispatch_ref("id", "d1") != remote_dispatch.result_ref("id")
    assert remote_dispatch.dispatch_ref("id", "d1").startswith(remote_dispatch.PUSH_REF_PREFIX)
    assert remote_dispatch.result_ref("id").startswith(remote_dispatch.PUSH_REF_PREFIX)


def test_dispatch_ref_is_unique_per_invocation_for_the_same_sandbox():
    a, b = remote_dispatch.new_dispatch_id(), remote_dispatch.new_dispatch_id()
    assert a != b
    assert remote_dispatch.dispatch_ref("sbox", a) != remote_dispatch.dispatch_ref("sbox", b)
    for dispatch_id in (a, b):
        assert remote_dispatch.dispatch_ref("sbox", dispatch_id) == (
            f"{remote_dispatch.PUSH_REF_PREFIX}/sbox/{dispatch_id}/in"
        )


def test_open_session_fetches_the_exact_per_dispatch_ref(tmp_path, monkeypatch):
    root = make_repo(tmp_path)
    calls: list[list[str]] = []
    cred_path = tmp_path / "cred.json"
    fake_run = real_node_fake_run(tmp_path / "node_home", calls)

    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: root)
    session = remote_dispatch.open_session(
        "codex",
        ["exec", "hi"],
        credential=make_credential(cred_path),
        registry_path=write_registry(tmp_path),
        run=fake_run,
        probe=lambda a: 1.0,
    )
    checkout_scripts = [
        c[-1] for c in calls if c[:1] == ["ssh"] and "checkout -qf" in c[-1]
    ]
    assert len(checkout_scripts) == 1
    expected = f"+{remote_dispatch.PUSH_REF_PREFIX}/{session.sandbox_id}/"
    assert expected in checkout_scripts[0]
    assert ":refs/cdx/in" in checkout_scripts[0]
    assert "/in:refs/cdx/in" in checkout_scripts[0]
    # never the bare, sandbox-wide ref the old shared-tip design used
    assert f"+{remote_dispatch.PUSH_REF_PREFIX}/{session.sandbox_id}/in:" not in checkout_scripts[0]


def test_gc_dispatch_refs_deletes_only_stale_refs_never_the_in_flight_one(tmp_path, monkeypatch):
    root = make_repo(tmp_path)
    cred_path = tmp_path / "cred.json"
    fake_run = real_node_fake_run(tmp_path / "node_home")

    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: root)
    session = remote_dispatch.open_session(
        "codex",
        ["exec", "hi"],
        credential=make_credential(cred_path),
        registry_path=write_registry(tmp_path),
        run=fake_run,
        probe=lambda a: 1.0,
    )
    bare_dir = tmp_path / "node_home" / remote_dispatch.remote_bare_dir(
        remote_dispatch.repo_key(remote_dispatch.git_common_dir(root))
    )
    sbox = session.sandbox_id
    prefix = f"{remote_dispatch.PUSH_REF_PREFIX}/{sbox}"
    stale_ref = f"{prefix}/{int(1)}-{'a' * 32}/in"
    fresh_ref = f"{prefix}/{int(remote_dispatch.time.time())}-{'b' * 32}/in"
    head = git("rev-parse", "HEAD", cwd=root).stdout.strip()
    for ref in (stale_ref, fresh_ref):
        git("push", "-q", str(bare_dir), f"{head}:{ref}", cwd=root)

    warning = remote_dispatch.gc_dispatch_refs(
        root, str(bare_dir), sbox, remote_dispatch.dispatch_ref(sbox, "keep-me"), env=os.environ, run=subprocess.run
    )
    assert warning is None
    remaining = {
        line.split()[1]
        for line in git("ls-remote", str(bare_dir), f"{prefix}/*/in", cwd=root).stdout.splitlines()
    }
    assert stale_ref not in remaining
    assert fresh_ref in remaining


def test_gc_dispatch_refs_never_deletes_the_ref_just_pushed(tmp_path):
    root = make_repo(tmp_path)
    bare_dir = tmp_path / "bare.git"
    git("init", "-q", "--bare", str(bare_dir), cwd=tmp_path)
    sbox = "sbox"
    keep_id = remote_dispatch.new_dispatch_id()
    keep_ref = remote_dispatch.dispatch_ref(sbox, keep_id)
    old_ref = remote_dispatch.dispatch_ref(sbox, f"1-{'c' * 32}")
    head = git("rev-parse", "HEAD", cwd=root).stdout.strip()
    for ref in (keep_ref, old_ref):
        git("push", "-q", str(bare_dir), f"{head}:{ref}", cwd=root)

    warning = remote_dispatch.gc_dispatch_refs(
        root, str(bare_dir), sbox, keep_ref, env=os.environ, run=subprocess.run
    )
    assert warning is None
    remaining = {
        line.split()[1]
        for line in git("ls-remote", str(bare_dir), f"refs/cdx/{sbox}/*/in", cwd=root).stdout.splitlines()
    }
    assert keep_ref in remaining
    assert old_ref not in remaining


def test_gc_dispatch_refs_failure_is_a_non_fatal_warning(tmp_path):
    root = make_repo(tmp_path)

    def failing_run(argv, **kwargs):
        return subprocess.CompletedProcess(argv, 1, stdout="", stderr="connection refused")

    warning = remote_dispatch.gc_dispatch_refs(
        root, "ssh://u@h/gone.git", "sbox", "refs/cdx/sbox/x/in",
        env=os.environ, host_name="debian9", run=failing_run,
    )
    assert warning is not None
    # owner-safe: no internal stage name, no ssh URL, no raw git stderr on this surface
    assert "debian9" in warning
    assert "connection refused" not in warning
    assert "ssh://" not in warning


def test_open_session_still_succeeds_when_gc_cannot_reach_the_remote(tmp_path, monkeypatch):
    root = make_repo(tmp_path)
    cred_path = tmp_path / "cred.json"
    base_fake_run = real_node_fake_run(tmp_path / "node_home")

    def fake_run(argv, **kwargs):
        if argv[:1] == ["/usr/bin/git"] and "ls-remote" in argv:
            return subprocess.CompletedProcess(argv, 128, stdout="", stderr="could not resolve host")
        return base_fake_run(argv, **kwargs)

    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: root)
    session = remote_dispatch.open_session(
        "codex",
        ["exec", "hi"],
        credential=make_credential(cred_path),
        registry_path=write_registry(tmp_path),
        run=fake_run,
        probe=lambda a: 1.0,
    )
    assert session.git_push is True
    assert session.git_warning is not None
    assert session.host_name in session.git_warning
    assert "could not resolve host" not in session.git_warning


def test_remote_checkout_script_scopes_clean_to_sync_excludes():
    script = remote_dispatch.remote_checkout_script(
        "sandbox/workspaces/x", "/home/n/cdx-offload/repos/k.git", "refs/cdx/x/in", "deadbeef"
    )
    assert "clean -qffd" in script
    for excluded in remote_dispatch.SYNC_EXCLUDES:
        assert f"-e {excluded}" in script


def test_checkout_and_result_scripts_lock_inside_git_dir_not_the_worktree():
    """A lock file in the worktree root is an untracked path `git add -A` in
    remote_result_script would pick up and round-trip home as spurious state."""
    checkout = remote_dispatch.remote_checkout_script(
        "sandbox/workspaces/x", "/home/n/cdx-offload/repos/k.git", "refs/cdx/x/in", "deadbeef"
    )
    result = remote_dispatch.remote_result_script(
        "sandbox/workspaces/x", "/home/n/cdx-offload/repos/k.git", "refs/cdx/x/out"
    )
    for script in (checkout, result):
        assert '"$w/.git/cdx-offload.lock"' in script
        assert '"$w/.cdx-offload.lock"' not in script


def test_result_push_forces_the_scratch_ref():
    """The out ref points at the latest result, not at a history. A reused workspace whose
    base moved makes the push a non-fast-forward, and a rejected push strands the run."""
    script = remote_dispatch.remote_result_script(
        "sandbox/workspaces/x", "/home/n/cdx-offload/repos/k.git", "refs/cdx/x/out"
    )
    assert 'git push -q --force "$b" "$sha:refs/cdx/x/out"' in script


def flat_load_selection(claims: Path):
    """Every node reports the same lagging load average — placement must still spread."""
    return remote_dispatch.select_host(
        remote_dispatch.candidate_hosts(THREE_REACHABLE),
        probe=lambda a: 0.5,
        claims=claims,
    )


def test_selections_with_identical_probe_results_spread_across_hosts(isolated_claim_dir):
    held = [flat_load_selection(isolated_claim_dir) for _ in range(3)]
    assert sorted(host["name"] for host, _access, _claim in held) == [
        "debian1",
        "debian2",
        "debian3",
    ]
    assert all(claim is not None for _host, _access, claim in held)


def test_a_released_claim_frees_its_host_for_the_next_selection(isolated_claim_dir):
    held = [flat_load_selection(isolated_claim_dir) for _ in range(3)]
    freed_host, _access, claim = held[1]
    claim.release()
    host, _access, _claim = flat_load_selection(isolated_claim_dir)
    assert host["name"] == freed_host["name"]


def _child_selection(claims: str, results: str, start, hold) -> None:
    random.seed()  # forked children inherit one RNG stream; real dispatches do not
    start.wait(timeout=30)
    host, _access, _claim = flat_load_selection(Path(claims))
    (Path(results) / str(os.getpid())).write_text(host["name"], encoding="utf-8")
    hold.wait(timeout=30)


def test_concurrent_dispatches_do_not_all_stack_on_one_host(isolated_claim_dir, tmp_path):
    """The measured defect: seven simultaneous probes saw one idle node and all took it."""
    ctx = multiprocessing.get_context("spawn")
    results = tmp_path / "results"
    results.mkdir()
    start, hold = ctx.Barrier(7), ctx.Barrier(7)
    children = [
        ctx.Process(
            target=_child_selection,
            args=(str(isolated_claim_dir), str(results), start, hold),
        )
        for _ in range(7)
    ]
    for child in children:
        child.start()
    for child in children:
        child.join(timeout=60)
    assert all(child.exitcode == 0 for child in children), [c.exitcode for c in children]
    chosen = [path.read_text(encoding="utf-8") for path in results.iterdir()]
    assert len(chosen) == 7
    assert max(Counter(chosen).values()) <= 6
    assert len(set(chosen)) >= 2


def _plant_claim(directory: Path, host: str, pid: int, start_ticks: str) -> Path:
    path = directory / f"{host}__{pid}__{start_ticks}__planted.claim"
    path.touch()
    return path


def test_a_claim_whose_pid_is_dead_is_reaped_and_ignored(isolated_claim_dir):
    """A crashed or killed dispatch must not leave its node looking busy forever."""
    gone = subprocess.Popen(["/bin/true"])
    gone.wait()
    corpse = _plant_claim(isolated_claim_dir, "debian1", gone.pid, "0")
    live_pid_wrong_identity = _plant_claim(
        isolated_claim_dir, "debian1", os.getpid(), "999999999"
    )
    # two unreaped claims would push debian1 (0.1) above debian2 (1.2)
    loads = {"100.0.0.1": 0.1, "100.0.0.2": 1.2, "100.0.0.3": 1.2}

    host, _access, _claim = remote_dispatch.select_host(
        remote_dispatch.candidate_hosts(THREE_REACHABLE),
        probe=lambda a: loads[a["host"]],
        claims=isolated_claim_dir,
    )

    assert host["name"] == "debian1"
    assert not corpse.exists()
    assert not live_pid_wrong_identity.exists()


def test_a_live_claim_is_counted_and_survives_reaping(isolated_claim_dir):
    mine = _plant_claim(
        isolated_claim_dir,
        "debian1",
        os.getpid(),
        remote_dispatch._process_start_ticks(os.getpid()) or "0",
    )
    host, _access, _claim = flat_load_selection(isolated_claim_dir)
    assert host["name"] != "debian1"
    assert mine.exists()


def test_claim_bookkeeping_failure_degrades_to_plain_least_loaded(tmp_path):
    unusable = tmp_path / "not-a-directory"
    unusable.write_text("", encoding="utf-8")
    loads = {"100.0.0.1": 9.0, "100.0.0.2": 0.2, "100.0.0.3": 4.0}
    host, _access, claim = remote_dispatch.select_host(
        remote_dispatch.candidate_hosts(THREE_REACHABLE),
        probe=lambda a: loads[a["host"]],
        claims=unusable,
    )
    assert host["name"] == "debian2"
    assert claim is None


def test_open_session_releases_the_claim_when_the_mirror_push_fails(tmp_path, monkeypatch):
    claims = tmp_path / "claims"

    def fake_run(argv, **kwargs):
        rc = 1 if argv[0] == "rsync" else 0
        return subprocess.CompletedProcess(argv, rc, stdout="", stderr="disk full")

    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: tmp_path / "repo")
    with pytest.raises(remote_dispatch.OffloadUnavailable):
        remote_dispatch.open_session(
            "codex",
            ["exec", "hi"],
            credential=make_credential(tmp_path / "cred.json"),
            registry_path=write_registry(tmp_path, THREE_REACHABLE),
            run=fake_run,
            probe=lambda a: 1.0,
            claims=claims,
        )
    assert list(claims.glob("*.claim")) == []


def test_session_release_frees_the_claim_at_the_end_of_a_run(tmp_path, monkeypatch):
    claims = tmp_path / "claims"
    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: tmp_path / "repo")
    session = remote_dispatch.open_session(
        "codex",
        ["exec", "hi"],
        credential=make_credential(tmp_path / "cred.json"),
        registry_path=write_registry(tmp_path, THREE_REACHABLE),
        run=lambda argv, **kw: subprocess.CompletedProcess(
            argv, 0, stdout="/home/nodeuser\n", stderr=""
        ),
        probe=lambda a: 1.0,
        claims=claims,
    )
    assert len(list(claims.glob("*.claim"))) == 1
    session.release()
    assert list(claims.glob("*.claim")) == []
    session.release()


def _workdir_session(tmp_path, monkeypatch, forwarded_argv, root):
    def fake_run(argv, **kwargs):
        if "printf '%s\\n' \"$HOME\"" in " ".join(argv):
            return subprocess.CompletedProcess(argv, 0, stdout="/home/nodeuser\n", stderr="")
        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")

    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: root)
    monkeypatch.setattr(remote_dispatch, "git_toplevel", lambda cwd: root)
    return remote_dispatch.open_session(
        "codex",
        forwarded_argv,
        credential=make_credential(tmp_path / "cred.json"),
        registry_path=write_registry(tmp_path),
        run=fake_run,
        probe=lambda a: 1.0,
    )


@pytest.mark.parametrize("flag", ["-C", "--cd"])
def test_open_session_translates_the_workdir_flag_to_its_container_path(
    tmp_path, monkeypatch, flag, isolated_claim_dir
):
    """A workstation path forwarded verbatim makes codex die with a bare `os error 2`."""
    root = tmp_path / "repo"
    (root / "sub").mkdir(parents=True)
    (tmp_path / "cred.json").write_text("{}")
    session = _workdir_session(
        tmp_path, monkeypatch, ["exec", flag, str(root / "sub"), "hi"], root
    )
    expected = f"{remote_dispatch.container_workspace(root)}/sub"
    assert expected in session.argv
    assert str(root / "sub") not in session.argv


def test_open_session_refuses_a_workdir_outside_the_mirrored_checkout(
    tmp_path, monkeypatch, isolated_claim_dir
):
    root = tmp_path / "repo"
    outside = tmp_path / "elsewhere"
    root.mkdir(parents=True)
    outside.mkdir()
    (tmp_path / "cred.json").write_text("{}")
    with pytest.raises(remote_dispatch.OffloadUnavailable) as exc_info:
        _workdir_session(tmp_path, monkeypatch, ["exec", "-C", str(outside), "hi"], root)
    assert exc_info.value.code == remote_dispatch.EXIT_MIRROR
    assert "outside the mirrored checkout" in str(exc_info.value)


def test_open_session_refuses_a_workdir_that_is_not_in_a_checkout(
    tmp_path, monkeypatch, isolated_claim_dir
):
    """Without this guard a non-repo `-C` makes repo_root fall back to it and mirror it whole."""
    loose = tmp_path / "loose"
    loose.mkdir()
    (tmp_path / "cred.json").write_text("{}")
    monkeypatch.setattr(remote_dispatch, "git_toplevel", lambda cwd: None)
    with pytest.raises(remote_dispatch.OffloadUnavailable) as exc_info:
        remote_dispatch.open_session(
            "codex",
            ["exec", "-C", str(loose), "hi"],
            credential=make_credential(tmp_path / "cred.json"),
            registry_path=write_registry(tmp_path),
        )
    assert exc_info.value.code == remote_dispatch.EXIT_MIRROR
    assert "not inside a git checkout" in str(exc_info.value)


def test_find_workdir_takes_the_last_occurrence_and_stops_at_the_prompt_separator():
    assert remote_dispatch.find_workdir(["exec", "-C", "/a", "--cd=/b", "x"]) == (3, 1, "/b")
    assert remote_dispatch.find_workdir(["exec", "--", "-C", "/a"]) is None
    assert remote_dispatch.find_workdir(["exec", "hi"]) is None


def test_push_excludes_the_tmpjail_overlay_scratch_root():
    """`.tmpjail-work/work` is mode 000 by kernel requirement; rsync aborts on it."""
    argv = remote_dispatch.push_argv({"host": "h"}, Path("/repo"), "rel")
    pairs = list(zip(argv, argv[1:]))
    assert ("--exclude", ".tmpjail-work") in pairs


def test_no_workspace_requested_reads_the_env_var(monkeypatch):
    monkeypatch.delenv(remote_dispatch.NO_WORKSPACE_ENV, raising=False)
    assert remote_dispatch.no_workspace_requested() is False
    monkeypatch.setenv(remote_dispatch.NO_WORKSPACE_ENV, "1")
    assert remote_dispatch.no_workspace_requested() is True


def test_remote_scratch_dir_never_collides_with_the_real_mirror(tmp_path):
    root = tmp_path / "repo"
    assert remote_dispatch.remote_scratch_dir(root) != remote_dispatch.remote_rel_dir(root)
    assert remote_dispatch.remote_scratch_dir(root).startswith(remote_dispatch.remote_rel_dir(root))


def test_open_session_no_workspace_transfers_nothing(tmp_path, monkeypatch):
    calls: list[list[str]] = []
    cred_path = tmp_path / "cred.json"

    def fake_run(argv, **kwargs):
        calls.append(argv)
        if "printf '%s\\n' \"$HOME\"" in " ".join(argv):
            return subprocess.CompletedProcess(argv, 0, stdout="/home/nodeuser\n", stderr="")
        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")

    monkeypatch.setenv(remote_dispatch.NO_WORKSPACE_ENV, "1")
    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: tmp_path / "repo")
    session = remote_dispatch.open_session(
        "codex",
        ["exec", "reply OK"],
        credential=make_credential(cred_path),
        registry_path=write_registry(tmp_path),
        run=fake_run,
        probe=lambda a: 1.0 if a["host"] == "100.0.0.1" else 5.0,
    )
    assert session.host_name == "debian1"
    assert session.no_workspace is True
    assert not any(c and c[0] in ("rsync", "scp") for c in calls)
    assert any("mkdir -p" in " ".join(c) for c in calls)
    assert session.rel_dir == remote_dispatch.remote_scratch_dir(tmp_path / "repo")
    assert session.mounts == ()
    assert session.git_warning is None
    assert session.pull_back(run=fake_run) is None
    assert not any(c and c[0] in ("rsync", "scp") for c in calls)
    # agent-sandbox derives the container's workspace from --id, not from rel_dir: these
    # three must name the same scratch identity or the container runs somewhere else.
    scratch_id = remote_dispatch.scratch_sandbox_id(tmp_path / "repo")
    assert session.sandbox_id == scratch_id
    assert session.rel_dir.endswith(scratch_id)
    assert f"--id {scratch_id} " in " ".join(session.argv) or scratch_id in session.argv


def test_open_session_no_workspace_refuses_a_workdir_flag(tmp_path, monkeypatch):
    monkeypatch.setenv(remote_dispatch.NO_WORKSPACE_ENV, "1")
    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: tmp_path / "repo")
    with pytest.raises(remote_dispatch.OffloadUnavailable) as exc_info:
        remote_dispatch.open_session(
            "codex",
            ["exec", "-C", str(tmp_path), "hi"],
            credential=make_credential(tmp_path / "cred.json"),
            registry_path=write_registry(tmp_path),
            probe=lambda a: 1.0,
        )
    assert exc_info.value.code == remote_dispatch.EXIT_MIRROR
    assert remote_dispatch.NO_WORKSPACE_ENV in str(exc_info.value)


def test_open_session_no_workspace_still_fails_closed_with_no_reachable_host(tmp_path, monkeypatch):
    monkeypatch.setenv(remote_dispatch.NO_WORKSPACE_ENV, "1")
    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: tmp_path / "repo")
    with pytest.raises(remote_dispatch.OffloadUnavailable) as exc_info:
        remote_dispatch.open_session(
            "codex",
            ["exec", "hi"],
            credential=make_credential(tmp_path / "cred.json"),
            registry_path=write_registry(tmp_path, {"hosts": []}),
            probe=lambda a: 1.0,
        )
    assert exc_info.value.code == remote_dispatch.EXIT_NO_NODE


def test_open_session_normal_path_still_pushes_when_env_unset(tmp_path, monkeypatch):
    monkeypatch.delenv(remote_dispatch.NO_WORKSPACE_ENV, raising=False)
    calls: list[list[str]] = []

    def fake_run(argv, **kwargs):
        calls.append(argv)
        if argv[:1] == ["rsync"]:
            return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")
        if "printf '%s\\n' \"$HOME\"" in " ".join(argv):
            return subprocess.CompletedProcess(argv, 0, stdout="/home/nodeuser\n", stderr="")
        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")

    monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: tmp_path / "repo")
    session = remote_dispatch.open_session(
        "codex",
        ["exec", "hi"],
        credential=make_credential(tmp_path / "cred.json"),
        registry_path=write_registry(tmp_path),
        run=fake_run,
        probe=lambda a: 1.0 if a["host"] == "100.0.0.1" else 5.0,
    )
    assert session.no_workspace is False
    assert any(c and c[0] == "rsync" for c in calls)


def test_sandbox_id_survives_an_uppercase_checkout_name(tmp_path):
    """sandbox-run enforces ^[a-z0-9][a-z0-9._-]{0,63}$; a capital letter in the
    checkout directory name must fold, not kill the dispatch before it starts."""
    sid = remote_dispatch.sandbox_id(tmp_path / "Botmaster")
    assert re.fullmatch(r"[a-z0-9][a-z0-9._-]{0,63}", sid), sid
    assert sid.startswith("botmaster-")