239 lines
7.4 KiB
Python
239 lines
7.4 KiB
Python
"""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"
|