fix(klaud): verify older and aggregate baselines, free abandoned work / fix(klaud): 验证旧版与聚合基线并释放放弃的工作

.github/workflows/klaud-candidate.yml

@@ -78,9 +78,14 @@ jobs:
       - name: Install Claude Code 2.1.282
         id: claude_cli
         run: | # zizmor: ignore[adhoc-packages] Claude CLI and its native packages are pinned to 2.1.282
-          npm install --prefix "$RUNNER_TEMP/claude-code" --no-audit --no-fund @anthropic-ai/claude-code@2.1.282
           claude_cli="$RUNNER_TEMP/claude-code/node_modules/.bin/claude"
-          test "$("$claude_cli" --version)" = "2.1.282 (Claude Code)"
+          for attempt in 1 2; do
+            rm -rf "$RUNNER_TEMP/claude-code"
+            npm install --prefix "$RUNNER_TEMP/claude-code" --no-audit --no-fund @anthropic-ai/claude-code@2.1.282 &&
+              [ "$("$claude_cli" --version 2>&1)" = "2.1.282 (Claude Code)" ] && break
+            [ "$attempt" = 2 ] && exit 1
+            sleep 15
+          done
           echo "path=$claude_cli" >> "$GITHUB_OUTPUT"
       - name: Let Klaud Cold own the candidate PR
         id: claude

.github/workflows/klaud-plan.yml

@@ -136,9 +136,14 @@ jobs:
         id: claude_cli
         if: steps.prepare.outputs.has_candidates == 'true'
         run: | # zizmor: ignore[adhoc-packages] Claude CLI and its native packages are pinned to 2.1.282
-          npm install --prefix "$RUNNER_TEMP/claude-code" --no-audit --no-fund @anthropic-ai/claude-code@2.1.282
           claude_cli="$RUNNER_TEMP/claude-code/node_modules/.bin/claude"
-          test "$("$claude_cli" --version)" = "2.1.282 (Claude Code)"
+          for attempt in 1 2; do
+            rm -rf "$RUNNER_TEMP/claude-code"
+            npm install --prefix "$RUNNER_TEMP/claude-code" --no-audit --no-fund @anthropic-ai/claude-code@2.1.282 &&
+              [ "$("$claude_cli" --version 2>&1)" = "2.1.282 (Claude Code)" ] && break
+            [ "$attempt" = 2 ] && exit 1
+            sleep 15
+          done
           echo "path=$claude_cli" >> "$GITHUB_OUTPUT"
       - name: Check for overlapping open PRs
         id: review

inferencex-e2e/infx/klaud/__main__.py

@@ -35,7 +35,7 @@
 )
 
 PRODUCERS = "producers.json"
-RETRY_NEXT_WAVE = frozenset({"capacity-deferred"})
+RETRY_NEXT_WAVE = frozenset({"capacity-deferred", "agent-not-started"})
 
 
 def observation_key(row: dict) -> tuple:
