diff --git a/docs/development/ci-impact-selection.md b/docs/development/ci-impact-selection.md
index dfa915d82a..3d6fef84a8 100644
--- a/docs/development/ci-impact-selection.md
+++ b/docs/development/ci-impact-selection.md
@@ -170,7 +170,18 @@ full qualification. A PR is classified from its merge base through its complete
 head, including changes from earlier commits, not just its latest commit.
 
 The `ci-impact-plan` artifact and job summary report exact revisions, change kind,
-per-job execution flags, reason and coverage scope. The merge gate requires
+per-job execution flags, reason and coverage scope. `python_shards` reads the
+static `test-shard.strategy.matrix.shard` axis from the immutable workflow at
+`checkout_sha`, currently six, rather than retaining a separate historical count.
+Unsupported or noncontiguous topology fails classification instead of publishing
+a fallback count. Docs/presentation exemptions still report zero. The number
+describes the workflow topology, not how many jobs actually ran.
+
+`python_shards` 从 `checkout_sha` 对应的不可变 workflow 读取静态分片轴，当前为六片，
+不另存历史计数；不支持或不连续的拓扑直接报错，不输出猜测值。文档／前端豁免仍为零。
+它描述 workflow 拓扑，不能证明任何分片实际执行或 JUnit／coverage 已完整。
+
+The merge gate requires
 success for required jobs and an explicit skip for exempt ones; failure,
 cancellation, missing outputs, contradictory flags or unexpected skips fail.
 
diff --git a/scripts/ci/impact_plan.py b/scripts/ci/impact_plan.py
index b1e451afda..4992b47e26 100644
--- a/scripts/ci/impact_plan.py
+++ b/scripts/ci/impact_plan.py
@@ -78,6 +78,25 @@ def trusted_policy(base: str, head: str) -> bool:
     return True
 
 
+def declared_python_shards(checkout_sha: str) -> int:
+    # Only classification needs the parser already installed in the changes
+    # job. Aggregate verification must remain usable with the standard library.
+    import yaml
+
+    workflow = yaml.load(git("show", f"{checkout_sha}:{POLICY_PATHS[-1]}"), Loader=yaml.BaseLoader)
+    message = "test-shard.strategy.matrix must declare only a static contiguous shard axis [1..N]"
+    try:
+        matrix = workflow["jobs"]["test-shard"]["strategy"]["matrix"]
+    except (KeyError, TypeError):
+        raise ValueError(message) from None
+    if not isinstance(matrix, dict) or set(matrix) != {"shard"}:
+        raise ValueError(message)
+    shards = matrix["shard"]
+    if not isinstance(shards, list) or not shards or shards != [str(i) for i in range(1, len(shards) + 1)]:
+        raise ValueError(message)
+    return len(shards)
+
+
 def plan(base: str, head: str, *, pull_request: bool = True, force_full: bool = False) -> dict:
     base_sha, head_sha = revision(base), revision(head)
     merge_base = git("merge-base", base_sha, head_sha).decode().strip()
@@ -96,12 +115,13 @@ def plan(base: str, head: str, *, pull_request: bool = True, force_full: bool =
     elif kind == "presentation" and not trusted_policy(base_sha, head_sha):
         kind, reason = "full", "presentation exemption policy is not already trusted on the target branch"
     flags = job_flags(kind, presentation=presentation or policy_change)
+    checkout_sha = revision("HEAD")
     return {"schema_version": SCHEMA, "base_sha": base_sha, "head_sha": head_sha,
-            "merge_base_sha": merge_base, "checkout_sha": revision("HEAD"), "change_kind": kind,
+            "merge_base_sha": merge_base, "checkout_sha": checkout_sha, "change_kind": kind,
             "force_full": force_full, "reason": reason, **flags,
             "changes": [{"status": item.status, "path": item.path} for item in changes],
             "coverage_scope": "full Python suite" if flags["python_tests"] else "no Python coverage claim",
-            "python_shards": 4 if flags["python_tests"] else 0}
+            "python_shards": declared_python_shards(checkout_sha) if flags["python_tests"] else 0}
 
 
 def write_plan(packet: dict, output: str) -> None:
diff --git a/scripts/ci/test_impact_plan.py b/scripts/ci/test_impact_plan.py
index 069be0af5c..f52b2353ce 100644
--- a/scripts/ci/test_impact_plan.py
+++ b/scripts/ci/test_impact_plan.py
@@ -55,8 +55,8 @@ def write(path, content):
                 target = root / path
                 target.parent.mkdir(parents=True, exist_ok=True)
                 target.write_text(content)
