Streaming as a decorator, first half. Each invocation of reach gets an id and every event it emits is tagged with it, which is what will let a detached run's log be read back and what correlates the lines of a run that streamed for nine minutes. No command signature changed and no command imports core.jobs — that is the point, per the D-263 amendment: a command must not know jobs exist, because the alternative is call-site discipline wearing a different hat. A ContextVar rather than a module global. A global is correct only until something runs two invocations in one process — which a test harness or a future batch verb does immediately, and which would then interleave two jobs' events under one id with nothing reporting an error. The job context is the OUTERMOST wrapper, and it has to be. @logged emits from its finally and @handle_errors emits its verdict while unwinding, so a context established inside either would already be reset by the time the two most important events are written — leaving them the only untagged lines in the log, and they are precisely the ones a detached run gets read back for. Fixed in passing: the job id used local time while every event's ts is UTC, so an id read 155327 beside its own first log line reading 13:53:27. Two hours apart reads as a logging bug every time someone correlates them by eye. New conformance invariant — nothing outside core/ may import core.jobs. My first version of it inspected only the module path, so it missed `from tooling.core import jobs`, where the name is in the import LIST and which is the form anyone would actually write. It passed while checking nothing. Rewritten to catch all three reachable forms and then verified by committing a real violation, which it named by file and line. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
168 lines
5.6 KiB
Python
168 lines
5.6 KiB
Python
"""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
|
|
|
|
from tooling.core import jobs
|
|
|
|
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:
|
|
# Tagged here rather than at each call site, so nothing can emit an
|
|
# uncorrelated line. The two events that matter most for a detached run —
|
|
# the verdict and the invocation record — are written while unwinding, and
|
|
# would be the ones a call-site approach missed.
|
|
job_id = jobs.current()
|
|
if job_id:
|
|
payload = {**payload, "job": job_id}
|
|
|
|
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"
|