fix(klaud): unblock AgentX smoke and stop re-picking blocked candidates / fix(klaud): 修复 AgentX 冒烟测试并停止重复选择受阻候选

.github/workflows/klaud-plan.yml

@@ -4,7 +4,7 @@ name: Klaud Cold auto-sweep
 on: # zizmor: ignore[concurrency-limits]
   workflow_dispatch:
   schedule:
-    - cron: '0 */6 * * *'
+    - cron: '0 */2 * * *'
 
 permissions:
   contents: read
@@ -81,9 +81,9 @@ jobs:
       actions: read # Read workflow runs and artifacts.
       pull-requests: read # Read PR metadata.
     env:
-      # Bound one semantic-overlap review; recently attempted same-base candidates sort last.
       REVIEW_BATCH_SIZE: '64'
       CANDIDATE_COOLDOWN_HOURS: '24'
+      STALE_PR_DAYS: '7'
     outputs:
       review: ${{ steps.review.outputs.structured_output }}
       outcome: ${{ steps.review.outcome }}
@@ -122,6 +122,7 @@ jobs:
           --directory "$RUNNER_TEMP/klaud"
           --review-batch-size "$REVIEW_BATCH_SIZE"
           --cooldown-hours "$CANDIDATE_COOLDOWN_HOURS"
+          --stale-pr-days "$STALE_PR_DAYS"
       # Uploaded before the agent runs, so selection reads the planner's files, not agent-touched ones.
       - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
         with:
@@ -163,8 +164,9 @@ jobs:
             --json-schema '${{ steps.prepare.outputs.review_schema }}'
           prompt: |
             Use Read on candidates.json, open-prs.json and capacity.json under ${{ runner.temp }}/klaud/.
-            Review candidates in the supplied shuffled order against current recipes and the complete open-PR index,
-            including drafts and arbitrary branches. Changed-file paths are already fetched;
+            Review candidates in the supplied shuffled order against current recipes and the open-PR index,
+            including drafts and arbitrary branches. The index omits PRs idle for ${{ env.STALE_PR_DAYS }} days
+            (other than Klaud candidate PRs); those no longer own any family. Changed-file paths are already fetched;
             inspect relevant PR bodies/diffs with direct gh pr view/diff commands. For API reads,
             use gh api --method GET. Use Read/Glob/Grep for local inspection, not shell file commands
             or Python. Run one allowed gh/git command per Bash call, without cd/env wrappers,
@@ -186,7 +188,10 @@ jobs:
             to fill the cap. Never copy private eligibility data into reasons or reports.
             Mark duplicate when an open PR owns that family's image refresh or overlapping
             compatibility work, even with another image tag; unrelated work on the same hardware
-            is not a duplicate. Return the supplied schema: candidate-id, decision
+            is not a duplicate. A cross-cutting PR (launcher, telemetry, refactor or shared-schema
+            work spanning many families) is a duplicate only if its diff changes this family's
+            image or the exact engine flags/environment a refresh edits; touching the same recipe
+            file for unrelated settings is not overlap. Return the supplied schema: candidate-id, decision
             (proceed/duplicate/uncertain), family (null if unresolved), telemetry-clusters (exact
             target IDs, empty if unresolved), pull-requests (overlapping
             PR numbers), baseline-model (required for proceed; otherwise null), reason (brief evidence).

inferencex-e2e/infx/klaud/__main__.py

@@ -11,13 +11,19 @@
 import secrets
 import shlex
 import subprocess
+import tempfile
 from collections import Counter
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from types import SimpleNamespace
 
 from .api import PUBLIC, ReadError, capacity_context, fetch_capacity, fetch_catalog
-from .github import VerificationError, items as github_items, read as github_read
+from .github import (
+    VerificationError,
+    download_json,
+    items as github_items,
+    read as github_read,
+)
 from .models import (
     CandidateOutcome,
     OwnedCandidate,
@@ -29,6 +35,7 @@
 )
 
 PRODUCERS = "producers.json"
+RETRY_NEXT_WAVE = frozenset({"capacity-deferred"})
 
 
 def observation_key(row: dict) -> tuple:
