state-hub/scripts/consistency_check.py

#!/usr/bin/env python3
"""consistency_check.py — ADR-001 consistency checking engine.

Runs bidirectional checks between workplan files in a registered repo and the
state-hub DB. The file is always authoritative; the DB is the cache/index layer.

Checks:
  C-01  workplans-dir            FAIL  No   workplans/ directory missing
  C-02  workplan-parse           FAIL  No   Workplan file cannot be parsed
  C-03  workstream-stale-ref     FAIL  No   state_hub_workstream_id in file not in DB
  C-04  workstream-status-drift  WARN  Yes  File status != DB status (file wins)
  C-05  workstream-title-drift   WARN  Yes  File title != DB title (file wins)
  C-06  workstream-unlinked      WARN  Yes  Workplan has no state_hub_workstream_id
        --fix derives the UUID from the canonical record id and writes it back
  C-07  orphan-db-active         FAIL  No   Active DB workstream, no backing file
  C-08  orphan-db-closed         INFO  No   Finished/archived DB workstream, no file
  C-09  workstream-repo-mismatch FAIL  Yes  DB workstream repo_id != file location
  C-10  task-status-drift        WARN  Yes  Task status differs between file and DB
  C-11  task-unlinked            WARN  Yes  Task block has no state_hub_task_id
  C-12  orphan-db-task           WARN  Yes  DB task in workstream has no file backing unless terminal in a closed workstream
  C-13  workstream-auto-complete WARN  Yes  All DB tasks done but workstream still active
  C-14  ghost-duplicate          WARN  No   Active topic workstream with no repo_id matches a file-backed title — probable ghost from premature create_workstream() call
  C-15  task-db-ahead            WARN  No   DB task is terminal and file is not — report only; file wins (ADR-001)
  C-16  repo-behind-remote       WARN  No   Local repo is behind remote tracking branch — --fix skipped to avoid clobbering remote progress
  C-17  repo-ahead-push-failed   WARN  No   Local repo has unpushed commits and push failed — writes skipped to prevent runaway divergence
  C-19  workstream-planning-drift WARN Yes  planning_priority/planning_order differs between file and DB
  C-20  workstream-dependency-missing WARN Yes Workplan dependency frontmatter missing from DB graph
  C-22  task-description-drift  WARN  Yes  Task description/content differs between file and DB
  C-23  workstream-active-task-planning-status WARN Yes Workstream/workplan is planning while a task is progress or wait
  C-24  repo-classification-missing WARN  No   Registered repo lacks a valid .repo-classification.yaml on disk
  C-25  blocked-unblock-sweep    WARN  No   Blocked workplan has unread inbox from counterpart — blocker may have cleared
  C-26  workplan-id-prefix       WARN  No   Workplan frontmatter id uses non-canonical prefix for this repo
  C-27  workplan-id-collision    WARN  No   Same workplan id appears in multiple repos
  C-28  inbox-stale-unread       WARN  No   Unread inbox messages older than INBOX_STALE_DAYS
  C-29  inbox-work-unpromoted    WARN  No   Stale unread message looks like a multi-step work request without a workplan
  C-30  scope-current-state-stale WARN No   SCOPE.md Current State contradicts live workplan statuses
  C-31  work-record-unregistered  WARN No   YAML-block id matches no kind in the canon work-record type registry (sidetrack detector, CUST-WP-0060)
  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:
    python scripts/consistency_check.py --repo SLUG [--fix] [--no-writeback] [--json] [--api-base URL]
    python scripts/consistency_check.py --all [--fix] [--no-writeback] [--json] [--api-base URL]
    python scripts/consistency_check.py --here [PATH] [--fix] [--no-writeback] [--json] [--api-base URL]

Identifiers (ADR-007 / RMGR-WP-0005):
    C-06 / C-11 / C-32 derive UUIDv5 identifiers from the fleet namespace and
    canonical record id. Any instance may reconcile; independent instances
    produce the same identifier and byte-identical writeback.

Exit codes (single-repo / local CLI):
    0 — clean (no FAILs or WARNs; INFOs are allowed)
    1 — one or more assessment FAILs or automation ERRORs (C-00) present
    2 — warnings-only strict CLI result (no FAILs, but WARNs present)

Exit codes (--remote --all scheduled sweep):
    0 — automation completed and documented results (assessment failures OK)
    1 — automation error: API unreachable, repo list fetch failed, C-00 on
        any repo, or other infrastructure fault that prevented a full run

Assessment failures (C-01..C-24 except C-00) are repo hygiene gaps recorded
in the sweep report for later improvement. They do not fail the scheduler.

Agent/operator Make wrappers normalize exit code 2 to shell success while
preserving visible warning output. Use the direct script when a machine caller
needs to distinguish clean from warnings-only.
"""
from __future__ import annotations

import argparse
import os
import json
import re
import shutil
import socket
import subprocess
import sys
import time
import uuid
from contextlib import contextmanager
from dataclasses import dataclass, field
from collections import Counter
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

_REPO_ROOT = Path(__file__).resolve().parent.parent
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

def _sweep_hostname() -> str:
    """Hostname for host_paths resolution; override when pod hostname != fleet FQDN."""
    return os.environ.get("STATE_HUB_SWEEP_HOSTNAME", "").strip() or socket.gethostname()


from api.workplan_status import (  # noqa: E402
    CANONICAL_WORKSTREAM_STATUSES,
    CLOSED_WORKSTREAM_STATUSES,
    LEGACY_WORKSTREAM_STATUS_ALIASES,
    OPEN_WORKSTREAM_STATUSES,
    SUPPORTED_WORKSTREAM_STATUSES,
    normalize_workstream_status as _normalize_workstream_status,
    ready_review_status,
)
from api.classification import (  # noqa: E402
    CLASSIFICATION_FILENAME,
    load_classification_file,
    resolve_topic_domain_slug,
)
from api.services.lifecycle import should_activate_parent_for_active_tasks  # noqa: E402
from api.task_status import (  # noqa: E402
    CANONICAL_TASK_STATUSES,
    OPEN_TASK_STATUSES,
    TASK_STATUS_ORDER,
    TERMINAL_TASK_STATUSES,
    normalize_task_status,
)

_WORK_RECORD_NAMESPACE_UUID = uuid.UUID("a4058507-5c4a-5a00-ab06-fffa4fb46009")


def _derived_work_record_uuid(record_id: str) -> str:
    return str(uuid.uuid5(_WORK_RECORD_NAMESPACE_UUID, f"helixforge\n{record_id}"))

try:
    import yaml as _yaml
    _HAS_YAML = True
except ImportError:
    _HAS_YAML = False

try:
    import httpx as _httpx
    _HAS_HTTPX = True
except ImportError:
    _HAS_HTTPX = False


# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

_LEGACY_METER_COMPONENT = "state-hub.fix-consistency"


def _legacy_meter_headers() -> dict[str, str]:
    return {"X-StateHub-Component": _LEGACY_METER_COMPONENT}


_TASK_BLOCK_RE = re.compile(r"```task\s*\n(.*?)\n```", re.DOTALL)
_HEADING_RE = re.compile(r"^(#{1,4})\s+(.+?)$", re.MULTILINE)
_ARCHIVED_WP_RE = re.compile(r"^\d{6}-(.+\.md)$")
VALID_WP_STATUSES = set(CANONICAL_WORKSTREAM_STATUSES)
SUPPORTED_WP_STATUSES = set(SUPPORTED_WORKSTREAM_STATUSES)
VALID_TASK_STATUSES = set(CANONICAL_TASK_STATUSES)
VALID_TASK_PRIORITIES = {"low", "medium", "high", "critical"}
VALID_DEP_RELATIONSHIPS = {"blocks", "starts_after", "informs", "soft_dependency"}
DEFAULT_REMOTE_ALL_MAX_SECONDS = int(os.environ.get("CONSISTENCY_REMOTE_ALL_MAX_SECONDS", "300"))
STALE_UNREAD_DAYS = int(os.environ.get("INBOX_STALE_DAYS", "3"))
_API_GET_RETRIES = int(os.environ.get("CONSISTENCY_API_GET_RETRIES", "3"))
_API_GET_RETRY_BASE_DELAY = float(os.environ.get("CONSISTENCY_API_GET_RETRY_DELAY", "0.5"))

_WP_FILE_PREFIX_RE = re.compile(r"^([A-Za-z][A-Za-z0-9-]*-WP)-\d+", re.IGNORECASE)
_WP_FILE_BARE_RE = re.compile(r"^(WP)-\d+", re.IGNORECASE)
_WP_ID_PREFIX_RE = re.compile(r"^([A-Z][A-Z0-9-]*-WP)-\d+", re.IGNORECASE)
_WP_ID_BARE_RE = re.compile(r"^(WP)-\d+", re.IGNORECASE)
_BLOCKED_ON_RE = re.compile(r"^message-from:(?P<agent>.+)$")
_WORK_REQUEST_RE = re.compile(
    r"\b(implement|workplan|multi-?step|please\s+(implement|add|create|fix|build))\b",
    re.IGNORECASE,
)
_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)

# Ordinal ranking for task statuses used by the no-regress rule (T01/C-15).
STATUS_ORDER: dict[str, int] = dict(TASK_STATUS_ORDER)


def normalise_workstream_status(status: str, *, has_started: bool | None = None) -> str:
    """Translate a workplan file status value to its DB-canonical equivalent."""
    return _normalize_workstream_status(status, has_started=has_started)


def normalise_task_status(status: Any, *, default: str = "todo") -> str:
    try:
        return normalize_task_status(status, default=default)
    except ValueError:
        return default


def canonical_workplan_filename(path: Path) -> str:
    """Return the workplan filename without an archive completion-date prefix."""
    return _ARCHIVED_WP_RE.sub(r"\1", path.name)


def workplan_display_path(repo_dir: Path, path: Path) -> str:
    """Stable relative path for reports, including archived/ when applicable."""
    try:
        return str(path.relative_to(repo_dir))
    except ValueError:
        return path.name


def iter_workplan_files(workplans_dir: Path, include_archived: bool = True) -> list[Path]:
    """Return active root workplans plus archived workplans when requested."""
    files = [
        path for path in sorted(workplans_dir.glob("*.md"))
        if path.name not in _NON_WORKPLAN_WORKPLAN_FILES
    ]
    archived_dir = workplans_dir / "archived"
    if include_archived and archived_dir.is_dir():
        files.extend(sorted(archived_dir.glob("*.md")))
    return files


# C-00 marks infrastructure/automation faults (API down, repo missing in DB).
# All other FAIL severities are assessment findings for follow-up.
_AUTOMATION_ERROR_CHECKS: frozenset[str] = frozenset({"C-00"})
_NON_WORKPLAN_WORKPLAN_FILES: frozenset[str] = frozenset({"README.md"})


# ---------------------------------------------------------------------------
# Data types
# ---------------------------------------------------------------------------

@dataclass
class Issue:
    severity: str           # FAIL | WARN | INFO
    check_id: str           # C-01 through C-12
    message: str
    file_path: str = ""
    db_id: str = ""
    file_value: str = ""
    db_value: str = ""
    fixable: bool = False
    _fix_context: dict = field(default_factory=dict, repr=False)


@dataclass
class ConsistencyReport:
    repo_slug: str
    repo_path: str
    issues: list[Issue] = field(default_factory=list)
    fixes_applied: list[str] = field(default_factory=list)

    def add(self, **kwargs) -> Issue:
        issue = Issue(**kwargs)
        self.issues.append(issue)
        return issue

    @property
    def failures(self) -> list[Issue]:
        return [i for i in self.issues if i.severity == "FAIL"]

    @property
    def automation_errors(self) -> list[Issue]:
        return [
            i for i in self.issues
            if i.severity == "FAIL" and i.check_id in _AUTOMATION_ERROR_CHECKS
        ]

    @property
    def assessment_failures(self) -> list[Issue]:
        return [
            i for i in self.issues
            if i.severity == "FAIL" and i.check_id not in _AUTOMATION_ERROR_CHECKS
        ]

    @property
    def warnings(self) -> list[Issue]:
        return [i for i in self.issues if i.severity == "WARN"]

    @property
    def infos(self) -> list[Issue]:
        return [i for i in self.issues if i.severity == "INFO"]


@dataclass(frozen=True)
class RenormalizationRule:
    check_id: str
    invariant: str
    detection: str
    repair: str
    test_anchor: str


RENORMALIZATION_RULES: tuple[RenormalizationRule, ...] = (
    RenormalizationRule(
        check_id="C-23",
        invariant="Planning-state workplans cannot contain active task work.",
        detection=(
            "Workplan status is proposed, ready, or backlog while a linked "
            "task is progress or wait."
        ),
        repair=(
            "ready/backlog: patch DB and file to active. "
            "proposed: report only (ADR-007 decision 3)."
        ),
        test_anchor="tests/test_consistency_check.py::TestLifecycleRenormalization",
    ),
)


RENORMALIZATION_NEXT_GUARD_CHECKLIST: tuple[str, ...] = (
    "Name the invariant and assign the next C-id.",
    "Add rule metadata to RENORMALIZATION_RULES.",
    "Add detection in check_repo before generic drift rules that could fight it.",
    "Add the repair branch in fix_repo, using file writes only through helpers.",
    "Add detection and repair tests under TestLifecycleRenormalization.",
)


@contextmanager
def run_lock(name: str):
    """Hold a nonblocking process lock for long-running consistency modes."""
    try:
        import fcntl
    except ImportError:
        yield True
        return

    lock_path = Path(os.environ.get("CONSISTENCY_LOCK_DIR", "/tmp")) / f"custodian-{name}.lock"
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    handle = lock_path.open("w", encoding="utf-8")
    try:
        try:
            fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            yield False
            return
        handle.seek(0)
        handle.truncate()
        handle.write(f"{os.getpid()} {datetime.utcnow().isoformat()}Z\n")
        handle.flush()
        yield True
    finally:
        try:
            fcntl.flock(handle, fcntl.LOCK_UN)
        except OSError:
            pass
        handle.close()


# ---------------------------------------------------------------------------
# YAML / frontmatter parsing
# ---------------------------------------------------------------------------

def _parse_yaml_block(raw: str) -> dict:
    """Parse a YAML string; fallback to simple key:value if pyyaml unavailable."""
    if _HAS_YAML:
        try:
            return _yaml.safe_load(raw) or {}
        except _yaml.YAMLError:
            return {"_parse_error": True}
    result: dict = {}
    for line in raw.splitlines():
        if ":" in line and not line.startswith(" "):
            k, _, v = line.partition(":")
            result[k.strip()] = v.strip().strip('"').strip("'")
    return result


def parse_frontmatter(text: str) -> tuple[dict, str]:
    """Split YAML frontmatter from body. Returns ({}, text) if no frontmatter."""
    if not text.startswith("---"):
        return {}, text
    parts = text.split("---", 2)
    if len(parts) < 3:
        return {}, text
    meta = _parse_yaml_block(parts[1].strip())
    return meta, parts[2]


def _clean_task_description(raw: str) -> str | None:
    """Trim section chrome while preserving task markdown content."""
    lines = raw.splitlines()
    while lines and not lines[0].strip():
        lines.pop(0)
    while lines and not lines[-1].strip():
        lines.pop()
    while lines and lines[-1].strip() in {"---", "***", "___"}:
        lines.pop()
        while lines and not lines[-1].strip():
            lines.pop()
    while lines and lines[0].strip() in {"---", "***", "___"}:
        lines.pop(0)
        while lines and not lines[0].strip():
            lines.pop(0)
    text = "\n".join(lines).strip()
    return text or None


def parse_task_blocks(body: str) -> list[dict]:
    """Extract task blocks, injecting title and markdown content from the task section."""
    headings = [
        (m.start(), len(m.group(1)), m.group(2).strip())
        for m in _HEADING_RE.finditer(body)
    ]
    task_matches = list(_TASK_BLOCK_RE.finditer(body))
    results = []
    for idx, m in enumerate(task_matches):
        task = _parse_yaml_block(m.group(1).strip())
        prev = [(pos, level, text) for pos, level, text in headings if pos < m.start()]
        prev_heading = prev[-1] if prev else None
        if "title" not in task:
            if prev_heading:
                task["title"] = prev_heading[2]

        content_end = len(body)
        if idx + 1 < len(task_matches):
            content_end = min(content_end, task_matches[idx + 1].start())
        if prev_heading:
            current_level = prev_heading[1]
            next_peer_heading = next(
                (
                    pos
                    for pos, level, _text in headings
                    if pos > m.end() and level <= current_level
                ),
                None,
            )
            if next_peer_heading is not None:
                content_end = min(content_end, next_peer_heading)
        else:
            next_heading = next((pos for pos, _level, _text in headings if pos > m.end()), None)
            if next_heading is not None:
                content_end = min(content_end, next_heading)

        description = _clean_task_description(body[m.end():content_end])
        if description:
            task["description"] = description
        results.append(task)
    return results


def get_tasks_from_workplan(meta: dict, body: str) -> list[dict]:
    """Get tasks from workplan — handles both ```task``` blocks and tasks: YAML list."""
    blocks = parse_task_blocks(body)
    if blocks:
        return blocks
    # Fallback: tasks embedded as a YAML list in frontmatter (activity-core style)
    tasks = meta.get("tasks")
    if isinstance(tasks, list):
        return tasks
    return []


