atlas-iac/testing/tests/test_hermes_cli_quarantine.py

"""Identity-safe quarantine and replacement races."""

from __future__ import annotations

from testing.tests.test_hermes_cli_support import (
    Path,
    SimpleNamespace,
    _completed_result,
    _install_terminal_recovery_db,
    _pending_terminal_record,
    hashlib,
    json,
    lanes,
    nullcontext,
    os,
    pytest,
    stat,
    sys,
)


@pytest.mark.parametrize(
    "structured",
    [
        {"status": "completed", "blockers": {}},
        {**_completed_result(), "changed_files": "src/a.py"},
        {**_completed_result(), "unexpected": True},
        {**_completed_result(), "blockers": ["work remains"]},
        {**_completed_result("tests are still running")},
    ],
)
def test_terminal_record_requires_exact_completed_result_contract(structured):
    record = {
        "board": "cassandra",
        "task_id": "t_schema",
        "expected_run_id": 1,
        "result": json.dumps(structured),
        "summary": str(structured.get("summary") or "done"),
        "metadata": {},
        "kanban_state": "pending",
    }

    assert lanes._terminal_record_valid(record) is False


def test_minimal_completed_result_with_mapping_blockers_quarantines_without_completion(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    path = lanes._terminal_path(lanes.state_path("cassandra", "t_schema"), 6)
    lanes.atomic_json(
        path,
        {
            "board": "cassandra",
            "task_id": "t_schema",
            "expected_run_id": 6,
            "result": json.dumps({"status": "completed", "blockers": {}}),
            "summary": "done",
            "metadata": {},
            "kanban_state": "pending",
        },
    )
    task = SimpleNamespace(
        id="t_schema", status="running", current_run_id=6, assignee="cli-auto"
    )
    completed = []
    reclaimed = []

    class Connection:
        def close(self):
            return None

    fake_db = SimpleNamespace(
        scoped_current_board=lambda _board: nullcontext(),
        connect=lambda board: Connection(),
        get_task=lambda _conn, _task_id: task,
        complete_task=lambda *_args, **_kwargs: completed.append(True),
        reclaim_task=lambda *_args, **_kwargs: (reclaimed.append(True) or True),
    )
    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))

    assert lanes.recover_pending_finalizations() == 0
    assert completed == []
    assert reclaimed == [True]
    assert not path.exists()
    quarantined = list((path.parent / "quarantine").glob("*.quarantine"))
    assert len(quarantined) == 1
    assert quarantined[0].stat().st_mode & 0o777 == 0o600


