orchestrata_ultimate.factory.evidence.red_first

Reference › orchestrata_ultimate › factory › evidence › red_first

red_first is the red-first evidence ledger (SPEC-033 section 9, row B6). A test counts as evidence for a change only if it was observed failing at a commit that is an ancestor of the commit where it was observed passing; otherwise it may be a test that can never fail. RedFirstLedger wraps the existing governed-write EventLedger and adds no schema: its two event kinds, test_red_observed and test_green_observed, are stored in the byproducts["kind"] field of ordinary records with empty materials and products, so the enforcement check never misreads them.

record_red and record_green store a RedObservation or GreenObservation. record_from_report reads a pytest report and records one observation per passed or failed test. The report format is the NDJSON stream written by the project's live pytest observer plugin, read per test by iter_test_outcomes. parse_failure extracts a failure type and message from a pytest longrepr, and failure_digest hashes the type with a number-scrubbed message template so equivalent failures compare equal. evidence_status classifies the evidence quality for a test at a change commit, using _reds_for, _greens_for and _is_ancestor (a git ancestry check). The scripts.factory_evidence CLI imports it; tests are in tests/factory/test_red_first.py.

At a glance

Class diagram

Dependency diagram

API

Classes

RedObservation (dataclass)

Decorators: dataclass(frozen=True) · Line 73

One observed-failing outcome for a test, at a commit.

GreenObservation (dataclass)

Decorators: dataclass(frozen=True) · Line 84

One observed-passing outcome for a test, at a commit.

RedFirstLedger

Line 191

Wraps a B1c :class:EventLedger to record + query test red/green observations (see module docstring for the ledger-reuse rationale).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/factory/evidence/red_first.py

"""Red-first evidence ledger (SPEC-033 §9, B6).

A test only counts as evidence for a change if it was observed FAILING
(red) at some commit ANCESTOR to the commit where it was observed PASSING
(green) -- otherwise it may be a test that can never fail (or one whose
green outcome has nothing to do with the change under review).

Report format decision
-----------------------
Two candidate report formats exist in this repo: pytest JUnit-XML (only
``scripts/flaky_test_hunter.py`` writes it, via ``--junit-xml``, and nothing
else consumes it) and the NDJSON stream
``orchestrata_ultimate.core.threading.pytest_live_observer`` writes when its
plugin is loaded (``-p ...pytest_live_observer``) -- ONE line of JSON per
pytest hook event (``start``/``report``/``collect_error``/``session_finish``).
The project's batched executors (``scripts/run_tests_streaming.py``,
``scripts/run_tests.py``) run tests under this plugin, and
``read_live_results`` already reduces its stream to per-file outcomes. This
module reads the SAME NDJSON stream (per-TEST, not per-file) because it is
the format the batched executor the project actually runs already emits --
JUnit-XML would require a second, unused writer. See
:func:`iter_test_outcomes`.

Ledger reuse (B1c)
-------------------
``RedFirstLedger`` wraps the existing B1c
:class:`~orchestrata_ultimate.factory.enforcement.ledger.EventLedger` --
SAME storage (append-only JSONL of :class:`GovernedWriteRecord`), no schema
change. ``GovernedWriteRecord`` already carries a free-form ``byproducts``
dict (ADR-0030's in-toto-link shape reserves it for exactly this kind of
extension); the two new event kinds, ``test_red_observed`` and
``test_green_observed``, are hosted there as ``byproducts["kind"]``.
``materials``/``products`` are left empty -- a test observation has no file
products, so B1a's ``enforce()`` (which only matches delta paths against
recorded ``products``) never misinterprets these records; they coexist
harmlessly in the same ledger file as ordinary governed-write records.
"""

from __future__ import annotations

import hashlib
import json
import re
import subprocess
from collections.abc import Iterator
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Literal

from orchestrata_ultimate.factory.enforcement.ledger import EventLedger
from orchestrata_ultimate.factory.enforcement.models import WriterIdentity

# The writer identity every red/green observation is recorded under. Like
# any B1c record it carries a WriterIdentity, but observations are never
# subject to enforce()'s registered-writer check (they have no products),
# so this identity is never registered in a WriterRegistry.
_OBSERVER_WRITER = WriterIdentity(
    name="red_first_ledger", binary_hash="n/a", config_hash="n/a"
)

_RED_KIND = "test_red_observed"
_GREEN_KIND = "test_green_observed"

EvidenceStatus = Literal["red_first", "green_only", "never_red", "unobserved"]


def _utc_now_iso() -> str:
    return datetime.now(UTC).isoformat()


