diff --git a/api/routers/tasks.py b/api/routers/tasks.py index b13eb06..7a54a5c 100644 --- a/api/routers/tasks.py +++ b/api/routers/tasks.py @@ -263,6 +263,33 @@ async def update_task( parent_workstream=ws, previous_task_status=previous_status, ) + # RMGR-WP-0002 dual-run: optional checkout writeback via repo-manager + if new_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 + + ws_full = ws or await session.get(Workplan, task.workplan_id) + repo = None + if ws_full and ws_full.repo_id: + repo = await session.get(ManagedRepo, ws_full.repo_id) + if repo is not None: + repo_path = resolve_repo_path(repo) + try_writeback_for_task( + repo_path=repo_path, + repo_slug=repo.slug, + task_id=str(task.id), + status=new_status, + reason="state-hub PATCH /tasks dual-run", + ) + except Exception: + # Dual-run must not break native DB update path + import logging + + logging.getLogger(__name__).exception( + "repo-manager dual-run writeback failed for task %s", task_id + ) await session.commit() await session.refresh(task) diff --git a/api/services/repo_manager_dual_run.py b/api/services/repo_manager_dual_run.py new file mode 100644 index 0000000..b6a8731 --- /dev/null +++ b/api/services/repo_manager_dual_run.py @@ -0,0 +1,240 @@ +"""State Hub dual-run adapter → Repo Manager (RMGR-WP-0002 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. + +Shared env vars (see repo-manager dual_run.py): + RM_WRITEBACK, RM_RECONCILE, RM_PILOT_REPOS, RM_METER_PATH, RMGR_BIN +""" + +from __future__ import annotations + +import json +import logging +import os +import subprocess +import sys +import uuid +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + + +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: + 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: 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]: + """Run rmgr CLI. Prefer PYTHONPATH to sibling repo-manager checkout.""" + 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 = False, + 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()) + 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 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: + 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=False, # SH fix-consistency push-seal remains authority for push in pilot + ) + 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"), + }, + ) + else: + logger.warning( + "RM dual-run writeback failed for %s task=%s: %s", + repo_slug, + task_id, + result.get("error") or result, + ) + return result diff --git a/scripts/consistency_check.py b/scripts/consistency_check.py index c530f2e..3bc4ace 100644 --- a/scripts/consistency_check.py +++ b/scripts/consistency_check.py @@ -2607,6 +2607,23 @@ def fix_repo( if any(i.check_id == "C-00" for i in report.failures): return report + # RMGR-WP-0002: optional repo-manager reconcile proxy (index rebuild) for pilots + try: + from api.services.repo_manager_dual_run import reconcile_for_repo, rm_reconcile + + if reconcile_for_repo(repo_slug) and report.repo_path: + rm_rec = rm_reconcile(repo_path=report.repo_path, repo_slug=repo_slug) + if rm_rec.get("ok") or rm_rec.get("exit_code") == 0: + report.fixes_applied.append( + f"RM dual-run reconcile ok: index={rm_rec.get('index_path') or 'written'}" + ) + else: + report.fixes_applied.append( + f"RM dual-run reconcile note: {rm_rec.get('error') or rm_rec}" + ) + except Exception as rec_exc: # noqa: BLE001 + report.fixes_applied.append(f"RM dual-run reconcile skipped: {rec_exc}") + # Auto-register this machine's path in host_paths so future runs work # without --repo-path. Idempotent: skipped when already correct. repo_path = report.repo_path @@ -3043,22 +3060,68 @@ def fix_repo( task_block_id = ctx["task_block_id"] db_status = ctx["db_status"] old_status = issue.file_value - if _patch_task_status_in_file(wp_file, task_block_id, db_status): - committed = _git_commit_writeback( - repo_path, - wp_file, - [f"{task_block_id}: {old_status} → {db_status}"], + # RMGR-WP-0002 dual-run: prefer repo-manager writeback when flagged + rm_done = False + try: + from api.services.repo_manager_dual_run import ( + try_writeback_for_task, + writeback_for_repo, ) - suffix = " (committed)" if committed else " (file patched, commit failed)" + + if writeback_for_repo(repo_slug) and repo_path: + # Prefer hub UUID when available for robust matching + task_ref = ctx.get("task_id") or task_block_id + rm_result = try_writeback_for_task( + repo_path=repo_path, + repo_slug=repo_slug, + task_id=str(task_ref), + status=db_status, + reason="state-hub C-15 dual-run writeback", + ) + if rm_result and rm_result.get("status") == "applied": + git_sha = (rm_result.get("evidence") or {}).get("git_sha") + suffix = f" (repo-manager git_sha={git_sha})" if git_sha else " (repo-manager)" + report.fixes_applied.append( + f"C-15 fixed: task '{task_block_id}' " + f"{old_status} → {db_status}{suffix}" + ) + rm_done = True + except Exception as rm_exc: # noqa: BLE001 report.fixes_applied.append( - f"C-15 fixed: task '{task_block_id}' " - f"{old_status} → {db_status}{suffix}" - ) - else: - report.fixes_applied.append( - f"C-15 SKIP: could not locate task block '{task_block_id}' " - f"in {wp_file.name}" + f"C-15 dual-run fallback to native: {rm_exc}" ) + if not rm_done: + if _patch_task_status_in_file(wp_file, task_block_id, db_status): + committed = _git_commit_writeback( + repo_path, + wp_file, + [f"{task_block_id}: {old_status} → {db_status}"], + ) + try: + from api.services.repo_manager_dual_run import record_mutation + + record_mutation( + source="state-hub", + kind="task_status_writeback_c15", + repo_slug=repo_slug, + detail={ + "task_block_id": task_block_id, + "status": db_status, + "committed": committed, + }, + ) + except Exception: + pass + suffix = " (committed)" if committed else " (file patched, commit failed)" + report.fixes_applied.append( + f"C-15 fixed: task '{task_block_id}' " + f"{old_status} → {db_status}{suffix}" + ) + else: + report.fixes_applied.append( + f"C-15 SKIP: could not locate task block '{task_block_id}' " + f"in {wp_file.name}" + ) except Exception as e: report.fixes_applied.append(f"{issue.check_id} ERROR: {e}") diff --git a/workplans/STATE-WP-0079-retirement-strangler.md b/workplans/STATE-WP-0079-retirement-strangler.md index 15bcd74..82af266 100644 --- a/workplans/STATE-WP-0079-retirement-strangler.md +++ b/workplans/STATE-WP-0079-retirement-strangler.md @@ -16,6 +16,7 @@ related: - SHR-INV-0001 - RMGR-WP-0001 - HUB-WP-0004 +state_hub_workstream_id: "749beac6-3c62-4284-aab5-9ed7fce900c2" --- # State Hub retirement strangler and disposition execution @@ -27,12 +28,23 @@ Execute the **keep/move/replace/retire** dispositions from without adding new permanent authorities here. End state: freeze window with no normal traffic, then archive. +## Dual-run handoff (RMGR-WP-0002) + +Repo Manager Stage B dual-run is available: + +- Flags: `RM_WRITEBACK`, `RM_RECONCILE`, `RM_PILOT_REPOS` (see `repo-manager/docs/dual-run.md`) +- SH adapter: `api/services/repo_manager_dual_run.py` (PATCH `/tasks` + C-15 writeback + reconcile proxy) +- Evidence: `repo-manager/docs/evidence/wp0002-completion.md` + +Expand pilot list here as cutover progresses; do not add new checkout mutators in State Hub. + ## Freeze policy for new scope ```task id: STATE-WP-0079-T01 status: todo priority: high +state_hub_task_id: "375c0b66-98b0-4d5b-98dd-4ad2883a4051" ``` Document and enforce: changes during retirement must preserve compatibility, @@ -45,6 +57,7 @@ permanent ownership (INTENT retirement status). id: STATE-WP-0079-T02 status: todo priority: high +state_hub_task_id: "595c3936-20b6-453f-92b5-84f86798054a" ``` Group SHR-INV-0001 items into cutover slices (repo/work → repo-manager; @@ -57,6 +70,7 @@ Per slice: adapter flag, owner API, rollback. id: STATE-WP-0079-T03 status: todo priority: medium +state_hub_task_id: "55716f12-fb12-4a1e-a770-076c490db111" ``` Adapt `statehub register` / scaffolding so `prj-` repos with `GOAL.md` and @@ -69,6 +83,7 @@ Adapt `statehub register` / scaffolding so `prj-` repos with `GOAL.md` and id: STATE-WP-0079-T04 status: todo priority: high +state_hub_task_id: "db5291f2-3801-40fa-abf2-ceb0e77687c9" ``` Implement strangler handoffs coordinated with RMGR-WP-0001 and HUB-WP-0004. @@ -80,6 +95,7 @@ Compatibility tests for each dispositioned route family. id: STATE-WP-0079-T05 status: todo priority: medium +state_hub_task_id: "02e508ed-3cde-4487-907e-d324a8a877d6" ``` Complete retirement of suggestions, workstream aliases, and other `retire` @@ -91,6 +107,7 @@ inventory items once meters/callers allow. Keep historical rows archive-readable id: STATE-WP-0079-T06 status: todo priority: high +state_hub_task_id: "d52c95c6-af3f-4b2c-804e-e07763c9a8ab" ``` With T06 gates: zero normal read/write window, final dump, backup/restore