From 9693946755729a0d135f49a1cd4cad352d8a6868 Mon Sep 17 00:00:00 2001 From: tegwick Date: Wed, 8 Jul 2026 00:56:21 +0200 Subject: [PATCH] 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 | 11 + mcp_server/server.py | 171 ++++++++--- scripts/consistency_check.py | 514 +++++++++++++++++++++++++++++++- tests/test_consistency_check.py | 106 +++++++ tests/test_mcp_write_tools.py | 75 ++++- 5 files changed, 819 insertions(+), 58 deletions(-) diff --git a/docs/workplan-convention.md b/docs/workplan-convention.md index 3067cc5..95b6584 100644 --- a/docs/workplan-convention.md +++ b/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:` 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 diff --git a/mcp_server/server.py b/mcp_server/server.py index 5ee3aec..d2abb8a 100644 --- a/mcp_server/server.py +++ b/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: - """List all waiting tasks, optionally filtered by workstream_id.""" - return json.dumps(_get("/tasks", {"status": "wait", "workstream_id": workstream_id}), indent=2) +def list_blocked_tasks( + workplan_id: str | None = None, + 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, - to_workplan_id=to_workstream_id, + from_workplan_id=source_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: - """Return all dependency edges touching a workstream (both directions). +def list_dependencies( + 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: - body["fulfilling_workstream_id"] = fulfilling_workstream_id + if parent_id is not None: + 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 = "task" if task_id else ("workstream" if workstream_id else None) + scope_id = task_id or parent_id + 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: diff --git a/scripts/consistency_check.py b/scripts/consistency_check.py index f954a82..e3b4925 100644 --- a/scripts/consistency_check.py +++ b/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.+)$") +_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: - with _httpx.Client(base_url=api_base, timeout=10.0, follow_redirects=True) as c: - filtered = {k: v for k, v in (params or {}).items() if v is not None} - r = c.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: + 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: + with _httpx.Client(base_url=api_base, timeout=10.0, follow_redirects=True) as c: + r = c.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 - if return_error: - return {"_error": str(exc)} - return None - except Exception as exc: - 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: @@ -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 = "" _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 += [ "", "---", diff --git a/tests/test_consistency_check.py b/tests/test_consistency_check.py index 0ce1973..e9cf0bd 100644 --- a/tests/test_consistency_check.py +++ b/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 diff --git a/tests/test_mcp_write_tools.py b/tests/test_mcp_write_tools.py index c0034bc..2b927f5 100644 --- a/tests/test_mcp_write_tools.py +++ b/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"