"""State Hub dual-run adapter → Repo Manager (RMGR-WP-0002 Stage B). When RM_WRITEBACK / RM_RECONCILE are enabled (and optional RM_PILOT_REPOS matches), State Hub delegates checkout mutation to the ``rmgr`` CLI instead of native file writeback. Rollback: unset the env flags. Shared env vars (see repo-manager dual_run.py): RM_WRITEBACK, RM_RECONCILE, RM_PILOT_REPOS, RM_METER_PATH, RMGR_BIN """ from __future__ import annotations import json import logging import os import subprocess import sys import uuid from pathlib import Path from typing import Any logger = logging.getLogger(__name__) def _truthy(name: str) -> bool: return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"} def writeback_enabled() -> bool: return _truthy("RM_WRITEBACK") def reconcile_enabled() -> bool: return _truthy("RM_RECONCILE") def pilot_slugs() -> set[str] | None: raw = os.environ.get("RM_PILOT_REPOS") if raw is None: return None raw = raw.strip() if not raw: return set() return {s.strip() for s in raw.split(",") if s.strip()} def slug_allowed(slug: str | None) -> bool: if not slug: return False pilots = pilot_slugs() if pilots is None: return True return slug in pilots def writeback_for_repo(slug: str | None) -> bool: return writeback_enabled() and slug_allowed(slug) def reconcile_for_repo(slug: str | None) -> bool: return reconcile_enabled() and slug_allowed(slug) def meter_path() -> Path: raw = os.environ.get("RM_METER_PATH", "~/.repo-manager/mutation-meter.jsonl") return Path(raw).expanduser() def record_mutation( *, source: str, kind: str, repo_slug: str | None, detail: dict[str, Any] | None = None, ) -> None: try: from datetime import datetime, timezone path = meter_path() path.parent.mkdir(parents=True, exist_ok=True) row = { "ts": datetime.now(timezone.utc).isoformat(), "source": source, "kind": kind, "repo_slug": repo_slug, "detail": detail or {}, } with path.open("a", encoding="utf-8") as fh: fh.write(json.dumps(row, default=str) + "\n") except OSError: pass def run_rmgr(args: list[str], *, timeout: int = 120) -> tuple[int, str, str]: """Run rmgr CLI. Prefer PYTHONPATH to sibling repo-manager checkout.""" env = os.environ.copy() custom = os.environ.get("RMGR_BIN", "").strip() if custom: cmd = custom.split() + args else: src = os.environ.get("REPO_MANAGER_SRC", "").strip() sibling = Path(src) if src else (Path.home() / "repo-manager" / "src") if not sibling.is_dir() and (Path.home() / "repo-manager" / "src").is_dir(): sibling = Path.home() / "repo-manager" / "src" if sibling.is_dir(): env["PYTHONPATH"] = str(sibling) + ( os.pathsep + env["PYTHONPATH"] if env.get("PYTHONPATH") else "" ) cmd = [ sys.executable, "-c", "from repo_manager.cli import main; import sys; raise SystemExit(main(sys.argv[1:]))", *args, ] else: cmd = ["rmgr", *args] proc = subprocess.run( cmd, capture_output=True, text=True, timeout=timeout, check=False, env=env, ) return proc.returncode, proc.stdout, proc.stderr def rm_update_task_status( *, repo_path: str | Path, task_id: str, status: str, repo_slug: str | None = None, reason: str = "state-hub dual-run", push: bool = False, correlation_id: str | None = None, ) -> dict[str, Any]: """Invoke repo-manager task status writeback. Returns parsed command result dict.""" correlation_id = correlation_id or str(uuid.uuid4()) args = [ "update-task-status", "--path", str(repo_path), "--task-id", str(task_id), "--status", status, "--reason", reason, "--correlation-id", correlation_id, "--idempotency-key", f"sh-dual-{task_id}-{status}-{correlation_id}", ] if repo_slug: args.extend(["--slug", repo_slug]) if push: args.append("--push") code, out, err = run_rmgr(args) try: result = json.loads(out.strip() or "{}") except json.JSONDecodeError: result = { "status": "failed", "error": { "code": "internal", "message": f"rmgr non-json exit={code} stderr={err!r} stdout={out[:500]!r}", }, "correlation_id": correlation_id, } if code != 0 and result.get("status") not in ("applied", "rejected"): result.setdefault("status", "failed") result.setdefault( "error", {"code": "internal", "message": f"rmgr exit={code} stderr={err!r}"}, ) return result def rm_reconcile(*, repo_path: str | Path, repo_slug: str | None = None) -> dict[str, Any]: args = ["reconcile", "--path", str(repo_path)] if repo_slug: args.extend(["--slug", repo_slug]) code, out, err = run_rmgr(args) try: result = json.loads(out.strip() or "{}") except json.JSONDecodeError: result = {"ok": False, "error": err or out, "exit_code": code} result["exit_code"] = code return result def try_writeback_for_task( *, repo_path: str | Path | None, repo_slug: str | None, task_id: str, status: str, reason: str = "state-hub dual-run PATCH /tasks", ) -> dict[str, Any] | None: """If dual-run writeback applies, run RM and return result; else None (use native SH).""" if not writeback_for_repo(repo_slug): return None if not repo_path: logger.warning("RM dual-run writeback skipped: no repo_path for %s", repo_slug) return None path = Path(repo_path) if not path.is_dir(): logger.warning("RM dual-run writeback skipped: missing path %s", path) return None result = rm_update_task_status( repo_path=path, task_id=task_id, status=status, repo_slug=repo_slug, reason=reason, push=False, # SH fix-consistency push-seal remains authority for push in pilot ) if result.get("status") == "applied": record_mutation( source="repo-manager", kind="task_status_writeback_via_sh_facade", repo_slug=repo_slug, detail={ "task_id": task_id, "status": status, "git_sha": (result.get("evidence") or {}).get("git_sha"), }, ) else: logger.warning( "RM dual-run writeback failed for %s task=%s: %s", repo_slug, task_id, result.get("error") or result, ) return result