control_plane.py

from __future__ import annotations

import json
import re
import sqlite3
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

import httpx

from .config import (
    ACTIVE_PROJECTS_PATH,
    DOCS_DIR,
    INVENTORY_DB_PATH,
    OLLARMA_BASE_URL,
    OLLARMA_TIMEOUT_SECONDS,
    OVERWATCH_ROOT,
    RESOURCE_REGISTRY_PATH,
    ROOT,
    SEEDGRAPH_BASE_URL,
    SEEDGRAPH_TIMEOUT_SECONDS,
    SEEDGRAPH_ROOT,
)
from .hackathon_demo import (
    build_demo_package as build_agenthon_demo_package,
    build_demo_project_detail,
    get_demo_project,
    get_provider_presence,
    get_sample_run,
    list_demo_projects,
)

PLANNING_DIR = ROOT / ".planning"
STATE_PATH = PLANNING_DIR / "STATE.md"
ROADMAP_PATH = PLANNING_DIR / "ROADMAP.md"
MILESTONE_AUDIT_PATH = PLANNING_DIR / "v0.1-MILESTONE-AUDIT.md"
PENDING_TODOS_DIR = PLANNING_DIR / "todos" / "pending"
DEMO_VERIFICATION_FRESHNESS = timedelta(days=3)
AGENTHON_PROJECT_ID = "agenthon-001"
DEMO_PRIORITY_PROJECTS = (
    AGENTHON_PROJECT_ID,
    "watchtower",
    "ollarma",
    "seedgraph",
    "overwatch",
    "gettingsciencedone",
    "antigence",
)
PROJECT_DEMO_HINTS: dict[str, dict[str, Any]] = {
    AGENTHON_PROJECT_ID: {
        "dashboard_url": f"/projects/{AGENTHON_PROJECT_ID}",
        "dashboard_label": "Agenthon project page",
        "kb_candidates": [],
        "howto_candidates": ["docs/AGENTHON_001.md", "docs/AGENTHON_001_DEMO_SCRIPT.md"],
        "launch_command": (
            "curl -X POST http://127.0.0.1:8000/api/demo-projects/agenthon-001/run "
            "-H 'Content-Type: application/json' "
            "-d '{\"task_request\":\"Find public sources on low-cost biological age biomarkers and produce a public-safe shortlist.\"}'"
        ),
        "launch_source_candidates": ["docs/AGENTHON_001.md", "docs/AGENTHON_001_DEMO_SCRIPT.md"],
        "probe": {
            "url": "http://127.0.0.1:8000/api/demo-projects/agenthon-001",
            "expected_status": 200,
            "expected_text": AGENTHON_PROJECT_ID,
            "method": "live_probe:GET /api/demo-projects/agenthon-001",
        },
        "dependencies": [
            "OPENAI_API_KEY for review synthesis",
            "TAVILY_API_KEY for live public-source retrieval",
            "DAYTONA_API_KEY and DAYTONA_API_URL for live sandbox execution",
        ],
        "limitations": [
            "Bundled sample output is public-safe fixture data, not a persisted live run.",
            "SeedGraph and Antigence remain supporting lanes in this slice.",
        ],
        "artifact_candidates": [
            "demo/agenthon-001/sample-request.json",
            "demo/agenthon-001/sample-run.json",
            "docs/AGENTHON_001.md",
        ],
        "escalation_candidates": [
            "docs/HUMAN_AI_INTERVENTION_POLICY.md",
            "docs/CONTROL_PLANE_BOUNDARY.md",
        ],
    },
    "watchtower": {
        "dashboard_url": "http://127.0.0.1:8000",
        "dashboard_label": "Watchtower dashboard",
        "kb_candidates": ["docs/KB_HOME.md"],
        "howto_candidates": ["docs/KB_HOWTO_RUN_WATCHTOWER_2026-04-10.md", "README.md"],
        "launch_command": (
            "uv run --no-project --python 3.13 --with fastapi --with httpx --with jinja2 "
            "--with pydantic --with uvicorn uvicorn watchtower.api:app --reload"
        ),
        "launch_source_candidates": ["docs/KB_HOWTO_RUN_WATCHTOWER_2026-04-10.md", "README.md"],
        "probe": {
            "url": "http://127.0.0.1:8000/healthz",
            "expected_status": 200,
            "expected_text": "ok",
            "method": "live_probe:GET /healthz",
        },
        "dependencies": [
            "WATCHTOWER_OVERWATCH_ROOT points at Overwatch exports",
            "WATCHTOWER_OLLARMA_URL points at local Ollarma",
            "WATCHTOWER_SEEDGRAPH_URL is optional for provenance reads",
        ],
        "limitations": [
            "Downstream-only surface; no canonical writes",
            "SeedGraph visibility depends on configured service URL",
        ],
        "artifact_candidates": [
            ".planning/STATE.md",
            "tests/test_api.py",
            "tests/test_control_plane.py",
        ],
        "escalation_candidates": [
            "docs/HUMAN_AI_INTERVENTION_POLICY.md",
            "AGENTS.md",
        ],
    },
    "ollarma": {
        "dashboard_url": "http://127.0.0.1:8484/dashboard",
        "dashboard_label": "Ollarma dashboard",
        "kb_candidates": ["docs/OLLARMA_KB_CONTRACT.md"],
        "howto_candidates": ["docs/LOCAL_ADOPTION.md", "docs/OLLARMA_DASHBOARD.md", "README.md"],
        "launch_command": "ollarma serve",
        "launch_source_candidates": ["README.md", "docs/OLLARMA_DASHBOARD.md"],
        "probe": {
            "url": "http://127.0.0.1:8484/dashboard/overview",
            "expected_status": 200,
            "expected_text": "generated_at",
            "method": "live_probe:GET /dashboard/overview",
        },
        "dependencies": [
            "Local Ollama runtime",
            "Validated model selection artifact",
            "Repo-local KB artifacts",
        ],
        "limitations": [
            "Bounded helper lane only; no arbitrary shell execution",
            "KB gaps should block routed help instead of falling back silently",
        ],
        "artifact_candidates": [
            "results/run-2026-04-11T15:45:32Z.json",
            "results/run-2026-04-11T15:45:32Z.jsonl",
            "results/model_selection.md",
        ],
        "escalation_candidates": [
            "docs/DETERMINISTIC_EXECUTION_BOUNDARY.md",
            "src/ollarma/escalation.py",
        ],
    },
    "seedgraph": {
        "dashboard_url": "http://127.0.0.1:8011/triage/publication-batches/pubbatch-724017d6090c4c15",
        "dashboard_label": "SeedGraph accepted-run dashboard",
        "kb_candidates": [],
        "howto_candidates": [".planning/artifacts/v6.2/agenthon-001-science-ops-demo/HANDOFF.md"],
        "launch_command": (
            "bash /Users/byron/projects/active/seedgraph/.planning/artifacts/v6.2/"
            "agenthon-001-science-ops-demo/run-demo.sh"
        ),
        "launch_source_candidates": [
            ".planning/artifacts/v6.2/agenthon-001-science-ops-demo/HANDOFF.md",
            ".planning/quick/260411-cs8-demo-launcher-and-verification-for-accep/260411-cs8-SUMMARY.md",
        ],
        "probe": {
            "url": "http://127.0.0.1:8012/api/v1/publication-families/pubfam-a71de205cf71abea/provenance",
            "expected_status": 200,
            "expected_text": "publication-family-provenance-v1",
            "method": "live_probe:accepted family provenance",
        },
        "dependencies": [
            "Accepted v6.2 artifacts under .planning/artifacts/v6.2",
            "Repo .venv for triage and service APIs",
        ],
        "limitations": [
            "Read-focused accepted-run demo only",
            "No review mutation or canonical truth writeback",
        ],
        "artifact_candidates": [
            ".planning/artifacts/v6.2/phase-63-final.scorecard.json",
            ".planning/artifacts/v6.2/phase-64-final.drift.json",
            ".planning/quick/260411-cs8-demo-launcher-and-verification-for-accep/260411-cs8-SUMMARY.md",
        ],
        "escalation_candidates": [
            "src/seedgraph/review/ui.py",
            "src/seedgraph/triage/ui.py",
        ],
    },
    "overwatch": {
        "dashboard_url": "http://127.0.0.1:8484/dashboard",
        "dashboard_label": "Overwatch dashboard route from README",
        "kb_candidates": ["docs/governance/kb/KB_HOME.md", "KB_INDEX.md"],
        "howto_candidates": ["docs/governance/KB_HOWTO_OVERWATCH_MCP_DATABASE_2026-03-07.md", "README.md"],
        "launch_command": "python scripts/control_plane_dashboard_web.py",
        "launch_source_candidates": ["README.md"],
        "probe": {
            "url": "http://127.0.0.1:8484/api/dashboard",
            "expected_status": 200,
            "expected_text": "projects",
            "method": "live_probe:GET /api/dashboard",
        },
        "dependencies": [
            "Overwatch local environment and database stack",
            "Canonical control-plane exports",
        ],
        "limitations": [
            "README dashboard port conflicts with live Ollarma service in this environment",
            "Treat stale or conflicting local dashboard docs as demo blockers",
        ],
        "artifact_candidates": [
            "results/exp001_20260410T192943Z_bootstrap_smoke.json",
            "data/control_plane/active_projects.json",
            "data/control_plane/control_plane_resource_registry.json",
        ],
        "escalation_candidates": [
            "scripts/antigence_review.py",
            "README.md",
        ],
    },
    "gettingsciencedone": {
        "dashboard_url": None,
        "dashboard_label": "No browser dashboard documented",
        "kb_candidates": [],
        "howto_candidates": ["pyproject.toml", ".planning/STATE.md", "homebrew/README.md"],
        "launch_command": "gsigmad serve",
        "launch_source_candidates": ["pyproject.toml", ".planning/STATE.md"],
        "dependencies": [
            "Python 3.11+",
            "gsigmad CLI installed from the local package",
        ],
        "limitations": [
            "CLI or MCP surface only in current repo evidence",
            "No repo-local browser dashboard documented",
        ],
        "artifact_candidates": [
            ".planning/STATE.md",
            ".planning/research/SUMMARY.md",
        ],
        "escalation_candidates": [
            "src/gsigmad/commands/redteam.py",
            "src/gsigmad/commands/review.py",
        ],
    },
    "antigence": {
        "dashboard_url": "http://127.0.0.1:5055/",
        "dashboard_label": "Antigence dashboard",
        "kb_candidates": ["docs/kb/README.md"],
        "howto_candidates": ["docs/ANTIGENCE_STARTUP.md", "INSTALL.md", "README.md"],
        "launch_command": "antigence start",
        "launch_source_candidates": ["README.md", "INSTALL.md"],
        "probe": {
            "url": "http://127.0.0.1:5055/",
            "expected_status": 200,
            "expected_text": "Antigence",
            "method": "live_probe:GET /",
        },
        "dependencies": [
            'Package installed with the "web" extra',
            "Optional Ollama improves analysis quality",
        ],
        "limitations": [
            "Web UI is optional and may not be running by default",
            "Verification remains review-oriented, not canonical truth",
        ],
        "artifact_candidates": [
            "results/release/public-docs-alignment.json",
            "results/review_backfill_20260220T022549Z",
        ],
        "escalation_candidates": [
            "docs/portfolio/antigence-escalation-bundle.schema.json",
            "src/antigence/upstream/variant_review.py",
        ],
    },
}


