ci: retain sanitized failure context before pytest interruption

docs/development/ci-impact-selection.md

@@ -52,13 +52,14 @@ Frontstage Pages 只验收并发布公共网站和 projection demo，保留构
 预算、静态宿主 prompt 和后端 Python/TS 逻辑暂不享受免跑；它们仍可能改变核心行为。
 新增一种豁免只需审阅其业务边界和保留的验收，不要求列举全部替代测试文件。
 
-## Four complete Python shards
+## Six complete Python shards
 
-Full Python qualification uses four runners with two xdist workers each:
-`--splits 4 --group N --splitting-algorithm least_duration`. It still partitions
+Full Python qualification uses six shard jobs with two xdist workers each:
+`--splits 6 --group N --splitting-algorithm least_duration`. It still partitions
 the whole collection, excluding only the separately executed Stage2c marker.
 No tests are removed. Without timing history the splitter uses equal weights;
-four-way parallelism is not a claim of perfect duration balancing.
+six shards do not establish perfect duration balancing. Each numbered shard
+shares its queue across workflow runs without replacing another current head.
 
 The shard command explicitly loads `scripts.ci.pytest_timing`. Its controller
 streams pytest's setup/call/teardown durations to Actions logs as each report
@@ -88,6 +89,19 @@ only a lower bound; use job/step timestamps for setup and runner occupancy.
 These observations do not become a timing cache, change selection or establish
 qualification. Missing durations still use the existing equal-weight fallback.
 
+The same observer flushes failed reports as separate `LOOPX_PYTEST_FAILURE`
+JSON lines, including nested failures without counting their duration again.
+They retain the same hashed identity, phase, a known exception class and up to
+eight repository-relative source positions with literal-redacted assertion
+expressions. Messages, parameters, locals and captured IO are omitted from the
+new JSON; pytest's existing output is unchanged. Unknown classes or unavailable
+source context remain null, and an observer error may omit a diagnostic record.
+Only reports received by the controller are covered; collection failures,
+worker death before sending and unseen reports at interruption are excluded.
+These locations are diagnostic context, not complete cause attribution or
+authenticated evidence. JUnit, exit status, full collection and gates remain
+required. The timing reader ignores this separate prefix.
+
 分片在 controller 中随 pytest 的 setup/call/teardown 报告写入 Actions 日志，
 无需等待最终 JUnit 或耗时汇总。下载单个 job 日志后运行
 `node scripts/ci/pytest_timing.ts JOB_LOG`，可回读选中／完成数量、已观察耗时、
@@ -96,6 +110,10 @@ worker 秒数、各阶段总和与最慢用例。缺失、损坏或矛盾记录
 仅收集项的报告提供 phase 耗时；pytest/unittest 子测试仍由 pytest 输出结果，
 其成本已包含在父项 call 中，不重复记入 phase。旧 observer 把子测试混入普通 phase
 的日志仍作为矛盾证据拒绝，reader 不猜测应丢弃哪条重复记录。
+同一 observer 收到失败报告后立即输出 `LOOPX_PYTEST_FAILURE` 脱敏源码位置、异常类和
+断言结构；子测试单独保留失败位置，不重复计时。新增 JSON 不含消息、参数、locals 或
+捕获的 IO，原 pytest 输出不变。未知类、缺失源码或诊断异常仍是证据缺口；未送达的
+报告与收集失败不在覆盖范围，不能据此完成原因归因、证明真实性或豁免完整门禁。
 并发 phase 总和不是
 runner-minutes、排队时长或 CPU 时间；不完整记录仅给出下界，环境准备与占用须结合
 job/step 时间戳。记录不进入 timing cache、不改变测试选择、不替代门禁；没有耗时
@@ -110,15 +128,16 @@ timestamp/ANSI framing; phase identity, ordering and completion checks remain st
 reader 接受 GitHub 日志分块拼接产生的行首 BOM，再读取时间戳和 ANSI 包装；
 phase 身份、顺序与完整性检查保持严格。
 
-The aggregate requires every shard to succeed and all four coverage files to
+The aggregate requires every shard to succeed and all six coverage files to
 exist before combining them. The existing full-suite coverage floor remains.
 No Python or backend TypeScript coverage artifact or Sonar run is manufactured
 when those lanes are exempt. Dashboard coverage remains scoped to the frontend.
 
