atlas-iac/testing/tests/test_hermes_cli_capabilities.py

"""Mixed-image capability gates for mounted Hermes CLI lane scripts."""

from __future__ import annotations

import json
import runpy
import sys
from contextlib import nullcontext
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace

import pytest

from testing.tests.test_hermes_cli_support import SCRIPTS, _completed_result, lanes


def _old_complete(
    _conn,
    _task_id,
    *,
    result=None,
    summary=None,
    metadata=None,
    expected_run_id=None,
):
    return bool(result or summary or metadata or expected_run_id)


def _new_complete(
    _conn,
    _task_id,
    *,
    result=None,
    summary=None,
    metadata=None,
    expected_run_id=None,
    replay_ended_run_id=None,
):
    return bool(result or summary or metadata or expected_run_id or replay_ended_run_id)


def _old_reclaim(_conn, _task_id, *, reason, signal_fn=None):
    return bool(reason or signal_fn)


def _new_reclaim(
    _conn,
    _task_id,
    *,
    reason,
    signal_fn=None,
    expected_run_id=None,
):
    return bool(reason or signal_fn or expected_run_id)


def _db(*, patched: bool):
    return SimpleNamespace(
        complete_task=_new_complete if patched else _old_complete,
        reclaim_task=_new_reclaim if patched else _old_reclaim,
    )


def _terminal_document(run_id: int = 7) -> dict:
    structured = _completed_result("accepted result")
    return {
        "board": "cassandra",
        "task_id": "t_compat",
        "expected_run_id": run_id,
        "result": json.dumps(structured, sort_keys=True),
        "summary": "accepted result",
        "metadata": {},
        "kanban_state": "pending",
        "recorded_at": lanes.utc_now(),
    }


def test_capability_detection_requires_explicit_safety_keywords():
    old = lanes.detect_kanban_capabilities(_db(patched=False))
    new = lanes.detect_kanban_capabilities(_db(patched=True))

    assert old.exact_run_completion is True
    assert old.ended_run_replay is False
    assert old.exact_run_reclaim is False
    assert old.ready is False
    assert old.deferred_features == ("ended-run-replay", "exact-run-reclaim")
    assert new.ready is True
    assert new.deferred_features == ()

    variadic = SimpleNamespace(
        complete_task=lambda *_args, **_kwargs: True,
        reclaim_task=lambda *_args, **_kwargs: True,
    )
    assert lanes.detect_kanban_capabilities(variadic).ready is False
    assert lanes._explicit_keyword(object(), "expected_run_id") is False


def test_deferred_features_name_every_absent_capability():
    """A fully legacy image defers all three safety features by exact name."""
    fully_legacy = lanes.KanbanCapabilities(False, False, False)

    assert fully_legacy.ready is False
    assert fully_legacy.deferred_features == (
        "exact-run-completion",
        "ended-run-replay",
        "exact-run-reclaim",
    )


