atlas-iac/testing/tests/test_hermes_cli_matrix_capabilities.py

"""Legacy/patched image compatibility matrix for mixed Hermes rollouts."""

from __future__ import annotations

from contextlib import nullcontext
from pathlib import Path
from types import SimpleNamespace

import pytest

from testing.tests.test_hermes_cli_capabilities import (
    _new_reclaim,
    _old_reclaim,
    _terminal_document,
)
from testing.tests.test_hermes_cli_support import lanes

def _coordinator_db(*, patched: bool):
    """Model the image decomposition API used by the coordinator container."""
    if patched:

        def specify(
            _conn,
            _task_id,
            *,
            title=None,
            body=None,
            assignee=None,
            author=None,
            require_no_runs=False,
        ):
            return not require_no_runs

        def decompose(
            _conn,
            _task_id,
            *,
            root_assignee,
            children,
            author=None,
            auto_promote=True,
            require_no_runs=False,
        ):
            return None if require_no_runs else [f"{_task_id}.1"]

    else:

        def specify(
            _conn,
            _task_id,
            *,
            title=None,
            body=None,
            assignee=None,
            author=None,
        ):
            return True

        def decompose(
            _conn,
            _task_id,
            *,
            root_assignee,
            children,
            author=None,
            auto_promote=True,
        ):
            return [f"{_task_id}.1"]

    return SimpleNamespace(
        specify_triage_task=specify,
        decompose_triage_task=decompose,
    )


@pytest.mark.parametrize("worker_patched", [False, True])
@pytest.mark.parametrize("coordinator_patched", [False, True])
def test_mixed_image_rollout_compatibility_matrix(
    tmp_path: Path,
    monkeypatch,
    coordinator_patched: bool,
    worker_patched: bool,
):
    """Exact legacy/patched image matrix for decomposition and finalization.

    The hermes coordinator container and the cli-lane-runner worker container
    can run different image generations during a rollout while sharing one
    Kanban SQLite database and these mounted scripts.  The behaviors that
    differ between the legacy and the patched image on this path are:

    * legacy ``complete_task`` accepts ``expected_run_id`` but not
      ``replay_ended_run_id``: active exact runs complete on both images,
      ended-run replay is image-gated.
    * legacy ``reclaim_task`` lacks ``expected_run_id``: orphan reclaim is
      image-gated (covered by test_old_image_never_calls_unguarded_reclaim).
    * legacy ``specify_triage_task``/``decompose_triage_task`` lack
      ``require_no_runs``: only the patched coordinator refuses to redefine
      a task that already has execution history.

    Combination behavior proven here:

    * legacy coordinator + legacy worker: mid-run decomposition succeeds
      unguarded; the worker defers the ended run's accepted result (journal
      preserved for a patched replay) and reports deferred health.
    * legacy coordinator + patched worker: mid-run decomposition succeeds
      unguarded; the patched worker replays under ``replay_ended_run_id``,
      the database refuses the voided run, and the result converges to
      ``stale`` (conflict evidence) instead of completing a redefined task.
    * patched coordinator + legacy worker: decomposition is refused while
      the run exists, so the still-running exact run completes under
      ``expected_run_id``; replay/reclaim stay deferred and the readiness
      probe keeps the worker pod unready for new claims.
    * patched coordinator + patched worker: decomposition is refused while
      the run exists, the exact run completes, and the worker is ready.
    """
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    task = SimpleNamespace(
        id="t_matrix",
        status="running",
        current_run_id=7,
        completed_run_id=None,
        result=None,
        assignee="cli-auto",
    )

    def old_matrix_complete(
        _conn,
        _task_id,
        *,
        result=None,
        summary=None,
        metadata=None,
        expected_run_id=None,
    ):
        if task.status == "running" and task.current_run_id == expected_run_id:
            task.status = "done"
            task.completed_run_id = expected_run_id
            task.current_run_id = None
            task.result = result
            return True
        return False

    def new_matrix_complete(
        _conn,
        _task_id,
        *,
        result=None,
        summary=None,
        metadata=None,
        expected_run_id=None,
        replay_ended_run_id=None,
    ):
        if replay_ended_run_id is not None:
            # The decomposed run is no longer the latest ended run.
            return False
        return old_matrix_complete(
            _conn,
            _task_id,
            result=result,
            summary=summary,
            metadata=metadata,
            expected_run_id=expected_run_id,
        )

    worker_db = SimpleNamespace(
        scoped_current_board=lambda _board: nullcontext(),
        connect=lambda board: SimpleNamespace(close=lambda: None),
        get_task=lambda _conn, _task_id: task,
        complete_task=new_matrix_complete if worker_patched else old_matrix_complete,
        reclaim_task=_new_reclaim if worker_patched else _old_reclaim,
    )
    coordinator_db = _coordinator_db(patched=coordinator_patched)

    assert (
        lanes._explicit_keyword(
            coordinator_db.decompose_triage_task,
            "require_no_runs",
        )
        is coordinator_patched
    )
    assert (
        lanes._explicit_keyword(
            coordinator_db.specify_triage_task,
            "require_no_runs",
        )
        is coordinator_patched
    )
    health = tmp_path / "runtime-health.json"
    capabilities = lanes.initialize_kanban_capabilities(
        worker_db,
        health_path=health,
    )
    assert capabilities.ready is worker_patched
    assert capabilities.exact_run_completion is True
    if worker_patched:
        assert lanes.readiness_issue(health) is None
    else:
        assert capabilities.deferred_features == (
            "ended-run-replay",
            "exact-run-reclaim",
        )
        assert (
            lanes.readiness_issue(health) == "runtime compatibility is deferred"
        )

    if coordinator_patched:
        refused = coordinator_db.decompose_triage_task(
            object(),
            "t_matrix",
            root_assignee="cli-auto",
            children=[{"title": "split"}],
            require_no_runs=True,
        )
        assert refused is None
    else:
        created = coordinator_db.decompose_triage_task(
            object(),
            "t_matrix",
            root_assignee="cli-auto",
            children=[{"title": "split"}],
        )
        assert created == ["t_matrix.1"]
        task.status = "triage"
        task.current_run_id = None

    identity = lanes.TerminalIdentity("cassandra", "t_matrix", 7, "pending")
    outcome = lanes._finalize_document_db(worker_db, identity, _terminal_document())

    if coordinator_patched:
        assert outcome == "committed"
        assert task.status == "done"
        assert task.completed_run_id == 7
    elif worker_patched:
        assert outcome == "stale"
        assert task.status == "triage"
    else:
        assert outcome == "deferred"
        assert task.status == "triage"