def _as_list(value: Any) -> list[str]:
    if value is None:
        return []
    if isinstance(value, list):
        return [str(item).strip().strip('"') for item in value if str(item).strip()]
    if isinstance(value, str):
        return [item.strip().strip('"') for item in value.split(",") if item.strip()]
    return [str(value).strip().strip('"')]


def _as_int_or_none(value: Any) -> int | None:
    if value in (None, "", "~", "null", "None", "none"):
        return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


# ---------------------------------------------------------------------------
# File update helpers
# ---------------------------------------------------------------------------

def _add_frontmatter_field(file_path: Path, key: str, value: str) -> None:
    """Insert key: "value" into frontmatter before the closing --- line."""
    text = file_path.read_text(encoding="utf-8")
    lines = text.split("\n")
    close_idx = None
    for i, line in enumerate(lines[1:], 1):
        if line.strip() == "---":
            close_idx = i
            break
    if close_idx is None:
        return
    lines.insert(close_idx, f'{key}: "{value}"')
    file_path.write_text("\n".join(lines), encoding="utf-8")


def _patch_frontmatter_field(file_path: Path, key: str, value: str) -> bool:
    """Update or insert a scalar frontmatter field without rewriting the file."""
    text = file_path.read_text(encoding="utf-8")
    if not text.startswith("---"):
        return False
    lines = text.split("\n")
    close_idx = None
    for i, line in enumerate(lines[1:], 1):
        if line.strip() == "---":
            close_idx = i
            break
    if close_idx is None:
        return False

    new_line = f"{key}: {value}"
    for i in range(1, close_idx):
        if re.match(rf"^\s*{re.escape(key)}\s*:", lines[i]):
            if lines[i] == new_line:
                return False
            lines[i] = new_line
            file_path.write_text("\n".join(lines), encoding="utf-8")
            return True

    lines.insert(close_idx, new_line)
    file_path.write_text("\n".join(lines), encoding="utf-8")
    return True


def _inject_task_id_into_block(
    file_path: Path, field_name: str, field_value: str, match_id: str
) -> bool:
    """Inject state_hub_task_id into the ```task``` block whose id == match_id."""
    text = file_path.read_text(encoding="utf-8")

    def _replace(m: re.Match) -> str:
        block_content = m.group(1)
        task_meta = _parse_yaml_block(block_content.strip())
        if str(task_meta.get("id", "")) != match_id:
            return m.group(0)
        existing_val = task_meta.get(field_name)
        if existing_val is not None and str(existing_val).strip() not in ("", "~", "null", "None", "none"):
            return m.group(0)
        # Replace existing null/~ line if present, otherwise append
        new_content = re.sub(
            rf"^{re.escape(field_name)}:.*$",
            f'{field_name}: "{field_value}"',
            block_content,
            flags=re.MULTILINE,
        )
        if new_content == block_content:
            new_content = block_content.rstrip() + f"\n{field_name}: \"{field_value}\""
        return f"```task\n{new_content}\n```"

    new_text = _TASK_BLOCK_RE.sub(_replace, text)
    if new_text != text:
        file_path.write_text(new_text, encoding="utf-8")
        return True
    return False


def _inject_task_id_frontmatter_list(
    file_path: Path, field_value: str, match_id: str
) -> bool:
    """Inject state_hub_task_id into a task entry in frontmatter tasks: list."""
    if not _HAS_YAML:
        return False
    import yaml
    text = file_path.read_text(encoding="utf-8")
    meta, body = parse_frontmatter(text)
    tasks = meta.get("tasks", [])
    changed = False
    for t in tasks:
        if str(t.get("id", "")) == match_id and "state_hub_task_id" not in t:
            t["state_hub_task_id"] = field_value
            changed = True
    if not changed:
        return False
    try:
        new_meta_str = yaml.dump(meta, allow_unicode=True, default_flow_style=False)
        parts = text.split("---", 2)
        if len(parts) < 3:
            return False
        new_text = f"---\n{new_meta_str}---{parts[2]}"
        file_path.write_text(new_text, encoding="utf-8")
        return True
    except Exception:
        return False


# ---------------------------------------------------------------------------
# API helpers
# ---------------------------------------------------------------------------

_API_CLIENTS: dict[tuple[str, object], Any] = {}


def _api_client(api_base: str) -> Any:
    """Reuse one keep-alive pool per API/client implementation for this run."""
    key = (api_base.rstrip("/"), _httpx.Client)
    client = _API_CLIENTS.get(key)
    if client is None:
        client = _httpx.Client(
            base_url=api_base,
            timeout=10.0,
            follow_redirects=True,
            headers=_legacy_meter_headers(),
        )
        _API_CLIENTS[key] = client
    return client


def _api_get(
    api_base: str,
    path: str,
    params: dict | None = None,
    *,
    return_error: bool = False,
) -> Any:
    if not _HAS_HTTPX:
        return {"_error": "httpx is not installed"} if return_error else None
    # Only append trailing slash to the path component, not to query strings
    if "?" not in path and not path.endswith("/"):
        path += "/"
    filtered = {k: v for k, v in (params or {}).items() if v is not None}
    last_error: Exception | None = None
    for attempt in range(_API_GET_RETRIES):
        try:
            r = _api_client(api_base).get(path, params=filtered if filtered else None)
            r.raise_for_status()
            return r.json()
        except _httpx.HTTPStatusError as exc:
            if exc.response.status_code == 404:
                return None
            last_error = exc
            if exc.response.status_code >= 500 and attempt < _API_GET_RETRIES - 1:
                time.sleep(_API_GET_RETRY_BASE_DELAY * (attempt + 1))
                continue
            if return_error:
                return {"_error": str(exc)}
            return None
        except (_httpx.TimeoutException, _httpx.ConnectError, _httpx.NetworkError) as exc:
            last_error = exc
            if attempt < _API_GET_RETRIES - 1:
                time.sleep(_API_GET_RETRY_BASE_DELAY * (attempt + 1))
                continue
            if return_error:
                return {"_error": str(exc)}
            return None
        except Exception as exc:
            last_error = exc
            if return_error:
                return {"_error": str(exc)}
            return None
    if return_error and last_error is not None:
        return {"_error": str(last_error)}
    return None


def _api_patch(api_base: str, path: str, body: dict) -> Any:
    if not _HAS_HTTPX:
        return None
    if not path.endswith("/"):
        path += "/"
    try:
        r = _api_client(api_base).patch(path, json=body)
        r.raise_for_status()
        return r.json()
    except Exception as exc:
        # Return a sentinel dict so callers can distinguish "API error" from "success"
        # and report it rather than silently dropping the fix.
        return {"_error": str(exc)}


def _api_put(api_base: str, path: str, body: dict) -> Any:
    if not _HAS_HTTPX:
        return {"_error": "httpx is not installed"}
    if not path.endswith("/"):
        path += "/"
    try:
        r = _api_client(api_base).put(path, json=body, timeout=30.0)
        r.raise_for_status()
        return r.json()
    except Exception as exc:
        return {"_error": str(exc)}


def _api_post(api_base: str, path: str, body: dict) -> Any:
    if not _HAS_HTTPX:
        return {"_error": "httpx is not installed"}
    if not path.endswith("/"):
        path += "/"
    try:
        r = _api_client(api_base).post(path, json=body)
        r.raise_for_status()
        return r.json()
    except _httpx.HTTPStatusError as exc:
        detail = exc.response.text
        if len(detail) > 500:
            detail = detail[:497] + "..."
        return {"_error": f"{exc.response.status_code} {exc.response.reason_phrase}: {detail}"}
    except Exception as exc:
        return {"_error": str(exc)}


# ---------------------------------------------------------------------------
# Core check engine
# ---------------------------------------------------------------------------

def resolve_repo_path(repo: dict, override: str | None = None) -> str:
    """Resolve the local filesystem path for a repo on the current machine.

    Priority:
    1. Explicit --repo-path CLI override
    2. host_paths[current_hostname] — per-machine path registered via POST /repos/{slug}/paths/
    3. local_path — legacy single-path field (backward compat)
    """
    if override:
        return override
    hostname = _sweep_hostname()
    host_paths = repo.get("host_paths") or {}
    return host_paths.get(hostname) or repo.get("local_path") or ""


def _infer_slug_from_path(api_base: str, path: str) -> "tuple[str, str] | None":
    """Identify a registered repo from a local checkout path.

    Strategy (in order):
    1. Root-commit fingerprint — ``git rev-list --max-parents=0 HEAD`` produces
       the same SHA-1 on every clone, independent of remote URL or checkout path.
       Looked up via ``GET /repos/by-fingerprint?hash=<sha>``.
    2. Remote URL fallback — exact string match against ``remote_url`` field.
       Less reliable across machines (SSH aliases, HTTP vs HTTPS, etc.) but
       useful when fingerprint is not yet stored.

    Returns ``(slug, git_root)`` on success, ``None`` if no match found.
    """
    try:
        git_root = subprocess.check_output(
            ["git", "rev-parse", "--show-toplevel"],
            cwd=path, stderr=subprocess.DEVNULL, text=True,
        ).strip()
    except (subprocess.CalledProcessError, FileNotFoundError, OSError):
        return None

    # Strategy 1: fingerprint lookup (most reliable)
    try:
        fingerprint = subprocess.check_output(
            ["git", "rev-list", "--max-parents=0", "HEAD"],
            cwd=git_root, stderr=subprocess.DEVNULL, text=True,
        ).strip()
    except (subprocess.CalledProcessError, FileNotFoundError, OSError):
        fingerprint = ""

    # Get local remote URL once — used for both disambiguation and fallback
    try:
        remote_url = subprocess.check_output(
            ["git", "remote", "get-url", "origin"],
            cwd=git_root, stderr=subprocess.DEVNULL, text=True,
        ).strip()
    except (subprocess.CalledProcessError, FileNotFoundError, OSError):
        remote_url = ""

    if fingerprint:
        # Try fingerprint + remote URL for precise match
        if remote_url:
            import urllib.parse as _up
            candidates = _api_get(
                api_base,
                f"/repos/by-fingerprint?hash={fingerprint}&remote_url={_up.quote(remote_url, safe='')}",
            )
            if isinstance(candidates, list) and len(candidates) == 1:
                return candidates[0]["slug"], git_root

        # Fingerprint alone (works when repos don't share ancestry)
        candidates = _api_get(api_base, f"/repos/by-fingerprint?hash={fingerprint}")
        if isinstance(candidates, list):
            if len(candidates) == 1:
                return candidates[0]["slug"], git_root
            if len(candidates) > 1:
                # Disambiguate: prefer the repo whose slug appears in the git_root path
                for repo in candidates:
                    if repo["slug"] in git_root:
                        return repo["slug"], git_root
                # Can't disambiguate — return first match with a warning
                print(
                    f"  WARNING: {len(candidates)} repos share fingerprint {fingerprint[:12]}… "
                    f"— using '{candidates[0]['slug']}'. "
                    "Set remote_url on each repo for accurate matching.",
                    file=sys.stderr,
                )
                return candidates[0]["slug"], git_root

    # Strategy 2: remote URL exact match (fallback)
    if remote_url:
        import urllib.parse as _up
        repo = _api_get(api_base, f"/repos/by-remote?url={_up.quote(remote_url, safe='')}")
        if repo and isinstance(repo, dict) and "slug" in repo:
            return repo["slug"], git_root

    return None


_WORK_RECORD_FENCE_RE = re.compile(r"```(?:yaml|task)\n(.*?)```", re.S)
_WORK_RECORD_ID_LINE_RE = re.compile(r"^id:\s*[\"']?([A-Za-z0-9_-]+)[\"']?\s*$", re.M)
_WORK_RECORD_ID_LIKE_RE = re.compile(r"^[A-Z][A-Z0-9]*(-[A-Z0-9]+)+$")
_WORK_RECORD_SKIP_DIRS = frozenset({".git", "node_modules", ".venv", "history"})


def _load_work_record_kind_registry() -> list[dict] | None:
    """Load [{kind, patterns}] from the canon work-record type registry.

    Location: $WORK_RECORD_REGISTRY, else ~/the-custodian/canon/standards/
    work-record-types.yaml. Returns None (checks skipped) when unavailable.
    Kind-aware: shared by C-31 (flat sidetrack detection) and C-32
    (kind-aware registration — needs to know intake vs. decision vs.
    engagement to pick the right hub entity).
    """
    if not _HAS_YAML:
        return None
    candidates = []
    env = os.environ.get("WORK_RECORD_REGISTRY")
    if env:
        candidates.append(Path(env))
    candidates.append(
        Path.home() / "the-custodian/canon/standards/work-record-types.yaml")
    for path in candidates:
        if not path.is_file():
            continue
        try:
            reg = _yaml.safe_load(path.read_text())
            kinds = []
            for kind in reg.get("kinds", []):
                patterns = [re.compile(p) for p in kind.get("id_patterns", [])]
                patterns += [re.compile(lp["pattern"])
                             for lp in kind.get("legacy_patterns", [])]
                kinds.append({"kind": kind["kind"], "patterns": patterns})
            return kinds
        except Exception:
            return None
    return None


def _load_work_record_patterns() -> list[re.Pattern] | None:
    """Flat id-pattern list across all kinds — used by C-31's sidetrack
    detector, which only needs to know "is this id registered at all",
    not which kind it belongs to."""
    kinds = _load_work_record_kind_registry()
    if kinds is None:
        return None
    return [p for k in kinds for p in k["patterns"]]


def _check_unregistered_work_records(repo_dir: Path, report: "ConsistencyReport") -> None:
    """C-31: warn on YAML-block ids matching no registered work-record kind.

    Detection only (CUST-WP-0060-T05); registration of non-workplan kinds is
    a later stage. An unregistered id scheme is a coordination sidetrack that
    will need manual reintegration — surface it the day it appears.
    """
    patterns = _load_work_record_patterns()
    if patterns is None:
        return
    seen: set[str] = set()
    for md in sorted(repo_dir.rglob("*.md")):
        if any(part in _WORK_RECORD_SKIP_DIRS for part in md.parts):
            continue
        try:
            text = md.read_text(errors="replace")
        except OSError:
            continue
        for body in _WORK_RECORD_FENCE_RE.findall(text):
            for rid in _WORK_RECORD_ID_LINE_RE.findall(body):
                if rid in seen or "NNN" in rid or rid.endswith("-TNN"):
                    continue
                if not _WORK_RECORD_ID_LIKE_RE.match(rid):
                    continue
                if any(p.match(rid) for p in patterns):
                    continue
                seen.add(rid)
                report.add(
                    severity="WARN",
                    check_id="C-31",
                    message=(
                        f"[{md.relative_to(repo_dir)}] id '{rid}' matches no "
                        f"kind in the canon work-record type registry — "
                        f"unregistered species become sidetracks needing "
                        f"manual reintegration (work-record-types_v0.1)"
                    ),
                    fixable=False,
                )


_YAML_ONLY_FENCE_RE = re.compile(r"```yaml\n(.*?)\n```", re.S)
_WORK_RECORD_ID_FIELD_BY_KIND = {
    "intake": "state_hub_intake_id",
    "decision": "state_hub_decision_id",
    "engagement": "state_hub_engagement_id",
}
_WORK_RECORD_CREATE_ENDPOINT_BY_KIND = {
    "intake": "/intakes",
    "decision": "/decisions",
}


def _classify_work_record_kind(record_id: str, kind_registry: list[dict]) -> str | None:
    for k in kind_registry:
        if any(p.match(record_id) for p in k["patterns"]):
            return k["kind"]
    return None


def _inject_yaml_block_field(
    file_path: Path, field_name: str, field_value: str, match_id: str
) -> bool:
    """Inject a field into the ```yaml``` block whose id == match_id.
    Mirrors _inject_task_id_into_block, adapted for fenced yaml blocks
    (intake/decision/engagement records live as ```yaml, not ```task)."""
    text = file_path.read_text(encoding="utf-8")

    def _replace(m: re.Match) -> str:
        block_content = m.group(1)
        meta = _parse_yaml_block(block_content.strip())
        if not isinstance(meta, dict) or str(meta.get("id", "")) != match_id:
            return m.group(0)
        existing_val = meta.get(field_name)
        if existing_val is not None and str(existing_val).strip() not in ("", "~", "null", "None", "none"):
            return m.group(0)
        new_content = re.sub(
            rf"^{re.escape(field_name)}:.*$",
            f'{field_name}: "{field_value}"',
            block_content,
            flags=re.MULTILINE,
        )
        if new_content == block_content:
            new_content = block_content.rstrip() + f"\n{field_name}: \"{field_value}\""
        return f"```yaml\n{new_content}\n```"

    new_text = _YAML_ONLY_FENCE_RE.sub(_replace, text)
    if new_text != text:
        file_path.write_text(new_text, encoding="utf-8")
        return True
    return False


