ci: expose test stall stacks and bound fixture progress IDs

.github/workflows/python-tests.yml

@@ -570,7 +570,7 @@ jobs:
     needs: [changes, chat-bundle]
     if: needs.changes.outputs.python_tests == 'true'
     runs-on: ubuntu-latest
-    # Share four long-test lanes across current PRs and other workflow runs.
+    # Share six long-test lanes across current PRs and other workflow runs.
     # Do not replace another candidate's pending shard or cancel its active head.
     # Workflow-level concurrency still retires obsolete heads of the same PR.
     concurrency:
@@ -628,9 +628,11 @@ jobs:
         # Split the whole collection, not a hand-maintained list of directories.
         # Without timing history least_duration alternates equal-weight tests.
         # Each runner retains the measured two-worker pool.
+        # Cooperative per-test Python stacks diagnose stalls without terminating tests.
         run: >-
           python -m pytest -v -n 2 -m "not stage2c_e2e"
           -p scripts.ci.pytest_timing
+          --ci-stall-after 120
           --splits 6 --group ${{ matrix.shard }}
           --splitting-algorithm least_duration
           --durations=25 --durations-min=1

docs/development/testing-and-quality.md

@@ -649,7 +649,7 @@ network latency, provider availability, or a two-hour matrix.
 它刻意不包含真实模型调用和 full smoke catalog，因此普通迭代不依赖凭证、网络
 时延、模型服务可用性或两小时级测试矩阵。
 
-The Linux suite uses four hosted runners with two xdist workers each.
+The Linux suite uses six hosted runners with two xdist workers each.
 `pytest-split` partitions the complete collection using `least_duration`;
 without a timing file, tests have equal weight and alternate between shards.
 Lint, type checks, and the CLI budget run separately. The required `pytest`
@@ -664,13 +664,28 @@ 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).
+Full Python shards also opt into `--ci-stall-after 120` on the existing pytest
+observer: a long-running test attempts to print its executing thread's Python stack without
+terminating it or changing its result. This cooperative diagnostic needs the
+interpreter to schedule its observer thread; it does not diagnose native code
+that holds the GIL indefinitely. The dump is diagnostic evidence, not proof that the last named test
+caused a stall. Reproduce with `-p scripts.ci.pytest_timing --ci-stall-after 120`; omit
+`--ci-stall-after` or set it to `0` to keep this default-off diagnostic disabled. This grants no authority to
+inspect other processes. Large text-boundary fixtures use short parameter IDs
+so verbose progress records their cases without printing the payload.
+Each triggered snapshot uses an isolated same-interpreter writer with a 100ms
+self-exit deadline, plus a 500ms parent communication timeout with kill/reap.
+This bounds pipe backpressure without changing shared stderr flags or leaving
+a persistently blocked writer after pytest exits; native process startup can add latency.
+Snapshots may be dropped on IO failure or backpressure. Default `0` starts no
+observer timer or writer process; pytest's own results and reports remain authoritative.
 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
 the former Python and Sonar paths. Client-only PRs use the exemption above;
 Sonar-configuration changes remain full, including on forks without a token.
 
-Linux 全套测试分到四台 hosted runner，每台保留两个 xdist worker。`pytest-split`
+Linux 全套测试分到六台 hosted runner，每台保留两个 xdist worker。`pytest-split`
 按完整 collection 分片；没有历史耗时时，等权测试交替分配。lint、类型检查和 CLI
 预算独立执行。必需的 `pytest` 汇总检查会拒绝失败／跳过的分片和缺失的 coverage，
 合并后再执行原有 19.6% 门槛；不要求单个分片达到全套覆盖率。coverage 使用相对路径，
