#!/usr/bin/env python3 """Queue a trusted same-PR Hermes continuation from an existing root card.""" from __future__ import annotations import argparse import json import re import sys from typing import Any import scm_broker_client import supervisor_state BOARD = re.compile(r"[a-z0-9][a-z0-9-]{0,63}\Z") TASK = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}\Z") PR_URL = re.compile( r"https://scm\.bstein\.dev/titan/(?P[A-Za-z0-9][A-Za-z0-9_.-]{0,99})/pulls/(?P[1-9][0-9]{0,9})\Z" ) MAX_OBJECTIVE = 8_000 class ContinueError(ValueError): """Reject a continuation that cannot be tied to durable coordinator state.""" def _text(value: Any) -> str: return value.strip() if isinstance(value, str) else "" def _validated_args(board: str, root_task: str, objective: str) -> tuple[str, str, str]: """Keep the CLI surface limited to a board, root ID, and user objective.""" board, root_task, objective = _text(board), _text(root_task), _text(objective) if not BOARD.fullmatch(board) or not TASK.fullmatch(root_task): raise ContinueError("board or root task ID is invalid") if not objective or len(objective) > MAX_OBJECTIVE or "\x00" in objective: raise ContinueError("objective must be non-empty and at most 8000 characters") return board, root_task, objective def _live_head(lineage: Any) -> str: """Prove the exact open, same-repository PR that the root recorded.""" match = PR_URL.fullmatch(lineage.pull_request) if match is None or match.group("project") != lineage.project: raise ContinueError("root lineage does not carry a canonical pull request") try: document = json.loads(scm_broker_client.read( f"/api/v1/repos/titan/{lineage.project}/pulls/{match.group('number')}" )) except Exception as error: # noqa: BLE001 - read proof must fail closed raise ContinueError(f"live pull-request proof is unavailable: {type(error).__name__}") from error if not isinstance(document, dict): raise ContinueError("live pull-request proof is malformed") head, base = document.get("head"), document.get("base") full_name = f"titan/{lineage.project}" valid = ( document.get("state") == "open" and isinstance(head, dict) and isinstance(base, dict) and _text(head.get("ref")) == lineage.branch and isinstance(head.get("repo"), dict) and head["repo"].get("full_name") == full_name and _text(base.get("ref")) == lineage.base_branch and isinstance(base.get("repo"), dict) and base["repo"].get("full_name") == full_name ) sha = _text(head.get("sha")) if isinstance(head, dict) else "" if not valid or not re.fullmatch(r"[0-9a-fA-F]{7,64}", sha): raise ContinueError("live pull request no longer matches trusted branch, base, and repository") return sha def _body(root_task: str, lineage: Any, head: str, objective: str) -> str: """Retain human-visible PR/root context; authority remains in supervisor state.""" return ( "Hermes-Task-Role: repair\n\n" f"Trusted continuation of root task {root_task} on {lineage.pull_request}.\n" f"Verified current PR head: {head}\n\n" "Use the coordinator-issued private workspace from the latest remote branch. " "Keep this existing PR and branch; do not create a branch or PR. Inspect the " "current diff, implement and verify the requested repair, then submit through " "the signed mediator path.\n\n" f"Objective:\n{objective}\n" ) def queue( kanban_db: Any, *, board: str, root_task: str, objective: str ) -> tuple[str, bool]: """Create one idempotent root-parent repair card using only trusted lineage.""" board, root_task, objective = _validated_args(board, root_task, objective) with kanban_db.scoped_current_board(board): conn = kanban_db.connect(board=board) try: root = kanban_db.get_task(conn, root_task) if root is None: raise ContinueError("root task does not exist on the requested board") lineage = supervisor_state.get_root(board, root_task) if lineage is None or lineage.root_task_id != root_task: raise ContinueError("root task has no coordinator-issued PR lineage") head = _live_head(lineage) existing = supervisor_state.existing_child(board, root_task, head, objective) if existing: return existing, False # A live read proved this is the current same PR. It invalidates any # prior approval before the new child can be dispatched. supervisor_state.record_live_head(board, root_task, head) key = f"supervisor:continue:{root_task}:{head}:{supervisor_state.objective_digest(objective)}" child_id = kanban_db.create_task( conn, title=f"Continue PR {lineage.pull_request.rsplit('/', 1)[-1]}: {objective[:96]}", body=_body(root_task, lineage, head, objective), assignee="cli-auto", created_by="hermes-supervisor", parents=[root_task], idempotency_key=key, # Native Hermes derives ready/todo from the root parent; its # only non-blocked initial status is the explicit running mode. initial_status="running", ) if not isinstance(child_id, str) or not child_id: raise ContinueError("Kanban did not return a continuation task ID") supervisor_state.record_child( board, child_id, root_task, root_task, "repair", head, objective ) supervisor_state.clear_ready(board, root_task) try: kanban_db.add_comment( conn, root_task, "hermes-supervisor", f"supervisor: queued trusted continuation {child_id} for current PR head {head}.", ) except Exception: pass return child_id, True finally: conn.close() def main(argv: list[str] | None = None) -> int: """Parse a minimal local CLI then queue a continuation or print its ID.""" parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--board", required=True) parser.add_argument("--root-task", required=True) parser.add_argument("--objective", required=True) args = parser.parse_args(argv) try: from hermes_cli import kanban_db child, created = queue( kanban_db, board=args.board, root_task=args.root_task, objective=args.objective ) except ContinueError as error: print(f"kanban continuation rejected: {error}", file=sys.stderr) return 2 print(json.dumps({"task_id": child, "created": created}, sort_keys=True)) return 0 if __name__ == "__main__": # pragma: no cover - process entry point raise SystemExit(main())