def test_startup_health_moves_from_bounded_deferred_to_ready(
    tmp_path: Path,
    capsys,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    health = tmp_path / "runtime-health.json"
    old = lanes.initialize_kanban_capabilities(_db(patched=False), health_path=health)
    old_document = lanes.runtime_health(health)

    assert old.ready is False
    assert old_document["state"] == "deferred"
    assert old_document["schema_version"] == 1
    assert old_document["ready"] is False
    assert old_document["active_run_completion_safe"] is True
    assert old_document["deferred_features"] == [
        "ended-run-replay",
        "exact-run-reclaim",
    ]
    assert health.stat().st_size < 2048
    assert health.stat().st_mode & 0o777 == 0o600
    assert "compatibility deferred" in capsys.readouterr().err

    new_db = _db(patched=True)
    new = lanes.kanban_capabilities(new_db)
    lanes.initialize_kanban_capabilities(new_db, health_path=health)
    assert new.ready is True
    assert lanes.runtime_health(health)["state"] == "ready"


def test_readiness_fails_closed_without_disclosing_health_values(
    tmp_path: Path,
    capsys,
    monkeypatch,
):
    """Readiness distinguishes startup, malformed, deferred, stale, and ready."""
    health = tmp_path / "runtime-health.json"
    now = datetime(2026, 8, 17, 12, 0, tzinfo=timezone.utc)
    assert lanes.readiness_issue(health, now=now) == "runtime health is missing"

    health.write_text("not-json", encoding="utf-8")
    assert lanes.readiness_issue(health, now=now) == "runtime health is malformed"

    deferred = lanes._health_document(lanes.KanbanCapabilities(True, False, False))
    deferred["observed_at"] = now.isoformat()
    health.write_text(json.dumps(deferred), encoding="utf-8")
    assert lanes.readiness_issue(health, now=now) == "runtime compatibility is deferred"

    ready = lanes._health_document(lanes.KanbanCapabilities(True, True, True))
    ready["observed_at"] = (now - timedelta(seconds=61)).isoformat()
    health.write_text(json.dumps(ready), encoding="utf-8")
    assert lanes.readiness_issue(health, now=now) == "runtime health is stale"

    ready["observed_at"] = now.isoformat()
    health.write_text(json.dumps(ready), encoding="utf-8")
    assert lanes.readiness_issue(health, now=now) is None

    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path)
    monkeypatch.setenv("HERMES_CLI_HEALTH_MAX_AGE_SECONDS", "31536000")
    assert sys.modules["cli_lane_capabilities"].main() == 0
    assert capsys.readouterr().out == ""


@pytest.mark.parametrize(
    "mutation",
    [
        {"schema_version": 2},
        {"state": "ready", "ready": False},
        {"state": "ready", "capabilities": []},
        {"observed_at": 5},
        {"observed_at": "not-a-date"},
        {"observed_at": "2026-08-17T12:00:00"},
    ],
)
def test_readiness_rejects_inconsistent_documents(
    tmp_path: Path,
    mutation: dict,
):
    """Inconsistent ready documents remain not-ready."""
    now = datetime(2026, 8, 17, 12, 0, tzinfo=timezone.utc)
    document = lanes._health_document(lanes.KanbanCapabilities(True, True, True))
    document["observed_at"] = now.isoformat()
    document.update(mutation)
    health = tmp_path / "runtime-health.json"
    health.write_text(json.dumps(document), encoding="utf-8")

    assert lanes.readiness_issue(health, now=now) == "runtime health is malformed"


def test_readiness_command_fails_with_only_a_bounded_reason(
    tmp_path: Path,
    capsys,
    monkeypatch,
):
    """The probe command exits nonzero and discloses nothing but the reason."""
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path)

    assert sys.modules["cli_lane_capabilities"].main() == 1

    captured = capsys.readouterr()
    assert captured.out == ""
    assert captured.err.strip() == "runtime health is missing"


def test_capabilities_module_is_the_deployed_probe_entrypoint(
    tmp_path: Path,
    monkeypatch,
):
    """`python cli_lane_capabilities.py` runs the readiness probe directly."""
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path)

    with pytest.raises(SystemExit) as excinfo:
        runpy.run_path(
            str(SCRIPTS / "cli_lane_capabilities.py"),
            run_name="__main__",
        )

    assert excinfo.value.code == 1


def test_readiness_rejects_future_health(tmp_path: Path):
    """A clock-skewed future observation cannot make a stale process ready."""
    now = datetime(2026, 8, 17, 12, 0, tzinfo=timezone.utc)
    document = lanes._health_document(lanes.KanbanCapabilities(True, True, True))
    document["observed_at"] = (now + timedelta(seconds=6)).isoformat()
    health = tmp_path / "runtime-health.json"
    health.write_text(json.dumps(document), encoding="utf-8")

    assert lanes.readiness_issue(health, now=now) == "runtime health is stale"


