feat: dual-run config file, bulk writeback, writeback_push

RMGR-WP-0003: load ~/.repo-manager/dual-run.yaml; bulk-status-sync uses
RM writeback; optional push after writeback when configured.

api/routers/tasks.py

@@ -169,6 +169,8 @@ async def bulk_status_sync(
     updated: list[Task] = []
     events: list[ProgressEvent] = []
     author = body.author or "custodian"
+    # Cache repo resolution for dual-run writeback (RMGR-WP-0003)
+    repo_cache: dict = {}
     for update in body.updates:
         task = tasks_by_id[update.task_id]
         previous_status = status_value(task.status)
@@ -182,6 +184,32 @@ async def bulk_status_sync(
             parent_workstream=ws,
             previous_task_status=previous_status,
         )
+        if target_status != previous_status:
+            try:
+                from api.models.managed_repo import ManagedRepo
+                from api.services.repo_manager_dual_run import try_writeback_for_task
+                from api.services.workplan_files import resolve_repo_path
+
+                repo_id = ws.repo_id if ws else None
+                if repo_id is not None:
+                    if repo_id not in repo_cache:
+                        repo_cache[repo_id] = await session.get(ManagedRepo, repo_id)
+                    repo = repo_cache[repo_id]
+                    if repo is not None:
+                        try_writeback_for_task(
+                            repo_path=resolve_repo_path(repo),
+                            repo_slug=repo.slug,
+                            task_id=str(task.id),
+                            status=target_status,
+                            reason="state-hub bulk-status-sync dual-run",
+                        )
+            except Exception:
+                import logging
+
+                logging.getLogger(__name__).exception(
+                    "repo-manager dual-run bulk writeback failed for task %s",
+                    update.task_id,
+                )
         event = ProgressEvent(
             task_id=task.id,
             workplan_id=task.workplan_id,

api/services/repo_manager_dual_run.py

@@ -1,11 +1,11 @@
-"""State Hub dual-run adapter → Repo Manager (RMGR-WP-0002 Stage B).
+"""State Hub dual-run adapter → Repo Manager (RMGR-WP-0002/0003 Stage B).
 
-When RM_WRITEBACK / RM_RECONCILE are enabled (and optional RM_PILOT_REPOS matches),
+Config file (same as repo-manager): ~/.repo-manager/dual-run.yaml
-State Hub delegates checkout mutation to the ``rmgr`` CLI instead of native
+or RM_DUAL_RUN_CONFIG. Env overrides when set:
-file writeback. Rollback: unset the env flags.
 
-Shared env vars (see repo-manager dual_run.py):
+  RM_WRITEBACK, RM_RECONCILE, RM_WRITEBACK_PUSH, RM_PILOT_REPOS, RM_METER_PATH
-  RM_WRITEBACK, RM_RECONCILE, RM_PILOT_REPOS, RM_METER_PATH, RMGR_BIN
+
+Rollback: disable in config / unset env → native State Hub path.
 """
 
 from __future__ import annotations
@@ -16,32 +16,97 @@ 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 _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:
+    _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 _truthy("RM_WRITEBACK")
+    return _cfg_bool("writeback", "RM_WRITEBACK", False)
 
 
 def reconcile_enabled() -> bool:
-    return _truthy("RM_RECONCILE")
+    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 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()
+    val = cfg["pilot_repos"]
-    if not raw:
+    if val is None:
-        return set()
+        return None
-    return {s.strip() for s in raw.split(",") if s.strip()}
+    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 +127,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")
+    raw = os.environ.get("RM_METER_PATH")
-    return Path(raw).expanduser()
+    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(
@@ -92,7 +162,6 @@ def record_mutation(
 
 
 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:
@@ -133,11 +202,12 @@ def rm_update_task_status(
     status: str,
     repo_slug: str | None = None,
     reason: str = "state-hub dual-run",
-    push: bool = False,
+    push: bool | None = None,
     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())
+    if push is None:
+        push = writeback_push_enabled()
     args = [
         "update-task-status",
         "--path",
@@ -200,7 +270,6 @@ def try_writeback_for_task(
     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:
@@ -217,7 +286,7 @@ def try_writeback_for_task(
         status=status,
         repo_slug=repo_slug,
         reason=reason,
-        push=False,  # SH fix-consistency push-seal remains authority for push in pilot
+        push=writeback_push_enabled(),
     )
     if result.get("status") == "applied":
         record_mutation(
@@ -228,6 +297,7 @@ def try_writeback_for_task(
                 "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: