chemenu/tools/chemenu/cli.py

"""wikitool - deterministic operations for Chemenu.

The root AGENTS.md holds the invariants that say when these commands are
mandatory; tools/CONTRACT.md is the full per-command reference.
"""
import sys
import time

import typer

try:
    from chemenu.commands import (
        _util,
        cite_cmd,
        dist_cmd,
        doctor,
        docs_verify,
        eval_cmd,
        git_publish,
        index_build,
        instructions_cmd,
        lint as lint_module,
        links_cmd,
        log_append,
        migrate_cmd,
        new_page,
        page_ops,
        provenance_cmd,
        raw_cmd,
        run_budget,
        search as search_module,
        touch as touch_module,
        types_cmd,
        upload_cmd,
        upstream_cmd,
        version_cmd,
        work_cmd,
        xref,
    )
except ModuleNotFoundError as exc:
    # jsonschema/PyYAML are hard, non-optional dependencies (schema validation
    # is the tool's whole safety net) - fail loudly with a fix, not a silent
    # degradation or a raw traceback.
    sys.stderr.write(
        f"wikitool: missing required dependency '{exc.name}'.\n"
        "This is not optional - schema validation depends on it. Run:\n"
        "  cd tools && .venv/bin/pip install -r requirements.txt\n"
    )
    sys.exit(1)

from chemenu.telemetry import emit  # noqa: E402 - after the dependency check

app = typer.Typer(
    help="wikitool - deterministic operations for Chemenu (see AGENTS.md).",
    no_args_is_help=True,
)

app.add_typer(xref.app, name="xref")
app.add_typer(cite_cmd.app, name="cite")
app.add_typer(links_cmd.app, name="links")
app.add_typer(index_build.app, name="index")
app.add_typer(log_append.app, name="log")
app.add_typer(provenance_cmd.app, name="sources")
app.add_typer(raw_cmd.app, name="raw")
app.add_typer(upload_cmd.app, name="upload")
app.add_typer(instructions_cmd.app, name="instructions")
app.add_typer(run_budget.app, name="budget")
app.add_typer(types_cmd.app, name="types")
app.add_typer(docs_verify.app, name="docs")
app.add_typer(work_cmd.app, name="work")
app.add_typer(eval_cmd.app, name="eval")
app.add_typer(dist_cmd.app, name="dist")
app.add_typer(version_cmd.app, name="version")
app.add_typer(migrate_cmd.app, name="migrate")
app.add_typer(upstream_cmd.app, name="upstream")
app.command("new")(new_page.new_page_command)
app.command("touch")(touch_module.touch_command)
app.command("rename")(page_ops.rename_command)
app.command("rm")(page_ops.rm_command)
app.command("move")(page_ops.move_command)
app.command("lint")(lint_module.lint_command)
app.command("search")(search_module.search_command)
app.command("publish")(git_publish.publish_command)
app.command("sync")(git_publish.sync_command)
app.command("doctor")(doctor.doctor_command)


def main() -> None:
    # Iteration Budget Gate / Loop-Breaker (see the tooling contract's
    # "Iteration and Cost Limits"): recorded and enforced here, once per
    # process, before Typer dispatches to any subcommand - so it covers every
    # command uniformly and cannot be bypassed by the calling agent skipping a
    # step. Help output is never counted: discovering a command's options is
    # not iteration on the wiki, and charging for it would discourage exactly
    # the behavior the skills ask for.
    #
    # Tracing sits at the same point for the same reason - one place that no
    # command can route around. It is not the same set, though: the budget
    # exempts read-only retrieval, while the trace records it, because what an
    # agent looked at before acting is exactly what a trajectory scorer needs.
    argv = sys.argv[1:]
    is_help = any(arg in ("--help", "-h") for arg in argv)
    if argv and not is_help:
        override = "--override-budget" in argv
        filtered = [a for a in argv if a != "--override-budget"]
        command = filtered[0] if filtered else ""
        charged = run_budget.record_and_check(command, filtered[1:], override)
        sys.argv = [sys.argv[0], *filtered]
        _run_traced(command, filtered[1:], charged)
        return
    if is_help:
        sys.argv = [sys.argv[0], *[a for a in argv if a != "--override-budget"]]
    app()


def _run_traced(command: str, args: list[str], charged: bool = False) -> None:
    """Dispatch to Typer and record the call, whatever way it ends.

    Typer leaves through SystemExit on every path, success included, so the
    exit code is read there rather than from a return value.

    A call that left through `_util.fail()` declined instead of acting - a
    rejected argument, or a read-only check reporting findings - so its budget
    slot is handed back here. The trace still records it: what the session
    tried is exactly what a trajectory scorer needs, and the loop-breaker keeps
    the call in its history either way.
    """
    started = time.monotonic()
    exit_code = 0
    try:
        app()
    except SystemExit as exc:
        code = exc.code
        exit_code = code if isinstance(code, int) else (0 if code is None else 1)
        raise
    except BaseException:
        exit_code = 1
        raise
    finally:
        if charged and _util.declined():
            run_budget.refund()
        emit(
            "wikitool",
            "wikitool.call",
            {
                "command": command,
                "args": args,
                "exit_code": exit_code,
                "duration_ms": round((time.monotonic() - started) * 1000, 1),
            },
        )


if __name__ == "__main__":
    main()