fix(klaud): wait for capacity instead of discarding passing smoke runs / fix(klaud): 等待容量而不是丢弃已通过的冒烟测试

.github/klaud-candidate-prompt.md

@@ -48,12 +48,16 @@ maintainer ownership: leave PR/branch/labels/jobs intact, report handoff and sto
 Never add or remove it yourself.
 
 Check capacity before edits/branch/PR creation, every targeted dispatch, the final label
-transition and capacity-related recovery: `check-capacity --cluster ID` (repeat for ALL
-actual targets). Require fresh, available telemetry and utilization strictly below 80%.
+transition and capacity-related recovery: `check-capacity --cluster ID` (repeat --cluster for
+ALL actual targets). Require fresh, available telemetry and utilization strictly below 80%.
 Queue eligible work with normal scheduler controls; no priority overrides.
-If the check fails, report capacity-deferred and call finish. A utilization increase after
-dispatch never justifies cancelling healthy work. A benchmark capacity error alone is
-insufficient: recheck capacity before deciding to defer.
+Before the PR exists, a failed check means report capacity-deferred and call finish.
+Once the PR exists, wait instead: rerun the same command with `--wait` and a 600000 ms Bash
+timeout while it exits 1;
+exit 0 means proceed, exit 2 means the KLAUD_CAPACITY_WAIT_MINUTES budget is spent, so
+report capacity-deferred and call finish. Keep passing smoke evidence while waiting; never
+redispatch it. A utilization increase after dispatch never justifies cancelling healthy work.
+A benchmark capacity error alone is insufficient: recheck capacity before deciding to defer.
 
 Freeze the COMPLETE original public baseline point roster before attempts, using
 candidate.source.date, verified old-image producer IDs/SHAs and full recipe/workload/topology/

.github/workflows/klaud-candidate.yml

@@ -30,6 +30,7 @@ jobs:
     env:
       # Repair attempts after the initial update; the baseline is already published.
       MAX_REPAIRS: '5'
+      KLAUD_CAPACITY_WAIT_MINUTES: '120'
       KLAUD_BRANCH: klaud/auto-${{ inputs.candidate }}
       KLAUD_TEST_NAME: klaud-${{ github.run_id }}-${{ inputs.candidate }}
     steps:

inferencex-e2e/infx/klaud/__main__.py

@@ -12,6 +12,7 @@
 import shlex
 import subprocess
 import tempfile
+import time
 from collections import Counter
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
@@ -36,6 +37,31 @@
 
 PRODUCERS = "producers.json"
 RETRY_NEXT_WAVE = frozenset({"capacity-deferred"})
+CAPACITY_LEDGER = "capacity-wait.json"
+CAPACITY_POLL_SECONDS = 60
+CAPACITY_CALL_SECONDS = 540
+
+
+def capacity_available(clusters: set[str]) -> bool:
+    try:
+        return clusters <= fetch_capacity(Policy())
+    except ReadError:
+        return False
+
+
+def wait_for_capacity(clusters: set[str], ledger: Path, budget_minutes: int) -> int:
+    waited = json.loads(ledger.read_text())["waited-seconds"] if ledger.exists() else 0
+    deadline = time.monotonic() + CAPACITY_CALL_SECONDS
+    while True:
+        if capacity_available(clusters):
+            return 0
+        if waited + CAPACITY_POLL_SECONDS > budget_minutes * 60:
+            return 2
+        if time.monotonic() + CAPACITY_POLL_SECONDS > deadline:
+            return 1
+        waited += CAPACITY_POLL_SECONDS
+        ledger.write_text(json.dumps({"waited-seconds": waited}) + "\n")
+        time.sleep(CAPACITY_POLL_SECONDS)
 
 
 def observation_key(row: dict) -> tuple:
@@ -854,14 +880,19 @@ def main() -> int:
     )
     capacity = commands.add_parser(
         "check-capacity",
-        help="Exit 0 with available nodes below 80%% utilization; otherwise nonzero, without printing telemetry",
+        help="Exit 0 with available nodes below 80%% utilization; otherwise nonzero, without printing telemetry. With --wait: 1 means call again, 2 means the wait budget is spent",
     )
     capacity.add_argument(
         "--cluster",
         required=True,
         action="append",
         help="Exact telemetry cluster; repeat for every possible recipe target",
     )
+    capacity.add_argument(
+        "--wait",
+        action="store_true",
+        help="Poll for up to 9 minutes, charged to the KLAUD_CAPACITY_WAIT_MINUTES budget",
+    )
     commands.add_parser(
         "check-stop",
         help="Claude Stop hook: require finished runs and validated, closed or handed-off PRs",
@@ -1036,6 +1067,15 @@ def main() -> int:
                 parser.error("--max-candidates-per-run must be between 1 and 256")
             select(args.directory, args.max_candidates_per_run, args.execution_file)
             return 0
+        if args.wait:
+            budget = os.environ.get("KLAUD_CAPACITY_WAIT_MINUTES", "")
+            if not budget.isdigit() or "KLAUD_EVIDENCE" not in os.environ:
+                parser.error("--wait needs KLAUD_EVIDENCE and KLAUD_CAPACITY_WAIT_MINUTES")
+            return wait_for_capacity(
+                set(args.cluster),
+                Path(os.environ["KLAUD_EVIDENCE"]) / CAPACITY_LEDGER,
+                int(budget),
+            )
         return 0 if set(args.cluster) <= fetch_capacity(Policy()) else 1
     except ReadError as error:
         if args.command == "plan":

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

@@ -276,6 +276,36 @@ def pull(number, branch, updated):
     assert [pr["number"] for pr in klaud.active_pulls(pulls, now, 7)] == [1, 3, 4]
 
 
+def test_capacity_wait_charges_one_budget_across_calls(tmp_path, monkeypatch):
+    clock = [0.0]
+    polls = iter(
+        [api.ReadError("clusters"), set(), set(), {"other"}, {"mi355x"}]
+        + [set()] * 20
+    )
+
+    def fetch(_policy):
+        result = next(polls)
+        if isinstance(result, Exception):
+            raise result
+        return result
+
+    monkeypatch.setattr(klaud, "fetch_capacity", fetch)
+    monkeypatch.setattr(klaud.time, "monotonic", lambda: clock[0])
+    monkeypatch.setattr(klaud.time, "sleep", lambda seconds: clock.__setitem__(0, clock[0] + seconds))
+    monkeypatch.setattr(klaud, "CAPACITY_CALL_SECONDS", 180)
+    ledger = tmp_path / klaud.CAPACITY_LEDGER
+    wait = lambda: klaud.wait_for_capacity({"mi355x"}, ledger, 6)  # noqa: E731
+
+    assert wait() == 1
+    assert wait() == 0
+    assert json.loads(ledger.read_text()) == {"waited-seconds": 180}
+    assert wait() == 2
+    assert json.loads(ledger.read_text()) == {"waited-seconds": 360}
+    spent_at = clock[0]
+    assert wait() == 2
+    assert clock[0] == spent_at
+
+
 def test_select_continues_after_one_baseline_state_failure(tmp_path, monkeypatch):
     first_id = "1" * 16 + "-" + "2" * 16
     second_id = "3" * 16 + "-" + "4" * 16