ci: bound full Python shards across concurrent runs

.github/workflows/python-tests.yml

@@ -499,6 +499,13 @@ 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.
+    # 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:
+      group: loopx-python-test-shard-${{ matrix.shard }}
+      queue: max
+      cancel-in-progress: false
     # All four shards were still progressing at 83-94% when the 45m job
     # deadline cancelled run 37916979857. Reserve setup/reporting headroom
     # without changing test-level deadlines, collection or worker count.

docs/development/testing-and-quality.md

@@ -619,6 +619,31 @@ Sonar 只复用同一次 run 的 XML，不重复测试、不跨 run 取产物。
 Sonar，测试 job 不接收 Sonar secret。触发范围取原有两套 workflow 的并集；纯前端
 PR 使用前述豁免，Sonar 配置变更仍全量运行，包括没有 token 的 fork。
 
+Full Python shards share four 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
+candidate. Workflow-level cancellation still retires an obsolete head of the
+same PR. Queue overflow, cancellation and missing coverage remain failures of
+the required gate; collection, four 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
+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 组，适用于
+携带本版 workflow 的 PR、main、merge queue 和手动运行。每组允许一个运行中 job，
+用 `queue: max` 保留至多 100 个等待 job，不替换其他有效候选。workflow 级取消
+仍清退同 PR 的旧 head；队列溢出、取消和缺失 coverage 仍不能通过必需检查。
+完整 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
 run the complete suite locally. 全量本地测试仍省略分片参数即可。

scripts/ci/test_review_gate.py

@@ -2,9 +2,11 @@
 from __future__ import annotations
 import copy
 import json
+import os
 from pathlib import Path
 import re
 import subprocess
+import tempfile
 from types import SimpleNamespace
 import unittest
 import yaml
@@ -205,5 +207,74 @@ def test_backend_flag_cannot_waive_a_full_pr_or_create_client_coverage(self):
                     verifier(value)
 
 
+class PythonShardAdmissionTests(unittest.TestCase):
+    """Check the shipped workflow's admission scope and executable gate."""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.workflow = yaml.load(
+            (Path(__file__).resolve().parents[2] / ".github/workflows/python-tests.yml").read_text(),
+            Loader=yaml.BaseLoader,
+        )
+
+    def script(self, job, step):
+        return next(row["run"] for row in self.workflow["jobs"][job]["steps"]
+                    if row.get("name") == step)
+
+    def test_required_python_gate_rejects_cancelled_overflow_and_missing_shards(self):
+        script = self.script("pytest", "Require every upstream check")
+        for checks in ("success", "failure", "cancelled", "skipped", ""):
+            for shards in ("success", "failure", "cancelled", "skipped", ""):
+                result = subprocess.run(
+                    ["bash", "--noprofile", "--norc", "-eo", "pipefail", "-c", script],
+                    env={**os.environ, "CHECKS_RESULT": checks, "SHARDS_RESULT": shards},
+                    capture_output=True,
+                )
+                with self.subTest(checks=checks, shards=shards):
+                    self.assertEqual(result.returncode == 0,
+                                     checks == shards == "success")
+        gate = self.workflow["jobs"]["pytest"]
+        self.assertEqual(gate["if"], "always() && needs.changes.outputs.python_tests == 'true'")
+        self.assertEqual(gate["needs"], ["changes", "checks", "test-shard"])
+
+    def test_required_python_gate_reads_every_nonempty_coverage_partition(self):
+        script = self.script("pytest", "Combine complete coverage and enforce the existing floor")
+        # Execute the workflow's file guard, before the coverage tool itself.
+        guard = script.split("python -m coverage combine", 1)[0]
+        for absent in (None, 1, 2, 3, 4):
+            for empty in (False, True):
+                with tempfile.TemporaryDirectory() as directory:
+                    for shard in range(1, 5):
+                        path = Path(directory) / f"coverage-shards/python-coverage-{shard}/.coverage"
+                        path.parent.mkdir(parents=True)
+                        if shard != absent or empty:
+                            path.write_bytes(b"" if shard == absent else b"partition")
+                    result = subprocess.run(
+                        ["bash", "--noprofile", "--norc", "-eo", "pipefail", "-c", guard],
+                        cwd=directory, capture_output=True,
+                    )
+                    with self.subTest(absent=absent, empty=empty):
+                        self.assertEqual(result.returncode == 0, absent is None)
+
+    def test_long_shards_share_groups_across_current_heads_without_replacement(self):
+        job = self.workflow["jobs"]["test-shard"]
+        fields = job.get("concurrency", {})
+        self.assertEqual(fields.get("queue"), "max")
+        self.assertEqual(fields.get("cancel-in-progress"), "false")
+        template = fields["group"]
+        # A ref/run/PR component would silently restore per-run admission.
+        self.assertEqual(re.findall(r"\$\{\{\s*(.*?)\s*\}\}", template), ["matrix.shard"])
+        groups = {re.sub(r"\$\{\{.*?\}\}", str(shard), template).casefold()
+                  for shard in range(1, 5)}
+        self.assertEqual(len(groups), 4)
+        self.assertEqual(job["strategy"]["matrix"]["shard"], ["1", "2", "3", "4"])
+        self.assertEqual(job["strategy"]["fail-fast"], "false")
+        # GateTests executes the current workflow's PR lifecycle expressions,
+        # including closed events whose ref changes to main. Keep that semantic
+        # qualification instead of requiring the superseded ref-only spelling.
+        self.assertEqual(self.workflow["permissions"], {"contents": "read"})
+        self.assertNotIn("secrets.", json.dumps(job))
+
+
 if __name__ == "__main__":
     unittest.main()