@@ -680,11 +695,21 @@ CI 同时在 xdist controller 流式记录 pytest 原生 phase 耗时。下载
 `node scripts/ci/pytest_timing.ts JOB_LOG` 回读；不完整证据保持不完整，不能替代
 JUnit、coverage 或必过门禁。归属与测量边界见
 [分片耗时记录](ci-impact-selection.md#four-complete-python-shards)。
+全量 Python 分片在既有 pytest observer 上显式启用 `--ci-stall-after 120`：测试
+超过该时长就尝试输出其执行线程的 Python 栈，不终止测试或改变结果。此协作式诊断
+需要解释器调度观察线程，无法诊断始终占有 GIL 的原生代码。栈用于诊断，不能只凭
+最后一个测试名认定卡顿根因。本地加 `-p scripts.ci.pytest_timing --ci-stall-after 120`
+复现，省略此参数或设为 `0` 可保持默认关闭；它不授予检查其他进程的权限。大文本边界 fixture 使用简短参数 ID，详细进度保留用例身份，
+不打印整个载荷。
+每次触发栈输出时，使用同解释器的隔离 writer，100ms 自行退出；父进程另设
+500ms 通信期限，超时则终止并回收。管道背压不会改变共享 stderr 标志或留下
+持续阻塞的 writer；原生进程启动仍可能增加延迟。IO 失败或背压时可丢弃栈输出。
+默认 `0` 不启动观察 timer 或 writer 进程；pytest 原生结果和报告仍是权威。
 Sonar 只复用同一次 run 的 XML，不重复测试、不跨 run 取产物。缺少 token 仍成功跳过
 Sonar，测试 job 不接收 Sonar secret。触发范围取原有两套 workflow 的并集；纯前端
 PR 使用前述豁免，Sonar 配置变更仍全量运行，包括没有 token 的 fork。
 
-Full Python shards share four repository-wide job concurrency groups, one per
+Full Python shards share six repository-wide job concurrency groups, one per
 `matrix.shard`, across PR, main, merge-queue and manual runs carrying this
 workflow version. Each group admits one running job and uses `queue: max`
 to retain up to 100 pending jobs instead of replacing another current
@@ -699,16 +724,16 @@ unchanged: cancelled, skipped or incomplete dependencies cannot become a green
 aggregate. This adds one short trusted cleanup job per update without checkout,
 PR-code execution or new permissions on qualification jobs. Older workflow
 versions on the base branch retain only closed-event cleanup. Queue overflow, cancellation and missing coverage remain failures of
-the required gate; collection, four partitions, two workers, reports and the
+the required gate; collection, six partitions, two workers, reports and the
 coverage floor stay unchanged. Contributors can wait longer for a shard during
-overlapping full runs. This limits long-shard occupancy to four runners; it
+overlapping full runs. This limits long-shard occupancy to six runners; it
 does not reserve release capacity or prioritize releases over other jobs or
 repositories in the organization's shared hosted pool. Older workflow versions
 do not participate. Roll back by removing this job-level concurrency block,
 retaining workflow-level cancellation and the required gate. See GitHub's
 [concurrency queue contract](https://docs.github.com/en/actions/how-tos/write-workflows/choose-when-workflows-run/control-workflow-concurrency).
 
-全量 Python 分片按 `matrix.shard` 共用四个仓库级 job concurrency 组，适用于
+全量 Python 分片按 `matrix.shard` 共用六个仓库级 job concurrency 组，适用于
 携带本版 workflow 的 PR、main、merge queue 和手动运行。每组允许一个运行中 job，
 用 `queue: max` 保留至多 100 个等待 job，不替换其他有效候选。workflow 级取消
 请求清退同 PR 的旧 head；可信 base 分支的 `synchronize` 清理还会取消遗留
@@ -718,14 +743,14 @@ PR、当前 head 和无关联 run 均被保护。必过汇总条件不变，取
 不能变绿。每次更新增加一个短清理 job，不 checkout 或执行 PR 代码，不增加普通
 验证 job 的权限；base 分支旧版 workflow 仍只清理 closed 事件。队列溢出、取消
 和缺失 coverage 仍不能通过必需检查。
-完整 collection、四分片、双 worker、报告和覆盖率门槛保持不变；重叠的全量运行
-会增加贡献者等待分片的时间。此限制只把长分片占用约束到四台 runner，不预留发布
+完整 collection、六分片、双 worker、报告和覆盖率门槛保持不变；重叠的全量运行
+会增加贡献者等待分片的时间。此限制只把长分片占用约束到六台 runner，不预留发布
 容量，也不保证发布优先于组织共享 hosted 池中的其他 job 或仓库。旧版 workflow
 不参与该限制。回退时移除此 job 的 concurrency 块，保留 workflow 级取消和必需
 检查；队列行为遵循上述 GitHub 官方契约。
 
-Reproduce one shard locally with `uv run --extra test python -m pytest -v -n 2 --splits 4 --group 1
---splitting-algorithm least_duration --cov=loopx`. Omit the split arguments to
+Reproduce one shard locally with `uv run --extra test python -m pytest -v -n 2 --splits 6 --group 1
+--splitting-algorithm least_duration -p scripts.ci.pytest_timing --ci-stall-after 120 --cov=loopx`. Omit the split arguments to
 run the complete suite locally. 全量本地测试仍省略分片参数即可。
 
 ## Smokes And Canary / Smoke 与 Canary

scripts/ci/pytest_timing.py

@@ -1,16 +1,22 @@
 """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.
+Load explicitly with ``-p scripts.ci.pytest_timing``. Optional cooperative
+stacks do not change test selection or results; timing values come from pytest.
 """
 from __future__ import annotations
 
 import json
 import hashlib
 import math
+import os
 import re
+import subprocess
 import time
+import sys
+import threading
+from collections.abc import Generator, Sequence
+from typing import Any
 
 import pytest
 from _pytest.subtests import SubtestReport
@@ -20,10 +26,66 @@
 
 
 PREFIX = "LOOPX_PYTEST_TIMING "
+_STACK_FD = pytest.StashKey[int]()
+_STACK_WRITE_PROGRAM = (
+    "import os,sys,threading;"
+    "deadline=threading.Timer(0.1,lambda:os._exit(0));"
+    "deadline.daemon=True;deadline.start();"
+    "os.write(2,sys.stdin.buffer.read())"
+)
+
+
+def pytest_addoption(parser: pytest.Parser) -> None:
+    parser.addoption("--ci-stall-after", type=float, default=0,
+                     help="Print the executing test thread's Python stack after this many seconds; never stop the test.")
+
+
+@pytest.hookimpl(wrapper=True, trylast=True)
+def pytest_runtest_protocol(item: pytest.Item) -> Generator[None, object, object]:
+    delay = item.config.getoption("ci_stall_after")
+    if not delay:
+        return (yield)
+    owner_thread = threading.get_ident()
+    finished = threading.Event()
+
+    def dump() -> None:
+        # Strong frame references avoid asynchronous native frame traversal.
+        # Only this test's executing thread is inspected; no locals are read.
+        frame = sys._current_frames().get(owner_thread)
+        rows = ["LOOPX_PYTEST_STACK " + item.nodeid.partition("[")[0][:256]]
+        while frame is not None and len(rows) <= 24:
+            rows.append(f"  {frame.f_code.co_filename}:{frame.f_lineno} in {frame.f_code.co_name}")
+            frame = frame.f_back
+        del frame
+        if not finished.is_set():
+            try:
+                # A duplicate shares stderr's flags: never change them or
+                # write a potentially full pipe in this process. The isolated
+                # writer has its own exit deadline, even if pytest is killed;
+                # run's timeout also kills/reaps it before releasing this FD.
+                subprocess.run(
+                    [sys.executable, "-I", "-S", "-c", _STACK_WRITE_PROGRAM],
+                    input=("\n" + "\n".join(rows) + "\n").encode("utf-8", errors="backslashreplace"),
+                    stdout=subprocess.DEVNULL, stderr=item.config.stash[_STACK_FD],
+                    timeout=0.5, check=False,
+                )
+            except (OSError, subprocess.TimeoutExpired):
+                # Optional diagnostics cannot replace pytest's result on IO failure.
+                pass
+
+    timer = threading.Timer(delay, dump)
+    timer.daemon = True
+    timer.start()
+    try:
+        return (yield)
+    finally:
+        finished.set()
+        timer.cancel()
+        timer.join()
 
 
 @pytest.hookimpl(wrapper=True)
-def pytest_runtest_makereport(item, call):
+def pytest_runtest_makereport(item: pytest.Item, call: Any) -> Generator[None, Any, Any]:
     report = yield
     if call.excinfo is not None:
         try:
@@ -34,20 +96,39 @@ def pytest_runtest_makereport(item, call):
     return report
 
 
-def pytest_configure(config):
+def pytest_configure(config: pytest.Config) -> None:
+    delay = config.getoption("ci_stall_after")
+    if not math.isfinite(delay) or delay < 0:
+        raise pytest.UsageError("--ci-stall-after must be finite and nonnegative")
+    if delay:
+        # Capture the original descriptor before per-test fd capture starts,
+        # so a blocked worker's diagnostic is visible before it returns.
+        if sys.__stderr__ is None:
+            raise pytest.UsageError("--ci-stall-after requires an original stderr file descriptor")
+        try:
+            config.stash[_STACK_FD] = os.dup(sys.__stderr__.fileno())
+        except (AttributeError, OSError, ValueError) as error:
+            raise pytest.UsageError("--ci-stall-after requires an original stderr file descriptor") from error
     # 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")
 
 
+def pytest_unconfigure(config: pytest.Config) -> None:
+    descriptor = config.stash.get(_STACK_FD, None)
+    if descriptor is not None:
+        os.close(descriptor)
+        del config.stash[_STACK_FD]
+
+
 class TimingObserver:
-    def __init__(self, config):
+    def __init__(self, config: pytest.Config) -> None:
         self.config = config
         self.started = time.perf_counter()
-        self.collection = None
+        self.collection: tuple[str, ...] | None = None
 
-    def emit(self, *, _prefix=PREFIX, **row):
+    def emit(self, *, _prefix: str = PREFIX, **row: Any) -> None:
         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")
@@ -61,10 +142,10 @@ def emit(self, *, _prefix=PREFIX, **row):
             reporter._tw.flush()
 
     @pytest.hookimpl(trylast=True)
-    def pytest_sessionstart(self, session):
+    def pytest_sessionstart(self, session: pytest.Session) -> None:
         self.emit(event="start", version=1)
 
-    def record_collection(self, ids):
+    def record_collection(self, ids: Sequence[str]) -> None:
         if self.collection is None:
             self.collection = tuple(ids)
             self.emit(event="collection", selected=len(ids))
@@ -74,14 +155,14 @@ def record_collection(self, ids):
             self.emit(event="collection_mismatch")
 
     @pytest.hookimpl(optionalhook=True)
-    def pytest_xdist_node_collection_finished(self, node, ids):
+    def pytest_xdist_node_collection_finished(self, node: Any, ids: Sequence[str]) -> None:
         self.record_collection(ids)
 
-    def pytest_collection_finish(self, session):
+    def pytest_collection_finish(self, session: pytest.Session) -> None:
         if not getattr(self.config.option, "numprocesses", None):
             self.record_collection([item.nodeid for item in session.items])
 
-    def pytest_runtest_logreport(self, report):
+    def pytest_runtest_logreport(self, report: Any) -> None:
         test_id = hashlib.sha256(report.nodeid.encode("utf-8", errors="surrogatepass")).hexdigest()
         display = report.nodeid.partition("[")[0][:256]
         if report.failed:
@@ -108,5 +189,5 @@ def pytest_runtest_logreport(self, report):
                   seconds=duration if math.isfinite(duration) and duration >= 0 else None)
 
     @pytest.hookimpl(trylast=True)
-    def pytest_sessionfinish(self, session, exitstatus):
+    def pytest_sessionfinish(self, session: pytest.Session, exitstatus: int) -> None:
         self.emit(event="end", exit_code=int(session.exitstatus))

tests/extensions/test_lark_reply_handoff.py

@@ -187,7 +187,8 @@ def sender(original, returned_session, returned_turn, text):
         controller.close()
 
 
-@pytest.mark.parametrize("source", ["字" * 32000, "😀" * 24000, "\u0000" * 16000])
+@pytest.mark.parametrize("source", ["字" * 32000, "😀" * 24000, "\u0000" * 16000],
+                         ids=["cjk", "emoji", "escaped-control"])
 def test_source_context_encoded_boundary_still_reads_the_real_ingress_record(tmp_path: Path, source: str) -> None:
     register_ingress(tmp_path, session_id="session", client_turn_id="turn", channel="manager.external.fixture",
                      sender_id="owner", message="Current request", source_id="fixture:current", source_message=source)
@@ -196,7 +197,8 @@ def test_source_context_encoded_boundary_still_reads_the_real_ingress_record(tmp
     assert _read(path)["source_message"] == source
 
 
-@pytest.mark.parametrize("source", ["x" * 32001, "😀" * 26000, "\u0000" * 17000])
+@pytest.mark.parametrize("source", ["x" * 32001, "😀" * 26000, "\u0000" * 17000],
+                         ids=["ascii", "emoji", "escaped-control"])
 def test_oversized_source_is_rejected_before_any_ingress_record(tmp_path: Path, source: str) -> None:
     with pytest.raises(ValueError, match="source context exceeds.*scoped artifact"):
         register_ingress(tmp_path, session_id="session", client_turn_id="turn", channel="manager.external.fixture",

tests/extensions/test_lark_reply_size.py

@@ -8,13 +8,15 @@
 )
 
 
-@pytest.mark.parametrize("value", ["x" * 149992, "测" * 49997, "😀" * 37498])
+@pytest.mark.parametrize("value", ["x" * 149992, "测" * 49997, "😀" * 37498],
+                         ids=["ascii", "cjk", "emoji"])
 def test_provider_byte_budget_accepts_large_text(value):
     # {"x":""} costs eight bytes; UTF-8 code points have different costs.
     validate_lark_text_request_size({"x": value})
 
 
-@pytest.mark.parametrize("value", ["x" * 149993, "测" * 49998, "😀" * 37499])
+@pytest.mark.parametrize("value", ["x" * 149993, "测" * 49998, "😀" * 37499],
+                         ids=["ascii", "cjk", "emoji"])
 def test_provider_byte_budget_rejects_oversize_utf8(value):
     with pytest.raises(LarkOutboundTextError):
         validate_lark_text_request_size({"x": value})