feat(consistency): coordination hygiene checks and MCP workplan_id aliases

Add C-25..C-30 fix-consistency checks for blocked-workplan inbox sweeps,
stale unread triage, workplan ID prefix/collision lint, and SCOPE freshness.
Extend brief generation and get_domain_summary with inbox hygiene warnings.
Complete workplan_id aliases on remaining MCP tools and retry transient
_api_get failures to reduce false stale-reference errors under load.

docs/workplan-convention.md

@@ -23,6 +23,17 @@ During extraction, legacy `CUST-WP-*` plans may be bridged or migrated with
 their existing `state_hub_workstream_id` values. Write files first, then run
 State Hub consistency sync after this repo is registered.
 
+When a workplan is `blocked`, record the unblock condition in frontmatter:
+
+```yaml
+status: blocked
+blocked_on: message-from:llm-connect
+```
+
+`blocked_on` uses the `message-from:<agent>` form so fix-consistency can
+cross-check unread inbox messages from that counterpart and warn when the
+blocker may have cleared.
+
 Canonical workplan/workstream statuses are:
 
 ```text

mcp_server/server.py

@@ -344,6 +344,31 @@ def get_domain_summary(domain_slug: str) -> str:
     if goal_guidance:
         result["goal_guidance"] = goal_guidance
 
+    inbox_hygiene: dict[str, Any] = {}
+    try:
+        from scripts.consistency_check import collect_inbox_hygiene, STALE_UNREAD_DAYS
+    except ImportError:
+        collect_inbox_hygiene = None  # type: ignore[assignment]
+        STALE_UNREAD_DAYS = 3
+    if collect_inbox_hygiene is not None:
+        for repo in repos:
+            repo_slug = repo["slug"]
+            hygiene = collect_inbox_hygiene(API_BASE, repo_slug)
+            if (
+                hygiene["stale_unread_count"]
+                or hygiene["missing_thread"]
+                or hygiene["work_requests_unpromoted"]
+            ):
+                inbox_hygiene[repo_slug] = {
+                    "stale_unread_count": hygiene["stale_unread_count"],
+                    "stale_unread_days": STALE_UNREAD_DAYS,
+                    "stale_unread": hygiene["stale_unread"][:5],
+                    "missing_thread_count": len(hygiene["missing_thread"]),
+                    "work_requests_unpromoted": hygiene["work_requests_unpromoted"][:3],
+                }
+    if inbox_hygiene:
+        result["inbox_hygiene"] = inbox_hygiene
+
     # Compact capabilities list (type + title + repo_slug only, capped at 20)
     caps_raw = _get("/capability-catalog/", {"domain": domain_slug, "status": "active"})
     if isinstance(caps_raw, list):
@@ -416,9 +441,21 @@ def list_tasks(
 
 
 @mcp.tool()
-def list_blocked_tasks(workstream_id: str | None = None) -> str:
+def list_blocked_tasks(
-    """List all waiting tasks, optionally filtered by workstream_id."""
+    workplan_id: str | None = None,
-    return json.dumps(_get("/tasks", {"status": "wait", "workstream_id": workstream_id}), indent=2)
+    workstream_id: str | None = None,
+) -> str:
+    """List all waiting tasks, optionally filtered by workplan.
+
+    Args:
+        workplan_id: UUID of the workplan (preferred).
+        workstream_id: legacy alias for workplan_id.
+    """
+    parent_id = workplan_id or workstream_id
+    return json.dumps(
+        _get("/tasks", {"status": "wait", "workplan_id": parent_id, "workstream_id": parent_id}),
+        indent=2,
+    )
 
 
 @mcp.tool()
@@ -879,17 +916,25 @@ def clear_human_flag(task_id: str) -> str:
 
 
 @mcp.tool()
-def list_human_interventions(workstream_id: str | None = None) -> str:
+def list_human_interventions(
+    workplan_id: str | None = None,
+    workstream_id: str | None = None,
+) -> str:
     """List all tasks flagged for human intervention.
 
-    Returns tasks where needs_human=True, optionally filtered to one workstream.
+    Returns tasks where needs_human=True, optionally filtered to one workplan.
     Use this at session start to surface Bernd's action items.
 
     Args:
-        workstream_id: optional UUID to scope results to one workstream
+        workplan_id: optional UUID to scope results to one workplan (preferred).
+        workstream_id: legacy alias for workplan_id.
     """
+    parent_id = workplan_id or workstream_id
     return json.dumps(
-        _get("/tasks", {"needs_human": "true", "workstream_id": workstream_id}),
+        _get(
+            "/tasks",
+            {"needs_human": "true", "workplan_id": parent_id, "workstream_id": parent_id},
+        ),
         indent=2,
     )
 
@@ -899,6 +944,7 @@ def record_decision(
     title: str,
     decision_type: str = "pending",
     topic_id: str | None = None,
+    workplan_id: str | None = None,
     workstream_id: str | None = None,
     description: str | None = None,
     rationale: str | None = None,
@@ -914,17 +960,19 @@ def record_decision(
         title: decision title
         decision_type: made | pending
         topic_id: optional topic UUID
-        workstream_id: optional workstream UUID (at least one required)
+        workplan_id: optional workplan UUID (preferred; at least one required)
+        workstream_id: legacy alias for workplan_id
         description: optional context
         rationale: reasoning behind the decision
         decided_by: person/agent who decided
         deadline: ISO datetime string for when decision is needed
     """
+    parent_id = workplan_id or workstream_id
     decision = _post("/decisions", {
         "title": title,
         "decision_type": decision_type,
         "topic_id": topic_id,
-        "workstream_id": workstream_id,
+        "workplan_id": parent_id,
         "description": description,
         "rationale": rationale,
         "decided_by": decided_by,
@@ -935,7 +983,8 @@ def record_decision(
 
     progress_error = _emit_progress_event("record_decision", decision, {
         "topic_id": topic_id,
-        "workstream_id": workstream_id,
+        "workplan_id": parent_id,
+        "workstream_id": parent_id,
         "decision_id": decision["id"],
         "event_type": "decision_recorded",
         "summary": f"Decision recorded ({decision_type}): {title}",
@@ -1075,9 +1124,16 @@ def update_workplan_status(workplan_id: str, status: str) -> str:
 
 
 @mcp.tool()
-def update_workstream_status(workstream_id: str, status: str) -> str:
+def update_workstream_status(
+    status: str,
+    workplan_id: str | None = None,
+    workstream_id: str | None = None,
+) -> str:
     """Legacy alias for update_workplan_status."""
-    return _update_workplan_status_impl(workstream_id, status, tool_name="update_workstream_status")
+    parent_id = workplan_id or workstream_id
+    if not parent_id:
+        return _json_result(_mcp_error("update_workstream_status", "workplan_id is required"))
+    return _update_workplan_status_impl(parent_id, status, tool_name="update_workstream_status")
 
 
 @mcp.tool()
@@ -1104,7 +1160,8 @@ def update_workplan(
 
 @mcp.tool()
 def update_workstream(
-    workstream_id: str,
+    workplan_id: str | None = None,
+    workstream_id: str | None = None,
     title: str | None = None,
     description: str | None = None,
     owner: str | None = None,
@@ -1113,8 +1170,11 @@ def update_workstream(
     status: str | None = None,
 ) -> str:
     """Legacy alias for update_workplan."""
+    parent_id = workplan_id or workstream_id
+    if not parent_id:
+        return _json_result(_mcp_error("update_workstream", "workplan_id is required"))
     return _update_workplan_impl(
-        workstream_id,
+        parent_id,
         title=title,
         description=description,
         owner=owner,
@@ -1277,16 +1337,22 @@ def create_workplan_dependency(
 
 @mcp.tool()
 def create_dependency(
-    from_workstream_id: str,
+    from_workplan_id: str | None = None,
+    from_workstream_id: str | None = None,
+    to_workplan_id: str | None = None,
     to_workstream_id: str | None = None,
     to_task_id: str | None = None,
     relationship_type: str = "blocks",
     description: str | None = None,
 ) -> str:
     """Legacy alias for create_workplan_dependency."""
+    source_id = from_workplan_id or from_workstream_id
+    target_id = to_workplan_id or to_workstream_id
+    if not source_id:
+        return _json_result(_mcp_error("create_dependency", "from_workplan_id is required"))
     return _create_dependency_impl(
-        from_workplan_id=from_workstream_id,
+        from_workplan_id=source_id,
-        to_workplan_id=to_workstream_id,
+        to_workplan_id=target_id,
         to_task_id=to_task_id,
         relationship_type=relationship_type,
         description=description,
@@ -1294,23 +1360,30 @@ def create_dependency(
 
 
 @mcp.tool()
-def list_dependencies(workstream_id: str) -> str:
+def list_dependencies(
-    """Return all dependency edges touching a workstream (both directions).
+    workplan_id: str | None = None,
+    workstream_id: str | None = None,
+) -> str:
+    """Return all dependency edges touching a workplan (both directions).
 
-    The response distinguishes edges where this workstream is the dependent
+    The response distinguishes edges where this workplan is the dependent
     (depends_on) from edges where it is the blocker (blocks).
 
     Args:
-        workstream_id: UUID of the workstream to inspect
+        workplan_id: UUID of the workplan to inspect (preferred).
+        workstream_id: legacy alias for workplan_id.
     """
-    edges = _get(f"/workplans/{workstream_id}/dependencies")
+    parent_id = workplan_id or workstream_id
+    if not parent_id:
+        return _json_result(_mcp_error("list_dependencies", "workplan_id is required"))
+    edges = _get(f"/workplans/{parent_id}/dependencies")
     depends_on = [
         e for e in edges
-        if e.get("from_workplan_id", e.get("from_workstream_id")) == workstream_id
+        if e.get("from_workplan_id", e.get("from_workstream_id")) == parent_id
     ]
     blocks = [
         e for e in edges
-        if e.get("to_workplan_id", e.get("to_workstream_id")) == workstream_id
+        if e.get("to_workplan_id", e.get("to_workstream_id")) == parent_id
     ]
     return json.dumps({"depends_on": depends_on, "blocks": blocks}, indent=2)
 
@@ -1329,6 +1402,7 @@ def register_extension_point(
     priority: str = "medium",
     ep_id: str | None = None,
     topic_id: str | None = None,
+    workplan_id: str | None = None,
     workstream_id: str | None = None,
 ) -> str:
     """Register a discovered extension point — optional future functionality not yet committed.
@@ -1345,13 +1419,15 @@ def register_extension_point(
         priority: low | medium | high | critical
         ep_id: optional human-readable ID, e.g. EP-CUST-001 (auto-assigned if omitted)
         topic_id: UUID of related topic
-        workstream_id: UUID of related workstream
+        workplan_id: UUID of related workplan (preferred)
+        workstream_id: legacy alias for workplan_id
     """
+    parent_id = workplan_id or workstream_id
     ep = _post("/extension-points", {
         "domain": domain, "title": title, "ep_type": ep_type,
         "description": description, "location": location,
         "priority": priority, "ep_id": ep_id,
-        "topic_id": topic_id, "workstream_id": workstream_id,
+        "topic_id": topic_id, "workplan_id": parent_id,
     })
     _post("/progress", {
         "summary": f"Extension point registered: [{ep.get('ep_id') or ep['id'][:8]}] {title} ({ep_type}, {domain})",
@@ -1406,6 +1482,7 @@ def register_technical_debt(
     severity: str = "medium",
     td_id: str | None = None,
     topic_id: str | None = None,
+    workplan_id: str | None = None,
     workstream_id: str | None = None,
 ) -> str:
     """Register a technical debt item — a known quality compromise to address later.
@@ -1422,13 +1499,15 @@ def register_technical_debt(
         severity: low | medium | high | critical
         td_id: optional human-readable ID, e.g. TD-CUST-001
         topic_id: UUID of related topic
-        workstream_id: UUID of related workstream
+        workplan_id: UUID of related workplan (preferred)
+        workstream_id: legacy alias for workplan_id
     """
+    parent_id = workplan_id or workstream_id
     td = _post("/technical-debt", {
         "domain": domain, "title": title, "debt_type": debt_type,
         "description": description, "location": location,
         "severity": severity, "td_id": td_id,
-        "topic_id": topic_id, "workstream_id": workstream_id,
+        "topic_id": topic_id, "workplan_id": parent_id,
     })
     _post("/progress", {
         "summary": f"Technical debt registered: [{td.get('td_id') or td['id'][:8]}] {title} ({debt_type}, {severity}, {domain})",
@@ -2032,6 +2111,7 @@ def register_contribution(
     target_org: str | None = None,
     target_repo: str | None = None,
     body_path: str | None = None,
+    related_workplan_id: str | None = None,
     related_workstream_id: str | None = None,
     notes: str | None = None,
 ) -> str:
@@ -2043,20 +2123,22 @@ def register_contribution(
         target_org: GitHub org or owner of the upstream project
         target_repo: Repository name of the upstream project
         body_path: Relative path to the Markdown artifact file in the repo
-        related_workstream_id: UUID of the related workstream (optional)
+        related_workplan_id: UUID of the related workplan (preferred, optional)
+        related_workstream_id: legacy alias for related_workplan_id
         notes: Any additional notes (optional)
     """
+    parent_id = related_workplan_id or related_workstream_id
     contrib = _post("/contributions", {
         "type": type,
         "title": title,
         "target_org": target_org,
         "target_repo": target_repo,
         "body_path": body_path,
-        "related_workstream_id": related_workstream_id,
+        "related_workplan_id": parent_id,
         "notes": notes,
     })
     _post("/progress", {
-        "workstream_id": related_workstream_id,
+        "workplan_id": parent_id,
         "event_type": "contribution_registered",
         "summary": f"Contribution registered [{type.upper()}]: {title}",
         "author": "custodian",
@@ -2441,6 +2523,7 @@ def request_capability(
     capability_type: str,
     requesting_agent: str,
     requesting_domain: str,
+    requesting_workplan_id: str | None = None,
     requesting_workstream_id: str | None = None,
     priority: str = "medium",
     blocking_task_id: str | None = None,
@@ -2454,17 +2537,19 @@ def request_capability(
         capability_type: Category (e.g. 'infrastructure', 'api', 'data', 'security')
         requesting_agent: Your agent identifier (e.g. 'net-kingdom-worker')
         requesting_domain: Your domain slug (e.g. 'custodian')
-        requesting_workstream_id: UUID of your workstream (optional)
+        requesting_workplan_id: UUID of your workplan (preferred, optional)
+        requesting_workstream_id: legacy alias for requesting_workplan_id
         priority: low | medium | high | critical (default: medium)
         blocking_task_id: UUID of the task blocked until this is fulfilled (optional)
     """
+    parent_id = requesting_workplan_id or requesting_workstream_id
     req = _post("/capability-requests", {
         "title": title,
         "description": description,
         "capability_type": capability_type,
         "requesting_agent": requesting_agent,
         "requesting_domain": requesting_domain,
-        "requesting_workstream_id": requesting_workstream_id,
+        "requesting_workplan_id": parent_id,
         "priority": priority,
         "blocking_task_id": blocking_task_id,
     })
@@ -2487,6 +2572,7 @@ def patch_capability_request(
     catalog_entry_id: Optional[str] = None,
     priority: Optional[str] = None,
     blocking_task_id: Optional[str] = None,
+    fulfilling_workplan_id: Optional[str] = None,
     fulfilling_workstream_id: Optional[str] = None,
 ) -> dict:
     """Correct mutable metadata on a capability request.
@@ -2500,11 +2586,13 @@ def patch_capability_request(
         catalog_entry_id: Correct catalog entry UUID. Re-derives fulfilling domain.
         priority: New priority (low/medium/high/critical).
         blocking_task_id: UUID of the task this request unblocks on completion.
-        fulfilling_workstream_id: UUID of the workstream delivering this capability.
+        fulfilling_workplan_id: UUID of the workplan delivering this capability (preferred).
+        fulfilling_workstream_id: legacy alias for fulfilling_workplan_id.
 
     Returns:
         Updated capability request dict, or {"error": "..."}.
     """
+    parent_id = fulfilling_workplan_id or fulfilling_workstream_id
     body: dict = {}
     if catalog_entry_id is not None:
         body["catalog_entry_id"] = catalog_entry_id
@@ -2512,8 +2600,8 @@ def patch_capability_request(
         body["priority"] = priority
     if blocking_task_id is not None:
         body["blocking_task_id"] = blocking_task_id
-    if fulfilling_workstream_id is not None:
+    if parent_id is not None:
-        body["fulfilling_workstream_id"] = fulfilling_workstream_id
+        body["fulfilling_workplan_id"] = parent_id
 
     if not body:
         return {"error": "no fields provided to patch"}
@@ -2923,6 +3011,7 @@ def record_token_event(
     tokens_in: int,
     tokens_out: int,
     task_id: Optional[str] = None,
+    workplan_id: Optional[str] = None,
     workstream_id: Optional[str] = None,
     repo_id: Optional[str] = None,
     model: Optional[str] = None,
@@ -2942,7 +3031,8 @@ def record_token_event(
         tokens_in: Input token count
         tokens_out: Output token count
         task_id: UUID of the task (nullable)
-        workstream_id: UUID of the workstream (nullable; auto-filled from task)
+        workplan_id: UUID of the workplan (preferred; nullable; auto-filled from task)
+        workstream_id: legacy alias for workplan_id
         repo_id: UUID of the managed repo (nullable)
         model: Model identifier, e.g. 'claude-sonnet-4-6'
         agent: Agent name, e.g. 'custodian', 'ralph'
@@ -2951,11 +3041,12 @@ def record_token_event(
         note: Free-text note
         session_id: Agent session identifier
     """
+    parent_id = workplan_id or workstream_id
     body = {
         "tokens_in": tokens_in,
         "tokens_out": tokens_out,
         "task_id": task_id,
-        "workstream_id": workstream_id,
+        "workplan_id": parent_id,
         "repo_id": repo_id,
         "model": model,
         "agent": agent,
@@ -2976,8 +3067,8 @@ def record_token_event(
     }
 
     # Append running total for the task if available
-    scope_id = task_id or workstream_id
+    scope_id = task_id or parent_id
-    scope = "task" if task_id else ("workstream" if workstream_id else None)
+    scope = "task" if task_id else ("workstream" if parent_id else None)
     if scope and scope_id:
         summary = _get("/token-events/summary", {"scope": scope, "id": scope_id})
         if "error" not in summary:

scripts/consistency_check.py

@@ -27,6 +27,12 @@ Checks:
   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
 
 Usage:
     python scripts/consistency_check.py --repo SLUG [--fix] [--no-writeback] [--json] [--api-base URL]
@@ -62,7 +68,8 @@ import sys
 import time
 from contextlib import contextmanager
 from dataclasses import dataclass, field
-from datetime import datetime
+from collections import Counter
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any
 
@@ -124,6 +131,20 @@ 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"}
 
 # Legacy file/API aliases translated before comparison and PATCHing.
 FILE_TO_DB_WORKSTREAM_STATUS: dict[str, str] = dict(LEGACY_WORKSTREAM_STATUS_ALIASES)
@@ -541,22 +562,40 @@ def _api_get(
     # Only append trailing slash to the path component, not to query strings
     if "?" not in path and not path.endswith("/"):
         path += "/"
-    try:
+    filtered = {k: v for k, v in (params or {}).items() if v is not None}
-        with _httpx.Client(base_url=api_base, timeout=10.0, follow_redirects=True) as c:
+    last_error: Exception | None = None
-            filtered = {k: v for k, v in (params or {}).items() if v is not None}
+    for attempt in range(_API_GET_RETRIES):
-            r = c.get(path, params=filtered if filtered else None)
+        try:
-            r.raise_for_status()
+            with _httpx.Client(base_url=api_base, timeout=10.0, follow_redirects=True) as c:
-            return r.json()
+                r = c.get(path, params=filtered if filtered else None)
-    except _httpx.HTTPStatusError as exc:
+                r.raise_for_status()
-        if exc.response.status_code == 404:
+                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
-        if return_error:
+        except (_httpx.TimeoutException, _httpx.ConnectError, _httpx.NetworkError) as exc:
-            return {"_error": str(exc)}
+            last_error = exc
-        return None
+            if attempt < _API_GET_RETRIES - 1:
-    except Exception as exc:
+                time.sleep(_API_GET_RETRY_BASE_DELAY * (attempt + 1))
-        if return_error:
+                continue
-            return {"_error": str(exc)}
+            if return_error:
-        return None
+                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:
@@ -1282,6 +1321,12 @@ def check_repo(api_base: str, repo_slug: str, repo_path_override: str | None = N
     # 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
@@ -1532,6 +1577,350 @@ def _git_commit_writeback(
 # 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)
@@ -1704,6 +2093,99 @@ def _write_custodian_brief(api_base: str, repo_slug: str, repo_path: str) -> boo
     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 += [
         "",
         "---",

tests/test_consistency_check.py

@@ -31,6 +31,8 @@ from consistency_check import (
     RENORMALIZATION_RULES,
     STATUS_ORDER,
     _BACKGROUND_CHECKS,
+    _api_get,
+    _check_scope_freshness,
     _detect_behind_remote,
     _git_pull,
     _patch_frontmatter_field,
@@ -39,11 +41,14 @@ from consistency_check import (
     archive_closed_workplans,
     canonical_workplan_filename,
     check_repo,
+    collect_inbox_hygiene,
     consistency_exit_code,
     fix_repo,
     get_tasks_from_workplan,
+    infer_wp_prefix,
     iter_workplan_files,
     normalise_workstream_status,
+    parse_blocked_on,
     parse_frontmatter,
     parse_task_blocks,
     render_text,
@@ -1402,3 +1407,104 @@ class TestReportNeedsAction:
         """Diverged repo (both behind and ahead) needs action."""
         r = self._make_report([])
         assert _report_needs_action(r, behind_remote=True, ahead_of_remote=5) is True
+
+
+class TestCoordinationHygieneHelpers:
+    def test_parse_blocked_on_extracts_agent(self):
+        assert parse_blocked_on("message-from:llm-connect") == "llm-connect"
+        assert parse_blocked_on("other-format") is None
+
+    def test_infer_wp_prefix_prefers_on_disk_ids(self, tmp_path):
+        workplans = tmp_path / "workplans"
+        workplans.mkdir()
+        (workplans / "CUST-WP-0001-alpha.md").write_text(
+            "---\nid: CUST-WP-0001\nstatus: active\n---\n",
+            encoding="utf-8",
+        )
+        (workplans / "CUST-WP-0002-beta.md").write_text(
+            "---\nid: STATE-WP-0002\nstatus: active\n---\n",
+            encoding="utf-8",
+        )
+        assert infer_wp_prefix(tmp_path, "state-hub") == "CUST-WP"
+
+    def test_collect_inbox_hygiene_flags_stale_unread(self, monkeypatch):
+        from datetime import datetime, timedelta, timezone
+
+        stale = (datetime.now(timezone.utc) - timedelta(days=5)).isoformat()
+        recent = datetime.now(timezone.utc).isoformat()
+
+        def fake_get(api_base, path, params=None, **kwargs):
+            if path.startswith("/messages"):
+                return [
+                    {
+                        "id": "11111111-1111-1111-1111-111111111111",
+                        "from_agent": "llm-connect",
+                        "subject": "Please implement the workplan",
+                        "body": "multi-step implementation needed",
+                        "created_at": stale,
+                    },
+                    {
+                        "id": "22222222-2222-2222-2222-222222222222",
+                        "from_agent": "ops-warden",
+                        "subject": "FYI",
+                        "body": "status update",
+                        "created_at": recent,
+                    },
+                ]
+            return None
+
+        monkeypatch.setattr("consistency_check._api_get", fake_get)
+        hygiene = collect_inbox_hygiene("http://example", "the-custodian")
+        assert hygiene["stale_unread_count"] == 1
+        assert hygiene["work_requests_unpromoted"]
+        assert hygiene["missing_thread"]
+
+    def test_scope_freshness_warns_on_contradiction(self, tmp_path):
+        (tmp_path / "SCOPE.md").write_text(
+            "# SCOPE\n\n## Current State\n\n- Status: active — stable\n",
+            encoding="utf-8",
+        )
+        report = ConsistencyReport(repo_slug="demo", repo_path=str(tmp_path))
+        _check_scope_freshness(tmp_path, [], report)
+        assert any(issue.check_id == "C-30" for issue in report.warnings)
+
+    def test_api_get_retries_transient_errors(self, monkeypatch):
+        attempts = {"count": 0}
+
+        class FakeResponse:
+            def __init__(self, status_code: int):
+                self.status_code = status_code
+                self.reason_phrase = "Server Error"
+
+            def raise_for_status(self):
+                if self.status_code < 400:
+                    return
+                import httpx
+
+                request = httpx.Request("GET", "http://example/workstreams/ws-1/")
+                response = httpx.Response(self.status_code, request=request)
+                raise httpx.HTTPStatusError("boom", request=request, response=response)
+
+            def json(self):
+                return {"id": "ws-1"}
+
+        class FakeClient:
+            def __init__(self, *args, **kwargs):
+                pass
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, exc_type, exc, tb):
+                return False
+
+            def get(self, path, params=None):
+                attempts["count"] += 1
+                if attempts["count"] < 2:
+                    return FakeResponse(503)
+                return FakeResponse(200)
+
+        monkeypatch.setattr("consistency_check._httpx.Client", FakeClient)
+        result = _api_get("http://example", "/workstreams/ws-1")
+        assert result == {"id": "ws-1"}
+        assert attempts["count"] == 2

tests/test_mcp_write_tools.py

@@ -195,7 +195,7 @@ class TestMCPWriteTools:
                 "/progress",
                 {
                     "topic_id": None,
-                    "workstream_id": "ws-1",
+                    "workplan_id": "ws-1",
                     "task_id": None,
                     "event_type": "note",
                     "summary": "MCP progress",
@@ -260,7 +260,7 @@ class TestMCPWriteTools:
                     "title": body["title"],
                     "decision_type": body["decision_type"],
                     "topic_id": body["topic_id"],
-                    "workstream_id": body["workstream_id"],
+                    "workplan_id": body.get("workplan_id"),
                     "status": "open",
                     "escalation_note": None,
                 }
@@ -321,3 +321,74 @@ class TestMCPWriteTools:
         assert body["tool"] == "record_decision"
         assert body["error"] == "API response missing required field(s): id"
         assert [path for path, _ in calls] == ["/decisions"]
+
+    async def test_record_decision_accepts_workplan_id_alias(self, monkeypatch):
+        calls: list[tuple[str, dict[str, Any]]] = []
+
+        def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
+            calls.append((path, body))
+            if path == "/decisions":
+                return {
+                    "id": "decision-1",
+                    "title": body["title"],
+                    "decision_type": body["decision_type"],
+                    "topic_id": body["topic_id"],
+                    "workplan_id": body["workplan_id"],
+                    "status": "open",
+                    "escalation_note": None,
+                }
+            if path == "/progress":
+                return {"id": "event-1", **body}
+            raise AssertionError(f"unexpected POST {path}")
+
+        monkeypatch.setattr(server, "_post", fake_post)
+
+        body = await _call_tool(
+            "record_decision",
+            {
+                "title": "Alias check",
+                "decision_type": "made",
+                "topic_id": "topic-1",
+                "workplan_id": "wp-1",
+            },
+        )
+
+        assert body["id"] == "decision-1"
+        assert calls[0][1]["workplan_id"] == "wp-1"
+        assert calls[1][1]["workplan_id"] == "wp-1"
+
+    async def test_list_blocked_tasks_accepts_workplan_id_alias(self, monkeypatch):
+        captured: dict[str, Any] = {}
+
+        def fake_get(path: str, params: dict[str, Any] | None = None) -> Any:
+            captured["path"] = path
+            captured["params"] = params
+            return []
+
+        monkeypatch.setattr(server, "_get", fake_get)
+
+        await _call_tool("list_blocked_tasks", {"workplan_id": "wp-1"})
+
+        assert captured["params"]["workplan_id"] == "wp-1"
+        assert captured["params"]["workstream_id"] == "wp-1"
+
+    async def test_record_token_event_accepts_workplan_id_alias(self, monkeypatch):
+        calls: list[tuple[str, dict[str, Any]]] = []
+
+        def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
+            calls.append((path, body))
+            return {"id": "token-1", "tokens_total": 1500}
+
+        def fake_get(path: str, params: dict[str, Any] | None = None) -> Any:
+            return {"tokens_total": 1500, "event_count": 1}
+
+        monkeypatch.setattr(server, "_post", fake_post)
+        monkeypatch.setattr(server, "_get", fake_get)
+
+        body = await _call_tool(
+            "record_token_event",
+            {"tokens_in": 1000, "tokens_out": 500, "workplan_id": "wp-1"},
+        )
+
+        assert body["event_id"] == "token-1"
+        assert calls[0][1]["workplan_id"] == "wp-1"