akcli

"""`akcli capabilities` — one machine-readable manifest of the whole CLI surface.

The single root document an agent reads to drive the tool blind: every
subcommand + flag (introspected from the live ``argparse`` parser, so it cannot
drift from reality), the frozen exit-code and error-code tables, the op-list
vocabulary, the calculator registry, and the packaged JSON Schemas with their
version fields. Environment probing stays in ``akcli doctor``; this command is
pure static surface description and never touches the filesystem outside the
package's own resources.
"""

from __future__ import annotations

import argparse
import json

from .. import contract as _contract
from ..errors import ERROR_CODES, EXIT, exit_for_code
from ._shared import _dumps, _emit

CAPABILITIES_VERSION = _contract.SCHEMA_VERSIONS["capabilities"]

# Option strings owned by the shared global-flags parent (reported once under
# ``global_flags``, filtered out of every per-command flag list).
_GLOBAL_OPTS = {"-C", "--config", "-v", "--verbose", "-q", "--quiet",
                "--json", "--no-color", "--debug"}

_EXIT_MEANING = {
    "OK": "success / no findings",
    "FINDINGS": "check findings present (lint-style; tune with --fail-on)",
    "USAGE": "usage / argument / config error",
    "PARSE": "parse error (corrupt OLE2 or S-expression)",
    "NOT_FOUND": "file not found",
    "UNSUPPORTED_FORMAT": "unsupported format",
    "OPLIST": "op-list or verify failure",
    "TOOL_MISSING": "required external tool missing or network failure",
    "QUERY_MISS": "named net/component does not exist in the (valid) file",
}


def _json_safe(value: object) -> object | None:
    """Defaults worth reporting; anything exotic (or SUPPRESS) becomes None."""
    if value is argparse.SUPPRESS or value is None:
        return None
    if isinstance(value, (str, int, float, bool)):
        return value
    if isinstance(value, (list, tuple)):
        return [v for v in value if isinstance(v, (str, int, float, bool))]
    return None


def _describe_action(action: argparse.Action) -> dict:
    entry: dict = {}
    if action.option_strings:
        entry["flags"] = list(action.option_strings)
    else:
        entry["name"] = action.dest
        entry["required"] = action.nargs not in ("?", "*")
    if action.help and action.help is not argparse.SUPPRESS:
        entry["help"] = " ".join(str(action.help).split())
    if action.choices is not None:
        entry["choices"] = [str(c) for c in action.choices]
    if action.metavar:
        mv = action.metavar
        entry["metavar"] = list(mv) if isinstance(mv, tuple) else mv
    # nargs == 0 covers store_true/store_false/count: the flag takes no value.
    entry["takes_value"] = action.nargs != 0
    default = _json_safe(action.default)
    if default is not None:
        entry["default"] = default
    if getattr(action, "required", False) and action.option_strings:
        entry["required"] = True
    return entry


def _describe_parser(parser: argparse.ArgumentParser) -> tuple[list, list, list]:
    """Return ``(positionals, flags, subcommands)`` for one (sub)parser."""
    positionals: list = []
    flags: list = []
    subcommands: list = []
    for action in parser._actions:  # noqa: SLF001 - argparse has no public walk API
        if isinstance(action, (argparse._HelpAction, argparse._VersionAction)):
            continue
        if isinstance(action, argparse._SubParsersAction):
            helps = {c.dest: c.help for c in action._choices_actions}
            for name, sub in action.choices.items():
                subcommands.append(_describe_command(name, sub, helps.get(name)))
            continue
        if action.option_strings:
            if set(action.option_strings) & _GLOBAL_OPTS:
                continue  # reported once under global_flags
            flags.append(_describe_action(action))
        else:
            positionals.append(_describe_action(action))
    return positionals, flags, subcommands


def _describe_command(name: str, parser: argparse.ArgumentParser,
                      help_text: str | None) -> dict:
    positionals, flags, subcommands = _describe_parser(parser)
    entry: dict = {"name": name}
    if help_text:
        entry["help"] = " ".join(str(help_text).split())
    if positionals:
        entry["positionals"] = positionals
    if flags:
        entry["flags"] = flags
    if subcommands:
        entry["subcommands"] = subcommands
    return entry


