@@ -19,6 +19,7 @@
 import argparse
 import json
 import os
+import re
 import subprocess
 from urllib.parse import urlencode
 
@@ -33,31 +34,69 @@ def api(path):
 
 def decide(repository, run_id, expected_attempt, max_reruns, fetch=api):
     run = fetch(f"repos/{repository}/actions/runs/{run_id}")
-    if run.get("path") not in {".github/workflows/server-memory-ci.yml", ".github/workflows/licence-checker.yml"}:
-        return "skip", "only required workflows automatically retry"
+    if run.get("path") != ".github/workflows/server-ci.yml":
+        return "skip", "only Server CI required jobs automatically retry", None
     if (run.get("status") != "completed" or run.get("conclusion") != "failure"
             or run.get("run_attempt") != expected_attempt):
-        return "skip", "source run changed or is no longer a completed failure"
+        return "skip", "source run changed or is no longer a completed failure", None
     if expected_attempt > max_reruns:
-        return "skip", "retry limit reached"
+        return "skip", "retry limit reached", None
     sha = run.get("head_sha")
     if not sha:
-        return "skip", "missing head SHA"
+        return "skip", "missing head SHA", None
     if run.get("event") == "pull_request":
-        return "skip", "PR automatic retry paused: CI artifacts cannot independently prove the tested code"
+        return "skip", "PR automatic retry paused: CI artifacts cannot independently prove the tested code", None
     if run.get("event") == "push":
         if (run.get("repository", {}).get("full_name") != repository
                 or run.get("head_repository", {}).get("full_name") != repository):
-            return "skip", "push source repository differs"
+            return "skip", "push source repository differs", None
         branch = run.get("head_branch")
         if not branch:
-            return "skip", "missing push branch"
+            return "skip", "missing push branch", None
         # Read from the workflow repository, never a similarly named fork branch.
         commits = fetch(f"repos/{repository}/commits?" + urlencode({"sha": branch, "per_page": 1}))
-        if commits and commits[0].get("sha") == sha:
-            return "rerun", "push branch head unchanged"
-        return "skip", "push branch head moved"
-    return "skip", "unsupported source event"
+        if not commits or commits[0].get("sha") != sha:
+            return "skip", "push branch head moved", None
+        # A native job rerun includes its dependent jobs. Prefer a failed producer
+        # over its failed reporter; never select the advisory branch of Server CI.
+        targets = []
+        page = 1
+        while True:
+            jobs = fetch(f"repos/{repository}/actions/runs/{run_id}/attempts/{expected_attempt}/jobs"
+                         f"?per_page=100&page={page}")["jobs"]
+            for job in jobs:
+                if job.get("status") != "completed" or job.get("conclusion") not in {"failure", "timed_out"}:
+                    continue
+                name = job.get("name", "")
+                if name == "check-license":
+                    priority = 0
+                elif name == "memory-plan":
+                    priority = 1
+                elif name == "memory-java-runtime / read-version":
+                    priority = 2
+                elif re.fullmatch(r"build-server / server_memory \(Java [1-9][0-9]*(?:\.[0-9]+)*\)", name):
+                    priority = 3
+                elif name == "Server memory tests":
+                    priority = 4
+                else:
+                    continue
+                # Only failed reporters retry themselves; a timed-out producer
+                # must be retried before its dependent failure can recover.
+                if priority == 4 and job.get("conclusion") != "failure":
+                    continue
+                if job["run_id"] != run_id or job["head_sha"] != sha:
+                    return "skip", "job does not match source run", None
+                job_id = job["id"]
+                if not isinstance(job_id, int) or job_id <= 0:
+                    return "skip", "invalid job ID", None
+                targets.append((priority, job_id))
+            if len(jobs) < 100:
+                break
+            page += 1
+        if targets:
+            return "rerun", "failed required job and its dependencies", min(targets)[1]
+        return "skip", "no failed required job in this attempt", None
+    return "skip", "unsupported source event", None
 
 
 def main():
@@ -68,12 +107,12 @@ def main():
     parser.add_argument("--max-reruns", required=True, type=int)
     args = parser.parse_args()
     try:
-        action, reason = decide(args.repository, args.run_id, args.run_attempt, args.max_reruns)
+        action, reason, job_id = decide(args.repository, args.run_id, args.run_attempt, args.max_reruns)
     except (subprocess.SubprocessError, OSError, ValueError, KeyError, TypeError, AttributeError):
-        action, reason = "skip", "unable to verify current run and source metadata"
+        action, reason, job_id = "skip", "unable to verify current run and source metadata", None
     print(f"{action}: {reason}")
     with open(os.environ["GITHUB_OUTPUT"], "a", encoding="utf-8") as output:
-        output.write(f"action={action}\nreason={reason}\n")
+        output.write(f"action={action}\nreason={reason}\njob_id={job_id or ''}\n")
 
 
 if __name__ == "__main__":
 
@@ -21,6 +21,7 @@
 import os
 import subprocess
 import tempfile
+import json
 from unittest.mock import patch
 
 spec = importlib.util.spec_from_file_location("check_rerun", Path(__file__).with_name("check-rerun.py"))
@@ -32,31 +33,113 @@ class FreshnessTest(unittest.TestCase):
     def setUp(self):
         self.repository = "apache/hugegraph"
         self.run = {"status": "completed", "conclusion": "failure", "run_attempt": 1,
-                    "path": ".github/workflows/server-memory-ci.yml", "event": "push", "head_sha": "old", "head_branch": "release/test",
+                    "path": ".github/workflows/server-ci.yml", "event": "push", "head_sha": "old", "head_branch": "release/test",
                     "repository": {"full_name": self.repository},
                     "head_repository": {"full_name": self.repository}}
         self.current_head = "old"
         self.calls = []
+        self.jobs = [self.job(117, "build-server / server_memory (Java 17)"),
+                     self.job(118, "Server memory tests")]
+
+    def job(self, job_id, name, conclusion="failure"):
+        return {"id": job_id, "name": name, "status": "completed", "conclusion": conclusion,
+                "run_id": 42, "head_sha": "old"}
 
     def fetch(self, path):
         self.calls.append(path)
         if path == f"repos/{self.repository}/actions/runs/42":
             return self.run
         if path == f"repos/{self.repository}/commits?sha=release%2Ftest&per_page=1":
             return [{"sha": self.current_head}] if self.current_head else []
+        if path.startswith(f"repos/{self.repository}/actions/runs/42/attempts/"):
+            page = int(path.rsplit("page=", 1)[1])
+            return {"jobs": self.jobs[(page - 1) * 100:page * 100]}
         raise AssertionError("Unexpected lookup: " + path)
 
     def decide(self):
         return checker.decide(self.repository, 42, 1, 2, self.fetch)[0]
 
-    def test_only_required_workflows_retry(self):
-        for path in [".github/workflows/server-ci.yml", ".github/workflows/codeql-analysis.yml",
-                     ".github/workflows/server-compatibility-ci.yml"]:
+    def test_only_server_required_jobs_retry(self):
+        for path in [".github/workflows/server-memory-ci.yml", ".github/workflows/licence-checker.yml",
+                     ".github/workflows/codeql-analysis.yml", ".github/workflows/server-compatibility-ci.yml"]:
             self.run["path"] = path
             self.assertEqual("skip", self.decide())
-        self.run["path"] = ".github/workflows/licence-checker.yml"
+        self.run["path"] = ".github/workflows/server-ci.yml"
         self.assertEqual("rerun", self.decide())
 
