codex-mmo/libexec/mmo_cli_output.py

"""Terminal-aware presentation helpers for the Codex MMO command line."""

from __future__ import annotations

import json
import shutil
import sys
from collections.abc import Mapping, Sequence
from typing import Any, TextIO


def json_text(value: Any) -> str:
    """Serialize one strict, deterministic JSON value."""

    return json.dumps(
        value,
        ensure_ascii=False,
        indent=2,
        sort_keys=True,
        allow_nan=False,
    )


def emit_json(value: Any, *, stream: TextIO | None = None) -> None:
    print(json_text(value), file=stream or sys.stdout)


def stdout_is_tty() -> bool:
    return bool(getattr(sys.stdout, "isatty", lambda: False)())


def stderr_is_tty() -> bool:
    return bool(getattr(sys.stderr, "isatty", lambda: False)())


def progress(message: str, *, quiet: bool = False) -> None:
    """Emit stable, non-animated progress only to an interactive diagnostic stream."""

    if not quiet and stderr_is_tty():
        print(message, file=sys.stderr, flush=True)


def emit_error(
    *,
    category: str,
    message: str,
    hint: str | None = None,
    as_json: bool = False,
    exception_type: str | None = None,
    traceback_text: str | None = None,
) -> None:
    """Write one user-facing error without contaminating stdout."""

    if as_json:
        value: dict[str, Any] = {"error_type": category, "error": message}
        if hint:
            value["hint"] = hint
        if exception_type:
            value["exception_type"] = exception_type
        if traceback_text:
            value["traceback"] = traceback_text
        emit_json(value, stream=sys.stderr)
        return
    print(f"error: {message}", file=sys.stderr)
    if hint:
        print(f"hint: {hint}", file=sys.stderr)
    if exception_type:
        print(f"exception: {exception_type}", file=sys.stderr)
    if traceback_text:
        print(traceback_text.rstrip(), file=sys.stderr)


def emit_usage_error(
    usage: str,
    message: str,
    *,
    hint: str | None = None,
    as_json: bool = False,
) -> None:
    if not as_json:
        print(usage.rstrip(), file=sys.stderr)
    emit_error(
        category="usage",
        message=message,
        hint=hint,
        as_json=as_json,
    )


def _display(value: Any) -> str:
    if value is None:
        return "-"
    if isinstance(value, bool):
        return "yes" if value else "no"
    if isinstance(value, float):
        return f"{value:g}"
    if isinstance(value, (list, tuple)):
        return ", ".join(_display(item) for item in value) if value else "-"
    return str(value)


def _clip(value: str, width: int) -> str:
    if width < 4 or len(value) <= width:
        return value
    return value[: width - 3].rstrip() + "..."


def _terminal_width() -> int:
    return max(40, shutil.get_terminal_size(fallback=(100, 24)).columns)


# (heading, mapping key, required, optional maximum width)
Column = tuple[str, str, bool, int | None]


def _table(
    rows: Sequence[Mapping[str, Any]],
    columns: Sequence[Column],
    *,
    empty: str,
) -> None:
    if not rows:
        print(empty)
        return
    values = [
        {key: _display(row.get(key)) for _heading, key, _required, _maximum in columns}
        for row in rows
    ]
    widths: dict[str, int] = {}
    for heading, key, _required, maximum in columns:
        natural = max(len(heading), *(len(row[key]) for row in values))
        widths[key] = min(natural, maximum) if maximum is not None else natural

    selected = list(columns)

    def table_width(items: Sequence[Column]) -> int:
        return sum(widths[key] for _heading, key, _required, _maximum in items) + 2 * (
            len(items) - 1
        )

    available = _terminal_width()
    for column in reversed(columns):
        if table_width(selected) <= available:
            break
        if not column[2] and column in selected:
            selected.remove(column)

    if table_width(selected) > available:
        # Identifiers and other required values remain lossless in narrow terminals.
        for index, row in enumerate(values):
            if index:
                print()
            for heading, key, _required, _maximum in columns:
                print(f"{heading.title()}: {row[key]}")
        return

    print("  ".join(f"{heading:<{widths[key]}}" for heading, key, _r, _m in selected))
    for row in values:
        print(
            "  ".join(
                f"{_clip(row[key], widths[key]):<{widths[key]}}"
                for _heading, key, _required, _maximum in selected
            ).rstrip()
        )


def _mapping_rows(value: Mapping[str, Any], *, key_name: str) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for key, item in value.items():
        if isinstance(item, Mapping):
            rows.append({key_name: key, **item})
        else:
            rows.append({key_name: key, "value": item})
    return rows


