settled-reach/tooling/core/console.py

"""The single output path (D-263). Nothing else in `reach` prints.

Two channels, and keeping them apart is the whole design:

    stdout   the command's ACTUAL OUTPUT — the data the caller asked for.
             Nothing else ever goes here, so `reach ... | jq` keeps working.
    stderr   the EVENT STREAM — progress, and the final verdict — as JSONL,
             one object per line.

**Rendering happens at the sink, not at the emit site.** When stderr is a
terminal the events are rendered for a human; otherwise they are written as raw
JSONL. A live terminal and a job log are then the same artefact in two
presentations, which is what lets `reach jobs log` render for a person while a
conformance test asserts on the same bytes.

**Emitting is optional.** A command that never calls `event()` works normally,
and the push-gate checks deliberately emit nothing. This is a channel, not an
obligation — the point is that a long command *has somewhere to speak*, not that
every command must.

**The stream never replaces the verdict.** A remedy emitted at line 400 of 900
is technically printed and practically invisible, so `verdict()` prints once,
last, and is what a caller reads when it reads only one thing.

Stdlib only, and cheap to import: this module is on the gate path.
"""

from __future__ import annotations

import json
import os
import sys
from datetime import datetime, timezone
from typing import Any, TextIO

LEVELS: dict[str, int] = {"debug": 10, "info": 20, "warn": 30, "error": 40}

ENV_LEVEL = "SR_LOG_LEVEL"
ENV_FORMAT = "SR_OUTPUT_FORMAT"  # "json" | "text" — overrides TTY detection

# Quiet by default so hooks are not spammed. T-1249's global --verbose lowers it.
_threshold = LEVELS.get(os.environ.get(ENV_LEVEL, "info").lower(), LEVELS["info"])


def set_level(name: str) -> None:
    """Set the minimum level that reaches the stream. Unknown names are a no-op."""
    global _threshold
    if name.lower() in LEVELS:
        _threshold = LEVELS[name.lower()]


def is_verbose() -> bool:
    """True when debug-level events are being emitted.

    Read by the error handler to decide whether an unexpected failure gets a
    traceback or a one-liner. Verbosity is one setting, not two — a --verbose
    that showed debug events but hid tracebacks would be a puzzle.
    """
    return _threshold <= LEVELS["debug"]


def out(text: str = "") -> None:
    """Write to stdout — the command's actual output, never commentary."""
    print(text, file=sys.stdout, flush=True)


def event(
    message: str,
    *,
    level: str = "info",
    phase: str | None = None,
    progress: float | None = None,
    **extra: Any,
) -> None:
    """Emit one progress event onto the stream.

    Suppressed entirely if below the current level, so a debug-chatty service
    costs nothing in a hook.
    """
    if LEVELS.get(level, LEVELS["info"]) < _threshold:
        return
    _write(
        {
            "ts": _now(),
            "level": level,
            "phase": phase,
            "message": message,
            "progress": progress,
            **extra,
        }
    )


def verdict(message: str, *, ok: bool = True, fix: str | None = None) -> None:
    """Print the final summary: once, last, and never suppressed by the level.

    `fix` is the command that would resolve a failure — the contract at the top
    of D-263. It is carried as a structured field so a caller can extract it
    without parsing prose.
    """
    _write(
        {
            "ts": _now(),
            "level": "info" if ok else "error",
            "kind": "verdict",
            "ok": ok,
            "message": message,
            "fix": fix,
        }
    )


def _now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds")


def _write(payload: dict[str, Any]) -> None:
    stream: TextIO = sys.stderr
    if _render_as_text(stream):
        stream.write(_render(payload))
    else:
        # Drop None-valued optionals so a log line carries what happened, not a
        # census of the fields that did not apply.
        compact = {k: v for k, v in payload.items() if v is not None}
        stream.write(json.dumps(compact, ensure_ascii=False) + "\n")
    stream.flush()


def _render_as_text(stream: TextIO) -> str | bool:
    override = os.environ.get(ENV_FORMAT, "").lower()
    if override == "text":
        return True
    if override == "json":
        return False
    try:
        return stream.isatty()
    except (AttributeError, ValueError):
        # A closed or substituted stream: prefer the machine format, which is
        # the one that stays parseable when nobody is watching.
        return False


def _render(payload: dict[str, Any]) -> str:
    message = payload.get("message", "")
    if payload.get("kind") == "verdict":
        if payload.get("ok"):
            return f"{message}\n"
        fix = payload.get("fix")
        tail = f"\n\nFix: {fix}\n" if fix else "\n"
        return f"{message}{tail}"

    prefix = {"warn": "warning: ", "error": "error: "}.get(payload.get("level", ""), "")
    phase = payload.get("phase")
    scope = f"[{phase}] " if phase else ""
    progress = payload.get("progress")
    pct = f" ({progress * 100:.0f}%)" if isinstance(progress, (int, float)) else ""
    return f"{scope}{prefix}{message}{pct}\n"