From 1f5d20c024b421f95bc6264eeda6e2c5e7ad8329 Mon Sep 17 00:00:00 2001 From: adibarra <93070681+adibarra@users.noreply.github.com> Date: Wed, 14 Oct 2026 18:10:28 -0500 Subject: [PATCH 1/2] fix(klaud): unblock multi-node AgentX smoke and stop re-picking blocked candidates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Smoke eval rows no longer carry eval-all-concs, which the multi-node AgentX dispatch schema rejects; 65 live families failed get-jobs before any GPU work. - The cooldown now spans every base and skips candidates whose last attempt was not a capacity deferral, instead of only reordering same-SHA attempts. - The overlap index drops non-Klaud PRs idle for 7 days, and the review treats cross-cutting PRs as overlap only when they touch the family's image or flags. - Planner waves run every 2 hours instead of every 6. - 冒烟评测行不再携带 eval-all-concs，多节点 AgentX 调度 schema 会拒绝该字段； 此前 65 个现行配置族在任何 GPU 任务开始前就在 get-jobs 失败。 - 冷却期现在覆盖所有基准提交，并跳过最近一次尝试不是容量推迟的候选， 而不是只对同一 SHA 的尝试重新排序。 - 重叠索引会剔除 7 天无活动的非 Klaud PR；评审仅在跨领域 PR 修改该配置族的 镜像或引擎参数时才视为重叠。 - 规划任务改为每 2 小时运行一次，而不是每 6 小时。 --- .github/workflows/klaud-plan.yml | 18 ++- inferencex-e2e/infx/klaud/__main__.py | 128 ++++++++++++++---- inferencex-e2e/infx/matrix/generate.py | 2 +- .../infx/tests/klaud/test_klaud_github.py | 82 ++++++++--- .../infx/tests/matrix/test_validation.py | 12 ++ 5 files changed, 190 insertions(+), 52 deletions(-) diff --git a/.github/workflows/klaud-plan.yml b/.github/workflows/klaud-plan.yml index f8e3251e4f..2df17ad7a1 100644 --- a/.github/workflows/klaud-plan.yml +++ b/.github/workflows/klaud-plan.yml @@ -4,7 +4,8 @@ name: Klaud Cold auto-sweep on: # zizmor: ignore[concurrency-limits] workflow_dispatch: schedule: - - cron: '0 */6 * * *' + # Frequent waves catch short idle-capacity windows; the capacity gate and claims bound load. + - cron: '0 */2 * * *' permissions: contents: read @@ -81,9 +82,11 @@ 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. + # Bound one semantic-overlap review. Attempted candidates skip the cooldown unless only + # capacity deferred them; other open PRs idle past the stale window stop owning families. 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 +125,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 +167,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 +191,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). diff --git a/inferencex-e2e/infx/klaud/__main__.py b/inferencex-e2e/infx/klaud/__main__.py index c58ff6e796..b86244413d 100644 --- a/inferencex-e2e/infx/klaud/__main__.py +++ b/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,8 @@ ) PRODUCERS = "producers.json" +# Outcomes that say nothing about the candidate itself, so the next wave may retry them. +RETRY_NEXT_WAVE = frozenset({"capacity-deferred"}) def observation_key(row: dict) -> tuple: @@ -152,8 +160,12 @@ 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]: + """Map candidates given an agent within the cooldown to their newest recorded outcome. + + Every base counts: main moves faster than the cooldown, so a same-base filter never matched. + An attempt without a readable verified outcome maps to ``None``. + """ cutoff = datetime.now(UTC) - timedelta(hours=cooldown_hours) try: runs = github_items( @@ -163,38 +175,86 @@ 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: + """The verified outcome in a candidate diagnostics artifact, or ``None`` if unreadable.""" + 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]: + """Skip recently attempted candidates; capacity deferrals only sort last. + + A deterministic block (missing weights, a broken matrix, a failed sweep) repeats on every + wave, so it waits out the cooldown instead of taking a slot from an untried candidate. + """ + 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]: + """Open PRs that can own a family's refresh in the overlap review. + + A PR idle for ``stale_days`` no longer blocks families; Klaud's own candidate PRs always + count, because they hold a validated or in-flight refresh awaiting review. + """ + 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 +287,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 +315,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 +841,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 +1035,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) diff --git a/inferencex-e2e/infx/matrix/generate.py b/inferencex-e2e/infx/matrix/generate.py index 1d5b7ffeda..5865039af3 100644 --- a/inferencex-e2e/infx/matrix/generate.py +++ b/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) diff --git a/inferencex-e2e/infx/tests/klaud/test_klaud_github.py b/inferencex-e2e/infx/tests/klaud/test_klaud_github.py index 6def2cf6c5..6fa0043a69 100644 --- a/inferencex-e2e/infx/tests/klaud/test_klaud_github.py +++ b/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): diff --git a/inferencex-e2e/infx/tests/matrix/test_validation.py b/inferencex-e2e/infx/tests/matrix/test_validation.py index 3509be0141..b71351b7d3 100644 --- a/inferencex-e2e/infx/tests/matrix/test_validation.py +++ b/inferencex-e2e/infx/tests/matrix/test_validation.py @@ -1335,6 +1335,18 @@ 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): + """Klaud dispatches --smoke for every family; each row must reach the GPU jobs.""" + 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: From e322315b6050a1b309c047290ff96cd58053216b Mon Sep 17 00:00:00 2001 From: adibarra <93070681+adibarra@users.noreply.github.com> Date: Wed, 14 Oct 2026 18:27:23 -0500 Subject: [PATCH 2/2] chore(klaud): drop added comments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 删除新增的注释。 --- .github/workflows/klaud-plan.yml | 3 --- inferencex-e2e/infx/klaud/__main__.py | 17 ----------------- .../infx/tests/matrix/test_validation.py | 1 - 3 files changed, 21 deletions(-) diff --git a/.github/workflows/klaud-plan.yml b/.github/workflows/klaud-plan.yml index 2df17ad7a1..bd1d5c6ecc 100644 --- a/.github/workflows/klaud-plan.yml +++ b/.github/workflows/klaud-plan.yml @@ -4,7 +4,6 @@ name: Klaud Cold auto-sweep on: # zizmor: ignore[concurrency-limits] workflow_dispatch: schedule: - # Frequent waves catch short idle-capacity windows; the capacity gate and claims bound load. - cron: '0 */2 * * *' permissions: @@ -82,8 +81,6 @@ jobs: actions: read # Read workflow runs and artifacts. pull-requests: read # Read PR metadata. env: - # Bound one semantic-overlap review. Attempted candidates skip the cooldown unless only - # capacity deferred them; other open PRs idle past the stale window stop owning families. REVIEW_BATCH_SIZE: '64' CANDIDATE_COOLDOWN_HOURS: '24' STALE_PR_DAYS: '7' diff --git a/inferencex-e2e/infx/klaud/__main__.py b/inferencex-e2e/infx/klaud/__main__.py index b86244413d..1c6c7aa46a 100644 --- a/inferencex-e2e/infx/klaud/__main__.py +++ b/inferencex-e2e/infx/klaud/__main__.py @@ -35,7 +35,6 @@ ) PRODUCERS = "producers.json" -# Outcomes that say nothing about the candidate itself, so the next wave may retry them. RETRY_NEXT_WAVE = frozenset({"capacity-deferred"}) @@ -161,11 +160,6 @@ def choose( def recent_candidate_outcomes(repository: str, cooldown_hours: int) -> dict[str, str | None]: - """Map candidates given an agent within the cooldown to their newest recorded outcome. - - Every base counts: main moves faster than the cooldown, so a same-base filter never matched. - An attempt without a readable verified outcome maps to ``None``. - """ cutoff = datetime.now(UTC) - timedelta(hours=cooldown_hours) try: runs = github_items( @@ -206,7 +200,6 @@ def recent_candidate_outcomes(repository: str, cooldown_hours: int) -> dict[str, def recorded_outcome(repository: str, artifact: dict) -> str | None: - """The verified outcome in a candidate diagnostics artifact, or ``None`` if unreadable.""" with tempfile.TemporaryDirectory() as scratch: try: download_json(repository, artifact, Path(scratch)) @@ -225,11 +218,6 @@ def recorded_outcome(repository: str, artifact: dict) -> str | None: def order_by_cooldown(candidates: list[dict], recent: dict[str, str | None]) -> list[dict]: - """Skip recently attempted candidates; capacity deferrals only sort last. - - A deterministic block (missing weights, a broken matrix, a failed sweep) repeats on every - wave, so it waits out the cooldown instead of taking a slot from an untried candidate. - """ 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 @@ -238,11 +226,6 @@ def order_by_cooldown(candidates: list[dict], recent: dict[str, str | None]) -> def active_pulls(pulls: list[dict], now: datetime, stale_days: int) -> list[dict]: - """Open PRs that can own a family's refresh in the overlap review. - - A PR idle for ``stale_days`` no longer blocks families; Klaud's own candidate PRs always - count, because they hold a validated or in-flight refresh awaiting review. - """ cutoff = now - timedelta(days=stale_days) return [ pr diff --git a/inferencex-e2e/infx/tests/matrix/test_validation.py b/inferencex-e2e/infx/tests/matrix/test_validation.py index b71351b7d3..ac78746d81 100644 --- a/inferencex-e2e/infx/tests/matrix/test_validation.py +++ b/inferencex-e2e/infx/tests/matrix/test_validation.py @@ -1336,7 +1336,6 @@ def test_validate_agentic_matrix_entry_dispatches_on_prefill_key(self): validate_agentic_matrix_entry(without_prefill) def test_smoke_rows_satisfy_the_dispatch_schema(self): - """Klaud dispatches --smoke for every family; each row must reach the GPU jobs.""" from infx.matrix.generate import smoke_entries row = {**MULTINODE_AGENTIC_EVAL_ROW, "conc": [8, 32]}