def _render_nested(value: Any, *, indent: int = 0) -> None:
    prefix = " " * indent
    if isinstance(value, Mapping):
        if not value:
            print(prefix + "(none)")
            return
        for key, item in value.items():
            label = str(key).replace("_", " ").capitalize()
            if isinstance(item, (Mapping, list, tuple)):
                print(f"{prefix}{label}:")
                _render_nested(item, indent=indent + 2)
            else:
                print(f"{prefix}{label}: {_display(item)}")
        return
    if isinstance(value, (list, tuple)):
        if not value:
            print(prefix + "(none)")
            return
        for item in value:
            if isinstance(item, (Mapping, list, tuple)):
                print(prefix + "-")
                _render_nested(item, indent=indent + 2)
            else:
                print(f"{prefix}- {_display(item)}")
        return
    print(prefix + _display(value))


def _render_pass_report(value: Mapping[str, Any]) -> None:
    passed = value.get("passed", value.get("valid"))
    if passed is not None:
        print(f"Status: {'passed' if passed else 'failed'}")
    for key in ("profile", "profile_id", "suite", "suite_id", "run_id", "report_path"):
        if value.get(key) is not None:
            print(f"{key.replace('_', ' ').title()}: {_display(value[key])}")
    checks = value.get("checks")
    if isinstance(checks, Mapping):
        print("Checks:")
        for name, check in checks.items():
            check_passed = check.get("passed") if isinstance(check, Mapping) else bool(check)
            print(f"  {'pass' if check_passed else 'FAIL'}  {name}")
    for key in ("errors", "warnings", "failure", "error"):
        item = value.get(key)
        if item:
            print(f"{key.capitalize()}:")
            _render_nested(item, indent=2)
    remaining = {
        key: item
        for key, item in value.items()
        if key
        not in {
            "passed",
            "valid",
            "profile",
            "profile_id",
            "suite",
            "suite_id",
            "run_id",
            "report_path",
            "checks",
            "errors",
            "warnings",
            "failure",
            "error",
        }
    }
    if remaining:
        print("Details:")
        _render_nested(remaining, indent=2)


def render_human(command: str, value: Any) -> None:
    """Render one complete structured result for an interactive terminal."""

    if command == "profile.list" and isinstance(value, Mapping):
        _table(
            _mapping_rows(value, key_name="id"),
            (
                ("ACTIVE", "active", False, 6),
                ("PROFILE", "id", True, None),
                ("MATURITY", "maturity", False, 10),
                ("ROOT", "root", False, 24),
                ("SOURCE", "source", False, 8),
                ("DESCRIPTION", "description", False, 60),
            ),
            empty="No profiles found.",
        )
        return
    if command == "tool-mcp.list" and isinstance(value, Mapping):
        servers = value.get("servers")
        rows = _mapping_rows(servers, key_name="server") if isinstance(servers, Mapping) else []
        for row in rows:
            row["ready"] = bool(row.get("ready"))
            row["tools"] = len(row.get("enabled_tools") or [])
        _table(
            rows,
            (
                ("SERVER", "server", True, None),
                ("TRANSPORT", "transport", False, 16),
                ("READY", "ready", False, 5),
                ("TOOLS", "tools", False, 5),
                ("SOURCE", "source", False, 40),
            ),
            empty=f"No Tool MCP servers are defined under {value.get('registry_root', '-')}",
        )
        return
    if command == "catalog.models" and isinstance(value, Mapping):
        _table(
            _mapping_rows(value, key_name="key"),
            (
                ("MODEL", "key", True, None),
                ("MAKER", "maker", False, 16),
                ("ROUTE", "route", False, 28),
                ("INVENTORY", "inventory", False, 18),
                ("AVAILABLE", "availability", False, 12),
            ),
            empty="No matching catalog models found.",
        )
        return
    if command == "catalog.routes" and isinstance(value, Mapping):
        _table(
            _mapping_rows(value, key_name="route"),
            (
                ("ROUTE", "route", True, None),
                ("DRIVER", "driver", False, 20),
                ("PROTOCOL", "wire_protocol", False, 20),
                ("ACCESS", "access_product", False, 22),
                ("ENDPOINT", "base_url", False, 50),
            ),
            empty="No catalog routes found.",
        )
        return
    if command == "catalog.resources" and isinstance(value, Mapping):
        _table(
            _mapping_rows(value, key_name="resource"),
            (
                ("RESOURCE", "resource", True, None),
                ("CAPACITY", "max_active", False, 8),
                ("LOCK", "lock_key", False, 32),
                ("DESCRIPTION", "description", False, 60),
            ),
            empty="No catalog resources found.",
        )
        return
    if command == "gateway.list" and isinstance(value, Sequence):
        _table(
            [item for item in value if isinstance(item, Mapping)],
            (
                ("SNAPSHOT", "snapshot_hash", True, None),
                ("STATUS", "status", True, 12),
                ("PROFILE", "profile_id", False, 30),
                ("PID", "pid", False, 8),
                ("ENDPOINT", "base_url", False, 36),
            ),
            empty="No gateways found.",
        )
        return
    if command == "session.list" and isinstance(value, Sequence):
        _table(
            [item for item in value if isinstance(item, Mapping)],
            (
                ("SESSION", "session_id", True, None),
                ("STATUS", "status", True, 12),
                ("PROFILE", "profile_id", False, 30),
                ("KIND", "session_kind", False, 14),
                ("LAST ACTIVE (UTC)", "last_active_at", False, 26),
                ("RESUMABLE", "resumable", False, 9),
            ),
            empty="No sessions found.",
        )
        return
    if command == "session.runs" and isinstance(value, Sequence):
        _table(
            [item for item in value if isinstance(item, Mapping)],
            (
                ("RUN", "run_id", True, None),
                ("STATUS", "status", True, 12),
                ("KIND", "kind", False, 12),
                ("CREATED (UTC)", "created_at", False, 26),
                ("EXIT", "exit_code", False, 6),
            ),
            empty="No runs found for this session.",
        )
        return
    if command in {"jobs.list", "jobs.status"} and isinstance(value, Sequence):
        _table(
            [item for item in value if isinstance(item, Mapping)],
            (
                ("JOB", "job_id", True, None),
                ("STATUS", "status", True, 12),
                ("AGENT", "agent", False, 28),
                ("BACKEND", "backend", False, 12),
                ("MODEL", "model", False, 42),
                ("LAST PROGRESS (UTC)", "last_progress_at", False, 26),
            ),
            empty="No jobs found.",
        )
        return
    if command == "jobs.result" and isinstance(value, Mapping):
        text = value.get("text", value.get("result", ""))
        if text:
            print(str(text), end="" if str(text).endswith("\n") else "\n")
        else:
            print("No result text is available.")
        cursor = value.get("next_cursor")
        if cursor is not None:
            print(
                f"More result data is available; rerun with --cursor {cursor}.",
                file=sys.stderr,
            )
        return
    if command == "eval.suites" and isinstance(value, Mapping):
        _table(
            _mapping_rows(value, key_name="id"),
            (
                ("SUITE", "id", True, None),
                ("TASKS", "task_count", False, 7),
                ("SOURCE", "source", False, 8),
                ("NAME", "name", False, 36),
                ("DESCRIPTION", "description", False, 60),
            ),
            empty="No evaluation suites found.",
        )
        return
    if command == "eval.list" and isinstance(value, Sequence):
        evaluation_rows: list[Mapping[str, Any]] = [
            item for item in value if isinstance(item, Mapping)
        ]
        _table(
            evaluation_rows,
            (
                ("RUN", "run_id", True, None),
                ("STATUS", "status", True, 12),
                ("PROFILE", "profile_id", False, 30),
                ("SUITE", "suite_id", False, 28),
                ("CREATED (UTC)", "created_at", False, 26),
            ),
            empty="No evaluation runs found.",
        )
        return
    if command in {
        "doctor",
        "profile.doctor",
        "profile.validate",
        "profile.smoke",
        "validate",
        "tool-mcp.validate",
        "catalog.inventory",
        "catalog.verify",
        "catalog.discover",
        "catalog.refresh",
        "eval.validate",
        "eval.run",
    } and isinstance(value, Mapping):
        _render_pass_report(value)
        return
    if isinstance(value, Mapping):
        _render_nested(value)
        return
    if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
        _render_nested(value)
        return
    print(_display(value))


def emit_structured(command: str, value: Any, *, force_json: bool = False) -> None:
    if force_json or not stdout_is_tty():
        emit_json(value)
    else:
        render_human(command, value)


def emit_scalar(value: str, *, json_value: Any | None = None, force_json: bool = False) -> None:
    if force_json:
        emit_json(json_value if json_value is not None else {"value": value})
    else:
        print(value)


def emit_tty_success(
    message: str, *, json_value: Any | None = None, force_json: bool = False
) -> None:
    if force_json:
        emit_json(json_value if json_value is not None else {"status": "ok"})
    elif stdout_is_tty():
        print(message)