-            def classify(base, *extra):
-                result = subprocess.run([sys.executable, script, "classify", "--base", base, "--head", "HEAD",
+            def classify(base, *extra, head="HEAD"):
+                result = subprocess.run([sys.executable, script, "classify", "--base", base, "--head", head,
                     "--plan", str(root / "plan.json"), *extra], cwd=root, env=env, capture_output=True, text=True)
                 self.assertEqual(result.returncode, 0, result.stderr)
                 return json.loads((root / "plan.json").read_text()), result.stdout
@@ -67,7 +67,8 @@ def classify(base, *extra):
             path = "loopx/web/chat/index.html"
             write(path, "old")
             for policy in POLICY_PATHS:
-                write(policy, "reviewed policy")
+                write(policy, "jobs:\n  test-shard:\n    strategy:\n      matrix:\n        shard: [1, 2, 3, 4, 5, 6]\n"
+                      if policy.endswith(".yml") else "reviewed policy")
             git("add", ".")
             git("commit", "-qm", "base")
             base = git("rev-parse", "HEAD")
@@ -78,6 +79,7 @@ def classify(base, *extra):
             self.assertFalse(packet["python_tests"])
             self.assertFalse(packet["backend_tests"])
             self.assertFalse(packet["stage2c_tests"])
+            self.assertEqual(packet["python_shards"], 0)
             self.assertTrue(packet["presentation_tests"])
             self.assertEqual(packet["head_sha"], git("rev-parse", "HEAD"))
             for flag in ("--force-full", "--non-pr"):
@@ -85,7 +87,7 @@ def classify(base, *extra):
                 self.assertTrue(forced["python_tests"])
                 self.assertTrue(forced["backend_tests"])
                 self.assertTrue(forced["stage2c_tests"])
-                self.assertEqual(forced["python_shards"], 4)
+                self.assertEqual(forced["python_shards"], 6)
             # A policy change cannot authorize its own exemption.
             write(POLICY_PATHS[0], "changed policy")
             git("commit", "-qam", "policy")
@@ -116,6 +118,39 @@ def classify(base, *extra):
             failed = subprocess.run([sys.executable, script, "classify", "--base", "missing", "--head", "HEAD"],
                                     cwd=root, env=env, capture_output=True)
             self.assertNotEqual(failed.returncode, 0)
+            # The workflow executing this checkout owns topology, even when
+            # the candidate head differs or an uncommitted file is misleading.
+            published = git("rev-parse", "HEAD")
+            workflow = POLICY_PATHS[-1]
+            prefix = "jobs:\n  test-shard:\n    strategy:\n      matrix:\n"
+            write(workflow, prefix + "        shard: [1, 2, 3, 4, 5, 6, 7, 8]\n")
+            self.assertEqual(classify(published, "--non-pr")[0]["python_shards"], 6)
+            git("commit", "-qam", "executing workflow has eight shards")
+            packet, _ = classify(published, "--non-pr", head=published)
+            self.assertEqual(packet["python_shards"], 8)
+            self.assertEqual(packet["head_sha"], published)
+            self.assertEqual(packet["checkout_sha"], git("rev-parse", "HEAD"))
+            for axis in ("shard: []", "shard: [1, 1]", "shard: [1, 3]", "shard: [1, '02']",
+                         "shard: '${{ inputs.shards }}'", "shard: [1, 2]\n        python: ['3.11']", None):
+                with self.subTest(axis=axis):
+                    write(workflow, prefix + "        " + axis + "\n" if axis else "jobs: {}\n")
+                    git("commit", "-qam", "unsupported execution topology")
+                    output = root / "rejected.json"
+                    result = subprocess.run([sys.executable, script, "classify", "--base", "HEAD",
+                        "--head", "HEAD", "--non-pr", "--plan", str(output)],
+                        cwd=root, env=env, capture_output=True, text=True)
+                    self.assertNotEqual(result.returncode, 0)
+                    self.assertIn("test-shard.strategy.matrix", result.stderr)
+                    self.assertFalse(output.exists(), "invalid topology must not publish a fallback plan")
+            # The docs exemption still reports no Python coverage or shards.
+            docs_base = git("rev-parse", "HEAD")
+            write("README.md", "fixture documentation")
+            git("add", "--", "README.md")
+            git("commit", "-qm", "docs-only change")
+            packet, _ = classify(docs_base)
+            self.assertEqual(packet["change_kind"], "docs")
+            self.assertFalse(packet["python_tests"])
+            self.assertEqual(packet["python_shards"], 0)
 
 
 if __name__ == "__main__":
diff --git a/scripts/ci/test_review_gate.py b/scripts/ci/test_review_gate.py
index 9efe533523..edfc84e056 100644
--- a/scripts/ci/test_review_gate.py
+++ b/scripts/ci/test_review_gate.py
@@ -6,6 +6,7 @@
 from pathlib import Path
 import re
 import subprocess
+import sys
 import tempfile
 from types import SimpleNamespace
 import unittest
@@ -37,6 +38,16 @@ def needs(kind="full", *, presentation=False, jobs=JOB_OUTPUT):
 
 
 class GateTests(unittest.TestCase):
+    def test_real_aggregate_entrypoints_do_not_need_the_classification_yaml_parser(self):
+        script = Path(__file__).with_name("review_gate.py")
+        for command, jobs in (("verify", JOB_OUTPUT), ("verify-core", CHECK_OUTPUT)):
+            with self.subTest(command=command):
+                result = subprocess.run([sys.executable, "-S", str(script), command],
+                    env={**os.environ, "PYTHONPATH": "", "NEEDS_JSON": json.dumps(needs(jobs=jobs))},
+                    capture_output=True, text=True)
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertIn("qualification complete", result.stdout)
+
     def test_closed_pr_cancels_same_lane_and_skips_qualification(self):
         workflow = yaml.load(
             (Path(__file__).resolve().parents[2] / ".github/workflows/python-tests.yml").read_text(),