diff --git a/api/routers/tasks.py b/api/routers/tasks.py index 7a54a5c..b76746c 100644 --- a/api/routers/tasks.py +++ b/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, diff --git a/api/services/repo_manager_dual_run.py b/api/services/repo_manager_dual_run.py index b6a8731..3af5e40 100644 --- a/api/services/repo_manager_dual_run.py +++ b/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), -State Hub delegates checkout mutation to the ``rmgr`` CLI instead of native -file writeback. Rollback: unset the env flags. +Config file (same as repo-manager): ~/.repo-manager/dual-run.yaml +or RM_DUAL_RUN_CONFIG. Env overrides when set: -Shared env vars (see repo-manager dual_run.py): - RM_WRITEBACK, RM_RECONCILE, RM_PILOT_REPOS, RM_METER_PATH, RMGR_BIN + 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 @@ -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() - 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 +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") - 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( @@ -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: