CLI Commands

akms.cli.commands

CLI commands for AKMS developer operations.

Commands

akms promote — tentative → established (local nodes only) akms suppress — → draft akms deprecate — → deprecated akms status — run graph_status health report akms query — query the compiled knowledge graph akms loadout — query and generate a task loadout akms resolve-task — deterministic task knowledge → loadout + manifest akms mirror-status — show configured code-mirror provider identity akms generate-mirror — refresh code mirrors via configured provider akms orchestrate — run the orchestrator pipeline

cmd_promote

cmd_promote(args: Namespace) -> int

Promote a tentative node to established (local nodes only).

Source code in packages/akms/src/akms/cli/commands.py

def cmd_promote(args: argparse.Namespace) -> int:
    """Promote a tentative node to established (local nodes only)."""
    repo_root = Path(args.repo).resolve()
    node_path = _find_local_node(repo_root, args.node_id)

    if node_path is None:
        print(
            f"Local node '{args.node_id}' not found in {repo_root}/knowledge/local-nodes/",
            file=sys.stderr,
        )
        return 1

    # Verify current status is tentative
    post = fm.load(str(node_path))
    current = post.metadata.get("status", "")
    if current != "tentative":
        print(
            f"Cannot promote: node '{args.node_id}' is '{current}', not 'tentative'",
            file=sys.stderr,
        )
        return 1

    return _set_status_and_invalidate(repo_root, node_path, "established")

cmd_suppress

cmd_suppress(args: Namespace) -> int

Suppress a node (set to draft).

Source code in packages/akms/src/akms/cli/commands.py

def cmd_suppress(args: argparse.Namespace) -> int:
    """Suppress a node (set to draft)."""
    repo_root = Path(args.repo).resolve()
    node_path = _find_local_node(repo_root, args.node_id)

    if node_path is None:
        print(
            f"Local node '{args.node_id}' not found in {repo_root}/knowledge/local-nodes/",
            file=sys.stderr,
        )
        return 1

    return _set_status_and_invalidate(repo_root, node_path, "draft")

cmd_deprecate

cmd_deprecate(args: Namespace) -> int

Deprecate a node.

Source code in packages/akms/src/akms/cli/commands.py

def cmd_deprecate(args: argparse.Namespace) -> int:
    """Deprecate a node."""
    repo_root = Path(args.repo).resolve()
    node_path = _find_local_node(repo_root, args.node_id)

    if node_path is None:
        print(
            f"Local node '{args.node_id}' not found in {repo_root}/knowledge/local-nodes/",
            file=sys.stderr,
        )
        return 1

    return _set_status_and_invalidate(repo_root, node_path, "deprecated")

cmd_status

cmd_status(args: Namespace) -> int

Run graph_status health report.

Source code in packages/akms/src/akms/cli/commands.py

def cmd_status(args: argparse.Namespace) -> int:
    """Run graph_status health report."""
    from akms.schema.validators import parse_propagation_config

    repo_root = Path(args.repo).resolve()

    # F-06: honor config.global_vault as the third step in the precedence
    # chain (explicit arg > env > config > default). `cmd_status` has no
    # --vault flag today; env var and config are the only sources.
    config_path = repo_root / "knowledge" / "graph" / "propagation_config.yaml"
    config = parse_propagation_config(config_path) if config_path.exists() else None

    report = graph_status(repo_root, config=config)
    print(format_report(report))

    return 0

cmd_query

cmd_query(args: Namespace) -> int

Query the compiled graph and print ranked node summaries as JSON.

Source code in packages/akms/src/akms/cli/commands.py

def cmd_query(args: argparse.Namespace) -> int:
    """Query the compiled graph and print ranked node summaries as JSON."""
    from akms.graph.query_subgraph import query_subgraph

    repo_root = Path(args.repo).resolve()
    try:
        config = _load_cli_config(repo_root)
        graph, graph_path = _load_cli_graph(repo_root, args.graph, config)
        ranked = query_subgraph(
            graph,
            args.tags,
            args.role,
            config=config,
            max_depth=args.max_depth,
        )
    except Exception as exc:
        print(f"Query failed: {exc}", file=sys.stderr)
        return 1

    print(
        json.dumps(
            {
                "count": len(ranked),
                # Relative to the repo when inside it: the JSON output is shareable,
                # and an absolute path would leak the local directory layout.
                "graph_path": _display_path(graph_path, repo_root),
                "nodes": _serialize_ranked_nodes(ranked),
            },
            indent=2,
            sort_keys=True,
        )
    )
    return 0