@pytest.mark.parametrize(
    ("status", "run_id", "expected"),
    [("running", 7, "committed"), ("blocked", None, "deferred")],
)
def test_old_image_completes_only_the_exact_active_run(
    tmp_path: Path,
    monkeypatch,
    status: str,
    run_id: int | None,
    expected: str,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    task = SimpleNamespace(
        id="t_compat",
        status=status,
        current_run_id=run_id,
        completed_run_id=None,
        result=None,
        assignee="cli-auto",
    )
    calls = []

    def complete(
        _conn,
        _task_id,
        *,
        result=None,
        summary=None,
        metadata=None,
        expected_run_id=None,
    ):
        calls.append(expected_run_id)
        task.status = "done"
        task.current_run_id = None
        task.completed_run_id = expected_run_id
        task.result = result
        return True

    db = SimpleNamespace(
        scoped_current_board=lambda _board: nullcontext(),
        connect=lambda board: SimpleNamespace(close=lambda: None),
        get_task=lambda _conn, _task_id: task,
        complete_task=complete,
        reclaim_task=_old_reclaim,
    )
    identity = lanes.TerminalIdentity("cassandra", "t_compat", 7, "pending")

    outcome = lanes._finalize_document_db(db, identity, _terminal_document())

    assert outcome == expected
    assert calls == ([7] if status == "running" else [])
    if expected == "deferred":
        assert task.status == "blocked"


def test_old_image_preserves_pending_and_prepared_journals(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    task = SimpleNamespace(
        id="t_compat",
        status="blocked",
        current_run_id=None,
        completed_run_id=None,
        result=None,
        assignee="cli-auto",
    )
    db = SimpleNamespace(
        scoped_current_board=lambda _board: nullcontext(),
        connect=lambda board: SimpleNamespace(close=lambda: None),
        get_task=lambda _conn, _task_id: task,
        complete_task=_old_complete,
        reclaim_task=_old_reclaim,
    )
    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db))
    pending, _record = lanes._write_terminal_record(
        lanes.state_path("cassandra", "t_compat"),
        board="cassandra",
        task_id="t_compat",
        run_id=7,
        structured=_completed_result("accepted result"),
        summary="accepted result",
        metadata={},
    )

    assert lanes.recover_pending_finalizations() == 0
    assert pending.exists()
    assert len(list(pending.parent.glob("*.terminal.prepared-*.json"))) == 1
    assert task.status == "blocked"


def test_old_image_never_calls_unguarded_reclaim(tmp_path: Path, monkeypatch):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    task = SimpleNamespace(
        id="t_running",
        status="running",
        current_run_id=11,
        assignee="cli-auto",
    )
    calls = []

    def reclaim(_conn, _task_id, *, reason, signal_fn=None):
        calls.append((reason, signal_fn))
        return True

    db = SimpleNamespace(
        complete_task=_old_complete,
        reclaim_task=reclaim,
        list_boards=lambda include_archived=False: [{"slug": "cassandra"}],
        scoped_current_board=lambda _board: nullcontext(),
        connect=lambda board: SimpleNamespace(close=lambda: None),
        list_tasks=lambda _conn: [task],
    )
    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db))
    monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0)

    lanes.recover_orphans()

    assert calls == []


def test_main_detects_capabilities_before_startup_recovery(
    tmp_path: Path,
    monkeypatch,
):
    events = []
    db = _db(patched=False)
    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db))
    monkeypatch.setattr(lanes, "RESULT_SCHEMA_PATH", tmp_path / "result.schema.json")
    monkeypatch.setattr(
        lanes,
        "initialize_kanban_capabilities",
        lambda candidate: events.append(("detect", candidate))
        or lanes.KanbanCapabilities(True, False, False),
    )

    def stop_after_detection():
        events.append(("recover", db))
        raise RuntimeError("stop after startup boundary")

    monkeypatch.setattr(lanes, "recover_orphans", stop_after_detection)

    with pytest.raises(RuntimeError, match="startup boundary"):
        lanes.main()

    assert events == [("detect", db), ("recover", db)]