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 errno
import os
import sys
import time

import typer
import typer.core as _typer_core

# GNU-style, TTY-independent help for every command (Gitea #121 B5): one
# format for humans and agents alike, no rich frames on either `--help` or a
# usage error (`typer.core.HAS_RICH` is what both `TyperCommand.format_help`
# and `TyperGroup.format_help` check before choosing rich rendering over the
# plain-Click fallback - see their own source). Set at import time, not only
# in `main()`, so a test driving `app()` directly (CliRunner) sees the same
# behavior as a real invocation.
_typer_core.HAS_RICH = False

from chemenu import cli_contract  # noqa: E402 - after the HAS_RICH patch, which must land first

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,
        review_cmd,
        run_budget,
        search as search_module,
        task_cmd,
        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


class _BrokenPipeSwallow:
    """Wraps a stream so a write into a closed pipe is dropped instead of
    raised - installed on `sys.stdout`/`sys.stderr` before Typer/Click ever
    run, so Click's own broken-pipe handling (`click.core.BaseCommand.main`)
    never gets the chance to fire.

    Why not just read Click's outcome afterwards: Click already catches this
    exact case (`OSError` with `errno.EPIPE`) and turns it into `sys.exit(1)`
    to avoid a traceback - a clean-looking exit, but indistinguishable from a
    real failure to whatever reads that exit code next. `cli._run_traced`
    does exactly that: it is the trace, which recorded a truncated-but-
    otherwise-successful `types describe source | head -1` as a tool error
    (Gitea #110, measured against a real trace: `exit_code: 1` for a call the
    very next, unpiped, retry of which showed `exit_code: 0`).

    Swallowing the write here instead means Click's own handler never
    triggers, so the command finishes through its normal exit path - `0` for
    an otherwise-successful run - and `sigpipe` on this wrapper is the signal
    `_run_traced` reads to note the truncation without miscasting it as an
    error.
    """

    def __init__(self, wrapped):
        self._wrapped = wrapped
        self.sigpipe = False

    def _is_epipe(self, exc: OSError) -> bool:
        return exc.errno == errno.EPIPE

    def write(self, data):
        try:
            return self._wrapped.write(data)
        except OSError as exc:
            if not self._is_epipe(exc):
                raise
            self.sigpipe = True
            return len(data)

    def flush(self):
        try:
            self._wrapped.flush()
        except OSError as exc:
            if not self._is_epipe(exc):
                raise
            self.sigpipe = True

    def __getattr__(self, attr):
        return getattr(self._wrapped, attr)


def _pacify_real_fd(stream) -> None:
    """Redirect a broken stream's real file descriptor to `os.devnull`.

    Swallowing the write in `_BrokenPipeSwallow` is not enough on its own:
    CPython still flushes the *real* underlying stream automatically at
    interpreter shutdown, by code this module does not control, and that
    flush hits the same closed pipe - printing "Exception ignored while
    flushing sys.stdout" (the well-known CPython caveat; see the standard
    library docs' "Note on SIGPIPE"). Once a pipe is known broken there is
    nothing left worth writing to it, so pointing the fd at `/dev/null`
    makes every later flush - ours or the interpreter's own - a normal
    write that always succeeds.
    """
    try:
        devnull = os.open(os.devnull, os.O_WRONLY)
        try:
            os.dup2(devnull, stream.fileno())
        finally:
            os.close(devnull)
    except (OSError, AttributeError):
        # AttributeError: a stream with no real fd at all (a test double, or
        # a harness that already replaced sys.stdout with something that
        # isn't a file) - nothing to redirect, same as the OSError case.
        pass

# What usage lines and "Try '... -h'" hints name. Without it Click takes
# argv[0], which under `tools/wikitool` (`python -m chemenu.cli`) printed
# `python -m chemenu.cli` - a command nobody should copy.
PROG_NAME = "wikitool"

app = typer.Typer(
    help="wikitool - deterministic operations for Chemenu (see AGENTS.md).",
    no_args_is_help=True,
    # `-h` alongside `--help` on every command (Gitea #121 B5) - Linux
    # convention. Click's context settings inherit down the whole command
    # tree from the root Typer, so this one declaration covers every nested
    # group and command; no command declares its own `-h` (checked).
    context_settings={"help_option_names": ["-h", "--help"]},
)

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.add_typer(task_cmd.app, name="task")
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("review")(review_cmd.review_command)
app.command("publish")(git_publish.publish_command)
app.command("sync")(git_publish.sync_command)
app.command("doctor")(doctor.doctor_command)


def _render_options_text(command, ctx) -> str:
    """Click's own Arguments/Options sections, plain-formatted, for splicing
    into a `cli_contract` record's OPTIONS section. A fixed width (not the
    real terminal's) is what keeps `wikitool <cmd> -h` byte-identical with
    and without a TTY - the whole point of a GNU-style, script-friendly
    format."""
    formatter = ctx.formatter_class(width=80, max_width=100)
    command.format_options(ctx, formatter)
    text = formatter.getvalue().strip()
    # A lone "Options:" label is redundant under our own OPTIONS heading;
    # kept only when Arguments are also present, where it distinguishes the
    # two groups.
    if text.startswith("Options:\n") and "Arguments:\n" not in text:
        text = text[len("Options:\n"):]
    return text


def _render_root_help() -> str:
    """`wikitool -h`/`--help`/no-args: usage, the full index, and where the
    per-command record lives - never Click's default subcommand listing,
    which cannot show a command's typed properties."""
    return (
        "Usage: wikitool <command> [ARGS]...\n\n"
        "wikitool - deterministic operations for Chemenu (see AGENTS.md).\n\n"
        + cli_contract.render_index()
        + "\n\nRun `wikitool <command> -h` for a command's full record "
        "(synopsis, properties, exit status, retry policy, ...).\n"
    )


# The one global override B5 needs (Gitea #121): every leaf command's
# `-h`/`--help` renders from its `cli_contract` record instead of Click's
# default composition, and the bare root command renders the index. A
# command or group with no record (there is currently exactly one such
# case - an intermediate group like `xref` on its own, never asked for by
# name in normal use) falls through to Click's own formatting unchanged.
#
# Patched on `typer._click.core.Command` - typer 0.27 vendors its own
# internal fork of click (`typer._click`), so `TyperCommand`/`TyperGroup`
# (see `typer.core`) resolve `format_help` there, not on the top-level
# `click` package's `Command` class. Both already fall through to this same
# base implementation via `super().format_help(...)` once `HAS_RICH` is
# False (see their own source) - one patch point covers every command and
# group uniformly.
#
# This reaches into a private, underscore-prefixed module with no version
# pin (`requirements.txt` allows any `typer>=0.12`), so a future typer that
# restructures or drops `_click` must not crash every `wikitool` invocation
# at import time. If the shape this needs is not there, skip the patch: help
# falls back to plain, unframed Click output (HAS_RICH is already False)
# without the contract-based rendering - degraded, not broken.
try:
    import typer._click.core as _typer_click_core  # noqa: E402

    _original_format_help = _typer_click_core.Command.format_help

    def _contract_format_help(self, ctx, formatter) -> None:
        if ctx.parent is None:
            formatter.write(_render_root_help())
            return
        record = cli_contract.get(cli_contract.path_of(ctx))
        if record is None:
            _original_format_help(self, ctx, formatter)
            return
        formatter.write(
            cli_contract.render_text(record, options_text=_render_options_text(self, ctx))
        )

    _typer_click_core.Command.format_help = _contract_format_help
except (ImportError, AttributeError):
    pass


_typer_click_core.Command.format_help = _contract_format_help


def main() -> None:
    # Iteration Budget Gate / Loop-Breaker (see instructions/gates.md
    # "Iteration Budget Gate and loop-breaker"): 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 ""
        # A gate refusal leaves `record_and_check` through `_util.fail()`,
        # which raises `typer.Exit` (Gitea #147). That is fine inside Typer's
        # own dispatch - Click catches it - but this call runs *before*
        # `app()` ever starts, so nothing catches it here: left alone, the
        # process would exit 1 correctly but print a Python traceback right
        # after the `ERROR` line, which AGENTS.md § Gates asks a session to
        # show a human and stop on. A traceback reads as a crash, not a gate,
        # and invites exactly the retry the message forbids (see #52).
        try:
            charged = run_budget.record_and_check(command, filtered[1:], override)
        except typer.Exit as exc:
            code = exc.exit_code
            sys.exit(code if isinstance(code, int) else (0 if code is None else 1))
        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(prog_name=PROG_NAME)


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
    real_stdout, real_stderr = sys.stdout, sys.stderr
    stdout_wrap = _BrokenPipeSwallow(real_stdout)
    stderr_wrap = _BrokenPipeSwallow(real_stderr)
    sys.stdout, sys.stderr = stdout_wrap, stderr_wrap
    try:
        app(prog_name=PROG_NAME)
    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 stdout_wrap.sigpipe:
            _pacify_real_fd(real_stdout)
        if stderr_wrap.sigpipe:
            _pacify_real_fd(real_stderr)
        sys.stdout, sys.stderr = real_stdout, real_stderr
        if charged and _util.declined():
            run_budget.refund()
        attrs = {
            "command": command,
            "args": args,
            "exit_code": exit_code,
            "duration_ms": round((time.monotonic() - started) * 1000, 1),
        }
        if stdout_wrap.sigpipe or stderr_wrap.sigpipe:
            attrs["stdout_truncated"] = True
        emit("wikitool", "wikitool.call", attrs)


if __name__ == "__main__":
    main()