taskplane/tp.py

#!/usr/bin/env python3
"""Taskplane: human-gated delivery, native observations, graph and dashboard."""
from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import subprocess
import sys
from typing import Any

if sys.version_info < (3, 10):
    raise SystemExit("Taskplane requires Python 3.10 or newer")
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "taskplane"))

from taskplane import depgraph, flow, graph_primitives, primitives, storage, workspace_binding


def _git(workspace: str, *args: str) -> str:
    result = subprocess.run(["git", *args], cwd=workspace, text=True,
                            encoding="utf-8", errors="replace", capture_output=True)
    if result.returncode:
        raise ValueError(result.stderr.strip() or "Git could not read this workspace")
    return result.stdout


def _changed(workspace: str, base: str) -> list[str]:
    return sorted(set(_git_paths(workspace, "diff", "--name-only", "-z", base, "--")
                      + _git_paths(workspace, "ls-files", "-z", "--others", "--exclude-standard")))


def _git_paths(workspace: str, *args: str) -> list[str]:
    result = subprocess.run(["git", *args], cwd=workspace, capture_output=True)
    if result.returncode:
        raise ValueError(os.fsdecode(result.stderr).strip() or "Git could not read this workspace")
    return [os.fsdecode(path) for path in result.stdout.split(b"\0") if path]


def _version(verify: bool) -> dict[str, Any]:
    manifests = [p for p in (ROOT / ".codex-plugin/plugin.json", ROOT / ".claude-plugin/plugin.json") if p.is_file()]
    versions = [json.loads(p.read_text())["version"] for p in manifests]
    market = ROOT / ".claude-plugin/marketplace.json"
    if market.is_file():
        data = json.loads(market.read_text())
        versions.extend([data["version"], data["plugins"][0]["version"]])
    result: dict[str, Any] = {"version": versions[0]}
    if verify:
        result["ok"] = len(set(versions)) == 1
    return result