@dataclass(frozen=True)
class RedObservation:
    """One observed-failing outcome for a test, at a commit."""

    test_id: str
    outcome: Literal["failed"]
    failure_digest: str
    commit: str
    timestamp: str


@dataclass(frozen=True)
class GreenObservation:
    """One observed-passing outcome for a test, at a commit.

    ``failure_digest`` is always ``None`` -- a passing test raised nothing to
    digest; the field exists so both observation shapes are structurally
    symmetric for callers that key on either.
    """

    test_id: str
    outcome: Literal["passed"]
    failure_digest: None
    commit: str
    timestamp: str


# ---------------------------------------------------------------------------
# Failure digest: stable hash of (exception type, message TEMPLATE) -- never
# the full text, since line numbers embedded in pytest's longrepr churn on
# every reformat/rewrite without the underlying assertion changing.
# ---------------------------------------------------------------------------

_NUM_RE = re.compile(r"\d+")
_TYPE_MSG_RE = re.compile(
    r"^([A-Za-z_][\w.]*(?:Error|Exception|Warning))(?::\s*(.*))?$"
)


def parse_failure(longrepr: str) -> tuple[str, str]:
    """Extract ``(failure_type, message)`` from a pytest ``longrepr`` string.

    Pytest's failure text ends in one or more lines prefixed ``E   `` (the
    assertion-rewrite / traceback summary). The LAST line matching
    ``<Type>[: message]`` wins (it is the final, most specific summary); if
    none matches (a bare ``assert <expr>`` with no custom message never gets
    an explicit ``AssertionError:`` prefix), the last ``E   assert ...`` line
    is used verbatim as the message, under the implicit ``AssertionError``
    type pytest's assertion rewriting always raises.
    """
    e_lines = [
        ln.strip()[2:].strip() if ln.strip().startswith("E ") else ln.strip()[1:]
        for ln in longrepr.splitlines()
        if ln.lstrip().startswith("E")
    ]
    for ln in reversed(e_lines):
        m = _TYPE_MSG_RE.match(ln)
        if m:
            failure_type = m.group(1).rsplit(".", 1)[-1]
            return failure_type, (m.group(2) or "")
    for ln in reversed(e_lines):
        if ln.startswith("assert "):
            return "AssertionError", ln
    return "AssertionError", longrepr.strip()[-200:]


def failure_digest(failure_type: str, message: str) -> str:
    """Stable digest of ``failure_type`` + a number-scrubbed message template.

    Numbers are replaced with ``#`` so a value (line number, count, id) that
    changes between runs of the SAME assertion does not change the digest.
    """
    template = _NUM_RE.sub("#", message.strip())
    payload = f"{failure_type}:{template}"
    return hashlib.sha256(payload.encode()).hexdigest()[:16]


# ---------------------------------------------------------------------------
# NDJSON report parsing (pytest_live_observer's stream)
# ---------------------------------------------------------------------------


