scripts.error_report

Reference › scripts › error_report

"What is broken right now": one report, printed every session. It collects the red state from cached artifacts instead of rerunning anything. No module or test imports it.

Invocation: main takes only --exit-code, which makes it exit 1 when anything is red so it can gate a workflow; otherwise it always exits 0. build_report assembles the sections in order and returns the text and a red flag. red_tests lists failing tests from the last recorded test run (_failing_tests), gates lists failing gates from the gate cache, types reports type-check results, blocked_work reports blocked work from the project context file and tree reports the state of the working tree. Each source is a file, so _age and _stale_note state how old it is and warn when it is stale, and _load reads JSON artifacts.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/error_report.py

"""What is broken right now — one report, printed every session.

This exists because of a measured failure mode, not a preference. On
2026-08-09 the answer to "how many tests are red?" was "three" for most of a
session while the artifact on disk said three FAILING MODULES holding ten
failing tests, four of which were regressions introduced that same day. The
gap was never a lie about a number; it was a habit of quoting the last thing
said out loud instead of reading the last thing measured.

So every claim here comes from an artifact on disk, and every section carries
the age of the artifact it read. A stale artifact is reported as stale rather
than as a verdict — "no failures recorded 4 hours ago" is not "nothing is
red", and the difference is the whole point.

Sections
--------
RED TESTS      ``reports/run-tests-latest.json`` — failing modules, and the
               individual test ids parsed out of each module's captured
               output, because a module count understates the damage.
GATES          ``.cache/gates.json`` — the battery's per-gate verdicts, with
               reused/cached entries marked as such. A cached PASS is not
               evidence that the gate passes against the current tree.
TYPES          ``.cache/mypy`` — error count if a run has been recorded.
BLOCKED WORK   ``context.json`` — open P1/high items, which is where defects
               that were surfaced-but-not-fixed go to be forgotten.
TREE           git dirty/ahead state, so an unpushed fix is not mistaken for
               a landed one.

Exit status is 0 unless ``--exit-code`` is passed, so this can be a
non-blocking session hook and a gate from the same entry point.
"""

from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
import time
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parent.parent

TEST_ARTIFACT = REPO_ROOT / "reports" / "run-tests-latest.json"
GATE_CACHE = REPO_ROOT / ".cache" / "gates.json"
MYPY_CACHE = REPO_ROOT / ".cache" / "mypy"
CONTEXT = REPO_ROOT / "context.json"

#: Past this, an artifact describes a tree that has since moved on. Chosen to
#: be shorter than a working session: the question this report answers is
#: "what is red NOW", and an answer from before the last round of edits
#: cannot address it.
STALE_AFTER_SECONDS = 3600

#: At or below this many modules, the artifact came from a targeted run
#: rather than the suite. The full suite is ~1500 modules, so anything this
#: small is someone debugging one file — and it overwrites the same artifact.
PARTIAL_RUN_MODULES = 20

#: pytest's short summary line: ``FAILED path::test - message``.
FAILED_LINE = re.compile(r"^FAILED (\S+)", re.M)
#: The batched runner's own per-test line: ``module::test — traceback``.
RUNNER_LINE = re.compile(r"^(\S+::\S+) — ", re.M)

HIGH_PRIORITIES = frozenset({"P1", "high"})


def _age(path: Path) -> tuple[float, str]:
    """Seconds since *path* was written, and a human rendering of it."""
    if not path.exists():
        return (float("inf"), "absent")
    seconds = time.time() - path.stat().st_mtime
    if seconds < 90:
        return (seconds, f"{int(seconds)}s ago")
    if seconds < 5400:
        return (seconds, f"{int(seconds / 60)}m ago")
    return (seconds, f"{seconds / 3600:.1f}h ago")


def _stale_note(seconds: float) -> str:
    if seconds == float("inf"):
        return "  (no run recorded — this is not evidence of green)"
    if seconds > STALE_AFTER_SECONDS:
        return "  STALE — predates recent edits; re-run before trusting it"
    return ""


def _load(path: Path) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except OSError, json.JSONDecodeError:
        return None


def _failing_tests(captured: str) -> list[str]:
    """Individual failing test ids from one module's captured output."""
    ids = set(FAILED_LINE.findall(captured)) | set(RUNNER_LINE.findall(captured))
    return sorted(ids)


def _label(path: Path) -> str:
    """Repo-relative where possible; an injected tmp path prints in full."""
    try:
        return str(path.relative_to(REPO_ROOT))
    except ValueError:
        return str(path)


