From bd5f261402772f4cf0e31017a3eab06423782fbc Mon Sep 17 00:00:00 2001 From: jenkins Date: Sun, 13 Sep 2026 16:42:49 -0500 Subject: [PATCH] hermes: use native task links throughout PR follow-ups --- .../scripts/execution_pool_coordinator.py | 17 +++++- services/hermes/scripts/kanban_supervisor.py | 37 ++++++++---- ...st_hermes_execution_pool_coordinator_v2.py | 57 ++++++++++++++++++- .../tests/test_hermes_kanban_supervisor.py | 43 +++++++++++++- 4 files changed, 137 insertions(+), 17 deletions(-) diff --git a/services/hermes/scripts/execution_pool_coordinator.py b/services/hermes/scripts/execution_pool_coordinator.py index 5fb5e187..c2015edc 100644 --- a/services/hermes/scripts/execution_pool_coordinator.py +++ b/services/hermes/scripts/execution_pool_coordinator.py @@ -59,6 +59,21 @@ def _task_parents(task: Any) -> set[str]: return result +def _native_task_parents(kanban_db: Any, connection: Any, task: Any) -> set[str]: + """Read durable links through native Kanban when that API is available.""" + task_id = str(_task_value(task, "id", "") or "") + parent_ids = getattr(kanban_db, "parent_ids", None) + if callable(parent_ids) and task_id: + try: + value = parent_ids(connection, task_id) + except (OSError, ValueError) as error: + raise RuntimeError("native continuation parents are unavailable") from error + if not isinstance(value, (list, tuple, set)): + raise RuntimeError("native continuation parents are malformed") + return {str(item) for item in value if isinstance(item, (str, int))} + return _task_parents(task) + + def _submission_lineage( binding: dict[str, Any], assignment: Any, submission: Any ) -> tuple[supervisor_lineage.Lineage, str] | None: @@ -135,7 +150,7 @@ def assignment_payload( repo = re.fullmatch(r"https://scm\.bstein\.dev/titan/([A-Za-z0-9][A-Za-z0-9_.-]{0,99})\.git", repo_url) if repo is None or lineage.project != repo.group(1): raise RuntimeError("supervisor continuation project is invalid") - native_parents = _task_parents(task) + native_parents = _native_task_parents(kanban_db, connection, task) if parent not in native_parents or lineage.root_task_id not in native_parents: raise RuntimeError("supervisor continuation parents are invalid") if kanban_db.get_task(connection, lineage.root_task_id) is None: diff --git a/services/hermes/scripts/kanban_supervisor.py b/services/hermes/scripts/kanban_supervisor.py index 060ffa08..e155436e 100644 --- a/services/hermes/scripts/kanban_supervisor.py +++ b/services/hermes/scripts/kanban_supervisor.py @@ -178,11 +178,21 @@ def _comment(kanban_db: Any, conn: Any, task_id: str, body: str) -> None: _log(f"could not comment on {task_id}: {error}") -def _with_supervisor_state(board: str, task: Any) -> Any: - """Overlay coordinator-owned lineage on native task rows lacking metadata.""" +def _with_supervisor_state(kanban_db: Any, conn: Any, board: str, task: Any) -> Any: + """Overlay durable links and coordinator lineage on a native task row.""" task_id = str(getattr(task, "id", "") if not isinstance(task, dict) else task.get("id", "")) if not task_id: return task + values = dict(task) if isinstance(task, dict) else dict(vars(task)) + parent_ids = getattr(kanban_db, "parent_ids", None) + if callable(parent_ids): + try: + raw_parents = parent_ids(conn, task_id) + except Exception as error: # noqa: BLE001 - a missing link proof may duplicate work + raise RuntimeError("native supervisor parent links are unavailable") from error + if not isinstance(raw_parents, (list, tuple, set)): + raise RuntimeError("native supervisor parent links are malformed") + values["parents"] = [str(parent) for parent in raw_parents if isinstance(parent, (str, int))] try: root = supervisor_state.get_root(board, task_id) child = supervisor_state.get_child(board, task_id) @@ -192,8 +202,7 @@ def _with_supervisor_state(board: str, task: Any) -> Any: # a possible continuation as a new root task. raise RuntimeError("supervisor integrity state is unavailable") from error if root is None and child is None: - return task - values = dict(task) if isinstance(task, dict) else dict(vars(task)) + return values metadata = dict(values.get("metadata") or {}) if root is not None: metadata["supervisor_lineage"] = root.stamp_fields() @@ -289,7 +298,7 @@ def _escalate(kanban_db: Any, conn: Any, task_id: str, reason: str) -> None: _comment(kanban_db, conn, task_id, body) -def _already_created(kanban_db: Any, conn: Any, payload: dict[str, Any]) -> bool: +def _already_created(kanban_db: Any, conn: Any, payload: dict[str, Any], board: str = "") -> bool: """True if a card matching this spawn payload's stamp already exists.""" supervisor_stamp = (payload.get("metadata") or {}).get("supervisor") or {} kind = supervisor_stamp.get("kind") @@ -298,10 +307,13 @@ def _already_created(kanban_db: Any, conn: Any, payload: dict[str, Any]) -> bool if not kind or not root: return False try: - tasks = list(kanban_db.list_tasks(conn)) - except Exception as error: # noqa: BLE001 - if we cannot confirm, do not skip + tasks = [ + _with_supervisor_state(kanban_db, conn, board, task) + for task in kanban_db.list_tasks(conn) + ] + except Exception as error: # noqa: BLE001 - never retry an unverified create _log(f"could not re-scan before spawn retry: {error}") - return False + raise RuntimeError("cannot verify existing follow-up") from error return policy.existing_followup(tasks, kind, root, commit) @@ -327,7 +339,7 @@ def _spawn(kanban_db: Any, conn: Any, decision: policy.Decision, board: str = "" # raised at call binding, before any insert, but a post-insert TypeError # is also possible, so re-run the dedup scan first: only retry the create # when no matching card exists, so a partial insert is never doubled. - if _already_created(kanban_db, conn, decision.payload or {}): + if _already_created(kanban_db, conn, decision.payload or {}, board): return # Native Hermes has no task metadata column. The state table records # this stamp after creation; old compatible test/runtime APIs may still @@ -336,7 +348,7 @@ def _spawn(kanban_db: Any, conn: Any, decision: policy.Decision, board: str = "" try: child_id = kanban_db.create_task(conn, **payload) except TypeError: - if _already_created(kanban_db, conn, decision.payload or {}): + if _already_created(kanban_db, conn, decision.payload or {}, board): return payload.pop("idempotency_key", None) child_id = kanban_db.create_task(conn, **payload) @@ -410,7 +422,10 @@ def supervise_board( with kanban_db.scoped_current_board(board): conn = kanban_db.connect(board=board) try: - tasks = [_with_supervisor_state(board, task) for task in kanban_db.list_tasks(conn)] + tasks = [ + _with_supervisor_state(kanban_db, conn, board, task) + for task in kanban_db.list_tasks(conn) + ] for task in tasks: decision = policy.plan(task, tasks, limits) try: diff --git a/testing/tests/test_hermes_execution_pool_coordinator_v2.py b/testing/tests/test_hermes_execution_pool_coordinator_v2.py index d221a91c..328cc523 100644 --- a/testing/tests/test_hermes_execution_pool_coordinator_v2.py +++ b/testing/tests/test_hermes_execution_pool_coordinator_v2.py @@ -178,17 +178,70 @@ def test_assignment_payload_requires_state_and_native_parents_for_continuations( kanban = SimpleNamespace( build_worker_context=lambda *_args: "objective", get_task=lambda _connection, task_id: object() if task_id == "root" else None, + parent_ids=lambda _connection, task_id: ["root", "review"] + if task_id == "repair" else [], ) - accepted = task(id="repair", parents=("root", "review")) + accepted = task(id="repair") value = coordinator.assignment_payload(kanban, object(), accepted, "metis") assert value["repo_url"] == "https://scm.bstein.dev/titan/metis.git" assert value["branch"] == "wt/root" assert value["continuation_kind"] == "repair" - rejected = task(id="repair", parents=("root",)) + monkeypatch.setattr(kanban, "parent_ids", lambda *_args: ["root"]) + rejected = task(id="repair") with pytest.raises(RuntimeError, match="parents"): coordinator.assignment_payload(kanban, object(), rejected, "metis") +@pytest.mark.parametrize( + ("parent_ids", "message"), + [ + ( + lambda *_args: (_ for _ in ()).throw(OSError("board unavailable")), + "unavailable", + ), + (lambda *_args: "root", "malformed"), + ], +) +def test_continuation_parent_api_never_falls_back_after_an_error( + monkeypatch, parent_ids, message +): + lineage = coordinator.supervisor_lineage.Lineage( + "root", "wt/root", "https://scm.bstein.dev/titan/metis/pulls/7", "metis", "main" + ) + child = {"lineage": lineage, "parent_task_id": "root", "kind": "repair"} + monkeypatch.setattr(coordinator.supervisor_state, "get_child", lambda *_args: child) + monkeypatch.setattr(coordinator.supervisor_state, "get_root", lambda *_args: lineage) + kanban = SimpleNamespace( + build_worker_context=lambda *_args: "objective", + get_task=lambda *_args: object(), + parent_ids=parent_ids, + ) + + with pytest.raises(RuntimeError, match=message): + coordinator.assignment_payload( + kanban, object(), task(id="repair", parents=("root",)), "metis" + ) + + +def test_continuation_parent_attribute_fallback_requires_no_native_api(monkeypatch): + lineage = coordinator.supervisor_lineage.Lineage( + "root", "wt/root", "https://scm.bstein.dev/titan/metis/pulls/7", "metis", "main" + ) + child = {"lineage": lineage, "parent_task_id": "root", "kind": "repair"} + monkeypatch.setattr(coordinator.supervisor_state, "get_child", lambda *_args: child) + monkeypatch.setattr(coordinator.supervisor_state, "get_root", lambda *_args: lineage) + kanban = SimpleNamespace( + build_worker_context=lambda *_args: "objective", + get_task=lambda *_args: object(), + ) + + value = coordinator.assignment_payload( + kanban, object(), task(id="repair", parents=("root",)), "metis" + ) + + assert value["root_task_id"] == "root" + + def test_poll_uses_per_ordinal_key_for_empty_and_active_assignment(tmp_path): store = protocol.PoolStore(tmp_path / "pool.db") pool = coordinator.Coordinator(MASTER, store) diff --git a/testing/tests/test_hermes_kanban_supervisor.py b/testing/tests/test_hermes_kanban_supervisor.py index 63d20742..79c08038 100644 --- a/testing/tests/test_hermes_kanban_supervisor.py +++ b/testing/tests/test_hermes_kanban_supervisor.py @@ -36,7 +36,7 @@ def _isolate_ledger(tmp_path, monkeypatch): class RecordingDb: """Minimal in-memory kanban_db stub capturing every mutating call.""" - def __init__(self, tasks, boards=None, raise_on=None): + def __init__(self, tasks, boards=None, raise_on=None, parent_links=None): self._tasks = tasks self._boards = boards if boards is not None else [{"slug": "cassandra"}] self.created = [] @@ -44,6 +44,7 @@ class RecordingDb: self.blocked = [] self.metadata_sets = [] self._raise_on = raise_on or set() + self._parent_links = parent_links or {} def list_boards(self, include_archived=False): if "list_boards" in self._raise_on: @@ -61,6 +62,11 @@ class RecordingDb: def list_tasks(self, _conn): return list(self._tasks) + def parent_ids(self, _conn, task_id): + if "parent_ids" in self._raise_on: + raise RuntimeError("links unavailable") + return self._parent_links.get(task_id, []) + def create_task(self, _conn, **kwargs): if "create_task" in self._raise_on: raise RuntimeError("write failed") @@ -213,6 +219,36 @@ def test_impl_done_spawns_review_and_comments(): assert db.blocked == [] +def test_native_parent_links_hydrate_real_shaped_followups_before_planning(): + impl = _task( + id="impl", result={"changed_files": ["a.py"], "head_commit": "c1"}, + ) + chain = policy.lineage.Lineage( + "impl", "feature/impl", "https://scm.bstein.dev/titan/atlas-iac/pulls/1", "atlas-iac", "main" + ) + supervisor.supervisor_state.record_submission("cassandra", "impl", chain, "c1") + external = SimpleNamespace( + id="existing-review", status="ready", metadata={}, result=None, + title="Review c1", body="existing external review for c1", + ) + db = RecordingDb([impl, external], parent_links={"existing-review": ["impl"]}) + + assert supervisor.supervise_once(db, policy.Limits()) == 0 + assert db.created == [] + + +def test_native_parent_link_lookup_error_skips_the_board(capsys): + impl = _task( + id="impl", result={"changed_files": ["a.py"], "head_commit": "c1", "pull_request": "pr/1"}, + ) + _seed_root(impl) + db = RecordingDb([impl], raise_on={"parent_ids"}) + + assert supervisor.supervise_once(db, policy.Limits()) == 0 + assert db.created == [] + assert "parent links are unavailable" in capsys.readouterr().err + + def test_ship_marks_ready_for_human_without_merging(): review = _task( id="rev", @@ -482,7 +518,7 @@ def test_spawn_typeerror_does_not_double_insert_when_row_already_exists(): assert len(db.created) == 1 # retry suppressed by the re-scan guard -def test_already_created_returns_false_without_stamp_or_on_scan_error(): +def test_already_created_returns_false_without_stamp_and_fails_closed_on_scan_error(): db = RecordingDb([]) conn = SimpleNamespace(close=lambda: None) assert supervisor._already_created(db, conn, {"metadata": {}}) is False @@ -492,7 +528,8 @@ def test_already_created_returns_false_without_stamp_or_on_scan_error(): raise RuntimeError("scan down") payload = {"metadata": {"supervisor": {"kind": "review", "root": "impl", "head_commit": "c1"}}} - assert supervisor._already_created(BrokenScanDb([]), conn, payload) is False + with pytest.raises(RuntimeError, match="cannot verify existing follow-up"): + supervisor._already_created(BrokenScanDb([]), conn, payload) # --- fault isolation ------------------------------------------------------