tegwick
6b82215ae7
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a049a4-ee9f-78e1-9d66-2cb0f9bea3e3
962 lines
40 KiB
Python
962 lines
40 KiB
Python
#!/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.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
from repository_rename_cli import configure_repo_commands
|
|
from statehub_register import run_register as run_statehub_register
|
|
|
|
STATE_HUB_DIR = Path(__file__).resolve().parent
|
|
API_BASE = os.environ.get("API_BASE", "http://127.0.0.1:8000")
|
|
_LEGACY_METER_HEADERS = {"X-StateHub-Component": "state-hub.custodian-cli"}
|
|
TEMPLATE = STATE_HUB_DIR / "scripts" / "project_claude_md.template"
|
|
PATCH_CWD = STATE_HUB_DIR / "scripts" / "patch_mcp_cwd.py"
|
|
|
|
_SUGGESTION_PREAMBLE = """\
|
|
<!--
|
|
CUSTODIAN SUGGESTION — generated by `custodian register-project`
|
|
|
|
This file was written by the Custodian State Hub registration workflow.
|
|
It contains the standard State Hub integration block for this repository.
|
|
|
|
A Claude Code agent in this repo should:
|
|
1. Read this file and the existing CLAUDE.md side by side
|
|
2. Integrate the sections below into CLAUDE.md — preserve all existing
|
|
project-specific conventions; add or update the hub integration block
|
|
3. Delete this file once integration is complete and commit the result
|
|
|
|
Do NOT add this file to .gitignore. It is a temporary artefact that signals
|
|
pending integration work to the repo agent.
|
|
-->
|
|
|
|
"""
|
|
|
|
_ONBOARDING_TASKS = [
|
|
(
|
|
"Integrate CLAUDE.custodian.md → CLAUDE.md",
|
|
"high",
|
|
"A CLAUDE.custodian.md suggestion file was written by the custodian registration workflow. "
|
|
"Read both files, merge the hub integration block into the existing CLAUDE.md "
|
|
"(preserve all project-specific conventions), then delete CLAUDE.custodian.md and commit.",
|
|
),
|
|
(
|
|
"Write first workplan and initialise workplans/",
|
|
"high",
|
|
"Create a workplans/ directory and write the first workplan file following ADR-001 "
|
|
"(~/the-custodian/canon/architecture/adr-001-workplans-as-repo-artefacts.md). "
|
|
"Cover the repo's primary near-term work strand. Register the workstream in the state hub via MCP.",
|
|
),
|
|
(
|
|
"Ingest SBOM",
|
|
"medium",
|
|
# path substituted at call time
|
|
"",
|
|
),
|
|
(
|
|
"Register known EPs and TDs",
|
|
"low",
|
|
"Catalogue any known extension points (future enhancement hooks) and technical debt items "
|
|
"using the register_extension_point() and register_technical_debt() MCP tools.",
|
|
),
|
|
]
|
|
|
|
# ── Helpers ────────────────────────────────────────────────────────────────────
|
|
|
|
def _api_get(path: str) -> object:
|
|
url = API_BASE.rstrip("/") + path
|
|
try:
|
|
req = urllib.request.Request(url, headers=_LEGACY_METER_HEADERS)
|
|
with urllib.request.urlopen(req, 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", **_LEGACY_METER_HEADERS},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=10) as r:
|
|
return json.loads(r.read())
|
|
|
|
|
|
def _api_patch(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", **_LEGACY_METER_HEADERS},
|
|
method="PATCH",
|
|
)
|
|
with urllib.request.urlopen(req, timeout=10) as r:
|
|
return json.loads(r.read())
|
|
|
|
|
|
def _find_repo_by_slug(repo_slug: str) -> dict | None:
|
|
repos = _api_get("/repos/")
|
|
return next((r for r in repos if r.get("slug") == repo_slug), None)
|
|
|
|
|
|
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:
|
|
from scripts.mcp_registration import load_claude_json, mcp_server_registered
|
|
|
|
return mcp_server_registered(load_claude_json())
|
|
|
|
|
|
# ── Subcommands ────────────────────────────────────────────────────────────────
|
|
|
|
def cmd_register(args: argparse.Namespace) -> None:
|
|
"""Register a project/repo with the State Hub and generate onboarding tasks."""
|
|
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
|
|
repo_slug = re.sub(r"-+", "-", re.sub(r"[^a-z0-9]", "-", project_name.lower())).strip("-")
|
|
|
|
# ── Step 1: API health ─────────────────────────────────────────────────────
|
|
print(f"==> Checking API at {API_BASE} ...")
|
|
_api_get("/state/health")
|
|
print(" API OK")
|
|
|
|
# ── Step 2: Domain ─────────────────────────────────────────────────────────
|
|
domain = args.domain
|
|
valid_domains = [d["slug"] for d in _api_get("/domains/?status=active")]
|
|
if not domain:
|
|
print("==> Auto-detecting domain from project charter ...")
|
|
domain = _detect_domain(project_path)
|
|
if domain:
|
|
print(f" Detected: {domain}")
|
|
else:
|
|
print("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 (auto-create if new domain) ───────────────────
|
|
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_slug") == domain), None)
|
|
if not match:
|
|
print(f" No topic found — creating one for domain '{domain}' ...")
|
|
t_slug = re.sub(r"[^a-z0-9]+", "-", domain.lower()).strip("-")
|
|
try:
|
|
match = _api_post("/topics/", {
|
|
"slug": t_slug,
|
|
"title": project_name,
|
|
"domain": domain,
|
|
"status": "active",
|
|
})
|
|
print(f" Topic created: {match['title']} ({match['id']})")
|
|
except Exception as e:
|
|
print(f"ERROR: Could not create topic for domain '{domain}': {e}")
|
|
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: 'dev-hub' (or legacy 'state-hub') not in ~/.claude.json.")
|
|
print(" Run: python scripts/migrate_mcp_config.py # if upgrading legacy config")
|
|
print(" See ~/.claude/CLAUDE.md → MCP Server Registration section.")
|
|
|
|
# ── Step 5: Write CLAUDE.custodian.md ─────────────────────────────────────
|
|
suggestion_file = project_path / "CLAUDE.custodian.md"
|
|
print(f"==> Writing custodian suggestion to {suggestion_file} ...")
|
|
content = (
|
|
_SUGGESTION_PREAMBLE
|
|
+ TEMPLATE.read_text()
|
|
.replace("{PROJECT_NAME}", project_name)
|
|
.replace("{DOMAIN}", domain)
|
|
.replace("{TOPIC_ID}", topic_id)
|
|
.replace("{REPO_SLUG}", repo_slug)
|
|
)
|
|
suggestion_file.write_text(content)
|
|
print(" Written. The repo agent integrates it into CLAUDE.md then deletes it.")
|
|
|
|
# ── Step 6: Register repo ─────────────────────────────────────────────────
|
|
print(f"==> Registering repo '{repo_slug}' under domain '{domain}' ...")
|
|
repo = None
|
|
try:
|
|
repo = _api_post("/repos/", {
|
|
"domain_slug": domain,
|
|
"slug": repo_slug,
|
|
"name": project_name,
|
|
"local_path": str(project_path),
|
|
})
|
|
print(" Registered.")
|
|
except urllib.error.HTTPError as e:
|
|
if e.code != 409:
|
|
print(f" NOTE: {e} — repo registration failed, continuing.")
|
|
else:
|
|
print(" Repo already registered, reusing existing record.")
|
|
repo = _find_repo_by_slug(repo_slug)
|
|
except Exception as e:
|
|
print(f" NOTE: {e} — repo may already be registered, continuing.")
|
|
repo = _find_repo_by_slug(repo_slug)
|
|
|
|
repo_id = repo.get("id") if isinstance(repo, dict) else None
|
|
if repo_id:
|
|
print(f" repo_id: {repo_id}")
|
|
else:
|
|
print(" WARNING: Could not resolve repo_id; onboarding workstream will remain domain-level.")
|
|
|
|
# ── Step 7: Onboarding workstream + tasks ─────────────────────────────────
|
|
ws_slug = f"repo-integration-{repo_slug}"
|
|
print(f"==> Creating onboarding workstream '{ws_slug}' ...")
|
|
# Check if it already exists
|
|
existing_ws = next(
|
|
(w for w in _api_get("/workplans/") if w.get("slug") == ws_slug and w.get("status") == "active"),
|
|
None,
|
|
)
|
|
if existing_ws:
|
|
print(" Onboarding workstream already exists — skipping task creation.")
|
|
if repo_id and not existing_ws.get("repo_id"):
|
|
existing_owner = existing_ws.get("owner")
|
|
_api_patch(f"/workplans/{existing_ws['id']}/", {
|
|
"repo_id": repo_id,
|
|
"owner": repo_slug if existing_owner in (None, domain) else existing_owner,
|
|
})
|
|
print(" Attached existing onboarding workstream to repo.")
|
|
elif repo_id and existing_ws.get("repo_id") != repo_id:
|
|
print(
|
|
" WARNING: Existing onboarding workstream is attached to a different repo_id; "
|
|
"leaving it unchanged."
|
|
)
|
|
else:
|
|
try:
|
|
ws = _api_post("/workplans/", {
|
|
"topic_id": topic_id,
|
|
"title": f"Repo Integration: {repo_slug}",
|
|
"slug": ws_slug,
|
|
"description": (
|
|
f"Bootstrapping workstream created by the custodian during registration of "
|
|
f"'{repo_slug}'. Contains onboarding tasks for the repo agent to execute. "
|
|
f"ADR-001 exception: this workstream is DB-first because the repo has no "
|
|
f"workplans/ directory yet. Task T2 produces the first workplan file."
|
|
),
|
|
"owner": repo_slug,
|
|
"status": "active",
|
|
"repo_id": repo_id,
|
|
})
|
|
ws_id = ws["id"]
|
|
sbom_desc = (
|
|
f"Capture the repo's dependency snapshot. From state-hub dir: "
|
|
f"make ingest-sbom REPO={repo_slug} SCAN=1 REPO_PATH={project_path}"
|
|
)
|
|
tasks = [
|
|
(_ONBOARDING_TASKS[0][0], _ONBOARDING_TASKS[0][1], _ONBOARDING_TASKS[0][2]),
|
|
(_ONBOARDING_TASKS[1][0], _ONBOARDING_TASKS[1][1], _ONBOARDING_TASKS[1][2]),
|
|
(_ONBOARDING_TASKS[2][0], _ONBOARDING_TASKS[2][1], sbom_desc),
|
|
(_ONBOARDING_TASKS[3][0], _ONBOARDING_TASKS[3][1], _ONBOARDING_TASKS[3][2]),
|
|
]
|
|
for title, priority, description in tasks:
|
|
_api_post("/tasks/", {
|
|
"workplan_id": ws_id,
|
|
"title": title,
|
|
"priority": priority,
|
|
"description": description,
|
|
})
|
|
print(f" Created with {len(tasks)} onboarding tasks.")
|
|
print(f" The {domain} repo agent will see these at next session start.")
|
|
except Exception as e:
|
|
print(f" WARNING: Could not create onboarding tasks: {e}")
|
|
ws_id = None
|
|
|
|
# ── Step 8: Progress event ─────────────────────────────────────────────────
|
|
print("==> Recording registration event ...")
|
|
try:
|
|
_api_post("/progress/", {
|
|
"topic_id": topic_id,
|
|
"event_type": "milestone",
|
|
"summary": f"Repo registered: {project_name} ({domain}) — onboarding tasks created",
|
|
"author": "custodian",
|
|
"detail": {
|
|
"project_path": str(project_path),
|
|
"suggestion_file": str(suggestion_file),
|
|
"repo_slug": repo_slug,
|
|
"domain": domain,
|
|
"onboarding_workstream_slug": ws_slug,
|
|
},
|
|
})
|
|
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" Repo slug: {repo_slug}")
|
|
print(f" Topic ID: {topic_id}")
|
|
print(f" Suggestion: {suggestion_file}")
|
|
print()
|
|
print("Next: open the repo in Claude Code.")
|
|
print(" The repo agent will pick up 4 onboarding tasks and integrate autonomously.")
|
|
|
|
|
|
def cmd_ingest_sbom(args: argparse.Namespace) -> None:
|
|
"""Ingest SBOM for the current (or specified) repo. Auto-detects slug from registration."""
|
|
project_path = Path(args.path).resolve()
|
|
|
|
_api_get("/state/health")
|
|
|
|
# Resolve repo slug: explicit override, or look up by local_path
|
|
repo_slug = args.slug
|
|
if not repo_slug:
|
|
repos = _api_get("/repos/")
|
|
repo = next((r for r in repos if r.get("local_path") == str(project_path)), None)
|
|
if not repo:
|
|
print(f"ERROR: No registered repo found for path '{project_path}'.")
|
|
print(" Register first: custodian register-project --domain <slug>")
|
|
print(" Or pass --slug explicitly.")
|
|
sys.exit(1)
|
|
repo_slug = repo["slug"]
|
|
|
|
print(f"==> Ingesting SBOM for '{repo_slug}' from {project_path} ...")
|
|
|
|
python = STATE_HUB_DIR / ".venv" / "bin" / "python"
|
|
ingest_script = STATE_HUB_DIR / "scripts" / "ingest_sbom.py"
|
|
|
|
if not python.exists():
|
|
print(f"ERROR: .venv not found at {STATE_HUB_DIR}. Run 'make install' in the state-hub directory.")
|
|
sys.exit(1)
|
|
|
|
cmd = [str(python), str(ingest_script), "--repo", repo_slug, "--scan", "--repo-path", str(project_path)]
|
|
if args.dry_run:
|
|
cmd.append("--dry-run")
|
|
|
|
result = subprocess.run(cmd)
|
|
sys.exit(result.returncode)
|
|
|
|
|
|
def cmd_fix_consistency(args: argparse.Namespace) -> None:
|
|
"""Run ADR-001 consistency repair from any registered repo checkout."""
|
|
checker = STATE_HUB_DIR / "scripts" / "consistency_check.py"
|
|
if not checker.exists():
|
|
print(f"ERROR: consistency checker not found at {checker}")
|
|
print(" Run this command from an editable state-hub install or the state-hub repo.")
|
|
sys.exit(1)
|
|
|
|
if args.remote and not (args.repo or args.all):
|
|
print("ERROR: --remote requires --repo or --all.")
|
|
print(" From a local checkout, run: statehub fix-consistency")
|
|
print(" For pull-before-fix, run: statehub fix-consistency --repo <slug> --remote")
|
|
sys.exit(1)
|
|
|
|
cmd = [sys.executable, str(checker)]
|
|
if args.all:
|
|
cmd.append("--all")
|
|
elif args.repo:
|
|
cmd.extend(["--repo", args.repo])
|
|
if args.repo_path:
|
|
cmd.extend(["--repo-path", str(Path(args.repo_path).expanduser().resolve())])
|
|
else:
|
|
cmd.append("--here")
|
|
if args.path:
|
|
cmd.append(str(Path(args.path).expanduser().resolve()))
|
|
|
|
cmd.append("--fix")
|
|
if args.remote:
|
|
cmd.append("--remote")
|
|
if args.no_writeback:
|
|
cmd.append("--no-writeback")
|
|
if getattr(args, "bootstrap_empty_projection", False):
|
|
cmd.append("--bootstrap-empty-projection")
|
|
if args.archive_closed:
|
|
cmd.append("--archive-closed")
|
|
if args.archive_workplan:
|
|
cmd.extend(["--archive-workplan", args.archive_workplan])
|
|
if args.archive_date:
|
|
cmd.extend(["--archive-date", args.archive_date])
|
|
if args.api_base:
|
|
cmd.extend(["--api-base", args.api_base])
|
|
if args.as_json:
|
|
cmd.append("--json")
|
|
if args.max_seconds is not None:
|
|
cmd.extend(["--max-seconds", str(args.max_seconds)])
|
|
|
|
result = subprocess.run(cmd)
|
|
exit_code = result.returncode
|
|
if exit_code == 2 and not args.strict_warnings:
|
|
exit_code = 0
|
|
sys.exit(exit_code)
|
|
|
|
|
|
def cmd_quality_debt(args: argparse.Namespace) -> None:
|
|
"""List DoX quality debt (STATE-WP-0077) — ready without DoR-Ok, etc."""
|
|
script = STATE_HUB_DIR / "scripts" / "quality_debt.py"
|
|
if not script.exists():
|
|
print(f"ERROR: quality_debt.py not found at {script}")
|
|
sys.exit(1)
|
|
cmd = [sys.executable, str(script)]
|
|
if args.repo_path:
|
|
cmd.extend(["--repo-path", str(Path(args.repo_path).expanduser().resolve())])
|
|
else:
|
|
cmd.append("--here")
|
|
if args.api_base:
|
|
cmd.extend(["--api-base", args.api_base])
|
|
if args.no_hub:
|
|
cmd.append("--no-hub")
|
|
if args.as_json:
|
|
cmd.append("--json")
|
|
if args.strict:
|
|
cmd.append("--strict")
|
|
result = subprocess.run(cmd)
|
|
sys.exit(result.returncode)
|
|
|
|
|
|
def cmd_promote_intake(args: argparse.Namespace) -> None:
|
|
"""Promote a routed intake into a workplan, task, decision, or engagement."""
|
|
script = STATE_HUB_DIR / "scripts" / "promote_intake.py"
|
|
if not script.exists():
|
|
print(f"ERROR: promote_intake.py not found at {script}")
|
|
sys.exit(1)
|
|
|
|
cmd = [
|
|
sys.executable, str(script), args.intake_id,
|
|
"--to", args.to,
|
|
"--repo-path", str(Path(args.repo_path).expanduser().resolve()),
|
|
"--repo-slug", args.repo_slug,
|
|
"--domain", args.domain,
|
|
"--api-base", args.api_base,
|
|
]
|
|
if args.target_file:
|
|
cmd.extend(["--target-file", args.target_file])
|
|
if args.workplan_file:
|
|
cmd.extend(["--workplan-file", args.workplan_file])
|
|
|
|
result = subprocess.run(cmd)
|
|
sys.exit(result.returncode)
|
|
|
|
|
|
def cmd_create_workstream(args: argparse.Namespace) -> None:
|
|
"""Create a workstream under a domain's topic."""
|
|
_api_get("/state/health")
|
|
|
|
# Resolve topic_id from domain
|
|
topics = _api_get("/topics/?status=active")
|
|
match = next((t for t in topics if t.get("domain_slug") == args.domain), None)
|
|
if not match:
|
|
print(f"ERROR: No active topic for domain '{args.domain}'.")
|
|
sys.exit(1)
|
|
topic_id = match["id"]
|
|
|
|
slug = args.slug or re.sub(r"[^a-z0-9]+", "-", args.title.lower()).strip("-")
|
|
|
|
ws = _api_post("/workplans/", {
|
|
"topic_id": topic_id,
|
|
"title": args.title,
|
|
"slug": slug,
|
|
"description": args.description,
|
|
"owner": args.owner,
|
|
"status": "active",
|
|
})
|
|
_api_post("/progress/", {
|
|
"topic_id": topic_id,
|
|
"workplan_id": ws["id"],
|
|
"event_type": "workplan_created",
|
|
"summary": f"Workplan created: {args.title}",
|
|
"author": "custodian",
|
|
"detail": {"owner": args.owner, "slug": slug},
|
|
})
|
|
print(f"Created workstream: {ws['title']}")
|
|
print(f" id: {ws['id']}")
|
|
print(f" slug: {ws['slug']}")
|
|
print(f" domain: {args.domain}")
|
|
print(f" owner: {ws.get('owner') or '—'}")
|
|
|
|
|
|
def cmd_create_task(args: argparse.Namespace) -> None:
|
|
"""Create a task under a workstream (by ID or slug)."""
|
|
_api_get("/state/health")
|
|
|
|
# Resolve workstream: accept UUID or slug
|
|
workstream_id = args.workstream
|
|
if not _is_uuid(workstream_id):
|
|
wss = _api_get("/workplans/")
|
|
match = next((w for w in wss if w.get("slug") == workstream_id), None)
|
|
if not match:
|
|
print(f"ERROR: No workstream found with slug '{workstream_id}'.")
|
|
print(" Use 'custodian status' or check the dashboard for valid slugs.")
|
|
sys.exit(1)
|
|
workstream_id = match["id"]
|
|
|
|
task = _api_post("/tasks/", {
|
|
"workplan_id": workstream_id,
|
|
"title": args.title,
|
|
"priority": args.priority,
|
|
"description": args.description,
|
|
"assignee": args.assignee,
|
|
})
|
|
_api_post("/progress/", {
|
|
"workplan_id": workstream_id,
|
|
"task_id": task["id"],
|
|
"event_type": "task_created",
|
|
"summary": f"Task created: {args.title}",
|
|
"author": "custodian",
|
|
"detail": {"priority": args.priority},
|
|
})
|
|
print(f"Created task: {task['title']}")
|
|
print(f" id: {task['id']}")
|
|
print(f" priority: {task['priority']}")
|
|
print(f" status: {task['status']}")
|
|
|
|
|
|
def _is_uuid(s: str) -> bool:
|
|
import uuid as _uuid
|
|
try:
|
|
_uuid.UUID(s)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
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', '?')}")
|
|
# Which instance answered, not merely that one did.
|
|
role = health.get("instance_role") or "undeclared"
|
|
label = health.get("instance_label")
|
|
print(f"Instance: {role}{f' ({label})' if label else ''}")
|
|
summary = _api_get("/state/summary")
|
|
t = summary["totals"]
|
|
topics = t.get("topics", {})
|
|
workplans = t.get("workplans") or t.get("workstreams", {})
|
|
tasks = t.get("tasks", {})
|
|
decisions = t.get("decisions", {})
|
|
print(f"API base: {API_BASE}")
|
|
print(f"Topics: {topics.get('active', 0)} active")
|
|
print(f"Workplans: {workplans.get('active', 0)} active, {workplans.get('blocked', 0)} blocked")
|
|
# Task statuses are wait|todo|progress|done|cancel (see workplan-convention.md).
|
|
print(f"Tasks: {tasks.get('progress', 0)} in-progress, {tasks.get('todo', 0)} todo, {tasks.get('wait', 0)} waiting")
|
|
print(f"Decisions: {decisions.get('open', 0)} open, {decisions.get('escalated', 0)} 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']}")
|
|
|
|
|
|
def _load_json_file(path: str) -> dict:
|
|
target = Path(path)
|
|
try:
|
|
value = json.loads(target.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
print(f"ERROR: invalid JSON file {target}: {exc}")
|
|
sys.exit(2)
|
|
if not isinstance(value, dict):
|
|
print(f"ERROR: {target} must contain a JSON object")
|
|
sys.exit(2)
|
|
return value
|
|
|
|
|
|
def _git_source(path: str, repo_slug: str | None = None) -> tuple[Path, dict]:
|
|
target = Path(path).resolve()
|
|
try:
|
|
root = Path(subprocess.check_output(
|
|
["git", "-C", str(target.parent), "rev-parse", "--show-toplevel"],
|
|
text=True,
|
|
).strip())
|
|
revision = subprocess.check_output(
|
|
["git", "-C", str(root), "rev-parse", "HEAD"], text=True
|
|
).strip()
|
|
relative = target.relative_to(root).as_posix()
|
|
except (subprocess.CalledProcessError, ValueError) as exc:
|
|
print(f"ERROR: {target} must be inside a Git repository: {exc}")
|
|
sys.exit(2)
|
|
return target, {"repo": repo_slug or root.name, "path": relative, "revision": revision}
|
|
|
|
|
|
def cmd_review_project(args: argparse.Namespace) -> None:
|
|
"""Project a file-authoritative contract wrapper through the direct API."""
|
|
target, source = _git_source(args.file, args.source_repo)
|
|
contract = _load_json_file(str(target))
|
|
payload = contract if "contract" in contract and "source" in contract else {
|
|
"contract": contract,
|
|
"source": source,
|
|
"decision_id": args.decision_id,
|
|
"workplan_id": args.workplan_id,
|
|
"task_id": args.task_id,
|
|
"required_for_decision": args.required_for_decision,
|
|
}
|
|
print(json.dumps(
|
|
_api_post("/review-contracts/projections", payload),
|
|
indent=2,
|
|
))
|
|
|
|
|
|
def cmd_review_submit(args: argparse.Namespace) -> None:
|
|
"""Submit a file-authoritative receipt against an active contract."""
|
|
key = urllib.parse.quote(args.contract_key, safe="")
|
|
target, source = _git_source(args.file, args.source_repo)
|
|
receipt = _load_json_file(str(target))
|
|
if "source" not in receipt:
|
|
receipt["source"] = source
|
|
print(json.dumps(
|
|
_api_post(f"/review-contracts/{key}/receipts", receipt),
|
|
indent=2,
|
|
))
|
|
|
|
|
|
def cmd_review_status(args: argparse.Namespace) -> None:
|
|
"""Print the derived owner and gate matrix; never an execution authorization."""
|
|
key = urllib.parse.quote(args.contract_key, safe="")
|
|
print(json.dumps(_api_get(f"/review-contracts/{key}/aggregate"), indent=2))
|
|
|
|
|
|
|
|
def _outbox_store(args):
|
|
from api.edge.outbox import OutboxStore, default_outbox_path
|
|
|
|
return OutboxStore(args.outbox_path or default_outbox_path())
|
|
|
|
|
|
def cmd_outbox_status(args: argparse.Namespace) -> None:
|
|
store = _outbox_store(args)
|
|
print(json.dumps(store.summary(), indent=2))
|
|
|
|
|
|
def cmd_outbox_list(args: argparse.Namespace) -> None:
|
|
store = _outbox_store(args)
|
|
rows = store.export(status=args.status, limit=args.limit)
|
|
print(json.dumps(rows, indent=2))
|
|
|
|
|
|
def cmd_outbox_export(args: argparse.Namespace) -> None:
|
|
store = _outbox_store(args)
|
|
payload = store.export(status=args.status, limit=args.limit)
|
|
if args.output:
|
|
Path(args.output).write_text(json.dumps(payload, indent=2) + "\n")
|
|
print(f"Exported {len(payload)} envelope(s) to {args.output}")
|
|
else:
|
|
print(json.dumps(payload, indent=2))
|
|
|
|
|
|
def cmd_outbox_replay(args: argparse.Namespace) -> None:
|
|
import asyncio
|
|
from api.edge.relay import replay_pending
|
|
|
|
store = _outbox_store(args)
|
|
upstream = args.upstream_url or os.environ.get("STATEHUB_UPSTREAM_URL") or API_BASE
|
|
result = asyncio.run(replay_pending(store, upstream_url=upstream, limit=args.limit))
|
|
print(json.dumps(result, indent=2))
|
|
|
|
|
|
def cmd_outbox_retry(args: argparse.Namespace) -> None:
|
|
store = _outbox_store(args)
|
|
store.retry(args.envelope_id)
|
|
print(f"Queued {args.envelope_id} for retry")
|
|
|
|
|
|
def cmd_outbox_cancel(args: argparse.Namespace) -> None:
|
|
store = _outbox_store(args)
|
|
store.cancel(args.envelope_id)
|
|
print(f"Cancelled {args.envelope_id}")
|
|
|
|
|
|
def cmd_dev_up(args: argparse.Namespace) -> None:
|
|
"""Start files-first local dev hub (compose path via dev_hub_up.sh)."""
|
|
script = STATE_HUB_DIR / "scripts" / "dev_hub_up.sh"
|
|
if not script.exists():
|
|
print(f"ERROR: dev hub script not found at {script}")
|
|
sys.exit(1)
|
|
|
|
env = os.environ.copy()
|
|
if args.with_edge:
|
|
env["WITH_EDGE"] = "1"
|
|
if args.with_mcp:
|
|
env["WITH_MCP"] = "1"
|
|
if args.repo_root:
|
|
env["REPO_ROOT"] = str(Path(args.repo_root).expanduser().resolve())
|
|
|
|
cmd = ["bash", str(script), args.profile]
|
|
result = subprocess.run(cmd, env=env, cwd=STATE_HUB_DIR)
|
|
sys.exit(result.returncode)
|
|
|
|
|
|
# ── Entry point ────────────────────────────────────────────────────────────────
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
prog=Path(sys.argv[0]).name,
|
|
description="Custodian State Hub CLI",
|
|
)
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
configure_repo_commands(sub)
|
|
|
|
# register
|
|
statehub_reg = sub.add_parser(
|
|
"register",
|
|
help="Register the current repo with State Hub and prime it for Codex",
|
|
)
|
|
statehub_reg.add_argument("--path", default=os.getcwd(), help="Repo directory (defaults to cwd)")
|
|
statehub_reg.add_argument("--domain", default=None, help="State Hub domain slug")
|
|
statehub_reg.add_argument("--repo-slug", default=None, help="Repo slug (auto-detected if omitted)")
|
|
statehub_reg.add_argument("--wp-prefix", default=None, help="Workplan prefix, e.g. STATE-WP")
|
|
statehub_reg.add_argument("--description", default=None, help="One-sentence repo description")
|
|
statehub_reg.add_argument(
|
|
"--intent",
|
|
default=None,
|
|
help="Repo intent text to use when INTENT.md is absent and inference is insufficient",
|
|
)
|
|
statehub_reg.add_argument("--api-base", default=API_BASE, help="State Hub API base URL")
|
|
statehub_reg.add_argument(
|
|
"--llm-provider",
|
|
default=os.environ.get("STATEHUB_REGISTER_LLM_PROVIDER", "claude-code"),
|
|
help="llm-connect provider: claude-code, openrouter, openai, gemini, or mock",
|
|
)
|
|
statehub_reg.add_argument(
|
|
"--llm-model",
|
|
default=os.environ.get("STATEHUB_REGISTER_LLM_MODEL"),
|
|
help="Model name passed to llm-connect",
|
|
)
|
|
statehub_reg.add_argument(
|
|
"--llm-api-key",
|
|
default=os.environ.get("STATEHUB_REGISTER_LLM_API_KEY"),
|
|
help="API key for API-backed llm-connect providers",
|
|
)
|
|
statehub_reg.add_argument(
|
|
"--llm-timeout",
|
|
type=int,
|
|
default=int(os.environ.get("STATEHUB_REGISTER_LLM_TIMEOUT", "120")),
|
|
help="LLM timeout in seconds",
|
|
)
|
|
statehub_reg.add_argument("--no-llm", action="store_true", help="Skip LLM inference and use files/prompts")
|
|
statehub_reg.add_argument("--force", action="store_true", help="Overwrite generated repo files")
|
|
|
|
# register-project
|
|
reg = sub.add_parser("register-project", help="Register a project with the State Hub")
|
|
reg.add_argument(
|
|
"--domain",
|
|
default=None,
|
|
help="Project domain slug (auto-detected from charter if omitted)",
|
|
)
|
|
reg.add_argument(
|
|
"--path",
|
|
default=os.getcwd(),
|
|
help="Project directory (defaults to current directory)",
|
|
)
|
|
|
|
# ingest-sbom
|
|
ing = sub.add_parser("ingest-sbom", help="Ingest SBOM for the repo at the current directory")
|
|
ing.add_argument("--path", default=os.getcwd(), help="Repo directory (defaults to cwd)")
|
|
ing.add_argument("--slug", default=None, help="Repo slug (auto-detected from path if omitted)")
|
|
ing.add_argument("--dry-run", action="store_true", help="Parse lockfiles but do not submit to API")
|
|
|
|
# fix-consistency
|
|
fix = sub.add_parser(
|
|
"fix-consistency",
|
|
help="Reconcile workplan files with State Hub from the current repo",
|
|
)
|
|
target = fix.add_mutually_exclusive_group()
|
|
target.add_argument("--repo", default=None, help="Registered repo slug; defaults to inferring from --path")
|
|
target.add_argument("--all", action="store_true", help="Fix all registered repos with a visible path")
|
|
fix.add_argument("--path", default=os.getcwd(), help="Repo checkout to infer from (defaults to cwd)")
|
|
fix.add_argument("--repo-path", default=None, help="Override repo path when using --repo")
|
|
fix.add_argument("--remote", action="store_true", help="Pull before fixing; requires --repo or --all")
|
|
fix.add_argument("--max-seconds", type=int, default=None, help="Wall-clock budget for --remote --all")
|
|
fix.add_argument("--no-writeback", action="store_true", help="Disable DB-to-file status writeback")
|
|
fix.add_argument(
|
|
"--bootstrap-empty-projection",
|
|
action="store_true",
|
|
help="Rebuild file UUIDs only after proving the repo projection is empty",
|
|
)
|
|
fix.add_argument("--archive-closed", action="store_true", help="Archive closed root workplans after fixing")
|
|
fix.add_argument("--archive-workplan", default=None, help="Archive only the matching workplan id or filename")
|
|
fix.add_argument("--archive-date", default=None, help="YYMMDD archive prefix for --archive-closed")
|
|
fix.add_argument("--api-base", default=API_BASE, help="State Hub API base URL")
|
|
fix.add_argument("--json", action="store_true", dest="as_json", help="Output JSON from the checker")
|
|
fix.add_argument(
|
|
"--strict-warnings",
|
|
action="store_true",
|
|
help="Preserve checker exit code 2 for warnings-only runs",
|
|
)
|
|
|
|
# quality-debt (STATE-WP-0077)
|
|
qdebt = sub.add_parser(
|
|
"quality-debt",
|
|
help="List DoX quality debt: ready without DoR-Ok, finished without DoD-Ok, intakes without DoC-Ok",
|
|
)
|
|
qdebt.add_argument("--repo-path", default=None, help="Repo root (default: cwd)")
|
|
qdebt.add_argument("--api-base", default=API_BASE, help="State Hub API base URL")
|
|
qdebt.add_argument("--no-hub", action="store_true", help="Skip hub intake scan")
|
|
qdebt.add_argument("--json", action="store_true", dest="as_json")
|
|
qdebt.add_argument("--strict", action="store_true", help="Exit 1 if any debt found")
|
|
qdebt.set_defaults(func=cmd_quality_debt)
|
|
|
|
# promote-intake
|
|
promote = sub.add_parser(
|
|
"promote-intake",
|
|
help="Promote a routed intake into a workplan, task, decision, or engagement (CUST-WP-0061-T03)",
|
|
)
|
|
promote.add_argument("intake_id", help="UUID of the routed intake")
|
|
promote.add_argument("--to", required=True, choices=["workplan", "task", "decision", "engagement"])
|
|
promote.add_argument("--repo-path", default=os.getcwd(), help="Local checkout of the target repo")
|
|
promote.add_argument("--repo-slug", required=True, help="Registered repo slug")
|
|
promote.add_argument("--domain", required=True, help="Market domain slug, e.g. infotech")
|
|
promote.add_argument("--target-file", default=None, help="Repo-relative path (required for decision/engagement)")
|
|
promote.add_argument("--workplan-file", default=None, help="Repo-relative workplan file path (required for task)")
|
|
promote.add_argument("--api-base", default=API_BASE, help="State Hub API base URL")
|
|
|
|
# create-workstream
|
|
cws = sub.add_parser("create-workstream", help="Create a workstream under a domain topic")
|
|
cws.add_argument("--domain", required=True, help="Domain slug to create the workstream under")
|
|
cws.add_argument("--title", required=True, help="Workstream title")
|
|
cws.add_argument("--slug", default=None, help="URL slug (auto-generated from title if omitted)")
|
|
cws.add_argument("--owner", default=None, help="Owner name")
|
|
cws.add_argument("--description", default=None, help="Optional description")
|
|
|
|
# create-task
|
|
ctask = sub.add_parser("create-task", help="Create a task under a workstream")
|
|
ctask.add_argument("--workstream", required=True, metavar="ID_OR_SLUG", help="Workstream UUID or slug")
|
|
ctask.add_argument("--title", required=True, help="Task title")
|
|
ctask.add_argument("--priority", choices=["low", "medium", "high", "critical"], default="medium")
|
|
ctask.add_argument("--assignee", default=None)
|
|
ctask.add_argument("--description", default=None)
|
|
|
|
|
|
# outbox
|
|
outbox = sub.add_parser("outbox", help="Inspect and replay the local State Hub edge outbox")
|
|
outbox.add_argument("--outbox-path", default=None, help="SQLite outbox path (defaults to ~/.statehub/edge-outbox.sqlite3)")
|
|
out_sub = outbox.add_subparsers(dest="outbox_command", required=True)
|
|
|
|
out_status = out_sub.add_parser("status", help="Show pending, conflict, and ack counts")
|
|
out_status.set_defaults(func=cmd_outbox_status)
|
|
|
|
out_list = out_sub.add_parser("list", help="List outbox envelopes as JSON")
|
|
out_list.add_argument("--status", default=None, help="Filter by status")
|
|
out_list.add_argument("--limit", type=int, default=100)
|
|
out_list.set_defaults(func=cmd_outbox_list)
|
|
|
|
out_export = out_sub.add_parser("export", help="Export non-secret envelopes")
|
|
out_export.add_argument("--status", default=None, help="Filter by status")
|
|
out_export.add_argument("--limit", type=int, default=1000)
|
|
out_export.add_argument("--output", default=None, help="Write JSON to a file instead of stdout")
|
|
out_export.set_defaults(func=cmd_outbox_export)
|
|
|
|
out_replay = out_sub.add_parser("replay", help="Replay due queued envelopes")
|
|
out_replay.add_argument("--upstream-url", default=None, help="Central State Hub API base URL")
|
|
out_replay.add_argument("--limit", type=int, default=50)
|
|
out_replay.set_defaults(func=cmd_outbox_replay)
|
|
|
|
out_retry = out_sub.add_parser("retry", help="Force one envelope back to queued")
|
|
out_retry.add_argument("envelope_id")
|
|
out_retry.set_defaults(func=cmd_outbox_retry)
|
|
|
|
out_cancel = out_sub.add_parser("cancel", help="Cancel one envelope")
|
|
out_cancel.add_argument("envelope_id")
|
|
out_cancel.set_defaults(func=cmd_outbox_cancel)
|
|
|
|
# status
|
|
sub.add_parser("status", help="Show State Hub health and summary totals")
|
|
|
|
# review — file-backed multi-owner review projection
|
|
review = sub.add_parser("review", help="Project and inspect multi-owner review evidence")
|
|
review_sub = review.add_subparsers(dest="review_command", required=True)
|
|
review_project = review_sub.add_parser("project", help="Project an authoritative contract wrapper")
|
|
review_project.add_argument("file")
|
|
review_project.add_argument("--source-repo", default=None)
|
|
review_project.add_argument("--decision-id", default=None)
|
|
review_project.add_argument("--workplan-id", default=None)
|
|
review_project.add_argument("--task-id", default=None)
|
|
review_project.add_argument("--required-for-decision", action="store_true")
|
|
review_project.set_defaults(func=cmd_review_project)
|
|
review_submit = review_sub.add_parser("submit", help="Submit an authoritative receipt file")
|
|
review_submit.add_argument("contract_key")
|
|
review_submit.add_argument("file")
|
|
review_submit.add_argument("--source-repo", default=None)
|
|
review_submit.set_defaults(func=cmd_review_submit)
|
|
review_status = review_sub.add_parser("status", help="Print owner and gate status")
|
|
review_status.add_argument("contract_key")
|
|
review_status.set_defaults(func=cmd_review_status)
|
|
|
|
# dev up — files-first local hub (CUST-WP-0054-T07)
|
|
dev = sub.add_parser("dev", help="Local dev-hub commands")
|
|
dev_sub = dev.add_subparsers(dest="dev_command", required=True)
|
|
dev_up = dev_sub.add_parser("up", help="Start local dev hub from repo files")
|
|
dev_up.add_argument(
|
|
"--profile",
|
|
choices=["dev", "fleet"],
|
|
default="dev",
|
|
help="dev=local :8000; fleet=prefer tunnel :18000 when reachable",
|
|
)
|
|
dev_up.add_argument("--with-edge", action="store_true", help="Start edge relay on :18080")
|
|
dev_up.add_argument("--with-mcp", action="store_true", help="Register dev-hub MCP after API is up")
|
|
dev_up.add_argument(
|
|
"--repo-root",
|
|
default=None,
|
|
help="Parent directory for register-from-classification-all (default: $HOME)",
|
|
)
|
|
dev_up.set_defaults(func=cmd_dev_up)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if hasattr(args, "func"):
|
|
args.func(args)
|
|
elif args.command == "register":
|
|
run_statehub_register(args)
|
|
elif args.command == "register-project":
|
|
cmd_register(args)
|
|
elif args.command == "ingest-sbom":
|
|
cmd_ingest_sbom(args)
|
|
elif args.command == "fix-consistency":
|
|
cmd_fix_consistency(args)
|
|
elif args.command == "promote-intake":
|
|
cmd_promote_intake(args)
|
|
elif args.command == "create-workstream":
|
|
cmd_create_workstream(args)
|
|
elif args.command == "create-task":
|
|
cmd_create_task(args)
|
|
elif args.command == "status":
|
|
cmd_status(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|