diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml
index c944b7ab8c..8314a6549c 100644
--- a/.github/workflows/python-tests.yml
+++ b/.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
diff --git a/docs/development/ci-impact-selection.md b/docs/development/ci-impact-selection.md
index 0aef2015c9..d6a532308b 100644
--- a/docs/development/ci-impact-selection.md
+++ b/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
diff --git a/docs/development/testing-and-quality.md b/docs/development/testing-and-quality.md
index 92309efe35..7beecb51a9 100644
--- a/docs/development/testing-and-quality.md
+++ b/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。
diff --git a/scripts/ci/pytest_timing.py b/scripts/ci/pytest_timing.py
new file mode 100644
index 0000000000..03ca332a47
--- /dev/null
+++ b/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))
diff --git a/scripts/ci/pytest_timing.ts b/scripts/ci/pytest_timing.ts
new file mode 100644
index 0000000000..b28946abb9
--- /dev/null
+++ b/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;
+  }
+}
diff --git a/tests/test_ci_pytest_timing.py b/tests/test_ci_pytest_timing.py
new file mode 100644
index 0000000000..0ba61f2120
--- /dev/null
+++ b/tests/test_ci_pytest_timing.py
@@ -0,0 +1,255 @@
+"""Real pytest observations survive failure/interruption without qualifying CI."""
+from __future__ import annotations
+
+import json
+import hashlib
+import os
+from pathlib import Path
+import signal
+import subprocess
+import sys
+import time
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+READER = ROOT / "scripts/ci/pytest_timing.ts"
+PREFIX = "LOOPX_PYTEST_TIMING "
+
+
+def environment():
+    env = {key: value for key, value in os.environ.items()
+           if not key.startswith(("PYTEST", "COV_CORE", "COVERAGE"))}
+    env["PYTHONPATH"] = str(ROOT)
+    return env
+
+
+def read_log(path, *, success=True):
+    result = subprocess.run(["node", str(READER), str(path)], capture_output=True,
+                            text=True, timeout=10, check=False)
+    assert (result.returncode == 0) == success, result.stderr
+    return json.loads(result.stdout) if success else result.stderr
+
+
+@pytest.mark.parametrize("verbosity", ["-v", "-q"])
+@pytest.mark.parametrize("workers", [0, 2])
+def test_real_xdist_timings_match_native_pytest_owner(tmp_path, verbosity, workers):
+    (tmp_path / "test_subject.py").write_text(
+        "import time, pytest\n"
+        "def test_pass(): time.sleep(0.03)\n"
+        "def test_fail(): assert False\n"
+        "@pytest.mark.skip(reason='fixture')\n"
+        "def test_skip(): pass\n"
+        "@pytest.fixture\n"
+        "def bad(): raise ValueError('fixture')\n"
+        "def test_setup_error(bad): pass\n"
+        "import unittest\n"
+        "class TestNested(unittest.TestCase):\n"
+        "    def test_passed_subtests(self):\n"
+        "        for value in (1, 2):\n"
+        "            with self.subTest(value=value): self.assertGreater(value, 0)\n"
+        "    def test_failed_subtests(self):\n"
+        "        for value in (1, -1):\n"
+        "            with self.subTest(value=value): self.assertGreater(value, 0)\n",
+        encoding="utf-8")
+    log = tmp_path / "job.log"
+    with log.open("w") as output:
+        result = subprocess.run(
+            [sys.executable, "-m", "pytest", verbosity, "-n", str(workers),
+             "-p", "scripts.ci.pytest_timing", "--store-durations", "--clean-durations",
+             "--durations-path", "native.json", "--junitxml=junit.xml"],
+            cwd=tmp_path, env=environment(), stdout=output, stderr=subprocess.STDOUT,
+            timeout=30, check=False)
+    assert result.returncode == 1
+    assert all(line.startswith(PREFIX) for line in log.read_text().splitlines()
+               if PREFIX in line)
+    report = read_log(log)
+    # A completed traversal with test failures is not successful qualification.
+    assert report["execution_complete"] is True
+    assert report["pytest_exit_code"] == 1
+    assert report["selected_cases"] == report["finished_cases"] == 6
+    assert all(case["finished"] for case in report["longest_cases"])
+    assert report["invalid_duration_count"] == 0
+    native = json.loads((tmp_path / "native.json").read_text())
+    assert report["measured_worker_seconds"] == pytest.approx(sum(native.values()), abs=1e-6)
+    assert report["measured_worker_seconds"] >= 0.03
+    # Actual GitHub timestamps/ANSI and CRLF do not change the observation.
+    lines = log.read_text().splitlines()
+    timestamped = tmp_path / "hosted-format.log"
+    timestamped.write_bytes("\r\n".join(
+        "2026-01-01T00:00:00.0000000Z \x1b[32m" + line + "\x1b[0m"
+        for line in lines).encode())
+    assert read_log(timestamped) == report
+    # GitHub concatenates log chunks with a BOM before some timestamps.
+    # It is transport framing; the observation and its strict ordering match.
+    chunked = tmp_path / "hosted-chunks.log"
+    chunked.write_text("\n".join(
+        ("\ufeff" if index % 3 == 0 else "") + line
+        for index, line in enumerate(timestamped.read_text().splitlines())), encoding="utf-8")
+    assert read_log(chunked) == report
+
+
+def test_real_pytest_subtest_cost_is_included_once_in_parent_call(tmp_path):
+    (tmp_path / "test_subject.py").write_text(
+        "import time, pytest\n"
+        "@pytest.mark.parametrize('fail', [False, True])\n"
+        "def test_nested(subtests, fail):\n"
+        "    for value in (1, 2):\n"
+        "        with subtests.test(value=value):\n"
+        "            time.sleep(0.02)\n"
+        "            assert not (fail and value == 2)\n", encoding="utf-8")
+    # Independent native serialization records both parent and nested reports.
+    # SubtestReport supplies its own discriminator; no observer output is the oracle.
+    (tmp_path / "native_reports.py").write_text(
+        "import json\nfrom pathlib import Path\nreports = []\n"
+        "def pytest_runtest_logreport(report):\n"
+        "    reports.append({key: value for key, value in report._to_json().items()\n"
+        "                    if key in ('duration', 'when', 'outcome', '_report_type')})\n"
+        "def pytest_sessionfinish(session):\n"
+        "    if not hasattr(session.config, 'workerinput'):\n"
+        "        Path('native-reports.json').write_text(json.dumps(reports))\n",
+        encoding="utf-8")
+    log = tmp_path / "subtests.log"
+    with log.open("w") as output:
+        result = subprocess.run(
+            [sys.executable, "-m", "pytest", "-v", "-n", "2", "-p", "native_reports",
+             "-p", "scripts.ci.pytest_timing"], cwd=tmp_path, env=environment(),
+            stdout=output, stderr=subprocess.STDOUT, timeout=30, check=False)
+    assert result.returncode == 1
+    native = json.loads((tmp_path / "native-reports.json").read_text())
+    nested = [row for row in native if row.get("_report_type") == "SubTestReport"]
+    parents = [row for row in native if "_report_type" not in row]
+    assert len(nested) == 4 and sum(row["duration"] for row in nested) >= 0.08
+    assert len(parents) == 6  # three phases for each of two collected items
+    report = read_log(log)
+    assert report["execution_complete"] and report["pytest_exit_code"] == 1
+    assert report["selected_cases"] == report["finished_cases"] == 2
+    assert report["measured_worker_seconds"] == pytest.approx(
+        sum(row["duration"] for row in parents), abs=1e-6)
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX process-group kill; hosted shards are Linux")
+@pytest.mark.parametrize("verbosity", ["-v", "-q"])
+def test_killed_controller_leaves_partial_readable_native_observations(tmp_path, verbosity):
+    (tmp_path / "test_subject.py").write_text(
+        "import time\n"
+        "def test_first(): time.sleep(0.03)\n"
+        "def test_wait(): time.sleep(30)\n", encoding="utf-8")
+    log = tmp_path / "killed.log"
+    with log.open("w") as output:
+        process = subprocess.Popen(
+            [sys.executable, "-m", "pytest", verbosity, "-n", "2",
+             "-p", "scripts.ci.pytest_timing", "--junitxml=junit.xml"],
+            cwd=tmp_path, env=environment(), stdout=output, stderr=subprocess.STDOUT,
+            start_new_session=True)
+        try:
+            deadline = time.monotonic() + 10
+            while time.monotonic() < deadline:
+                rows = [json.loads(line.removeprefix(PREFIX))
+                        for line in log.read_text().splitlines() if line.startswith(PREFIX)]
+                if any(row.get("phase") == "teardown" for row in rows):
+                    break
+                assert process.poll() is None, log.read_text()
+                time.sleep(0.02)
+            else:
+                pytest.fail("No completed test timing before interruption")
+        finally:
+            if process.poll() is None:
+                os.killpg(process.pid, signal.SIGKILL)
+            process.wait(timeout=10)
+    report = read_log(log)
+    assert report["execution_complete"] is False
+    assert report["pytest_exit_code"] is None
+    assert report["selected_cases"] == 2
+    assert report["finished_cases"] == 1
+    assert report["measured_worker_seconds"] >= 0.03
+    assert not (tmp_path / "junit.xml").exists()
+
+
+def rows_log(tmp_path, rows):
+    log = tmp_path / "records.log"
+    log.write_text("\n".join(PREFIX + json.dumps({"elapsed_seconds": index, **row})
+                             for index, row in enumerate(rows)), encoding="utf-8")
+    return log
+
+
+START = {"event": "start", "version": 1}
+COLLECTION = {"event": "collection", "selected": 1}
+SETUP = {"event": "phase", "test_id": hashlib.sha256(b"test.py::test_example").hexdigest(),
+         "display_nodeid": "test.py::test_example", "phase": "setup",
+         "outcome": "passed", "seconds": 0.1}
+CALL = {**SETUP, "phase": "call", "seconds": 2}
+TEARDOWN = {**SETUP, "phase": "teardown", "seconds": 0.2}
+END = {"event": "end", "exit_code": 0}
+
+
+@pytest.mark.parametrize("rows", [
+    [], [COLLECTION], [START, START], [START, COLLECTION, SETUP, SETUP],
+    [START, COLLECTION, SETUP, CALL, CALL],
+    [START, COLLECTION, CALL], [START, COLLECTION, SETUP, {**CALL, "seconds": -1}],
+    [START, COLLECTION, {**SETUP, "phase": "unknown"}],
+    [START, COLLECTION, {**SETUP, "outcome": "unknown"}],
+    [START, COLLECTION, {**SETUP, "seconds": "2"}],
+    [START, COLLECTION, {**SETUP, "test_id": "unknown"}],
+    [START, COLLECTION, SETUP, {**CALL, "display_nodeid": "changed"}],
+    [START, {**COLLECTION, "selected": 0}, SETUP],
+    [START, {"event": "collection_mismatch"}],
+    [START, COLLECTION, END, SETUP], [START, {**END, "exit_code": 6}],
+])
+def test_malformed_or_contradictory_evidence_is_rejected(tmp_path, rows):
+    log = rows_log(tmp_path, rows)
+    read_log(log, success=False)
+    log.write_text("\n".join("\ufeff2026-01-01T00:00:00.0000000Z " + line
+                             for line in log.read_text().splitlines()), encoding="utf-8")
+    read_log(log, success=False)
+
+
+@pytest.mark.parametrize("rows,finished", [
+    ([START], 0), ([START, COLLECTION, SETUP], 0),
+    ([START, COLLECTION, SETUP, TEARDOWN, END], 0),
+    ([START, COLLECTION, SETUP, CALL, TEARDOWN, {**END, "exit_code": 2}], 1),
+    ([START, COLLECTION, SETUP, {**CALL, "seconds": None}, TEARDOWN, END], 1),
+    ([START, {**COLLECTION, "selected": 0}, {**END, "exit_code": 5}], 0),
+    ([START, {**COLLECTION, "selected": 2}, SETUP, CALL, TEARDOWN, END], 1),
+])
+def test_incomplete_or_invalid_timings_never_claim_complete_execution(tmp_path, rows, finished):
+    report = read_log(rows_log(tmp_path, rows))
+    assert report["execution_complete"] is False
+    assert report["finished_cases"] == finished
+    assert sum(case["finished"] for case in report["longest_cases"]) == finished
+
+
+def test_diagnostic_cli_rejects_missing_oversized_or_corrupt_inputs(tmp_path):
+    read_log(tmp_path / "missing.log", success=False)
+    log = tmp_path / "large.log"
+    with log.open("wb") as output:
+        output.truncate(32 * 1024 * 1024 + 1)
+    assert "32 MiB" in read_log(log, success=False)
+    log.write_text(PREFIX + "{broken", encoding="utf-8")
+    read_log(log, success=False)
+
+
+def test_real_oversized_parameter_ids_keep_distinct_identity_and_bounded_display(tmp_path):
+    (tmp_path / "test_subject.py").write_text(
+        "import pytest\n@pytest.mark.parametrize('value', ['x'*20000, 'x'*20000+'y',\n"
+        "    'curl -H \\\"Authorization: Bearer $TOKEN\\\"'])\n"
+        "def test_long(value): assert value\n", encoding="utf-8")
+    log = tmp_path / "long.log"
+    with log.open("w") as output:
+        subprocess.run(
+            [sys.executable, "-m", "pytest", "-v", "-n", "2", "-p", "scripts.ci.pytest_timing",
+             "--store-durations", "--clean-durations", "--durations-path", "native.json"],
+            cwd=tmp_path, env=environment(), stdout=output, stderr=subprocess.STDOUT,
+            timeout=30, check=True)
+    report = read_log(log)
+    assert report["execution_complete"] and report["finished_cases"] == 3
+    native = json.loads((tmp_path / "native.json").read_text())
+    assert {case["test_id"] for case in report["longest_cases"]} == {
+        hashlib.sha256(nodeid.encode()).hexdigest() for nodeid in native}
+    assert len({case["display_nodeid"] for case in report["longest_cases"]}) == 1
+    assert all(case["display_nodeid"].endswith("test_subject.py::test_long")
+               for case in report["longest_cases"])
+    assert all("Authorization" not in line and "$TOKEN" not in line
+               for line in log.read_text().splitlines() if line.startswith(PREFIX))
+    assert max(len(line) for line in log.read_text().splitlines() if line.startswith(PREFIX)) < 1200
diff --git a/tests/test_python_ci_workflow.py b/tests/test_python_ci_workflow.py
index 8f4174c052..a1e37eb8ad 100644
--- a/tests/test_python_ci_workflow.py
+++ b/tests/test_python_ci_workflow.py
@@ -306,6 +306,7 @@ def test_four_shards_execute_each_test_once_and_merge_portable_coverage(
         key: value for key, value in os.environ.items()
         if not key.startswith(("COVERAGE", "COV_CORE", "PYTEST"))
     }
+    env["PYTHONPATH"] = str(WORKFLOW_ROOT)
     seen: list[set[str]] = []
     for shard in (1, 2, 3, 4):
         root = tmp_path / f"checkout-{shard}"
@@ -404,6 +405,7 @@ def test_shard_names_failed_cases_before_the_remaining_tests_finish(tmp_path: Pa
     )
     env = {key: value for key, value in os.environ.items()
            if not key.startswith(("COVERAGE", "COV_CORE", "PYTEST"))}
+    env["PYTHONPATH"] = str(WORKFLOW_ROOT)
     log = tmp_path / "progress.log"
     with log.open("w") as output:
         process = subprocess.Popen(args, cwd=tmp_path, env=env, stdout=output, stderr=subprocess.STDOUT)