-全量 Python 从 2 个分片扩大到 4 个，每片仍为 2 个 worker，不提高单机进程争抢。
-真实 pytest-split/xdist/coverage 回归覆盖分片集合互斥、并集完整、四份报告合并以及
+全量 Python 采用 6 个分片，每片仍为 2 个 worker，不提高单机进程争抢。
+各分片号在 workflow 之间共享队列，不替换其他当前 head 的待执行资格。
+真实 pytest-split/xdist/coverage 回归覆盖分片集合互斥、并集完整、六份报告合并以及
 缺失任意报告时拒绝通过。分片增加会增加安装开销与同时占用的 runner；应看实际
-critical path 和 runner-minutes，而不是宣称“4 片必然快一倍”。
+critical path 和 runner-minutes，而不是根据分片数宣称必然加速。
 
 ## Override and evidence
 

scripts/ci/pytest_failure_context.py

@@ -0,0 +1,97 @@
+"""Bounded source context for failed pytest reports; no runtime payload text."""
+from __future__ import annotations
+
+import ast
+import builtins
+import inspect
+import re
+from functools import lru_cache
+from pathlib import Path
+
+import pytest
+
+
+PREFIX = "LOOPX_PYTEST_FAILURE "
+_PATH = re.compile(r"(?:tests|loopx|scripts|examples|packages)/[A-Za-z0-9_./-]+\.py")
+_NAME = re.compile(r"[A-Za-z_][A-Za-z0-9_]{0,127}")
+# Freeze identities and public names before tests can add/rename exceptions.
+_KNOWN_EXCEPTIONS = tuple((value, value.__name__) for value in vars(builtins).values()
+                          if isinstance(value, type) and issubclass(value, BaseException)) + (
+    (pytest.fail.Exception, pytest.fail.Exception.__name__),
+)
+
+
+def _source_path(value, root):
+    try:
+        path = Path(value)
+        relative = (path if path.is_absolute() else root / path).resolve().relative_to(root)
+    except (TypeError, ValueError, OSError, RuntimeError):
+        return None
+    name = relative.as_posix()
+    return name if _PATH.fullmatch(name) and len(name) <= 256 else None
+
+
+class _HideLiterals(ast.NodeTransformer):
+    def visit_Interpolation(self, node):
+        # Python 3.14 keeps original interpolation text separately from value.
+        self.generic_visit(node)
+        node.str = None
+        return node
+
+    def visit_Constant(self, node):
+        if node.value is None or isinstance(node.value, bool) or (
+            isinstance(node.value, (int, float)) and node.value in (0, 1)
+        ):
+            return node
+        return ast.copy_location(ast.Constant(value="<literal>"), node)
+
+
+@lru_cache(maxsize=32)
+def _source_context(root, path):
+    try:
+        with (root / path).open("rb") as stream:
+            source = stream.read(1_000_001)
+        if len(source) > 1_000_000:
+            return {}, frozenset()
+        tree = ast.parse(source.decode("utf-8"))
+        classes = frozenset(node.name for node in ast.walk(tree) if isinstance(node, ast.ClassDef))
+        assertions = {
+            node.lineno: ast.unparse(_HideLiterals().visit(node.test))[:256]
+            for node in ast.walk(tree) if isinstance(node, ast.Assert)
+        }
+        return assertions, classes
+    except (OSError, UnicodeError, SyntaxError, ValueError, RecursionError):
+        return {}, frozenset()
+
+
+def exception_type_name(exception, root):
+    for known, public_name in _KNOWN_EXCEPTIONS:
+        if exception is known:
+            return public_name
+    name = exception.__name__
+    if not _NAME.fullmatch(name):
+        return None
+    # A dynamically constructed exception name can itself contain a secret.
+    # Only disclose non-builtin names declared in in-root Python source.
+    path = _source_path(inspect.getsourcefile(exception), root)
+    return name if path is not None and name in _source_context(root, path)[1] else None
+
+
+def failure_context(report, root):
+    frames = []
+    traceback = getattr(report.longrepr, "reprtraceback", None)
+    for entry in getattr(traceback, "reprentries", ()):
+        location = getattr(entry, "reprfileloc", None)
+        path = _source_path(getattr(location, "path", None), root)
+        line = getattr(location, "lineno", None)
+        if path is not None and type(line) is int and 0 < line <= 1_000_000_000:
+            try:
+                assertion = _source_context(root, path)[0].get(line)
+            except Exception:
+                assertion = None
+            frames.append({"path": path, "line": line, "assertion": assertion})
+    name = getattr(report, "loopx_failure_type", None)
+    if not isinstance(name, str) or not _NAME.fullmatch(name):
+        name = None
+    return {"exception_type": name,
+            "frames": frames[-8:], "frames_truncated": len(frames) > 8}