def _check_work_record_registration(
    repo_dir: Path,
    report: "ConsistencyReport",
    api_base: str = "http://127.0.0.1:8000",
    bootstrap_empty_projection: bool = False,
) -> None:
    """C-32: register kind: intake / kind: decision YAML blocks found
    anywhere in the repo against the hub (CUST-WP-0061-T02, work-record
    stage 3) — writes state_hub_intake_id / state_hub_decision_id back into
    the source block, same write-back pattern as C-06 for workplans.

    kind: engagement has no hub entity yet (deferred; reported INFO, not
    fixable) — building one is a separate stage-3 follow-on, not this task.
    """
    kind_registry = _load_work_record_kind_registry()
    if kind_registry is None:
        return
    for md in sorted(repo_dir.rglob("*.md")):
        if any(part in _WORK_RECORD_SKIP_DIRS for part in md.parts):
            continue
        try:
            text = md.read_text(errors="replace")
        except OSError:
            continue
        for block in _YAML_ONLY_FENCE_RE.findall(text):
            meta = _parse_yaml_block(block.strip())
            if not isinstance(meta, dict) or meta.get("_parse_error"):
                continue
            rid = str(meta.get("id", "")).strip()
            if not rid or "NNN" in rid:
                continue
            kind = _classify_work_record_kind(rid, kind_registry)
            if kind not in _WORK_RECORD_ID_FIELD_BY_KIND:
                continue

            id_field = _WORK_RECORD_ID_FIELD_BY_KIND[kind]
            existing = str(meta.get(id_field, "")).strip().strip('"')
            if existing and existing not in ("~", "null", "None", "none"):
                if not bootstrap_empty_projection:
                    continue
                endpoint = _WORK_RECORD_CREATE_ENDPOINT_BY_KIND.get(kind)
                if endpoint and _api_get(api_base, f"{endpoint}/{existing}") is not None:
                    continue

            rel = md.relative_to(repo_dir)
            if kind not in _WORK_RECORD_CREATE_ENDPOINT_BY_KIND:
                report.add(
                    severity="INFO", check_id="C-32",
                    message=(
                        f"[{rel}] engagement '{rid}' has no hub entity yet — "
                        f"registration deferred (CUST-WP-0061 stage-3 seed)"
                    ),
                    fixable=False,
                )
                continue

            report.add(
                severity="WARN", check_id="C-32",
                message=(
                    f"[{rel}] {kind} '{rid}' is absent from the explicitly empty projection"
                    if existing
                    else f"[{rel}] {kind} '{rid}' has no {id_field} — not indexed in DB"
                ),
                file_path=str(rel),
                file_value=rid,
                fixable=True,
                _fix_context={
                    "md_path": md,
                    "kind": kind,
                    "rid": rid,
                    "meta": meta,
                    "desired_record_id": existing or None,
                },
            )


_WORK_RECORD_INDEX_NAME = "WORK-RECORDS.md"
_WORK_RECORD_INDEX_KIND_ORDER = {
    "workplan": 0, "task": 1, "intake": 2, "decision": 3, "engagement": 4,
}


def _generate_work_record_index(repo_dir: Path, repo_slug: str) -> str | None:
    """CUST-WP-0061-T04: the generated per-repo work-record index — every
    registered work record (workplan, task, intake, decision, engagement)
    with kind/id/status/lane/source, derivable purely from repo files (no
    hub query — matches the ADR-001 rebuild principle). Full transclusion
    rendering (markitect integration) is a follow-on; this is the flat
    first cut. Returns None (no file written) when the repo has no work
    records at all, to avoid clutter in small repos.
    """
    rows: list[tuple[str, str, str, str, str]] = []  # kind, id, status, lane, source

    workplans_dir = repo_dir / "workplans"
    if workplans_dir.is_dir():
        for wp_file in iter_workplan_files(workplans_dir, include_archived=False):
            try:
                text = wp_file.read_text(encoding="utf-8", errors="replace")
            except OSError:
                continue
            meta, body = parse_frontmatter(text)
            wp_id = str(meta.get("id", "")).strip()
            if not wp_id:
                continue
            source = str(wp_file.relative_to(repo_dir))
            rows.append(("workplan", wp_id, str(meta.get("status", "")).strip() or "—", "—", source))
            for task in get_tasks_from_workplan(meta, body):
                if task.get("_parse_error"):
                    continue
                t_id = str(task.get("id", "")).strip()
                if not t_id:
                    continue
                rows.append(("task", t_id, str(task.get("status", "")).strip() or "—", "—", source))

    kind_registry = _load_work_record_kind_registry()
    if kind_registry:
        for md in sorted(repo_dir.rglob("*.md")):
            if any(part in _WORK_RECORD_SKIP_DIRS for part in md.parts):
                continue
            if md.name == _WORK_RECORD_INDEX_NAME:
                continue
            try:
                text = md.read_text(errors="replace")
            except OSError:
                continue
            for block in _YAML_ONLY_FENCE_RE.findall(text):
                meta = _parse_yaml_block(block.strip())
                if not isinstance(meta, dict) or meta.get("_parse_error"):
                    continue
                rid = str(meta.get("id", "")).strip()
                if not rid or "NNN" in rid:
                    continue
                kind = _classify_work_record_kind(rid, kind_registry)
                if kind not in ("intake", "decision", "engagement"):
                    continue
                status = str(meta.get("status", "")).strip() or "—"
                lane = str(meta.get("lane", "")).strip() or "—"
                source = str(md.relative_to(repo_dir))
                rows.append((kind, rid, status, lane, source))

    if not rows:
        return None

    rows.sort(key=lambda r: (_WORK_RECORD_INDEX_KIND_ORDER.get(r[0], 99), r[4], r[1]))
    return _work_record_index_document(repo_slug, rows)


def _work_record_index_document(
    repo_slug: str, rows: list[tuple[str, str, str, str, str]]
) -> str:
    """Render WORK-RECORDS.md. An empty row list still produces the header
    and table so a leftover index can be rewritten after every live record
    is archived or removed.
    """
    lines = [
        f"# Work Records — {repo_slug}",
        "",
        "> Generated by `statehub fix-consistency` (CUST-WP-0061-T04, work-record",
        "> stage 3). Do not edit by hand — edit the source file/block listed for",
        "> each record and re-run fix-consistency to refresh this index. Archived",
        "> workplans are omitted; closed decisions/intakes/engagements stay listed",
        "> so recently-resolved work is still visible. [auto]",
        "",
        "| Kind | ID | Status | Lane | Source |",
        "| --- | --- | --- | --- | --- |",
    ]
    lines.extend(
        f"| {kind} | {rid} | {status} | {lane} | {source} |"
        for kind, rid, status, lane, source in rows
    )
    lines.append("")
    return "\n".join(lines)


def _check_work_record_index_freshness(
    repo_dir: Path, repo_slug: str, report: "ConsistencyReport"
) -> None:
    """C-33: WORK-RECORDS.md missing or stale relative to the current
    file-derived work-record set."""
    content = _generate_work_record_index(repo_dir, repo_slug)
    index_file = repo_dir / _WORK_RECORD_INDEX_NAME
    existing = index_file.read_text(encoding="utf-8") if index_file.is_file() else None
    if content is None:
        if existing is None:
            return
        empty = _work_record_index_document(repo_slug, [])
        if existing == empty:
            return
        content = empty
    elif existing == content:
        return
    report.add(
        severity="WARN", check_id="C-33",
        message=(
            f"{_WORK_RECORD_INDEX_NAME} is missing or stale — regenerate via "
            f"fix-consistency (CUST-WP-0061-T04 generated work-record index)"
        ),
        file_path=_WORK_RECORD_INDEX_NAME,
        fixable=True,
        _fix_context={"index_file": index_file, "content": content},
    )


