orchestrata_ultimate.core.threading.failure_detail

Reference › orchestrata_ultimate › core › threading › failure_detail

Gives the batched test runner full failure detail and scrubs that detail before the runner keeps it. The user asked on 2026-10-22 for the runner to return the full error, debug and trace information. By default each pytest batch now runs with the long traceback format, shows each frame's local variables, and writes captured stdout, stderr and log records for failing tests into the JUnit report; diagnostic_pytest_args returns only those defaults that the caller's own arguments do not already set (_sets tests one token). Detail is capped per module (FAILURE_DETAIL_CAP), per captured stream (CAPTURED_STREAM_CAP) and for the stderr of a killed batch (HUNG_STDERR_CAP), where the all-threads stack dump shows where a test was stuck.

The runner's per-module report is committed as run evidence, so the text first passes through redact_secrets, which hides the values of environment variables whose names look secret-bearing and strings shaped like known access tokens; _worth_hiding skips short values, numbers and booleans that would blank out ordinary text. redacted_tail(text, cap) keeps the last characters after redaction, and format_live_failure lays out a live-log failure in the same form as the JUnit path. The module uses only the standard library because the deployed tool tree vendors this package and checks its import closure at install time. test_executor and pytest_live_observer use it.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Constants

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/threading/failure_detail.py

"""Full failure detail for the batched test runner, and the secret scrub it
passes through before the runner keeps it.

Every batch runs with the full traceback (every frame, not ``--tb=short``'s
one line per frame), each frame's local variables, and the failing test's
captured stdout, stderr and log records, unless the caller set those options
itself. Mohamed, 2026-10-22: "change the runner, improve it, get full error
and debug and trace info".

The runner's per-module report (``reports/run-tests-latest.json``) is
committed by ``ship`` as run evidence, so all of that detail first goes
through :func:`redact_secrets`: a frame's locals routinely hold an ``env``
dict copied from ``os.environ``, which in this repo carries API keys.

Standard library only: the deployed tool tree vendors this package and its
import closure is checked at install time.
"""

from __future__ import annotations

import os
import re
from collections.abc import Mapping, Sequence
from typing import Any

#: Characters of failure detail kept per module. Long tracebacks with locals
#: run far past the 6,000 characters the short form was capped at.
FAILURE_DETAIL_CAP = 60_000
#: Characters kept of each captured stream (stdout, stderr, log) per test.
CAPTURED_STREAM_CAP = 20_000
#: Characters kept of a killed batch's stderr on its hung module: the
#: faulthandler all-threads stack dump that says where the test was stuck.
HUNG_STDERR_CAP = 40_000

#: pytest options every batch gets unless the caller's args already set them,
#: keyed by the option the caller would use to set it.
_DIAGNOSTIC_DEFAULTS: tuple[tuple[str, tuple[str, ...]], ...] = (
    ("--tb", ("--tb=long",)),
    ("--showlocals", ("--showlocals",)),
    # Captured log records, stdout and stderr go into the JUnit report the
    # executor reads failures from -- for failing tests only, so a green
    # batch's report does not carry every passing test's output.
    ("junit_logging", ("-o", "junit_logging=all")),
    ("junit_log_passing_tests", ("-o", "junit_log_passing_tests=false")),
)

#: An environment variable whose value is hidden wherever it appears in a
#: failure. Broader than ``scripts/check_staged_secrets.py``'s name rule on
#: purpose: that gate must not flag ordinary code, while hiding a harmless
#: value in a traceback costs nothing.
_SECRET_ENV_NAME_RE = re.compile(
    r"KEY|TOKEN|SECRET|PASSWORD|PASSWD|PASSPHRASE|CREDENTIAL", re.IGNORECASE
)
#: Credential shapes hidden even when no environment variable holds them.
_TOKEN_RE = re.compile(
    r"sk-(?:or-v1-|ant-|proj-)?[A-Za-z0-9_-]{20,}"  # OpenRouter, Anthropic, OpenAI
    r"|glpat-[A-Za-z0-9_-]{20,}"  # GitLab personal access token
    r"|gh[pousr]_[A-Za-z0-9]{36,}"  # GitHub token
    r"|xox[abprs]-[A-Za-z0-9-]{10,}"  # Slack token
    r"|AKIA[0-9A-Z]{16}"  # AWS access key id
)
#: Extra characters scanned before a tail's cut, so a secret straddling the
#: cut is still whole when it is redacted.
_REDACT_MARGIN = 4096


def diagnostic_pytest_args(pytest_args: Sequence[str]) -> list[str]:
    """The diagnostic defaults that *pytest_args* do not already set."""
    extra: list[str] = []
    for key, args in _DIAGNOSTIC_DEFAULTS:
        if not any(_sets(arg, key) for arg in pytest_args):
            extra.extend(args)
    return extra


def _sets(arg: str, key: str) -> bool:
    """Whether the single command-line token *arg* sets option *key*."""
    if key == "--tb":
        return arg == "--tb" or arg.startswith("--tb=")
    if key == "--showlocals":
        return arg in ("-l", "--showlocals", "--no-showlocals")
    return arg.startswith((f"{key}=", f"-o{key}=", f"--override-ini={key}="))


def redact_secrets(text: str, environ: Mapping[str, str] | None = None) -> str:
    """*text* with each secret-named environment value, and each known token
    shape, replaced by a ``<redacted:...>`` marker.

    Args:
        text: Failure detail about to be kept.
        environ: The environment whose secret values to hide (defaults to
            ``os.environ``, which the runner loaded ``.env`` into).
    """
    if not text:
        return text
    env = os.environ if environ is None else environ
    secrets = sorted(
        (
            (name, value)
            for name, value in env.items()
            if _SECRET_ENV_NAME_RE.search(name) and _worth_hiding(value)
        ),
        # Longest first, so a value that contains another is replaced whole.
        key=lambda item: len(item[1]),
        reverse=True,
    )
    for name, value in secrets:
        if value in text:
            text = text.replace(value, f"<redacted:{name}>")
    return _TOKEN_RE.sub("<redacted:token>", text)


def _worth_hiding(value: str) -> bool:
    """Short values, numbers and booleans would only blank out ordinary text."""
    return len(value) >= 8 and not value.isdigit() and value.lower() != "false"


def redacted_tail(text: str, cap: int, environ: Mapping[str, str] | None = None) -> str:
    """The last *cap* characters of *text*, redacted.

    Redacting a tail after cutting it can leave the front half of a secret
    the cut sliced through. So a wider window is redacted first, starting at
    a line boundary, and only then cut to *cap*.
    """
    if len(text) > cap + _REDACT_MARGIN:
        window = text[-(cap + _REDACT_MARGIN) :]
        text = window[window.find("\n") + 1 :]
    return redact_secrets(text, environ)[-cap:]


def format_live_failure(failure: Mapping[str, Any]) -> str:
    """One live-log failure laid out as the JUnit path lays one out: the test
    and phase, the full traceback, then each captured stream."""
    nodeid = failure.get("nodeid", "?")
    when = failure.get("when", "?")
    parts = [f"{nodeid} — {when} failed\n{failure.get('longrepr', '')}".rstrip()]
    captured: Mapping[str, str] = failure.get("captured") or {}
    for stream, body in captured.items():
        if body:
            parts.append(f"--- captured {stream} ---\n{body.rstrip()}")
    return "\n".join(parts)