ci: preserve timing evidence from interrupted Python shards

.github/workflows/python-tests.yml

@@ -552,6 +552,7 @@ jobs:
         # Each runner retains the measured two-worker pool.
         run: >-
           python -m pytest -v -n 2 -m "not stage2c_e2e"
+          -p scripts.ci.pytest_timing
           --splits 4 --group ${{ matrix.shard }}
           --splitting-algorithm least_duration
           --durations=25 --durations-min=1

docs/development/ci-impact-selection.md

@@ -60,6 +60,56 @@ 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.
 
+The shard command explicitly loads `scripts.ci.pytest_timing`. Its controller
+streams pytest's setup/call/teardown durations to Actions logs as each report
+arrives, before final JUnit or duration summaries exist. To inspect a downloaded
+job log, run `node scripts/ci/pytest_timing.ts JOB_LOG`. The reader reports the
+selected and finished cases, observed elapsed time, measured worker-seconds,
+phase totals and longest observed cases.
+Only collected-item reports supply phase durations. Nested pytest/unittest
+subtests remain in pytest's result output; their cost is already included in
+the parent call, so they do not duplicate timing phases. Logs produced by the
+earlier observer that mixed subtest reports into ordinary phases remain rejected
+as contradictory evidence; the reader does not guess which duplicate to discard.
+The phase record identifies the full pytest node id by SHA-256 and displays the
+parent name without parameter payloads, bounded to 256 characters. Distinct
+parameter cases keep distinct hashes; huge values and hosted secret masking of
+parameter strings cannot expand or corrupt these JSON records. The existing
+verbose pytest result output remains available.
+Observation rows occupy their own lines with either quiet (`-q`) or verbose
+(`-v`) progress, including serial and xdist runs. The reader still requires
+line-framed records rather than guessing how to recover old progress-prefixed logs.
+Missing observations, malformed records or contradictory collections fail
+readback; an interrupted or incomplete run stays
+incomplete. A completed traversal with pytest exit code 1 still has test failures.
+Measured worker-seconds sum concurrent test phases and are not runner-minutes,
+queue delay, CPU time or the job's complete elapsed time. A partial log supplies
+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.
+
+分片在 controller 中随 pytest 的 setup/call/teardown 报告写入 Actions 日志，
+无需等待最终 JUnit 或耗时汇总。下载单个 job 日志后运行
+`node scripts/ci/pytest_timing.ts JOB_LOG`，可回读选中／完成数量、已观察耗时、
+worker 秒数、各阶段总和与最慢用例。缺失、损坏或矛盾记录拒绝回读；中断与不完整
+运行保持不完整。遍历完成且 exit code 为 1 仍代表测试失败。
+仅收集项的报告提供 phase 耗时；pytest/unittest 子测试仍由 pytest 输出结果，
+其成本已包含在父项 call 中，不重复记入 phase。旧 observer 把子测试混入普通 phase
+的日志仍作为矛盾证据拒绝，reader 不猜测应丢弃哪条重复记录。
+并发 phase 总和不是
+runner-minutes、排队时长或 CPU 时间；不完整记录仅给出下界，环境准备与占用须结合
+job/step 时间戳。记录不进入 timing cache、不改变测试选择、不替代门禁；没有耗时
+历史时仍按原等权规则分片。
+完整 pytest node id 用 SHA-256 标识；显示名去除参数内容并限制为 256 字符。
+不同参数仍有不同 hash；超长参数及宿主对参数字符串的脱敏不会扩大或破坏这些
+JSON 记录。原有 pytest 详细结果输出仍保留。
+无论使用 quiet（`-q`）还是 verbose（`-v`）进度，串行和 xdist 运行中的观察记录
+都独占一行。reader 仍要求完整行记录，不猜测如何修复旧的进度字符前缀日志。
+The reader accepts a line-leading BOM from concatenated GitHub log chunks before
+timestamp/ANSI framing; phase identity, ordering and completion checks remain strict.
+reader 接受 GitHub 日志分块拼接产生的行首 BOM，再读取时间戳和 ANSI 包装；
+phase 身份、顺序与完整性检查保持严格。
+
 The aggregate requires every shard to succeed and all four 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

docs/development/testing-and-quality.md

@@ -603,6 +603,11 @@ individual shards. Relative coverage paths make reports portable across runners.
 Each shard has a 60-minute job ceiling for setup, the full partition and report
 upload; individual test deadlines stay unchanged. Verbose progress names failed
 cases before the suite finishes, so an interrupted run retains their identities.