def check_repo(
    api_base: str,
    repo_slug: str,
    repo_path_override: str | None = None,
    bootstrap_empty_projection: bool = False,
) -> ConsistencyReport:
    """Run all consistency checks for a registered repo."""
    repo = _api_get(api_base, f"/repos/{repo_slug}", return_error=True)
    if isinstance(repo, dict) and "_error" in repo:
        report = ConsistencyReport(repo_slug=repo_slug, repo_path=repo_path_override or "(unknown)")
        report.add(
            severity="FAIL", check_id="C-00",
            message=(
                f"Could not query State Hub API at {api_base}: {repo['_error']}"
            ),
            fixable=False,
        )
        return report
    if repo is None:
        report = ConsistencyReport(repo_slug=repo_slug, repo_path="(unknown)")
        report.add(
            severity="FAIL", check_id="C-00",
            message=f"Repo '{repo_slug}' not found in state-hub DB",
            fixable=False,
        )
        return report

    # Repository aliases are accepted at the API boundary, but every new
    # binding, inbox write, and generated file must use the current identity.
    repo_slug = str(repo.get("canonical_slug") or repo.get("slug") or repo_slug)
    repo_id: str = repo["id"]
    repo_path: str = resolve_repo_path(repo, repo_path_override)
    report = ConsistencyReport(repo_slug=repo_slug, repo_path=repo_path)

    if bootstrap_empty_projection:
        projected = _api_get(api_base, "/workplans", {"repo_id": repo_id}, return_error=True)
        if not isinstance(projected, list):
            report.add(
                severity="FAIL",
                check_id="C-36",
                message="Could not prove that the repository projection is empty",
                fixable=False,
            )
            return report
        if projected:
            report.add(
                severity="FAIL",
                check_id="C-36",
                message=(
                    "Empty-projection bootstrap refused: repository already has "
                    f"{len(projected)} projected workplan(s)"
                ),
                fixable=False,
            )
            return report

    if not repo_path:
        report.add(
            severity="FAIL", check_id="C-00",
            message=f"Repo '{repo_slug}' has no local_path — cannot check workplan files",
            fixable=False,
        )
        _check_orphan_db(api_base, repo_id, set(), report)
        return report

    repo_dir = Path(repo_path)
    workplans_dir = repo_dir / "workplans"
    repo_market_domain = str(repo.get("domain_slug") or "").strip()

    # C-24: repo classification file missing or invalid (always WARN — migration rows too)
    class_data, class_errors, class_warnings = load_classification_file(repo_dir)
    if class_data is None:
        classified_by = str(repo.get("classified_by") or "").strip()
        if class_errors:
            detail = "; ".join(class_errors)
        else:
            detail = f"{CLASSIFICATION_FILENAME} missing on disk"
        if classified_by == "migration":
            detail = f"{detail} (DB row is migration-derived — commit a human-reviewed file when ready)"
        report.add(
            severity="WARN",
            check_id="C-24",
            message=f"Repo classification gap: {detail}",
            fixable=False,
        )
    for warning in class_warnings:
        report.add(
            severity="WARN",
            check_id="C-24",
            message=f"Repo classification advisory: {warning}",
            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)

    # C-32: intake/decision work-record registration (CUST-WP-0061-T02)
    _check_work_record_registration(
        repo_dir,
        report,
        api_base=api_base,
        bootstrap_empty_projection=bootstrap_empty_projection,
    )

    # C-33: generated per-repo work-record index (CUST-WP-0061-T04)
    _check_work_record_index_freshness(repo_dir, repo_slug, report)

    # C-01: workplans/ directory missing
    if not workplans_dir.is_dir():
        report.add(
            severity="FAIL", check_id="C-01",
            message=(
                "workplans/ directory missing — ADR-001 requires workplan files "
                "at <repo>/workplans/<ID>-<slug>.md"
            ),
            fixable=False,
        )
        _check_orphan_db(api_base, repo_id, set(), report)
        return report

    # Parse workplan files
    workplan_infos: list[tuple[Path, dict, str]] = []
    file_ws_ids: dict[str, tuple[Path, dict, str]] = {}  # ws_id → (file, meta, body)
    active_file_ws_ids: set[str] = set()

    for wp_file in iter_workplan_files(workplans_dir):
        try:
            text = wp_file.read_text(encoding="utf-8")
        except OSError as e:
            report.add(severity="FAIL", check_id="C-02",
                       message=f"Cannot read file: {e}", file_path=workplan_display_path(repo_dir, wp_file))
            continue
        if not text.startswith("---"):
            report.add(severity="FAIL", check_id="C-02",
                       message="No YAML frontmatter found", file_path=workplan_display_path(repo_dir, wp_file))
            continue
        meta, body = parse_frontmatter(text)
        if not meta or meta.get("_parse_error"):
            report.add(severity="FAIL", check_id="C-02",
                       message="YAML frontmatter parse error", file_path=workplan_display_path(repo_dir, wp_file))
            continue
        workplan_infos.append((wp_file, meta, body))
        ws_id = str(meta.get("state_hub_workstream_id", "")).strip().strip('"')
        if ws_id:
            file_ws_ids[ws_id] = (wp_file, meta, body)
            if wp_file.parent == workplans_dir:
                active_file_ws_ids.add(ws_id)

    workplan_id_to_ws_id: dict[str, str] = {}
    task_file_id_to_sh_id: dict[str, str] = {}
    for _wp_file, meta, body in workplan_infos:
        mapped_ws_id = str(meta.get("state_hub_workstream_id", "")).strip().strip('"')
        wp_id = str(meta.get("id", "")).strip()
        if wp_id and mapped_ws_id:
            workplan_id_to_ws_id[wp_id] = mapped_ws_id
        for task in get_tasks_from_workplan(meta, body):
            if task.get("_parse_error"):
                continue
            task_file_id = str(task.get("id", "")).strip()
            raw_sh = task.get("state_hub_task_id")
            task_sh_id = "" if raw_sh is None else str(raw_sh).strip().strip('"')
            if task_file_id and task_sh_id and task_sh_id not in ("~", "null", "None", "none"):
                task_file_id_to_sh_id[task_file_id] = task_sh_id

    # Modern hubs expose the entire repo-scoped read model in one response.
    # Keep the older per-record calls as a rolling-deploy compatibility path.
    projection_snapshot = _api_get(
        api_base,
        f"/repos/{repo_slug}/work-record-projection/snapshot",
        return_error=True,
    )
    snapshot_available = (
        isinstance(projection_snapshot, dict)
        and projection_snapshot.get("schema")
        == "state-hub.repository-projection-snapshot.v1"
        and str(projection_snapshot.get("repo_id")) == repo_id
    )
    snapshot_workplans_by_id: dict[str, dict] = {}
    snapshot_tasks_by_workplan: dict[str, list[dict]] = {}
    snapshot_dependencies_by_workplan: dict[str, list[dict]] = {}
    if snapshot_available:
        for row in projection_snapshot.get("workplans", []):
            if isinstance(row, dict) and row.get("id"):
                snapshot_workplans_by_id[str(row["id"])] = row
        for row in projection_snapshot.get("tasks", []):
            if isinstance(row, dict) and row.get("workplan_id"):
                snapshot_tasks_by_workplan.setdefault(
                    str(row["workplan_id"]), []
                ).append(row)
        for row in projection_snapshot.get("dependencies", []):
            if isinstance(row, dict) and row.get("from_workplan_id"):
                snapshot_dependencies_by_workplan.setdefault(
                    str(row["from_workplan_id"]), []
                ).append(row)

    # Per-workplan checks
    for wp_file, meta, body in workplan_infos:
        fname = workplan_display_path(repo_dir, wp_file)
        archived_file = wp_file.parent.name == "archived"
        ws_id = str(meta.get("state_hub_workstream_id", "")).strip().strip('"')
        file_status = str(meta.get("status", "")).strip()
        file_title = str(meta.get("title", "")).strip()
        file_domain = str(meta.get("domain", "")).strip()

        normalised_file_status = normalise_workstream_status(file_status)
        if archived_file and normalised_file_status not in CLOSED_WORKSTREAM_STATUSES:
            report.add(
                severity="FAIL", check_id="C-18",
                message="Archived workplan file has an open or planning status",
                file_path=fname,
                file_value=file_status,
                fixable=False,
            )

        if not ws_id:
            # C-06: workplan not linked to any DB workstream
            report.add(
                severity="WARN", check_id="C-06",
                message="Workplan has no state_hub_workstream_id — not indexed in DB",
                file_path=fname,
                file_value=file_title or fname,
                fixable=True,
                _fix_context={
                    "wp_file": str(wp_file),
                    "meta": meta,
                    "body": body,
                    "repo_id": repo_id,
                    "domain": file_domain,
                    "repo_market_domain": repo_market_domain,
                    "repo_slug": repo_slug,
                },
            )
            continue

        ws = (
            snapshot_workplans_by_id.get(ws_id)
            if snapshot_available
            else _api_get(api_base, f"/workplans/{ws_id}")
        )
        if ws is None:
            wp_id = str(meta.get("id", "")).strip()
            if wp_id and (
                ws_id == _derived_work_record_uuid(wp_id) or bootstrap_empty_projection
            ):
                # A deterministic file identifier missing from a replaceable
                # projection is registration work, not a stale reference.
                report.add(
                    severity="WARN",
                    check_id="C-06",
                    message=(
                        f"Authoritative workplan {ws_id[:8]}… is absent from this "
                        + (
                            "explicitly empty projection"
                            if bootstrap_empty_projection
                            else "projection"
                        )
                    ),
                    file_path=fname,
                    db_id=ws_id,
                    fixable=True,
                    _fix_context={
                        "wp_file": str(wp_file),
                        "meta": meta,
                        "body": body,
                        "repo_id": repo_id,
                        "domain": file_domain,
                        "repo_market_domain": repo_market_domain,
                        "repo_slug": repo_slug,
                        "desired_ws_id": ws_id,
                    },
                )
            else:
                # A non-derived missing identifier may be genuine stale state;
                # preserve the conservative manual-review behavior.
                report.add(
                    severity="FAIL", check_id="C-03",
                    message=f"state_hub_workstream_id {ws_id[:8]}… not found in DB (stale reference)",
                    file_path=fname,
                    db_id=ws_id,
                    fixable=False,
                )
            continue

        # C-09: repo mismatch — file is here but DB says different repo
        db_repo_id: str = ws.get("repo_id") or ""
        if db_repo_id != repo_id:
            report.add(
                severity="FAIL", check_id="C-09",
                message=(
                    f"Workstream '{ws.get('slug')}' repo_id in DB is "
                    f"{db_repo_id[:8] if db_repo_id else 'None'} "
                    f"but backing file lives in {repo_slug} ({repo_id[:8]}…)"
                ),
                file_path=fname,
                db_id=ws_id,
                file_value=repo_id,
                db_value=db_repo_id,
                fixable=True,
                _fix_context={"ws_id": ws_id, "correct_repo_id": repo_id},
            )
            # Continue to check drift even with mismatched repo

        tasks = get_tasks_from_workplan(meta, body)
        db_tasks = (
            snapshot_tasks_by_workplan.get(ws_id, [])
            if snapshot_available
            else _api_get(api_base, "/tasks", {"workplan_id": ws_id})
        )
        db_tasks_by_id = {
            str(row.get("id")): row
            for row in db_tasks
            if isinstance(row, dict) and row.get("id")
        } if isinstance(db_tasks, list) else {}
        file_task_statuses = [
            str(task.get("status", "")).strip()
            for task in tasks
            if not task.get("_parse_error")
        ]
        db_task_statuses = [
            str(task.get("status", "")).strip()
            for task in db_tasks
            if isinstance(db_tasks, list)
        ] if isinstance(db_tasks, list) else []
        active_task_requires_activation = should_activate_parent_for_active_tasks(
            parent_workstream_status=file_status,
            task_statuses=[*file_task_statuses, *db_task_statuses],
        )
        if active_task_requires_activation:
            # ADR-007 decision 3 / RMGR-WP-0005-T06: automation may report
            # that a proposed workplan has started work, but must not promote
            # proposed → active. proposed is a human review gate.
            proposed_gate = file_status == "proposed"
            report.add(
                severity="WARN",
                check_id="C-23",
                message=(
                    f"Lifecycle drift in '{ws.get('slug')}': workplan status "
                    f"{file_status!r} has an active task"
                    + (
                        " — report only (ADR-007: do not auto-promote proposed)"
                        if proposed_gate
                        else " — repair to 'active'"
                    )
                ),
                file_path=fname,
                db_id=ws_id,
                file_value=file_status,
                db_value=ws.get("status", ""),
                fixable=not proposed_gate,
                _fix_context={
                    "ws_id": ws_id,
                    "wp_file": str(wp_file),
                    "file_status": file_status,
                    "db_status": ws.get("status", ""),
                    "target_status": "active",
                },
            )

        # C-04: status drift — normalise file value before comparing so that
        # legacy file/API aliases are not treated as drift. Lifecycle repairs
        # take precedence so a stale planning file cannot regress active work.
        db_status = ws.get("status", "")
        normalised_db_status = normalise_workstream_status(db_status)
        if (
            not active_task_requires_activation
            and file_status
            and db_status
            and normalised_file_status != normalised_db_status
        ):
            report.add(
                severity="WARN", check_id="C-04",
                message=(
                    f"Status drift in '{ws.get('slug')}': "
                    f"file={file_status!r} db={db_status!r} (file wins)"
                ),
                file_path=fname,
                db_id=ws_id,
                file_value=file_status,
                db_value=db_status,
                fixable=True,
                _fix_context={
                    "ws_id": ws_id,
                    "field": "status",
                    "value": normalised_file_status,  # always send DB-valid value
                },
            )

        if normalised_file_status == "ready":
            review = ready_review_status(
                repo_dir,
                meta.get("reviewed_against_commit"),
                meta.get("context_paths"),
            )
            if review.needs_review:
                detail = f"Ready workplan may be stale: {review.reason}"
                if review.changed_paths:
                    preview = ", ".join(review.changed_paths[:5])
                    extra = "" if len(review.changed_paths) <= 5 else ", ..."
                    detail = f"{detail}; changed paths: {preview}{extra}"
                report.add(
                    severity="WARN",
                    check_id="C-21",
                    message=detail,
                    file_path=fname,
                    file_value=file_status,
                    db_value="needs_review",
                    fixable=False,
                )
            # C-34: soft DoR quality debt (STATE-WP-0077) — never blocks
            try:
                from quality_assessment import assessment_from_mapping, is_ok
            except ImportError:
                sys.path.insert(0, str(Path(__file__).resolve().parent))
                from quality_assessment import assessment_from_mapping, is_ok  # type: ignore
            dor = assessment_from_mapping(meta, "DoR")
            if not is_ok(dor):
                report.add(
                    severity="WARN",
                    check_id="C-34",
                    message=(
                        "status=ready without quality_dor DoR-Ok — process claim "
                        "without DoR assessment (soft; see docs/work-record-quality-gates.md)"
                    ),
                    file_path=fname,
                    file_value=str(dor or "unassessed"),
                    fixable=False,
                )

        # C-05: title drift
        db_title = ws.get("title", "")
        if file_title and db_title and file_title != db_title:
            report.add(
                severity="WARN", check_id="C-05",
                message=(
                    f"Title drift in '{ws.get('slug')}': "
                    f"file={file_title!r} db={db_title!r} (file wins)"
                ),
                file_path=fname,
                db_id=ws_id,
                file_value=file_title,
                db_value=db_title,
                fixable=True,
                _fix_context={"ws_id": ws_id, "field": "title", "value": file_title},
            )

        planning_priority = str(meta.get("planning_priority", "")).strip() or None
        if planning_priority != (ws.get("planning_priority") or None):
            report.add(
                severity="WARN", check_id="C-19",
                message=(
                    f"Planning priority drift in '{ws.get('slug')}': "
                    f"file={planning_priority!r} db={ws.get('planning_priority')!r} (file wins)"
                ),
                file_path=fname,
                db_id=ws_id,
                file_value=planning_priority,
                db_value=ws.get("planning_priority"),
                fixable=True,
                _fix_context={"ws_id": ws_id, "field": "planning_priority", "value": planning_priority},
            )

        planning_order = _as_int_or_none(meta.get("planning_order"))
        if planning_order != ws.get("planning_order"):
            report.add(
                severity="WARN", check_id="C-19",
                message=(
                    f"Planning order drift in '{ws.get('slug')}': "
                    f"file={planning_order!r} db={ws.get('planning_order')!r} (file wins)"
                ),
                file_path=fname,
                db_id=ws_id,
                file_value=planning_order,
                db_value=ws.get("planning_order"),
                fixable=True,
                _fix_context={"ws_id": ws_id, "field": "planning_order", "value": planning_order},
            )

        # C-10, C-11, C-12: task-level checks
        db_task_by_id: dict[str, dict] = {}
        if isinstance(db_tasks, list):
            for t in db_tasks:
                db_task_by_id[t["id"]] = t

        existing_deps = (
            snapshot_dependencies_by_workplan.get(ws_id, [])
            if snapshot_available
            else _api_get(api_base, f"/workplans/{ws_id}/dependencies") or []
        )
        existing_dep_keys = set()
        if isinstance(existing_deps, list):
            for dep in existing_deps:
                from_id = dep.get("from_workstream_id") or dep.get("from_workplan_id")
                if from_id != ws_id:
                    continue
                rel = dep.get("relationship_type") or "blocks"
                to_workplan_id = dep.get("to_workstream_id") or dep.get("to_workplan_id")
                if to_workplan_id:
                    existing_dep_keys.add(("workstream", to_workplan_id, rel))
                if dep.get("to_task_id"):
                    existing_dep_keys.add(("task", dep["to_task_id"], rel))

        for target_wp_id in _as_list(meta.get("depends_on_workplans")):
            target_ws_id = workplan_id_to_ws_id.get(target_wp_id)
            if not target_ws_id:
                report.add(
                    severity="WARN",
                    check_id="C-20",
                    message=f"Workplan dependency target '{target_wp_id}' is not linked to State Hub",
                    file_path=fname,
                    file_value=target_wp_id,
                    fixable=False,
                )
                continue
            dep_key = ("workstream", target_ws_id, "blocks")
            if dep_key not in existing_dep_keys:
                report.add(
                    severity="WARN",
                    check_id="C-20",
                    message=f"Missing DB dependency edge: {ws_id[:8]}… depends on workplan {target_wp_id}",
                    file_path=fname,
                    db_id=ws_id,
                    file_value=target_wp_id,
                    fixable=True,
                    _fix_context={
                        "from_workstream_id": ws_id,
                        "to_workstream_id": target_ws_id,
                        "relationship_type": "blocks",
                    },
                )

        for target_task_id in _as_list(meta.get("depends_on_tasks")):
            target_sh_id = task_file_id_to_sh_id.get(target_task_id)
            if not target_sh_id:
                report.add(
                    severity="WARN",
                    check_id="C-20",
                    message=f"Task dependency target '{target_task_id}' is not linked to State Hub",
                    file_path=fname,
                    file_value=target_task_id,
                    fixable=False,
                )
                continue
            dep_key = ("task", target_sh_id, "starts_after")
            if dep_key not in existing_dep_keys:
                report.add(
                    severity="WARN",
                    check_id="C-20",
                    message=f"Missing DB dependency edge: {ws_id[:8]}… starts after task {target_task_id}",
                    file_path=fname,
                    db_id=ws_id,
                    file_value=target_task_id,
                    fixable=True,
                    _fix_context={
                        "from_workstream_id": ws_id,
                        "to_task_id": target_sh_id,
                        "relationship_type": "starts_after",
                    },
                )

        file_task_sh_ids: set[str] = set()

        for task in tasks:
            if task.get("_parse_error"):
                continue
            t_id = str(task.get("id", "")).strip()
            _raw_sh = task.get("state_hub_task_id")
            t_sh_id = "" if _raw_sh is None else str(_raw_sh).strip().strip('"')
            if t_sh_id in ("~", "null", "None", "none"):
                t_sh_id = ""
            t_status = normalise_task_status(task.get("status", "todo"))

            if t_sh_id:
                file_task_sh_ids.add(t_sh_id)
                db_task = db_tasks_by_id.get(t_sh_id)
                if db_task is None:
                    if t_id and t_sh_id == _derived_work_record_uuid(t_id):
                        report.add(
                            severity="WARN",
                            check_id="C-11",
                            message=f"Derived task '{t_id}' is absent from this projection",
                            file_path=f"{fname}#{t_id}",
                            db_id=t_sh_id,
                            fixable=True,
                            _fix_context={
                                "ws_id": ws_id,
                                "ws_status": ws.get("status", ""),
                                "task": task,
                                "wp_file": str(wp_file),
                                "meta": meta,
                                "body": body,
                                "desired_task_id": t_sh_id,
                            },
                        )
                    else:
                        report.add(
                            severity="FAIL", check_id="C-03",
                            message=f"state_hub_task_id {t_sh_id[:8]}… not found in DB",
                            file_path=f"{fname}#{t_id}",
                            db_id=t_sh_id,
                            fixable=False,
                        )
                    continue
                # C-10 / C-15: task status drift. ADR-001: the file originates
                # work. Same-rank wait/progress is not "DB ahead" — writing
                # wait over progress/todo is the CFED/RMGR-WP-0004-T02 bug.
                # C-15 remains a report-only review signal when the DB is
                # terminal and the file is not.
                db_t_status = normalise_task_status(db_task.get("status", "todo"))
                if t_status and db_t_status and t_status != db_t_status:
                    db_rank = STATUS_ORDER.get(db_t_status, 0)
                    file_rank = STATUS_ORDER.get(t_status, 0)
                    db_terminal = db_t_status in {"done", "cancel"}
                    file_terminal = t_status in {"done", "cancel"}
                    if db_terminal and not file_terminal:
                        report.add(
                            severity="WARN", check_id="C-15",
                            message=(
                                f"DB task '{t_id}' is terminal "
                                f"(db={db_t_status!r}, file={t_status!r}) "
                                f"— report only; file wins (ADR-001 / RMGR-WP-0005-T06)"
                            ),
                            file_path=f"{fname}#{t_id}",
                            db_id=t_sh_id,
                            file_value=t_status,
                            db_value=db_t_status,
                            fixable=False,
                        )
                    # File always wins, including same-rank wait ↔ progress.
                    report.add(
                        severity="WARN", check_id="C-10",
                        message=(
                            f"Task status drift '{t_id}': "
                            f"file={t_status!r} db={db_t_status!r} (file wins)"
                        ),
                        file_path=f"{fname}#{t_id}",
                        db_id=t_sh_id,
                        file_value=t_status,
                        db_value=db_t_status,
                        fixable=True,
                        _fix_context={"task_id": t_sh_id, "status": t_status},
                    )
                file_description = task.get("description")
                if isinstance(file_description, str):
                    file_description = file_description.strip() or None
                else:
                    file_description = None
                db_description = db_task.get("description")
                if isinstance(db_description, str):
                    db_description = db_description.strip() or None
                else:
                    db_description = None
                if file_description and file_description != db_description:
                    report.add(
                        severity="WARN", check_id="C-22",
                        message=f"Task description drift '{t_id}': file content differs from DB (file wins)",
                        file_path=f"{fname}#{t_id}",
                        db_id=t_sh_id,
                        file_value=file_description[:120],
                        db_value=(db_description or "")[:120],
                        fixable=True,
                        _fix_context={"task_id": t_sh_id, "description": file_description},
                    )
            elif t_id:
                # C-11: task exists in file but not linked to DB
                ws_status = ws.get("status", "")
                report.add(
                    severity="WARN", check_id="C-11",
                    message=f"Task '{t_id}' has no state_hub_task_id",
                    file_path=f"{fname}#{t_id}",
                    fixable=True,
                    _fix_context={
                        "ws_id": ws_id,
                        "ws_status": ws_status,
                        "task": task,
                        "wp_file": str(wp_file),
                        "meta": meta,
                        "body": body,
                    },
                )

        # C-12: DB tasks with no file backing
        if isinstance(db_tasks, list):
            ws_status = ws.get("status", "")
            ws_finished = normalise_workstream_status(ws_status) in CLOSED_WORKSTREAM_STATUSES
            for db_t in db_tasks:
                if db_t["id"] not in file_task_sh_ids:
                    db_t_status = normalise_task_status(db_t.get("status", "todo"))
                    open_task = db_t_status not in TERMINAL_TASK_STATUSES
                    # Terminal orphans (cancel/done) are historical duplicates after
                    # file linkage or explicit retirement. The public task DELETE route
                    # is a cancel operation, so these are not further actionable.
                    if not open_task:
                        continue
                    # Auto-cancel fixable when workstream is finished and task is open.
                    fixable = ws_finished and open_task
                    report.add(
                        severity="WARN", check_id="C-12",
                        message=(
                            f"DB task '{db_t.get('title', '')}' "
                            f"(id={db_t['id'][:8]}…, status={db_t_status}) "
                            f"in workstream '{ws.get('slug')}' has no file backing"
                        ),
                        db_id=db_t["id"],
                        fixable=fixable,
                        _fix_context={
                            "task_id": db_t["id"],
                            "ws_finished": ws_finished,
                        },
                    )

        # C-13: all DB tasks done but workstream still active — worker forgot to close
        db_status = ws.get("status", "")
        if normalise_workstream_status(db_status) == "active" and isinstance(db_tasks, list) and db_tasks:
            non_terminal = [
                t for t in db_tasks
                if normalise_task_status(t.get("status", "todo")) not in TERMINAL_TASK_STATUSES
            ]
            if not non_terminal:
                report.add(
                    severity="WARN", check_id="C-13",
                    message=(
                        f"All {len(db_tasks)} DB tasks for '{ws.get('slug')}' are "
                        f"done/cancel but workstream status is still 'active' — "
                        f"worker likely forgot update_workstream_status()"
                    ),
                    file_path=fname,
                    db_id=ws_id,
                    db_value="active",
                    fixable=True,
                    _fix_context={
                        "ws_id": ws_id,
                        "field": "status",
                        "value": "finished",
                    },
                )

    # C-07 / C-08: orphan DB workstreams (have repo_id=this_repo but no backing file)
    _check_orphan_db(api_base, repo_id, set(file_ws_ids.keys()), report, active_file_ws_ids)

    # C-14: ghost duplicate — active workstream on same topic with no repo_id whose
    # title matches a file-backed workstream.  Root cause: create_workstream() called
    # before the workplan file was written; fix-consistency then created a second
    # workstream from the file, leaving the first as an invisible orphan.
    _check_ghost_duplicates(api_base, workplan_infos, file_ws_ids, report)

    _check_blocked_unblock_sweep(api_base, repo_slug, repo_dir, workplan_infos, report)
    _check_inbox_hygiene(api_base, repo_slug, report)
    _check_workplan_prefixes(repo_dir, repo_slug, workplan_infos, report)
    _check_workplan_id_collisions(api_base, repo_slug, repo_dir, workplan_infos, report)
    _check_scope_freshness(repo_dir, workplan_infos, report)

    _sync_workplan_bindings(api_base, repo_slug, workplan_infos, repo_dir, report)

    return report


