From 4877a30dc0a1eaa464cafbf428592ee4b07479d5 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Sun, 6 Sep 2026 03:27:42 +0800 Subject: [PATCH] rebase: improve CI baseline reconciliation Signed-off-by: tzhouam --- .../SPEC/engine/steps/rebase_v3.md | 4 +- doc/architecture/SPEC/rebase_engine.md | 9 +- .../engine/steps/rebase_v3.py | 1 + .../rebase_engine/ci_loop.py | 149 ++++++++++++------ test/test_ci_wiring.py | 71 ++++++++- 5 files changed, 183 insertions(+), 51 deletions(-) diff --git a/doc/architecture/SPEC/engine/steps/rebase_v3.md b/doc/architecture/SPEC/engine/steps/rebase_v3.md index e764ac87f..f58cb8aa3 100644 --- a/doc/architecture/SPEC/engine/steps/rebase_v3.md +++ b/doc/architecture/SPEC/engine/steps/rebase_v3.md @@ -1,6 +1,6 @@ # engine/steps/rebase_v3.py —— 规范 - + `LOC ~2204 · step 库（v3 rebase 装配层） · refactor-status: oversized` @@ -21,7 +21,7 @@ step —— 薄的受治理 wrapper，substate-first、类型化失败、发布 | `rebase.v3_test_loop` | script/write_workspace | `phase3_failed`；substate `tests.pipeline`/`infra_failures`；空 manifest ⇒ `manifest_empty` | | `rebase.v3_precommit` | script/write_workspace | 无发布 —— substate `tests.precommit`（passed/failed/failed_preexisting/not_declared） | | `rebase.v3_push_gate` | deterministic/read | `push_gate_flagged`、`push_gate_overrides`；不许 ⇒ FORBIDDEN；override 记 trace | -| `rebase.v3_ci` | script/push | `ci_result`、`ci_build_urls`；substate `ci`（rounds/adopted/unfixed） | +| `rebase.v3_ci` | script/push | `ci_result`、`ci_build_urls`；substate `ci`（rounds/adopted/unfixed，以及逐 job 分类证据） | | `rebase.v3_finalize` | deterministic/read | 无发布 —— substate `phase=done|needs_human`；有失败 ⇒ BLOCKED（复用 exit 3） | ## 公开契约（注册的 step 之外） diff --git a/doc/architecture/SPEC/rebase_engine.md b/doc/architecture/SPEC/rebase_engine.md index 70eb52663..f778c4e4a 100644 --- a/doc/architecture/SPEC/rebase_engine.md +++ b/doc/architecture/SPEC/rebase_engine.md @@ -1,6 +1,6 @@ # rebase_engine/ —— 规范 - + `LOC ~7500（26 个模块） · repo-rebase-v3 的原生 rebase 引擎 · refactor-status: ok` @@ -86,6 +86,13 @@ runner/LLM/CI client 全部可注入 —— 每个模块都能离线测试。 shell 的 venv 状态绝不影响装到哪；只有 import 验证失败才重试。 - CI 创建受守卫：durable op 记录先于 API 调用，恢复按 op id 精确匹配， **不确定时绝不重复创建**。 +- CI baseline 使用最高可信来源（schedule > API > other）的最近窗口：最新 + 构建失败立即生效，较旧失败只有在窗口内重复出现才保留；单次陈旧红构建不会 + 越过更新的绿构建复活。 +- baseline 分类以同名 job 的日志根因为身份，provider exit code 只作候选排序； + 支持 `FAILED`/`ERROR` pytest node 的原因相关子集，但缺坐标、日志、当前签名 + 或共同异常证据时一律 fail-closed。每个 job 的分类理由与 baseline build/job + 坐标随 round 持久化，供恢复和审计使用。 - **A5** —— 全包仓库中立：仓库值经 `WheelSpec`/`PinSpec`/`ManifestSpec`/ `ModulePromptData`/`tool_schemas.json`/hooks 注入；`test_repo_neutral_core` 钉住。 diff --git a/src/infermatrix_copilot/engine/steps/rebase_v3.py b/src/infermatrix_copilot/engine/steps/rebase_v3.py index 5bf85f073..d998c1086 100644 --- a/src/infermatrix_copilot/engine/steps/rebase_v3.py +++ b/src/infermatrix_copilot/engine/steps/rebase_v3.py @@ -2171,6 +2171,7 @@ async def _abort_ci_cleanup(outcome, _client=client, _ops=ops_dir): "budget_timeouts": list(r.budget_timeouts), "ignored": r.ignored, "ignored_baseline": r.ignored_baseline, + "jobs": list(r.jobs), } for r in result.rounds]}}) updates = {"ci_result": result.result, "ci_build_urls": [r.build_url for r in result.rounds diff --git a/src/infermatrix_copilot/rebase_engine/ci_loop.py b/src/infermatrix_copilot/rebase_engine/ci_loop.py index b1ffeb350..940a71aed 100644 --- a/src/infermatrix_copilot/rebase_engine/ci_loop.py +++ b/src/infermatrix_copilot/rebase_engine/ci_loop.py @@ -221,6 +221,9 @@ class JobResult: classification: str = "" # passed|failed|ignored|ignored_baseline| # budget_timeout|incomplete|retrying log_file: str = "" + classification_reason: str = "" + baseline_build_id: str = "" + baseline_job_id: str = "" @dataclass @@ -325,21 +328,26 @@ def _save_job_log(log_dir, name: str, text: str) -> str: return str(path) -def _baseline_match(spec: CIClassifySpec, name: str, - exit_status: int) -> BaselineFailure | None: +def _baseline_matches(spec: CIClassifySpec, name: str, + exit_status: int) -> tuple[BaselineFailure, ...]: + """Same-name baseline candidates, preferring the same provider exit code. + + Exit status is supporting evidence, not failure identity: a real pytest + failure can be followed by a teardown timeout and therefore land as -1 on + main but 1 on the branch. Root-cause comparison below remains mandatory. + """ lower = name.lower() coded = exit_status or -1 - for bf in spec.baseline: - if bf.name == lower and bf.exit_status == coded: - return bf - return None + matches = [bf for bf in spec.baseline if bf.name == lower] + return tuple(sorted(matches, key=lambda bf: bf.exit_status != coded)) def _same_root_cause(client: CIClient, entry: BaselineFailure, - log_text: str, spec: CIClassifySpec, log_dir) -> bool: + log_text: str, spec: CIClassifySpec, log_dir) -> str: """Root-cause comparison against the baseline job's log: identical - normalized error signatures, or identical (non-empty) failing pytest - node-id sets, mean the same pre-existing failure. + normalized error signatures, or cause-correlated pytest node-id subsets, + mean the same pre-existing failure. The returned string is durable + evidence for the classification; an empty string means no safe match. FAIL-CLOSED on missing evidence (round-4 review — a deliberate divergence from the parent's lenient defaults): missing baseline @@ -348,22 +356,32 @@ def _same_root_cause(client: CIClient, entry: BaselineFailure, comparison CANNOT be made — the failure stays actionable rather than being waved through as pre-existing.""" if not entry.job_id or not entry.build_id: - return False + return "" fetch = spec.baseline_log_fn or client.get_job_log main_log = fetch(entry.build_id, entry.job_id) if not main_log: - return False - _save_job_log(log_dir, f"baseline_{entry.name}", main_log) + return "" + _save_job_log(log_dir, + f"baseline_{entry.build_id}_{entry.job_id}_{entry.name}", + main_log) cur_sig = extract_error_signature(log_text or "", spec.extra_exception_names) if not cur_sig: - return False + return "" main_sig = extract_error_signature(main_log, spec.extra_exception_names) if cur_sig == main_sig: - return True + return "identical normalized error signature" cur_tests = extract_failed_test_ids(log_text or "") main_tests = extract_failed_test_ids(main_log) - return bool(cur_tests) and cur_tests == main_tests + if not cur_tests or not cur_tests <= main_tests: + return "" + cur_errors = extract_exception_lines(log_text, spec.extra_exception_names) + main_errors = extract_exception_lines(main_log, + spec.extra_exception_names) + if cur_errors & main_errors: + return ("current pytest failure/error nodes are a cause-matched " + "subset of the baseline") + return "" def _classify_terminal(client: CIClient, build_id: str, job: dict, @@ -416,11 +434,17 @@ def _classify_terminal(client: CIClient, build_id: str, job: dict, elif jr.state == "timed_out" and is_budget_timeout(log_text): jr.classification = "budget_timeout" else: - entry = _baseline_match(spec, name, jr.exit_status) - if entry is not None and _same_root_cause(client, entry, - log_text, spec, - log_dir): + match = next(((entry, reason) + for entry in _baseline_matches( + spec, name, jr.exit_status) + if (reason := _same_root_cause( + client, entry, log_text, spec, log_dir))), None) + if match is not None: + entry, reason = match jr.classification = "ignored_baseline" + jr.classification_reason = reason + jr.baseline_build_id = entry.build_id + jr.baseline_job_id = entry.job_id else: jr.classification = "failed" return jr @@ -615,10 +639,11 @@ def normalize_log_line(line: str) -> str: def extract_failed_test_ids(log_text: str) -> set[str]: - """pytest node ids (`path::Class::test[param]`) from FAILED lines.""" + """Pytest node ids from terminal FAILED and setup/teardown ERROR lines.""" ids: set[str] = set() for line in log_text.split("\n"): - if "FAILED" not in line or "::" not in line: + if not any(token in line for token in ("FAILED", "ERROR")) \ + or "::" not in line: continue m = re.search(r"([\w./-]+\.py::[\w:]+(?:\[[^\]]*\])?)", normalize_log_line(line)) @@ -627,19 +652,30 @@ def extract_failed_test_ids(log_text: str) -> set[str]: return ids -def extract_error_signature(log_text: str, - extra_exception_names: Sequence[str] = ()) -> str: - """Key error lines, normalized so two runs of one failure compare equal. - Repo-specific exception class names arrive as adapter data.""" +def extract_exception_lines( + log_text: str, + extra_exception_names: Sequence[str] = ()) -> set[str]: + """All normalized exception lines used to corroborate pytest node matches.""" base = (r"RuntimeError|AssertionError|TypeError|ValueError" r"|ModuleNotFoundError|ImportError|CalledProcessError" r"|torch\.OutOfMemoryError|CUDA out of memory" r"|exit status|SyntaxError|KeyError|AttributeError") names = "|".join([base, *map(re.escape, extra_exception_names)]) rx = re.compile(rf".*({names}).*") + return {line for line in ( + normalize_log_line(raw) for raw in log_text.split("\n")) + if line and rx.match(line)} + + +def extract_error_signature(log_text: str, + extra_exception_names: Sequence[str] = ()) -> str: + """Key error lines, normalized so two runs of one failure compare equal. + Repo-specific exception class names arrive as adapter data.""" lines = log_text.split("\n") + exception_lines = extract_exception_lines(log_text, + extra_exception_names) sig_lines = [n for n in (normalize_log_line(ln) for ln in lines) - if n and rx.match(n)] + if n in exception_lines] failures = [normalize_log_line(ln) for ln in lines if "FAILED" in ln and "::" in ln] result = sig_lines[-3:] if sig_lines else [] @@ -677,35 +713,54 @@ def pick_best_build(builds: Sequence[dict]) -> dict | None: return scored[0][2] -def fetch_baseline_failures(client: CIClient, - branch: str) -> tuple[BaselineFailure, ...]: +def fetch_baseline_failures(client: CIClient, branch: str, *, + window: int = 5) -> tuple[BaselineFailure, ...]: """Pre-existing failures on the baseline pipeline's `branch`, with coordinates so the monitor can root-cause-compare logs. The CLIENT is scoped to the BASELINE pipeline (adapter data), not the build-under-test's. - The best trustworthy COMPLETED build is selected first (schedule > - api > completed > newest) and failures are extracted only when THAT - build failed — querying failed builds alone would resurrect stale - failures that a newer green baseline run already proved fixed - (round-4 review).""" + Use a window from the best available trust tier (schedule > api > other). + A failure from the newest trusted build is current baseline evidence. + Older failures survive a newer green only when they recur in at least two + builds, distinguishing an intermittent baseline from one stale red run. + """ + if window < 1: + raise ValueError("baseline window must be positive") builds = client.latest_builds(branch, states=("passed", "failed"), per_page=10) - build = pick_best_build(builds) - if not build or str(build.get("state", "")) != "failed": + if not builds: return () - build_id = str(build.get("id", "")) + def trust(build: dict) -> int: + source = str(build.get("source", "")) + if source == "schedule": + return 3 + if source == "api": + return 2 + return 1 + + tier = max(map(trust, builds)) + selected = [b for b in builds if trust(b) == tier][:window] + failures: dict[str, list[tuple[int, BaselineFailure]]] = {} + for index, build in enumerate(selected): + build_id = str(build.get("id", "")) + for job in client.list_jobs(build_id): + if str(job.get("state", "")) not in ( + "failed", "broken", "timed_out"): + continue + name = str(job.get("name", "") or "").lower() + if not name: + continue + raw = job.get("exit_status", -1) + failure = BaselineFailure( + name=name, exit_status=int(raw or -1), + job_id=str(job.get("id", "")), build_id=build_id) + failures.setdefault(name, []).append((index, failure)) + out: list[BaselineFailure] = [] - for job in client.list_jobs(build_id): - if str(job.get("state", "")) not in ("failed", "broken", "timed_out"): - continue - name = str(job.get("name", "") or "").lower() - if not name: - continue - raw = job.get("exit_status", -1) - out.append(BaselineFailure(name=name, exit_status=int(raw or -1), - job_id=str(job.get("id", "")), - build_id=build_id)) + for entries in failures.values(): + if entries[0][0] == 0 or len(entries) >= 2: + out.extend(failure for _, failure in entries) return tuple(out) @@ -725,6 +780,7 @@ class CIBuildRound: budget_timeouts: list[str] = field(default_factory=list) ignored: int = 0 ignored_baseline: int = 0 + jobs: list[dict] = field(default_factory=list) @dataclass @@ -748,6 +804,7 @@ def _record_outcome(rec: CIBuildRound, outcome: MonitorOutcome) -> None: if j.classification == "ignored") rec.ignored_baseline = sum(1 for j in outcome.jobs if j.classification == "ignored_baseline") + rec.jobs = [asdict(j) for j in outcome.jobs] def op_index_base(ops: Sequence[BuildOp], run_id: str) -> int: diff --git a/test/test_ci_wiring.py b/test/test_ci_wiring.py index 08ac60189..22016ef2b 100644 --- a/test/test_ci_wiring.py +++ b/test/test_ci_wiring.py @@ -492,13 +492,50 @@ def test_monitor_baseline_fails_closed_on_missing_evidence(): out = ci_loop.monitor_build(ci, "b", spec=CIClassifySpec( baseline=baseline), poll_sec=0, timeout_sec=0, sleep=lambda s: None) assert out.jobs[0].classification == "failed" - # exit-status mismatch means NO baseline match at all + # Exit-status mismatch alone is not failure identity: the same pytest + # fingerprint may be followed by a teardown timeout on only one side. baseline = (BaselineFailure(name="lane c", exit_status=2, job_id="mainC", build_id="mb"),) ci = ScriptedCI(snapshots=[{"state": "failed", "jobs": [ - _job("Lane C", "failed", 1)]}], logs={"Lane C": log}) + _job("Lane C", "failed", 1)]}], + logs={"Lane C": log, "mainC": log}) out = ci_loop.monitor_build(ci, "b", spec=CIClassifySpec( baseline=baseline), poll_sec=0, timeout_sec=0, sleep=lambda s: None) + assert out.jobs[0].classification == "ignored_baseline" + assert out.jobs[0].baseline_build_id == "mb" + assert out.jobs[0].baseline_job_id == "mainC" + assert out.jobs[0].classification_reason + + +def test_monitor_baseline_matches_cause_checked_error_subset(): + """A current ERROR-only subset matches a noisier baseline job, while a + new cause on the same node remains actionable.""" + baseline = (BaselineFailure(name="lane", exit_status=1, + job_id="main", build_id="mb"),) + baseline_log = ( + "ValueError: triton is not supported for NvFP4 MoE\n" + "ERROR tests/nvfp4.py::test_marlin - RuntimeError: init failed\n" + "AssertionError: audio mismatch\n" + "FAILED tests/audio.py::test_stream - AssertionError: audio mismatch") + current = ( + "ValueError: triton is not supported for NvFP4 MoE\n" + "ERROR tests/nvfp4.py::test_marlin - RuntimeError: init failed") + ci = ScriptedCI( + snapshots=[{"state": "failed", + "jobs": [_job("Lane", "failed", 1)]}], + logs={"Lane": current, "main": baseline_log}) + out = ci_loop.monitor_build(ci, "b", spec=CIClassifySpec( + baseline=baseline), poll_sec=0, timeout_sec=0, + sleep=lambda s: None) + assert out.jobs[0].classification == "ignored_baseline" + assert "subset" in out.jobs[0].classification_reason + + ci.logs["Lane"] = ( + "ValueError: a new regression\n" + "ERROR tests/nvfp4.py::test_marlin - RuntimeError: different") + out = ci_loop.monitor_build(ci, "b", spec=CIClassifySpec( + baseline=baseline), poll_sec=0, timeout_sec=0, + sleep=lambda s: None) assert out.jobs[0].classification == "failed" @@ -731,6 +768,12 @@ def cancel_build(self, build_id): assert fresh[0].state == "created" # the monitored build is the FRESH one at the new commit assert result.rounds[0].build_id == fresh[0].build_id + assert result.rounds[0].jobs == [{ + "name": "Green", "job_id": "Green", "state": "passed", + "exit_status": 0, "classification": "passed", "log_file": "", + "classification_reason": "", "baseline_build_id": "", + "baseline_job_id": "", + }] def test_rounds_active_intent_orphan_is_owned_at_prepush(tmp_path): @@ -827,6 +870,9 @@ def latest_builds(self, branch, states=(), per_page=30): return self.builds def list_jobs(self, build_id): + if build_id == "9": + return [_job("Lane A", "passed", 0, id="jA9"), + _job("Zero Exit", "passed", 0, id="jZ9")] assert build_id == "3" return [_job("Lane A", "failed", 1, id="jA"), _job("Zero Exit", "failed", 0, id="jZ"), @@ -846,6 +892,27 @@ def list_jobs(self, build_id): assert ci_loop.fetch_baseline_failures( BaselineClient(fresh_green), "main") == () + # A recurring older failure remains valid intermittent-baseline evidence + # even when the newest same-tier build happens to pass. + recurring = [ + {"id": "9", "number": 9, "state": "passed", "source": "schedule"}, + {"id": "3", "number": 3, "state": "failed", "source": "schedule"}, + {"id": "2", "number": 2, "state": "failed", "source": "schedule"}, + ] + + class RecurringBaselineClient(BaselineClient): + def list_jobs(self, build_id): + if build_id == "9": + return [_job("Lane A", "passed", 0, id="jA9")] + return [_job("Lane A", "failed", 1, id=f"jA{build_id}")] + + got = ci_loop.fetch_baseline_failures( + RecurringBaselineClient(recurring), "main") + assert got == ( + BaselineFailure("lane a", 1, "jA3", "3"), + BaselineFailure("lane a", 1, "jA2", "2"), + ) + # ── run_ci_rounds ────────────────────────────────────────────────────────────