rebase: guard automatic debug patches

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

@@ -57,8 +57,9 @@ step —— 薄的受治理 wrapper，substate-first、类型化失败、发布
 - **E1/E2**：能力缺失记 `capability_gap` 并声明式降级（无 tier key/CI
   token/客户端、debug backend 缺失 ⇒ 回归记为结构性失败），绝不静默跳过。
 - debug 尝试**快照护栏**：snapshot → agent → 内容 digest 变更检查（staged+
-  unstaged+untracked 字节+mode+symlink）→ 复跑/本地验证；被拒或验证失败的
-  尝试**回滚** —— 只有绿色复跑算修好。
+  unstaged+untracked 字节+mode+symlink）→ 复跑/本地验证 → patch policy；自动
+  patch 不得修改 assertion/tolerance，test 文件编辑必须有明确本地 passed。
+  被拒、验证失败或策略违规的尝试**回滚**。
 - 空/损坏 manifest ⇒ `manifest_empty`（push gate 阻塞，绝不空洞通过）；
   不可运行的命令归 STRUCTURAL —— 绝不借 bash rc=0 假通过。
 - **A5**：一切仓库知识来自 adapter manifest；parity 词汇泄漏上限 14，

doc/architecture/SPEC/rebase_engine.md

@@ -19,6 +19,7 @@
 | `agent_loop.py` | rebase agent 循环（流式、`.decision.md` 计划闸、150 轮预算，cache-parity） |
 | `assign.py` | commit→模块 确定性归类 + 路径漂移检查 + 报告渲染 |
 | `ci_loop.py` | CI 构建生命周期：受守卫的创建/恢复、monitor、日志分类器、round 编排 |
+| `debug_patch_policy.py` | 自动 CI debug 补丁策略：禁止 oracle 弱化，未本地验证时禁止 test 编辑 |
 | `gitio.py` | git 机械层：暂存纪律、签名提交重试、token 头传输、执行已授权 decision |
 | `hooks.py` | `RebaseHooks` —— adapter 可定制的窄行为面，fail-closed 加载 |
 | `knowledge_attest.py` | 知识层的 WAL 安全逻辑摘要 + 快照/恢复（只读） |
@@ -93,6 +94,8 @@ runner/LLM/CI client 全部可注入 —— 每个模块都能离线测试。
   支持 `FAILED`/`ERROR` pytest node 的原因相关子集，但缺坐标、日志、当前签名
   或共同异常证据时一律 fail-closed。每个 job 的分类理由与 baseline build/job
   坐标随 round 持久化，供恢复和审计使用。
+- 自动 debug agent 不能修改 assertion/tolerance oracle；任何 test 文件编辑只有
+  在对应本地验证明确 passed 时才可进入远端重试。拒绝的尝试由调用方恢复快照。
 - **A5** —— 全包仓库中立：仓库值经 `WheelSpec`/`PinSpec`/`ManifestSpec`/
   `ModulePromptData`/`tool_schemas.json`/hooks 注入；`test_repo_neutral_core` 钉住。
 

src/infermatrix_copilot/engine/steps/rebase_v3.py

@@ -19,6 +19,8 @@
 
 from ...rebase_engine import test_loop as tl
 from ...rebase_engine import wheel as wheel_mod
+from ...rebase_engine.debug_patch_policy import (capture_patch_policy,
+                                                 evaluate_debug_patch)
 from ...rebase_engine.modes import MODES, MUTATING_MODES, mode_state_flags
 from ...rebase_engine.phase1_steps import Phase1Config, run_commit_assignment
 from ...rebase_engine.push_gate import evaluate_push_gate
@@ -2036,8 +2038,9 @@ def _verify_locally(slug: str) -> str:
     async def debug_fn(jr) -> bool:
         """One serialized CI-debug dispatch (parent `_dispatch_sdk_ci_debug`
         parity): snapshot → agent → tree-change check → local harness
-        verification where possible; rejected/unverified-failed attempts
-        are REVERTED so they cannot leak into the retry commit."""
+        verification where possible → patch-policy check. Assertion/tolerance
+        edits and locally unverified test edits are REVERTED so they cannot
+        leak into the retry commit."""
         tier = _tier_client(ctx)
         if isinstance(tier, StepResult):
             ctx.trace.record("capability_gap",
@@ -2061,6 +2064,12 @@ async def debug_fn(jr) -> bool:
         # `status --porcelain` (round-1 review)
         before = _worktree_digest(repo)
         snap, untracked = tl.snapshot_worktree(Path(repo))
+        try:
+            policy_before = capture_patch_policy(Path(repo))
+        except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
+            ctx.trace.record("ci_debug_policy_unavailable", job=jr.name,
+                             detail=str(exc))
+            return False
         verdict = await _run_debug_agent(
             ctx, manifest, module, slug,
             log_tail or f"CI job {jr.name} failed "
@@ -2077,6 +2086,20 @@ async def debug_fn(jr) -> bool:
             ctx.trace.record("ci_debug_no_changes", job=jr.name)
             return False
         local = _verify_locally(slug)
+        try:
+            policy = evaluate_debug_patch(
+                Path(repo), policy_before, local_verdict=local)
+        except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
+            tl.restore_worktree(Path(repo), snap, untracked)
+            ctx.trace.record("ci_debug_policy_unavailable", job=jr.name,
+                             detail=str(exc))
+            return False
+        if not policy.allowed:
+            tl.restore_worktree(Path(repo), snap, untracked)
+            ctx.trace.record("ci_debug_policy_rejected", job=jr.name,
+                             reason=policy.reason,
+                             paths=list(policy.paths))
+            return False
         if local == "failed":
             tl.restore_worktree(Path(repo), snap, untracked)
             ctx.trace.record("ci_debug_verify_failed", job=jr.name)

src/infermatrix_copilot/rebase_engine/debug_patch_policy.py

@@ -0,0 +1,130 @@
+"""Fail-closed policy for patches produced by automatic debug agents.
+
+Remote CI remains the final judge for product-code changes that cannot run on
+the local host. Tests are different: weakening the oracle can manufacture a
+green result. This module snapshots the small policy-sensitive surface before
+an agent runs and evaluates only the delta introduced by that attempt.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import re
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+_SENSITIVE_LINE = re.compile(
+    r"\bassert\b|pytest\.(?:raises|approx)|"
+    r"\b(?:atol|rtol|tolerance|threshold|similarity|ssim|psnr|wer)\b",
+    re.IGNORECASE,
+)
+_POLICY_SUFFIXES = {
+    ".cfg", ".ini", ".j2", ".jinja", ".json", ".md", ".py", ".sh",
+    ".toml", ".txt", ".yaml", ".yml",
+}
+
+
+@dataclass(frozen=True)
+class PatchPolicySnapshot:
+    """Relevant file state immediately before one debug-agent attempt."""
+
+    sensitive_lines: dict[str, tuple[str, ...]]
+    test_hashes: dict[str, str]
+
+
+@dataclass(frozen=True)
+class PatchPolicyDecision:
+    allowed: bool
+    reason: str = ""
+    paths: tuple[str, ...] = ()
+
+
+def _repo_files(repo: Path) -> tuple[str, ...]:
+    proc = subprocess.run(
+        ["git", "-C", str(repo), "ls-files", "-co", "--exclude-standard", "-z"],
+        capture_output=True,
+        timeout=60,
+        check=False,
+    )
+    if proc.returncode != 0:
+        raise RuntimeError("cannot enumerate repository files for debug-patch policy")
+    return tuple(
+        p.decode("utf-8", errors="surrogateescape")
+        for p in proc.stdout.split(b"\0")
+        if p
+    )
+
+
+def _is_test_file(relative: str) -> bool:
+    path = Path(relative)
+    parts = {part.lower() for part in path.parts[:-1]}
+    name = path.name.lower()
+    return (
+        "test" in parts
+        or "tests" in parts
+        or name.startswith("test_")
+        or name.endswith("_test.py")
+    )
+
+
+def _read(repo: Path, relative: str) -> bytes:
+    path = repo / relative
+    try:
+        return path.read_bytes() if path.is_file() else b""
+    except OSError:
+        return b""
+
+
+def capture_patch_policy(repo: Path) -> PatchPolicySnapshot:
+    """Capture assertion/tolerance lines and complete test-file hashes."""
+    repo = Path(repo)
+    sensitive: dict[str, tuple[str, ...]] = {}
+    tests: dict[str, str] = {}
+    for relative in _repo_files(repo):
+        is_test = _is_test_file(relative)
+        if not is_test and Path(relative).suffix.lower() not in _POLICY_SUFFIXES:
+            continue
+        content = _read(repo, relative)
+        if Path(relative).suffix.lower() in _POLICY_SUFFIXES:
+            text = content.decode("utf-8", errors="replace")
+            lines = tuple(line.strip() for line in text.splitlines()
+                          if _SENSITIVE_LINE.search(line))
+            if lines:
+                sensitive[relative] = lines
+        if is_test:
+            tests[relative] = hashlib.sha256(content).hexdigest()
+    return PatchPolicySnapshot(sensitive, tests)
+
+
+def evaluate_debug_patch(
+    repo: Path,
+    before: PatchPolicySnapshot,
+    *,
+    local_verdict: str,
+) -> PatchPolicyDecision:
+    """Reject oracle changes, and reject any unverified test-file change."""
+    after = capture_patch_policy(repo)
+    sensitive_paths = tuple(sorted(
+        path for path in set(before.sensitive_lines) | set(after.sensitive_lines)
+        if before.sensitive_lines.get(path) != after.sensitive_lines.get(path)
+    ))
+    if sensitive_paths:
+        return PatchPolicyDecision(
+            False,
+            "automatic debug patches may not change assertions or tolerances",
+            sensitive_paths,
+        )
+
+    test_paths = tuple(sorted(
+        path for path in set(before.test_hashes) | set(after.test_hashes)
+        if before.test_hashes.get(path) != after.test_hashes.get(path)
+    ))
+    if test_paths and local_verdict != "passed":
+        return PatchPolicyDecision(
+            False,
+            f"test-file changes require a passing local verification "
+            f"(got {local_verdict})",
+            test_paths,
+        )
+    return PatchPolicyDecision(True)