def main(argv: list[str] | None = None) -> int:
    arguments = list(sys.argv[1:] if argv is None else argv)
    if arguments[:1] == ["workspace"]:
        return workspace_binding.main(arguments[1:])
    if arguments[:1] == ["flow"]:
        return flow.main(arguments[1:], compact=True)
    if arguments and arguments[0] in flow.HOOK_NAMES:
        if len(arguments) != 1:
            print(json.dumps({"error": "Named native hooks accept their event on stdin only."}))
            return 2
        return flow.run_hook(arguments[0])
    parser = argparse.ArgumentParser(description=__doc__)
    commands = parser.add_subparsers(dest="command", required=True)
    commands.add_parser("flow", help="Human-gated delivery and observations; use flow --help")
    commands.add_parser("workspace", help="Inspect, bind or recover the selected workspace; use workspace --help")
    version = commands.add_parser("version", help="Report the installed version")
    version.add_argument("--verify", action="store_true")
    help_command = commands.add_parser("help", help="Show the supported commands")
    help_command.add_argument("--md", action="store_true")
    graph = commands.add_parser("graph", help="Scan, inspect, and render source dependencies")
    graph.add_argument("--workspace")
    actions = graph.add_subparsers(dest="action", required=True)
    scan = actions.add_parser("scan")
    scan.add_argument("--decompose", action="store_true")
    scan.add_argument("--strict", action="store_true")
    impact = actions.add_parser("impact")
    impact.add_argument("--files")
    impact.add_argument("--base", default="HEAD")
    impact.add_argument("--depth", type=int, default=3)
    impact.add_argument("--json", action="store_true")
    html = actions.add_parser("html")
    html.add_argument("--files")
    html.add_argument("--base", default="HEAD")
    html.add_argument("--out")
    html.add_argument("--focus")
    edge = actions.add_parser("edge")
    edge.add_argument("src")
    edge.add_argument("dst")
    edge.add_argument("--kind", default="runtime")
    edge.add_argument("--note", default="")
    edge.add_argument("--confidence", default="high")
    board = commands.add_parser("dashboard", help="Render the shared delivery dashboard")
    board.add_argument("--workspace")
    board.add_argument("--out")
    board.add_argument("--run")
    review = commands.add_parser("review", help="Capture a source inventory for native review")
    review.add_argument("action", choices=["start"])
    review.add_argument("--workspace")
    review.add_argument("--scope", choices=["repository", "diff"], default="diff")
    review.add_argument("--base", default="HEAD")
    review.add_argument("--paths", help="Comma-separated source paths")
    lens = commands.add_parser("lens", help="Suggest lenses from source signals")
    lens.add_argument("--workspace")
    lens.add_argument("--files", required=True)
    lens.add_argument("--stage")
    args = parser.parse_args(arguments)
    try:
        if args.command == "help":
            if args.md:
                print((ROOT / "docs/cli-reference.md").read_text(), end="")
            else:
                parser.print_help()
            return 0
        if args.command == "version":
            result = _version(args.verify)
            print(json.dumps(result, indent=2))
            return 0 if result.get("ok", True) else 1
        workspace = str(workspace_binding.resolve_workspace(args.workspace))
        workspace_binding.ensure(workspace)
        if args.command == "graph":
            if args.action == "scan":
                data = depgraph.scan(workspace, decompose=args.decompose)
                flow.record_scan(Path(workspace), data)
                result = {"modules": len(data["modules"]), "edges": len(data["edges"]),
                          "files": len(data["files"]), "stored": depgraph._path(workspace)}
                if args.decompose:
                    result["components"] = len(data.get("components", []))
                quality = depgraph.scan_quality(data)
                if quality.get("degraded"):
                    result["graph_quality"] = quality
                print(json.dumps(result, indent=2))
                return int(args.strict and bool(quality.get("degraded")))
            if args.action == "edge":
                result = depgraph.record_edge(workspace, args.src, args.dst,
                    kind=args.kind, note=args.note, confidence=args.confidence)
            else:
                files = args.files.split(",") if args.files else _changed(workspace, args.base)
                if args.action == "impact":
                    result = depgraph.impact(workspace, files, max_depth=args.depth)
                else:
                    if not depgraph.load(workspace).get("modules"):
                        raise ValueError("No graph recorded; run graph scan first")
                    print(depgraph.to_html(workspace, files, out=args.out, focus=args.focus))
                    return 0
        elif args.command == "dashboard":
            output = flow.publish_dashboard(Path(workspace), args.run,
                output=Path(args.out) if args.out else None, select=bool(args.run))
            print(str(output))
            return 0
        elif args.command == "lens":
            result = graph_primitives.route_verdicts(workspace, args.files.split(","), stage=args.stage)
        else:
            head = _git(workspace, "rev-parse", "HEAD").strip()
            if args.scope == "repository":
                if _git(workspace, "diff", "HEAD", "--").strip():
                    raise ValueError("Tracked source differs from HEAD; use a diff review")
                files = _git_paths(workspace, "ls-files", "-z")
            else:
                files = _changed(workspace, args.base)
            if args.paths:
                wanted = args.paths.split(",")
                files = [p for p in files if p in wanted]
            files = [p for p in files if not p.startswith(".taskplane/")]
            for filename in files:
                (Path(workspace) / filename).resolve().relative_to(Path(workspace))
            if not files:
                raise ValueError("No selected source; use --scope repository to review tracked source")
            patch = "" if args.scope == "repository" else _git(workspace, "--literal-pathspecs", "diff", args.base, "--", *files)
            source = {"head": head, "scope": args.scope, "base": args.base,
                      "files": files, "patch": patch}
            output = Path(storage.tp_dir(workspace)) / "review-source.json"
            primitives.atomic_json(output, source)
            result = {"status": "ready", "source": str(output), "files": len(files)}
        print(json.dumps(result, indent=2))
        return 0
    except (OSError, ValueError, KeyError, primitives.StateError) as exc:
        print(f"taskplane: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())