+    def test_failed_producer_is_selected_before_reporter(self):
+        for name in ["check-license", "memory-plan", "memory-java-runtime / read-version",
+                     "build-server / server_memory (Java 17)"]:
+            with self.subTest(name=name):
+                self.jobs = [self.job(118, "Server memory tests"), self.job(117, name)]
+                self.assertEqual(117, checker.decide(self.repository, 42, 1, 2, self.fetch)[2])
+        self.jobs = [self.job(118, "Server memory tests")]
+        self.assertEqual(118, checker.decide(self.repository, 42, 1, 2, self.fetch)[2])
+
+    def test_timed_out_producer_is_retried_only_inside_failed_run(self):
+        self.jobs = [self.job(118, "Server memory tests"),
+                     self.job(117, "build-server / server_memory (Java 17)", "timed_out")]
+        self.assertEqual(117, checker.decide(self.repository, 42, 1, 2, self.fetch)[2])
+        self.run["conclusion"] = "timed_out"
+        self.assertEqual("skip", self.decide())
+
+    def test_timed_out_reporter_alone_is_not_retried(self):
+        self.jobs = [self.job(118, "Server memory tests", "timed_out")]
+        self.assertEqual("skip", self.decide())
+
+    def test_advisory_success_and_non_failure_jobs_are_not_retried(self):
+        for name in ["plan", "java-runtime / read-version", "server-rocksdb / server_rocksdb (Java 17)",
+                     "pd_store / struct", "affected-module-tests", "build-server / server_memory (Java 17) / extra"]:
+            self.jobs = [self.job(118, name)]
+            self.assertEqual("skip", self.decide())
+        for conclusion in ["success", "cancelled", "skipped", None]:
+            self.jobs = [self.job(118, "Server memory tests", conclusion)]
+            self.assertEqual("skip", self.decide())
+
+    def test_job_selection_paginates_the_exact_attempt(self):
+        self.jobs = [self.job(i + 1, "advisory") for i in range(100)] + [self.job(117, "memory-plan")]
+        self.run["run_attempt"] = 2
+        self.assertEqual(117, checker.decide(self.repository, 42, 2, 2, self.fetch)[2])
+        self.assertIn(f"repos/{self.repository}/actions/runs/42/attempts/2/jobs?per_page=100&page=2", self.calls)
+        self.assertFalse(any("/attempts/1/" in path for path in self.calls))
+
+    def test_job_identity_and_metadata_must_match(self):
+        for key, value in [("run_id", 99), ("head_sha", "different"), ("id", "117"), ("id", -1)]:
+            with self.subTest(key=key):
+                self.jobs = [dict(self.job(117, "memory-plan"), **{key: value})]
+                self.assertEqual("skip", self.decide())
+
+    def test_workflow_posts_native_rerun_for_checker_selected_job(self):
+        workflow = Path(__file__).resolve().parents[1] / "workflows/rerun-ci.yml"
+        source = workflow.read_text()
+        self.assertIn('      - "HugeGraph-Server CI"', source)
+        self.assertIn("github.event.workflow_run.event == 'push'", source)
+        self.assertIn("ref: ${{ github.sha }}", source)
+        command = source.split("      - name: Rerun the required job and its dependencies", 1)[1]
+        command = command.split("        run: |\n", 1)[1].strip()
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            output = root / "output"
+            decide = checker.decide
+            with patch.dict(os.environ, {"GITHUB_OUTPUT": str(output)}), patch(
+                    "sys.argv", ["check-rerun.py", "--repository", self.repository,
+                                 "--run-id", "42", "--run-attempt", "1", "--max-reruns", "2"]), patch.object(
+                    checker, "decide", side_effect=lambda *args: decide(*args, fetch=self.fetch)):
+                checker.main()
+            selected = dict(line.split("=", 1) for line in output.read_text().splitlines())
+            self.assertEqual("rerun", selected["action"])
+            command = command.replace("${{ steps.freshness.outputs.job_id }}", selected["job_id"])
+            gh = root / "gh"
+            gh.write_text('#!/usr/bin/env python3\nimport json, os, sys\n'
+                          'open(os.environ["ARGS_PATH"], "w").write(json.dumps(sys.argv[1:]))\n')
+            gh.chmod(0o700)
+            args_path = root / "args.json"
+            subprocess.run(["bash", "-eu", "-c", command], check=True,
+                           env={**os.environ, "GH_REPO": self.repository, "ARGS_PATH": str(args_path),
+                                "PATH": str(root) + os.pathsep + os.environ["PATH"]})
+            self.assertEqual(["api", "--method", "POST", f"repos/{self.repository}/actions/jobs/117/rerun"],
+                             json.loads(args_path.read_text()))
+
     def test_api_failure_fails_closed(self):
         with tempfile.TemporaryDirectory() as directory:
             output = Path(directory) / "output"
 