scripts/ci/pytest_timing.py

@@ -9,15 +9,31 @@
 import json
 import hashlib
 import math
+import re
 import time
 
 import pytest
 from _pytest.subtests import SubtestReport
 
+from scripts.ci.pytest_failure_context import PREFIX as FAILURE_PREFIX
+from scripts.ci.pytest_failure_context import exception_type_name, failure_context
+
 
 PREFIX = "LOOPX_PYTEST_TIMING "
 
 
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_makereport(item, call):
+    report = yield
+    if call.excinfo is not None:
+        try:
+            report.loopx_failure_type = exception_type_name(call.excinfo.type, item.config.rootpath.resolve())
+        except Exception:
+            # Optional diagnostic metadata never replaces pytest's own report.
+            report.loopx_failure_type = None
+    return report
+
+
 def pytest_configure(config):
     # xdist sends TestReports to the controller. Worker output would duplicate
     # reports and can be lost when a worker is killed.
@@ -31,9 +47,9 @@ def __init__(self, config):
         self.started = time.perf_counter()
         self.collection = None
 
-    def emit(self, **row):
+    def emit(self, *, _prefix=PREFIX, **row):
         row["elapsed_seconds"] = round(time.perf_counter() - self.started, 6)
-        line = PREFIX + json.dumps(row, ensure_ascii=True, allow_nan=False, separators=(",", ":"))
+        line = _prefix + json.dumps(row, ensure_ascii=True, allow_nan=False, separators=(",", ":"))
         reporter = self.config.pluginmanager.get_plugin("terminalreporter")
         if reporter is not None:
             reporter.ensure_newline()
@@ -66,6 +82,18 @@ def pytest_collection_finish(self, session):
             self.record_collection([item.nodeid for item in session.items])
 
     def pytest_runtest_logreport(self, report):
+        test_id = hashlib.sha256(report.nodeid.encode("utf-8", errors="surrogatepass")).hexdigest()
+        display = report.nodeid.partition("[")[0][:256]
+        if report.failed:
+            try:
+                self.emit(_prefix=FAILURE_PREFIX, version=1, test_id=test_id,
+                          display_nodeid=display if re.fullmatch(r"[A-Za-z0-9_./:-]+", display) else None,
+                          phase=report.when, subtest=isinstance(report, SubtestReport),
+                          **failure_context(report, self.config.rootpath.resolve()))
+            except Exception:
+                # Do not mask the original failure, halt remaining tests or
+                # lose timing/JUnit if optional context construction/IO fails.
+                pass
         # A subtest is nested within its collected item's call phase. Its
         # duration is already included there; it is not another selected item
         # or another call phase. Keep pytest's own subtest result reporting.
@@ -74,9 +102,8 @@ def pytest_runtest_logreport(self, report):
         duration = report.duration
         # Parameter payloads can be huge or trigger hosted secret masking,
         # corrupting JSON. Hash full identity; display only the parent name.
-        self.emit(event="phase", test_id=hashlib.sha256(
-            report.nodeid.encode("utf-8", errors="surrogatepass")).hexdigest(),
-                  display_nodeid=report.nodeid.partition("[")[0][:256], phase=report.when,
+        self.emit(event="phase", test_id=test_id,
+                  display_nodeid=display, phase=report.when,
                   outcome=report.outcome,
                   seconds=duration if math.isfinite(duration) and duration >= 0 else None)