def _schema_inventory() -> list[dict]:
    """Packaged JSON Schemas with their declared version fields."""
    from importlib import resources

    out: list[dict] = []
    root = resources.files("akcli.schemas")
    for res in sorted(root.iterdir(), key=lambda r: r.name):
        if not res.name.endswith(".json"):
            continue
        try:
            doc = json.loads(res.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):  # pragma: no cover - packaged data
            continue
        entry: dict = {"name": res.name}
        if isinstance(doc, dict):
            if doc.get("title"):
                entry["title"] = doc["title"]
            for field in ("protocol_version", "proposals_version", "facts_version"):
                if field in doc:
                    entry["version_field"] = field
                    entry["version"] = doc[field]
                    break
            else:
                props = doc.get("properties", {})
                sv = props.get("schema_version", {})
                if "const" in sv:
                    entry["version_field"] = "schema_version"
                    entry["version"] = sv["const"]
                elif sv:
                    entry["version_field"] = "schema_version"
                    entry["version"] = "additive (no const pin)"
        out.append(entry)
    return out


def build_manifest() -> dict:
    """The full capabilities document (also the seam tests introspect)."""
    from .. import __version__
    from .. import cli as _cli   # lazy: cli imports this module at registration
    from .. import ops as _ops
    from .. import opspec as _opspec
    from ..calc import CALCS

    parser = _cli.build_parser()
    _, top_flags, commands = _describe_parser(parser)
    # The shared parent's flags are filtered out of every per-command list;
    # describe them once from the parent parser itself (+ top-level --version).
    global_flags = [
        _describe_action(a) for a in _cli._global_flags()._actions  # noqa: SLF001
        if not isinstance(a, argparse._HelpAction)
    ] + top_flags

    # straight off the ONE vocabulary source (akcli.opspec), not a copy
    macro_names = sorted(_opspec.macro_names())
    core_names = sorted(_opspec.core_names())
    sugar_names = sorted(_opspec.sugar_names())

    return {
        "schema_version": CAPABILITIES_VERSION,
        "akcli_version": __version__,
        "protocol_version": _ops.PROTOCOL_VERSION,
        "exit_codes": [
            {"code": code, "name": name,
             "meaning": _EXIT_MEANING.get(name, name.lower())}
            for name, code in sorted(EXIT.items(), key=lambda kv: kv[1])
        ],
        "error_codes": [
            {"code": code, "exit": exit_for_code(code)}
            for code in sorted(ERROR_CODES)
        ],
        "global_flags": global_flags,
        "commands": commands,
        "ops": {
            "protocol_version": _ops.PROTOCOL_VERSION,
            "core": core_names,
            "sugar": sugar_names,
            "macros": macro_names,
            "capabilities": _ops.load_capabilities().get("ops", {}),
            # Hard vocabulary limits, published here so an agent can branch
            # BEFORE attempting an op instead of learning each constraint
            # from a failed draw (BAD_ANGLE / NON_ORTHOGONAL_WIRE /
            # HIERARCHICAL_UNSUPPORTED / OFF_GRID).
            # ``rotation_enum`` comes from opspec (the ONE vocabulary source)
            # and ``grid_mil`` from ops' own constant, so neither can drift.
            # ``wire_orthogonal_only``
            # and the two ``hierarchy`` fields are hand-maintained prose —
            # there is no constant to export — so whoever changes the writer's
            # hierarchy rule (writers/instances.instances_path,
            # writers/connectivity's INVALID_INSTANCES_PATH) must update these
            # strings and the pin in tests/test_json_version_stamps.py.
            "constraints": {
                "rotation_enum": sorted(_opspec.ENUMS["rotation"]),
                "wire_orthogonal_only": True,
                "grid_mil": int(_ops.GRID_MIL),
                "hierarchy": "hierarchical_write_v2",
                "sheet_reuse": True,
                "hierarchy_note": "one op-list can edit SEVERAL files: the "
                                  "envelope's 'sheets' map + per-op 'sheet' "
                                  "tags route ops onto declared child "
                                  ".kicad_sch files in one atomic "
                                  "transaction (run backup + `undo --run`); "
                                  "a child .kicad_sch is also still a normal "
                                  "single-file draw target. A sheet "
                                  "instantiated more than once needs "
                                  "per-instance references "
                                  "(instance_designators — a scalar ref "
                                  "refuses INSTANCE_REFS_REQUIRED). An op "
                                  "naming a sub-sheet path directly still "
                                  "fails HIERARCHICAL_UNSUPPORTED: route by "
                                  "'sheet' tag instead",
            },
            # HONESTY: Altium is import-only — the per-op "altium" support
            # matrix is informational and there is NO Altium write executor;
            # plan/draw always build target_format "kicad". Read from
            # ops.capabilities.json (the single source of truth) so the manifest
            # cannot drift from the published matrix.
            "altium_write_wired": bool(
                _ops.load_capabilities().get("altium_write_wired", False)),
        },
        "calculators": {"count": len(CALCS), "names": sorted(CALCS)},
        "schemas": _schema_inventory(),
        "conventions": {
            "stdout": "data", "stderr": "logs",
            "json_flag": "--json is accepted by every subcommand",
            "dry_run": "plan never writes; draw writes only with --apply",
            "version_stamps": "every JSON object payload carries "
                              "schema_version or a family version field "
                              "(protocol_version, journal_version); the "
                              "documented exceptions are array payloads "
                              "(jlc search, calc batch) and name-keyed "
                              "tables (calc list, expected)",
            "json_errors": "with --json, a failing command that would "
                           "otherwise leave stdout empty emits "
                           '{"error": {code, message, exit, remediation}} '
                           "on stdout; that covers argparse usage failures "
                           "(unknown command/flag/choice → code USAGE) and "
                           "unexpected exceptions (code INTERNAL); plan/draw "
                           "op-list errors emit the normal draw-result shape "
                           "with status refused",
        },
    }


# Top-level manifest sections addressable via --section (the full manifest is
# ~96 KB — mostly `commands` — while an agent usually needs ONE section or one
# command's flags; a slice is a few hundred bytes instead of the whole tax).
_SECTIONS = ("exit_codes", "error_codes", "global_flags", "commands", "ops",
             "calculators", "schemas", "conventions")


def _slice_payload(manifest: dict, section: str, verb: str | None) -> tuple[dict, int]:
    """One section (or one command) of the manifest, in a stamped envelope."""
    payload = {
        "schema_version": manifest["schema_version"],
        "akcli_version": manifest["akcli_version"],
        "protocol_version": manifest["protocol_version"],
        "section": section,
    }
    if verb is not None:
        match = [c for c in manifest["commands"] if c["name"] == verb]
        payload["command"] = verb
        payload["found"] = bool(match)
        if not match:
            return payload, EXIT["QUERY_MISS"]
        payload["commands"] = match
        return payload, EXIT["OK"]
    payload[section] = manifest[section]
    return payload, EXIT["OK"]


def _cmd_capabilities(args: argparse.Namespace) -> int:
    manifest = build_manifest()
    verb = getattr(args, "command_filter", None)
    section = "commands" if verb else getattr(args, "section", None)
    if section:
        # A slice is machine-facing by definition: JSON on stdout either way.
        payload, code = _slice_payload(manifest, section, verb)
        _emit(_dumps(payload))
        return code
    if args.json:
        _emit(_dumps(manifest))
        return EXIT["OK"]

    lines = [
        f"akcli {manifest['akcli_version']} (protocol {manifest['protocol_version']}, "
        f"capabilities {manifest['schema_version']})",
        "",
        f"commands ({len(manifest['commands'])}):",
    ]
    for cmd in manifest["commands"]:
        subs = ""
        if cmd.get("subcommands"):
            subs = "  [" + " ".join(s["name"] for s in cmd["subcommands"]) + "]"
        lines.append(f"  {cmd['name']:<16} {cmd.get('help', '')}{subs}")
    ops = manifest["ops"]
    lines += [
        "",
        "exit codes: " + " · ".join(
            f"{e['code']} {e['name']}" for e in manifest["exit_codes"]),
        f"ops: {len(ops['core'])} core + {len(ops['sugar'])} sugar "
        f"+ {len(ops['macros'])} macros (protocol {ops['protocol_version']})",
        f"calculators: {manifest['calculators']['count']}",
        "schemas: " + ", ".join(s["name"] for s in manifest["schemas"]),
        "",
        "full manifest: akcli capabilities --json",
    ]
    _emit("\n".join(lines))
    return EXIT["OK"]


def register(sub, common) -> None:
    p = sub.add_parser(
        "capabilities", parents=[common],
        help="print the full machine-readable CLI surface manifest")
    p.add_argument("--section", choices=_SECTIONS,
                   help="print ONE manifest section (stamped JSON slice) "
                        "instead of the whole document")
    p.add_argument("--command", dest="command_filter", metavar="VERB",
                   help="print one command's manifest entry (implies "
                        "--section commands; unknown VERB exits 8)")
    p.set_defaults(handler=_cmd_capabilities)