test/test_assembly.py

@@ -1533,6 +1533,66 @@ async def fake_rounds(**kwargs):
     assert captured["token"] == "gh-push-tok"
 
 
+def test_v3_ci_debug_rejects_weakened_assertion(
+        v3_agent_env, settings, trace, monkeypatch):
+    """The CI-debug acceptance boundary restores an oracle-weakening patch,
+    even when the target test cannot be reproduced on the local host."""
+    import subprocess
+    from types import SimpleNamespace
+    from infermatrix_copilot.engine.registry import StepRegistry
+    from infermatrix_copilot.engine.step import StepContext
+    from infermatrix_copilot.engine.steps import register_builtin_steps
+    from infermatrix_copilot.engine.steps import rebase_v3
+    from infermatrix_copilot.rebase_engine import ci_loop
+    _, _, repo, run_dir = v3_agent_env
+    subprocess.run(["git", "checkout", "-qb", "dev/vllm-align"],
+                   cwd=repo, check=True)
+    helper = repo / "tests" / "audio_helpers.py"
+    helper.write_text("def check(score):\n    assert score > 0.8\n")
+    subprocess.run(["git", "add", "-A"], cwd=repo, check=True)
+    subprocess.run(["git", "-c", "user.name=t", "-c", "user.email=t@t",
+                    "commit", "-qm", "add test helper"], cwd=repo,
+                   check=True)
+    original = helper.read_text()
+    monkeypatch.setattr(settings, "buildkite_api_token", "bk",
+                        raising=False)
+    monkeypatch.setattr(
+        rebase_v3, "_tier_client",
+        lambda ctx: (object(), SimpleNamespace(
+            model="m", api_key="k", base_url="", source="global")))
+
+    async def fake_agent(*args, **kwargs):
+        helper.write_text("def check(score):\n    assert score > 0.7\n")
+        return "done"
+
+    accepted = []
+
+    async def fake_rounds(**kwargs):
+        accepted.append(await kwargs["debug_fn"](SimpleNamespace(
+            name="Quick", log_file="", state="failed", exit_status=1)))
+        return SimpleNamespace(result="failed", reason="stub",
+                               fixed_jobs=[], unfixed_jobs=["Quick"],
+                               rounds=[])
+
+    monkeypatch.setattr(rebase_v3, "_run_debug_agent", fake_agent)
+    monkeypatch.setattr(ci_loop, "run_ci_rounds", fake_rounds)
+    registry = register_builtin_steps(StepRegistry())
+    rd = run_dir.parent / "dir-ci-policy"
+    rd.mkdir(exist_ok=True)
+    ctx = StepContext(
+        settings=settings, params={}, run_dir=rd, trace=trace,
+        state={"task_spec": {"repo": "vllm-omni",
+                             "params": {"rebase_mode": "remote_ci",
+                                        "upstream_commit": "f" * 40}},
+               "run_id": "run-ci-policy", "repo_path": str(repo),
+               "upstream_commit": "f" * 40})
+    result = asyncio.run(registry.get("rebase.v3_ci").handler(ctx))
+    assert result.ok and accepted == [False]
+    assert helper.read_text() == original
+    rejected = list(trace.events("ci_debug_policy_rejected"))
+    assert rejected and rejected[-1]["paths"] == ["tests/audio_helpers.py"]
+
+
 def test_v3_ci_arms_adoption_exemption_only_on_proven_remote_noop(
         v3_env, settings, trace, monkeypatch, tmp_path):
     """The pre-push adoption exemption may only arm when the push is a

test/test_ci_wiring.py

@@ -14,7 +14,8 @@
 import pytest
 
 from infermatrix_copilot.ci.buildkite import BuildkiteCI, BuildkiteError
-from infermatrix_copilot.rebase_engine import ci_loop, push_to_ci
+from infermatrix_copilot.rebase_engine import (ci_loop, debug_patch_policy,
+                                               push_to_ci)
 from infermatrix_copilot.rebase_engine.ci_loop import (BaselineFailure,
                                                        CIClassifySpec)
 
@@ -1333,6 +1334,54 @@ def test_worktree_digest_sees_content_edits_porcelain_misses(tmp_path):
     assert _worktree_digest(repo) == d4
 
 
+def test_debug_patch_policy_rejects_oracle_weakening(tmp_path):
+    """A locally green run cannot authorize an agent to lower a test oracle."""
+    import subprocess
+
+    repo = tmp_path / "r"
+    (repo / "tests").mkdir(parents=True)
+    subprocess.run(["git", "init", "-q", repo], check=True)
+    helper = repo / "tests" / "audio_helpers.py"
+    helper.write_text("def check(score):\n    assert score > 0.8\n")
+    subprocess.run(["git", "-C", repo, "add", "-A"], check=True)
+    before = debug_patch_policy.capture_patch_policy(repo)
+
+    helper.write_text("def check(score):\n    assert score > 0.7\n")
+    decision = debug_patch_policy.evaluate_debug_patch(
+        repo, before, local_verdict="passed")
+    assert not decision.allowed
+    assert decision.paths == ("tests/audio_helpers.py",)
+    assert "assertions or tolerances" in decision.reason
+
+
+def test_debug_patch_policy_requires_local_pass_for_test_edits(tmp_path):
+    import subprocess
+
+    repo = tmp_path / "r"
+    (repo / "tests").mkdir(parents=True)
+    subprocess.run(["git", "init", "-q", repo], check=True)
+    test_file = repo / "tests" / "test_widget.py"
+    test_file.write_text("def test_widget():\n    pass\n")
+    source = repo / "widget.py"
+    source.write_text("VALUE = 1\n")
+    subprocess.run(["git", "-C", repo, "add", "-A"], check=True)
+
+    before = debug_patch_policy.capture_patch_policy(repo)
+    test_file.write_text("def test_widget():\n    # new coverage\n    pass\n")
+    skipped = debug_patch_policy.evaluate_debug_patch(
+        repo, before, local_verdict="skipped")
+    assert not skipped.allowed
+    assert "passing local verification" in skipped.reason
+    assert debug_patch_policy.evaluate_debug_patch(
+        repo, before, local_verdict="passed").allowed
+
+    subprocess.run(["git", "-C", repo, "checkout", "--", "."], check=True)
+    before = debug_patch_policy.capture_patch_policy(repo)
+    source.write_text("VALUE = 2\n")
+    assert debug_patch_policy.evaluate_debug_patch(
+        repo, before, local_verdict="unavailable").allowed
+
+
 def test_cancel_owned_active_builds_on_abort(tmp_path):
     """Abort cleanup cancels ONLY our still-active op-recorded builds:
     terminal builds and adopted (ledger-less) builds are untouched, and a