feat(RMGR-WP-0003): production pilot dual-run config, bulk, push

Config file dual-run (env override), writeback_push, bulk-status facade
on State Hub, pilot example config, evidence and finished workplan.

src/repo_manager/dual_run.py

@@ -1,13 +1,15 @@
-"""Dual-run flags and checkout-mutation meter (RMGR-WP-0002 / ArchitectureBlueprint Stage B).
+"""Dual-run flags and checkout-mutation meter (RMGR-WP-0002/0003).
 
-Environment variables (shared with State Hub adapter):
+Precedence (highest first):
+  1. Environment variables when set
+  2. Config file (~/.repo-manager/dual-run.yaml or RM_DUAL_RUN_CONFIG)
+  3. Defaults (all off)
 
-  RM_WRITEBACK=1|true|yes   — file+git task writeback via repo-manager
-  RM_RECONCILE=1|true|yes   — reconcile path prefers repo-manager for pilot repos
-  RM_PILOT_REPOS=slug1,slug2 — if set, flags only apply to these slugs; empty = all
-  RM_METER_PATH=~/.repo-manager/mutation-meter.jsonl — append-only meter log
+Environment variables:
+  RM_WRITEBACK, RM_RECONCILE, RM_WRITEBACK_PUSH, RM_PILOT_REPOS, RM_METER_PATH,
+  RM_DUAL_RUN_CONFIG
 
-Rollback: unset flags or set to 0 → State Hub native path only.
+Rollback: writeback/reconcile false in config and unset env → State Hub native.
 """
 
 from __future__ import annotations
@@ -15,33 +17,106 @@ from __future__ import annotations
 import json
 import os
 from datetime import datetime, timezone
+from functools import lru_cache
 from pathlib import Path
 from typing import Any, Literal
 
+import yaml
+
 Source = Literal["state-hub", "repo-manager"]
 
+_TRUE = frozenset({"1", "true", "yes", "on"})
+_FALSE = frozenset({"0", "false", "no", "off"})
 
-def _truthy(name: str) -> bool:
-    return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"}
+
+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:
+    """Clear cached config (tests / SIGHUP-style)."""
+    _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, default: bool = False) -> bool:
+    env_map = {
+        "writeback": "RM_WRITEBACK",
+        "reconcile": "RM_RECONCILE",
+        "writeback_push": "RM_WRITEBACK_PUSH",
+    }
+    env_name = env_map.get(key)
+    if env_name:
+        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 _truthy("RM_WRITEBACK")
+    return _cfg_bool("writeback", False)
 
 
 def reconcile_enabled() -> bool:
-    return _truthy("RM_RECONCILE")
+    return _cfg_bool("reconcile", False)
+
+
+def writeback_push_enabled() -> bool:
+    return _cfg_bool("writeback_push", False)
 
 
 def pilot_slugs() -> set[str] | None:
-    """None means all repos; empty set after parse of blank list means none."""
+    """None = all repos; empty set = none."""
     raw = os.environ.get("RM_PILOT_REPOS")
-    if raw is None:
+    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
-    raw = raw.strip()
-    if not raw:
-        return set()
-    return {s.strip() for s in raw.split(",") if s.strip()}
+    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:
@@ -62,8 +137,13 @@ def reconcile_for_repo(slug: str | None) -> bool:
 
 
 def meter_path() -> Path:
-    raw = os.environ.get("RM_METER_PATH", "~/.repo-manager/mutation-meter.jsonl")
-    return Path(raw).expanduser()
+    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(
@@ -73,7 +153,6 @@ def record_mutation(
     repo_slug: str | None,
     detail: dict[str, Any] | None = None,
 ) -> None:
-    """Append one meter line. Best-effort; never raises to callers."""
     try:
         path = meter_path()
         path.parent.mkdir(parents=True, exist_ok=True)
@@ -112,9 +191,18 @@ def meter_summary(path: Path | None = None) -> dict[str, int]:
 
 def flags_status() -> dict[str, Any]:
     return {
+        "config_path": str(config_path()),
+        "config_exists": config_path().is_file(),
+        "writeback": writeback_enabled(),
+        "reconcile": reconcile_enabled(),
+        "writeback_push": writeback_push_enabled(),
+        "pilot_repos": sorted(pilot_slugs()) if pilot_slugs() is not None else None,
+        "meter_path": str(meter_path()),
+        "meter": meter_summary(),
+        # legacy env-style keys for operators
         "RM_WRITEBACK": writeback_enabled(),
         "RM_RECONCILE": reconcile_enabled(),
+        "RM_WRITEBACK_PUSH": writeback_push_enabled(),
         "RM_PILOT_REPOS": sorted(pilot_slugs()) if pilot_slugs() is not None else None,
         "RM_METER_PATH": str(meter_path()),
-        "meter": meter_summary(),
     }