cmd_loadout

cmd_loadout(args: Namespace) -> int

Query the graph and write a canonical task loadout.

Source code in packages/akms/src/akms/cli/commands.py

def cmd_loadout(args: argparse.Namespace) -> int:
    """Query the graph and write a canonical task loadout."""
    from akms.graph.generate_loadout import generate_loadout
    from akms.graph.qmd_cache import compute_graph_version
    from akms.graph.query_subgraph import query_subgraph

    repo_root = Path(args.repo).resolve()
    try:
        _validate_task_id(args.task_id)
        config = _load_cli_config(repo_root)
        graph, graph_path = _load_cli_graph(repo_root, args.graph, config)
        ranked = query_subgraph(
            graph,
            args.tags,
            args.role,
            config=config,
            max_depth=args.max_depth,
        )
        if args.output:
            output_path = _resolve_cli_path(repo_root, args.output, Path(args.output))
        else:
            output_path = _canonical_loadout_path(repo_root, args.phase, args.task_id)
        graph_version = compute_graph_version(graph_path)
        generate_loadout(
            G=graph,
            ranked_nodes=ranked,
            task_id=args.task_id,
            phase=args.phase,
            graph_version=graph_version,
            seed_tags=args.tags,
            agent_role=args.role,
            mode=args.mode,
            available_context=args.available_context,
            config=config,
            output_path=output_path,
            repo_root=repo_root,
        )
    except Exception as exc:
        print(f"Loadout generation failed: {exc}", file=sys.stderr)
        return 1

    print(
        json.dumps(
            {
                "graph_version": graph_version,
                "loadout_path": str(output_path),
                "mode": args.mode,
                "node_count": len(ranked),
            },
            indent=2,
            sort_keys=True,
        )
    )
    return 0

cmd_resolve_task

cmd_resolve_task(args: Namespace) -> int

Resolve exact task knowledge and write loadout + resolution manifest.

Stdout is always machine-readable JSON (no human logs mixed in). Errors are reported via status: error with a non-zero exit code.

Source code in packages/akms/src/akms/cli/commands.py

def cmd_resolve_task(args: argparse.Namespace) -> int:
    """Resolve exact task knowledge and write loadout + resolution manifest.

    Stdout is **always** machine-readable JSON (no human logs mixed in). Errors
    are reported via ``status: error`` with a non-zero exit code.
    """
    from akms.task_context.resolve_task_service import resolve_task

    repo_root = Path(args.repo).resolve()
    result = resolve_task(
        repo_root=repo_root,
        task=args.task_json,
        route_index=args.routes,
        agent_role=args.role,
        changed_paths=getattr(args, "changed_paths", None),
        base=getattr(args, "base", None),
        head=getattr(args, "head", None),
        graph_path=getattr(args, "graph", None),
        loadout_path=getattr(args, "output", None),
        manifest_path=getattr(args, "manifest", None),
        mode=getattr(args, "mode", "routing"),
        available_context=getattr(args, "available_context", 0),
        max_depth=getattr(args, "max_depth", 2),
        phase=getattr(args, "phase", None),
    )
    # Pure JSON on stdout — never mix human logs here.
    print(json.dumps(result.to_json_dict(), indent=2, sort_keys=True))
    return 0 if result.status == "ok" else 1

cmd_orchestrate

cmd_orchestrate(args: Namespace) -> int

Run the orchestrator pipeline.

Source code in packages/akms/src/akms/cli/commands.py

def cmd_orchestrate(args: argparse.Namespace) -> int:
    """Run the orchestrator pipeline."""
    import asyncio

    # Fail fast, before the pipeline starts and before anything is written to
    # disk. Without this the first stage dispatches an agent, hits the missing
    # SDK deep inside execute(), writes a failed AgentMemory, and then blocks
    # on the stage checkpoint gate. The deep ImportError in akms/agents/base.py
    # stays as defense for library users; this only short-circuits the CLI.
    # --agent (an external class) is exempt: only its author knows its deps.
    if not args.agent:
        reason = _backend_unavailable_reason(
            getattr(args, "backend", None) or DEFAULT_BACKEND
        )
        if reason:
            print(reason, file=sys.stderr)
            return 1

    from akms.agents.base import AKMSAgent
    from akms.orchestrator.checkpoint import (
        FileCheckpointHandler,
        TerminalCheckpointHandler,
    )
    from akms.orchestrator.orchestrator import run_pipeline
    from akms.schema.validators import parse_propagation_config

    repo_root = Path(args.repo).resolve()

    # Load config
    config_path = (
        Path(args.config)
        if args.config
        else (repo_root / "knowledge" / "graph" / "propagation_config.yaml")
    )
    if config_path.exists():
        config = parse_propagation_config(config_path)
    else:
        print(f"Config not found at {config_path}, using defaults", file=sys.stderr)
        from akms.schema.models import PropagationConfig

        config = PropagationConfig()

    # Resolve agent class. --agent (explicit dotted path) wins; otherwise
    # --backend maps a friendly name to a built-in runtime; default = AKMSAgent.
    dotted_path = args.agent or (
        BACKENDS.get(args.backend) if getattr(args, "backend", None) else None
    )
    if dotted_path:
        try:
            agent_cls = _import_agent_class(dotted_path)
        except (ImportError, AttributeError) as exc:
            print(
                f"Failed to import agent class '{dotted_path}': {exc}", file=sys.stderr
            )
            return 1
        except TypeError as exc:
            print(str(exc), file=sys.stderr)
            return 1
    else:
        agent_cls = AKMSAgent

    # Override plan_name if provided
    if args.plan:
        config.orchestrator.plan_name = args.plan

    # Select checkpoint handler
    checkpoint_handler = (
        TerminalCheckpointHandler() if args.terminal else FileCheckpointHandler()
    )

    # Run the pipeline. Preflight and stage failures surface as short
    # messages with a nonzero exit; an aborted pipeline (developer abort or
    # checkpoint timeout) is a failure too — only a completed run exits 0.
    from akms.agents.base import AgentPreflightError
    from akms.orchestrator.orchestrator import StageFailedError

    try:
        state = asyncio.run(
            run_pipeline(
                repo_root=repo_root,
                spec_path=args.spec or "",
                goal=args.goal or "",
                plan_name=args.plan or config.orchestrator.plan_name,
                resume=args.resume,
                config=config,
                agent_cls=agent_cls,
                model=args.model,
                checkpoint_handler=checkpoint_handler,
            )
        )
    except AgentPreflightError as exc:
        print(str(exc), file=sys.stderr)
        return 1
    except StageFailedError as exc:
        print(f"{exc} — state saved; resume with --resume.", file=sys.stderr)
        return 1

    if getattr(state, "aborted", False):
        print("Pipeline aborted — state saved; resume with --resume.", file=sys.stderr)
        return 1
    return 0

build_parser

build_parser() -> argparse.ArgumentParser

Build the CLI argument parser.

Source code in packages/akms/src/akms/cli/commands.py

