feat(consistency): consume repo-manager conformance
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Successful in 1s

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
This commit is contained in:
tegwick 2026-08-21 22:34:24 +02:00
parent ec780c05a2
commit b9d9ffed5f
2 changed files with 93 additions and 0 deletions

View file

@ -38,6 +38,7 @@ Checks:
C-32 work-record-not-indexed WARN Yes kind: intake/decision YAML block has no hub id not indexed in DB (registration, CUST-WP-0061-T02)
C-33 work-record-index-stale WARN Yes WORK-RECORDS.md missing or stale generated per-repo index (CUST-WP-0061-T04)
C-34 quality-dor-ready WARN No status=ready without quality_dor DoR-Ok (STATE-WP-0077 soft)
C-35 repo-manager-conformance WARN No Repo Manager flavor/standards contract reports findings
(finished¬DoD-Ok is listed by `statehub quality-debt`, not per-file C-warn avoids historical flood)
Usage:
@ -73,6 +74,7 @@ import argparse
import os
import json
import re
import shutil
import socket
import subprocess
import sys
@ -164,6 +166,56 @@ _WORK_REQUEST_RE = re.compile(
)
_OPEN_WORKPLAN_STATUSES = {"proposed", "ready", "active", "blocked", "backlog"}
def _check_repo_manager_conformance(repo_dir: Path, report: "ConsistencyReport") -> None:
"""C-35: consume Repo Manager's conformance contract without reimplementing it."""
configured = os.environ.get("RMGR_BIN", "").strip()
candidates = [configured] if configured else []
candidates.extend(
[
str(Path.home() / "repo-manager" / ".venv" / "bin" / "rmgr"),
shutil.which("rmgr") or "",
]
)
executable = next((item for item in candidates if item and Path(item).is_file()), None)
if executable is None:
report.add(
severity="INFO",
check_id="C-35",
message="Repo Manager conformance adapter unavailable; set RMGR_BIN or install rmgr",
fixable=False,
)
return
try:
completed = subprocess.run(
[executable, "conform", "--path", str(repo_dir)],
capture_output=True,
text=True,
check=False,
timeout=30,
)
payload = json.loads(completed.stdout)
except (OSError, subprocess.TimeoutExpired, json.JSONDecodeError) as exc:
report.add(
severity="INFO",
check_id="C-35",
message=f"Repo Manager conformance adapter could not run: {exc}",
fixable=False,
)
return
for finding in payload.get("findings") or []:
severity = str(finding.get("severity") or "warning")
report.add(
severity="WARN" if severity in {"missing", "contradictory", "warning"} else "INFO",
check_id="C-35",
message=(
f"Repo Manager {severity}: {finding.get('path') or '(repository)'}: "
f"{finding.get('message') or finding.get('code') or 'conformance finding'}"
),
file_path=finding.get("path"),
fixable=False,
)
# Legacy file/API aliases translated before comparison and PATCHing.
FILE_TO_DB_WORKSTREAM_STATUS: dict[str, str] = dict(LEGACY_WORKSTREAM_STATUS_ALIASES)
@ -1158,6 +1210,9 @@ def check_repo(api_base: str, repo_slug: str, repo_path_override: str | None = N
fixable=False,
)
# C-35: consume the repository-standard answer from Repo Manager.
_check_repo_manager_conformance(repo_dir, report)
# C-31: work-record sidetrack detector (canon work-record-types registry)
_check_unregistered_work_records(repo_dir, report)