def _sync_workplan_bindings(
    api_base: str,
    repo_slug: str,
    workplan_infos: list[tuple[Path, dict, str]],
    repo_dir: Path,
    report: ConsistencyReport,
) -> None:
    bindings: list[dict[str, Any]] = []
    for wp_file, meta, _ in workplan_infos:
        ws_id = str(meta.get("state_hub_workstream_id", "")).strip().strip('"')
        if not ws_id:
            continue
        archived = wp_file.parent.name == "archived"
        file_status = normalise_workstream_status(str(meta.get("status", "")).strip())
        bindings.append(
            {
                "workplan_id": ws_id,
                "filename": wp_file.name,
                "relative_path": workplan_display_path(repo_dir, wp_file),
                "repo_slug": repo_slug,
                "archived": archived,
                "status": file_status or None,
            }
        )
    if not bindings:
        return
    result = _api_put(api_base, "/workplans/index/bindings", {"bindings": bindings})
    if isinstance(result, dict) and "_error" in result:
        report.fixes_applied.append(f"bindings WARN: {result['_error']}")
    elif isinstance(result, dict):
        report.fixes_applied.append(
            f"bindings: synced {result.get('updated', 0)}/{result.get('received', len(bindings))}"
        )


def _check_orphan_db(
    api_base: str,
    repo_id: str,
    file_ws_ids: set[str],
    report: ConsistencyReport,
    active_file_ws_ids: set[str] | None = None,
) -> None:
    """Flag DB workstreams with repo_id=this_repo that have no backing workplan file."""
    active_file_ws_ids = active_file_ws_ids or file_ws_ids
    all_ws = _api_get(api_base, "/workplans", {"repo_id": repo_id})
    if not isinstance(all_ws, list):
        return
    for ws in all_ws:
        ws_id = ws["id"]
        ws_status = ws.get("status", "")
        normalised_status = normalise_workstream_status(ws_status)
        if normalised_status not in CLOSED_WORKSTREAM_STATUSES and ws_id in active_file_ws_ids:
            continue
        if normalised_status in CLOSED_WORKSTREAM_STATUSES and ws_id in file_ws_ids:
            continue
        ws_slug = ws.get("slug", "")
        if normalised_status not in CLOSED_WORKSTREAM_STATUSES:
            report.add(
                severity="FAIL", check_id="C-07",
                message=(
                    f"Non-closed DB workstream '{ws_slug}' (id={ws_id[:8]}…) "
                    f"has no backing workplan file — ADR-001 violation"
                ),
                db_id=ws_id,
                fixable=False,
            )
        elif normalised_status in CLOSED_WORKSTREAM_STATUSES:
            report.add(
                severity="INFO", check_id="C-08",
                message=(
                    f"Closed DB workstream '{ws_slug}' "
                    f"(id={ws_id[:8]}…, status={ws_status}) has no backing workplan file"
                ),
                db_id=ws_id,
                fixable=False,
            )


def _check_ghost_duplicates(
    api_base: str,
    workplan_infos: list[tuple],
    file_ws_ids: dict[str, tuple],
    report: ConsistencyReport,
) -> None:
    """C-14: detect active workstreams with repo_id=null whose title matches a
    file-backed workstream — these are ghosts created by premature create_workstream()
    calls before the workplan file existed.
    """
    # Build lookup: normalised title → file-backed workstream id
    file_titles: dict[str, str] = {}
    for _, meta, _ in workplan_infos:
        ws_id = str(meta.get("state_hub_workstream_id", "")).strip().strip('"')
        title = str(meta.get("title", "")).strip().lower()
        if title and ws_id:
            file_titles[title] = ws_id

    if not file_titles:
        return

    # Gather topic_ids from all file-backed workstreams so we can query by topic
    topic_ids: set[str] = set()
    for ws_id in file_ws_ids:
        ws = _api_get(api_base, f"/workplans/{ws_id}")
        if ws and ws.get("topic_id"):
            topic_ids.add(ws["topic_id"])

    for topic_id in topic_ids:
        topic_ws: list[dict] = []
        for status in OPEN_WORKSTREAM_STATUSES:
            status_rows = _api_get(api_base, "/workplans", {"topic_id": topic_id, "status": status})
            if isinstance(status_rows, list):
                topic_ws.extend(status_rows)
        for ws in topic_ws:
            ws_id = ws["id"]
            if ws_id in file_ws_ids:
                continue  # legitimately linked
            if ws.get("repo_id"):
                continue  # C-07 covers repo-scoped orphans
            ws_title = ws.get("title", "").strip().lower()
            if ws_title in file_titles:
                file_backed_id = file_titles[ws_title]
                report.add(
                    severity="WARN",
                    check_id="C-14",
                    message=(
                        f"Ghost duplicate: active workstream '{ws.get('slug')}' "
                        f"(id={ws_id[:8]}…, repo_id=null) has same title as "
                        f"file-backed workstream {file_backed_id[:8]}… — "
                        f"likely created via create_workstream() before workplan file existed; "
                        f"archive it"
                    ),
                    db_id=ws_id,
                    fixable=False,
                )


# ---------------------------------------------------------------------------
# Git sync (T02–T04: pull gate, writeback, push seal) — see repo_sync.py
# ---------------------------------------------------------------------------
# repo_sync.py owns the push-seal invariant and all git lifecycle primitives.
# The aliases below keep internal call sites and existing tests unchanged.

from repo_sync import (  # noqa: E402 (import after top-level imports intentional)
    count_local_ahead as _detect_ahead_of_remote,
    count_remote_ahead as _count_remote_ahead,
    pull_ff as _git_pull,
    push_ff as _git_push,
)


def _detect_behind_remote(repo_path: str) -> bool:
    """True if remote has commits the local repo lacks (C-16 predicate).

    Delegates to repo_sync.count_remote_ahead, which fetches before counting.
    Returns False on any error so C-16 is never spuriously triggered.
    """
    return _count_remote_ahead(repo_path) > 0


def _patch_task_status_in_file(
    file_path: Path, task_block_id: str, new_status: str
) -> bool:
    """Update the ``status:`` field inside a task block identified by its id.

    Only modifies lines inside a ```task … ``` fenced block whose ``id:``
    matches *task_block_id*.  Returns True if the file was changed.
    """
    text = file_path.read_text(encoding="utf-8")

    def _replace(m: re.Match) -> str:
        # m.group(0) is the full ```task...``` block including fences.
        # m.group(1) is the inner YAML content (no fences).
        full_block = m.group(0)
        inner = m.group(1).strip()
        task_meta = _parse_yaml_block(inner)
        if str(task_meta.get("id", "")).strip() != task_block_id:
            return full_block
        # Replace the status line only, leave everything else untouched.
        return re.sub(
            r"^(status:\s*)\S+",
            rf"\g<1>{new_status}",
            full_block,
            flags=re.MULTILINE,
        )

    new_text = _TASK_BLOCK_RE.sub(_replace, text)
    if new_text != text:
        file_path.write_text(new_text, encoding="utf-8")
        return True
    return False


def _git_identity_config() -> list[str]:
    import os as _os

    name = _os.environ.get("GIT_SYNC_USER_NAME", "custodian-sync")
    email = _os.environ.get("GIT_SYNC_USER_EMAIL", "custodian-sync@railiance.local")
    return ["-c", f"user.name={name}", "-c", f"user.email={email}"]


def _git_commit_writeback(
    repo_path: str,
    file_path: Path,
    changes: list[str],
    *,
    subject: str = "chore(consistency): sync task status from DB [auto]",
) -> bool:
    """Stage *file_path* and commit with a standard writeback message.

    Returns True on success, False on any error (errors are logged to stderr
    but do not abort the consistency run).
    """
    from datetime import date as _date
    summary = "\n".join(f"  - {c}" for c in changes)
    msg = (
        f"{subject}\n\n"
        f"Updated by fix-consistency on {_date.today().isoformat()}:\n"
        f"{summary}"
    )
    import os as _os
    # Pass GIT_CUSTODIAN_SYNC=1 so the post-commit hook can detect it is
    # running from within a sync pass and exit early, preventing re-entrancy.
    sync_env = {**_os.environ, "GIT_CUSTODIAN_SYNC": "1"}
    try:
        subprocess.run(
            ["git", "-C", repo_path, "add", str(file_path)],
            check=True, capture_output=True, env=sync_env,
        )
        subprocess.run(
            ["git", "-C", repo_path, *_git_identity_config(), "commit", "-m", msg],
            check=True, capture_output=True, env=sync_env,
        )
        return True
    except subprocess.CalledProcessError as e:
        print(
            f"WARN: git commit failed for writeback: {e.stderr.decode().strip()}",
            file=sys.stderr,
        )
        return False


# ---------------------------------------------------------------------------
# Worker orientation brief (.custodian-brief.md)
# ---------------------------------------------------------------------------

def default_wp_prefix(repo_slug: str) -> str:
    first = repo_slug.split("-", 1)[0].upper()
    return f"{first}-WP"


def infer_wp_prefix(repo_path: Path, repo_slug: str) -> str:
    """Prefer established on-disk workplan prefixes over first-token derivation."""
    counts: Counter[str] = Counter()
    workplans_dir = repo_path / "workplans"
    if workplans_dir.is_dir():
        for workplan in iter_workplan_files(workplans_dir):
            if workplan.name.startswith("ADHOC"):
                continue
            try:
                text = workplan.read_text(encoding="utf-8", errors="replace")
            except OSError:
                continue
            id_match = _WP_ID_PREFIX_RE.search(text) or _WP_ID_BARE_RE.search(text)
            if id_match:
                counts[id_match.group(1).upper()] += 1
                continue
            file_match = _WP_FILE_PREFIX_RE.match(workplan.name) or _WP_FILE_BARE_RE.match(workplan.name)
            if file_match:
                counts[file_match.group(1).upper()] += 1
    if not counts:
        return default_wp_prefix(repo_slug)
    return counts.most_common(1)[0][0]


def parse_blocked_on(value: str) -> str | None:
    if not value:
        return None
    match = _BLOCKED_ON_RE.match(value.strip())
    return match.group("agent").strip() if match else None


def _parse_message_timestamp(value: Any) -> datetime | None:
    if not value:
        return None
    if isinstance(value, datetime):
        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
    try:
        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except ValueError:
        return None
    return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)


def _summarise_inbox_message(msg: dict[str, Any], *, now: datetime) -> dict[str, Any]:
    created = _parse_message_timestamp(msg.get("created_at"))
    age_days = (now - created).days if created else None
    return {
        "id": str(msg.get("id", ""))[:8],
        "from_agent": msg.get("from_agent", ""),
        "subject": msg.get("subject", ""),
        "age_days": age_days,
    }


def collect_inbox_hygiene(api_base: str, repo_slug: str) -> dict[str, Any]:
    messages = _api_get(
        api_base,
        "/messages",
        {"to_agent": repo_slug, "unread_only": True, "limit": 100},
    ) or []
    if not isinstance(messages, list):
        return {
            "stale_unread": [],
            "missing_thread": [],
            "work_requests_unpromoted": [],
            "stale_unread_count": 0,
        }

    now = datetime.now(timezone.utc)
    stale_cutoff = now - timedelta(days=STALE_UNREAD_DAYS)
    thread_cutoff = now - timedelta(days=1)

    stale_unread: list[dict[str, Any]] = []
    missing_thread: list[dict[str, Any]] = []
    work_requests_unpromoted: list[dict[str, Any]] = []

    for msg in messages:
        created = _parse_message_timestamp(msg.get("created_at"))
        if created is None:
            continue
        summary = _summarise_inbox_message(msg, now=now)
        if created < stale_cutoff:
            stale_unread.append(summary)
        if not msg.get("thread_id") and created < thread_cutoff:
            missing_thread.append(summary)
        body = f"{msg.get('subject', '')} {msg.get('body', '')}"
        if _WORK_REQUEST_RE.search(body) and created < stale_cutoff:
            work_requests_unpromoted.append(summary)

    return {
        "stale_unread": stale_unread,
        "missing_thread": missing_thread,
        "work_requests_unpromoted": work_requests_unpromoted,
        "stale_unread_count": len(stale_unread),
    }


def _check_blocked_unblock_sweep(
    api_base: str,
    repo_slug: str,
    repo_dir: Path,
    workplan_infos: list[tuple[Path, dict, str]],
    report: ConsistencyReport,
) -> None:
    for wp_file, meta, _ in workplan_infos:
        file_status = normalise_workstream_status(str(meta.get("status", "")).strip())
        if file_status != "blocked":
            continue
        blocked_on = str(meta.get("blocked_on", "")).strip()
        counterpart = parse_blocked_on(blocked_on)
        if not counterpart:
            continue
        messages = _api_get(
            api_base,
            "/messages",
            {
                "to_agent": repo_slug,
                "from_agent": counterpart,
                "unread_only": True,
                "limit": 20,
            },
        ) or []
        if not isinstance(messages, list) or not messages:
            continue
        fname = workplan_display_path(repo_dir, wp_file)
        wp_id = str(meta.get("id", wp_file.stem)).strip()
        report.add(
            severity="WARN",
            check_id="C-25",
            message=(
                f"Blocked workplan '{wp_id}' waiting on {blocked_on!r} has "
                f"{len(messages)} unread message(s) from {counterpart} — "
                "blocker may have cleared"
            ),
            file_path=fname,
            file_value=blocked_on,
            fixable=False,
        )


def _check_inbox_hygiene(api_base: str, repo_slug: str, report: ConsistencyReport) -> None:
    hygiene = collect_inbox_hygiene(api_base, repo_slug)
    if hygiene["stale_unread_count"]:
        preview = ", ".join(
            f"{m['from_agent']}:{m['id']}" for m in hygiene["stale_unread"][:5]
        )
        extra = "" if hygiene["stale_unread_count"] <= 5 else ", ..."
        report.add(
            severity="WARN",
            check_id="C-28",
            message=(
                f"{hygiene['stale_unread_count']} unread inbox message(s) older than "
                f"{STALE_UNREAD_DAYS} day(s): {preview}{extra}"
            ),
            fixable=False,
        )
    if hygiene["missing_thread"]:
        preview = ", ".join(
            f"{m['from_agent']}:{m['id']}" for m in hygiene["missing_thread"][:5]
        )
        report.add(
            severity="WARN",
            check_id="C-28",
            message=(
                f"{len(hygiene['missing_thread'])} unread message(s) lack thread_id "
                f"for supersession tracking: {preview}"
            ),
            fixable=False,
        )
    for msg in hygiene["work_requests_unpromoted"]:
        report.add(
            severity="WARN",
            check_id="C-29",
            message=(
                f"Unread work request from {msg['from_agent']} ({msg['id']}) may need "
                f"a workplan file: {msg['subject'][:120]}"
            ),
            fixable=False,
        )


def _check_workplan_prefixes(
    repo_dir: Path,
    repo_slug: str,
    workplan_infos: list[tuple[Path, dict, str]],
    report: ConsistencyReport,
) -> None:
    canonical = infer_wp_prefix(repo_dir, repo_slug)
    for wp_file, meta, _ in workplan_infos:
        if wp_file.name.startswith("ADHOC"):
            continue
        wp_id = str(meta.get("id", "")).strip()
        if not wp_id:
            continue
        match = _WP_ID_PREFIX_RE.match(wp_id) or _WP_ID_BARE_RE.match(wp_id)
        if not match:
            continue
        prefix = match.group(1).upper()
        if prefix == canonical.upper():
            continue
        report.add(
            severity="WARN",
            check_id="C-26",
            message=(
                f"Workplan id '{wp_id}' uses prefix '{prefix}' but repo canonical "
                f"prefix is '{canonical}' — new plans must conform"
            ),
            file_path=workplan_display_path(repo_dir, wp_file),
            file_value=wp_id,
            db_value=canonical,
            fixable=False,
        )


def _scan_repo_workplan_ids(repos: list[dict[str, Any]]) -> dict[str, list[tuple[str, str]]]:
    id_map: dict[str, list[tuple[str, str]]] = {}
    for repo in repos:
        slug = repo["slug"]
        path = resolve_repo_path(repo)
        if not path or not Path(path).is_dir():
            continue
        workplans_dir = Path(path) / "workplans"
        if not workplans_dir.is_dir():
            continue
        for wp_file in iter_workplan_files(workplans_dir):
            if wp_file.name.startswith("ADHOC"):
                continue
            try:
                text = wp_file.read_text(encoding="utf-8")
            except OSError:
                continue
            if not text.startswith("---"):
                continue
            meta, _ = parse_frontmatter(text)
            if not meta or meta.get("_parse_error"):
                continue
            wp_id = str(meta.get("id", "")).strip()
            if wp_id:
                id_map.setdefault(wp_id, []).append((slug, wp_file.name))
    return id_map