def iter_test_outcomes(report_path: Path) -> Iterator[tuple[str, str, str | None]]:
    """Yield ``(nodeid, outcome, longrepr)`` -- one per test in the NDJSON
    report at ``report_path``.

    Mirrors ``pytest_live_observer.read_live_results``'s authoritative-
    outcome rule: the ``call`` phase report is authoritative unless
    ``setup`` failed first (in which case ``call`` never runs and ``setup``
    is used instead). Tolerates a truncated final line and malformed JSON --
    such lines are skipped, never raised.
    """
    best: dict[str, tuple[str, str, str | None]] = {}
    with Path(report_path).open(encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                record = json.loads(line)
            except json.JSONDecodeError:
                continue
            if record.get("event") != "report":
                continue
            nodeid = record.get("nodeid")
            when = record.get("when")
            outcome = record.get("outcome")
            if not nodeid or when not in ("setup", "call") or not outcome:
                continue
            longrepr = record.get("longrepr")
            existing = best.get(nodeid)
            if when == "call":
                best[nodeid] = (when, outcome, longrepr)
            elif when == "setup" and outcome == "failed" and existing is None:
                best[nodeid] = (when, outcome, longrepr)
    for nodeid, (_when, outcome, longrepr) in best.items():
        yield nodeid, outcome, longrepr


class RedFirstLedger:
    """Wraps a B1c :class:`EventLedger` to record + query test red/green
    observations (see module docstring for the ledger-reuse rationale).

    ``repo_root`` is the git repository ancestor checks run against --
    ``evidence_status`` needs real commit-graph ancestry, which the ledger
    itself has no notion of.
    """

    def __init__(self, ledger: EventLedger, repo_root: Path):
        self._ledger = ledger
        self._repo_root = Path(repo_root)

    def record_red(
        self,
        test_id: str,
        commit: str,
        worktree: str,
        failure_type: str,
        message: str,
    ) -> RedObservation:
        """Record a RED (failing) observation for ``test_id`` at ``commit``."""
        digest = failure_digest(failure_type, message)
        timestamp = _utc_now_iso()
        self._ledger.record(
            writer=_OBSERVER_WRITER,
            materials={},
            products={},
            worktree=worktree,
            commit=commit,
            byproducts={
                "kind": _RED_KIND,
                "test_id": test_id,
                "failure_digest": digest,
                "observed_at": timestamp,
            },
        )
        return RedObservation(
            test_id=test_id,
            outcome="failed",
            failure_digest=digest,
            commit=commit,
            timestamp=timestamp,
        )

    def record_green(
        self, test_id: str, commit: str, worktree: str
    ) -> GreenObservation:
        """Record a GREEN (passing) observation for ``test_id`` at ``commit``."""
        timestamp = _utc_now_iso()
        self._ledger.record(
            writer=_OBSERVER_WRITER,
            materials={},
            products={},
            worktree=worktree,
            commit=commit,
            byproducts={
                "kind": _GREEN_KIND,
                "test_id": test_id,
                "observed_at": timestamp,
            },
        )
        return GreenObservation(
            test_id=test_id,
            outcome="passed",
            failure_digest=None,
            commit=commit,
            timestamp=timestamp,
        )

    def record_from_report(
        self, report_path: Path, commit: str, worktree: str
    ) -> tuple[list[RedObservation], list[GreenObservation]]:
        """Record a red or green observation for every passed/failed test in
        the NDJSON report at ``report_path`` (skipped/errored tests are
        neither -- they carry no red-first evidence)."""
        reds: list[RedObservation] = []
        greens: list[GreenObservation] = []
        for nodeid, outcome, longrepr in iter_test_outcomes(report_path):
            if outcome == "failed":
                failure_type, message = parse_failure(longrepr or "")
                reds.append(
                    self.record_red(nodeid, commit, worktree, failure_type, message)
                )
            elif outcome == "passed":
                greens.append(self.record_green(nodeid, commit, worktree))
        return reds, greens

    def _reds_for(self, test_id: str) -> list[RedObservation]:
        out = []
        for record in self._ledger.all():
            bp = record.byproducts
            if bp.get("kind") == _RED_KIND and bp.get("test_id") == test_id:
                out.append(
                    RedObservation(
                        test_id=test_id,
                        outcome="failed",
                        failure_digest=bp.get("failure_digest", ""),
                        commit=record.commit,
                        timestamp=bp.get("observed_at", record.recorded_at),
                    )
                )
        return out

    def _greens_for(self, test_id: str) -> list[GreenObservation]:
        out = []
        for record in self._ledger.all():
            bp = record.byproducts
            if bp.get("kind") == _GREEN_KIND and bp.get("test_id") == test_id:
                out.append(
                    GreenObservation(
                        test_id=test_id,
                        outcome="passed",
                        failure_digest=None,
                        commit=record.commit,
                        timestamp=bp.get("observed_at", record.recorded_at),
                    )
                )
        return out

    def _is_ancestor(self, ancestor_commit: str, descendant_commit: str) -> bool:
        result = subprocess.run(
            [
                "git",
                "-C",
                str(self._repo_root),
                "merge-base",
                "--is-ancestor",
                ancestor_commit,
                descendant_commit,
            ],
            capture_output=True,
            check=False,
        )
        return result.returncode == 0

    def evidence_status(self, test_id: str, change_commit: str) -> EvidenceStatus:
        """Classify the red-first evidence quality for ``test_id`` at
        ``change_commit`` (see the four outcomes in the module/class docs).

        - ``red_first``: a red observation at a commit strictly ANCESTOR to
          ``change_commit``, plus a green observation AT ``change_commit``.
        - ``green_only``: a green observation at ``change_commit``, and the
          only red observation(s) found are at that SAME commit (the change
          cannot be what flipped a test that was never seen failing before
          it landed).
        - ``never_red``: a green observation at ``change_commit`` with no
          red observation at all (ancestor or otherwise) for this test.
        - ``unobserved``: no green observation at ``change_commit``.
        """
        greens_here = [
            g for g in self._greens_for(test_id) if g.commit == change_commit
        ]
        if not greens_here:
            return "unobserved"

        reds = self._reds_for(test_id)
        same_commit_red = False
        for red in reds:
            if red.commit == change_commit:
                same_commit_red = True
                continue
            if self._is_ancestor(red.commit, change_commit):
                return "red_first"
        if same_commit_red:
            return "green_only"
        return "never_red"