@@ -152,8 +159,7 @@ def choose(
     return selected
 
 
-def recent_candidate_ids(repository: str, base: str, cooldown_hours: int) -> set[str]:
-    """Return same-base candidates recently given an agent, as a soft ordering hint."""
+def recent_candidate_outcomes(repository: str, cooldown_hours: int) -> dict[str, str | None]:
     cutoff = datetime.now(UTC) - timedelta(hours=cooldown_hours)
     try:
         runs = github_items(
@@ -163,38 +169,75 @@ def recent_candidate_ids(repository: str, base: str, cooldown_hours: int) -> set
         )
     except (OSError, ValueError, KeyError, TypeError, subprocess.SubprocessError):
         print("::warning::Klaud cooldown history unavailable; continuing without cooldown")
-        return set()
-    result = set()
+        return {}
+    dated = []
     for run in runs:
         try:
             created = datetime.fromisoformat(run["created_at"].replace("Z", "+00:00"))
-            if created < cutoff or run.get("head_sha") != base:
-                continue
+            if created >= cutoff:
+                dated.append((created, int(run["id"])))
+        except (AttributeError, KeyError, TypeError, ValueError):
+            continue
+    result: dict[str, str | None] = {}
+    for _, run_id in sorted(dated, reverse=True):
+        try:
             artifacts = github_items(
-                repository, f"actions/runs/{int(run['id'])}/artifacts?per_page=100", "artifacts"
+                repository, f"actions/runs/{run_id}/artifacts?per_page=100", "artifacts"
             )
-        except (
-            AttributeError,
-            KeyError,
-            OSError,
-            TypeError,
-            ValueError,
-            subprocess.SubprocessError,
-        ):
+        except (OSError, ValueError, KeyError, TypeError, subprocess.SubprocessError):
             continue
         for artifact in artifacts:
             try:
                 match = re.fullmatch(
                     r"klaud-candidate-([0-9a-f]{16}-[0-9a-f]{16})", artifact["name"]
                 )
-                if match and not artifact.get("expired", True):
-                    result.add(match[1])
+                if not match or artifact.get("expired", True) or match[1] in result:
+                    continue
             except (AttributeError, KeyError, TypeError):
                 continue
+            result[match[1]] = recorded_outcome(repository, artifact)
     return result
 
 
-def plan(root: Path, directory: Path, review_batch_size: int, cooldown_hours: int) -> None:
+def recorded_outcome(repository: str, artifact: dict) -> str | None:
+    with tempfile.TemporaryDirectory() as scratch:
+        try:
+            download_json(repository, artifact, Path(scratch))
+            outcome = json.loads((Path(scratch) / "candidate-diagnostics.json").read_text())[
+                "candidate-outcome"
+            ]
+            return CandidateOutcome.model_validate(outcome).outcome
+        except (
+            OSError,
+            ValueError,
+            KeyError,
+            TypeError,
+            subprocess.SubprocessError,
+        ):
+            return None
+
+
+def order_by_cooldown(candidates: list[dict], recent: dict[str, str | None]) -> list[dict]:
+    fresh = [candidate for candidate in candidates if candidate["id"] not in recent]
+    deferred = [
+        candidate for candidate in candidates if recent.get(candidate["id"], "") in RETRY_NEXT_WAVE
+    ]
+    return fresh + deferred
+
+
+def active_pulls(pulls: list[dict], now: datetime, stale_days: int) -> list[dict]:
+    cutoff = now - timedelta(days=stale_days)
+    return [
+        pr
+        for pr in pulls
+        if re.match(r"^klaud[e]?/auto-", pr["head"]["ref"])
+        or datetime.fromisoformat(pr["updated_at"].replace("Z", "+00:00")) >= cutoff
+    ]
+
+
+def plan(
+    root: Path, directory: Path, review_batch_size: int, cooldown_hours: int, stale_pr_days: int
+) -> None:
     from . import claims
 
     policy = Policy()
@@ -227,10 +270,9 @@ def plan(root: Path, directory: Path, review_batch_size: int, cooldown_hours: in
     if recovery_file.exists():
         blocked.update(json.loads(recovery_file.read_text()))
     candidates = [candidate for candidate in candidates if candidate["family"] not in blocked]
-    recent = recent_candidate_ids(repository, base, cooldown_hours)
-    candidates = [candidate for candidate in candidates if candidate["id"] not in recent] + [
-        candidate for candidate in candidates if candidate["id"] in recent
-    ]
+    candidates = order_by_cooldown(
+        candidates, recent_candidate_outcomes(repository, cooldown_hours)
+    )
     candidates = candidates[:review_batch_size]
     contexts = [
         {
@@ -256,8 +298,9 @@ def plan(root: Path, directory: Path, review_batch_size: int, cooldown_hours: in
             "branch": pr["head"]["ref"],
             "head": pr["head"]["sha"],
         }
-        for page in pulls
-        for pr in page
+        for pr in active_pulls(
+            [pr for page in pulls for pr in page], datetime.now(UTC), stale_pr_days
+        )
     ]
     if candidates:
         for pr in open_prs:
@@ -781,7 +824,13 @@ def main() -> int:
         "--cooldown-hours",
         type=int,
         required=True,
-        help="Soft same-base candidate cooldown before review (1-168)",
+        help="Hours a recently attempted candidate waits before another agent (1-168)",
+    )
+    prepare.add_argument(
+        "--stale-pr-days",
+        type=int,
+        required=True,
+        help="Days without activity after which another open PR stops owning a family (1-90)",
     )
     regeneration = commands.add_parser(
         "regenerate-producers",
@@ -969,7 +1018,15 @@ def main() -> int:
                 parser.error("--review-batch-size must be between 1 and 256")
             if not 1 <= args.cooldown_hours <= 168:
                 parser.error("--cooldown-hours must be between 1 and 168")
-            plan(args.root, args.directory, args.review_batch_size, args.cooldown_hours)
+            if not 1 <= args.stale_pr_days <= 90:
+                parser.error("--stale-pr-days must be between 1 and 90")
+            plan(
+                args.root,
+                args.directory,
+                args.review_batch_size,
+                args.cooldown_hours,
+                args.stale_pr_days,
+            )
             return 0
         if args.command == "regenerate-producers":
             regenerate_producers(args.directory)

inferencex-e2e/infx/matrix/generate.py

@@ -143,7 +143,7 @@ def smoke_entries(entries: list[dict]) -> list[dict]:
         row.pop(Fields.REQUIRE_POWER.value, None)
         if row.get("prefill") is not None:
             row["conc"] = [row["eval-conc"]]
-            row["eval-all-concs"] = False
+            row.pop("eval-all-concs", None)
         evals.append(row)
     return benchmarks + trim_conc(evals)
 

inferencex-e2e/infx/tests/klaud/test_klaud_github.py

@@ -203,33 +203,77 @@ def test_finish_requires_specific_baseline_reason_before_touching_session(
     )
 
 
-def test_recent_candidates_use_all_current_base_workflow_artifacts(monkeypatch):
-    candidate_id = "1" * 16 + "-" + "2" * 16
-    base = "a" * 40
-    calls = []
+def test_cooldown_skips_recent_blocks_across_bases_and_retries_capacity_deferrals(
+    monkeypatch,
+):
+    blocked, deferred, unreadable, superseded, untried = (
+        f"{digit * 16}-{digit * 16}" for digit in "12345"
+    )
+    outcomes = {
+        1: ("readiness-blocked", "targeted"),
+        2: ("capacity-deferred", "targeted"),
+        4: ("capacity-deferred", "targeted"),
+        5: ("failed", "targeted"),
+    }
 
     def items(_repository, path, key):
-        calls.append((path, key))
         if path.startswith("actions/workflows/klaud-plan.yml/runs?"):
             return [
-                {"id": 42, "created_at": "9999-01-01T00:00:00Z", "head_sha": base},
-                {"id": 43, "created_at": "9999-01-01T00:00:00Z", "head_sha": "b" * 40},
+                {"id": 41, "created_at": "9999-01-01T00:00:00Z", "head_sha": "a" * 40},
+                {"id": 42, "created_at": "9999-01-02T00:00:00Z", "head_sha": "b" * 40},
+                {"id": 40, "created_at": "2000-01-01T00:00:00Z", "head_sha": "c" * 40},
             ]
-        if path == "actions/runs/42/artifacts?per_page=100":
-            return [
-                {
-                    "name": f"klaud-candidate-{candidate_id}",
-                    "expired": False,
-                    "workflow_run": None,
-                },
-                {"name": "other-artifact", "expired": False},
-            ]
-        raise AssertionError(path)
+        run_artifacts = {
+            41: [(1, blocked), (3, unreadable), (5, superseded)],
+            42: [(2, deferred), (4, superseded)],
+            40: [(6, untried)],
+        }
+        run_id = int(path.split("/")[2])
+        return [
+            {"id": artifact, "name": f"klaud-candidate-{candidate}", "expired": False}
+            for artifact, candidate in run_artifacts[run_id]
+        ]
+
+    def download(_repository, artifact, destination):
+        if artifact["id"] not in outcomes:
+            raise github.VerificationError("Invalid artifact archive")
+        outcome, phase = outcomes[artifact["id"]]
+        record = {"outcome": outcome, "phase": phase, "pull-request": None, "run-ids": [],
+                  "repairs-used": 0}
+        (destination / "candidate-diagnostics.json").write_text(
+            json.dumps({"candidate-outcome": record})
+        )
 
     monkeypatch.setattr(klaud, "github_items", items)
+    monkeypatch.setattr(klaud, "download_json", download)
+
+    recent = klaud.recent_candidate_outcomes("example/project", 24)
+    assert recent == {
+        blocked: "readiness-blocked",
+        unreadable: None,
+        deferred: "capacity-deferred",
+        superseded: "capacity-deferred",
+    }
+    candidates = [{"id": candidate} for candidate in (deferred, blocked, untried, unreadable,
+                                                      superseded)]
+    assert [candidate["id"] for candidate in klaud.order_by_cooldown(candidates, recent)] == [
+        untried, deferred, superseded]
+
+
+def test_overlap_index_drops_idle_prs_but_keeps_klaud_candidate_prs():
+    now = klaud.datetime(2026, 10, 14, tzinfo=klaud.UTC)
+
+    def pull(number, branch, updated):
+        return {"number": number, "head": {"ref": branch}, "updated_at": updated}
+
+    pulls = [
+        pull(1, "feature/active", "2026-10-07T00:00:00Z"),
+        pull(2, "klaud/qwen3.5-nightly", "2026-10-06T23:59:59Z"),
+        pull(3, "klaud/auto-0123456789abcdef-0123456789abcdef", "2026-09-01T00:00:00Z"),
+        pull(4, "klaude/auto-fedcba9876543210-fedcba9876543210", "2026-09-01T00:00:00Z"),
+    ]
 
-    assert klaud.recent_candidate_ids("example/project", base, 24) == {candidate_id}
-    assert calls[-1] == ("actions/runs/42/artifacts?per_page=100", "artifacts")
+    assert [pr["number"] for pr in klaud.active_pulls(pulls, now, 7)] == [1, 3, 4]
 
 
 def test_select_continues_after_one_baseline_state_failure(tmp_path, monkeypatch):

inferencex-e2e/infx/tests/matrix/test_validation.py

@@ -1335,6 +1335,17 @@ def test_validate_agentic_matrix_entry_dispatches_on_prefill_key(self):
         with pytest.raises(ValueError):
             validate_agentic_matrix_entry(without_prefill)
 
+    def test_smoke_rows_satisfy_the_dispatch_schema(self):
+        from infx.matrix.generate import smoke_entries
+
+        row = {**MULTINODE_AGENTIC_EVAL_ROW, "conc": [8, 32]}
+        del row["eval-only"]
+        rows = smoke_entries([row])
+
+        assert [(entry["conc"], entry.get("eval-only", False)) for entry in rows] == [
+            ([8], False), ([32], True)]
+        for entry in rows:
+            validate_agentic_matrix_entry(entry)
 
 
 class TestLoadConfigFiles: