feat(RMGR-WP-0002): dual-run writeback, flags, meter, SH facade

Add dual-run flags/meter, harden task-status (idempotency, UUID, head,
push), State Hub adapter for PATCH /tasks and C-15/reconcile proxy, pilot
evidence, and finish RMGR-WP-0002.

src/repo_manager/dual_run.py

@@ -0,0 +1,120 @@
+"""Dual-run flags and checkout-mutation meter (RMGR-WP-0002 / ArchitectureBlueprint Stage B).
+
+Environment variables (shared with State Hub adapter):
+
+  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
+
+Rollback: unset flags or set to 0 → State Hub native path only.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Literal
+
+Source = Literal["state-hub", "repo-manager"]
+
+
+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:
+    """None means all repos; empty set after parse of blank list means 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: Source,
+    kind: str,
+    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)
+        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 meter_summary(path: Path | None = None) -> dict[str, int]:
+    path = path or meter_path()
+    counts: dict[str, int] = {"state-hub": 0, "repo-manager": 0, "total": 0}
+    if not path.is_file():
+        return counts
+    for line in path.read_text(encoding="utf-8").splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            row = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+        src = row.get("source")
+        if src in counts:
+            counts[src] += 1
+        counts["total"] += 1
+    return counts
+
+
+def flags_status() -> dict[str, Any]:
+    return {
+        "RM_WRITEBACK": writeback_enabled(),
+        "RM_RECONCILE": reconcile_enabled(),
+        "RM_PILOT_REPOS": sorted(pilot_slugs()) if pilot_slugs() is not None else None,
+        "RM_METER_PATH": str(meter_path()),
+        "meter": meter_summary(),
+    }