"""State Hub dual-run adapter → Repo Manager (RMGR-WP-0002/0003 Stage B). Config file (same as repo-manager): ~/.repo-manager/dual-run.yaml or RM_DUAL_RUN_CONFIG. Env overrides when set: RM_WRITEBACK, RM_RECONCILE, RM_WRITEBACK_PUSH, RM_PILOT_REPOS, RM_METER_PATH Rollback: disable in config / unset env → native State Hub path. """ from __future__ import annotations import json import logging import os import subprocess import sys import uuid from functools import lru_cache from pathlib import Path from typing import Any import yaml logger = logging.getLogger(__name__) _TRUE = frozenset({"1", "true", "yes", "on"}) _FALSE = frozenset({"0", "false", "no", "off"}) def config_path() -> Path: raw = os.environ.get("RM_DUAL_RUN_CONFIG", "~/.repo-manager/dual-run.yaml") return Path(raw).expanduser() @lru_cache(maxsize=1) def _load_file_config() -> dict[str, Any]: path = config_path() if not path.is_file(): return {} try: data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} except (OSError, yaml.YAMLError): return {} return data if isinstance(data, dict) else {} def reload_config() -> None: _load_file_config.cache_clear() def _env_bool(name: str) -> bool | None: raw = os.environ.get(name) if raw is None or raw.strip() == "": return None v = raw.strip().lower() if v in _TRUE: return True if v in _FALSE: return False return None def _cfg_bool(key: str, env_name: str, default: bool = False) -> bool: ev = _env_bool(env_name) if ev is not None: return ev cfg = _load_file_config() if key in cfg: val = cfg[key] if isinstance(val, bool): return val if isinstance(val, str): return val.strip().lower() in _TRUE return default def writeback_enabled() -> bool: return _cfg_bool("writeback", "RM_WRITEBACK", False) def reconcile_enabled() -> bool: return _cfg_bool("reconcile", "RM_RECONCILE", False) def writeback_push_enabled() -> bool: return _cfg_bool("writeback_push", "RM_WRITEBACK_PUSH", False) def pilot_slugs() -> set[str] | None: raw = os.environ.get("RM_PILOT_REPOS") if raw is not None: raw = raw.strip() if not raw: return set() return {s.strip() for s in raw.split(",") if s.strip()} cfg = _load_file_config() if "pilot_repos" not in cfg: return None val = cfg["pilot_repos"] if val is None: return None if isinstance(val, str): if not val.strip(): return set() return {s.strip() for s in val.split(",") if s.strip()} if isinstance(val, list): return {str(s).strip() for s in val if str(s).strip()} return None 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") if raw: return Path(raw).expanduser() cfg = _load_file_config() if cfg.get("meter_path"): return Path(str(cfg["meter_path"])).expanduser() return Path("~/.repo-manager/mutation-meter.jsonl").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]: 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 | None = None, correlation_id: str | None = None, ) -> dict[str, Any]: correlation_id = correlation_id or str(uuid.uuid4()) if push is None: push = writeback_push_enabled() 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 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=writeback_push_enabled(), ) 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"), "push_ok": (result.get("evidence") or {}).get("push_ok"), }, ) else: logger.warning( "RM dual-run writeback failed for %s task=%s: %s", repo_slug, task_id, result.get("error") or result, ) return result