+An explicitly loaded CI observer also streams pytest-owned phase durations from
+the xdist controller. Read the downloaded log with
+`node scripts/ci/pytest_timing.ts JOB_LOG`; incomplete evidence remains incomplete
+and does not replace JUnit, coverage or the required gate. See the
+[timing observation boundary](ci-impact-selection.md#four-complete-python-shards).
 The reusable Sonar workflow consumes that same run's XML and never reruns
 pytest or reads cross-run artifacts. Missing Sonar tokens still skip analysis
 successfully; test jobs receive no Sonar secret. The trigger is the union of
@@ -615,6 +620,10 @@ Linux 全套测试分到四台 hosted runner，每台保留两个 xdist worker
 合并后再执行原有 19.6% 门槛；不要求单个分片达到全套覆盖率。coverage 使用相对路径，
 每个分片的作业上限为 60 分钟，覆盖环境准备、完整分片和报告上传；单项测试超时
 保持不变。详细进度在套件结束前就记录失败用例名称，运行中断后仍可定位这些用例。
+CI 同时在 xdist controller 流式记录 pytest 原生 phase 耗时。下载日志后用
+`node scripts/ci/pytest_timing.ts JOB_LOG` 回读；不完整证据保持不完整，不能替代
+JUnit、coverage 或必过门禁。归属与测量边界见
+[分片耗时记录](ci-impact-selection.md#four-complete-python-shards)。
 Sonar 只复用同一次 run 的 XML，不重复测试、不跨 run 取产物。缺少 token 仍成功跳过
 Sonar，测试 job 不接收 Sonar secret。触发范围取原有两套 workflow 的并集；纯前端
 PR 使用前述豁免，Sonar 配置变更仍全量运行，包括没有 token 的 fork。

scripts/ci/pytest_timing.py

@@ -0,0 +1,85 @@
+"""Stream pytest-owned timings from the controller, including interrupted runs.
+
+This is a CI observation adapter, not a splitter, timing cache or test gate.
+Load explicitly with ``-p scripts.ci.pytest_timing``. No configuration or test
+selection changes; node ids and phase/outcome values come from pytest itself.
+"""
+from __future__ import annotations
+
+import json
+import hashlib
+import math
+import time
+
+import pytest
+from _pytest.subtests import SubtestReport
+
+
+PREFIX = "LOOPX_PYTEST_TIMING "
+
+
+def pytest_configure(config):
+    # xdist sends TestReports to the controller. Worker output would duplicate
+    # reports and can be lost when a worker is killed.
+    if not hasattr(config, "workerinput"):
+        config.pluginmanager.register(TimingObserver(config), "loopx-ci-timing-observer")
+
+
+class TimingObserver:
+    def __init__(self, config):
+        self.config = config
+        self.started = time.perf_counter()
+        self.collection = None
+
+    def emit(self, **row):
+        row["elapsed_seconds"] = round(time.perf_counter() - self.started, 6)
+        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()
+            # Quiet progress writes directly to the terminal writer without
+            # currentfspath, so ensure_newline alone cannot end that line.
+            if reporter._tw.width_of_current_line:
+                reporter._tw.line()
+            reporter.write_line(line)
+            reporter._tw.flush()
+
+    @pytest.hookimpl(trylast=True)
+    def pytest_sessionstart(self, session):
+        self.emit(event="start", version=1)
+
+    def record_collection(self, ids):
+        if self.collection is None:
+            self.collection = tuple(ids)
+            self.emit(event="collection", selected=len(ids))
+        elif self.collection != tuple(ids):
+            # xdist owns rejecting different worker collections. Preserve the
+            # observation without changing that decision or claiming coverage.
+            self.emit(event="collection_mismatch")
+
+    @pytest.hookimpl(optionalhook=True)
+    def pytest_xdist_node_collection_finished(self, node, ids):
+        self.record_collection(ids)
+
+    def pytest_collection_finish(self, session):
+        if not getattr(self.config.option, "numprocesses", None):
+            self.record_collection([item.nodeid for item in session.items])
+
+    def pytest_runtest_logreport(self, report):
+        # 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.
+        if isinstance(report, SubtestReport):
+            return
+        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,
+                  outcome=report.outcome,
+                  seconds=duration if math.isfinite(duration) and duration >= 0 else None)
+
+    @pytest.hookimpl(trylast=True)
+    def pytest_sessionfinish(self, session, exitstatus):
+        self.emit(event="end", exit_code=int(session.exitstatus))

scripts/ci/pytest_timing.ts

@@ -0,0 +1,108 @@
+/** Read CI timing observations; never supplies a test gate or reusable cache. */
+import { readFileSync, statSync } from "node:fs";
+import { pathToFileURL } from "node:url";
+
+const prefix = "LOOPX_PYTEST_TIMING ";
+type Phase = "setup" | "call" | "teardown"; // pytest TestReport vocabulary, local adapter
+type Outcome = "passed" | "failed" | "skipped";
+type Observation = { seconds: number | null; outcome: Outcome };
+type Case = { display_nodeid: string } & Partial<Record<Phase, Observation>>;
+
+function finite(value: unknown): value is number {
+  return typeof value === "number" && Number.isFinite(value) && value >= 0;
+}
+
+export function summarizeTimingLog(log: string) {
+  const cases = new Map<string, Case>();
+  let started = false, ended = false;
+  let selected: number | null = null, exitCode: number | null = null;
+  let elapsed = 0, records = 0;
+  for (const original of log.split(/\r?\n/)) {
+    const line = original.replace(/^\uFEFF/, "").replace(/\x1b\[[0-9;]*m/g, "")
+      .replace(/^\d{4}-\d{2}-\d{2}T\S+Z\s+/, "");
+    if (!line.startsWith(prefix)) continue;
+    if (++records > 200_000) throw new Error("Timing log exceeds the observation limit.");
+    const row = JSON.parse(line.slice(prefix.length));
+    if (!row || Array.isArray(row) || typeof row !== "object" ||
+        !finite(row.elapsed_seconds) || row.elapsed_seconds < elapsed || ended) {
+      throw new Error("Invalid timing record order or elapsed duration.");
+    }
+    elapsed = row.elapsed_seconds;
+    if (row.event === "start") {
+      if (started || records !== 1 || row.version !== 1) throw new Error("Invalid timing start.");
+      started = true;
+    } else if (!started) {
+      throw new Error("Timing record has no start.");
+    } else if (row.event === "collection") {
+      if (selected !== null || cases.size || !Number.isSafeInteger(row.selected) || row.selected < 0) {
+        throw new Error("Invalid selected collection count.");
+      }
+      selected = row.selected;
+    } else if (row.event === "phase") {
+      if (selected === null || typeof row.test_id !== "string" || !/^[a-f0-9]{64}$/.test(row.test_id) ||
+          typeof row.display_nodeid !== "string" || !row.display_nodeid ||
+          row.display_nodeid.length > 512 || !["setup", "call", "teardown"].includes(row.phase) ||
+          !["passed", "failed", "skipped"].includes(row.outcome) ||
+          (row.seconds !== null && !finite(row.seconds))) throw new Error("Invalid pytest phase.");
+      const phase = row.phase as Phase;
+      const entry: Case = cases.get(row.test_id) ?? { display_nodeid: row.display_nodeid };
+      if (entry.display_nodeid !== row.display_nodeid || entry[phase] || (phase !== "setup" && !entry.setup) ||
+          (phase === "call" && entry.setup?.outcome !== "passed") || entry.teardown) {
+        throw new Error("Duplicate or out-of-order pytest phase.");
+      }
+      entry[phase] = { seconds: row.seconds, outcome: row.outcome as Outcome };
+      cases.set(row.test_id, entry);
+      if (cases.size > selected) throw new Error("Reported cases exceed the selected collection.");
+    } else if (row.event === "end") {
+      if (!Number.isInteger(row.exit_code) || row.exit_code < 0 || row.exit_code > 5) {
+        throw new Error("Invalid pytest exit code.");
+      }
+      exitCode = row.exit_code;
+      ended = true;
+    } else {
+      throw new Error(`Unusable timing event: ${String(row.event)}`);
+    }
+  }
+  if (!started) throw new Error("No CI timing observations found; this is not a zero-cost run.");
+  let measured = 0, invalid = 0, finished = 0;
+  const phaseSeconds = { setup: 0, call: 0, teardown: 0 };
+  const longest = [];
+  for (const [testId, entry] of cases) {
+    let seconds = 0;
+    for (const phase of ["setup", "call", "teardown"] as const) {
+      const value = entry[phase]?.seconds;
+      if (value === null) invalid++;
+      else if (value !== undefined) { seconds += value; phaseSeconds[phase] += value; }
+    }
+    measured += seconds;
+    const caseFinished = !!(entry.teardown && (entry.call || entry.setup?.outcome !== "passed"));
+    if (caseFinished) finished++;
+    longest.push({ test_id: testId, display_nodeid: entry.display_nodeid,
+      measured_seconds: seconds, finished: caseFinished });
+  }
+  longest.sort((a, b) => b.measured_seconds - a.measured_seconds || a.test_id.localeCompare(b.test_id));
+  return {
+    execution_complete: ended && (exitCode === 0 || exitCode === 1) &&
+      selected !== null && selected > 0 && finished === selected && invalid === 0,
+    pytest_exit_code: exitCode,
+    selected_cases: selected,
+    observed_cases: cases.size,
+    finished_cases: finished,
+    invalid_duration_count: invalid,
+    observed_elapsed_seconds: elapsed,
+    measured_worker_seconds: measured,
+    phase_seconds: phaseSeconds,
+    longest_cases: longest.slice(0, 25),
+  };
+}
+
+if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
+  try {
+    if (process.argv.length !== 3) throw new Error("Usage: node scripts/ci/pytest_timing.ts JOB_LOG");
+    if (statSync(process.argv[2]).size > 32 * 1024 * 1024) throw new Error("Job log exceeds 32 MiB.");
+    process.stdout.write(JSON.stringify(summarizeTimingLog(readFileSync(process.argv[2], "utf8")), null, 2) + "\n");
+  } catch (error) {
+    process.stderr.write(`${error instanceof Error ? error.message : error}\n`);
+    process.exitCode = 1;
+  }
+}