def _check_workplan_id_collisions(
    api_base: str,
    repo_slug: str,
    repo_dir: Path,
    workplan_infos: list[tuple[Path, dict, str]],
    report: ConsistencyReport,
) -> None:
    repos = _api_get(api_base, "/repos") or []
    if not isinstance(repos, list):
        return
    id_map = _scan_repo_workplan_ids(repos)
    for wp_file, meta, _ in workplan_infos:
        wp_id = str(meta.get("id", "")).strip()
        if not wp_id:
            continue
        locations = id_map.get(wp_id, [])
        if len(locations) <= 1:
            continue
        report.add(
            severity="WARN",
            check_id="C-27",
            message=(
                f"Workplan id '{wp_id}' collides across repos: "
                + ", ".join(f"{slug}/{fname}" for slug, fname in locations)
            ),
            file_path=workplan_display_path(repo_dir, wp_file),
            file_value=wp_id,
            fixable=False,
        )


def _scope_current_state_lines(repo_dir: Path) -> dict[str, str]:
    scope_path = repo_dir / "SCOPE.md"
    if not scope_path.exists():
        return {}
    text = scope_path.read_text(encoding="utf-8", errors="replace")
    match = re.search(r"## Current State\s*\n(.*?)(?:\n## |\Z)", text, re.DOTALL)
    if not match:
        return {}
    result: dict[str, str] = {}
    for line in match.group(1).splitlines():
        stripped = line.strip()
        if not stripped.startswith("-"):
            continue
        parts = stripped[1:].split(":", 1)
        if len(parts) != 2:
            continue
        result[parts[0].strip().lower()] = parts[1].strip()
    return result


def _repo_has_open_workplans(workplan_infos: list[tuple[Path, dict, str]]) -> bool:
    for wp_file, meta, _ in workplan_infos:
        if wp_file.parent.name == "archived":
            continue
        status = normalise_workstream_status(str(meta.get("status", "")).strip())
        if status in _OPEN_WORKPLAN_STATUSES:
            return True
    return False


def _check_scope_freshness(
    repo_dir: Path,
    workplan_infos: list[tuple[Path, dict, str]],
    report: ConsistencyReport,
) -> None:
    scope = _scope_current_state_lines(repo_dir)
    if not scope:
        return
    has_open_workplans = _repo_has_open_workplans(workplan_infos)
    scope_status = scope.get("status", "").lower()
    warnings: list[str] = []
    if "active" in scope_status and not has_open_workplans:
        warnings.append(
            "SCOPE.md Current State says active but no open workplans remain — may be stale"
        )
    if has_open_workplans and any(
        token in scope_status for token in ("finished", "archived", "idle", "dormant")
    ):
        warnings.append(
            "SCOPE.md Current State does not reflect active workplans — may be stale"
        )
    implementation = scope.get("implementation", "").lower()
    if has_open_workplans and implementation and "not yet started" in implementation:
        warnings.append(
            "SCOPE.md mentions work not yet started but active workplans exist — may be stale"
        )
    for warning in warnings:
        report.add(
            severity="WARN",
            check_id="C-30",
            message=warning,
            file_path="SCOPE.md",
            fixable=False,
        )


_BRIEF_HEADER = "<!-- custodian-brief: generated by fix-consistency — do not edit manually -->"
_TASK_STATUS_ICON = {"done": "✓", "cancel": "✗", "progress": "►", "wait": "!", "todo": "·"}
_OPEN_STATUSES = set(OPEN_TASK_STATUSES)


def _commit_working_memory_notes(repo_slug: str, repo_path: str) -> bool:
    """Stage and commit activity-core daily-triage working-memory notes.

    Only the-custodian carries the fleet working-memory sink. Notes are
    repo-relative under memory/working/ and are written by activity-core on
    Railiance via hostPath into the clone checkout.
    """
    if repo_slug != "the-custodian":
        return False

    wm_dir = Path(repo_path) / "memory" / "working"
    if not wm_dir.is_dir():
        return False

    notes = sorted(wm_dir.glob("daily-triage-*.md"))
    if not notes:
        return False

    import os as _os

    sync_env = {**_os.environ, "GIT_CUSTODIAN_SYNC": "1"}
    rel_paths = [str(note.relative_to(repo_path)) for note in notes]
    try:
        subprocess.run(
            ["git", "-C", repo_path, "add", "--", *rel_paths],
            check=True,
            capture_output=True,
            env=sync_env,
        )
        staged = subprocess.run(
            ["git", "-C", repo_path, "diff", "--cached", "--quiet"],
            capture_output=True,
            env=sync_env,
        )
        if staged.returncode == 0:
            return False
        from datetime import date as _date

        msg = (
            "chore(working-memory): sync daily-triage notes from activity-core "
            f"[auto {_date.today().isoformat()}]"
        )
        subprocess.run(
            ["git", "-C", repo_path, *_git_identity_config(), "commit", "-m", msg],
            check=True,
            capture_output=True,
            env=sync_env,
        )
        return True
    except subprocess.CalledProcessError as e:
        print(
            f"WARN: working-memory commit failed: {e.stderr.decode().strip()}",
            file=sys.stderr,
        )
        return False


def _write_custodian_brief(api_base: str, repo_slug: str, repo_path: str) -> bool:
    """Generate .custodian-brief.md at the repo root and git-commit if changed.

    The brief gives any agent — including subagents without MCP access and
    workers on remote machines — instant orientation without a live hub
    connection.  Returns True if the file was written (content changed).
    """
    import datetime as _dt
    from datetime import timezone as _tz

    repo = _api_get(api_base, f"/repos/{repo_slug}")
    if not repo:
        return False

    repo_id: str = repo.get("id", "")
    domain_slug = str(repo.get("domain_slug") or "")

    # The repository projection owns domain identity. Retain the workplan/topic
    # lookup only for compatibility with older API responses that omitted it.
    workstreams: list[dict] = []
    for status in OPEN_WORKSTREAM_STATUSES:
        rows = _api_get(api_base, "/workplans", {"repo_id": repo_id, "status": status}) or []
        if isinstance(rows, list):
            workstreams.extend(rows)
    _ws_for_domain = workstreams if workstreams else []
    if not _ws_for_domain:
        all_ws = _api_get(api_base, "/workplans", {"repo_id": repo_id}) or []
        _ws_for_domain = all_ws if isinstance(all_ws, list) else []
    if not domain_slug and _ws_for_domain:
        topic = _api_get(api_base, f"/topics/{_ws_for_domain[0].get('topic_id', '')}")
        if topic:
            domain_slug = topic.get("domain_slug", "")

    # Active repo goal (first active one if multiple)
    goal_text = ""
    goals = _api_get(api_base, "/repo-goals", {"repo_slug": repo_slug}) or []
    if isinstance(goals, list):
        active_goals = [g for g in goals if g.get("status") == "active"]
        if active_goals:
            g = active_goals[0]
            goal_text = g.get("title", "") or g.get("description", "")

    now_utc = _dt.datetime.now(_tz.utc)
    ts = now_utc.strftime("%Y-%m-%d %H:%M UTC")

    lines = [
        _BRIEF_HEADER,
        f"# Custodian Brief — {repo_slug}",
        "",
        f"**Domain:** {domain_slug or '(unknown)'}  ",
        f"**Last synced:** {ts}  ",
        "**State Hub:** http://127.0.0.1:8000 *(adjust if running on a remote machine)*",
        "",
    ]

    if goal_text:
        lines += ["## Current Goal", "", goal_text, ""]

    if isinstance(workstreams, list) and workstreams:
        lines.append("## Active Workstreams")
        for ws in workstreams:
            ws_title = ws.get("title", ws.get("slug", "?"))
            ws_id = ws["id"]
            tasks = _api_get(api_base, "/tasks", {"workplan_id": ws_id}) or []
            if not isinstance(tasks, list):
                tasks = []

            done = sum(
                1 for t in tasks
                if normalise_task_status(t.get("status", "todo")) in TERMINAL_TASK_STATUSES
            )
            total = len(tasks)
            pct = f"{done}/{total}" if total else "no tasks"

            open_tasks = [
                t for t in tasks
                if normalise_task_status(t.get("status", "todo")) in _OPEN_STATUSES
            ]
            # Show wait first, then progress, then todo (cap at 5).
            priority_order = {"wait": 0, "progress": 1, "todo": 2}
            open_tasks.sort(
                key=lambda t: priority_order.get(
                    normalise_task_status(t.get("status", "todo")), 9
                )
            )

            lines += [
                "",
                f"### {ws_title}",
                f"Progress: {pct} done  |  workplan_id: `{ws_id}`",
            ]

            if open_tasks:
                lines.append("")
                lines.append("**Open tasks:**")
                for t in open_tasks[:7]:
                    status = normalise_task_status(t.get("status", "todo"))
                    icon = _TASK_STATUS_ICON.get(status, "·")
                    title = t.get("title", t["id"])
                    tid = t["id"]
                    blocker = t.get("blocking_reason", "")
                    task_line = f"- {icon} {title}  `{tid[:8]}`"
                    if status == "wait" and blocker:
                        task_line += f"\n  *(wait: {blocker})*"
                    lines.append(task_line)
                if len(open_tasks) > 7:
                    lines.append(f"- … and {len(open_tasks) - 7} more open tasks")
    else:
        lines += ["## Active Workstreams", "", "*(none — repo may need first-session setup)*"]

    hygiene = collect_inbox_hygiene(api_base, repo_slug)
    if hygiene["stale_unread_count"] or hygiene["missing_thread"] or hygiene["work_requests_unpromoted"]:
        lines += ["", "## Inbox Hygiene", ""]
        if hygiene["stale_unread_count"]:
            lines.append(
                f"**Stale unread:** {hygiene['stale_unread_count']} message(s) older than "
                f"{STALE_UNREAD_DAYS} day(s) — triage at session start."
            )
        if hygiene["missing_thread"]:
            lines.append(
                f"**Missing thread_id:** {len(hygiene['missing_thread'])} unread message(s) "
                "lack supersession chains."
            )
        for msg in hygiene["work_requests_unpromoted"][:3]:
            lines.append(
                f"- ! {msg['from_agent']}: {msg['subject'][:100]}  `{msg['id']}`"
            )

    blocked_warnings: list[str] = []
    workplans_dir = Path(repo_path) / "workplans"
    if workplans_dir.is_dir():
        for wp_file in iter_workplan_files(workplans_dir):
            try:
                text = wp_file.read_text(encoding="utf-8")
            except OSError:
                continue
            if not text.startswith("---"):
                continue
            meta, _ = parse_frontmatter(text)
            if not meta or meta.get("_parse_error"):
                continue
            if normalise_workstream_status(str(meta.get("status", "")).strip()) != "blocked":
                continue
            blocked_on = str(meta.get("blocked_on", "")).strip()
            counterpart = parse_blocked_on(blocked_on)
            if not counterpart:
                continue
            messages = _api_get(
                api_base,
                "/messages",
                {
                    "to_agent": repo_slug,
                    "from_agent": counterpart,
                    "unread_only": True,
                    "limit": 5,
                },
            ) or []
            if isinstance(messages, list) and messages:
                wp_id = str(meta.get("id", wp_file.stem)).strip()
                blocked_warnings.append(
                    f"- ! **{wp_id}** — blocker may have cleared: "
                    f"{len(messages)} unread message(s) from `{counterpart}`"
                )
    if blocked_warnings:
        lines += ["", "## Blocked Workplans", ""]
        lines.extend(blocked_warnings)

    scope_warnings: list[str] = []
    scope = _scope_current_state_lines(Path(repo_path))
    has_open_workplans = False
    if workplans_dir.is_dir():
        for wp_file in iter_workplan_files(workplans_dir):
            try:
                text = wp_file.read_text(encoding="utf-8")
            except OSError:
                continue
            if not text.startswith("---"):
                continue
            meta, _ = parse_frontmatter(text)
            if not meta or meta.get("_parse_error"):
                continue
            if wp_file.parent.name == "archived":
                continue
            status = normalise_workstream_status(str(meta.get("status", "")).strip())
            if status in _OPEN_WORKPLAN_STATUSES:
                has_open_workplans = True
                break
    if scope:
        scope_status = scope.get("status", "").lower()
        if "active" in scope_status and not has_open_workplans:
            scope_warnings.append(
                "SCOPE.md says active but no open workplans remain — refresh Current State."
            )
        if has_open_workplans and any(
            token in scope_status for token in ("finished", "archived", "idle", "dormant")
        ):
            scope_warnings.append(
                "SCOPE.md Current State does not reflect active workplans — may be stale."
            )
    if scope_warnings:
        lines += ["", "## SCOPE Freshness", ""]
        lines.extend(f"- {warning}" for warning in scope_warnings)

    lines += [
        "",
        "---",
        "## MCP Orientation (when available)",
        "",
        "If the state-hub MCP server is reachable, call:",
        f"`get_domain_summary(\"{domain_slug}\")`",
        "This provides richer cross-domain context.",
        "If the MCP call fails, use this file as your orientation source.",
    ]

    content = "\n".join(lines) + "\n"

    brief_path = Path(repo_path) / ".custodian-brief.md"
    existing = brief_path.read_text(encoding="utf-8") if brief_path.exists() else ""

    # Strip the timestamp line before comparing to avoid spurious writes
    def _strip_ts(text: str) -> str:
        return "\n".join(
            ln for ln in text.splitlines()
            if not ln.startswith("**Last synced:**")
        )

    if _strip_ts(content) == _strip_ts(existing):
        return False  # no meaningful change

    brief_path.write_text(content, encoding="utf-8")

    # Commit the brief so remote workers can pull it
    _git_commit_writeback(
        repo_path,
        brief_path,
        [f"update .custodian-brief.md for {repo_slug}"],
    )
    return True


# ---------------------------------------------------------------------------
# Fix engine
# ---------------------------------------------------------------------------

