diff --git a/.github/klaud-candidate-prompt.md b/.github/klaud-candidate-prompt.md index 238a763d51..0df0aefb0e 100644 --- a/.github/klaud-candidate-prompt.md +++ b/.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/ diff --git a/.github/workflows/klaud-candidate.yml b/.github/workflows/klaud-candidate.yml index a177a9b064..6435cd50d2 100644 --- a/.github/workflows/klaud-candidate.yml +++ b/.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: diff --git a/inferencex-e2e/infx/klaud/__main__.py b/inferencex-e2e/infx/klaud/__main__.py index 1c6c7aa46a..8a1ab40f10 100644 --- a/inferencex-e2e/infx/klaud/__main__.py +++ b/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,7 +880,7 @@ 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", @@ -862,6 +888,11 @@ def main() -> int: 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": diff --git a/inferencex-e2e/infx/tests/klaud/test_klaud_github.py b/inferencex-e2e/infx/tests/klaud/test_klaud_github.py index 6fa0043a69..65bac14b33 100644 --- a/inferencex-e2e/infx/tests/klaud/test_klaud_github.py +++ b/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