def build_parser() -> argparse.ArgumentParser:
    """Build the CLI argument parser."""
    parser = argparse.ArgumentParser(
        prog="akms",
        description="AKMS — Adaptive Knowledge Management System CLI",
    )
    _add_repo_argument(parser, top_level=True)
    # Host adapters probe `akms --version` before `--help` to decide whether
    # they can enforce a supported-version range, so the output shape
    # (`akms <version>` on stdout, exit 0) is part of the CLI contract.
    parser.add_argument(
        "--version",
        action="version",
        version=f"%(prog)s {__version__}",
    )

    subparsers = parser.add_subparsers(dest="command", help="Available commands")

    # promote
    promote_parser = subparsers.add_parser(
        "promote",
        help="Promote tentative node to established",
    )
    _add_repo_argument(promote_parser)
    promote_parser.add_argument("node_id", help="Node ID to promote")
    promote_parser.set_defaults(func=cmd_promote)

    # suppress
    suppress_parser = subparsers.add_parser(
        "suppress",
        help="Suppress node (set to draft)",
    )
    _add_repo_argument(suppress_parser)
    suppress_parser.add_argument("node_id", help="Node ID to suppress")
    suppress_parser.set_defaults(func=cmd_suppress)

    # deprecate
    deprecate_parser = subparsers.add_parser(
        "deprecate",
        help="Deprecate a node",
    )
    _add_repo_argument(deprecate_parser)
    deprecate_parser.add_argument("node_id", help="Node ID to deprecate")
    deprecate_parser.set_defaults(func=cmd_deprecate)

    # status
    status_parser = subparsers.add_parser(
        "status",
        help="Run graph health report",
    )
    _add_repo_argument(status_parser)
    status_parser.set_defaults(func=cmd_status)

    # query
    query_parser = subparsers.add_parser(
        "query",
        help="Query the compiled graph by seed tags",
    )
    _add_repo_argument(query_parser)
    query_parser.add_argument("tags", nargs="+", help="One or more seed tags")
    query_parser.add_argument(
        "--role",
        choices=("implementer", "code_reviewer", "physics_reviewer"),
        default="implementer",
        help="Agent role query profile (default: implementer)",
    )
    query_parser.add_argument(
        "--max-depth",
        type=int,
        default=2,
        help="Maximum traversal depth from seed nodes (default: 2)",
    )
    query_parser.add_argument(
        "--graph",
        default=None,
        help="Compiled graph path relative to repo (default: knowledge/graph/graph.json)",
    )
    query_parser.set_defaults(func=cmd_query)

    # loadout
    loadout_parser = subparsers.add_parser(
        "loadout",
        help="Query the graph and generate a task loadout",
    )
    _add_repo_argument(loadout_parser)
    loadout_parser.add_argument("task_id", help="Task identifier for the loadout")
    loadout_parser.add_argument(
        "--phase", type=int, required=True, help="Task phase number"
    )
    loadout_parser.add_argument(
        "--tags",
        nargs="+",
        required=True,
        help="One or more query seed tags",
    )
    loadout_parser.add_argument(
        "--role",
        choices=("implementer", "code_reviewer", "physics_reviewer"),
        default="implementer",
        help="Agent role query profile (default: implementer)",
    )
    loadout_parser.add_argument(
        "--mode",
        choices=("routing", "full"),
        default="routing",
        help="Loadout content mode (default: routing)",
    )
    loadout_parser.add_argument(
        "--max-depth",
        type=int,
        default=2,
        help="Maximum traversal depth from seed nodes (default: 2)",
    )
    loadout_parser.add_argument(
        "--available-context",
        type=int,
        default=0,
        help="Estimated available context tokens recorded in the header",
    )
    loadout_parser.add_argument(
        "--graph",
        default=None,
        help="Compiled graph path relative to repo (default: knowledge/graph/graph.json)",
    )
    loadout_parser.add_argument(
        "--output",
        default=None,
        help="Exact output path relative to repo (default: canonical loadout path)",
    )
    loadout_parser.set_defaults(func=cmd_loadout)

    #   # ── resolve-task ─────────────────────────────────────────────────
    resolve_parser = subparsers.add_parser(
        "resolve-task",
        help="Resolve exact task knowledge into a loadout and resolution manifest",
    )
    _add_repo_argument(resolve_parser)
    resolve_parser.add_argument(
        "--task-json",
        required=True,
        help="Path to the task JSON file",
    )
    resolve_parser.add_argument(
        "--routes",
        required=True,
        help="Path to the task route index (JSON or YAML)",
    )
    resolve_parser.add_argument(
        "--role",
        choices=("implementer", "code_reviewer", "physics_reviewer"),
        default="implementer",
        help="Agent role query profile (default: implementer)",
    )
    resolve_parser.add_argument(
        "--phase",
        type=int,
        default=None,
        help="Phase number for output filenames (default: task.phase or 1)",
    )
    resolve_parser.add_argument(
        "--output",
        default=None,
        help="Loadout output path relative to repo "
        "(default: knowledge/loadouts/<phase>-<task>-<role>-loadout.md)",
    )
    resolve_parser.add_argument(
        "--manifest",
        default=None,
        help="Resolution manifest output path relative to repo "
        "(default: knowledge/resolution-manifests/<phase>-<task>-<role>-manifest.json)",
    )
    resolve_parser.add_argument(
        "--changed-paths",
        default=None,
        help="JSON file with a list (or {changed_paths: [...]}) of changed paths. "
        "A bare single path string is rejected — always pass a sequence.",
    )
    resolve_parser.add_argument(
        "--base",
        default=None,
        help="Git base revision for changed-path discovery (mutually exclusive "
        "with --changed-paths)",
    )
    resolve_parser.add_argument(
        "--head",
        default=None,
        help="Git head revision for changed-path discovery (default: HEAD)",
    )
    resolve_parser.add_argument(
        "--mode",
        choices=("routing", "full"),
        default="routing",
        help="Loadout content mode (default: routing)",
    )
    resolve_parser.add_argument(
        "--max-depth",
        type=int,
        default=2,
        help="Advisory query ego-graph depth (default: 2)",
    )
    resolve_parser.add_argument(
        "--available-context",
        type=int,
        default=0,
        help="Estimated available context tokens recorded in the loadout header",
    )
    resolve_parser.add_argument(
        "--graph",
        default=None,
        help="Compiled graph path relative to repo (default: knowledge/graph/graph.json)",
    )
    resolve_parser.set_defaults(func=cmd_resolve_task)

    # orchestrate
    orch_parser = subparsers.add_parser(
        "orchestrate",
        help="Run the orchestrator pipeline",
    )
    _add_repo_argument(orch_parser)
    orch_parser.add_argument(
        "--plan",
        "-p",
        default="",
        help="Plan name (used for branch naming)",
    )
    orch_parser.add_argument(
        "--config",
        "-c",
        default=None,
        help="Path to propagation_config.yaml (default: knowledge/graph/propagation_config.yaml)",
    )
    orch_parser.add_argument(
        "--backend",
        "-B",
        default=None,
        choices=sorted(BACKENDS),
        help=(
            "Built-in agent runtime: claude-sdk (default, Claude Agent SDK), "
            "claude-cli (drives the `claude` CLI headless — needs the claude "
            "binary on PATH, not the SDK), codex-sdk (OpenAI Agents SDK), "
            "codex-cli (drives `codex exec` — needs the codex binary on PATH), "
            "or local (openai-agents against a local OpenAI-compatible endpoint "
            "via AKMS_LLM_API_BASE). Overridden by --agent."
        ),
    )
    orch_parser.add_argument(
        "--agent",
        "-a",
        default=None,
        help=(
            "Dotted import path to an AKMSAgent subclass "
            "(e.g. myproject.akms_agent.MyProjectAgent or "
            "akms.agents.base_codex.AKMSCodexAgent). Overrides --backend. "
            "When both are omitted, uses the default AKMSAgent."
        ),
    )
    orch_parser.add_argument(
        "--model",
        "-m",
        default=None,
        help="Model override (e.g. claude-opus-4-6 for high-risk tasks)",
    )
    orch_parser.add_argument(
        "--resume",
        "-R",
        action="store_true",
        default=False,
        help="Resume from saved pipeline state",
    )
    orch_parser.add_argument(
        "--terminal",
        "-t",
        action="store_true",
        default=False,
        help="Use terminal checkpoint mode instead of file-based",
    )
    orch_parser.add_argument(
        "--spec",
        "-s",
        default=None,
        help="Path to specification file",
    )
    orch_parser.add_argument(
        "--goal",
        "-g",
        default=None,
        help="Goal description for the pipeline",
    )
    orch_parser.set_defaults(func=cmd_orchestrate)

    #   # Mirror provider status / refresh (separate module).
    from akms.cli.provider_commands import register_provider_commands

    register_provider_commands(subparsers)

    #   # Global vault status / install (separate module).
    from akms.cli.vault_commands import register_vault_commands

    register_vault_commands(subparsers)

    return parser

main

main(argv: list[str] | None = None) -> int

Main entry point for the AKMS CLI.

Source code in packages/akms/src/akms/cli/commands.py

def main(argv: list[str] | None = None) -> int:
    """Main entry point for the AKMS CLI."""
    parser = build_parser()
    args = parser.parse_args(argv)

    if not hasattr(args, "func"):
        parser.print_help()
        return 0

    return args.func(args)