@@ -203,10 +203,10 @@ 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
+            diagnostics = json.loads((Path(scratch) / "candidate-diagnostics.json").read_text())
+            if diagnostics.get("action-outcome") == "skipped":
+                return "agent-not-started"
+            return CandidateOutcome.model_validate(diagnostics["candidate-outcome"]).outcome
         except (
             OSError,
             ValueError,
@@ -662,6 +662,7 @@ def select(directory: Path, max_candidates: int, execution_file: Path | None = N
             deferred = "capacity-unavailable"
     capacity_deferred = []
     baseline_deferred = []
+    baseline_reasons: dict[str, str] = {}
     baseline_mismatches = []
     preflights = {}
     families = {decision.family for decision in review.decisions if decision.decision != "proceed"}
@@ -700,16 +701,24 @@ def select(directory: Path, max_candidates: int, execution_file: Path | None = N
                 ),
                 baseline,
             )
-        except VerificationError:
+        except (VerificationError, ReadError) as error:
             baseline_deferred.append(candidate["id"])
+            baseline_reasons[candidate["id"]] = str(error)
             families.add(decision.family)
             continue
-        except (OSError, ValueError, KeyError, TypeError, subprocess.SubprocessError, ReadError):
+        except subprocess.SubprocessError as error:
             baseline_deferred.append(candidate["id"])
+            baseline_reasons[candidate["id"]] = type(error).__name__
+            families.add(decision.family)
+            continue
+        except (OSError, ValueError, KeyError, TypeError) as error:
+            baseline_deferred.append(candidate["id"])
+            baseline_reasons[candidate["id"]] = f"{type(error).__name__}: {str(error)[:300]}"
             families.add(decision.family)
             continue
         if missing:
             baseline_deferred.append(candidate["id"])
+            baseline_reasons[candidate["id"]] = "baseline-point-mismatch"
             baseline_mismatches.append(
                 {
                     "id": candidate["id"],
@@ -759,6 +768,7 @@ def select(directory: Path, max_candidates: int, execution_file: Path | None = N
                 "deferred-reason": deferred,
                 "capacity-deferred-candidates": capacity_deferred,
                 "baseline-deferred-candidates": baseline_deferred,
+                "baseline-deferral-reasons": baseline_reasons,
                 "baseline-mismatch-candidates": baseline_mismatches,
                 **review.model_dump(by_alias=True),
             },
@@ -779,7 +789,13 @@ def select(directory: Path, max_candidates: int, execution_file: Path | None = N
             f" {len(capacity_deferred)} reviewed candidates deferred by the latest capacity check."
         )
     if baseline_deferred:
-        summary += f" {len(baseline_deferred)} candidates lacked a verifiable full baseline."
+        summary += f" {len(baseline_deferred)} candidates lacked a verifiable full baseline: " + (
+            "; ".join(
+                f"{count}× {reason[:120]}"
+                for reason, count in Counter(baseline_reasons.values()).most_common()
+            )
+            + "."
+        )
     if baseline_mismatches:
         summary += (
             f" {len(baseline_mismatches)} of them have frozen baseline points"

inferencex-e2e/infx/klaud/lifecycle.py

@@ -4,18 +4,23 @@
 
 import json
 import os
+import re
 import subprocess
 import tempfile
 from pathlib import Path
 from urllib.parse import urlencode
 
 from . import claims, github
 from .github import VerificationError
-from .models import CandidateOutcome, OwnedCandidate, Ownership, utc
+from .models import CandidateOutcome, OwnedCandidate, Ownership, identity, utc
 from .reporting import translated
 from .validation import verify_sweep
 
 BOT = "Klaud-Cold"
+COMMITTER = "github-actions[bot]"
+ACTIVE_STATUSES = ("requested", "waiting", "pending", "queued", "in_progress")
+CANDIDATE_ID = r"[0-9a-f]{16}-[0-9a-f]{16}"
+RECONCILE_JOB = "Reconcile interrupted candidates"
 SWEEP_LABELS = {
     "full-sweep-enabled",
     "non-canary-full-sweep-enabled",
@@ -87,7 +92,8 @@ def pulls(self) -> list[dict]:
         return pulls
 
     def runs(self) -> list[dict]:
-        query = {"per_page": 100, "created": ">=" + self.parent["created_at"]}
+        end = self.parent["updated_at"] if terminal(self.parent) else "*"
+        query = {"per_page": 100, "created": f"{self.parent['created_at']}..{end}"}
         targeted = github.items(
             self.repository,
             "actions/workflows/e2e-tests.yml/runs?"
@@ -105,7 +111,14 @@ def runs(self) -> list[dict]:
         sweeps = github.items(
             self.repository,
             "actions/workflows/run-sweep.yml/runs?"
-            + urlencode({**query, "event": "pull_request", "branch": self.branch}),
+            + urlencode(
+                {
+                    "per_page": 100,
+                    "created": ">=" + self.parent["created_at"],
+                    "event": "pull_request",
+                    "branch": self.branch,
+                }
+            ),
             "workflow_runs",
         )
         pulls = self.pulls()
@@ -531,6 +544,7 @@ def recover() -> None:
         raise VerificationError("Untrusted ownership parent")
     results = {}
     blocked = set()
+    live = set()
     for owner, _ in sorted(owners, key=lambda pair: pair[0].run_id, reverse=True):
         parent = parents[owner.run_id]
         for candidate in owner.candidates:
@@ -540,11 +554,13 @@ def recover() -> None:
             results[key] = False
             if not terminal(parent):
                 blocked.add(candidate.family)
+                live.add(candidate.id)
                 continue
             session = Session(repository, parent, candidate, recovering=True)
             try:
                 if not claims.recovery_lease(repository, parent["id"], candidate):
                     blocked.add(candidate.family)
+                    live.add(candidate.id)
                     continue
                 changed = reconcile(session)
                 claims.release_family(repository, candidate, parent["id"])
@@ -582,6 +598,136 @@ def recover() -> None:
             current = github.read(repository, f"actions/runs/{owner.run_id}")
             if terminal(current) and current["run_attempt"] == parents[owner.run_id]["run_attempt"]:
                 github.write(repository, f"actions/artifacts/{artifact['id']}", "DELETE")
+    try:
+        live |= sweep_abandoned(repository, live)
+    except (OSError, ValueError, KeyError, TypeError, subprocess.SubprocessError) as error:
+        print(f"::warning::Klaud abandoned-work sweep skipped: {type(error).__name__}")
+    else:
+        prefixes = {candidate.split("-", 1)[0] for candidate in live}
+        blocked = {family for family in blocked if identity(family)[:16] in prefixes}
     (Path(os.environ["RUNNER_TEMP"]) / "klaud-recovery.json").write_text(
         json.dumps(sorted(blocked))
     )
+
+
+def owned_activity(repository: str) -> tuple[set[str], dict[str, list[int]], set[int]]:
+    live: set[str] = set()
+    targeted: dict[str, list[int]] = {}
+    planners: set[int] = set()
+    for status in ACTIVE_STATUSES:
+        query = urlencode({"status": status, "per_page": 100})
+        for run in github.items(
+            repository, f"actions/workflows/e2e-tests.yml/runs?{query}", "workflow_runs"
+        ):
+            match = re.fullmatch(rf"e2e Test - klaud-\d+-({CANDIDATE_ID})", run["display_title"])
+            if match and run["actor"]["login"] == BOT:
+                targeted.setdefault(match[1], []).append(run["id"])
+        for run in github.items(
+            repository, f"actions/workflows/run-sweep.yml/runs?{query}", "workflow_runs"
+        ):
+            if match := re.fullmatch(rf"klaud/auto-({CANDIDATE_ID})", run["head_branch"] or ""):
+                live.add(match[1])
+        for run in github.items(
+            repository, f"actions/workflows/klaud-plan.yml/runs?{query}", "workflow_runs"
+        ):
+            for job in github.items(
+                repository, f"actions/runs/{run['id']}/jobs?per_page=100", "jobs"
+            ):
+                if match := re.search(rf"Run candidate ({CANDIDATE_ID})", job["name"]):
+                    live.add(match[1])
+                elif job["name"] == RECONCILE_JOB and job["status"] != "completed":
+                    planners.add(run["id"])
+    return live, targeted, planners
+
+
+def bot_committed(commits: list[dict]) -> bool:
+    return bool(commits) and all(
+        (commit.get("committer") or {}).get("login") == COMMITTER for commit in commits
+    )
+
+
+def sweep_abandoned(repository: str, live: set[str]) -> set[str]:
+    sessions, targeted, planners = owned_activity(repository)
+    live = live | sessions
+    pulls = github.items(repository, "pulls?state=open&per_page=100")
+    abandoned = []
+    for pull in pulls:
+        match = re.fullmatch(rf"klaud/auto-({CANDIDATE_ID})", pull["head"]["ref"])
+        if not match or match[1] in live:
+            continue
+        if (
+            pull["draft"]
+            and pull["user"]["login"] == BOT
+            and (pull["head"]["repo"] or {}).get("full_name") == repository
+            and not any(label["name"] == "klaud-handoff" for label in pull["labels"])
+            and bot_committed(
+                github.items(repository, f"pulls/{pull['number']}/commits?per_page=100")
+            )
+        ):
+            abandoned.append(pull)
+        else:
+            live.add(match[1])
+    for candidate, runs in targeted.items():
+        if candidate not in live:
+            for run in runs:
+                github.write(repository, f"actions/runs/{run}/cancel", "POST")
+    for owner in claims.family_owners(repository):
+        candidate = owner.candidates[0]
+        if candidate.id not in live and terminal(
+            github.read(repository, f"actions/runs/{owner.run_id}")
+        ):
+            claims.release_family(repository, candidate, owner.run_id)
+    if planners <= {int(os.environ.get("GITHUB_RUN_ID", "0"))}:
+        finished: dict[int, bool] = {}
+        for lease in github.items(repository, "git/matching-refs/heads/klaud/recovery-"):
+            match = re.fullmatch(rf"refs/heads/klaud/recovery-(\d+)-({CANDIDATE_ID})", lease["ref"])
+            if not match or match[2] in live:
+                continue
+            parent = int(match[1])
+            if parent not in finished:
+                finished[parent] = terminal(github.read(repository, f"actions/runs/{parent}"))
+            if finished[parent]:
+                github.write(repository, "git/" + lease["ref"], "DELETE")
+    for pull in abandoned:
+        github.write(
+            repository,
+            f"issues/{pull['number']}/comments",
+            "POST",
+            {
+                "body": translated(
+                    "**abandoned** · The owning Klaud session ended without cleanup. Closed, "
+                    "with any leftover smoke runs cancelled, so a new candidate can take this "
+                    "family.",
+                    "**abandoned** · 所属 Klaud 会话已结束但未完成清理。已关闭并取消残留的冒烟运行，"
+                    "新的候选可以接手该配置族。",
+                )
+            },
+        )
+        github.write(repository, f"pulls/{pull['number']}", "PATCH", {"state": "closed"})
+        github.write(repository, f"git/refs/heads/{pull['head']['ref']}", "DELETE")
+    open_heads = {pull["head"]["ref"] for pull in pulls}
+    for ref in github.items(repository, "git/matching-refs/heads/klaud/auto-"):
+        branch = ref["ref"].removeprefix("refs/heads/")
+        match = re.fullmatch(rf"klaud/auto-({CANDIDATE_ID})", branch)
+        if not match or match[1] in live or branch in open_heads:
+            continue
+        history = github.items(
+            repository,
+            "pulls?"
+            + urlencode(
+                {
+                    "state": "all",
+                    "head": repository.split("/", maxsplit=1)[0] + ":" + branch,
+                    "per_page": 100,
+                }
+            ),
+        )
+        if history:
+            latest = max(history, key=lambda pull: pull["created_at"])
+            closer = github.read(repository, f"issues/{latest['number']}").get("closed_by") or {}
+            if latest["user"]["login"] != BOT or closer.get("login") != BOT:
+                continue
+        elif not bot_committed([github.read(repository, f"commits/{ref['object']['sha']}")]):
+            continue
+        github.write(repository, f"git/refs/heads/{branch}", "DELETE")
+    return live

inferencex-e2e/infx/klaud/reporting.py

@@ -869,8 +869,9 @@ def public_point(entry: dict) -> dict:
         "conc": int(entry["conc"]),
         "image": normalized_image(entry["image"]),
     }
+    aggregate = multi and not point["disagg"] and not entry["decode"].get("num-worker")
     for role in ("prefill", "decode"):
-        topology = entry[role] if multi else entry
+        topology = entry["prefill" if aggregate else role] if multi else entry
         point.update(
             {
                 f"{role}_tp": topology["tp"],
@@ -886,14 +887,20 @@ def _matches_public_point(row: dict, point: dict) -> bool:
     """Compare a public row with a generated identity using canonical image spelling."""
     from .models import normalized_image
 
-    for key, value in point.items():
-        observed = row.get(key)
-        if key == "image":
-            if not isinstance(observed, str) or normalized_image(observed) != value:
-                return False
-        elif observed != value:
-            return False
-    return True
+    encodings = [point]
+    if point["is_multinode"] and not point["disagg"] and point["prefill_num_workers"] == 1:
+        encodings.append(
+            {**point, "is_multinode": False, "prefill_num_workers": 0, "decode_num_workers": 0}
+        )
+    return any(
+        all(
+            isinstance(row.get(key), str) and normalized_image(row[key]) == value
+            if key == "image"
+            else row.get(key) == value
+            for key, value in encoding.items()
+        )
+        for encoding in encodings
+    )
 
 
 def matrix_points(matrix: dict) -> list[dict]: