the-custodian/state-hub/custodian_cli.py

#!/usr/bin/env python3
"""
custodian — CLI for the Custodian State Hub.

Usage:
  custodian register-project [--domain DOMAIN] [--path PATH]

  Run from inside the project directory you want to connect.
  --domain defaults to auto-detection from the project charter.
  --path defaults to current working directory.
"""
from __future__ import annotations

import argparse
import json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.request
from pathlib import Path

STATE_HUB_DIR = Path(__file__).resolve().parent
API_BASE = os.environ.get("API_BASE", "http://127.0.0.1:8000")
TEMPLATE = STATE_HUB_DIR / "scripts" / "project_claude_md.template"
PATCH_CWD = STATE_HUB_DIR / "scripts" / "patch_mcp_cwd.py"

VALID_DOMAINS = [
    "custodian", "railiance", "markitect",
    "coulomb_social", "personhood", "foerster_capabilities",
]


# ── Helpers ────────────────────────────────────────────────────────────────────

def _api_get(path: str) -> object:
    url = API_BASE.rstrip("/") + path
    try:
        with urllib.request.urlopen(url, timeout=10) as r:
            return json.loads(r.read())
    except urllib.error.URLError as e:
        print(f"ERROR: Cannot reach API at {API_BASE}: {e}")
        print(f"  Start it: cd {STATE_HUB_DIR} && make api")
        sys.exit(1)


def _api_post(path: str, body: dict) -> object:
    url = API_BASE.rstrip("/") + path
    data = json.dumps({k: v for k, v in body.items() if v is not None}).encode()
    req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=10) as r:
        return json.loads(r.read())


def _detect_domain(project_path: Path) -> str | None:
    """Try to read domain from project charter frontmatter."""
    for charter in project_path.rglob("project_charter_v*.md"):
        text = charter.read_text()
        m = re.search(r"^domain:\s*(\S+)", text, re.MULTILINE)
        if m:
            return m.group(1).strip('"\'')
    return None


def _check_mcp() -> bool:
    claude_json = Path.home() / ".claude.json"
    if not claude_json.exists():
        return False
    config = json.loads(claude_json.read_text())
    return "state-hub" in config.get("mcpServers", {})


# ── Subcommands ────────────────────────────────────────────────────────────────

def cmd_register(args: argparse.Namespace) -> None:
    project_path = Path(args.path).resolve()
    if not project_path.is_dir():
        print(f"ERROR: {project_path} is not a directory.")
        sys.exit(1)

    project_name = project_path.name

    # ── Step 1: API health ─────────────────────────────────────────────────────
    print(f"==> Checking API at {API_BASE} ...")
    _api_get("/state/health")
    print("    API OK")

    # ── Step 2: Domain ─────────────────────────────────────────────────────────
    domain = args.domain
    if not domain:
        print("==> Auto-detecting domain from project charter ...")
        domain = _detect_domain(project_path)
        if domain:
            print(f"    Detected: {domain}")
        else:
            print(f"ERROR: Could not auto-detect domain. Pass --domain explicitly.")
            print(f"  Valid: {', '.join(VALID_DOMAINS)}")
            sys.exit(1)

    if domain not in VALID_DOMAINS:
        print(f"ERROR: Unknown domain '{domain}'. Valid: {', '.join(VALID_DOMAINS)}")
        sys.exit(1)

    # ── Step 3: Topic ID lookup ────────────────────────────────────────────────
    print(f"==> Looking up topic for domain '{domain}' ...")
    topics = _api_get("/topics/?status=active")
    match = next((t for t in topics if t.get("domain") == domain), None)
    if not match:
        print(f"ERROR: No active topic found for domain '{domain}'.")
        sys.exit(1)
    topic_id = match["id"]
    print(f"    topic_id: {topic_id}")

    # ── Step 4: MCP check ──────────────────────────────────────────────────────
    print("==> Checking MCP server registration ...")
    if _check_mcp():
        print("    MCP OK")
    else:
        print("WARNING: 'state-hub' not in ~/.claude.json.")
        print(f"  See ~/.claude/CLAUDE.md → MCP Server Registration section.")

    # ── Step 5: CLAUDE.md ──────────────────────────────────────────────────────
    claude_md = project_path / "CLAUDE.md"
    if claude_md.exists():
        print(f"==> CLAUDE.md already exists at {claude_md} — skipping.")
    else:
        print(f"==> Writing CLAUDE.md to {claude_md} ...")
        content = TEMPLATE.read_text()
        content = content.replace("{PROJECT_NAME}", project_name)
        content = content.replace("{DOMAIN}", domain)
        content = content.replace("{TOPIC_ID}", topic_id)
        claude_md.write_text(content)
        print("    Written.")

    # ── Step 6: Progress event ─────────────────────────────────────────────────
    print("==> Recording registration event ...")
    try:
        _api_post("/progress/", {
            "topic_id": topic_id,
            "event_type": "milestone",
            "summary": f"Project registered with State Hub: {project_name} ({domain})",
            "author": "custodian",
            "detail": {
                "project_path": str(project_path),
                "claude_md": str(claude_md),
                "domain": domain,
            },
        })
        print("    Event recorded.")
    except Exception as e:
        print(f"    WARNING: Could not record progress event: {e}")

    print()
    print("Registration complete!")
    print(f"  Project:   {project_name}")
    print(f"  Domain:    {domain}")
    print(f"  Topic ID:  {topic_id}")
    print(f"  CLAUDE.md: {claude_md}")
    print()
    print("Next: restart Claude Code for the MCP server to be active in this project.")


def cmd_status(_args: argparse.Namespace) -> None:
    """Quick status: API health + summary totals."""
    health = _api_get("/state/health")
    print(f"API: {health.get('status', '?')}  DB: {health.get('db', '?')}")
    summary = _api_get("/state/summary")
    t = summary["totals"]
    print(f"Topics: {t['topics']['active']} active")
    print(f"Workstreams: {t['workstreams']['active']} active, {t['workstreams']['blocked']} blocked")
    print(f"Tasks: {t['tasks']['in_progress']} in-progress, {t['tasks']['todo']} todo, {t['tasks']['blocked']} blocked")
    print(f"Decisions: {t['decisions']['open']} open, {t['decisions']['escalated']} escalated")
    blocking = summary.get("blocking_decisions", [])
    if blocking:
        print(f"\nBlocking decisions ({len(blocking)}):")
        for d in blocking:
            deadline = d.get("deadline") or "no deadline"
            print(f"  [{deadline}] {d['title']}")


# ── Entry point ────────────────────────────────────────────────────────────────

def main() -> None:
    parser = argparse.ArgumentParser(
        prog="custodian",
        description="Custodian State Hub CLI",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    # register-project
    reg = sub.add_parser("register-project", help="Register a project with the State Hub")
    reg.add_argument(
        "--domain",
        choices=VALID_DOMAINS,
        default=None,
        help="Project domain (auto-detected from charter if omitted)",
    )
    reg.add_argument(
        "--path",
        default=os.getcwd(),
        help="Project directory (defaults to current directory)",
    )

    # status
    sub.add_parser("status", help="Show State Hub health and summary totals")

    args = parser.parse_args()

    if args.command == "register-project":
        cmd_register(args)
    elif args.command == "status":
        cmd_status(args)


if __name__ == "__main__":
    main()