def fix_repo(
    api_base: str,
    repo_slug: str,
    repo_path_override: str | None = None,
    no_writeback: bool = False,
    bootstrap_empty_projection: bool = False,
) -> ConsistencyReport:
    """Run checks then apply all auto-fixable issues. Returns updated report."""
    report = check_repo(
        api_base,
        repo_slug,
        repo_path_override,
        bootstrap_empty_projection=bootstrap_empty_projection,
    )
    if any(i.check_id in {"C-00", "C-36"} 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
    if repo_path:
        repo_record = _api_get(api_base, f"/repos/{repo_slug}")
        if repo_record:
            hostname = _sweep_hostname()
            if (repo_record.get("host_paths") or {}).get(hostname) != repo_path:
                result = _api_post(
                    api_base, f"/repos/{repo_slug}/paths",
                    {"host": hostname, "path": repo_path},
                )
                if result and "_error" not in result:
                    report.fixes_applied.append(
                        f"host_paths[{hostname}] → {repo_path}"
                    )

    # T02 — pull gate: warn and skip all write operations when local repo is
    # behind its remote tracking branch.
    if repo_path and _detect_behind_remote(repo_path):
        report.add(
            severity="WARN", check_id="C-16",
            message=(
                f"Repo '{repo_slug}' is behind its remote tracking branch — "
                f"pull before fixing to avoid clobbering remote progress. "
                f"Run: git -C {repo_path} pull --ff-only"
            ),
            fixable=False,
        )
        report.fixes_applied.append(
            "C-16: all write operations skipped — local repo is behind remote"
        )
        return report

    # C-17: backlog guard — if local has unpushed commits from a prior failed push,
    # try to push them before making more. Skipping writes prevents runaway divergence.
    if repo_path:
        ahead = _detect_ahead_of_remote(repo_path)
        if ahead > 0:
            push_ok, push_msg = _git_push(repo_path)
            if not push_ok:
                report.add(
                    severity="WARN", check_id="C-17",
                    message=(
                        f"Repo '{repo_slug}' has {ahead} unpushed commit(s) and push "
                        f"failed ({push_msg}) — skipping writes to prevent runaway divergence"
                    ),
                    fixable=False,
                )
                report.fixes_applied.append(
                    "C-17: all write operations skipped — unpushed commits, push failed"
                )
                return report
            # Push succeeded — local is now in sync; proceed normally
            report.fixes_applied.append(f"C-17 cleared: pushed {ahead} backlogged commit(s)")

    fixable = [i for i in report.issues if i.fixable]

    for issue in fixable:
        ctx = issue._fix_context
        try:
            if issue.check_id in ("C-04", "C-05", "C-13", "C-19"):
                ws_id = ctx["ws_id"]
                result = _api_patch(api_base, f"/workplans/{ws_id}",
                                    {ctx["field"]: ctx["value"]})
                if result is not None and "_error" not in result:
                    report.fixes_applied.append(
                        f"{issue.check_id} fixed: workstream {ws_id[:8]}… "
                        f"{ctx['field']} → {ctx['value']!r}"
                    )
                elif result is not None:
                    report.fixes_applied.append(
                        f"{issue.check_id} FAILED: workstream {ws_id[:8]}… "
                        f"{ctx['field']} → {ctx['value']!r}: {result['_error']}"
                    )

            elif issue.check_id == "C-23":
                ws_id = ctx["ws_id"]
                target_status = ctx["target_status"]
                result = _api_patch(api_base, f"/workplans/{ws_id}", {"status": target_status})
                if result is not None and "_error" not in result:
                    report.fixes_applied.append(
                        f"C-23 fixed: workstream {ws_id[:8]}… status → {target_status!r}"
                    )
                elif result is not None:
                    report.fixes_applied.append(
                        f"C-23 FAILED: workstream {ws_id[:8]}… "
                        f"status → {target_status!r}: {result['_error']}"
                    )

                if no_writeback:
                    report.fixes_applied.append(
                        f"C-23 skipped file repair (--no-writeback): {Path(ctx['wp_file']).name}"
                    )
                else:
                    wp_file = Path(ctx["wp_file"])
                    old_status = ctx["file_status"]
                    if _patch_frontmatter_field(wp_file, "status", target_status):
                        committed = _git_commit_writeback(
                            repo_path,
                            wp_file,
                            [f"workplan status: {old_status} → {target_status}"],
                            subject="chore(consistency): renormalize lifecycle state [auto]",
                        )
                        suffix = " (committed)" if committed else " (file patched, commit failed)"
                        report.fixes_applied.append(
                            f"C-23 fixed: {wp_file.name} status "
                            f"{old_status} → {target_status}{suffix}"
                        )
                    else:
                        report.fixes_applied.append(
                            f"C-23 SKIP: {wp_file.name} already has status {target_status!r}"
                        )

            elif issue.check_id == "C-06":
                wp_file = Path(ctx["wp_file"])
                meta = ctx["meta"]
                domain = ctx["domain"]
                repo_market_domain = str(ctx.get("repo_market_domain") or "").strip()
                repo_id_val = ctx["repo_id"]
                body = ctx.get("body", "")
                wp_id = str(meta.get("id", "")).strip()
                desired_ws_id = str(
                    ctx.get("desired_ws_id") or _derived_work_record_uuid(wp_id)
                )
                title = str(meta.get("title", "")).strip()
                status = str(meta.get("status", "active")).strip()
                status = normalise_workstream_status(status)
                if status not in VALID_WP_STATUSES:
                    status = "active"

                # Find topic_id — workplan frontmatter may still use legacy
                # coordination slugs (e.g. custodian); map to market domain first.
                topic_domain = resolve_topic_domain_slug(
                    domain,
                    repo_market_domain=repo_market_domain or None,
                )
                topics = _api_get(api_base, "/topics")
                topic_id = None
                if isinstance(topics, list):
                    for t in topics:
                        if t.get("domain_slug") == topic_domain:
                            topic_id = t["id"]
                            break
                if topic_id is None:
                    report.fixes_applied.append(
                        f"C-06 SKIP {wp_id}: no topic found for domain "
                        f"'{topic_domain}' (workplan domain={domain!r})"
                    )
                    continue

                base_slug = re.sub(r"[^a-z0-9-]", "-", wp_id.lower()).strip("-") or "workplan"
                repo_slug_part = re.sub(
                    r"[^a-z0-9-]", "-", str(ctx.get("repo_slug") or "").lower()
                ).strip("-")
                slug_candidates = [base_slug]
                repo_qualified_slug = base_slug
                if repo_slug_part and not base_slug.startswith(f"{repo_slug_part}-"):
                    repo_qualified_slug = f"{repo_slug_part}-{base_slug}"
                    slug_candidates.append(repo_qualified_slug)
                for suffix in range(2, 21):
                    slug_candidates.append(f"{repo_qualified_slug}-{suffix}")

                ws_data = None
                last_error = None
                for slug in slug_candidates:
                    existing = _api_get(api_base, "/workplans", {"slug": slug}, return_error=True)
                    if isinstance(existing, dict) and "_error" in existing:
                        last_error = existing["_error"]
                        continue
                    if isinstance(existing, list) and existing:
                        existing_same_repo = next(
                            (w for w in existing if w.get("repo_id") == repo_id_val),
                            None,
                        )
                        if (
                            existing_same_repo
                            and existing_same_repo.get("title") == (title or wp_id)
                            and existing_same_repo.get("id") == desired_ws_id
                        ):
                            ws_data = existing_same_repo
                            report.fixes_applied.append(
                                f"C-06 reusing existing workstream {ws_data['id'][:8]}... for {wp_id}"
                            )
                            break
                        last_error = (
                            f"slug {slug!r} already belongs to a different workplan identity"
                        )
                        continue

                    ws_data = _api_post(api_base, "/workplans", {
                        "id": desired_ws_id,
                        "topic_id": topic_id,
                        "repo_id": repo_id_val,
                        "slug": slug,
                        "title": title or wp_id,
                        "status": status,
                        "owner": str(meta.get("owner", "")).strip() or None,
                        "planning_priority": str(meta.get("planning_priority", "")).strip() or None,
                        "planning_order": _as_int_or_none(meta.get("planning_order")),
                    })
                    if ws_data is None or (isinstance(ws_data, dict) and "_error" in ws_data):
                        last_error = ws_data.get("_error") if isinstance(ws_data, dict) else "no response"
                        ws_data = None
                        continue
                    break

                if ws_data is None:
                    report.fixes_applied.append(
                        f"C-06 FAIL {wp_id}: could not create workstream in DB: {last_error or 'no usable slug'}"
                    )
                    continue

                new_ws_id = ws_data["id"]
                if new_ws_id != desired_ws_id:
                    report.fixes_applied.append(
                        f"C-06 FAIL {wp_id}: projection returned unexpected UUID {new_ws_id}"
                    )
                    continue
                if not meta.get("state_hub_workstream_id"):
                    _add_frontmatter_field(wp_file, "state_hub_workstream_id", new_ws_id)
                report.fixes_applied.append(
                    f"C-06 fixed: created workstream {new_ws_id[:8]}… "
                    f"for {wp_id}, wrote ID to {wp_file.name}"
                )

                # Create tasks and inject IDs
                tasks = get_tasks_from_workplan(meta, body)
                for task in tasks:
                    if task.get("_parse_error"):
                        continue
                    t_id = str(task.get("id", "")).strip()
                    if not t_id:
                        continue
                    t_status = normalise_task_status(task.get("status", "todo"))
                    t_priority = str(task.get("priority", "medium")).strip()
                    if t_priority not in VALID_TASK_PRIORITIES:
                        t_priority = "medium"
                    raw_task_id = task.get("state_hub_task_id")
                    desired_task_id = (
                        str(raw_task_id).strip().strip('"')
                        if raw_task_id not in (None, "", "~", "null", "None", "none")
                        else _derived_work_record_uuid(t_id)
                    )
                    t_data = _api_post(api_base, "/tasks", {
                        "id": desired_task_id,
                        "workplan_id": new_ws_id,
                        "title": str(task.get("title", t_id)).strip() or t_id,
                        "description": task.get("description") or None,
                        "status": t_status,
                        "priority": t_priority,
                        "assignee": task.get("assignee") or None,
                    })
                    if t_data and "_error" not in t_data:
                        t_db_id = t_data["id"]
                        if t_db_id != desired_task_id:
                            report.fixes_applied.append(
                                f"  ! task {t_id} returned unexpected UUID {t_db_id}"
                            )
                            continue
                        if not raw_task_id:
                            injected = _inject_task_id_into_block(
                                wp_file, "state_hub_task_id", t_db_id, t_id
                            )
                            if not injected:
                                _inject_task_id_frontmatter_list(wp_file, t_db_id, t_id)
                        report.fixes_applied.append(f"  + task {t_id} → {t_db_id[:8]}…")
                    elif t_data:
                        report.fixes_applied.append(
                            f"  ! task {t_id} not created: {t_data.get('_error', t_data)}"
                        )

            elif issue.check_id == "C-32":
                md_path = ctx["md_path"]
                kind = ctx["kind"]
                rid = ctx["rid"]
                meta = ctx["meta"]
                id_field = _WORK_RECORD_ID_FIELD_BY_KIND[kind]
                title = str(meta.get("title") or rid)
                desired_record_id = ctx.get("desired_record_id")

                repo_record = _api_get(api_base, f"/repos/{repo_slug}")
                if not repo_record:
                    report.fixes_applied.append(
                        f"C-32 SKIP {rid}: could not look up repo '{repo_slug}'"
                    )
                    continue
                created = None
                if kind == "intake":
                    payload = {
                        "id": desired_record_id,
                        "title": title,
                        "repo_id": repo_record.get("id"),
                        "description": meta.get("description"),
                        "lane": meta.get("lane", "green"),
                        "origin": meta.get("origin"),
                        "origin_ref": meta.get("origin_ref"),
                        "source_repo_path": str(md_path.relative_to(Path(report.repo_path))),
                    }
                    created = _api_post(api_base, "/intakes", payload)
                elif kind == "decision":
                    repo_market_domain = str(repo_record.get("domain_slug") or "").strip()
                    topic_domain = resolve_topic_domain_slug(
                        repo_market_domain, repo_market_domain=repo_market_domain or None
                    )
                    topics = _api_get(api_base, "/topics")
                    topic_id = None
                    if isinstance(topics, list):
                        for t in topics:
                            if t.get("domain_slug") == topic_domain:
                                topic_id = t["id"]
                                break
                    if topic_id is None:
                        report.fixes_applied.append(
                            f"C-32 SKIP {rid}: no topic found for domain '{topic_domain}'"
                        )
                        continue
                    payload = {
                        "id": desired_record_id,
                        "title": title,
                        "topic_id": topic_id,
                        "decision_type": meta.get("decision_type", "pending"),
                        "description": meta.get("description"),
                        "rationale": meta.get("agent_recommendation") or meta.get("rationale"),
                    }
                    created = _api_post(api_base, "/decisions", payload)

                if not created or "_error" in created:
                    report.fixes_applied.append(
                        f"C-32 FAIL {rid}: could not create {kind} in DB: "
                        f"{(created or {}).get('_error', 'no response')}"
                    )
                    continue

                new_id = created["id"]
                if desired_record_id and new_id != desired_record_id:
                    report.fixes_applied.append(
                        f"C-32 FAIL {rid}: projection returned unexpected UUID {new_id}"
                    )
                    continue
                if desired_record_id:
                    report.fixes_applied.append(
                        f"C-32 fixed: created {kind} {new_id[:8]}… for {rid} "
                        "with the authoritative ID"
                    )
                    continue
                if _inject_yaml_block_field(md_path, id_field, new_id, rid):
                    report.fixes_applied.append(
                        f"C-32 fixed: created {kind} {new_id[:8]}… for {rid}, "
                        f"wrote {id_field} to {md_path.name}"
                    )
                else:
                    report.fixes_applied.append(
                        f"C-32 WARN {rid}: created {kind} {new_id[:8]}… but "
                        f"failed to write {id_field} back to {md_path.name}"
                    )

            elif issue.check_id == "C-33":
                ctx["index_file"].write_text(ctx["content"], encoding="utf-8")
                report.fixes_applied.append(
                    f"C-33 fixed: regenerated {ctx['index_file'].name}"
                )

            elif issue.check_id == "C-09":
                ws_id = ctx["ws_id"]
                correct_repo_id = ctx["correct_repo_id"]
                result = _api_patch(api_base, f"/workplans/{ws_id}",
                                    {"repo_id": correct_repo_id})
                if result is not None:
                    report.fixes_applied.append(
                        f"C-09 fixed: workstream {ws_id[:8]}… "
                        f"repo_id → {correct_repo_id[:8]}…"
                    )

            elif issue.check_id == "C-20":
                from_workstream_id = ctx["from_workstream_id"]
                body = {
                    "to_workstream_id": ctx.get("to_workstream_id"),
                    "to_task_id": ctx.get("to_task_id"),
                    "relationship_type": ctx["relationship_type"],
                }
                result = _api_post(api_base, f"/workplans/{from_workstream_id}/dependencies", body)
                if result is not None and "_error" not in result:
                    target = ctx.get("to_workstream_id") or ctx.get("to_task_id")
                    report.fixes_applied.append(
                        f"C-20 fixed: dependency {from_workstream_id[:8]}… "
                        f"{ctx['relationship_type']} → {target[:8]}…"
                    )
                elif result is not None:
                    report.fixes_applied.append(
                        f"C-20 FAILED: {result['_error']}"
                    )

            elif issue.check_id == "C-10":
                task_id = ctx["task_id"]
                status = ctx["status"]
                result = _api_patch(api_base, f"/tasks/{task_id}",
                                    {"status": status, "suppress_token_event": True})
                if result is not None and "_error" not in result:
                    report.fixes_applied.append(
                        f"C-10 fixed: task {task_id[:8]}… status → {status!r}"
                    )
                elif result is not None:
                    report.fixes_applied.append(
                        f"C-10 FAILED: task {task_id[:8]}… status → {status!r}: {result['_error']}"
                    )

            elif issue.check_id == "C-11":
                ws_id = ctx["ws_id"]
                ws_status = ctx.get("ws_status", "")
                task = ctx["task"]
                wp_file = Path(ctx["wp_file"])
                meta = ctx["meta"]
                body = ctx.get("body", "")
                t_id = str(task.get("id", "")).strip()
                # Skip creating tasks for finished workstreams — the workstream is
                # done/archived so unlinked tasks are stale file artefacts, not gaps.
                if normalise_workstream_status(ws_status) in CLOSED_WORKSTREAM_STATUSES:
                    report.fixes_applied.append(
                        f"C-11 skipped: task '{t_id}' in {ws_status} workstream — not created"
                    )
                else:
                    t_status = normalise_task_status(task.get("status", "todo"))
                    t_priority = str(task.get("priority", "medium")).strip()
                    if t_priority not in VALID_TASK_PRIORITIES:
                        t_priority = "medium"
                    desired_task_id = str(
                        ctx.get("desired_task_id") or _derived_work_record_uuid(t_id)
                    )
                    t_data = _api_post(api_base, "/tasks", {
                        "id": desired_task_id,
                        "workplan_id": ws_id,
                        "title": str(task.get("title", t_id)).strip() or t_id,
                        "description": task.get("description") or None,
                        "status": t_status,
                        "priority": t_priority,
                        "assignee": task.get("assignee") or None,
                    })
                    if t_data:
                        t_db_id = t_data["id"]
                        if t_db_id != desired_task_id:
                            report.fixes_applied.append(
                                f"C-11 FAIL: task '{t_id}' returned unexpected UUID {t_db_id}"
                            )
                            continue
                        if not task.get("state_hub_task_id"):
                            injected = _inject_task_id_into_block(
                                wp_file, "state_hub_task_id", t_db_id, t_id
                            )
                            if not injected:
                                _inject_task_id_frontmatter_list(wp_file, t_db_id, t_id)
                        report.fixes_applied.append(
                            f"C-11 fixed: task '{t_id}' → {t_db_id[:8]}…"
                        )

            elif issue.check_id == "C-12":
                task_id = ctx["task_id"]
                if ctx.get("ws_finished"):
                    result = _api_patch(api_base, f"/tasks/{task_id}", {"status": "cancel"})
                    if result is not None:
                        report.fixes_applied.append(
                            f"C-12 fixed: orphan task {task_id[:8]}… canceled (workstream finished)"
                        )

            elif issue.check_id == "C-22":
                task_id = ctx["task_id"]
                description = ctx["description"]
                result = _api_patch(api_base, f"/tasks/{task_id}", {"description": description})
                if result is not None and "_error" not in result:
                    report.fixes_applied.append(
                        f"C-22 fixed: task {task_id[:8]}… description updated"
                    )
                elif result is not None:
                    report.fixes_applied.append(
                        f"C-22 FAILED: task {task_id[:8]}… description update: {result['_error']}"
                    )

            elif issue.check_id == "C-15":
                # T03 — writeback: DB is ahead of file — patch file to match DB.
                if no_writeback:
                    report.fixes_applied.append(
                        f"C-15 skipped (--no-writeback): task '{ctx['task_block_id']}' "
                        f"db={ctx['db_status']!r}"
                    )
                else:
                    wp_file = Path(ctx["wp_file"])
                    task_block_id = ctx["task_block_id"]
                    db_status = ctx["db_status"]
                    old_status = issue.file_value
                    # 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,
                        )

                        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 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}")

    # Record that a sync run happened for this repo
    from datetime import timezone as _tz
    import datetime as _dt
    now_iso = _dt.datetime.now(_tz.utc).isoformat()
    _api_patch(api_base, f"/repos/{repo_slug}", {"last_state_synced_at": now_iso})

    # Write the worker orientation brief (.custodian-brief.md)
    if repo_path:
        brief_written = _write_custodian_brief(api_base, repo_slug, repo_path)
        if brief_written:
            report.fixes_applied.append("brief: .custodian-brief.md updated")

    if repo_path:
        wm_committed = _commit_working_memory_notes(repo_slug, repo_path)
        if wm_committed:
            report.fixes_applied.append("working-memory: daily-triage notes committed")

    # Push all commits made this run (writebacks + brief) to close the loop.
    # This ensures the next run starts with local == remote, making the
    # timer idempotent and preventing commit pile-up.
    if repo_path:
        push_ok, push_msg = _git_push(repo_path)
        if push_ok:
            report.fixes_applied.append(f"push: {push_msg}")
        else:
            report.fixes_applied.append(f"push WARN: {push_msg}")

    return report


# Check IDs that are known-background noise in multi-machine setups:
# C-08 = finished/archived DB workstream with no file (pre-ADR-001 legacy)
# These alone do not warrant a pull+fix cycle.
_BACKGROUND_CHECKS: frozenset[str] = frozenset({"C-08"})