def _read_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text())


def _read_text(path: Path) -> str:
    return path.read_text() if path.exists() else ""


def load_active_project_register() -> dict[str, Any]:
    return _read_json(ACTIVE_PROJECTS_PATH)


def load_resource_registry() -> dict[str, Any]:
    return _read_json(RESOURCE_REGISTRY_PATH)


def build_agenthon_live_readiness() -> dict[str, Any]:
    provider_presence = get_provider_presence()
    blockers = [
        item["env_var"]
        for item in provider_presence.values()
        if item.get("required") and not item.get("configured")
    ]
    next_steps = [
        f"Configure required env vars: {', '.join(blockers)}"
        if blockers
        else "Run the sample demo from the project page."
    ]
    return {
        "status": "ready_for_live_run" if not blockers else "provider_env_missing",
        "blockers": blockers,
        "next_steps": next_steps,
    }


def build_agenthon_project_record() -> dict[str, Any]:
    project = get_demo_project(AGENTHON_PROJECT_ID)
    sample_run = get_sample_run(AGENTHON_PROJECT_ID)
    readiness = build_agenthon_live_readiness()
    if project is None:
        raise FileNotFoundError(f"Missing demo project config for {AGENTHON_PROJECT_ID}")
    return {
        "project_id": project["project_id"],
        "name": project.get("public_title") or project["project_id"],
        "repo_path": str(ROOT),
        "project_type": "watchtower-local-demo",
        "mutation_scope": "watchtower-local-derived",
        "current_status": readiness["status"],
        "next_action": readiness["next_steps"][0],
        "notes": [project.get("summary") or "Public-safe Watchtower hackathon demo."],
        "direction_contract": {
            "primary_role": "hackathon-demo",
            "owner_lane": "watchtower",
            "operating_posture": "active" if not readiness["blockers"] else "staged",
            "direction": project.get("summary"),
            "boundary_notes": "Watchtower-local derived demo record; not canonical Overwatch truth.",
        },
        "state_metadata": {
            "status": readiness["status"],
            "milestone": project.get("version"),
            "last_activity": None,
        },
        "gaps": {
            "provider_env": "ok" if not readiness["blockers"] else "blocked",
            "seedgraph": "waived-by-type",
            "antigence": "waived-by-type",
        },
        "drift_items": [],
        "remediation_items": readiness["next_steps"],
        "evidence_sources": [
            f"/api/demo-projects/{AGENTHON_PROJECT_ID}",
            f"/api/demo-projects/{AGENTHON_PROJECT_ID}/sample-run",
        ],
        "surfaces": {
            "project_page": {"path": f"/projects/{AGENTHON_PROJECT_ID}", "status": "observed", "required": True},
            "project_json": {"path": f"/api/demo-projects/{AGENTHON_PROJECT_ID}", "status": "observed", "required": True},
            "run_api": {"path": f"/api/demo-projects/{AGENTHON_PROJECT_ID}/run", "status": "observed", "required": True},
        },
        "ownership_contract": {
            "owned_surface": {
                "surface_name": project.get("public_title") or AGENTHON_PROJECT_ID,
                "notes": "Public-safe Watchtower-owned hackathon demo surface.",
                "export_status": "derived_local",
                "scientific_authority": "watchtower-demo-only",
            },
            "imported_surfaces": [],
        },
    }


def _derived_project_records() -> list[dict[str, Any]]:
    return [build_agenthon_project_record()]


def list_projects() -> list[dict[str, Any]]:
    register = load_active_project_register()
    items: list[dict[str, Any]] = []
    for record in [*register.get("projects", []), *_derived_project_records()]:
        direction = record.get("direction_contract", {})
        ownership = record.get("ownership_contract", {})
        state_metadata = record.get("state_metadata", {})
        gaps = record.get("gaps", {})
        imported_surfaces = ownership.get("imported_surfaces", [])
        items.append(
            {
                "project_id": record.get("project_id"),
                "name": record.get("name"),
                "repo_path": record.get("repo_path"),
                "project_type": record.get("project_type"),
                "mutation_scope": record.get("mutation_scope"),
                "current_status": record.get("current_status"),
                "status": state_metadata.get("status"),
                "milestone": state_metadata.get("milestone"),
                "last_activity": state_metadata.get("last_activity"),
                "primary_role": direction.get("primary_role"),
                "owner_lane": direction.get("owner_lane"),
                "operating_posture": direction.get("operating_posture"),
                "direction": direction.get("direction"),
                "boundary_notes": direction.get("boundary_notes"),
                "next_action": record.get("next_action"),
                "notes": record.get("notes"),
                "gaps": gaps,
                "gap_count": sum(1 for value in gaps.values() if value not in {"ok", "waived-by-type"}),
                "imported_surface_count": len(imported_surfaces),
                "drift_items": record.get("drift_items", []),
                "remediation_items": record.get("remediation_items", []),
                "evidence_sources": record.get("evidence_sources", []),
                "surfaces": record.get("surfaces", {}),
            }
        )
    return sorted(items, key=lambda item: (item["project_id"] or "").lower())


def build_project_overview() -> dict[str, Any]:
    projects = list_projects()
    blocked = sum(1 for item in projects if item.get("operating_posture") == "blocked")
    active = sum(1 for item in projects if item.get("operating_posture") == "active")
    staged = sum(1 for item in projects if item.get("operating_posture") in {"staged", "adjacent"})
    return {
        "project_count": len(projects),
        "blocked_count": blocked,
        "active_count": active,
        "staged_count": staged,
    }


def list_resources(*, resource_class: str | None = None, owner: str | None = None) -> list[dict[str, Any]]:
    registry = load_resource_registry()
    items = registry.get("resources", [])
    if resource_class:
        items = [item for item in items if item.get("resource_class") == resource_class]
    if owner:
        items = [item for item in items if item.get("owner") == owner]
    return items


def build_resource_overview() -> dict[str, Any]:
    registry = load_resource_registry()
    return registry.get("summary", {})


def inventory_db_available() -> bool:
    return INVENTORY_DB_PATH.exists()


def _connect_inventory() -> sqlite3.Connection:
    connection = sqlite3.connect(f"file:{INVENTORY_DB_PATH}?mode=ro", uri=True)
    connection.row_factory = sqlite3.Row
    return connection


def _rows(query: str, params: tuple[Any, ...] = ()) -> list[dict[str, Any]]:
    if not inventory_db_available():
        return []
    with _connect_inventory() as connection:
        cursor = connection.execute(query, params)
        return [dict(row) for row in cursor.fetchall()]


def inventory_overview() -> dict[str, Any]:
    if not inventory_db_available():
        return {
            "available": False,
            "db_path": str(INVENTORY_DB_PATH),
            "root_count": 0,
            "complete_root_count": 0,
            "node_count": 0,
            "file_count": 0,
            "dir_count": 0,
            "symlink_count": 0,
        }

    with _connect_inventory() as connection:
        row = connection.execute(
            """
            SELECT
                COUNT(*) AS root_count,
                SUM(CASE WHEN scan_status = 'complete' THEN 1 ELSE 0 END) AS complete_root_count,
                COALESCE(SUM(node_count), 0) AS node_count,
                COALESCE(SUM(file_count), 0) AS file_count,
                COALESCE(SUM(dir_count), 0) AS dir_count,
                COALESCE(SUM(symlink_count), 0) AS symlink_count,
                MAX(updated_at) AS updated_at
            FROM root_scan_state
            """
        ).fetchone()

    return {
        "available": True,
        "db_path": str(INVENTORY_DB_PATH),
        "root_count": int(row["root_count"] or 0),
        "complete_root_count": int(row["complete_root_count"] or 0),
        "node_count": int(row["node_count"] or 0),
        "file_count": int(row["file_count"] or 0),
        "dir_count": int(row["dir_count"] or 0),
        "symlink_count": int(row["symlink_count"] or 0),
        "updated_at": row["updated_at"],
    }


def list_inventory_roots() -> list[dict[str, Any]]:
    rows = _rows(
        """
        SELECT
            root_id,
            label,
            path,
            scan_status,
            authority_source,
            file_count,
            dir_count,
            symlink_count,
            hidden_count,
            git_marker_count,
            error_count,
            total_file_bytes,
            node_count,
            updated_at
        FROM root_scan_state
        ORDER BY label COLLATE NOCASE
        """
    )
    return rows


def search_inventory_nodes(
    query: str | None = None,
    *,
    root_id: str | None = None,
    entry_type: str | None = None,
    include_hidden: bool = False,
    limit: int = 50,
) -> list[dict[str, Any]]:
    sql = """
        SELECT
            node_key,
            root_id,
            path,
            rel_path,
            name,
            depth,
            entry_type,
            extension,
            size_bytes,
            mtime_utc,
            is_hidden,
            is_symlink,
            symlink_target,
            scan_error
        FROM nodes
        WHERE 1 = 1
    """
    params: list[Any] = []
    if query:
        like = f"%{query}%"
        sql += " AND (path LIKE ? OR rel_path LIKE ? OR name LIKE ?)"
        params.extend([like, like, like])
    if root_id:
        sql += " AND root_id = ?"
        params.append(root_id)
    if entry_type:
        sql += " AND entry_type = ?"
        params.append(entry_type)
    if not include_hidden:
        sql += " AND is_hidden = 0"
    sql += " ORDER BY depth ASC, path ASC LIMIT ?"
    params.append(max(1, min(limit, 500)))
    return _rows(sql, tuple(params))


def build_watchtower_overview() -> dict[str, Any]:
    projects = list_projects()
    hackathon_projects = build_hackathon_projects()
    operator_contract = build_operator_contract()
    planning = build_planning_overview()
    active_blockers = build_active_blockers(operator_contract)
    next_actions = build_next_actions(projects, planning, active_blockers)
    tasks = build_task_feed(projects, planning)
    opportunities = build_opportunity_feed(planning)
    opportunity_lanes = build_opportunity_lanes()
    operator_queue = build_operator_queue(tasks, active_blockers, opportunities)
    helper_lane = next(
        (lane for lane in operator_contract["lanes"] if lane["lane_id"] == "ollarma-bounded-helper"),
        None,
    )
    seedgraph_lane = next(
        (lane for lane in operator_contract["lanes"] if lane["lane_id"] == "seedgraph-provenance-identity"),
        None,
    )
    ollarma_dashboard = _ollarma_dashboard_surface()
    readiness = build_operator_readiness(operator_contract, opportunity_lanes)
    approved_actions = build_approved_actions(operator_contract)
    demo_contracts = build_demo_contracts(
        projects,
        helper_lane=helper_lane,
        seedgraph_lane=seedgraph_lane,
        ollarma_dashboard=ollarma_dashboard,
    )
    demo_summary = build_demo_portfolio_summary(demo_contracts)
    operator_summary = build_operator_summary_rows(
        projects,
        helper_lane=helper_lane,
        seedgraph_lane=seedgraph_lane,
        ollarma_dashboard=ollarma_dashboard,
        next_actions=next_actions,
    )
    return {
        "projects": build_project_overview(),
        "resources": build_resource_overview(),
        "inventory": inventory_overview(),
        "operator_contract": operator_contract,
        "planning": planning,
        "active_blockers": active_blockers,
        "next_actions": next_actions,
        "tasks": tasks,
        "opportunities": opportunities,
        "opportunity_lanes": opportunity_lanes,
        "operator_queue": operator_queue,
        "readiness": readiness,
        "approved_actions": approved_actions,
        "helper_lane": helper_lane,
        "seedgraph_lane": seedgraph_lane,
        "ollarma_dashboard": ollarma_dashboard,
        "demo_contracts": demo_contracts,
        "demo_summary": demo_summary,
        "operator_summary": operator_summary,
        "hackathon_projects": hackathon_projects,
    }