def test_quarantine_avoids_symlink_and_mode_collision_destinations(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    path = lanes._terminal_path(lanes.state_path("cassandra", "t_collision"), 2)
    path.parent.mkdir(parents=True)
    path.write_bytes(b"invalid")
    quarantine = path.parent / "quarantine"
    quarantine.mkdir()
    fingerprint = hashlib.sha256(b"invalid").hexdigest()[:16]
    base = f"{path.name}.malformed-payload.{fingerprint}"
    victim = tmp_path / "victim"
    victim.write_text("unchanged", encoding="utf-8")
    (quarantine / f"{base}.0.quarantine").symlink_to(victim)
    collision = quarantine / f"{base}.1.quarantine"
    collision.write_text("attacker collision", encoding="utf-8")
    collision.chmod(0o644)
    monkeypatch.setattr(
        lanes.os,
        "chmod",
        lambda *_args, **_kwargs: pytest.fail("quarantine must not chmod foreign inodes"),
    )

    destination = lanes._quarantine_terminal(
        path,
        lanes._terminal_identity(path),
        "malformed-payload",
    )

    assert not path.exists()
    assert destination.name == f"{base}.2.quarantine"
    assert destination.stat().st_mode & 0o777 == 0o600
    diagnostic = json.loads(destination.read_text(encoding="utf-8"))
    assert diagnostic["size"] == len(b"invalid")
    assert diagnostic["sha256"] == hashlib.sha256(b"invalid").hexdigest()
    assert diagnostic["source_kind"] == "regular"
    assert victim.read_text(encoding="utf-8") == "unchanged"
    assert collision.read_text(encoding="utf-8") == "attacker collision"
    assert collision.stat().st_mode & 0o777 == 0o644


def test_hardlinked_terminal_source_is_never_chmodded_or_copied_as_data(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    path = lanes._terminal_path(lanes.state_path("cassandra", "t_hardlink"), 5)
    path.parent.mkdir(parents=True)
    foreign = tmp_path / "foreign"
    foreign.write_text("foreign inode contents", encoding="utf-8")
    foreign.chmod(0o644)
    os.link(foreign, path)
    monkeypatch.setattr(
        lanes.os,
        "chmod",
        lambda *_args, **_kwargs: pytest.fail("hardlinked source must not be chmodded"),
    )

    destination = lanes._quarantine_terminal(
        path,
        lanes._terminal_identity(path),
        "malformed-payload",
    )

    assert not path.exists()
    assert foreign.read_text(encoding="utf-8") == "foreign inode contents"
    assert foreign.stat().st_mode & 0o777 == 0o644
    assert destination.stat().st_mode & 0o777 == 0o600
    diagnostic = json.loads(destination.read_text(encoding="utf-8"))
    assert diagnostic["source_kind"] == "hardlink"
    assert "foreign inode contents" not in destination.read_text(encoding="utf-8")


def test_quarantine_preserves_and_replays_an_atomic_replacement(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    path = lanes._terminal_path(lanes.state_path("cassandra", "t_swap"), 21)
    path.parent.mkdir(parents=True)
    path.write_bytes(b"malformed")
    identity = lanes._terminal_identity(path)
    assert identity is not None
    structured = _completed_result("replacement result")
    replacement_record = {
        "board": "cassandra",
        "task_id": "t_swap",
        "expected_run_id": 21,
        "result": json.dumps(structured, sort_keys=True),
        "summary": structured["summary"],
        "metadata": {},
        "kanban_state": "pending",
        "recorded_at": lanes.utc_now(),
    }
    replacement = path.with_name("replacement.tmp")
    replacement.write_text(
        json.dumps(replacement_record, sort_keys=True),
        encoding="utf-8",
    )
    replacement.chmod(0o600)
    real_fsync = lanes.os.fsync
    swapped = {"value": False}

    def swap_on_quarantine_directory_fsync(descriptor):
        descriptor_stat = os.fstat(descriptor)
        if stat.S_ISDIR(descriptor_stat.st_mode) and not swapped["value"]:
            os.replace(replacement, path)
            swapped["value"] = True
        real_fsync(descriptor)

    monkeypatch.setattr(lanes.os, "fsync", swap_on_quarantine_directory_fsync)

    lanes._quarantine_terminal(path, identity, "malformed-payload")

    assert swapped["value"] is True
    assert path.exists()
    assert lanes._terminal_record_valid(
        lanes._load_terminal_json(path, identity), identity
    )

    task = SimpleNamespace(
        id="t_swap", status="running", current_run_id=21, result=None, assignee="cli-auto"
    )

    class Connection:
        def close(self):
            return None

    def complete_task(_conn, _task_id, **kwargs):
        task.status = "done"
        task.result = kwargs["result"]
        task.current_run_id = None
        return True

    fake_db = SimpleNamespace(
        scoped_current_board=lambda _board: nullcontext(),
        connect=lambda board: Connection(),
        get_task=lambda _conn, _task_id: task,
        complete_task=complete_task,
    )
    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))

    assert lanes.recover_pending_finalizations() == 1
    assert task.status == "done"
    assert not path.exists()


def test_recovery_quarantines_the_loaded_inode_not_a_before_open_replacement(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    path = lanes._terminal_path(lanes.state_path("cassandra", "t_gap"), 51)
    path.parent.mkdir(parents=True)
    path.write_bytes(b"malformed-original")
    replacement = path.with_name("replacement.valid")
    lanes.atomic_json(
        replacement,
        _pending_terminal_record("cassandra", "t_gap", 51, "preserved replacement"),
    )
    task = SimpleNamespace(
        id="t_gap",
        status="running",
        current_run_id=51,
        result=None,
        assignee="cli-auto",
    )
    completions = []
    reclaims = []
    _install_terminal_recovery_db(monkeypatch, task, completions, reclaims)
    real_quarantine = lanes._quarantine_terminal
    swapped = {"value": False}

    def swap_before_quarantine_reopens(*args, **kwargs):
        assert kwargs["snapshot"].prefix == b"malformed-original"
        if not swapped["value"]:
            os.replace(replacement, path)
            swapped["value"] = True
        return real_quarantine(*args, **kwargs)

    monkeypatch.setattr(lanes, "_quarantine_terminal", swap_before_quarantine_reopens)

    assert lanes.recover_pending_finalizations() == 1
    assert swapped["value"] is True
    assert task.status == "done"
    assert len(completions) == 1
    assert reclaims == []
    assert not path.exists()
    committed = next(path.parent.glob("*.terminal.committed.json"))
    assert json.loads(committed.read_text())["result"] == completions[0]


def test_recovery_replays_replacement_installed_after_quarantine_pins_source(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    path = lanes._terminal_path(lanes.state_path("cassandra", "t_after_open"), 52)
    path.parent.mkdir(parents=True)
    path.write_bytes(b"malformed-original")
    replacement = path.with_name("replacement.valid")
    lanes.atomic_json(
        replacement,
        _pending_terminal_record(
            "cassandra", "t_after_open", 52, "replacement after open"
        ),
    )
    task = SimpleNamespace(
        id="t_after_open",
        status="running",
        current_run_id=52,
        result=None,
        assignee="cli-auto",
    )
    completions = []
    reclaims = []
    _install_terminal_recovery_db(monkeypatch, task, completions, reclaims)
    real_fsync = lanes.os.fsync
    swapped = {"value": False}

    def swap_after_quarantine_open(descriptor):
        descriptor_stat = os.fstat(descriptor)
        if stat.S_ISDIR(descriptor_stat.st_mode) and not swapped["value"]:
            os.replace(replacement, path)
            swapped["value"] = True
        real_fsync(descriptor)

    monkeypatch.setattr(lanes.os, "fsync", swap_after_quarantine_open)

    assert lanes.recover_pending_finalizations() == 1
    assert swapped["value"] is True
    assert task.status == "done"
    assert len(completions) == 1
    assert reclaims == []
    assert not path.exists()


def test_recovery_survives_repeated_invalid_then_valid_replacements(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    path = lanes._terminal_path(lanes.state_path("cassandra", "t_churn"), 53)
    path.parent.mkdir(parents=True)
    path.write_bytes(b"malformed-zero")
    second = path.with_name("replacement.invalid")
    second.write_bytes(b"malformed-one")
    valid = path.with_name("replacement.valid")
    lanes.atomic_json(
        valid,
        _pending_terminal_record("cassandra", "t_churn", 53, "valid after churn"),
    )
    task = SimpleNamespace(
        id="t_churn",
        status="running",
        current_run_id=53,
        result=None,
        assignee="cli-auto",
    )
    completions = []
    reclaims = []
    _install_terminal_recovery_db(monkeypatch, task, completions, reclaims)
    real_quarantine = lanes._quarantine_terminal
    replacements = [second, valid]
    pinned = []

    def churn_before_quarantine(*args, **kwargs):
        pinned.append(kwargs["snapshot"].prefix)
        if replacements:
            os.replace(replacements.pop(0), path)
        return real_quarantine(*args, **kwargs)

    monkeypatch.setattr(lanes, "_quarantine_terminal", churn_before_quarantine)

    assert lanes.recover_pending_finalizations() == 1
    assert pinned == [b"malformed-zero", b"malformed-one"]
    assert replacements == []
    assert task.status == "done"
    assert len(completions) == 1
    assert reclaims == []
    assert not path.exists()
    diagnostics = list((path.parent / "quarantine").glob("*.quarantine"))
    assert len(diagnostics) == 2