def _report_needs_action(
    report: ConsistencyReport, behind_remote: bool, ahead_of_remote: int = 0
) -> bool:
    """Return True if the repo warrants a pull+fix cycle.

    A repo is considered clean (no action needed) when:
    - It is not behind its remote tracking branch, AND
    - It has no unpushed local commits (from a prior failed push), AND
    - It has no FAIL issues, AND
    - Every WARN/INFO issue is in the background-noise set (C-08).
    """
    if behind_remote or ahead_of_remote > 0:
        return True
    if report.assessment_failures or report.automation_errors:
        return True
    actionable_warns = [
        i for i in report.warnings + report.infos
        if i.check_id not in _BACKGROUND_CHECKS
    ]
    return bool(actionable_warns)


def fix_all_remote(
    api_base: str,
    no_writeback: bool = False,
    max_seconds: int = DEFAULT_REMOTE_ALL_MAX_SECONDS,
) -> list[ConsistencyReport]:
    """Pull-then-fix all registered repos that need attention.

    For each repo with a resolvable local path on this machine:
    1. Skip if the path does not exist (repo lives on another machine).
    2. Run check_repo() and detect_behind_remote() — read-only.
    3. If the repo is clean (no actionable issues, not behind remote): skip.
    4. Otherwise: git pull --ff-only, then fix_repo().

    Returns one ConsistencyReport per repo that was *not* skipped as clean.
    Repos skipped as clean are printed to stdout but not included in the return value.
    """
    repos = _api_get(api_base, "/repos")
    if not isinstance(repos, list):
        print("ERROR: Could not fetch repos from state-hub API", file=sys.stderr)
        return None

    started = time.monotonic()
    reports: list[ConsistencyReport] = []
    skipped_clean: list[str] = []
    skipped_missing: list[str] = []
    skipped_budget: list[str] = []

    for repo in repos:
        slug = repo["slug"]
        if max_seconds > 0 and time.monotonic() - started > max_seconds:
            skipped_budget.append(slug)
            skipped_budget.extend(r.get("slug", "?") for r in repos[repos.index(repo) + 1:])
            break

        # Resolve path using the same priority as check_repo
        path = resolve_repo_path(repo)
        if not path or not Path(path).is_dir():
            skipped_missing.append(slug)
            continue

        # Read-only pass: detect issues and remote staleness.
        # _detect_behind_remote does a fetch, so _detect_ahead_of_remote
        # after it sees an up-to-date tracking branch.
        pre_report = check_repo(api_base, slug)
        behind = _detect_behind_remote(path)
        ahead = _detect_ahead_of_remote(path)

        if not _report_needs_action(pre_report, behind, ahead):
            skipped_clean.append(slug)
            continue

        # Pull before fixing
        pull_ok, pull_msg = _git_pull(path)
        pull_tag = f"pull: {pull_msg}"
        if not pull_ok:
            pull_tag = f"pull WARN: {pull_msg} — proceeding anyway"

        report = fix_repo(api_base, slug, no_writeback=no_writeback)
        report.fixes_applied.insert(0, pull_tag)
        reports.append(report)

    # Print clean/missing summary lines before the detailed reports
    if skipped_clean:
        print(f"  CLEAN (skipped): {', '.join(skipped_clean)}")
    if skipped_missing:
        print(f"  NOT ON THIS HOST (skipped): {', '.join(skipped_missing)}")
    if skipped_budget:
        print(
            f"  BUDGET EXHAUSTED after {max_seconds}s (skipped): "
            f"{', '.join(skipped_budget)}"
        )
    if skipped_clean or skipped_missing or skipped_budget:
        print()

    return reports


def archive_closed_workplans(
    repo_path: str,
    completion_date: str | None = None,
    workplan: str | None = None,
) -> list[str]:
    """Move closed root workplans into workplans/archived/ with YYMMDD prefix.

    Only root-level files whose frontmatter status normalises to finished or
    archived are moved. Files with any open task blocks are left in place.
    """
    repo_dir = Path(repo_path)
    workplans_dir = repo_dir / "workplans"
    archived_dir = workplans_dir / "archived"
    if not workplans_dir.is_dir():
        return []

    date_prefix = completion_date or datetime.now().strftime("%y%m%d")
    archived_dir.mkdir(exist_ok=True)
    moved: list[str] = []

    for wp_file in sorted(workplans_dir.glob("*.md")):
        text = wp_file.read_text(encoding="utf-8")
        if not text.startswith("---"):
            continue
        meta, body = parse_frontmatter(text)
        if not meta or meta.get("_parse_error"):
            continue
        if workplan:
            wanted = workplan.removesuffix(".md")
            if wanted not in {str(meta.get("id", "")), wp_file.stem, wp_file.name}:
                continue
        status = normalise_workstream_status(str(meta.get("status", "")).strip())
        if status not in CLOSED_WORKSTREAM_STATUSES:
            continue
        tasks = get_tasks_from_workplan(meta, body)
        open_tasks = [
            t for t in tasks
            if normalise_task_status(t.get("status", "todo")) not in TERMINAL_TASK_STATUSES
        ]
        if open_tasks:
            continue

        target = archived_dir / f"{date_prefix}-{canonical_workplan_filename(wp_file)}"
        if target.exists():
            raise FileExistsError(f"Archived workplan already exists: {target}")
        wp_file.rename(target)
        moved.append(f"{wp_file.relative_to(repo_dir)} -> {target.relative_to(repo_dir)}")

    return moved


# ---------------------------------------------------------------------------
# Output / rendering
# ---------------------------------------------------------------------------

def render_renormalization_guide() -> str:
    lines = ["Lifecycle Renormalization Rules", "=" * 33]
    for rule in RENORMALIZATION_RULES:
        lines.extend(
            [
                "",
                f"{rule.check_id}: {rule.invariant}",
                f"  Detect: {rule.detection}",
                f"  Repair: {rule.repair}",
                f"  Tests:  {rule.test_anchor}",
            ]
        )
    lines.extend(["", "Add The Next Guard"])
    for idx, item in enumerate(RENORMALIZATION_NEXT_GUARD_CHECKLIST, 1):
        lines.append(f"  {idx}. {item}")
    return "\n".join(lines)


def render_text(report: ConsistencyReport, show_info: bool = True) -> str:
    SEP = "=" * 66
    lines = [
        f"ADR-001 Consistency Report",
        f"Repo:  {report.repo_slug}",
        f"Path:  {report.repo_path}",
        SEP,
    ]

    error_section = report.automation_errors
    fail_section = report.assessment_failures
    if error_section:
        lines.append(f"\n  AUTOMATION ERRORS ({len(error_section)}):")
        for i in error_section:
            loc = f"  [{i.file_path}]" if i.file_path else ""
            lines.append(f"    {i.check_id}{loc}")
            lines.append(f"      {i.message}")

    if fail_section:
        lines.append(f"\n  ASSESSMENT FAILURES ({len(fail_section)}):")
        for i in fail_section:
            loc = f"  [{i.file_path}]" if i.file_path else ""
            fix_tag = " [fixable]" if i.fixable else ""
            lines.append(f"    {i.check_id}{loc}{fix_tag}")
            lines.append(f"      {i.message}")
            if i.file_value or i.db_value:
                lines.append(f"      file={i.file_value!r}  db={i.db_value!r}")

    for sev in ("WARN", "INFO"):
        section = [i for i in report.issues if i.severity == sev]
        if not section or (sev == "INFO" and not show_info):
            continue
        lines.append(f"\n  {sev}S ({len(section)}):")
        for i in section:
            loc = f"  [{i.file_path}]" if i.file_path else ""
            fix_tag = " [fixable]" if i.fixable else ""
            lines.append(f"    {i.check_id}{loc}{fix_tag}")
            lines.append(f"      {i.message}")
            if i.file_value or i.db_value:
                lines.append(f"      file={i.file_value!r}  db={i.db_value!r}")

    if report.fixes_applied:
        lines.append(f"\n  FIXES APPLIED ({len(report.fixes_applied)}):")
        for f in report.fixes_applied:
            lines.append(f"    {f}")

    lines.append(f"\n{SEP}")
    n_err = len(report.automation_errors)
    n_fail = len(report.assessment_failures)
    n_warn = len(report.warnings)
    n_info = len(report.infos)
    lines.append(
        f"  {n_err} automation-error | {n_fail} assessment-fail | "
        f"{n_warn} warn | {n_info} info"
    )
    if n_err:
        lines.append("  RESULT: ✗ AUTOMATION ERROR")
    elif n_fail:
        lines.append("  RESULT: ✗ ASSESSMENT FAIL (follow-up needed)")
    elif n_warn:
        lines.append("  RESULT: ✓ PASS (with warnings)")
    else:
        lines.append("  RESULT: ✓ PASS")
    lines.append(SEP)
    return "\n".join(lines)


def report_to_dict(report: ConsistencyReport) -> dict:
    return {
        "repo_slug": report.repo_slug,
        "repo_path": report.repo_path,
        "issues": [
            {
                "severity": i.severity,
                "check_id": i.check_id,
                "message": i.message,
                "file_path": i.file_path,
                "db_id": i.db_id,
                "file_value": i.file_value,
                "db_value": i.db_value,
                "fixable": i.fixable,
            }
            for i in report.issues
        ],
        "fixes_applied": report.fixes_applied,
        "summary": {
            "fail": len(report.assessment_failures),
            "automation_error": len(report.automation_errors),
            "warn": len(report.warnings),
            "info": len(report.infos),
        },
        "result": (
            "error" if report.automation_errors else
            "fail" if report.assessment_failures else
            "warn" if report.warnings else
            "pass"
        ),
    }


def consistency_exit_code(reports: list[ConsistencyReport], *, remote_all: bool = False) -> int:
    """Return the strict CLI exit code for consistency reports."""
    any_automation_error = any(r.automation_errors for r in reports)
    any_assessment_fail = any(r.assessment_failures for r in reports)
    any_warn = any(r.warnings for r in reports)
    if remote_all:
        return 1 if any_automation_error else 0
    if any_automation_error or any_assessment_fail:
        return 1
    return 2 if any_warn else 0


# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(
        description="ADR-001 consistency checker — bidirectional file↔DB validation",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--repo", metavar="SLUG",
                       help="Registered repo slug (e.g. the-custodian)")
    group.add_argument("--all", action="store_true",
                       help="Run checks against all repos with a resolvable path on this host")
    group.add_argument("--here", metavar="PATH", nargs="?", const="",
                       help="Infer repo slug from git remote URL at PATH (default: CWD)")
    group.add_argument("--renormalization-guide", action="store_true",
                       help="Print lifecycle renormalization rules and the add-next-guard checklist")
    parser.add_argument("--fix", action="store_true",
                        help="Apply auto-fixable issues (status drift, repo mismatch, etc.)")
    parser.add_argument("--remote", action="store_true",
                        help="Pull each repo before fixing; when used with --all, skips repos "
                             "that are already clean (no actionable issues, not behind remote). "
                             "Implies --fix.")
    parser.add_argument("--max-seconds", type=int, default=DEFAULT_REMOTE_ALL_MAX_SECONDS,
                        help="Wall-clock budget for --remote --all before remaining repos are skipped "
                             f"(default: {DEFAULT_REMOTE_ALL_MAX_SECONDS}; 0 disables)")
    parser.add_argument("--no-writeback", action="store_true", dest="no_writeback",
                        help="Disable DB→file status writeback (C-15) while keeping other fixes")
    parser.add_argument(
        "--bootstrap-empty-projection",
        action="store_true",
        help=(
            "Rebuild authoritative legacy UUIDs only after proving the target "
            "repository has zero projected workplans"
        ),
    )
    parser.add_argument("--archive-closed", action="store_true",
                        help="Move closed root workplans to workplans/archived/YYMMDD-*.md")
    parser.add_argument("--archive-workplan", metavar="ID_OR_FILE", default=None,
                        help="When archiving, only move the matching workplan id or filename")
    parser.add_argument("--archive-date", metavar="YYMMDD", default=None,
                        help="Completion date prefix for --archive-closed (default: today)")
    parser.add_argument("--repo-path", metavar="PATH", default=None,
                        help="Override the local repo path (useful when the DB has a different "
                             "machine's path). Takes priority over host_paths and local_path.")
    parser.add_argument("--api-base", default="http://127.0.0.1:8000",
                        help="State Hub API base URL")
    parser.add_argument("--json", action="store_true", dest="as_json",
                        help="Output JSON instead of human-readable text")
    args = parser.parse_args()

    if args.renormalization_guide:
        if args.as_json:
            print(json.dumps({
                "rules": [
                    {
                        "check_id": rule.check_id,
                        "invariant": rule.invariant,
                        "detection": rule.detection,
                        "repair": rule.repair,
                        "test_anchor": rule.test_anchor,
                    }
                    for rule in RENORMALIZATION_RULES
                ],
                "add_next_guard": list(RENORMALIZATION_NEXT_GUARD_CHECKLIST),
            }, indent=2))
        else:
            print(render_renormalization_guide())
        sys.exit(0)

    import os as _os
    no_wb = getattr(args, "no_writeback", False)
    do_fix = args.fix or args.remote
    if args.bootstrap_empty_projection and (not do_fix or args.all or args.remote):
        parser.error(
            "--bootstrap-empty-projection requires --fix for one local repository "
            "and cannot be combined with --all or --remote"
        )

    # --here: infer slug from git remote URL, then run as single-repo check/fix
    if args.here is not None:
        search_path = args.here or _os.getcwd()
        inferred = _infer_slug_from_path(args.api_base, search_path)
        if inferred is None:
            print(
                f"ERROR: No registered repo with a matching remote_url found at '{search_path}'.",
                file=sys.stderr,
            )
            print(
                "  Register the repo first: POST /repos/ with remote_url set.",
                file=sys.stderr,
            )
            sys.exit(1)
        inferred_slug, git_root = inferred
        print(f"  Detected: {inferred_slug}  ({git_root})")
        if do_fix:
            reports = [
                fix_repo(
                    args.api_base,
                    inferred_slug,
                    git_root,
                    no_writeback=no_wb,
                    bootstrap_empty_projection=args.bootstrap_empty_projection,
                )
            ]
        else:
            reports = [check_repo(args.api_base, inferred_slug, git_root)]
        if args.archive_closed:
            moved = archive_closed_workplans(git_root, args.archive_date, args.archive_workplan)
            reports[0].fixes_applied.extend(f"archive: {m}" for m in moved)
    # --remote --all: smart pull+fix across all repos
    elif args.remote and args.all:
        with run_lock("consistency-remote-all") as acquired:
            if not acquired:
                print(
                    "SKIP: another fix-consistency-remote --all run is already active",
                    file=sys.stderr,
                )
                sys.exit(0)
            reports = fix_all_remote(
                args.api_base,
                no_writeback=no_wb,
                max_seconds=args.max_seconds,
            )
        if reports is None:
            sys.exit(1)
        if not reports:
            sys.exit(0)
    else:
        # Resolve repo list
        hostname = _sweep_hostname()
        repo_slugs: list[str] = []
        if args.all:
            repos = _api_get(args.api_base, "/repos")
            if not isinstance(repos, list):
                print("ERROR: Could not fetch repos from state-hub API", file=sys.stderr)
                sys.exit(1)
            repo_slugs = [
                r["slug"] for r in repos
                if r.get("local_path") or (r.get("host_paths") or {}).get(hostname)
            ]
            if not repo_slugs:
                print(
                    f"No repos with a path registered for host '{hostname}'.",
                    file=sys.stderr,
                )
                sys.exit(0)
        else:
            repo_slugs = [args.repo]

        # --repo-path only applies to single-repo runs; silently ignored with --all
        path_override = args.repo_path if not args.all else None

        if args.remote:
            # Single-repo remote: pull first, then fix
            slug = repo_slugs[0]
            repo = _api_get(args.api_base, f"/repos/{slug}")
            path = resolve_repo_path(repo or {}, path_override) if repo else (path_override or "")
            if path:
                pull_ok, pull_msg = _git_pull(path)
                prefix = "pull" if pull_ok else "pull WARN"
                print(f"  {prefix}: {pull_msg}")
            report = fix_repo(args.api_base, slug, path_override, no_writeback=no_wb)
            reports = [report]
        elif do_fix:
            reports = [
                fix_repo(
                    args.api_base,
                    slug,
                    path_override,
                    no_writeback=no_wb,
                    bootstrap_empty_projection=args.bootstrap_empty_projection,
                )
                for slug in repo_slugs
            ]
        else:
            reports = [check_repo(args.api_base, slug, path_override) for slug in repo_slugs]

        if args.archive_closed:
            for report in reports:
                moved = archive_closed_workplans(report.repo_path, args.archive_date, args.archive_workplan)
                report.fixes_applied.extend(f"archive: {m}" for m in moved)

    if args.as_json:
        output = (
            report_to_dict(reports[0])
            if len(reports) == 1
            else [report_to_dict(r) for r in reports]
        )
        print(json.dumps(output, indent=2))
    else:
        for report in reports:
            print(render_text(report))
            print()

    sys.exit(consistency_exit_code(reports, remote_all=args.remote and args.all))


if __name__ == "__main__":
    main()