project-orchestrator/scripts/fleet.py

#!/usr/bin/env python3
"""fleet.py — read, validate, and report the PO fleet registry (fleet.toml).

This is the PO's view of the fleet. It NEVER touches a project's repo and carries no project-side
state — the registry here is the single source of truth for project ↔ harness ↔ ref ↔ location.

Usage:
    python3 scripts/fleet.py list             # one line per project (also validates the file)
    python3 scripts/fleet.py status           # same, plus a one-line summary count
    python3 scripts/fleet.py validate         # parse + schema-check only; exit 1 on any error
    python3 scripts/fleet.py get <name>       # dump one project's full entry
    python3 scripts/fleet.py --file PATH ...  # use a non-default registry (default: ./fleet.toml)

Needs only the Python stdlib (tomllib → python >= 3.11).
"""
import sys
import tomllib
from pathlib import Path

REQUIRED = ["name", "location", "harness", "ref", "enabled", "secrets"]
OPTIONAL = ["config", "notes"]


def _registry_path(argv):
    path = Path("fleet.toml")
    out = []
    i = 0
    while i < len(argv):
        if argv[i] == "--file" and i + 1 < len(argv):
            path = Path(argv[i + 1]); i += 2; continue
        out.append(argv[i]); i += 1
    return path, out


def load(path):
    if not path.exists():
        sys.exit(f"fleet: registry not found: {path}")
    with open(path, "rb") as f:
        raw = tomllib.load(f)
    return raw


def validate(raw, path):
    errors = []
    projects = raw.get("project", [])
    if not isinstance(projects, list):
        errors.append("[[project]] must be an array of tables")
        projects = []
    seen = set()
    for i, p in enumerate(projects):
        tag = p.get("name", f"#{i}")
        for k in REQUIRED:
            if k not in p:
                errors.append(f"project {tag}: missing required field '{k}'")
        if not isinstance(p.get("enabled", False), bool):
            errors.append(f"project {tag}: 'enabled' must be a boolean")
        name = p.get("name")
        if name in seen:
            errors.append(f"duplicate project name: {name}")
        if name:
            seen.add(name)
        for k in p:
            if k not in REQUIRED + OPTIONAL:
                errors.append(f"project {tag}: unknown field '{k}'")
    return projects, errors


def cmd_list(projects):
    if not projects:
        print("(no projects registered)")
        return
    w = max(len(p.get("name", "?")) for p in projects)
    for p in projects:
        flag = "enabled " if p.get("enabled") else "disabled"
        print(f"  {p.get('name','?'):<{w}}  [{flag}]  {p.get('harness','?')}@{p.get('ref','?')}"
              f"  {p.get('location','?')}")


def main():
    path, argv = _registry_path(sys.argv[1:])
    cmd = argv[0] if argv else "list"
    raw = load(path)
    projects, errors = validate(raw, path)
    if cmd == "validate":
        if errors:
            for e in errors:
                print(f"  ERROR: {e}")
            sys.exit(1)
        print(f"fleet: OK — {len(projects)} project(s), schema v{raw.get('fleet', {}).get('version', '?')}")
        return
    # for list/status/get we still surface errors but don't hard-fail the read
    for e in errors:
        print(f"  ERROR: {e}", file=sys.stderr)
    if cmd == "list":
        cmd_list(projects)
    elif cmd == "status":
        cmd_list(projects)
        en = sum(1 for p in projects if p.get("enabled"))
        print(f"\n  total={len(projects)}  enabled={en}  disabled={len(projects) - en}"
              f"  (registry schema v{raw.get('fleet', {}).get('version', '?')})")
    elif cmd == "get" and len(argv) > 1:
        target = argv[1]
        for p in projects:
            if p.get("name") == target:
                for k in REQUIRED + OPTIONAL:
                    if k in p:
                        print(f"  {k:<9} = {p[k]!r}")
                return
        sys.exit(f"fleet: no project named {target!r}")
    else:
        sys.exit(__doc__)
    if errors:
        sys.exit(1)


if __name__ == "__main__":
    main()