@@ -21,6 +21,7 @@
 from pathlib import Path
 import subprocess
 import tempfile
+import textwrap
 import unittest
 from unittest.mock import patch
 
@@ -197,6 +198,45 @@ def plan(self):
                 "testedMergeSHA": "merge", "expected": ["server_memory", "server_rocksdb", "pd_store", "cluster"],
                 "selected": ["server", "pd", "store", "hstore", "cluster"]}
 
+    def memory_workflow_results(self, planner="success", runtime="success", build="success", selected=True):
+        workflow = Path(__file__).resolve().parents[1] / "workflows/server-ci.yml"
+        reporter = workflow.read_text().split("\n  server-memory-required:\n", 1)[1]
+        code = reporter.split("python3 - <<'PYGATE'\n", 1)[1].split("\n          PYGATE", 1)[0]
+        # Start with the actual needs keys, without generic gate aliases.
+        needs = {"memory-plan": {"result": planner, "outputs": {"server": str(selected).lower()}},
+                 "memory-java-runtime": {"result": runtime}, "build-server": {"result": build}}
+        with tempfile.TemporaryDirectory() as directory:
+            outcome = subprocess.run(["python3", "-c", textwrap.dedent(code)], cwd=directory,
+                                     env={**os.environ, "JOB_RESULTS": json.dumps(needs)},
+                                     capture_output=True, text=True)
+            if outcome.returncode:
+                return outcome, None
+            return outcome, json.loads((Path(directory) / "results.json").read_text())
+
+    def test_memory_workflow_results_feed_real_gate(self):
+        for state in ["success", "failure", "cancelled", "skipped", None]:
+            with self.subTest(build=state):
+                outcome, results = self.memory_workflow_results(build=state)
+                self.assertEqual(0, outcome.returncode, outcome.stderr)
+                if state == "success":
+                    self.assertEqual(["server_memory"], policy.gate(self.plan(), results, mode="memory")["executed"])
+                else:
+                    with self.assertRaisesRegex(policy.GateFailure, "server_memory"):
+                        policy.gate(self.plan(), results, mode="memory")
+        outcome, results = self.memory_workflow_results(runtime="skipped", build="skipped", selected=False)
+        self.assertEqual(0, outcome.returncode, outcome.stderr)
+        docs = dict(self.plan(), expected=[], selected=[])
+        self.assertEqual([], policy.gate(docs, results, mode="memory")["executed"])
+        for state in ["failure", "cancelled", "skipped"]:
+            with self.subTest(planner=state):
+                outcome, results = self.memory_workflow_results(planner=state)
+                self.assertNotEqual(0, outcome.returncode)
+                self.assertIn("CI planning failed", outcome.stderr)
+            with self.subTest(runtime=state):
+                outcome, results = self.memory_workflow_results(runtime=state)
+                self.assertNotEqual(0, outcome.returncode)
+                self.assertIn("Selected Java runtime preparation failed", outcome.stderr)
+
     def test_memory_gate_ignores_advisory_cancellation(self):
         plan = self.plan()
         results = {"plan": {"result": "success"}, "server_memory": {"result": "success"},
 
@@ -3,8 +3,7 @@ name: "Rerun CI"
 on:
   workflow_run:
     workflows:
-      - "Server Memory CI"
-      - "License Checker"
+      - "HugeGraph-Server CI"
     types:
       - completed
 
@@ -108,10 +107,10 @@ jobs:
             --repository "$GH_REPO" --run-id "$RUN_ID" \
             --run-attempt "$RUN_ATTEMPT" --max-reruns "$MAX_RERUNS"
 
-      - name: Rerun failed jobs
+      - name: Rerun the required job and its dependencies
         if: steps.freshness.outputs.action == 'rerun'
         env:
           GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           GH_REPO: ${{ github.repository }}
         run: |
-          gh run rerun ${{ github.event.workflow_run.id }} --failed
+          gh api --method POST "repos/$GH_REPO/actions/jobs/${{ steps.freshness.outputs.job_id }}/rerun"