diff --git a/.github/workflows/klaud-plan.yml b/.github/workflows/klaud-plan.yml index f8e3251e4f..bd1d5c6ecc 100644 --- a/.github/workflows/klaud-plan.yml +++ b/.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). diff --git a/inferencex-e2e/infx/klaud/__main__.py b/inferencex-e2e/infx/klaud/__main__.py index c58ff6e796..1c6c7aa46a 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,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) 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..ac78746d81 100644 --- a/inferencex-e2e/infx/tests/matrix/test_validation.py +++ b/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: