rebase: improve CI baseline reconciliation

doc/architecture/SPEC/engine/steps/rebase_v3.md

@@ -1,6 +1,6 @@
 # engine/steps/rebase_v3.py —— 规范
 
-<!-- verified-against: 2026-08-25 -->
+<!-- verified-against: 2026-09-06 -->
 
 `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 之外）

doc/architecture/SPEC/rebase_engine.md

@@ -1,6 +1,6 @@
 # rebase_engine/ —— 规范
 
-<!-- verified-against: 2026-08-25 -->
+<!-- verified-against: 2026-09-06 -->
 
 `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` 钉住。
 

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

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:

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 ────────────────────────────────────────────────────────────