def build_hackathon_projects() -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    for project in list_demo_projects():
        detail = build_demo_project_detail(project["project_id"])
        if detail is None:
            continue
        blockers = list(detail.get("live_blockers", []))
        sample_run = detail.get("sample_run", {})
        items.append(
            {
                "project_id": project["project_id"],
                "title": project.get("public_title") or project["project_id"],
                "summary": project.get("summary"),
                "version": project.get("version"),
                "status": "live_ready" if not blockers else "fixture_ready",
                "blockers": blockers,
                "next_step": (
                    "Run the sample demo from the project page."
                    if not blockers
                    else f"Configure required env vars: {', '.join(blockers)}"
                ),
                "tracks": {
                    "active": list(project.get("active_tracks", [])),
                    "supporting": list(project.get("supporting_lanes", [])),
                    "optional": list(project.get("optional_lanes", [])),
                },
                "project_page": f"/projects/{project['project_id']}",
                "project_json": f"/api/demo-projects/{project['project_id']}",
                "sample_run_status": sample_run.get("status"),
            }
        )
    return items


def _repo_root_for_project(project: dict[str, Any]) -> Path | None:
    repo_path = project.get("repo_path")
    if not isinstance(repo_path, str):
        return None
    repo_root = Path(repo_path)
    return repo_root if repo_root.exists() else None


def _resolve_candidates(repo_root: Path | None, candidates: list[str]) -> list[Path]:
    if repo_root is None:
        return []
    return [repo_root / candidate for candidate in candidates if (repo_root / candidate).exists()]


def _find_first_path(repo_root: Path | None, candidates: list[str]) -> Path | None:
    resolved = _resolve_candidates(repo_root, candidates)
    return resolved[0] if resolved else None


def _collect_artifacts(repo_root: Path | None, candidates: list[str]) -> list[dict[str, str]]:
    items: list[dict[str, str]] = []
    for path in _resolve_candidates(repo_root, candidates):
        items.append({"label": path.name, "path": str(path)})
    return items


def _isoformat_timestamp(value: float | None) -> str | None:
    if value is None:
        return None
    return datetime.fromtimestamp(value, tz=UTC).strftime("%Y-%m-%dT%H:%M:%SZ")


def _path_timestamp(path: Path | None) -> str | None:
    if path is None or not path.exists():
        return None
    return _isoformat_timestamp(path.stat().st_mtime)


def _latest_verification_path(repo_root: Path | None) -> Path | None:
    if repo_root is None:
        return None
    phase_root = repo_root / ".planning" / "phases"
    if not phase_root.exists():
        return None
    candidates = sorted(phase_root.glob("*/**/*VERIFICATION*.md"), key=lambda path: path.stat().st_mtime)
    return candidates[-1] if candidates else None


def _parse_repo_state(path: Path | None) -> dict[str, Any]:
    text = _read_text(path) if path else ""
    if not text:
        return {
            "milestone": None,
            "status": None,
            "phase": None,
            "next_commands": [],
            "last_updated": None,
            "last_activity": None,
        }
    milestone = (
        _extract_backtick_value(r"Current milestone:\s*`([^`]+)`", text)
        or _extract_backtick_value(r"Milestone:\s*`([^`]+)`", text)
        or re.search(r"^milestone:\s*(.+)$", text, flags=re.MULTILINE).group(1).strip()
        if re.search(r"^milestone:\s*(.+)$", text, flags=re.MULTILINE)
        else None
    )
    status = (
        _extract_backtick_value(r"Status:\s*`([^`]+)`", text)
        or re.search(r"^status:\s*(.+)$", text, flags=re.MULTILINE).group(1).strip()
        if re.search(r"^status:\s*(.+)$", text, flags=re.MULTILINE)
        else None
    )
    phase = (
        _extract_backtick_value(r"Current phase:\s*`([^`]+)`", text)
        or re.search(r"^Phase:\s*(.+)$", text, flags=re.MULTILINE).group(1).strip()
        if re.search(r"^Phase:\s*(.+)$", text, flags=re.MULTILINE)
        else None
    )
    last_updated_match = re.search(r"^last_updated:\s*\"?([^\n\"]+)\"?$", text, flags=re.MULTILINE)
    last_activity_match = re.search(r"^last_activity:\s*(.+)$", text, flags=re.MULTILINE)
    return {
        "milestone": milestone,
        "status": status,
        "phase": phase,
        "next_commands": _parse_bullet_block(text, "Next command:"),
        "last_updated": last_updated_match.group(1).strip() if last_updated_match else _path_timestamp(path),
        "last_activity": last_activity_match.group(1).strip() if last_activity_match else None,
    }


def _run_demo_probe(spec: dict[str, Any] | None) -> dict[str, Any] | None:
    if not isinstance(spec, dict):
        return None
    url = spec.get("url")
    if not isinstance(url, str):
        return None
    try:
        with httpx.Client(timeout=2.0) as client:
            response = client.get(url)
    except httpx.HTTPError as exc:
        return {
            "ok": False,
            "url": url,
            "status_code": None,
            "detail": str(exc),
            "checked_at": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
            "method": spec.get("method") or "live_probe",
        }

    text = response.text
    expected_status = spec.get("expected_status")
    expected_text = spec.get("expected_text")
    ok = True
    if isinstance(expected_status, int):
        ok = ok and response.status_code == expected_status
    if isinstance(expected_text, str):
        ok = ok and expected_text in text
    return {
        "ok": ok,
        "url": url,
        "status_code": response.status_code,
        "detail": None if ok else f"Unexpected response for {url}",
        "checked_at": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "method": spec.get("method") or "live_probe",
    }


def _verification_status_from(probe: dict[str, Any] | None, artifact_path: Path | None) -> tuple[str, str | None, str]:
    if probe and probe.get("ok"):
        return "pass", probe.get("checked_at"), probe.get("method") or "live_probe"
    artifact_timestamp = _path_timestamp(artifact_path)
    if artifact_timestamp:
        artifact_dt = datetime.strptime(artifact_timestamp, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=UTC)
        freshness = "pass" if datetime.now(UTC) - artifact_dt <= DEMO_VERIFICATION_FRESHNESS else "stale"
        return freshness, artifact_timestamp, f"artifact:{artifact_path.name}"
    if probe:
        return "fail", probe.get("checked_at"), probe.get("method") or "live_probe"
    return "missing", None, "no_verification_evidence"


def _classify_demo_status(
    *,
    has_launch: bool,
    probe_ok: bool,
    verification_status: str,
    blockers: list[str],
) -> str:
    if not has_launch:
        return "unknown"
    if blockers and not probe_ok:
        return "blocked"
    if verification_status == "stale":
        return "stale"
    if probe_ok and verification_status == "pass" and not blockers:
        return "runnable_now"
    if verification_status == "fail":
        return "blocked"
    return "partial"


def _trust_level_for_status(demo_status: str, verification_status: str) -> str:
    if demo_status == "runnable_now" and verification_status == "pass":
        return "production-demo-safe"
    if demo_status in {"runnable_now", "partial"} and verification_status in {"pass", "stale"}:
        return "verified"
    return "exploratory"


def _demo_docs(repo_root: Path | None, hint: dict[str, Any]) -> dict[str, Any]:
    kb_path = _find_first_path(repo_root, hint.get("kb_candidates", []))
    howto_path = _find_first_path(repo_root, hint.get("howto_candidates", []))
    return {
        "kb_link": str(kb_path) if kb_path else None,
        "howto_link": str(howto_path) if howto_path else None,
    }


def _demo_contract_for_project(
    project: dict[str, Any],
    *,
    helper_lane: dict[str, Any] | None,
    seedgraph_lane: dict[str, Any] | None,
    ollarma_dashboard: dict[str, Any] | None,
) -> dict[str, Any]:
    project_id = project.get("project_id")
    repo_root = _repo_root_for_project(project)
    hint = PROJECT_DEMO_HINTS.get(project_id, {})
    state_path = repo_root / ".planning" / "STATE.md" if repo_root else None
    state = _parse_repo_state(state_path)
    docs = _demo_docs(repo_root, hint)
    artifact_candidates = list(hint.get("artifact_candidates", []))
    verification_artifact = _find_first_path(repo_root, artifact_candidates) or _latest_verification_path(repo_root)
    probe = _run_demo_probe(hint.get("probe"))

    blockers: list[str] = []
    dashboard_url = hint.get("dashboard_url")
    if dashboard_url is None:
        blockers.append("DASHBOARD_LINK_MISSING")
    if docs["kb_link"] is None:
        blockers.append("KB_DOC_MISSING")
    if docs["howto_link"] is None:
        blockers.append("HOWTO_DOC_MISSING")
    launch_command = hint.get("launch_command")
    if not isinstance(launch_command, str):
        blockers.append("LAUNCH_PATH_UNKNOWN")
    if probe and not probe.get("ok"):
        blockers.append("LIVE_PROBE_FAILED")
    if project.get("operating_posture") == "blocked":
        blockers.append("PROJECT_BLOCKED")
    if project_id == "ollarma" and helper_lane and helper_lane.get("status") != "ready":
        if helper_lane.get("blocked_reason"):
            blockers.append(str(helper_lane["blocked_reason"]))
    if project_id == "seedgraph" and seedgraph_lane and seedgraph_lane.get("status") != "ready":
        if seedgraph_lane.get("blocked_reason"):
            blockers.append(str(seedgraph_lane["blocked_reason"]))
    if project_id == AGENTHON_PROJECT_ID:
        blockers.extend(build_agenthon_live_readiness().get("blockers", []))

    verification_status, verified_at, verification_method = _verification_status_from(probe, verification_artifact)
    demo_status = _classify_demo_status(
        has_launch=isinstance(launch_command, str),
        probe_ok=bool(probe and probe.get("ok")),
        verification_status=verification_status,
        blockers=blockers,
    )
    trust_level = _trust_level_for_status(demo_status, verification_status)

    source_of_truth = [str(ACTIVE_PROJECTS_PATH)]
    if state_path and state_path.exists():
        source_of_truth.append(str(state_path))
    launch_source_path = _find_first_path(repo_root, hint.get("launch_source_candidates", []))
    if launch_source_path:
        source_of_truth.append(str(launch_source_path))
    if docs["kb_link"]:
        source_of_truth.append(str(docs["kb_link"]))
    if docs["howto_link"] and docs["howto_link"] not in source_of_truth:
        source_of_truth.append(str(docs["howto_link"]))
    if verification_artifact:
        source_of_truth.append(str(verification_artifact))
    if project_id == AGENTHON_PROJECT_ID:
        source_of_truth = [
            f"/api/demo-projects/{AGENTHON_PROJECT_ID}",
            f"/api/demo-projects/{AGENTHON_PROJECT_ID}/sample-run",
        ]

    next_steps = project.get("next_action") or "No next action exported."
    if "LAUNCH_PATH_UNKNOWN" in blockers:
        next_steps = "Document an exact launch path before presenting this project as demo-ready."
    elif "LIVE_PROBE_FAILED" in blockers and launch_command:
        next_steps = f"Run the documented launch path and re-verify: {launch_command}"

    dashboard_links = _project_links(
        project,
        helper_lane=helper_lane,
        seedgraph_lane=seedgraph_lane,
        ollarma_dashboard=ollarma_dashboard,
    )
    if dashboard_url and not any(link["href"] == dashboard_url for link in dashboard_links):
        dashboard_links.insert(0, {"label": hint.get("dashboard_label") or "Dashboard", "href": dashboard_url})
    if project_id == AGENTHON_PROJECT_ID:
        dashboard_links.extend(
            [
                {"label": "Demo project JSON", "href": f"/api/demo-projects/{AGENTHON_PROJECT_ID}"},
                {"label": "Sample run JSON", "href": f"/api/demo-projects/{AGENTHON_PROJECT_ID}/sample-run"},
            ]
        )

    last_updated = (
        state.get("last_updated")
        or project.get("last_activity")
        or _path_timestamp(state_path)
    )
    known_limitations = list(hint.get("limitations", []))
    if "DASHBOARD_LINK_MISSING" in blockers:
        known_limitations.append("No dedicated dashboard surface is documented in the repo.")
    if verification_status in {"stale", "missing"}:
        known_limitations.append("Verification is stale or absent; do not present as fully demo-ready.")

    escalation_paths = [
        {"label": path.name, "path": str(path)}
        for path in _resolve_candidates(repo_root, hint.get("escalation_candidates", []))
    ]
    key_artifacts = _collect_artifacts(repo_root, hint.get("artifact_candidates", []))
    if project_id == AGENTHON_PROJECT_ID:
        key_artifacts = [
            {"label": "Sample request JSON", "path": f"/api/demo-projects/{AGENTHON_PROJECT_ID}"},
            {"label": "Sample run JSON", "path": f"/api/demo-projects/{AGENTHON_PROJECT_ID}/sample-run"},
            {"label": "Demo overview doc", "path": f"/demo-projects/{AGENTHON_PROJECT_ID}/docs/overview"},
        ]
        escalation_paths = [
            {"label": "Human/AI intervention policy", "path": "/api/operator/contracts"},
            {"label": "Demo project JSON", "path": f"/api/demo-projects/{AGENTHON_PROJECT_ID}"},
        ]

    return {
        "project_id": project_id,
        "project_name": project.get("name"),
        "owner_system": project_id,
        "owner_repo": str(repo_root) if repo_root else project.get("repo_path"),
        "dashboard_link": dashboard_url,
        "kb_link": docs["kb_link"],
        "howto_link": docs["howto_link"],
        "project_status": project.get("current_status") or project.get("status"),
        "milestone_status": state.get("milestone") or project.get("milestone"),
        "blockers": list(dict.fromkeys(blockers)),
        "next_steps": next_steps,
        "demo_status": demo_status,
        "launch_path": launch_command,
        "verification_status": verification_status,
        "verified_at": verified_at,
        "verification_method": verification_method,
        "source_of_truth": source_of_truth,
        "last_updated": last_updated,
        "dependencies": list(hint.get("dependencies", [])),
        "known_limitations": list(dict.fromkeys(known_limitations)),
        "key_artifacts": key_artifacts,
        "escalation_paths": escalation_paths,
        "trust_level": trust_level,
        "dashboard_links": dashboard_links,
        "probe": probe,
    }


def build_demo_contracts(
    projects: list[dict[str, Any]],
    *,
    helper_lane: dict[str, Any] | None,
    seedgraph_lane: dict[str, Any] | None,
    ollarma_dashboard: dict[str, Any] | None,
) -> list[dict[str, Any]]:
    keyed = {project.get("project_id"): project for project in projects}
    ordered_ids = list(DEMO_PRIORITY_PROJECTS) + sorted(
        project_id for project_id in keyed if project_id not in DEMO_PRIORITY_PROJECTS
    )
    contracts = [
        _demo_contract_for_project(
            keyed[project_id],
            helper_lane=helper_lane,
            seedgraph_lane=seedgraph_lane,
            ollarma_dashboard=ollarma_dashboard,
        )
        for project_id in ordered_ids
        if project_id in keyed
    ]
    return contracts


def build_demo_portfolio_summary(contracts: list[dict[str, Any]]) -> dict[str, Any]:
    summary = {
        "runnable_now": 0,
        "minor_fixes": 0,
        "blocked": 0,
        "no_demo_contract": 0,
        "global_blockers": [],
        "show_live_first": [],
    }
    priority_rank = {project_id: index for index, project_id in enumerate(DEMO_PRIORITY_PROJECTS)}
    for contract in contracts:
        status = contract.get("demo_status")
        if status == "runnable_now":
            summary["runnable_now"] += 1
        elif status in {"partial", "stale"}:
            summary["minor_fixes"] += 1
        elif status == "blocked":
            summary["blocked"] += 1
        else:
            summary["no_demo_contract"] += 1
        for blocker in contract.get("blockers", []):
            if blocker not in summary["global_blockers"]:
                summary["global_blockers"].append(blocker)
    ranked = sorted(
        contracts,
        key=lambda item: (
            {"production-demo-safe": 0, "verified": 1, "exploratory": 2}.get(item.get("trust_level"), 3),
            {"runnable_now": 0, "partial": 1, "stale": 2, "blocked": 3, "unknown": 4}.get(
                item.get("demo_status"),
                5,
            ),
            priority_rank.get(item.get("project_id"), 99),
        ),
    )
    summary["show_live_first"] = [
        {
            "project_id": item["project_id"],
            "project_name": item["project_name"],
            "demo_status": item["demo_status"],
            "trust_level": item["trust_level"],
            "launch_path": item["launch_path"],
        }
        for item in ranked[:5]
    ]
    return summary


def get_project_record(project_id: str) -> dict[str, Any] | None:
    register = load_active_project_register()
    for record in register.get("projects", []):
        if record.get("project_id") == project_id:
            return record
    for record in _derived_project_records():
        if record.get("project_id") == project_id:
            return record
    return None


def _project_surface_summary(project: dict[str, Any]) -> str:
    surfaces = project.get("surfaces", {})
    if not isinstance(surfaces, dict) or not surfaces:
        return "No canonical surface inventory exported."
    observed = 0
    required_missing = 0
    for surface in surfaces.values():
        if not isinstance(surface, dict):
            continue
        status = surface.get("status")
        required = bool(surface.get("required"))
        if isinstance(status, str) and "observed" in status and "not observed" not in status:
            observed += 1
        if required and status == "missing":
            required_missing += 1
    return f"{observed} observed • {required_missing} required missing"


def _project_links(
    project: dict[str, Any],
    *,
    helper_lane: dict[str, Any] | None,
    seedgraph_lane: dict[str, Any] | None,
    ollarma_dashboard: dict[str, Any] | None,
) -> list[dict[str, str]]:
    project_id = project.get("project_id")
    links = [
        {"label": "Watchtower project page", "href": f"/projects/{project_id}"},
        {"label": "Project JSON", "href": f"/api/projects/{project_id}"},
    ]
    if project_id == "watchtower":
        links.extend(
            [
                {"label": "Watchtower dashboard", "href": "/"},
                {"label": "Watchtower overview JSON", "href": "/api/overview"},
            ]
        )
    elif project_id == "ollarma" and isinstance(ollarma_dashboard, dict):
        dashboard_url = ollarma_dashboard.get("dashboard_url")
        overview_url = ollarma_dashboard.get("overview_url")
        if isinstance(dashboard_url, str):
            links.append({"label": "Ollarma dashboard", "href": dashboard_url})
        if isinstance(overview_url, str):
            links.append({"label": "Ollarma overview JSON", "href": overview_url})
        links.append({"label": "Ollarma helper status", "href": "/api/ollarma/status"})
    elif project_id == "seedgraph" and isinstance(seedgraph_lane, dict):
        links.append({"label": "SeedGraph status", "href": "/api/seedgraph/status"})
        route = seedgraph_lane.get("route")
        if isinstance(route, dict):
            base_url = route.get("base_url")
            path = route.get("path")
            if isinstance(base_url, str) and isinstance(path, str):
                links.append({"label": "SeedGraph queue", "href": f"{base_url}{path}"})
    elif project_id == "overwatch":
        links.extend(
            [
                {"label": "Canonical project register", "href": "/api/projects"},
                {"label": "Canonical resources", "href": "/api/resources"},
            ]
        )
    elif project_id == AGENTHON_PROJECT_ID:
        links.extend(
            [
                {"label": "Demo project JSON", "href": f"/api/demo-projects/{AGENTHON_PROJECT_ID}"},
                {"label": "Sample run JSON", "href": f"/api/demo-projects/{AGENTHON_PROJECT_ID}/sample-run"},
            ]
        )
    elif project_id == "antigence":
        links.append({"label": "Operator contracts", "href": "/api/operator/contracts"})
    if project_id and project_id not in {"ollarma"} and isinstance(helper_lane, dict) and project_id == helper_lane.get("owner", "").lower():
        links.append({"label": "Helper lane status", "href": "/api/ollarma/status"})
    return links


def build_operator_summary_rows(
    projects: list[dict[str, Any]],
    *,
    helper_lane: dict[str, Any] | None,
    seedgraph_lane: dict[str, Any] | None,
    ollarma_dashboard: dict[str, Any] | None,
    next_actions: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    next_actions_by_owner = {
        item.get("owner"): item for item in next_actions if isinstance(item, dict) and item.get("owner")
    }
    rows: list[dict[str, Any]] = []
    for project in projects:
        project_id = project.get("project_id")
        lane_status = None
        blocker = None
        if project_id == "ollarma" and helper_lane:
            lane_status = helper_lane.get("status")
            blocker = helper_lane.get("blocked_reason")
        elif project_id == "seedgraph" and seedgraph_lane:
            lane_status = seedgraph_lane.get("status")
            blocker = seedgraph_lane.get("blocked_reason")
        if project.get("operating_posture") == "blocked" and not blocker:
            blocker = "PROJECT_BLOCKED"
        next_item = next_actions_by_owner.get(project_id, {})
        rows.append(
            {
                "project_id": project_id,
                "name": project.get("name"),
                "owner": project.get("owner_lane") or project.get("primary_role") or project.get("project_type"),
                "status": lane_status or project.get("current_status") or project.get("status") or "unknown",
                "posture": project.get("operating_posture") or "unknown",
                "milestone": project.get("milestone"),
                "blocker": blocker or (f"{project.get('gap_count')} declared gaps" if project.get("gap_count") else None),
                "next_action": next_item.get("action") or project.get("next_action") or "No next action exported.",
                "surface_summary": _project_surface_summary(project),
                "links": _project_links(
                    project,
                    helper_lane=helper_lane,
                    seedgraph_lane=seedgraph_lane,
                    ollarma_dashboard=ollarma_dashboard,
                ),
            }
        )
    return rows


def build_project_detail(project_id: str) -> dict[str, Any] | None:
    demo_detail = build_demo_project_detail(project_id)
    if demo_detail is not None:
        project = demo_detail["project"]
        return {
            "project": {
                "project_id": project["project_id"],
                "name": project.get("public_title") or project["project_id"],
                "direction": project.get("summary"),
                "owner_lane": project.get("owner_surface"),
                "milestone": project.get("version"),
                "current_status": demo_detail.get("sample_run", {}).get("status"),
                "boundary_notes": "Watchtower-local derived demo record; not canonical Overwatch truth.",
            },
            "record": {
                "project_id": project["project_id"],
                "name": project.get("public_title") or project["project_id"],
                "record_source": "watchtower-local-demo",
            },
            "summary": None,
            "demo_contract": {
                "demo_status": demo_detail.get("sample_run", {}).get("status"),
                "trust_level": "public-safe-demo",
                "project_status": project.get("status"),
                "milestone_status": project.get("version"),
                "dashboard_links": [
                    {"label": "Project page", "href": f"/projects/{project_id}"},
                    {"label": "Project JSON", "href": f"/api/demo-projects/{project_id}"},
                    {"label": "Sample run JSON", "href": f"/api/demo-projects/{project_id}/sample-run"},
                ],
                "dashboard_link": f"/projects/{project_id}",
                "kb_link": f"/demo-projects/{project_id}/docs/overview",
                "howto_link": f"/demo-projects/{project_id}/docs/demo-script",
                "launch_path": f"GET /projects/{project_id}/demo-run",
                "verification_status": "fixture",
                "verified_at": None,
                "verification_method": "public-safe sample fixture",
                "blockers": demo_detail.get("live_blockers", []),
                "next_steps": (
                    "Configure required env vars for live Tavily/Daytona runs."
                    if demo_detail.get("live_blockers")
                    else "Run the sample demo from the project page."
                ),
                "last_updated": None,
                "dependencies": [
                    "OPENAI_API_KEY",
                    "TAVILY_API_KEY",
                    "DAYTONA_API_KEY",
                    "DAYTONA_API_URL",
                    "LANGSMITH_API_KEY (optional)",
                ],
                "known_limitations": [
                    "Watchtower-local derived demo record only",
                    "SeedGraph and Antigence are supporting lanes unless live integrations are added",
                ],
                "source_of_truth": [
                    str((ROOT / "demo" / project_id / "project.json")),
                    str((ROOT / "demo" / project_id / "sample-run.json")),
                ],
                "key_artifacts": [
                    {"label": "project.json", "path": str(ROOT / "demo" / project_id / "project.json")},
                    {"label": "sample-request.json", "path": str(ROOT / "demo" / project_id / "sample-request.json")},
                    {"label": "sample-run.json", "path": str(ROOT / "demo" / project_id / "sample-run.json")},
                ],
                "escalation_paths": [],
            },
            "surfaces": {},
            "owned_surface": None,
            "imported_surfaces": [],
        }
    record = get_project_record(project_id)
    if record is None:
        return None
    overview = build_watchtower_overview()
    project = next((item for item in list_projects() if item.get("project_id") == project_id), None)
    if project is None:
        return None
    summary_row = next(
        (item for item in overview.get("operator_summary", []) if item.get("project_id") == project_id),
        None,
    )
    demo_contract = next(
        (item for item in overview.get("demo_contracts", []) if item.get("project_id") == project_id),
        None,
    )
    return {
        "project": project,
        "record": record,
        "summary": summary_row,
        "demo_contract": demo_contract,
        "surfaces": record.get("surfaces", {}),
        "owned_surface": record.get("ownership_contract", {}).get("owned_surface"),
        "imported_surfaces": record.get("ownership_contract", {}).get("imported_surfaces", []),
        "agenthon_demo": build_agenthon_demo_package() if project_id == AGENTHON_PROJECT_ID else None,
    }


def _path_check(path: Path) -> dict[str, Any]:
    return {
        "path": str(path),
        "exists": path.exists(),
    }


def _canonical_read_lane() -> dict[str, Any]:
    checks = [
        _path_check(ACTIVE_PROJECTS_PATH),
        _path_check(RESOURCE_REGISTRY_PATH),
        _path_check(INVENTORY_DB_PATH),
    ]
    missing = [item["path"] for item in checks if not item["exists"]]
    return {
        "lane_id": "overwatch-canonical-read",
        "owner": "Overwatch",
        "watchtower_role": "read_and_render",
        "scope": "Canonical project, resource, and inventory reads",
        "status": "ready" if not missing else "blocked",
        "blocked_reason": None if not missing else "MISSING_CANONICAL_EXPORT",
        "detail": "Watchtower reads canonical control-plane exports and does not rewrite them.",
        "next_fix": None if not missing else f"Restore missing Overwatch export(s): {', '.join(missing)}",
        "checks": checks,
    }


def _canonical_write_lane() -> dict[str, Any]:
    return {
        "lane_id": "overwatch-canonical-write",
        "owner": "Overwatch",
        "watchtower_role": "never_write_directly",
        "scope": "Canonical truth, provenance, inventory, claim, and graph writes",
        "status": "review_gated",
        "blocked_reason": "WATCHTOWER_READ_ONLY_BY_POLICY",
        "detail": "Canonical mutations stay on the Overwatch-governed lane even when Watchtower launches approved flows.",
        "next_fix": "Use an Overwatch-approved writeback or review lane instead of Watchtower direct mutation.",
    }


def _safe_json(response: httpx.Response) -> tuple[dict[str, Any] | None, str | None]:
    try:
        payload = response.json()
    except ValueError:
        return None, "INVALID_JSON"
    if not isinstance(payload, dict):
        return None, "INVALID_JSON"
    return payload, None


def _probe_ollarma_route() -> dict[str, Any]:
    probe_timeout = min(OLLARMA_TIMEOUT_SECONDS, 4.0)
    try:
        with httpx.Client(base_url=OLLARMA_BASE_URL, timeout=probe_timeout) as client:
            health = client.get("/health")
    except httpx.TimeoutException:
        return {
            "status": "degraded",
            "blocked_reason": "OLLARMA_TIMEOUT",
            "detail": f"Watchtower could not complete the Ollarma helper readiness probe inside {probe_timeout:.1f}s.",
            "next_fix": "Keep Ollarma as the helper lane, but raise timeout budget or reduce helper health latency.",
            "health": {"ok": False, "status_code": None},
            "route": None,
        }
    except httpx.HTTPError as exc:
        return {
            "status": "blocked",
            "blocked_reason": "OLLARMA_UNAVAILABLE",
            "detail": str(exc),
            "next_fix": "Start or repair the Ollarma HTTP service on the configured local URL.",
            "health": {"ok": False, "status_code": None},
            "route": None,
        }

    if health.status_code != 200:
        return {
            "status": "degraded",
            "blocked_reason": f"OLLARMA_HEALTH_HTTP_{health.status_code}",
            "detail": f"Ollarma health probe returned HTTP {health.status_code}.",
            "next_fix": "Restore the Ollarma health endpoint before treating the helper lane as ready.",
            "health": {"ok": False, "status_code": health.status_code},
            "route": None,
        }

    health_body, health_error = _safe_json(health)
    if health_error:
        return {
            "status": "blocked",
            "blocked_reason": "OLLARMA_HEALTH_INVALID_JSON",
            "detail": "Ollarma health probe did not return a valid JSON body.",
            "next_fix": "Repair the Ollarma health API contract before using the helper lane.",
            "health": {"ok": True, "status_code": health.status_code},
            "route": None,
        }

    helper_chat = health_body.get("helper_chat") if isinstance(health_body.get("helper_chat"), dict) else {}
    chat_selection = (
        health_body.get("chat_selection") if isinstance(health_body.get("chat_selection"), dict) else {}
    )
    route_selection = (
        health_body.get("route_selection") if isinstance(health_body.get("route_selection"), dict) else {}
    )

    recent_receipt: dict[str, Any] | None = None
    try:
        with httpx.Client(base_url=OLLARMA_BASE_URL, timeout=probe_timeout) as client:
            overview = client.get("/dashboard/overview")
        overview_body, overview_error = _safe_json(overview)
        if not overview_error and overview.status_code < 400:
            receipts = (
                overview_body.get("recent_route_receipts")
                if isinstance(overview_body.get("recent_route_receipts"), list)
                else []
            )
            if receipts and isinstance(receipts[0], dict):
                recent_receipt = receipts[0]
    except httpx.HTTPError:
        recent_receipt = None

    effective_model = (
        route_selection.get("model")
        or chat_selection.get("model")
        or helper_chat.get("effective_model")
    )
    recovery_commands = (
        route_selection.get("recovery_commands")
        or chat_selection.get("recovery_commands")
        or helper_chat.get("recovery_commands")
        or []
    )

    selection_blocked = route_selection.get("status") == "blocked" or chat_selection.get("status") == "blocked"
    if selection_blocked:
        reason_code = (
            route_selection.get("reason_code")
            or chat_selection.get("reason_code")
            or helper_chat.get("reason_code")
            or "OLLARMA_SELECTION_BLOCKED"
        )
        detail = (
            route_selection.get("detail")
            or chat_selection.get("detail")
            or helper_chat.get("detail")
            or "Ollarma helper selection is blocked."
        )
        next_fix = "\n".join(recovery_commands) if recovery_commands else "Regenerate a valid Ollarma selection artifact."
        return {
            "status": "blocked",
            "blocked_reason": reason_code,
            "detail": detail,
            "next_fix": next_fix,
            "health": {"ok": True, "status_code": health.status_code},
            "route": {
                "lane": recent_receipt.get("lane") if recent_receipt else None,
                "kb_status": recent_receipt.get("kb_status") if recent_receipt else None,
                "status_code": 200,
                "query_class": recent_receipt.get("query_class") if recent_receipt else None,
                "selected_model": recent_receipt.get("selected_model") if recent_receipt else effective_model,
            },
        }

    detail = "Ollarma helper lane is ready."
    if effective_model:
        detail = f"Ollarma helper lane is ready with validated selection on {effective_model}."
    return {
        "status": "ready",
        "blocked_reason": None,
        "detail": detail,
        "next_fix": None,
        "health": {"ok": health.status_code == 200, "status_code": health.status_code},
        "route": {
            "lane": recent_receipt.get("lane") if recent_receipt else None,
            "kb_status": recent_receipt.get("kb_status") if recent_receipt else "ready",
            "status_code": 200,
            "query_class": recent_receipt.get("query_class") if recent_receipt else None,
            "selected_model": recent_receipt.get("selected_model") if recent_receipt else effective_model,
        },
    }


def _ollarma_helper_lane() -> dict[str, Any]:
    probe = _probe_ollarma_route()
    return {
        "lane_id": "ollarma-bounded-helper",
        "owner": "Ollarma",
        "watchtower_role": "default_local_lane",
        "scope": "Local helper routing, KB lookup, and low-cost operator assistance",
        **probe,
        "detail": probe["detail"],
    }


def _ollarma_workflow_lane() -> dict[str, Any]:
    return {
        "lane_id": "ollarma-deterministic-workflow",
        "owner": "Ollarma",
        "watchtower_role": "trigger_only",
        "scope": "Manifest-backed deterministic workflow execution",
        "status": "trigger_only",
        "blocked_reason": "MANIFEST_REQUIRED",
        "detail": "Watchtower can trigger approved Ollarma workflows, but execution remains manifest-bounded and local.",
        "next_fix": "Provide an approved manifest_ref and step_id when execution is actually intended.",
    }


def _ollarma_dashboard_surface() -> dict[str, Any]:
    probe_timeout = min(OLLARMA_TIMEOUT_SECONDS, 4.0)
    dashboard_url = f"{OLLARMA_BASE_URL.rstrip('/')}/dashboard"
    overview_url = f"{OLLARMA_BASE_URL.rstrip('/')}/dashboard/overview"

    try:
        with httpx.Client(base_url=OLLARMA_BASE_URL, timeout=probe_timeout) as client:
            response = client.get("/dashboard/overview")
    except httpx.TimeoutException:
        return {
            "owner": "Ollarma",
            "status": "degraded",
            "blocked_reason": "OLLARMA_DASHBOARD_TIMEOUT",
            "detail": f"Watchtower could not load the Ollarma dashboard overview inside {probe_timeout:.1f}s.",
            "next_fix": "Open the Ollarma dashboard directly or reduce dashboard overview latency.",
            "dashboard_url": dashboard_url,
            "overview_url": overview_url,
        }
    except httpx.HTTPError as exc:
        return {
            "owner": "Ollarma",
            "status": "blocked",
            "blocked_reason": "OLLARMA_DASHBOARD_UNAVAILABLE",
            "detail": str(exc),
            "next_fix": "Start or repair the Ollarma HTTP dashboard surface on the configured local URL.",
            "dashboard_url": dashboard_url,
            "overview_url": overview_url,
        }

    if response.status_code >= 400:
        return {
            "owner": "Ollarma",
            "status": "blocked",
            "blocked_reason": f"OLLARMA_DASHBOARD_HTTP_{response.status_code}",
            "detail": f"Ollarma dashboard overview returned HTTP {response.status_code}.",
            "next_fix": "Repair the Ollarma dashboard overview route before using Watchtower's connected view.",
            "dashboard_url": dashboard_url,
            "overview_url": overview_url,
        }

    payload, payload_error = _safe_json(response)
    if payload_error:
        return {
            "owner": "Ollarma",
            "status": "blocked",
            "blocked_reason": "OLLARMA_DASHBOARD_INVALID_JSON",
            "detail": "Ollarma dashboard overview did not return a valid JSON payload.",
            "next_fix": "Repair the Ollarma dashboard overview contract before consuming it from Watchtower.",
            "dashboard_url": dashboard_url,
            "overview_url": overview_url,
        }

    boundary = payload.get("boundary")
    if not isinstance(boundary, dict) or "dashboard_path" not in boundary or "overview_path" not in boundary:
        return {
            "owner": "Ollarma",
            "status": "dependency_pending",
            "blocked_reason": "OLLARMA_DASHBOARD_SCHEMA_INCOMPLETE",
            "detail": "Ollarma dashboard overview is reachable, but the typed boundary contract is incomplete.",
            "next_fix": "Expose the documented Ollarma dashboard boundary fields before relying on the connected Watchtower view.",
            "dashboard_url": dashboard_url,
            "overview_url": overview_url,
        }

    kb_status = payload.get("kb_status") if isinstance(payload.get("kb_status"), list) else []
    readiness = payload.get("readiness") if isinstance(payload.get("readiness"), list) else []
    recent_route_receipts = (
        payload.get("recent_route_receipts") if isinstance(payload.get("recent_route_receipts"), list) else []
    )
    workflow_runs = payload.get("workflow_runs") if isinstance(payload.get("workflow_runs"), list) else []
    autopilot_runs = payload.get("autopilot_runs") if isinstance(payload.get("autopilot_runs"), list) else []
    operator_resources = payload.get("operator_resources") if isinstance(payload.get("operator_resources"), list) else []

    return {
        "owner": "Ollarma",
        "status": "ready",
        "blocked_reason": None,
        "detail": "Watchtower is connected to the Ollarma-owned helper dashboard and its typed overview export.",
        "next_fix": None,
        "dashboard_url": f"{OLLARMA_BASE_URL.rstrip('/')}{boundary.get('dashboard_path', '/dashboard')}",
        "overview_url": f"{OLLARMA_BASE_URL.rstrip('/')}{boundary.get('overview_path', '/dashboard/overview')}",
        "run_detail_path_template": (
            f"{OLLARMA_BASE_URL.rstrip('/')}{boundary.get('run_detail_path_template', '/dashboard/runs/{{run_id}}')}"
        ),
        "interactive": bool(boundary.get("interactive")),
        "read_only": bool(boundary.get("read_only")),
        "integration_contract": boundary.get("integration_contract"),
        "review_gate": boundary.get("review_gate"),
        "helper_surfaces": boundary.get("helper_surfaces", []),
        "project_count": int(payload.get("project_count") or 0),
        "project_names": payload.get("projects", [])[:6],
        "kb_ready_count": sum(1 for item in kb_status if isinstance(item, dict) and item.get("status") == "ready"),
        "readiness": readiness[:3],
        "recent_route_receipts": recent_route_receipts[:3],
        "workflow_runs": workflow_runs[:3],
        "autopilot_runs": autopilot_runs[:3],
        "operator_resources": operator_resources[:3],
    }


def _seedgraph_identity_lane() -> dict[str, Any]:
    repo_present = SEEDGRAPH_ROOT.exists()
    triage_api = SEEDGRAPH_ROOT / "src" / "seedgraph" / "triage" / "api.py"
    service_api = SEEDGRAPH_ROOT / "src" / "seedgraph" / "service" / "api.py"
    detail = "SeedGraph owns content identity, Merkle roots, signed import proofs, and publication-truth hardening."
    checks = [
        {"path": str(SEEDGRAPH_ROOT), "exists": repo_present},
        {"path": str(triage_api), "exists": triage_api.exists()},
        {"path": str(service_api), "exists": service_api.exists()},
    ]
    if not repo_present or not triage_api.exists():
        return {
            "lane_id": "seedgraph-provenance-identity",
            "owner": "SeedGraph",
            "watchtower_role": "surface_when_ready",
            "scope": "Content-addressed identity, Merkle proofs, signed import receipts, and publication-truth contracts",
            "status": "dependency_pending",
            "blocked_reason": "SEEDGRAPH_READ_SURFACE_PENDING",
            "detail": detail,
            "next_fix": "Expose a stable SeedGraph read surface for triage/publication-family status before Watchtower consumes it.",
            "checks": checks,
        }

    if not SEEDGRAPH_BASE_URL:
        return {
            "lane_id": "seedgraph-provenance-identity",
            "owner": "SeedGraph",
            "watchtower_role": "surface_when_ready",
            "scope": "Content-addressed identity, Merkle proofs, signed import receipts, and publication-truth contracts",
            "status": "dependency_pending",
            "blocked_reason": "SEEDGRAPH_URL_UNCONFIGURED",
            "detail": f"{detail} Read surfaces exist in-repo (`/api/triage/queue`, `/api/triage/documents/{{seed_id}}`, `/api/v1/review/queue`), but Watchtower has no configured SeedGraph base URL.",
            "next_fix": "Set WATCHTOWER_SEEDGRAPH_URL to a running SeedGraph triage/service deployment.",
            "checks": checks,
        }

    probe_timeout = min(SEEDGRAPH_TIMEOUT_SECONDS, 4.0)
    try:
        with httpx.Client(base_url=SEEDGRAPH_BASE_URL, timeout=probe_timeout) as client:
            response = client.get("/api/triage/queue")
    except httpx.TimeoutException:
        return {
            "lane_id": "seedgraph-provenance-identity",
            "owner": "SeedGraph",
            "watchtower_role": "surface_when_ready",
            "scope": "Content-addressed identity, Merkle proofs, signed import receipts, and publication-truth contracts",
            "status": "degraded",
            "blocked_reason": "SEEDGRAPH_TIMEOUT",
            "detail": f"SeedGraph triage probe did not complete inside {probe_timeout:.1f}s.",
            "next_fix": "Reduce SeedGraph route latency or increase the configured probe timeout.",
            "checks": checks,
        }
    except httpx.HTTPError as exc:
        return {
            "lane_id": "seedgraph-provenance-identity",
            "owner": "SeedGraph",
            "watchtower_role": "surface_when_ready",
            "scope": "Content-addressed identity, Merkle proofs, signed import receipts, and publication-truth contracts",
            "status": "blocked",
            "blocked_reason": "SEEDGRAPH_UNAVAILABLE",
            "detail": str(exc),
            "next_fix": "Start or repair the configured SeedGraph read-surface service.",
            "checks": checks,
        }

    if response.status_code == 403:
        return {
            "lane_id": "seedgraph-provenance-identity",
            "owner": "SeedGraph",
            "watchtower_role": "surface_when_ready",
            "scope": "Content-addressed identity, Merkle proofs, signed import receipts, and publication-truth contracts",
            "status": "blocked",
            "blocked_reason": "SEEDGRAPH_AUTH_REQUIRED",
            "detail": "SeedGraph triage route is reachable but rejected the request without the required workspace permissions.",
            "next_fix": "Provide the approved SeedGraph auth/workspace context for Watchtower read access.",
            "checks": checks,
        }
    if response.status_code >= 400:
        return {
            "lane_id": "seedgraph-provenance-identity",
            "owner": "SeedGraph",
            "watchtower_role": "surface_when_ready",
            "scope": "Content-addressed identity, Merkle proofs, signed import receipts, and publication-truth contracts",
            "status": "blocked",
            "blocked_reason": "SEEDGRAPH_ROUTE_ERROR",
            "detail": f"SeedGraph triage probe returned HTTP {response.status_code}.",
            "next_fix": "Repair the SeedGraph triage API contract or reconfigure Watchtower to the correct route.",
            "checks": checks,
        }

    payload, payload_error = _safe_json(response)
    if payload_error:
        return {
            "lane_id": "seedgraph-provenance-identity",
            "owner": "SeedGraph",
            "watchtower_role": "surface_when_ready",
            "scope": "Content-addressed identity, Merkle proofs, signed import receipts, and publication-truth contracts",
            "status": "degraded",
            "blocked_reason": "SEEDGRAPH_ROUTE_INVALID_JSON",
            "detail": "SeedGraph triage probe returned a non-JSON response.",
            "next_fix": "Repair the SeedGraph triage API contract before treating the provenance lane as usable.",
            "checks": checks,
        }

    items = payload.get("items")
    if not isinstance(items, list) or "total" not in payload:
        return {
            "lane_id": "seedgraph-provenance-identity",
            "owner": "SeedGraph",
            "watchtower_role": "surface_when_ready",
            "scope": "Content-addressed identity, Merkle proofs, signed import receipts, and publication-truth contracts",
            "status": "dependency_pending",
            "blocked_reason": "SEEDGRAPH_TRIAGE_SCHEMA_INCOMPLETE",
            "detail": "SeedGraph triage queue is reachable, but its response schema is not sufficient for Watchtower's read contract.",
            "next_fix": "Expose the documented triage queue schema with `total` and `items` fields.",
            "checks": checks,
        }

    proof_keys = {
        "proof_id",
        "proof_status",
        "manifest_hash",
        "source_member_root",
        "derived_asset_root",
        "receipt_root",
        "proof_signature_digest",
    }
    declared_proof_fields = payload.get("proof_fields")
    declared_field_names = (
        {field for field in declared_proof_fields if isinstance(field, str)}
        if isinstance(declared_proof_fields, list)
        else set()
    )
    declared_proof_contract = bool(payload.get("proof_contract_version")) and proof_keys.issubset(declared_field_names)
    has_proof_contract = any(
        isinstance(item, dict) and any(key in item for key in proof_keys)
        for item in items
    )
    if not has_proof_contract and not declared_proof_contract:
        return {
            "lane_id": "seedgraph-provenance-identity",
            "owner": "SeedGraph",
            "watchtower_role": "surface_when_ready",
            "scope": "Content-addressed identity, Merkle proofs, signed import receipts, and publication-truth contracts",
            "status": "dependency_pending",
            "blocked_reason": "SEEDGRAPH_PROVENANCE_CONTRACT_PENDING",
            "detail": "SeedGraph triage queue is reachable, but Watchtower has not yet validated a provenance-specific proof contract from this route.",
            "next_fix": "Expose proof-bearing SeedGraph read fields or point Watchtower at a provenance-oriented read surface.",
            "checks": checks,
            "route": {
                "base_url": SEEDGRAPH_BASE_URL,
                "path": "/api/triage/queue",
                "status_code": response.status_code,
                "queue_total": payload.get("total"),
                "proof_contract_version": payload.get("proof_contract_version"),
                "proof_fields": sorted(declared_field_names) if declared_field_names else [],
            },
        }

    return {
        "lane_id": "seedgraph-provenance-identity",
        "owner": "SeedGraph",
        "watchtower_role": "surface_when_ready",
        "scope": "Content-addressed identity, Merkle proofs, signed import receipts, and publication-truth contracts",
        "status": "ready",
        "blocked_reason": None,
        "detail": "SeedGraph read surface is reachable and exposes proof-bearing provenance fields for downstream status reads.",
        "next_fix": None,
        "checks": checks,
        "route": {
            "base_url": SEEDGRAPH_BASE_URL,
            "path": "/api/triage/queue",
            "status_code": response.status_code,
            "queue_total": payload.get("total"),
            "proof_contract_version": payload.get("proof_contract_version"),
            "proof_fields": sorted(declared_field_names) if declared_field_names else [],
        },
    }


def _antigence_review_lane() -> dict[str, Any]:
    return {
        "lane_id": "antigence-risk-review",
        "owner": "Antigence",
        "watchtower_role": "surface_outcomes",
        "scope": "Risk review, escalation, and hard-stop decisions",
        "status": "review_gated",
        "blocked_reason": "HUMAN_OR_REVIEW_REQUIRED",
        "detail": "Risky or low-confidence work stays on Antigence or human review lanes rather than silently continuing in Watchtower.",
        "next_fix": "Escalate when governance, provenance, or safety rules require review.",
    }


def build_operator_contract() -> dict[str, Any]:
    lanes = [
        _canonical_read_lane(),
        _canonical_write_lane(),
        _ollarma_helper_lane(),
        _ollarma_workflow_lane(),
        _seedgraph_identity_lane(),
        _antigence_review_lane(),
    ]
    blockers = [
        {
            "lane_id": lane["lane_id"],
            "owner": lane["owner"],
            "blocked_reason": lane["blocked_reason"],
            "next_fix": lane.get("next_fix"),
        }
        for lane in lanes
        if lane["status"] in {"blocked", "degraded", "dependency_pending", "review_gated"}
    ]
    return {
        "systems": [
            {
                "system": "Overwatch",
                "owner": "Overwatch",
                "scope": "Canonical truth, governance, provenance, inventory, claims, resources, and graph writes",
                "watchtower_role": "Read canonical outputs and expose operator views.",
            },
            {
                "system": "Watchtower",
                "owner": "Watchtower",
                "scope": "Operator UI, query surfaces, lane selection, and approved orchestration triggers",
                "watchtower_role": "Own the downstream operator surface without becoming a second source of truth.",
            },
            {
                "system": "Ollarma",
                "owner": "Ollarma",
                "scope": "Bounded local helper routing, KB lookup, and deterministic workflow execution",
                "watchtower_role": "Use as the default low-cost helper and execution lane.",
            },
            {
                "system": "SeedGraph",
                "owner": "SeedGraph",
                "scope": "Content identity, Merkle-rooted integrity, signed import proofs, and publication-truth hardening",
                "watchtower_role": "Surface provenance state and future import status once stable read surfaces exist.",
            },
            {
                "system": "Antigence",
                "owner": "Antigence",
                "scope": "Risk review and escalation",
                "watchtower_role": "Show review gates and outcomes instead of bypassing them.",
            },
        ],
        "lanes": lanes,
        "blockers": blockers,
        "lessons": [
            {
                "owner": "Overwatch",
                "lesson": "Canonical truth stays upstream; Watchtower should consume exports and show whether they are present, never recreate them locally.",
            },
            {
                "owner": "Ollarma",
                "lesson": "Bounded local help is the default lane, but exact route blockers such as KB gaps must stay visible in the operator surface.",
            },
            {
                "owner": "SeedGraph",
                "lesson": "Content identity and import proofs belong to a dedicated provenance substrate, not to a Watchtower shadow model.",
            },
        ],
        "sources": [
            str(DOCS_DIR / "CONTROL_PLANE_BOUNDARY.md"),
            str(OVERWATCH_ROOT / "AGENTS.md"),
            str(SEEDGRAPH_ROOT / "AGENTS.md"),
        ],
    }


def _phase_sort_key(value: str | None) -> tuple[int, ...]:
    if not value:
        return (999,)
    parts: list[int] = []
    for token in value.split("."):
        try:
            parts.append(int(token))
        except ValueError:
            parts.append(999)
    return tuple(parts)


def _parse_phase_number(text: str | None) -> str | None:
    if not text:
        return None
    match = re.search(r"Phase\s+(\d+(?:\.\d+)*)", text, flags=re.IGNORECASE)
    return match.group(1) if match else None


def _parse_bullet_block(text: str, heading: str) -> list[str]:
    lines = text.splitlines()
    capture = False
    items: list[str] = []
    current: list[str] = []
    for raw_line in lines:
        line = raw_line.rstrip()
        if not capture:
            if line.strip() == heading:
                capture = True
            continue
        if not line.strip():
            if current:
                items.append(" ".join(current).strip())
                current = []
            if items:
                break
            continue
        if line.startswith("- "):
            if current:
                items.append(" ".join(current).strip())
            current = [line[2:].strip()]
            continue
        if line.startswith("  - "):
            if current:
                current.append(line[4:].strip())
            continue
        if current:
            items.append(" ".join(current).strip())
            current = []
        if items:
            break
    if current:
        items.append(" ".join(current).strip())
    return items


def _parse_roadmap_phases(text: str, current_phase: str | None) -> list[dict[str, Any]]:
    lines = text.splitlines()
    current_number = _parse_phase_number(current_phase)
    phases: list[dict[str, Any]] = []
    index = 0
    while index < len(lines):
        line = lines[index]
        match = re.match(r"### Phase (\d+(?:\.\d+)*)(?::\s*(.+))?$", line)
        if not match:
            index += 1
            continue
        number = match.group(1)
        title = match.group(2) or f"Phase {number}"
        plans = None
        summary = ""
        index += 1
        while index < len(lines) and not lines[index].startswith("### Phase "):
            current = lines[index].strip()
            if current.startswith("**Plans:**"):
                plans = current.removeprefix("**Plans:**").strip()
            elif not summary and current and not current.startswith("**") and not current.startswith("Plans:"):
                summary = current
            index += 1
        status = "upcoming"
        if current_number:
            if _phase_sort_key(number) < _phase_sort_key(current_number):
                status = "complete"
            elif number == current_number:
                status = "current"
        phases.append(
            {
                "phase_number": number,
                "title": title,
                "summary": summary,
                "plans": plans,
                "status": status,
            }
        )
    return phases


def _parse_todo(path: Path) -> dict[str, Any]:
    text = _read_text(path)
    title_match = re.search(r"^title:\s*(.+)$", text, flags=re.MULTILINE)
    created_match = re.search(r"^created:\s*(.+)$", text, flags=re.MULTILINE)
    problem_match = re.search(
        r"## Problem\s+(.*?)(?:\n## |\Z)",
        text,
        flags=re.DOTALL,
    )
    summary = ""
    if problem_match:
        summary = " ".join(problem_match.group(1).split())
    return {
        "title": title_match.group(1).strip() if title_match else path.stem,
        "created": created_match.group(1).strip() if created_match else None,
        "summary": summary,
        "body": text,
        "path": str(path),
    }


def build_planning_overview() -> dict[str, Any]:
    state_text = _read_text(STATE_PATH)
    roadmap_text = _read_text(ROADMAP_PATH)
    audit_text = _read_text(MILESTONE_AUDIT_PATH)

    milestone_match = re.search(r"^Current milestone:\s*`([^`]+)`", state_text, flags=re.MULTILINE)
    phase_match = re.search(r"^Current phase:\s*`([^`]+)`", state_text, flags=re.MULTILINE)
    status_match = re.search(r"^Status:\s*`([^`]+)`", state_text, flags=re.MULTILINE)
    audit_status_match = re.search(r"^status:\s*(.+)$", audit_text, flags=re.MULTILINE)

    next_commands = _parse_bullet_block(state_text, "Next command:")
    notes = _parse_bullet_block(state_text, "Notes:")
    state_todos = _parse_bullet_block(state_text, "### Pending Todos")
    pending_todos = [
        _parse_todo(path)
        for path in sorted(PENDING_TODOS_DIR.glob("*.md"))
    ]
    phases = _parse_roadmap_phases(roadmap_text, phase_match.group(1) if phase_match else None)

    return {
        "milestone": milestone_match.group(1) if milestone_match else None,
        "current_phase": phase_match.group(1) if phase_match else None,
        "current_phase_number": _parse_phase_number(phase_match.group(1) if phase_match else None),
        "status": status_match.group(1) if status_match else None,
        "next_commands": next_commands,
        "notes": notes,
        "audit_status": audit_status_match.group(1).strip() if audit_status_match else None,
        "pending_todos": pending_todos,
        "pending_todo_count": len(pending_todos),
        "state_todos": state_todos,
        "phases": phases,
        "source_paths": [str(STATE_PATH), str(ROADMAP_PATH), str(MILESTONE_AUDIT_PATH)],
    }


def build_active_blockers(operator_contract: dict[str, Any]) -> list[dict[str, Any]]:
    lanes = operator_contract.get("lanes", [])
    blockers = [
        {
            "lane_id": lane["lane_id"],
            "owner": lane["owner"],
            "scope": lane["scope"],
            "status": lane["status"],
            "blocked_reason": lane.get("blocked_reason"),
            "detail": lane.get("detail"),
            "next_fix": lane.get("next_fix"),
        }
        for lane in lanes
        if lane["status"] in {"blocked", "degraded", "dependency_pending"}
    ]
    return blockers


def build_next_actions(
    projects: list[dict[str, Any]],
    planning: dict[str, Any],
    active_blockers: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    actions: list[dict[str, Any]] = []
    for command in planning.get("next_commands", []):
        actions.append(
            {
                "owner": "Watchtower",
                "lane": "gsigmad",
                "title": "Advance the current dashboard phase",
                "action": command,
                "reason": planning.get("status") or "planning state",
            }
        )
    for blocker in active_blockers:
        if blocker.get("next_fix"):
            actions.append(
                {
                    "owner": blocker["owner"],
                    "lane": blocker["lane_id"],
                    "title": blocker["scope"],
                    "action": blocker["next_fix"],
                    "reason": blocker.get("blocked_reason"),
                }
            )
    for project in projects:
        next_action = project.get("next_action")
        if not next_action:
            continue
        actions.append(
            {
                "owner": project.get("project_id"),
                "lane": project.get("owner_lane") or project.get("primary_role"),
                "title": project.get("name") or project.get("project_id"),
                "action": next_action,
                "reason": project.get("operating_posture") or project.get("status"),
            }
        )
        if len(actions) >= 8:
            break
    return actions[:8]


def build_task_feed(projects: list[dict[str, Any]], planning: dict[str, Any]) -> list[dict[str, Any]]:
    tasks: list[dict[str, Any]] = []
    for todo in planning.get("pending_todos", []):
        tasks.append(
            {
                "owner": "Watchtower",
                "lane": "planning",
                "title": todo["title"],
                "detail": todo.get("summary"),
                "source": "watchtower-planning",
            }
        )
    for project in projects:
        if not project.get("next_action"):
            continue
        tasks.append(
            {
                "owner": project.get("project_id"),
                "lane": project.get("owner_lane") or project.get("primary_role"),
                "title": project.get("next_action"),
                "detail": project.get("name"),
                "source": "overwatch-project-register",
            }
        )
    return tasks[:12]


def _extract_backtick_value(pattern: str, text: str) -> str | None:
    match = re.search(pattern, text, flags=re.IGNORECASE)
    return match.group(1) if match else None


def _extract_backtick_values(pattern: str, text: str) -> list[str]:
    return re.findall(pattern, text, flags=re.IGNORECASE)


def _opportunity_lane_ids() -> tuple[str, ...]:
    return ("community_lab_events", "lab_ops_stack")


def _canonical_opportunity_items() -> list[dict[str, Any]]:
    register = load_active_project_register()
    registry = load_resource_registry()
    lane_ids = set(_opportunity_lane_ids())
    items: list[dict[str, Any]] = []

    for project in register.get("projects", []):
        lane_id = project.get("project_id")
        direction = project.get("direction_contract", {})
        owner_lane = direction.get("owner_lane")
        matched_lane = lane_id if lane_id in lane_ids else owner_lane if owner_lane in lane_ids else None
        if not matched_lane:
            continue
        items.append(
            {
                "title": project.get("name") or project.get("project_id") or matched_lane,
                "kind": "event_opportunity",
                "source": "overwatch-canonical",
                "canonical_status": "canonical_export",
                "blocked_reason": None,
                "primary_project": project.get("project_id") or matched_lane,
                "support_lanes": [],
                "detail": project.get("notes") or direction.get("direction") or "Canonical opportunity lane export from Overwatch.",
                "next_fix": None,
                "created": None,
                "path": ACTIVE_PROJECTS_PATH.as_posix(),
                "lane_id": matched_lane,
            }
        )

    for resource in registry.get("resources", []):
        resource_id = resource.get("resource_id")
        owner = resource.get("owner")
        family = resource.get("resource_family")
        matched_lane = resource_id if resource_id in lane_ids else owner if owner in lane_ids else family if family in lane_ids else None
        if not matched_lane:
            continue
        items.append(
            {
                "title": resource.get("display_name") or resource_id or matched_lane,
                "kind": "event_opportunity",
                "source": "overwatch-canonical",
                "canonical_status": "canonical_export",
                "blocked_reason": None,
                "primary_project": owner or matched_lane,
                "support_lanes": [],
                "detail": resource.get("description")
                or resource.get("scope")
                or f"Canonical Overwatch resource export for `{matched_lane}`.",
                "next_fix": None,
                "created": None,
                "path": RESOURCE_REGISTRY_PATH.as_posix(),
                "lane_id": matched_lane,
            }
        )

    deduped: dict[tuple[str, str], dict[str, Any]] = {}
    for item in items:
        key = (item["source"], item.get("lane_id") or item["title"])
        deduped[key] = item
    return list(deduped.values())


def build_opportunity_feed(planning: dict[str, Any]) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = _canonical_opportunity_items()
    for todo in planning.get("pending_todos", []):
        text = todo.get("body", "")
        lowered = f"{todo.get('title', '')} {todo.get('summary', '')}".lower()
        if not any(token in lowered for token in ("event", "hackathon", "demo", "track", "opportunity")):
            continue
        primary = (
            _extract_backtick_value(r"primary target repo:\s*`([^`]+)`", text)
            or _extract_backtick_value(r"intended project lane\s*\(`([^`]+)`\)", text)
            or _extract_backtick_value(r"target lane:\s*`([^`]+)`", text)
        )
        support = _extract_backtick_values(r"support lane[s]?:\s*`([^`]+)`", text)
        support += _extract_backtick_values(r"optional support:\s*`([^`]+)`", text)
        support += _extract_backtick_values(r"workflow/governance support:\s*`([^`]+)`", text)
        support += _extract_backtick_values(r"optional workflow/governance support:\s*`([^`]+)`", text)
        items.append(
            {
                "title": todo["title"],
                "kind": "event_opportunity",
                "source": "watchtower-derived",
                "canonical_status": "pending_overwatch_capture",
                "blocked_reason": "OVERWATCH_OPPORTUNITY_RECORD_MISSING",
                "primary_project": primary or "watchtower",
                "support_lanes": list(dict.fromkeys(support)),
                "detail": todo.get("summary"),
                "next_fix": "Create or refresh the canonical opportunity record in Overwatch when writable access is available.",
                "created": todo.get("created"),
                "path": todo.get("path"),
            }
        )
    return items


def _lane_export_exists(lane_id: str) -> bool:
    register = load_active_project_register()
    registry = load_resource_registry()
    for project in register.get("projects", []):
        if project.get("project_id") == lane_id:
            return True
        direction = project.get("direction_contract", {})
        if direction.get("owner_lane") == lane_id:
            return True
    for resource in registry.get("resources", []):
        if resource.get("resource_id") == lane_id:
            return True
        if resource.get("owner") == lane_id:
            return True
        if resource.get("resource_family") == lane_id:
            return True
    return False


def build_opportunity_lanes() -> list[dict[str, Any]]:
    lanes: list[dict[str, Any]] = []
    canonical_items = _canonical_opportunity_items()
    counts_by_lane: dict[str, int] = {}
    for item in canonical_items:
        lane_id = item.get("lane_id")
        if lane_id:
            counts_by_lane[lane_id] = counts_by_lane.get(lane_id, 0) + 1

    for lane_id in _opportunity_lane_ids():
        exported = _lane_export_exists(lane_id)
        lanes.append(
            {
                "lane_id": lane_id,
                "owner": "Overwatch",
                "watchtower_role": "surface_when_exported",
                "status": "ready" if exported else "blocked",
                "blocked_reason": None if exported else "OVERWATCH_LANE_NOT_EXPORTED",
                "detail": (
                    f"Canonical opportunity lane `{lane_id}` is exported by Overwatch."
                    if exported
                    else f"Watchtower cannot find a canonical Overwatch export for `{lane_id}` yet."
                ),
                "next_fix": (
                    None
                    if exported
                    else f"Expose `{lane_id}` through an Overwatch canonical project or resource export."
                ),
                "source": "overwatch-canonical-export",
                "item_count": counts_by_lane.get(lane_id, 0),
            }
        )
    return lanes


def build_operator_readiness(
    operator_contract: dict[str, Any],
    opportunity_lanes: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    lanes_by_id = {lane["lane_id"]: lane for lane in operator_contract.get("lanes", [])}
    items = []
    helper = lanes_by_id.get("ollarma-bounded-helper")
    workflow = lanes_by_id.get("ollarma-deterministic-workflow")
    provenance = lanes_by_id.get("seedgraph-provenance-identity")
    if helper:
        items.append(
            {
                "name": "Local helper",
                "owner": helper["owner"],
                "status": helper["status"],
                "detail": helper["detail"],
                "next_fix": helper.get("next_fix"),
            }
        )
    if workflow:
        items.append(
            {
                "name": "Approved orchestration",
                "owner": workflow["owner"],
                "status": workflow["status"],
                "detail": workflow["detail"],
                "next_fix": workflow.get("next_fix"),
            }
        )
    if provenance:
        items.append(
            {
                "name": "Provenance readiness",
                "owner": provenance["owner"],
                "status": provenance["status"],
                "detail": provenance["detail"],
                "next_fix": provenance.get("next_fix"),
            }
        )
    missing_opportunity_exports = [lane for lane in opportunity_lanes if lane["status"] != "ready"]
    if missing_opportunity_exports:
        items.append(
            {
                "name": "Opportunity lanes",
                "owner": "Overwatch",
                "status": "blocked",
                "detail": "Requested canonical opportunity lanes are not exported yet; Watchtower must use derived local capture only.",
                "next_fix": "Expose the requested opportunity lanes from Overwatch when ready.",
            }
        )
    return items


def build_operator_queue(
    tasks: list[dict[str, Any]],
    active_blockers: list[dict[str, Any]],
    opportunities: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    queue: list[dict[str, Any]] = []
    for blocker in active_blockers:
        queue.append(
            {
                "kind": "blocker",
                "title": blocker["scope"],
                "owner": blocker["owner"],
                "lane": blocker["lane_id"],
                "status": blocker["status"],
                "detail": blocker.get("detail"),
                "source": "operator-contract",
                "next_action": blocker.get("next_fix"),
            }
        )
    for opportunity in opportunities:
        queue.append(
            {
                "kind": "opportunity",
                "title": opportunity["title"],
                "owner": opportunity["primary_project"],
                "lane": "operator-opportunity",
                "status": opportunity["canonical_status"],
                "detail": opportunity.get("detail"),
                "source": opportunity["source"],
                "next_action": opportunity.get("next_fix"),
            }
        )
    for task in tasks:
        queue.append(
            {
                "kind": "task",
                "title": task["title"],
                "owner": task["owner"],
                "lane": task.get("lane"),
                "status": "active",
                "detail": task.get("detail"),
                "source": task.get("source"),
                "next_action": None,
            }
        )
    return queue[:18]


def build_approved_actions(operator_contract: dict[str, Any]) -> list[dict[str, Any]]:
    lanes_by_id = {lane["lane_id"]: lane for lane in operator_contract.get("lanes", [])}
    helper = lanes_by_id.get("ollarma-bounded-helper", {})
    workflow = lanes_by_id.get("ollarma-deterministic-workflow", {})
    canonical_write = lanes_by_id.get("overwatch-canonical-write", {})
    return [
        {
            "name": "Ask helper",
            "endpoint": "POST /api/chat",
            "owner": "Watchtower + Ollarma",
            "status": helper.get("status", "unknown"),
            "detail": "Send bounded local operator questions through Ollarma first and preserve exact fallback reasons.",
        },
        {
            "name": "Preview route",
            "endpoint": "POST /api/query",
            "owner": "Watchtower + Ollarma",
            "status": helper.get("status", "unknown"),
            "detail": "Route project-scoped questions and show whether local help is sufficient or fallback is required.",
        },
        {
            "name": "Trigger approved manifest step",
            "endpoint": "POST /api/orchestrate",
            "owner": "Watchtower + Ollarma",
            "status": workflow.get("status", "unknown"),
            "detail": "Only manifest-bounded orchestration is allowed; Watchtower stays out of unrestricted execution.",
        },
        {
            "name": "Write canonical truth",
            "endpoint": "not exposed",
            "owner": "Overwatch",
            "status": canonical_write.get("status", "review_gated"),
            "detail": "Canonical writes remain outside Watchtower and require an approved Overwatch or review lane.",
        },
    ]