def red_tests(lines: list[str], artifact: Path | None = None) -> int:
    """Report failing test modules and the tests inside them.

    Args:
        lines: Output accumulator, appended to in place.
        artifact: Test-run artifact to read. Defaults to the live one; tests
            pass their own rather than mutating module state.

    Returns:
        The number of failing tests found, so the caller can set exit status.
    """
    artifact = artifact if artifact is not None else TEST_ARTIFACT
    seconds, when = _age(artifact)
    lines.append(f"RED TESTS  ({_label(artifact)}, {when})")
    note = _stale_note(seconds)
    if note:
        lines.append(note)

    data = _load(artifact)
    if not isinstance(data, dict):
        lines.append("  unreadable artifact")
        return 0

    modules = data.get("modules")
    modules = modules if isinstance(modules, dict) else {}
    failing = {
        name: body
        for name, body in modules.items()
        if isinstance(body, dict) and body.get("returncode")
    }

    passed, failed = data.get("passed"), data.get("failed")
    lines.append(f"  totals: passed={passed} failed={failed} modules={len(modules)}")
    if 0 < len(modules) <= PARTIAL_RUN_MODULES:
        # A targeted run overwrites the same artifact the full suite writes.
        # Left unsaid, "passed=0 failed=1" off a one-module debug run reads as
        # a suite verdict, and a green one-module run reads as a green suite.
        lines.append(
            f"  PARTIAL RUN — covers {len(modules)} module(s), not the suite. "
            "Says nothing about the rest."
        )

    if not failing:
        lines.append("  no failing module in this artifact")
        return 0

    total = 0
    for name in sorted(failing):
        body = failing[name]
        counts = body.get("testcase_counts")
        counts = counts if isinstance(counts, dict) else {}
        tests = _failing_tests(str(body.get("stdout", "")))
        total += len(tests) or int(counts.get("failed", 0) or 0)
        lines.append(f"  FAIL {name}  ({counts.get('failed', '?')} failed)")
        for test in tests:
            lines.append(f"         {test}")
    lines.append(f"  -> {len(failing)} module(s), {total} failing test(s)")
    return total


def gates(lines: list[str], cache: Path | None = None) -> int:
    """Report the battery's per-gate verdicts, marking cached ones.

    Args:
        lines: Output accumulator, appended to in place.
        cache: Battery result to read. Defaults to the live one.
    """
    cache = cache if cache is not None else GATE_CACHE
    seconds, when = _age(cache)
    lines.append("")
    lines.append(f"GATES  ({_label(cache)}, {when})")
    note = _stale_note(seconds)
    if note:
        lines.append(note)

    data = _load(cache)
    if not isinstance(data, dict):
        lines.append("  no battery result recorded")
        return 0

    entries = data.get("gates") if isinstance(data.get("gates"), dict) else data
    if not isinstance(entries, dict):
        lines.append("  unreadable battery result")
        return 0

    failed = []
    cached = 0
    for name, body in sorted(entries.items()):
        if not isinstance(body, dict):
            continue
        status = str(body.get("status") or body.get("verdict") or "").upper()
        if body.get("cached") or body.get("reused"):
            cached += 1
        if status and status not in {"PASS", "OK", "SKIP"}:
            failed.append(f"{name}={status}")
    lines.append(f"  {len(entries)} gate(s); {cached} served from cache")
    if cached:
        lines.append("  a cached PASS is not proof against the CURRENT tree")
    for entry in failed:
        lines.append(f"  FAIL {entry}")
    return len(failed)


def types(lines: list[str], cache: Path | None = None) -> None:
    cache = cache if cache is not None else MYPY_CACHE
    seconds, when = _age(cache)
    lines.append("")
    lines.append(f"TYPES  ({_label(cache)}, {when})")
    if seconds == float("inf"):
        lines.append("  no mypy run recorded — run it before claiming type-clean")


def blocked_work(lines: list[str], context: Path | None = None) -> None:
    """Open P1/high items: where surfaced-but-unfixed defects accumulate."""
    context = context if context is not None else CONTEXT
    lines.append("")
    lines.append("BLOCKED WORK  (context.json, open P1/high)")
    data = _load(context)
    if not isinstance(data, dict):
        lines.append("  unreadable context.json")
        return
    items = data.get("deferred_work")
    items = items if isinstance(items, list) else []
    top = [
        item
        for item in items
        if isinstance(item, dict)
        and item.get("state") != "done"
        and item.get("priority") in HIGH_PRIORITIES
    ]
    total_open = sum(
        1 for item in items if isinstance(item, dict) and item.get("state") != "done"
    )
    lines.append(f"  {len(top)} at P1/high, {total_open} open overall")
    for item in top:
        lines.append(f"  - {str(item.get('text', '')).splitlines()[0][:110]}")


def tree(lines: list[str]) -> None:
    """Dirty/ahead state, so an unpushed fix is not read as a landed one."""
    lines.append("")
    lines.append("TREE")
    try:
        out = subprocess.run(
            ["git", "status", "-sb", "--porcelain=v1"],
            cwd=REPO_ROOT,
            capture_output=True,
            text=True,
            timeout=20,
            check=False,
        ).stdout.splitlines()
    except (OSError, subprocess.SubprocessError) as exc:
        lines.append(f"  git unavailable: {exc}")
        return
    branch = out[0] if out else "(unknown)"
    dirty = [line for line in out[1:] if line.strip()]
    lines.append(f"  {branch}")
    lines.append(f"  {len(dirty)} uncommitted file(s)")


def build_report() -> tuple[str, int]:
    """The whole report, and the number of red findings in it."""
    lines: list[str] = [
        "=" * 72,
        "ERROR REPORT — read this before reporting status",
        "=" * 72,
    ]
    red = red_tests(lines)
    red += gates(lines)
    types(lines)
    blocked_work(lines)
    tree(lines)
    lines.append("=" * 72)
    if red:
        lines.append(f"{red} red finding(s). Fixing these outranks new work.")
    else:
        lines.append("No red finding in the artifacts read. Check their ages above.")
    lines.append("=" * 72)
    return ("\n".join(lines), red)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=(__doc__ or "").splitlines()[0])
    parser.add_argument(
        "--exit-code",
        action="store_true",
        help="Exit 1 when anything is red, so this can gate a workflow.",
    )
    args = parser.parse_args(argv)

    report, red = build_report()
    print(report)
    return 1 if (args.exit_code and red) else 0


if __name__ == "__main__":
    sys.exit(main())