From 9e01b92bf08e505c259b948633d976019936dcac Mon Sep 17 00:00:00 2001
From: BeatSeat <137616995+BeatSeat@users.noreply.github.com>
Date: Sun, 15 Nov 2026 19:51:35 +0800
Subject: [PATCH 1/5] harness: report paired AB/BA statistics alongside the
 point verdict

Pair the k-th baseline and candidate samples that steps() already
alternates, and report the median paired speedup with a seeded percentile
bootstrap interval and its relation to the frozen gate. The field is
report-only: verdict, eligibility and selection are unchanged, and any value
that cannot be computed is reported as unavailable so records stay strict
JSON. Fewer than five pairs report no interval because simulated coverage at
three pairs is about 75%. tools/measurement_simulation.py reproduces the
synthetic ablation.

Agent-assisted-by: Claude Code / claude-opus-5-5 (author)
Agent-assisted-by: Codex CLI / gpt-6.1-sol [effort=high] (review)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: BeatSeat <137616995+BeatSeat@users.noreply.github.com>
---
 docs/optimization-workflow-unification.md    |   2 +
 src/inferrsi/harness/measurement.py          | 148 +++++++++++++++
 src/inferrsi/harness/optimization_session.py |   2 +-
 src/inferrsi/harness/project.py              |   6 +
 tests/test_harness_measurement.py            | 179 +++++++++++++++++++
 tests/test_project_optimization.py           |  60 +++++++
 tools/measurement_simulation.py              | 120 +++++++++++++
 7 files changed, 516 insertions(+), 1 deletion(-)
 create mode 100644 src/inferrsi/harness/measurement.py
 create mode 100644 tests/test_harness_measurement.py
 create mode 100644 tools/measurement_simulation.py

diff --git a/docs/optimization-workflow-unification.md b/docs/optimization-workflow-unification.md
index 18f00606e..f89db1ab8 100644
--- a/docs/optimization-workflow-unification.md
+++ b/docs/optimization-workflow-unification.md
@@ -176,6 +176,8 @@ inferrsi campaign start --directory <campaign-dir-outside-project> \
 
 项目适配器先检查基线和候选质量，再做 warmup 和交替顺序的基线/候选测量；每轮和 final 使用相同协议。每步骤启动新进程，因此跨进程 warmup 只能预热可持久缓存；设备/进程内 warmup 必须在测量脚本内完成并说明。性能取声明指标的中位数，当前没有通用服务生命周期管理器或统计显著性判定。
 
+assessment 另附只读的 `paired_statistics`：同一次重复内的第 k 个基线与第 k 个候选样本按顺序配对，计算逐对加速比的中位数，以及按重复重采样的 95% percentile bootstrap 区间（记录 resamples 与 seed），并给出区间与冻结门槛的关系（`above_gate`、`straddles_gate`、`below_gate`；门槛没有有限加速比时为 `gate_unreachable`；少于 5 次重复为 `insufficient_pairs`；样本或派生值无效为 `unavailable`）。`agrees_with_point_gate` 标出中位数点估计的判定与配对区间是否一致。该字段只描述留存样本，不改变 verdict、eligible 或选优，`statistical_significance` 仍为 `not_established`。区间是名义 95%，覆盖率未经校准：合成对照中 3 对时约 75%，因此少于 5 对不报告区间。区间假设各次重复独立，不能消除同一对内共同的设备争用，也不构成环境资格。合成对照可用 `python tools/measurement_simulation.py` 复算。
+
 结果协议：
 
 - 测量脚本正常结束并写 JSON，提供指定字段的有限正数。
diff --git a/src/inferrsi/harness/measurement.py b/src/inferrsi/harness/measurement.py
new file mode 100644
index 000000000..404be1512
--- /dev/null
+++ b/src/inferrsi/harness/measurement.py
@@ -0,0 +1,148 @@
+"""Paired, report-only statistics for alternating baseline/candidate measurements.
+
+The project protocol alternates AB/BA order inside each repetition, so the k-th
+baseline and k-th candidate sample share one repetition. Their per-pair ratio
+cancels drift shared within a repetition, which independent per-side medians
+retain. This module only describes retained samples: it never changes a verdict,
+eligibility or gate, and any value it cannot compute is reported as unavailable.
+
+The estimand is the median per-repetition speedup. The interval is a nominal 95%
+percentile bootstrap over repetitions with a recorded seed. Its coverage is not
+calibrated in general: with few repetitions it collapses towards the observed pair
+range and under-covers (about 75% at three pairs in tools/measurement_simulation.py),
+so no interval is reported below MIN_PAIRS. Repetitions are assumed independent;
+pairing cannot
+remove contention shared by both sides of one pair and does not qualify the
+environment.
+"""
+
+import math
+import random
+from statistics import median
+
+METHOD = "paired_median_speedup_percentile_bootstrap"
+CONFIDENCE = 0.95
+MIN_PAIRS = 5
+RESAMPLES = 10_000
+SEED = 0
+
+
+def gate_speedup(min_improvement_pct: float, direction: str) -> float:
+    """Express the frozen improvement gate as the equivalent speedup ratio.
+
+    A minimize gate of 100% or more has no finite speedup and returns ``inf``.
+    """
+    fraction = min_improvement_pct / 100
+    if direction == "maximize":
+        return 1 + fraction
+    if direction == "minimize":
+        return math.inf if fraction >= 1 else 1 / (1 - fraction)
+    raise ValueError("Metric direction must be explicit")
+
+
+def paired_speedups(baseline: list, candidate: list, direction: str) -> list[float]:
+    if len(baseline) != len(candidate):
+        raise ValueError("Paired statistics need one candidate sample per baseline sample")
+    if direction not in ("minimize", "maximize"):
+        raise ValueError("Metric direction must be explicit")
+    values = [*baseline, *candidate]
+    if any(type(v) not in (int, float) or not math.isfinite(v) or v <= 0 for v in values):
+        raise ValueError("Paired statistics need finite positive samples")
+    speedups = [b / c if direction == "minimize" else c / b
+                for b, c in zip(baseline, candidate)]
+    if any(not math.isfinite(s) or s <= 0 for s in speedups):
+        raise ValueError("A paired ratio overflowed or underflowed")
+    return speedups
+
+
+def relative_mad(values: list) -> float | None:
+    """Median absolute deviation divided by the median, a robust spread indicator."""
+    if not values:
+        return None
+    center = median(values)
+    return median(abs(v - center) for v in values) / center
+
+
+def _percentile(ordered: list, q: float) -> float:
+    position = q * (len(ordered) - 1)
+    low = math.floor(position)
+    high = min(low + 1, len(ordered) - 1)
+    return ordered[low] + (ordered[high] - ordered[low]) * (position - low)
+
+
+def bootstrap_median_interval(values: list, *, resamples: int = RESAMPLES,
+                              seed: int = SEED) -> list[float]:
+    if type(resamples) is not int or resamples < 1 or not values:
+        raise ValueError("Bootstrap needs a positive resample count and at least one value")
+    rng = random.Random(seed)
+    medians = sorted(median(rng.choices(values, k=len(values))) for _ in range(resamples))
+    tail = (1 - CONFIDENCE) / 2
+    return [_percentile(medians, tail), _percentile(medians, 1 - tail)]
+
+
+def _finite(value) -> bool:
+    if isinstance(value, bool) or value is None or isinstance(value, str):
+        return True
+    if isinstance(value, (int, float)):
+        return math.isfinite(value)
+    if isinstance(value, (list, tuple)):
+        return all(_finite(item) for item in value)
+    if isinstance(value, dict):
+        return all(_finite(item) for item in value.values())
+    return False
+
+
+def _describe(baseline, candidate, *, direction, min_improvement_pct, point_gate_passed,
+              resamples, seed) -> dict:
+    speedups = paired_speedups(baseline, candidate, direction)
+    gate = gate_speedup(min_improvement_pct, direction)
+    report = {
+        "status": "available", "pairs": len(speedups), "pair_speedups": speedups,
+        "median_speedup": median(speedups) if speedups else None,
+        "geomean_speedup": (math.exp(sum(map(math.log, speedups)) / len(speedups))
+                            if speedups else None),
+        "gate_speedup": gate if math.isfinite(gate) else None,
+        "baseline_relative_mad": relative_mad(baseline),
+        "candidate_relative_mad": relative_mad(candidate), "interval": None,
+    }
+    if len(speedups) < MIN_PAIRS:
+        return {**report, "status": "insufficient_pairs", "gate_relation": "insufficient_pairs",
+                "agrees_with_point_gate": None,
+                "reason": f"At least {MIN_PAIRS} repetitions are needed for an interval"}
+    low, high = bootstrap_median_interval(speedups, resamples=resamples, seed=seed)
+    if not math.isfinite(gate):
+        relation = "gate_unreachable"
+    elif low > gate:
+        relation = "above_gate"
+    elif high < gate:
+        relation = "below_gate"
+    else:
+        relation = "straddles_gate"
+    return {**report, "interval": [low, high], "gate_relation": relation,
+            "excludes_no_change": low > 1 or high < 1,
+            "agrees_with_point_gate": point_gate_passed == (relation == "above_gate")}
+
+
+def paired_statistics(baseline: list, candidate: list, *, direction: str,
+                      min_improvement_pct: float, point_gate_passed: bool,
+                      resamples: int = RESAMPLES, seed: int = SEED) -> dict:
+    """Describe paired samples against the frozen gate without changing any verdict.
+
+    Every returned number is finite so the record stays strict-JSON serializable;
+    anything that cannot be computed is reported with ``status="unavailable"``.
+    """
+    header = {"method": METHOD, "report_only": True, "confidence": CONFIDENCE,
+              "coverage": "nominal_uncalibrated",
+              "pairing": "kth baseline with kth candidate; AB/BA order alternates by repetition",
+              "resamples": resamples, "seed": seed}
+    try:
+        report = _describe(baseline, candidate, direction=direction,
+                           min_improvement_pct=min_improvement_pct,
+                           point_gate_passed=point_gate_passed,
+                           resamples=resamples, seed=seed)
+    except (ArithmeticError, TypeError, ValueError) as error:
+        return {**header, "status": "unavailable", "reason": str(error), "interval": None}
+    if not _finite(report):
+        return {**header, "status": "unavailable",
+                "reason": "A derived statistic is not finite", "interval": None}
+    return {**header, **report}
diff --git a/src/inferrsi/harness/optimization_session.py b/src/inferrsi/harness/optimization_session.py
index 9e7975c2c..b9fdee40b 100644
--- a/src/inferrsi/harness/optimization_session.py
+++ b/src/inferrsi/harness/optimization_session.py
@@ -115,7 +115,7 @@ def _runtime_identity():
              "infra/processes.py", "infra/artifacts.py", "harness/scenarios/external.py",
              "harness/atomic_json.py",
              "harness/usage.py", "harness/agent_sessions.py", "harness/task_protocol.py",
-             "harness/candidates.py", "harness/project.py",
+             "harness/candidates.py", "harness/project.py", "harness/measurement.py",
              "harness/records/knowledge_capture.py", "apps/campaign.py",
              "knowledge/context_projection.py")
     return {name: sha256(root / name) for name in names}
diff --git a/src/inferrsi/harness/project.py b/src/inferrsi/harness/project.py
index fa0da190d..df9d091a0 100644
--- a/src/inferrsi/harness/project.py
+++ b/src/inferrsi/harness/project.py
@@ -11,6 +11,7 @@
 from statistics import median
 
 from ..infra.artifacts import sha256
+from .measurement import paired_statistics
 
 
 def validate(settings: dict) -> None:
@@ -159,6 +160,10 @@ def result(status, reason, **fields):
     if not all(math.isfinite(value) for value in (improvement, speedup)):
         return result("infrastructure_failure", "nonfinite_derived_metric")
     accepted = improvement > 0 and improvement >= settings["min_improvement_pct"]
+    paired = paired_statistics(samples["baseline"], samples["candidate"],
+                               direction=settings["direction"],
+                               min_improvement_pct=settings["min_improvement_pct"],
+                               point_gate_passed=accepted)
     return result(
         "completed", "fixed_checks_satisfied" if accepted else "performance_gate_not_met",
         verdict="PASS" if accepted else "FAIL", score=after, baseline_score=before,
@@ -167,6 +172,7 @@ def result(status, reason, **fields):
                  "direction": settings["direction"], "baseline": before, "candidate": after,
                  "samples": samples},
         scope=settings["scope"], statistical_significance="not_established",
+        paired_statistics=paired,
         quality_contract="Only the supplied frozen quality script and its boolean result",
         environment_qualification="Caller-managed; no automatic hardware isolation claim",
     )
diff --git a/tests/test_harness_measurement.py b/tests/test_harness_measurement.py
new file mode 100644
index 000000000..d8a4a47db
--- /dev/null
+++ b/tests/test_harness_measurement.py
@@ -0,0 +1,179 @@
+"""Paired report-only statistics; every sample below is a synthetic software control."""
+
+import json
+import math
+import random
+from statistics import median
+
+import pytest
+
+from inferrsi.harness import optimization_session
+from inferrsi.harness.measurement import (
+    bootstrap_median_interval,
+    gate_speedup,
+    paired_speedups,
+    paired_statistics,
+)
+
+BASE = [10.0, 10.1, 9.9, 10.05, 9.95]
+
+
+def report(baseline, candidate, *, direction="minimize", gate_pct=1.0, passed=True, **kwargs):
+    return paired_statistics(baseline, candidate, direction=direction,
+                             min_improvement_pct=gate_pct, point_gate_passed=passed, **kwargs)
+
+
+def point_gate(baseline, candidate, gate_pct):
+    """The current project.assess rule: independent medians, minimize direction."""
+    improvement = (median(baseline) - median(candidate)) / median(baseline) * 100
+    return improvement > 0 and improvement >= gate_pct
+
+
+def test_gate_speedup_matches_improvement_pct_definition():
+    assert gate_speedup(10, "maximize") == pytest.approx(1.10)
+    # Minimize: a 10% lower latency is a 1/0.9 speedup, not 1.10.
+    assert gate_speedup(10, "minimize") == pytest.approx(1 / 0.9)
+    assert gate_speedup(0, "minimize") == 1
+    assert gate_speedup(100, "minimize") == math.inf
+    with pytest.raises(ValueError):
+        gate_speedup(1, "sideways")
+
+
+def test_pair_direction_and_input_validation():
+    assert paired_speedups([10, 20], [5, 10], "minimize") == [2.0, 2.0]
+    assert paired_speedups([10, 20], [5, 10], "maximize") == [0.5, 0.5]
+    for baseline, candidate in (([1, 2], [1]), ([1, 0], [1, 1]), ([1, math.nan], [1, 1])):
+        with pytest.raises(ValueError):
+            paired_speedups(baseline, candidate, "minimize")
+
+
+@pytest.mark.parametrize("baseline,candidate", [
+    ([10, 10, 10], [9, 9]),            # lost pair
+    ([10, 0, 10], [9, 9, 9]),          # zero sample
+    ([10, math.nan, 10], [9, 9, 9]),   # non-finite sample
+])
+def test_invalid_samples_report_unavailable_instead_of_raising(baseline, candidate):
+    result = report(baseline, candidate)
+    assert result["status"] == "unavailable" and result["interval"] is None
+    assert result["report_only"] is True
+
+
+@pytest.mark.parametrize("baseline,candidate", [
+    ([1e-300, 10, 10], [1e300, 5, 5]),   # paired ratio underflows to zero
+    ([1e300, 10, 10], [1e-300, 5, 5]),   # paired ratio overflows to infinity
+])
+def test_ratio_overflow_reports_unavailable_and_stays_strict_json(baseline, candidate):
+    result = report(baseline, candidate)
+    assert result["status"] == "unavailable"
+    json.dumps(result, allow_nan=False)
+
+
+def test_unreachable_minimize_gate_is_serializable_not_infinite():
+    result = report(BASE, [5.0, 5.1, 4.9, 5.02, 4.98], gate_pct=100, passed=False)
+    assert result["gate_speedup"] is None and result["gate_relation"] == "gate_unreachable"
+    assert result["agrees_with_point_gate"] is True
+    json.dumps(result, allow_nan=False)
+
+
+@pytest.mark.parametrize("pairs", [2, 4])
+def test_too_few_repetitions_report_no_interval(pairs):
+    # Below five pairs the simulated coverage of the nominal 95% interval is ~75%.
+    result = report(BASE[:pairs], [8.0, 8.1, 7.9, 8.05][:pairs])
+    assert result["status"] == "insufficient_pairs" and result["pairs"] == pairs
+    assert result["interval"] is None and result["agrees_with_point_gate"] is None
+
+
+def test_interval_is_deterministic_for_the_recorded_seed():
+    values = [1.02, 0.99, 1.05, 1.01, 0.97, 1.03]
+    assert bootstrap_median_interval(values, seed=7) == bootstrap_median_interval(values, seed=7)
+    result = report([10.0] * 6, [10 / v for v in values])
+    assert result["seed"] == 0 and result["resamples"] == 10_000
+
+
+def test_consistent_real_gain_is_above_gate():
+    baseline = [10.0, 10.1, 9.9, 10.05, 9.95]
+    candidate = [8.0, 8.1, 7.9, 8.02, 7.98]
+    result = report(baseline, candidate, gate_pct=5)
+    low, high = result["interval"]
+    assert low > result["gate_speedup"] and low <= result["median_speedup"] <= high
+    assert result["gate_relation"] == "above_gate" and result["excludes_no_change"] is True
+    assert result["agrees_with_point_gate"] is True
+
+
+def test_aa_noise_straddles_gate_even_when_point_estimate_clears_it():
+    # A/A control: identical code, noisy alternation. Independent medians differ by
+    # more than the 1% gate, so the current point gate passes; the pairs do not.
+    baseline = [10.0, 10.6, 9.6, 10.4, 9.9]
+    candidate = [9.7, 10.2, 9.8, 10.1, 9.75]
+    assert point_gate(baseline, candidate, 1)
+    result = report(baseline, candidate, gate_pct=1, passed=True)
+    assert result["gate_relation"] == "straddles_gate"
+    assert result["excludes_no_change"] is False
+    assert result["agrees_with_point_gate"] is False
+
+
+def test_pairing_cancels_shared_drift():
+    # Both sides slow down 2% per repetition; the true speedup is 1.25 in every pair.
+    drift = [1.0 + 0.02 * k for k in range(6)]
+    result = report([10.0 * d for d in drift], [8.0 * d for d in drift], gate_pct=10)
+    assert result["pair_speedups"] == pytest.approx([1.25] * 6)
+    assert result["interval"] == pytest.approx([1.25, 1.25])
+    assert result["baseline_relative_mad"] > 0.02
+
+
+def test_alternating_order_penalty_does_not_create_a_gain():
+    # A/A with a 5% penalty for whichever side runs second in a repetition.
+    baseline, candidate = [], []
+    for k in range(8):
+        first, second = 10.0, 10.5
+        baseline.append(first if k % 2 == 0 else second)
+        candidate.append(second if k % 2 == 0 else first)
+    result = report(baseline, candidate, gate_pct=1, passed=False)
+    assert result["gate_relation"] != "above_gate"
+
+
+def test_clear_regression_is_below_gate():
+    result = report(BASE, [12.0, 12.1, 11.9, 12.02, 11.98], gate_pct=1, passed=False)
+    assert result["gate_relation"] == "below_gate"
+    assert result["agrees_with_point_gate"] is True
+
+
+@pytest.mark.parametrize("noise", ["lognormal", "heavy_tail"])
+def test_aa_simulation_paired_false_gains_stay_rare_and_below_point_gate(noise):
+    """Ablation: independent medians vs paired interval on simulated A/A campaigns.
+
+    Each campaign has 10 repetitions with a shared random-walk drift, per-sample
+    3% noise and, for heavy_tail, a 10% chance of a 1.3x stall. Fixed settings:
+    200 campaigns, gate 1%, seed 20261115.
+    """
+    rng = random.Random(20261115)
+    point_false = paired_false = 0
+    campaigns = 200
+    for index in range(campaigns):
+        level, baseline, candidate = 10.0, [], []
+        for _ in range(10):
+            level *= math.exp(rng.gauss(0, 0.02))
+
+            def sample():
+                value = level * math.exp(rng.gauss(0, 0.03))
+                if noise == "heavy_tail" and rng.random() < 0.1:
+                    value *= 1.3
+                return value
+
+            baseline.append(sample())
+            candidate.append(sample())
+        point_false += point_gate(baseline, candidate, 1)
+        result = report(baseline, candidate, gate_pct=1, resamples=1000, seed=index)
+        paired_false += result["gate_relation"] == "above_gate"
+    assert paired_false / campaigns <= 0.05
+    assert paired_false < point_false
+
+
+def test_measurement_module_is_part_of_runtime_identity():
+    assert "harness/measurement.py" in optimization_session._runtime_identity()
+
+
+@pytest.mark.parametrize("resamples", [0, -1, 1.5])
+def test_invalid_resample_count_reports_unavailable(resamples):
+    result = report(BASE, [8.0, 8.1, 7.9, 8.02, 7.98], resamples=resamples)
+    assert result["status"] == "unavailable" and result["interval"] is None
diff --git a/tests/test_project_optimization.py b/tests/test_project_optimization.py
index 83df66362..6e42ef833 100644
--- a/tests/test_project_optimization.py
+++ b/tests/test_project_optimization.py
@@ -423,3 +423,63 @@ def test_final_reserve_blocks_search_but_allows_final(project, capsys, monkeypat
     assert main(["campaign", "final", "--directory", str(project[3]),
                  "--reason", "Search time exhausted"]) == 0
     capsys.readouterr()
+
+
+@pytest.mark.parametrize("direction,value", [("maximize", 20), ("minimize", 5)])
+def test_assessment_reports_paired_statistics_without_changing_the_verdict(
+    project, capsys, direction, value,
+):
+    start(project, capsys, direction=direction, extra=("--repetitions", "5"))
+    code, result = submit(project, capsys, value)
+    assert code == 0 and result["best"]["kind"] == "candidate"
+    assessment = result["latest_observation"]["assessment"]
+    assert assessment["verdict"] == "PASS" and assessment["speedup"] == 2
+    assert assessment["statistical_significance"] == "not_established"
+    paired = assessment["paired_statistics"]
+    assert paired["report_only"] is True and paired["pairs"] == 5
+    assert paired["status"] == "available"
+    assert paired["interval"] == pytest.approx([2.0, 2.0])
+    assert paired["gate_relation"] == "above_gate"
+    assert paired["agrees_with_point_gate"] is True
+    assert main(["campaign", "final", "--directory", str(project[3]),
+                 "--reason", "Synthetic search done"]) == 0
+    capsys.readouterr()
+    _, state, summary = read_records(project[3])
+    final = state["final"]["assessment"]
+    assert final["verdict"] == "PASS" and summary["optimization_speedup"] == 2
+    assert final["paired_statistics"]["gate_relation"] == "above_gate"
+
+
+def test_unreachable_gate_keeps_the_original_fail_and_strict_json_records(project, capsys):
+    # A minimize gate of 100% has no finite speedup; the report must not block the FAIL.
+    start(project, capsys, direction="minimize",
+          extra=("--repetitions", "5", "--min-improvement-pct", "100"))
+    code, result = submit(project, capsys, 5)
+    assert code == 0 and result["best"]["kind"] == "baseline"
+    assessment = result["latest_observation"]["assessment"]
+    assert assessment["verdict"] == "FAIL" and assessment["eligible"] is False
+    paired = assessment["paired_statistics"]
+    assert paired["gate_speedup"] is None and paired["gate_relation"] == "gate_unreachable"
+    json.dumps(read_records(project[3])[1], allow_nan=False)
+
+
+def test_records_without_paired_statistics_remain_readable(project, capsys, monkeypatch):
+    """Assessments retained before this field existed still load in records and dashboard."""
+    from inferrsi.harness import project as project_protocol
+
+    original = project_protocol.assess
+
+    def legacy_assess(*args, **kwargs):
+        result = original(*args, **kwargs)
+        result.pop("paired_statistics", None)
+        return result
+
+    monkeypatch.setattr(project_protocol, "assess", legacy_assess)
+    start(project, capsys)
+    code, result = submit(project, capsys, 20)
+    assert code == 0 and "paired_statistics" not in result["latest_observation"]["assessment"]
+    _, state, _ = read_records(project[3])
+    assert state["best"]["kind"] == "candidate"
+    snapshot = dashboard_snapshot(project[3])
+    assert snapshot["data"]["candidates"][0]["verdict"] == "PASS"
+    assert "Synthetic candidate" in render_dashboard(snapshot)
diff --git a/tools/measurement_simulation.py b/tools/measurement_simulation.py
new file mode 100644
index 000000000..d77591168
--- /dev/null
+++ b/tools/measurement_simulation.py
@@ -0,0 +1,120 @@
+"""Reproduce the synthetic ablation behind harness paired statistics.
+
+Simulated campaigns (software only, not device measurements) compare four rules
+on the same samples, so the contribution of pairing and of the interval can be
+separated:
+
+* point_rule        current project.assess: ratio of independent medians >= gate
+* paired_point      median of per-repetition speedups >= gate, no interval
+* unpaired_interval percentile bootstrap of the ratio of independently resampled
+                    medians; lower bound > gate
+* paired_interval   harness.measurement.paired_statistics: above_gate
+
+The data seed generates samples; the paired interval uses the production
+bootstrap configuration (RESAMPLES, SEED) unless overridden. Coverage is the
+fraction of campaigns whose paired bootstrap interval contains the true median
+speedup (computed even below MIN_PAIRS, where the harness reports no interval),
+which equals the configured true speedup under this symmetric noise model.
+
+    python tools/measurement_simulation.py            # production bootstrap config
+    python tools/measurement_simulation.py --resamples 1000 --campaigns 100   # quick
+"""
+
+import argparse
+import json
+import math
+import random
+from statistics import median
+
+from inferrsi.harness.measurement import (
+    MIN_PAIRS,
+    RESAMPLES,
+    SEED,
+    bootstrap_median_interval,
+    gate_speedup,
+    paired_statistics,
+)
+
+REPETITIONS = (3, 5, 10, 20)
+TRUE_SPEEDUPS = (1.0, 1.01, 1.03, 1.05, 1.10)
+NOISES = ("lognormal", "heavy_tail")
+
+
+def campaign(rng, repetitions, true_speedup, noise):
+    level, baseline, candidate = 10.0, [], []
+    for _ in range(repetitions):
+        level *= math.exp(rng.gauss(0, 0.02))  # drift shared by both sides of a pair
+
+        def sample(scale):
+            value = level * scale * math.exp(rng.gauss(0, 0.03))
+            if noise == "heavy_tail" and rng.random() < 0.1:
+                value *= 1.3
+            return value
+
+        baseline.append(sample(1.0))
+        candidate.append(sample(1 / true_speedup))
+    return baseline, candidate
+
+
+def unpaired_lower_bound(baseline, candidate, rng, resamples):
+    ratios = sorted(
+        median(rng.choices(baseline, k=len(baseline)))
+        / median(rng.choices(candidate, k=len(candidate)))
+        for _ in range(resamples)
+    )
+    return ratios[int(0.025 * (resamples - 1))]
+
+
+def rates(*, repetitions, true_speedup, noise, gate_pct, campaigns, data_seed, resamples,
+          bootstrap_seed):
+    rng = random.Random(f"{data_seed}-{noise}-{repetitions}-{true_speedup}")
+    unpaired_rng = random.Random(f"unpaired-{data_seed}")
+    gate = gate_speedup(gate_pct, "minimize")
+    counts = dict.fromkeys(("point_rule", "paired_point", "unpaired_interval",
+                            "paired_interval", "covered"), 0)
+    widths = []
+    for _ in range(campaigns):
+        baseline, candidate = campaign(rng, repetitions, true_speedup, noise)
+        point = median(baseline) / median(candidate)
+        counts["point_rule"] += point >= gate and point > 1
+        pairs = [b / c for b, c in zip(baseline, candidate)]
+        counts["paired_point"] += median(pairs) >= gate
+        counts["unpaired_interval"] += (
+            unpaired_lower_bound(baseline, candidate, unpaired_rng, resamples) > gate
+        )
+        result = paired_statistics(baseline, candidate, direction="minimize",
+                                   min_improvement_pct=gate_pct, point_gate_passed=True,
+                                   resamples=resamples, seed=bootstrap_seed)
+        # Coverage is computed for every repetition count, including those below
+        # MIN_PAIRS where the harness deliberately reports no interval.
+        low, high = bootstrap_median_interval(pairs, resamples=resamples, seed=bootstrap_seed)
+        counts["covered"] += low <= true_speedup <= high
+        widths.append(high / low)
+        counts["paired_interval"] += result.get("gate_relation") == "above_gate"
+    row = {"noise": noise, "repetitions": repetitions, "true_speedup": true_speedup,
+           "harness_reports_interval": repetitions >= MIN_PAIRS}
+    for name, count in counts.items():
+        p = count / campaigns
+        row[name] = {"rate": round(p, 4),
+                     "monte_carlo_se": round(math.sqrt(p * (1 - p) / campaigns), 4)}
+    row["median_interval_width_ratio"] = round(median(widths), 4) if widths else None
+    return row
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--campaigns", type=int, default=200)
+    parser.add_argument("--gate-pct", type=float, default=1.0)
+    parser.add_argument("--data-seed", type=int, default=20261115)
+    parser.add_argument("--resamples", type=int, default=RESAMPLES)
+    parser.add_argument("--bootstrap-seed", type=int, default=SEED)
+    args = parser.parse_args()
+    rows = [rates(repetitions=r, true_speedup=t, noise=n, gate_pct=args.gate_pct,
+                  campaigns=args.campaigns, data_seed=args.data_seed,
+                  resamples=args.resamples, bootstrap_seed=args.bootstrap_seed)
+            for n in NOISES for r in REPETITIONS for t in TRUE_SPEEDUPS]
+    print(json.dumps({"settings": vars(args), "rows": rows}, indent=2))
+
+
+if __name__ == "__main__":
+    main()

From 8019114b13c876f2f56e84400738803cf189f785 Mon Sep 17 00:00:00 2001
From: BeatSeat <137616995+BeatSeat@users.noreply.github.com>
Date: Sun, 15 Nov 2026 20:21:30 +0800
Subject: [PATCH 2/5] harness: explain gains with profile evidence and opt-in
 decision switches

Measurement scripts may add per-step region breakdowns and work counters.
A round may declare its mechanism and expected region/counter directions.
Regions declared unchanged act as an in-step clock, so ratios to them and
their share of the step total cancel device-wide clock or thermal drift;
counters such as launches or DRAM bytes show whether a small gain is
reproducible. Directions come only from paired intervals, with no
repository-wide tolerance.

Two switches are frozen at campaign start and default to the existing
behaviour: --require-explanation adopts only candidates whose mechanism is
confirmed by the data, and --decision paired requires the paired interval
to clear the gate, failing within-noise rounds without blocking the search.

Agent-assisted-by: Claude Code / claude-opus-5-5 (author)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: BeatSeat <137616995+BeatSeat@users.noreply.github.com>
---
 docs/optimization-workflow-unification.md    |  30 ++++
 src/inferrsi/arguments.py                    |   8 +
 src/inferrsi/harness/measurement.py          | 158 +++++++++++++++++
 src/inferrsi/harness/optimization.py         |  11 +-
 src/inferrsi/harness/optimization_session.py |  13 +-
 src/inferrsi/harness/project.py              |  72 +++++++-
 tests/test_harness_measurement.py            |  75 ++++++++
 tests/test_optimization_session.py           |  11 ++
 tests/test_project_explanation.py            | 175 +++++++++++++++++++
 9 files changed, 543 insertions(+), 10 deletions(-)
 create mode 100644 tests/test_project_explanation.py

diff --git a/docs/optimization-workflow-unification.md b/docs/optimization-workflow-unification.md
index f89db1ab8..ed5db388d 100644
--- a/docs/optimization-workflow-unification.md
+++ b/docs/optimization-workflow-unification.md
@@ -197,6 +197,36 @@ inferrsi dashboard --input <campaign-dir> --output <report.html>
 
 知识检索和 profiling 按当前问题需要进行，不要求每轮重查或每轮 profile。知识关闭时受管 `knowledge` 操作拒绝执行；这不是对宿主所有文件访问的隔离。
 
+### 用 profile 解释改进（可选）
+
+端到端时间会随时钟、温度和共享负载整体漂移，单凭数值很难区分真实改进与波动。测量脚本可以在结果 JSON 中额外输出同一步骤的分区域耗时和工作量计数，单位由脚本自定并在各步骤保持一致：
+
+```json
+{"metric": 8.1,
+ "breakdown": {"attention": 4.0, "norm": 0.1, "other": 4.0},
+ "counters": {"kernel_launches": 20, "dram_bytes": 8.0e8}}
+```
+
+每轮可以用 `--mechanism-file` 声明机制和可核对的预期特征：
+
+```json
+{"mechanism": "把 norm 融合进 attention 的 epilogue，少一次 HBM 往返",
+ "expect": [{"region": "norm", "direction": "decrease"},
+            {"region": "attention", "direction": "unchanged"},
+            {"counter": "kernel_launches", "direction": "decrease"}]}
+```
+
+assessment 的 `profile_evidence` 按重复配对比较每个区域和计数。声明为 `unchanged` 的区域作为同一步骤内的参照：其他区域先除以参照区域再比较，`share_based_speedup` 用参照区域在步骤总时间中的占比推算整体加速，因此整卡同比例的漂移会被抵消。计数一类的工作量指标基本不随时钟和温度变化，可以说明波动范围内的小幅改进是否可重复。方向只由配对区间判断（区间整体低于或高于 1），仓库内没有设置数值容差；参照区域如果实际也变了，推算会有偏差，`profile_evidence` 只描述数据，不判断机制文字本身。
+
+`explanation.status` 取值：`explained`（每条预期都被数据确认，且至少一条是变化）、`unsupported`（预期的变化没有在数据中出现）、`contradicted`（数据与预期方向相反，或声明不变的项出现明显变化）、`unverifiable`（缺少对应区域或计数，或少于 5 对）、`not_declared`（本轮未声明）。final 沿用被选中轮次的机制重新核对。
+
+两个开关在 `campaign start` 时冻结，默认都不改变原有判定：
+
+- `--require-explanation`：只有 `explained` 的候选可以被采纳。`unsupported`、`contradicted` 记为 FAIL，`unverifiable`、`not_declared` 记为 INCONCLUSIVE；两类都不阻塞后续轮次。
+- `--decision paired`：用 `paired_statistics` 判定，要求至少 5 次重复。区间整体高于门槛为 PASS；跨过门槛为 FAIL，`reason` 为 `within_noise`，搜索继续；统计不可用时为 INCONCLUSIVE。默认 `point` 保持中位数点估计规则。
+
+提交未修改的候选可以作为 A/A 对照，`paired_statistics` 的区间即给出该环境的噪声范围。
+
 ## 实际保留什么、保护什么
 
 `campaign.json` 关联计划，`plan.json` 固定协议、预算和候选格式，`state.json` 是唯一可写控制状态。`summary.json` 是可重建摘要，dashboard 从同一记录生成展示；不额外写一份 `rounds.jsonl`。原有知识交付和优化日志格式继续复用。
diff --git a/src/inferrsi/arguments.py b/src/inferrsi/arguments.py
index e943f2ec2..00dcbc6da 100644
--- a/src/inferrsi/arguments.py
+++ b/src/inferrsi/arguments.py
@@ -67,6 +67,12 @@ def _optimization_actions(stages, *, project=False):
         start.add_argument("--min-improvement-pct", type=float, required=True)
         start.add_argument("--repetitions", type=int, required=True)
         start.add_argument("--warmups", type=int, required=True)
+        start.add_argument("--decision", choices=("point", "paired"), default="point",
+                           help="point: independent medians (default); paired: the paired "
+                                "interval must clear the gate, within-noise rounds fail")
+        start.add_argument("--require-explanation", action="store_true",
+                           help="Adopt a candidate only when its declared mechanism is "
+                                "confirmed by measured breakdown/counters")
         start.add_argument("--round-mode", choices=("exact", "up_to"), default="up_to")
     else:
         start.add_argument("task")
@@ -89,6 +95,8 @@ def _optimization_actions(stages, *, project=False):
         if name == "round":
             stage.add_argument("--candidate", type=Path, required=True)
             stage.add_argument("--hypothesis", required=True)
+            stage.add_argument("--mechanism-file", type=Path,
+                               help="JSON mechanism with expected region/counter directions")
         if name == "final":
             stage.add_argument("--reason", help="Why the caller stopped an up-to-budget search")
         if name in ("round", "final"):
diff --git a/src/inferrsi/harness/measurement.py b/src/inferrsi/harness/measurement.py
index 404be1512..bf2f48b4f 100644
--- a/src/inferrsi/harness/measurement.py
+++ b/src/inferrsi/harness/measurement.py
@@ -146,3 +146,161 @@ def paired_statistics(baseline: list, candidate: list, *, direction: str,
         return {**header, "status": "unavailable",
                 "reason": "A derived statistic is not finite", "interval": None}
     return {**header, **report}
+
+
+# Profile evidence and declared mechanisms ---------------------------------------
+#
+# A measurement script may add per-step ``breakdown`` (region durations in one
+# unit) and ``counters`` (work metrics such as kernel launches or DRAM bytes).
+# Work metrics are largely independent of clock and thermal state, and region
+# shares within one step cancel device-wide fluctuation, so they can explain a
+# gain that end-to-end time alone cannot separate from noise. Directions are
+# decided only from the paired data; no numeric tolerance is built in.
+
+DIRECTIONS = ("decrease", "increase", "unchanged")
+
+
+def _numeric_map(value) -> dict | None:
+    if not isinstance(value, dict) or not value:
+        return None
+    if not all(isinstance(k, str) and k and type(v) in (int, float) and math.isfinite(v)
+               and v >= 0 for k, v in value.items()):
+        return None
+    return value
+
+
+def _paired_change(baseline: list, candidate: list, *, resamples: int, seed: int) -> dict:
+    ratios = []
+    for b, c in zip(baseline, candidate):
+        if b > 0:
+            ratios.append(c / b)
+        elif c == 0:
+            ratios.append(1.0)
+        else:
+            return {"ratio_median": None, "interval": None, "observed": "increase",
+                    "note": "baseline value is zero and candidate value is positive"}
+    report = {"ratio_median": median(ratios), "interval": None}
+    if len(ratios) < MIN_PAIRS:
+        return {**report, "observed": "insufficient_pairs"}
+    low, high = bootstrap_median_interval(ratios, resamples=resamples, seed=seed)
+    observed = "decrease" if high < 1 else "increase" if low > 1 else "no_clear_change"
+    return {**report, "interval": [low, high], "observed": observed}
+
+
+def _interval_report(values: list, *, resamples: int, seed: int) -> dict:
+    report = {"median": median(values), "interval": None}
+    if len(values) >= MIN_PAIRS:
+        report["interval"] = bootstrap_median_interval(values, resamples=resamples, seed=seed)
+    return report
+
+
+def profile_evidence(baseline_profiles: list, candidate_profiles: list, *,
+                     reference_regions=(), resamples: int = RESAMPLES,
+                     seed: int = SEED) -> dict:
+    """Compare per-repetition region durations and work counters between paired steps.
+
+    ``reference_regions`` are regions the round declares unchanged. Within one step
+    they act as an internal clock: dividing another region by them, and reading the
+    reference share of the step total, cancels device-wide changes such as clock or
+    temperature drift that scale every region of that step together.
+    """
+    if len(baseline_profiles) != len(candidate_profiles) or not baseline_profiles:
+        return {"status": "unavailable", "reason": "Profiles must be paired per repetition"}
+    sides = (baseline_profiles, candidate_profiles)
+    result = {"status": "available", "pairs": len(baseline_profiles), "regions": {},
+              "counters": {}, "reference_regions": [], "share_based_speedup": None}
+    for kind in ("breakdown", "counters"):
+        maps = [[_numeric_map((p or {}).get(kind)) for p in side] for side in sides]
+        if not all(m is not None for side in maps for m in side):
+            continue
+        shared = sorted(set.intersection(*(set(m) for side in maps for m in side)))
+        reference = [r for r in reference_regions if r in shared] if kind == "breakdown" else []
+        if kind == "breakdown":
+            result["reference_regions"] = reference
+        target = result["regions" if kind == "breakdown" else "counters"]
+        for name in shared:
+            b = [m[name] for m in maps[0]]
+            c = [m[name] for m in maps[1]]
+            entry = {"baseline_median": median(b), "candidate_median": median(c),
+                     **_paired_change(b, c, resamples=resamples, seed=seed), "basis": "raw"}
+            clock = [r for r in reference if r != name]
+            if clock:
+                bn = [m[name] / sum(m[r] for r in clock) for m in maps[0]
+                      if sum(m[r] for r in clock) > 0]
+                cn = [m[name] / sum(m[r] for r in clock) for m in maps[1]
+                      if sum(m[r] for r in clock) > 0]
+                if len(bn) == len(cn) == len(b):
+                    entry.update(_paired_change(bn, cn, resamples=resamples, seed=seed),
+                                 basis="normalized_to_reference_regions")
+            if kind == "breakdown":
+                totals = [[sum(m.values()) for m in side] for side in maps]
+                entry["baseline_share"] = median(v / t for v, t in zip(b, totals[0]) if t > 0)
+                entry["candidate_share"] = median(v / t for v, t in zip(c, totals[1]) if t > 0)
+            target[name] = entry
+        if reference:
+            ratios = []
+            for mb, mc in zip(*maps):
+                ref_b, ref_c = (sum(m[r] for r in reference) for m in (mb, mc))
+                tot_b, tot_c = sum(mb.values()), sum(mc.values())
+                if min(ref_b, ref_c, tot_b, tot_c) > 0:
+                    ratios.append((ref_c / tot_c) / (ref_b / tot_b))
+            if len(ratios) == len(maps[0]):
+                result["share_based_speedup"] = _interval_report(
+                    ratios, resamples=resamples, seed=seed)
+    if not result["regions"] and not result["counters"]:
+        return {**result, "status": "absent",
+                "reason": "No breakdown or counters were present in every measured step"}
+    return result if _finite(result) else {"status": "unavailable",
+                                           "reason": "A derived profile value is not finite"}
+
+
+def validate_mechanism(document) -> dict:
+    """Validate a declared mechanism: a statement plus checkable expected signatures."""
+    if not isinstance(document, dict) or set(document) - {"mechanism", "expect"}:
+        raise ValueError("A mechanism declares only 'mechanism' and 'expect'")
+    text = document.get("mechanism")
+    expect = document.get("expect")
+    if not isinstance(text, str) or not text.strip():
+        raise ValueError("Describe the mechanism that should cause the change")
+    if not isinstance(expect, list) or not expect:
+        raise ValueError("List at least one expected region or counter signature")
+    for item in expect:
+        keys = set(item) if isinstance(item, dict) else set()
+        kinds = keys & {"region", "counter"}
+        if (len(kinds) != 1 or keys - {"region", "counter", "direction"}
+                or not isinstance(item[next(iter(kinds))], str)
+                or not item[next(iter(kinds))].strip()
+                or item.get("direction") not in DIRECTIONS):
+            raise ValueError("Each expectation names one region or counter and a direction "
+                             "of decrease, increase or unchanged")
+    if all(item["direction"] == "unchanged" for item in expect):
+        raise ValueError("At least one expectation must predict a change")
+    return document
+
+
+def explain(mechanism: dict | None, evidence: dict) -> dict:
+    """Check a declared mechanism against observed profile evidence, without judging prose."""
+    if mechanism is None:
+        return {"status": "not_declared"}
+    checks, missing = [], False
+    for item in mechanism["expect"]:
+        kind = "region" if "region" in item else "counter"
+        table = evidence.get("regions" if kind == "region" else "counters") or {}
+        observed = table.get(item[kind], {}).get("observed")
+        if observed in (None, "insufficient_pairs"):
+            missing = True
+        checks.append({"kind": kind, "name": item[kind], "expected": item["direction"],
+                       "observed": observed or "missing"})
+    report = {"mechanism": mechanism["mechanism"], "checks": checks}
+    if any(c["expected"] in ("decrease", "increase") and c["observed"] in ("decrease", "increase")
+           and c["observed"] != c["expected"] for c in checks) or any(
+            c["expected"] == "unchanged" and c["observed"] in ("decrease", "increase")
+            for c in checks):
+        return {**report, "status": "contradicted"}
+    if missing or evidence.get("status") != "available":
+        return {**report, "status": "unverifiable"}
+    if all(c["observed"] == c["expected"] or (c["expected"] == "unchanged"
+                                              and c["observed"] == "no_clear_change")
+           for c in checks):
+        return {**report, "status": "explained"}
+    return {**report, "status": "unsupported"}
diff --git a/src/inferrsi/harness/optimization.py b/src/inferrsi/harness/optimization.py
index bbd439227..4a2c86548 100644
--- a/src/inferrsi/harness/optimization.py
+++ b/src/inferrsi/harness/optimization.py
@@ -287,7 +287,8 @@ def _settle(root, plan, state, pending, execution, assess):
 
 
 def evaluate_round(root: Path, *, candidate: Path | None, hypothesis: str,
-                   usage: Path | None, execute, assess, final=False, reason=None) -> dict:
+                   usage: Path | None, execute, assess, final=False, reason=None,
+                   mechanism: dict | None = None) -> dict:
     with _locked(root):
         plan, state = _load(root)
         _active(state, reserve=0 if final else plan["final_reserve_seconds"])
@@ -315,10 +316,18 @@ def evaluate_round(root: Path, *, candidate: Path | None, hypothesis: str,
             (directory / "agent-usage.json").write_bytes(usage_bytes)
         fingerprint = candidates.retain(source, directory / plan["candidate_name"],
                                         plan.get("candidate_includes"))
+        if final and state["best"]["kind"] == "candidate":
+            # The independent final rechecks the mechanism declared by the selected round.
+            selected = (root / state["best"]["path"]).parent / "mechanism.json"
+            if selected.is_file():
+                mechanism = json.loads(selected.read_text(encoding="utf-8"))
+        if mechanism is not None:
+            atomic_write_json(directory / "mechanism.json", mechanism)
         pending = {"kind": "final" if final else "round", "path": relative,
                    "candidate_id": state["best"].get("candidate_id", "baseline") if final
                    else f"candidate-{count + 1:03d}",
                    "hypothesis": hypothesis, "observations": observations,
+                   "mechanism": mechanism,
                    "candidate_sha256": fingerprint,
                    "contexts": list(state["contexts"]), "started_at": time.time()}
         state["pending"] = pending
diff --git a/src/inferrsi/harness/optimization_session.py b/src/inferrsi/harness/optimization_session.py
index b9fdee40b..de81d669a 100644
--- a/src/inferrsi/harness/optimization_session.py
+++ b/src/inferrsi/harness/optimization_session.py
@@ -154,7 +154,8 @@ def session_command(args, *, task_source: Path | None = None, knowledge_reader=N
             identity(repo, args.include)  # Reject invalid inputs before preparation writes.
             specification = {key: getattr(args, key) for key in (
                 "metric_key", "metric_unit", "quality_key", "scope", "direction",
-                "min_improvement_pct", "repetitions", "warmups")}
+                "min_improvement_pct", "repetitions", "warmups", "decision",
+                "require_explanation")}
             specification.update(measure_command=shlex.split(args.measure_command),
                                  quality_command=shlex.split(args.quality_command))
             settings = prepare_project(root / "frozen", args.protocol.resolve(),
@@ -215,11 +216,19 @@ def session_command(args, *, task_source: Path | None = None, knowledge_reader=N
     if args.session_action == "recover":
         return optimization.recover(root, _session_assess)
     final = args.session_action == "final"
+    mechanism = None
+    if not final and getattr(args, "mechanism_file", None) is not None:
+        if plan.get("task_source") != "project":
+            raise ValueError("Mechanism checks apply to project campaign rounds only")
+        from inferrsi.harness.measurement import validate_mechanism
+
+        mechanism = validate_mechanism(
+            json.loads(Path(args.mechanism_file).read_text(encoding="utf-8")))
     return optimization.evaluate_round(
         root, candidate=None if final else args.candidate,
         hypothesis="Independent final of the selected candidate" if final else args.hypothesis,
         usage=args.usage_export, execute=_session_execute, assess=_session_assess, final=final,
-        reason=getattr(args, "reason", None),
+        reason=getattr(args, "reason", None), mechanism=mechanism,
     )
 
 
diff --git a/src/inferrsi/harness/project.py b/src/inferrsi/harness/project.py
index df9d091a0..f3797b8eb 100644
--- a/src/inferrsi/harness/project.py
+++ b/src/inferrsi/harness/project.py
@@ -11,7 +11,9 @@
 from statistics import median
 
 from ..infra.artifacts import sha256
-from .measurement import paired_statistics
+from .measurement import MIN_PAIRS, explain, paired_statistics, profile_evidence, validate_mechanism
+
+DECISIONS = ("point", "paired")
 
 
 def validate(settings: dict) -> None:
@@ -35,6 +37,12 @@ def validate(settings: dict) -> None:
     value = settings["min_improvement_pct"]
     if type(value) not in (int, float) or not math.isfinite(value) or value < 0:
         raise ValueError("An explicit finite nonnegative improvement gate is required")
+    if settings.get("decision", "point") not in DECISIONS:
+        raise ValueError("Decision must be point or paired")
+    if settings.get("decision") == "paired" and settings["repetitions"] < MIN_PAIRS:
+        raise ValueError(f"A paired decision needs at least {MIN_PAIRS} repetitions")
+    if type(settings.get("require_explanation", False)) is not bool:
+        raise ValueError("require_explanation must be a boolean")
 
 
 def prepare(destination: Path, protocol: Path, instructions: Path, settings: dict) -> dict:
@@ -117,7 +125,19 @@ def inspect_step(directory: Path, settings: dict, definition: dict, execution: d
         type(value) not in (int, float) or not math.isfinite(value) or value <= 0
     ):
         return {**failure, "reason": "invalid_metric", "evidence_files": evidence}
-    return {"status": "completed", "value": value, "evidence_files": evidence}
+    profile = {key: document[key] for key in ("breakdown", "counters") if key in document}
+    return {"status": "completed", "value": value, "profile": profile,
+            "evidence_files": evidence}
+
+
+def read_mechanism(directory: Path) -> dict | None:
+    """Read the mechanism retained with an attempt, if the round declared one."""
+    path = directory / "mechanism.json"
+    if not path.exists():
+        return None
+    if path.is_symlink():
+        raise ValueError("Symbolic mechanism declaration")
+    return validate_mechanism(json.loads(path.read_bytes()))
 
 
 def assess(directory: Path, settings: dict, execution: dict) -> dict:
@@ -135,6 +155,7 @@ def result(status, reason, **fields):
             or any(c.get("verified") is not True for c in checks)):
         return result("infrastructure_failure", "incomplete_execution")
     samples = {"baseline": [], "candidate": []}
+    profiles = {"baseline": [], "candidate": []}
     candidate_quality = None
     for definition, record in zip(expected, actual):
         if record.get("id") != definition["id"]:
@@ -150,6 +171,7 @@ def result(status, reason, **fields):
             candidate_quality = True
         if definition["phase"] == "measurement":
             samples[definition["side"]].append(checked["value"])
+            profiles[definition["side"]].append(checked["profile"])
     if execution.get("status") != "completed" or len(actual) != len(expected):
         return result("infrastructure_failure", "incomplete_execution")
     before, after = median(samples["baseline"]), median(samples["candidate"])
@@ -159,20 +181,56 @@ def result(status, reason, **fields):
     speedup = before / after if settings["direction"] == "minimize" else after / before
     if not all(math.isfinite(value) for value in (improvement, speedup)):
         return result("infrastructure_failure", "nonfinite_derived_metric")
-    accepted = improvement > 0 and improvement >= settings["min_improvement_pct"]
+    point_accepted = improvement > 0 and improvement >= settings["min_improvement_pct"]
     paired = paired_statistics(samples["baseline"], samples["candidate"],
                                direction=settings["direction"],
                                min_improvement_pct=settings["min_improvement_pct"],
-                               point_gate_passed=accepted)
+                               point_gate_passed=point_accepted)
+    decision = settings.get("decision", "point")
+    verdict, reason = (("PASS", "fixed_checks_satisfied") if point_accepted
+                       else ("FAIL", "performance_gate_not_met"))
+    if decision == "paired":
+        relation = paired.get("gate_relation")
+        if paired.get("status") != "available":
+            verdict, reason = "INCONCLUSIVE", "paired_statistics_unavailable"
+        elif relation == "above_gate":
+            verdict, reason = "PASS", "paired_interval_above_gate"
+        elif relation == "straddles_gate":
+            verdict, reason = "FAIL", "within_noise"
+        else:
+            verdict, reason = "FAIL", "performance_gate_not_met"
+    try:
+        mechanism = read_mechanism(directory)
+        if mechanism is not None:
+            evidence["mechanism.json"] = sha256(directory / "mechanism.json")
+    except (OSError, ValueError) as error:
+        mechanism, mechanism_error = None, str(error)
+    else:
+        mechanism_error = None
+    reference = [item["region"] for item in (mechanism or {}).get("expect", [])
+                 if "region" in item and item["direction"] == "unchanged"]
+    observed = profile_evidence(profiles["baseline"], profiles["candidate"],
+                                reference_regions=reference)
+    explanation = explain(mechanism, observed)
+    if mechanism_error:
+        explanation = {"status": "unverifiable", "reason": mechanism_error}
+    if settings.get("require_explanation", False) and verdict == "PASS":
+        if explanation["status"] in ("contradicted", "unsupported"):
+            verdict, reason = "FAIL", "explanation_" + explanation["status"]
+        elif explanation["status"] != "explained":
+            verdict, reason = "INCONCLUSIVE", "explanation_" + explanation["status"]
+    accepted = verdict == "PASS"
     return result(
-        "completed", "fixed_checks_satisfied" if accepted else "performance_gate_not_met",
-        verdict="PASS" if accepted else "FAIL", score=after, baseline_score=before,
+        "completed", reason,
+        verdict=verdict, score=after, baseline_score=before,
         eligible=accepted, quality_passed=True, improvement_pct=improvement, speedup=speedup,
         metrics={"name": settings["metric_key"], "unit": settings["metric_unit"],
                  "direction": settings["direction"], "baseline": before, "candidate": after,
                  "samples": samples},
         scope=settings["scope"], statistical_significance="not_established",
-        paired_statistics=paired,
+        paired_statistics=paired, decision=decision,
+        require_explanation=settings.get("require_explanation", False),
+        profile_evidence=observed, explanation=explanation,
         quality_contract="Only the supplied frozen quality script and its boolean result",
         environment_qualification="Caller-managed; no automatic hardware isolation claim",
     )
diff --git a/tests/test_harness_measurement.py b/tests/test_harness_measurement.py
index d8a4a47db..abc83648c 100644
--- a/tests/test_harness_measurement.py
+++ b/tests/test_harness_measurement.py
@@ -10,9 +10,12 @@
 from inferrsi.harness import optimization_session
 from inferrsi.harness.measurement import (
     bootstrap_median_interval,
+    explain,
     gate_speedup,
     paired_speedups,
     paired_statistics,
+    profile_evidence,
+    validate_mechanism,
 )
 
 BASE = [10.0, 10.1, 9.9, 10.05, 9.95]
@@ -177,3 +180,75 @@ def test_measurement_module_is_part_of_runtime_identity():
 def test_invalid_resample_count_reports_unavailable(resamples):
     result = report(BASE, [8.0, 8.1, 7.9, 8.02, 7.98], resamples=resamples)
     assert result["status"] == "unavailable" and result["interval"] is None
+
+
+def profiles(rows, jitter=None):
+    """Per-repetition profiles; jitter scales every region of a step together."""
+    jitter = jitter or [1.0] * len(rows)
+    return [{"breakdown": {k: v * g for k, v in r["breakdown"].items()},
+             "counters": r.get("counters", {"launches": 10})} for r, g in zip(rows, jitter)]
+
+
+STEP_B = {"breakdown": {"gemm": 6.0, "norm": 2.0}, "counters": {"launches": 12, "bytes": 900}}
+STEP_C = {"breakdown": {"gemm": 6.0, "norm": 1.0}, "counters": {"launches": 8, "bytes": 900}}
+JITTER_B = [1.0, 1.08, 0.93, 1.05, 0.97]
+JITTER_C = [1.06, 0.94, 1.02, 0.95, 1.09]
+
+
+def test_reference_regions_cancel_device_wide_jitter():
+    evidence = profile_evidence(profiles([STEP_B] * 5, JITTER_B),
+                                profiles([STEP_C] * 5, JITTER_C), reference_regions=["gemm"])
+    assert evidence["status"] == "available" and evidence["reference_regions"] == ["gemm"]
+    norm = evidence["regions"]["norm"]
+    assert norm["basis"] == "normalized_to_reference_regions"
+    assert norm["ratio_median"] == pytest.approx(0.5) and norm["observed"] == "decrease"
+    assert evidence["share_based_speedup"]["median"] == pytest.approx(8 / 7)
+    assert evidence["share_based_speedup"]["interval"] == pytest.approx([8 / 7, 8 / 7])
+    assert evidence["counters"]["launches"]["observed"] == "decrease"
+    assert evidence["counters"]["bytes"]["observed"] == "no_clear_change"
+    # Without a reference the same region is only compared raw, jitter included.
+    raw = profile_evidence(profiles([STEP_B] * 5, JITTER_B), profiles([STEP_C] * 5, JITTER_C))
+    assert raw["regions"]["norm"]["basis"] == "raw" and raw["share_based_speedup"] is None
+
+
+def test_profile_evidence_absent_or_unpaired():
+    assert profile_evidence([{}] * 5, [{}] * 5)["status"] == "absent"
+    assert profile_evidence([STEP_B], [STEP_C, STEP_C])["status"] == "unavailable"
+    few = profile_evidence(profiles([STEP_B] * 2), profiles([STEP_C] * 2))
+    assert few["counters"]["launches"]["observed"] == "insufficient_pairs"
+
+
+MECHANISM = {"mechanism": "Fuse the norm into the GEMM epilogue",
+             "expect": [{"region": "norm", "direction": "decrease"},
+                        {"region": "gemm", "direction": "unchanged"},
+                        {"counter": "launches", "direction": "decrease"}]}
+
+
+@pytest.mark.parametrize("expect,status", [
+    (MECHANISM["expect"], "explained"),
+    ([{"region": "norm", "direction": "increase"}], "contradicted"),
+    ([{"counter": "bytes", "direction": "decrease"}], "unsupported"),
+    ([{"counter": "l2_hits", "direction": "increase"}], "unverifiable"),
+    ([{"counter": "launches", "direction": "decrease"},
+      {"counter": "bytes", "direction": "unchanged"}], "explained"),
+])
+def test_explain_classifies_declared_mechanisms(expect, status):
+    mechanism = validate_mechanism({"mechanism": "m", "expect": expect})
+    reference = [e["region"] for e in expect if "region" in e and e["direction"] == "unchanged"]
+    evidence = profile_evidence(profiles([STEP_B] * 5, JITTER_B),
+                                profiles([STEP_C] * 5, JITTER_C), reference_regions=reference)
+    assert explain(mechanism, evidence)["status"] == status
+    assert explain(None, evidence) == {"status": "not_declared"}
+
+
+@pytest.mark.parametrize("document", [
+    {"mechanism": "", "expect": [{"region": "a", "direction": "decrease"}]},
+    {"mechanism": "m", "expect": []},
+    {"mechanism": "m", "expect": [{"region": "a", "direction": "faster"}]},
+    {"mechanism": "m", "expect": [{"region": "a", "counter": "b", "direction": "decrease"}]},
+    {"mechanism": "m", "expect": [{"region": "a", "direction": "unchanged"}]},
+    {"mechanism": "m", "expect": [{"region": "a", "direction": "decrease"}], "extra": 1},
+])
+def test_invalid_mechanisms_are_rejected(document):
+    with pytest.raises(ValueError):
+        validate_mechanism(document)
diff --git a/tests/test_optimization_session.py b/tests/test_optimization_session.py
index b1a052e8f..c375e8d22 100644
--- a/tests/test_optimization_session.py
+++ b/tests/test_optimization_session.py
@@ -315,3 +315,14 @@ def test_runtime_identity_covers_migrated_process_domain_and_not_removed_executo
     assert "infra/executors/local.py" not in identity
     for name in identity:
         assert (root / name).is_file(), name
+
+
+def test_mechanism_checks_are_rejected_for_packaged_task_sessions(session, tmp_path):
+    mechanism = tmp_path / "mechanism.json"
+    mechanism.write_text(json.dumps({"mechanism": "m", "expect": [
+        {"region": "a", "direction": "decrease"}]}), encoding="utf-8")
+    with pytest.raises(SystemExit):
+        main(["benchmarks", "session", "round", "--session", str(session),
+              "--candidate", str(session / "work/candidate.py"), "--hypothesis", "h",
+              "--mechanism-file", str(mechanism)])
+    assert not (session / "rounds").exists() or not any((session / "rounds").iterdir())
diff --git a/tests/test_project_explanation.py b/tests/test_project_explanation.py
new file mode 100644
index 000000000..a77bd6f21
--- /dev/null
+++ b/tests/test_project_explanation.py
@@ -0,0 +1,175 @@
+"""Decision and mechanism switches through real campaign commands; values are synthetic."""
+
+import json
+
+import pytest
+
+from inferrsi.cli import main
+from inferrsi.harness.optimization import read_records
+
+# Device-wide factor applied to every timed value of one measurement call, cycling per
+# call. It scales all regions together, as clocks or temperature would.
+JITTER = [1.0, 1.06, 0.95, 1.04, 0.97, 1.05, 0.96, 1.03, 0.98, 1.02, 0.94, 1.07]
+
+BASELINE = {"metric": 10.0, "breakdown": {"attention": 4.0, "norm": 2.0, "other": 4.0},
+            "counters": {"kernel_launches": 30, "dram_bytes": 1000}}
+
+
+@pytest.fixture
+def project(tmp_path):
+    repo = tmp_path / "project"
+    repo.mkdir()
+    (repo / "engine.py").write_text("def compute(x):\n    return x * 2\n", encoding="utf-8")
+    (repo / "config.json").write_text(json.dumps(BASELINE), encoding="utf-8")
+    protocol = tmp_path / "protocol"
+    protocol.mkdir()
+    calls = tmp_path / "calls.txt"
+    (protocol / "measure.py").write_text(
+        "import json, sys\nfrom pathlib import Path\n"
+        f"JITTER = {JITTER!r}\nCALLS = Path({str(calls)!r})\n"
+        "k = int(CALLS.read_text()) if CALLS.exists() else 0\n"
+        "CALLS.write_text(str(k + 1))\n"
+        "g = JITTER[k % len(JITTER)]\n"
+        "c = json.loads((Path(sys.argv[1]) / 'config.json').read_text())\n"
+        "out = {'metric': c['metric'] * g,\n"
+        "       'breakdown': {n: v * g for n, v in c['breakdown'].items()},\n"
+        "       'counters': c['counters']}\n"
+        "Path(sys.argv[2]).write_text(json.dumps(out))\n",
+        encoding="utf-8",
+    )
+    (protocol / "quality.py").write_text(
+        "import json, runpy, sys\nfrom pathlib import Path\n"
+        "compute = runpy.run_path(str(Path(sys.argv[1])/'engine.py'))['compute']\n"
+        "Path(sys.argv[2]).write_text(json.dumps({'correct':compute(16)==32}))\n",
+        encoding="utf-8",
+    )
+    instructions = tmp_path / "task.md"
+    instructions.write_text("Synthetic fixture; preserve compute outputs.", encoding="utf-8")
+    return repo, protocol, instructions, tmp_path / "campaign"
+
+
+def start(project, capsys, *extra, repetitions=5):
+    repo, protocol, instructions, root = project
+    code = main([
+        "campaign", "start", "--directory", str(root), "--repo", str(repo),
+        "--include", "engine.py", "--include", "config.json", "--protocol", str(protocol),
+        "--instructions", str(instructions), "--measure-command",
+        '"{python}" "{protocol}/measure.py" "{candidate}" "{result}"',
+        "--quality-command", '"{python}" "{protocol}/quality.py" "{candidate}" "{result}"',
+        "--metric-key", "metric", "--metric-unit", "ms", "--direction", "minimize",
+        "--quality-key", "correct", "--scope", "synthetic-software-control",
+        "--repetitions", str(repetitions), "--warmups", "1", "--min-improvement-pct", "1",
+        "--rounds", "4", "--round-mode", "up_to", "--max-minutes", "5",
+        "--evaluation-timeout-seconds", "20", *extra,
+    ])
+    capsys.readouterr()
+    return code
+
+
+def submit(project, capsys, config, mechanism=None):
+    repo, _, _, root = project
+    (repo / "config.json").write_text(json.dumps(config), encoding="utf-8")
+    args = ["campaign", "round", "--directory", str(root), "--candidate", str(repo),
+            "--hypothesis", "Synthetic candidate"]
+    if mechanism is not None:
+        path = root.parent / "mechanism.json"
+        path.write_text(json.dumps(mechanism), encoding="utf-8")
+        args += ["--mechanism-file", str(path)]
+    assert main(args) == 0
+    return json.loads(capsys.readouterr().out)["latest_observation"]["assessment"]
+
+
+FUSED = {"metric": 8.0, "breakdown": {"attention": 4.0, "norm": 0.0, "other": 4.0},
+         "counters": {"kernel_launches": 20, "dram_bytes": 800}}
+FUSION = {"mechanism": "Fuse the norm into the attention epilogue, removing one HBM pass",
+          "expect": [{"region": "norm", "direction": "decrease"},
+                     {"region": "attention", "direction": "unchanged"},
+                     {"counter": "kernel_launches", "direction": "decrease"},
+                     {"counter": "dram_bytes", "direction": "decrease"}]}
+
+
+def test_confirmed_mechanism_is_adopted_and_rechecked_at_final(project, capsys):
+    assert start(project, capsys, "--require-explanation") == 0
+    assessment = submit(project, capsys, FUSED, FUSION)
+    assert assessment["verdict"] == "PASS" and assessment["eligible"] is True
+    assert assessment["explanation"]["status"] == "explained"
+    evidence = assessment["profile_evidence"]
+    assert evidence["counters"]["dram_bytes"]["ratio_median"] == pytest.approx(0.8)
+    # Device-wide jitter scales every region of a step. Raw cross-run ratios carry it,
+    # while ratios to the declared-unchanged region and its share of the step cancel it.
+    assert evidence["reference_regions"] == ["attention"]
+    norm = evidence["regions"]["norm"]
+    assert norm["basis"] == "normalized_to_reference_regions" and norm["observed"] == "decrease"
+    other = evidence["regions"]["other"]
+    assert other["ratio_median"] == pytest.approx(1.0) and other["interval"] == pytest.approx(
+        [1.0, 1.0])
+    share = evidence["share_based_speedup"]
+    assert share["median"] == pytest.approx(10 / 8) and share["interval"] == pytest.approx(
+        [10 / 8, 10 / 8])
+    attention = evidence["regions"]["attention"]  # raw medians alone carry the jitter
+    assert attention["baseline_median"] != pytest.approx(attention["candidate_median"])
+    assert main(["campaign", "final", "--directory", str(project[3]),
+                 "--reason", "Synthetic search done"]) == 0
+    capsys.readouterr()
+    _, state, _ = read_records(project[3])
+    assert state["final"]["assessment"]["explanation"]["status"] == "explained"
+    assert state["final"]["assessment"]["verdict"] == "PASS"
+
+
+@pytest.mark.parametrize("mechanism,status,verdict", [
+    ({"mechanism": "Claims the norm got slower", "expect": [
+        {"region": "norm", "direction": "increase"}]}, "contradicted", "FAIL"),
+    ({"mechanism": "Claims attention got faster", "expect": [
+        {"region": "attention", "direction": "decrease"}]}, "unsupported", "FAIL"),
+    ({"mechanism": "Names a counter that was never measured", "expect": [
+        {"counter": "l2_hit_rate", "direction": "increase"}]}, "unverifiable", "INCONCLUSIVE"),
+    (None, "not_declared", "INCONCLUSIVE"),
+])
+def test_unexplained_gain_is_not_adopted_and_search_continues(
+    project, capsys, mechanism, status, verdict,
+):
+    assert start(project, capsys, "--require-explanation") == 0
+    assessment = submit(project, capsys, FUSED, mechanism)
+    assert assessment["explanation"]["status"] == status
+    assert assessment["verdict"] == verdict and assessment["eligible"] is False
+    _, state, _ = read_records(project[3])
+    assert state["best"]["kind"] == "baseline" and state["state"] == "active"
+    assessment = submit(project, capsys, FUSED, FUSION)
+    assert assessment["verdict"] == "PASS"
+
+
+def test_explanation_is_report_only_without_the_switch(project, capsys):
+    assert start(project, capsys) == 0
+    contradicted = {"mechanism": "Wrong story", "expect": [
+        {"counter": "dram_bytes", "direction": "increase"}]}
+    assessment = submit(project, capsys, FUSED, contradicted)
+    assert assessment["verdict"] == "PASS" and assessment["eligible"] is True
+    assert assessment["explanation"]["status"] == "contradicted"
+    assert assessment["require_explanation"] is False
+
+
+def test_paired_decision_fails_within_noise_without_blocking(project, capsys):
+    assert start(project, capsys, "--decision", "paired") == 0
+    nominal = {**BASELINE, "metric": 9.85}  # 1.5% nominal gain inside ±6% device jitter
+    assessment = submit(project, capsys, nominal)
+    assert assessment["paired_statistics"]["gate_relation"] == "straddles_gate"
+    assert assessment["verdict"] == "FAIL" and assessment["reason"] == "within_noise"
+    _, state, _ = read_records(project[3])
+    assert state["state"] == "active" and state["best"]["kind"] == "baseline"
+    assessment = submit(project, capsys, FUSED)
+    assert assessment["verdict"] == "PASS"
+    assert assessment["reason"] == "paired_interval_above_gate"
+
+
+def test_paired_decision_requires_enough_repetitions(project, capsys):
+    with pytest.raises(SystemExit):
+        start(project, capsys, "--decision", "paired", repetitions=3)
+
+
+def test_invalid_mechanism_is_rejected_before_execution(project, capsys):
+    assert start(project, capsys, "--require-explanation") == 0
+    with pytest.raises(SystemExit):
+        submit(project, capsys, FUSED, {"mechanism": "x", "expect": [
+            {"region": "norm", "direction": "faster"}]})
+    _, state, _ = read_records(project[3])
+    assert state["rounds"] == []

From 0767ad452bf55942bbe2c183febf766a64e3dfd9 Mon Sep 17 00:00:00 2001
From: BeatSeat <137616995+BeatSeat@users.noreply.github.com>
Date: Sun, 15 Nov 2026 21:06:06 +0800
Subject: [PATCH 3/5] harness: bind declared mechanisms before execution and
 harden profile evidence

Bind the round's mechanism into control state before the candidate runs and
assess against that copy, so a candidate cannot rewrite its own declaration.
Require five pairs before any profile direction, treat zero-baseline ratios as
undefined, report unavailable shares instead of raising, keep raw evidence for
every region, and never normalize declared references against each other.

Agent-assisted-by: Claude Code / claude-opus-5-5 (author)
Agent-assisted-by: Codex CLI / gpt-6.1-sol [effort=xhigh] (review)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: BeatSeat <137616995+BeatSeat@users.noreply.github.com>
---
 docs/optimization-workflow-unification.md    |  4 +-
 src/inferrsi/harness/measurement.py          | 95 ++++++++++----------
 src/inferrsi/harness/optimization.py         |  8 +-
 src/inferrsi/harness/optimization_session.py |  6 +-
 src/inferrsi/harness/project.py              | 21 ++---
 tests/test_harness_measurement.py            | 40 ++++++++-
 tests/test_project_explanation.py            | 29 +++++-
 7 files changed, 128 insertions(+), 75 deletions(-)

diff --git a/docs/optimization-workflow-unification.md b/docs/optimization-workflow-unification.md
index ed5db388d..2a7634224 100644
--- a/docs/optimization-workflow-unification.md
+++ b/docs/optimization-workflow-unification.md
@@ -216,9 +216,9 @@ inferrsi dashboard --input <campaign-dir> --output <report.html>
             {"counter": "kernel_launches", "direction": "decrease"}]}
 ```
 
-assessment 的 `profile_evidence` 按重复配对比较每个区域和计数。声明为 `unchanged` 的区域作为同一步骤内的参照：其他区域先除以参照区域再比较，`share_based_speedup` 用参照区域在步骤总时间中的占比推算整体加速，因此整卡同比例的漂移会被抵消。计数一类的工作量指标基本不随时钟和温度变化，可以说明波动范围内的小幅改进是否可重复。方向只由配对区间判断（区间整体低于或高于 1），仓库内没有设置数值容差；参照区域如果实际也变了，推算会有偏差，`profile_evidence` 只描述数据，不判断机制文字本身。
+assessment 的 `profile_evidence` 按重复配对比较每个区域和计数，`raw` 保留原始比值。声明为 `unchanged` 的区域是本轮的假设，不是结论：其他区域另外给出除以同一步骤参照区域总和后的 `normalized` 比较，`share_based_speedup` 用参照区域在步骤总时间中的占比推算整体加速。只有当某种变化让同一步骤的所有区域按同一比例缩放时，它才会被抵消；若各区域受影响程度不同（例如降频对计算受限和访存受限的区域影响不同），抵消就不成立。参照区域本身始终只做原始比较，不会互相归一化，因此整体漂移较大时它可能显示为变化，使解释判为矛盾。计数一类的工作量指标基本不随时钟和温度变化，可以说明波动范围内的小幅改进是否可重复。方向只由配对区间判断（区间整体低于或高于 1），仓库内没有设置数值容差；基线为 0 的计数没有比值，记为 `undefined`。`profile_evidence` 只描述数据，不判断机制文字本身。
 
-`explanation.status` 取值：`explained`（每条预期都被数据确认，且至少一条是变化）、`unsupported`（预期的变化没有在数据中出现）、`contradicted`（数据与预期方向相反，或声明不变的项出现明显变化）、`unverifiable`（缺少对应区域或计数，或少于 5 对）、`not_declared`（本轮未声明）。final 沿用被选中轮次的机制重新核对。
+`explanation.status` 取值：`explained`（每条预期都被数据确认，且至少一条是变化）、`unsupported`（预期的变化没有在数据中出现）、`contradicted`（数据与预期方向相反，或声明不变的项出现明显变化）、`unverifiable`（缺少对应区域或计数、比值无定义，或少于 5 对）、`not_declared`（本轮未声明）。机制声明在执行前写入控制状态，判定使用这份绑定的声明，attempt 目录中的 `mechanism.json` 只是可读副本；final 沿用被选中轮次绑定的机制重新核对。
 
 两个开关在 `campaign start` 时冻结，默认都不改变原有判定：
 
diff --git a/src/inferrsi/harness/measurement.py b/src/inferrsi/harness/measurement.py
index bf2f48b4f..f40202634 100644
--- a/src/inferrsi/harness/measurement.py
+++ b/src/inferrsi/harness/measurement.py
@@ -170,28 +170,21 @@ def _numeric_map(value) -> dict | None:
 
 
 def _paired_change(baseline: list, candidate: list, *, resamples: int, seed: int) -> dict:
-    ratios = []
-    for b, c in zip(baseline, candidate):
-        if b > 0:
-            ratios.append(c / b)
-        elif c == 0:
-            ratios.append(1.0)
-        else:
-            return {"ratio_median": None, "interval": None, "observed": "increase",
-                    "note": "baseline value is zero and candidate value is positive"}
-    report = {"ratio_median": median(ratios), "interval": None}
-    if len(ratios) < MIN_PAIRS:
-        return {**report, "observed": "insufficient_pairs"}
+    """Paired ratio direction; a zero baseline makes the ratio undefined, never decisive."""
+    if len(baseline) < MIN_PAIRS:
+        return {"ratio_median": None, "interval": None, "observed": "insufficient_pairs"}
+    if any(b <= 0 for b in baseline):
+        return {"ratio_median": None, "interval": None, "observed": "undefined",
+                "note": "A zero baseline value has no paired ratio"}
+    ratios = [c / b for b, c in zip(baseline, candidate)]
     low, high = bootstrap_median_interval(ratios, resamples=resamples, seed=seed)
     observed = "decrease" if high < 1 else "increase" if low > 1 else "no_clear_change"
-    return {**report, "interval": [low, high], "observed": observed}
+    return {"ratio_median": median(ratios), "interval": [low, high], "observed": observed}
 
 
-def _interval_report(values: list, *, resamples: int, seed: int) -> dict:
-    report = {"median": median(values), "interval": None}
-    if len(values) >= MIN_PAIRS:
-        report["interval"] = bootstrap_median_interval(values, resamples=resamples, seed=seed)
-    return report
+def _shares(values: list, totals: list) -> float | None:
+    shares = [v / t for v, t in zip(values, totals) if t > 0]
+    return median(shares) if len(shares) == len(values) else None
 
 
 def profile_evidence(baseline_profiles: list, candidate_profiles: list, *,
@@ -199,10 +192,11 @@ def profile_evidence(baseline_profiles: list, candidate_profiles: list, *,
                      seed: int = SEED) -> dict:
     """Compare per-repetition region durations and work counters between paired steps.
 
-    ``reference_regions`` are regions the round declares unchanged. Within one step
-    they act as an internal clock: dividing another region by them, and reading the
-    reference share of the step total, cancels device-wide changes such as clock or
-    temperature drift that scale every region of that step together.
+    Every region and counter is compared raw. ``reference_regions`` are regions the
+    round declares unchanged; they are an assumption, not a conclusion. Other regions
+    are additionally compared after dividing by the reference total of the same step,
+    which cancels a change that scales every region of that step by the same factor.
+    References themselves are only ever checked raw, against each other never.
     """
     if len(baseline_profiles) != len(candidate_profiles) or not baseline_profiles:
         return {"status": "unavailable", "reason": "Profiles must be paired per repetition"}
@@ -214,39 +208,38 @@ def profile_evidence(baseline_profiles: list, candidate_profiles: list, *,
         if not all(m is not None for side in maps for m in side):
             continue
         shared = sorted(set.intersection(*(set(m) for side in maps for m in side)))
-        reference = [r for r in reference_regions if r in shared] if kind == "breakdown" else []
-        if kind == "breakdown":
-            result["reference_regions"] = reference
         target = result["regions" if kind == "breakdown" else "counters"]
+        reference = [r for r in dict.fromkeys(reference_regions) if r in shared] if (
+            kind == "breakdown") else []
+        clock = [[sum(m[r] for r in reference) for m in side] for side in maps]
+        usable_clock = bool(reference) and min(min(side) for side in clock) > 0
+        totals = [[sum(m.values()) for m in side] for side in maps]
         for name in shared:
             b = [m[name] for m in maps[0]]
             c = [m[name] for m in maps[1]]
+            raw = _paired_change(b, c, resamples=resamples, seed=seed)
             entry = {"baseline_median": median(b), "candidate_median": median(c),
-                     **_paired_change(b, c, resamples=resamples, seed=seed), "basis": "raw"}
-            clock = [r for r in reference if r != name]
-            if clock:
-                bn = [m[name] / sum(m[r] for r in clock) for m in maps[0]
-                      if sum(m[r] for r in clock) > 0]
-                cn = [m[name] / sum(m[r] for r in clock) for m in maps[1]
-                      if sum(m[r] for r in clock) > 0]
-                if len(bn) == len(cn) == len(b):
-                    entry.update(_paired_change(bn, cn, resamples=resamples, seed=seed),
-                                 basis="normalized_to_reference_regions")
+                     "raw": raw, "observed": raw["observed"], "basis": "raw"}
+            if usable_clock and name not in reference:
+                normalized = _paired_change([v / k for v, k in zip(b, clock[0])],
+                                            [v / k for v, k in zip(c, clock[1])],
+                                            resamples=resamples, seed=seed)
+                entry.update(normalized=normalized, observed=normalized["observed"],
+                             basis="normalized_to_reference_regions")
             if kind == "breakdown":
-                totals = [[sum(m.values()) for m in side] for side in maps]
-                entry["baseline_share"] = median(v / t for v, t in zip(b, totals[0]) if t > 0)
-                entry["candidate_share"] = median(v / t for v, t in zip(c, totals[1]) if t > 0)
+                entry["baseline_share"] = _shares(b, totals[0])
+                entry["candidate_share"] = _shares(c, totals[1])
             target[name] = entry
-        if reference:
-            ratios = []
-            for mb, mc in zip(*maps):
-                ref_b, ref_c = (sum(m[r] for r in reference) for m in (mb, mc))
-                tot_b, tot_c = sum(mb.values()), sum(mc.values())
-                if min(ref_b, ref_c, tot_b, tot_c) > 0:
-                    ratios.append((ref_c / tot_c) / (ref_b / tot_b))
-            if len(ratios) == len(maps[0]):
-                result["share_based_speedup"] = _interval_report(
-                    ratios, resamples=resamples, seed=seed)
+        if kind == "breakdown":
+            result["reference_regions"] = reference
+            if usable_clock and min(min(side) for side in totals) > 0:
+                ratios = [(kc / tc) / (kb / tb) for kb, tb, kc, tc in zip(
+                    clock[0], totals[0], clock[1], totals[1])]
+                report = {"median": median(ratios), "interval": None}
+                if len(ratios) >= MIN_PAIRS:
+                    report["interval"] = bootstrap_median_interval(
+                        ratios, resamples=resamples, seed=seed)
+                result["share_based_speedup"] = report
     if not result["regions"] and not result["counters"]:
         return {**result, "status": "absent",
                 "reason": "No breakdown or counters were present in every measured step"}
@@ -275,6 +268,10 @@ def validate_mechanism(document) -> dict:
                              "of decrease, increase or unchanged")
     if all(item["direction"] == "unchanged" for item in expect):
         raise ValueError("At least one expectation must predict a change")
+    names = [("region" if "region" in item else "counter",
+              item.get("region") or item.get("counter")) for item in expect]
+    if len(names) != len(set(names)):
+        raise ValueError("Declare each region or counter once")
     return document
 
 
@@ -287,7 +284,7 @@ def explain(mechanism: dict | None, evidence: dict) -> dict:
         kind = "region" if "region" in item else "counter"
         table = evidence.get("regions" if kind == "region" else "counters") or {}
         observed = table.get(item[kind], {}).get("observed")
-        if observed in (None, "insufficient_pairs"):
+        if observed in (None, "insufficient_pairs", "undefined"):
             missing = True
         checks.append({"kind": kind, "name": item[kind], "expected": item["direction"],
                        "observed": observed or "missing"})
diff --git a/src/inferrsi/harness/optimization.py b/src/inferrsi/harness/optimization.py
index 4a2c86548..914fcb268 100644
--- a/src/inferrsi/harness/optimization.py
+++ b/src/inferrsi/harness/optimization.py
@@ -317,11 +317,11 @@ def evaluate_round(root: Path, *, candidate: Path | None, hypothesis: str,
         fingerprint = candidates.retain(source, directory / plan["candidate_name"],
                                         plan.get("candidate_includes"))
         if final and state["best"]["kind"] == "candidate":
-            # The independent final rechecks the mechanism declared by the selected round.
-            selected = (root / state["best"]["path"]).parent / "mechanism.json"
-            if selected.is_file():
-                mechanism = json.loads(selected.read_text(encoding="utf-8"))
+            # The independent final rechecks the mechanism bound to the selected round.
+            mechanism = next((r.get("mechanism") for r in state["rounds"]
+                              if r["candidate_id"] == state["best"]["candidate_id"]), None)
         if mechanism is not None:
+            # A readable copy only; assessment uses the copy bound in state before execution.
             atomic_write_json(directory / "mechanism.json", mechanism)
         pending = {"kind": "final" if final else "round", "path": relative,
                    "candidate_id": state["best"].get("candidate_id", "baseline") if final
diff --git a/src/inferrsi/harness/optimization_session.py b/src/inferrsi/harness/optimization_session.py
index de81d669a..a37a39922 100644
--- a/src/inferrsi/harness/optimization_session.py
+++ b/src/inferrsi/harness/optimization_session.py
@@ -4,6 +4,7 @@
 projects and all subsequent operations use Harness and the retained inputs.
 """
 
+import json
 import os
 import runpy
 import shlex
@@ -50,7 +51,9 @@ def _session_assess(root, directory, plan, execution):
     if plan["task_source"] == "project":
         from inferrsi.harness.project import assess
 
-        return assess(directory, plan["evaluation_settings"], execution)
+        pending = json.loads((root / "state.json").read_text(encoding="utf-8"))["pending"]
+        return assess(directory, plan["evaluation_settings"], execution,
+                      mechanism=(pending or {}).get("mechanism"))
     return protocol(root / "frozen").assess(directory, plan["evaluation_settings"], execution)
 
 
@@ -198,7 +201,6 @@ def session_command(args, *, task_source: Path | None = None, knowledge_reader=N
             "Report summary.json and its curve. Import actual usage exports when available."
         )
         return result
-    import json
 
     plan = json.loads((root / "plan.json").read_text(encoding="utf-8"))
     if plan["runtime_identity"] != _runtime_identity():
diff --git a/src/inferrsi/harness/project.py b/src/inferrsi/harness/project.py
index f3797b8eb..2ccf148ba 100644
--- a/src/inferrsi/harness/project.py
+++ b/src/inferrsi/harness/project.py
@@ -130,17 +130,11 @@ def inspect_step(directory: Path, settings: dict, definition: dict, execution: d
             "evidence_files": evidence}
 
 
-def read_mechanism(directory: Path) -> dict | None:
-    """Read the mechanism retained with an attempt, if the round declared one."""
-    path = directory / "mechanism.json"
-    if not path.exists():
-        return None
-    if path.is_symlink():
-        raise ValueError("Symbolic mechanism declaration")
-    return validate_mechanism(json.loads(path.read_bytes()))
 
 
-def assess(directory: Path, settings: dict, execution: dict) -> dict:
+def assess(directory: Path, settings: dict, execution: dict, *,
+           mechanism: dict | None = None) -> dict:
+    """Judge one attempt; ``mechanism`` is the declaration bound before execution."""
     evidence = {}
 
     def result(status, reason, **fields):
@@ -200,13 +194,10 @@ def result(status, reason, **fields):
         else:
             verdict, reason = "FAIL", "performance_gate_not_met"
     try:
-        mechanism = read_mechanism(directory)
-        if mechanism is not None:
-            evidence["mechanism.json"] = sha256(directory / "mechanism.json")
-    except (OSError, ValueError) as error:
-        mechanism, mechanism_error = None, str(error)
-    else:
+        mechanism = None if mechanism is None else validate_mechanism(mechanism)
         mechanism_error = None
+    except ValueError as error:
+        mechanism, mechanism_error = None, str(error)
     reference = [item["region"] for item in (mechanism or {}).get("expect", [])
                  if "region" in item and item["direction"] == "unchanged"]
     observed = profile_evidence(profiles["baseline"], profiles["candidate"],
diff --git a/tests/test_harness_measurement.py b/tests/test_harness_measurement.py
index abc83648c..2bce671c6 100644
--- a/tests/test_harness_measurement.py
+++ b/tests/test_harness_measurement.py
@@ -201,7 +201,8 @@ def test_reference_regions_cancel_device_wide_jitter():
     assert evidence["status"] == "available" and evidence["reference_regions"] == ["gemm"]
     norm = evidence["regions"]["norm"]
     assert norm["basis"] == "normalized_to_reference_regions"
-    assert norm["ratio_median"] == pytest.approx(0.5) and norm["observed"] == "decrease"
+    assert norm["normalized"]["ratio_median"] == pytest.approx(0.5)
+    assert norm["observed"] == "decrease"
     assert evidence["share_based_speedup"]["median"] == pytest.approx(8 / 7)
     assert evidence["share_based_speedup"]["interval"] == pytest.approx([8 / 7, 8 / 7])
     assert evidence["counters"]["launches"]["observed"] == "decrease"
@@ -252,3 +253,40 @@ def test_explain_classifies_declared_mechanisms(expect, status):
 def test_invalid_mechanisms_are_rejected(document):
     with pytest.raises(ValueError):
         validate_mechanism(document)
+
+
+def test_zero_baseline_counter_is_undefined_not_decisive():
+    one = profile_evidence([{"counters": {"new_work": 0}}], [{"counters": {"new_work": 1}}])
+    assert one["counters"]["new_work"]["observed"] == "insufficient_pairs"
+    mixed = profile_evidence([{"counters": {"w": v}} for v in (0, 100, 100, 100, 100)],
+                             [{"counters": {"w": v}} for v in (1, 50, 50, 50, 50)])
+    assert mixed["counters"]["w"]["observed"] == "undefined"
+    mechanism = validate_mechanism({"mechanism": "m", "expect": [
+        {"counter": "w", "direction": "increase"}]})
+    assert explain(mechanism, mixed)["status"] == "unverifiable"
+
+
+def test_all_zero_candidate_breakdown_is_reported_not_raised():
+    evidence = profile_evidence([{"breakdown": {"only": 2.0}}] * 5,
+                                [{"breakdown": {"only": 0.0}}] * 5)
+    assert evidence["status"] == "available"
+    assert evidence["regions"]["only"]["candidate_share"] is None
+    assert evidence["regions"]["only"]["observed"] == "decrease"
+
+
+def test_references_are_checked_raw_never_against_each_other():
+    base = [{"breakdown": {"r1": 2.0, "r2": 2.0, "t": 2.0}}] * 5
+    cand = [{"breakdown": {"r1": 1.0, "r2": 1.0, "t": 0.5}}] * 5
+    evidence = profile_evidence(base, cand, reference_regions=["r1", "r2"])
+    assert evidence["regions"]["r1"]["basis"] == "raw"
+    assert evidence["regions"]["r1"]["observed"] == "decrease"
+    mechanism = validate_mechanism({"mechanism": "m", "expect": [
+        {"region": "r1", "direction": "unchanged"}, {"region": "r2", "direction": "unchanged"},
+        {"region": "t", "direction": "decrease"}]})
+    assert explain(mechanism, evidence)["status"] == "contradicted"
+
+
+def test_duplicate_expectations_are_rejected():
+    with pytest.raises(ValueError):
+        validate_mechanism({"mechanism": "m", "expect": [
+            {"region": "a", "direction": "unchanged"}, {"region": "a", "direction": "decrease"}]})
diff --git a/tests/test_project_explanation.py b/tests/test_project_explanation.py
index a77bd6f21..c40b582d0 100644
--- a/tests/test_project_explanation.py
+++ b/tests/test_project_explanation.py
@@ -94,15 +94,17 @@ def test_confirmed_mechanism_is_adopted_and_rechecked_at_final(project, capsys):
     assert assessment["verdict"] == "PASS" and assessment["eligible"] is True
     assert assessment["explanation"]["status"] == "explained"
     evidence = assessment["profile_evidence"]
-    assert evidence["counters"]["dram_bytes"]["ratio_median"] == pytest.approx(0.8)
+    assert evidence["counters"]["dram_bytes"]["raw"]["ratio_median"] == pytest.approx(0.8)
     # Device-wide jitter scales every region of a step. Raw cross-run ratios carry it,
     # while ratios to the declared-unchanged region and its share of the step cancel it.
     assert evidence["reference_regions"] == ["attention"]
     norm = evidence["regions"]["norm"]
     assert norm["basis"] == "normalized_to_reference_regions" and norm["observed"] == "decrease"
-    other = evidence["regions"]["other"]
+    other = evidence["regions"]["other"]["normalized"]
     assert other["ratio_median"] == pytest.approx(1.0) and other["interval"] == pytest.approx(
         [1.0, 1.0])
+    # The declared reference itself is checked raw, so device jitter keeps it unclear.
+    assert evidence["regions"]["attention"]["basis"] == "raw"
     share = evidence["share_based_speedup"]
     assert share["median"] == pytest.approx(10 / 8) and share["interval"] == pytest.approx(
         [10 / 8, 10 / 8])
@@ -173,3 +175,26 @@ def test_invalid_mechanism_is_rejected_before_execution(project, capsys):
             {"region": "norm", "direction": "faster"}]})
     _, state, _ = read_records(project[3])
     assert state["rounds"] == []
+
+
+def test_candidate_cannot_rewrite_its_declared_mechanism(project, capsys):
+    """The declaration is bound in state before execution; rewriting the copy changes nothing."""
+    repo = project[0]
+    (repo / "engine.py").write_text(
+        "import json\nfrom pathlib import Path\n"
+        "for parent in Path.cwd().parents:\n"
+        "    target = parent / 'mechanism.json'\n"
+        "    if target.is_file():\n"
+        "        target.write_text(json.dumps({'mechanism': 'forged', 'expect': [\n"
+        "            {'counter': 'dram_bytes', 'direction': 'decrease'}]}))\n"
+        "def compute(x):\n    return x * 2\n", encoding="utf-8")
+    assert start(project, capsys, "--require-explanation") == 0
+    wrong = {"mechanism": "Claims the norm got slower", "expect": [
+        {"region": "norm", "direction": "increase"}]}
+    assessment = submit(project, capsys, FUSED, wrong)
+    assert assessment["explanation"]["status"] == "contradicted"
+    assert assessment["explanation"]["mechanism"] == "Claims the norm got slower"
+    assert assessment["eligible"] is False
+    _, state, _ = read_records(project[3])
+    assert state["best"]["kind"] == "baseline"
+    assert state["rounds"][-1]["mechanism"]["mechanism"] == "Claims the norm got slower"

From f0c5d7ac59ef42b0f617ee0ad604347b5b6d657b Mon Sep 17 00:00:00 2001
From: BeatSeat <137616995+BeatSeat@users.noreply.github.com>
Date: Sun, 15 Nov 2026 21:25:14 +0800
Subject: [PATCH 4/5] harness: judge the in-memory declaration and seal its
 digest

The candidate runs with the caller's privileges and can write state.json or
the attempt copy. Pass the declaration held since submission to the assessor,
record its digest, and treat a copy that no longer matches as an integrity
failure.

Agent-assisted-by: Claude Code / claude-opus-5-5 (author)
Agent-assisted-by: Codex CLI / gpt-6.1-sol [effort=xhigh] (review)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: BeatSeat <137616995+BeatSeat@users.noreply.github.com>
---
 docs/optimization-workflow-unification.md    |  2 +-
 src/inferrsi/harness/optimization.py         | 20 ++++++--
 src/inferrsi/harness/optimization_session.py |  6 +--
 tests/test_project_explanation.py            | 52 ++++++++++++++------
 4 files changed, 58 insertions(+), 22 deletions(-)

diff --git a/docs/optimization-workflow-unification.md b/docs/optimization-workflow-unification.md
index 2a7634224..01ef4f58d 100644
--- a/docs/optimization-workflow-unification.md
+++ b/docs/optimization-workflow-unification.md
@@ -218,7 +218,7 @@ inferrsi dashboard --input <campaign-dir> --output <report.html>
 
 assessment 的 `profile_evidence` 按重复配对比较每个区域和计数，`raw` 保留原始比值。声明为 `unchanged` 的区域是本轮的假设，不是结论：其他区域另外给出除以同一步骤参照区域总和后的 `normalized` 比较，`share_based_speedup` 用参照区域在步骤总时间中的占比推算整体加速。只有当某种变化让同一步骤的所有区域按同一比例缩放时，它才会被抵消；若各区域受影响程度不同（例如降频对计算受限和访存受限的区域影响不同），抵消就不成立。参照区域本身始终只做原始比较，不会互相归一化，因此整体漂移较大时它可能显示为变化，使解释判为矛盾。计数一类的工作量指标基本不随时钟和温度变化，可以说明波动范围内的小幅改进是否可重复。方向只由配对区间判断（区间整体低于或高于 1），仓库内没有设置数值容差；基线为 0 的计数没有比值，记为 `undefined`。`profile_evidence` 只描述数据，不判断机制文字本身。
 
-`explanation.status` 取值：`explained`（每条预期都被数据确认，且至少一条是变化）、`unsupported`（预期的变化没有在数据中出现）、`contradicted`（数据与预期方向相反，或声明不变的项出现明显变化）、`unverifiable`（缺少对应区域或计数、比值无定义，或少于 5 对）、`not_declared`（本轮未声明）。机制声明在执行前写入控制状态，判定使用这份绑定的声明，attempt 目录中的 `mechanism.json` 只是可读副本；final 沿用被选中轮次绑定的机制重新核对。
+`explanation.status` 取值：`explained`（每条预期都被数据确认，且至少一条是变化）、`unsupported`（预期的变化没有在数据中出现）、`contradicted`（数据与预期方向相反，或声明不变的项出现明显变化）、`unverifiable`（缺少对应区域或计数、比值无定义，或少于 5 对）、`not_declared`（本轮未声明）。机制声明在执行前绑定：判定使用执行前保存的声明，并核对其摘要与 attempt 目录中的 `mechanism.json`；副本被改动时按完整性错误阻断 campaign。final 沿用被选中轮次绑定的机制重新核对。
 
 两个开关在 `campaign start` 时冻结，默认都不改变原有判定：
 
diff --git a/src/inferrsi/harness/optimization.py b/src/inferrsi/harness/optimization.py
index 914fcb268..b690e7a26 100644
--- a/src/inferrsi/harness/optimization.py
+++ b/src/inferrsi/harness/optimization.py
@@ -221,6 +221,15 @@ def _settle(root, plan, state, pending, execution, assess):
         _load(root)
         if candidates.identity(directory / plan["candidate_name"]) != pending["candidate_sha256"]:
             raise ValueError("Submitted candidate changed during evaluation")
+        bound = pending.get("mechanism_sha256")
+        if bound is not None and not (
+            digest(pending.get("mechanism")) == bound
+            and (directory / "mechanism.json").is_file()
+            and not (directory / "mechanism.json").is_symlink()
+            and digest(json.loads((directory / "mechanism.json").read_text(encoding="utf-8")))
+            == bound
+        ):
+            raise ValueError("Declared mechanism changed during evaluation")
     except (OSError, ValueError, TypeError, KeyError) as error:
         state["integrity_error"] = str(error)
     try:
@@ -228,7 +237,11 @@ def _settle(root, plan, state, pending, execution, assess):
             raise ValueError(state["integrity_error"])
         if execution.get("coordinator_error"):
             raise ValueError(execution["coordinator_error"])
-        assessment = assess(root, directory, plan, execution)
+        # Pass the declaration held in memory since before execution; the candidate
+        # can write files during execution, including state.json and attempt copies.
+        assessment = (assess(root, directory, plan, execution, mechanism=pending["mechanism"])
+                      if pending.get("mechanism") is not None
+                      else assess(root, directory, plan, execution))
         if not isinstance(assessment, dict) or assessment.get("status") not in (
             "completed", "candidate_failure", "infrastructure_failure"
         ):
@@ -320,14 +333,15 @@ def evaluate_round(root: Path, *, candidate: Path | None, hypothesis: str,
             # The independent final rechecks the mechanism bound to the selected round.
             mechanism = next((r.get("mechanism") for r in state["rounds"]
                               if r["candidate_id"] == state["best"]["candidate_id"]), None)
+        mechanism_sha256 = None
         if mechanism is not None:
-            # A readable copy only; assessment uses the copy bound in state before execution.
             atomic_write_json(directory / "mechanism.json", mechanism)
+            mechanism_sha256 = digest(mechanism)
         pending = {"kind": "final" if final else "round", "path": relative,
                    "candidate_id": state["best"].get("candidate_id", "baseline") if final
                    else f"candidate-{count + 1:03d}",
                    "hypothesis": hypothesis, "observations": observations,
-                   "mechanism": mechanism,
+                   "mechanism": mechanism, "mechanism_sha256": mechanism_sha256,
                    "candidate_sha256": fingerprint,
                    "contexts": list(state["contexts"]), "started_at": time.time()}
         state["pending"] = pending
diff --git a/src/inferrsi/harness/optimization_session.py b/src/inferrsi/harness/optimization_session.py
index a37a39922..a2ce3852f 100644
--- a/src/inferrsi/harness/optimization_session.py
+++ b/src/inferrsi/harness/optimization_session.py
@@ -45,15 +45,13 @@ def _session_execute(root, directory, plan, deadline):
                          None, environment=environment)
 
 
-def _session_assess(root, directory, plan, execution):
+def _session_assess(root, directory, plan, execution, mechanism=None):
     from inferrsi.harness.task_protocol import protocol
 
     if plan["task_source"] == "project":
         from inferrsi.harness.project import assess
 
-        pending = json.loads((root / "state.json").read_text(encoding="utf-8"))["pending"]
-        return assess(directory, plan["evaluation_settings"], execution,
-                      mechanism=(pending or {}).get("mechanism"))
+        return assess(directory, plan["evaluation_settings"], execution, mechanism=mechanism)
     return protocol(root / "frozen").assess(directory, plan["evaluation_settings"], execution)
 
 
diff --git a/tests/test_project_explanation.py b/tests/test_project_explanation.py
index c40b582d0..d43d0fc76 100644
--- a/tests/test_project_explanation.py
+++ b/tests/test_project_explanation.py
@@ -177,24 +177,48 @@ def test_invalid_mechanism_is_rejected_before_execution(project, capsys):
     assert state["rounds"] == []
 
 
-def test_candidate_cannot_rewrite_its_declared_mechanism(project, capsys):
-    """The declaration is bound in state before execution; rewriting the copy changes nothing."""
-    repo = project[0]
-    (repo / "engine.py").write_text(
+@pytest.mark.parametrize("target", ["state", "copy"])
+def test_rewriting_the_bound_declaration_is_detected(project, capsys, target):
+    """A candidate that rewrites state.json or the attempt copy cannot change the verdict."""
+    rewrite_state = (
+        "    state = parent / 'state.json'\n"
+        "    if state.is_file():\n"
+        "        data = json.loads(state.read_text())\n"
+        "        if data.get('pending') and data['pending'].get('mechanism'):\n"
+        "            data['pending']['mechanism'] = FORGED\n"
+        "            state.write_text(json.dumps(data))\n")
+    rewrite_copy = (
+        "    copy = parent / 'mechanism.json'\n"
+        "    if copy.is_file():\n"
+        "        copy.write_text(json.dumps(FORGED))\n")
+    (project[0] / "engine.py").write_text(
         "import json\nfrom pathlib import Path\n"
+        "FORGED = {'mechanism': 'forged', 'expect': [\n"
+        "    {'counter': 'dram_bytes', 'direction': 'decrease'}]}\n"
         "for parent in Path.cwd().parents:\n"
-        "    target = parent / 'mechanism.json'\n"
-        "    if target.is_file():\n"
-        "        target.write_text(json.dumps({'mechanism': 'forged', 'expect': [\n"
-        "            {'counter': 'dram_bytes', 'direction': 'decrease'}]}))\n"
+        + (rewrite_state if target == "state" else rewrite_copy) +
         "def compute(x):\n    return x * 2\n", encoding="utf-8")
     assert start(project, capsys, "--require-explanation") == 0
     wrong = {"mechanism": "Claims the norm got slower", "expect": [
         {"region": "norm", "direction": "increase"}]}
-    assessment = submit(project, capsys, FUSED, wrong)
-    assert assessment["explanation"]["status"] == "contradicted"
-    assert assessment["explanation"]["mechanism"] == "Claims the norm got slower"
-    assert assessment["eligible"] is False
-    _, state, _ = read_records(project[3])
+    repo, _, _, root = project
+    (repo / "config.json").write_text(json.dumps(FUSED), encoding="utf-8")
+    path = root.parent / "mechanism.json"
+    path.write_text(json.dumps(wrong), encoding="utf-8")
+    main(["campaign", "round", "--directory", str(root), "--candidate", str(repo),
+          "--hypothesis", "Synthetic candidate", "--mechanism-file", str(path)])
+    view = json.loads(capsys.readouterr().out)
+    state = json.loads((root / "state.json").read_text())
     assert state["best"]["kind"] == "baseline"
-    assert state["rounds"][-1]["mechanism"]["mechanism"] == "Claims the norm got slower"
+    if target == "copy":
+        # The sealed copy no longer matches its digest: an integrity failure blocks the
+        # campaign and nothing is adopted.
+        assert view["state"] == "blocked"
+        assert "mechanism changed" in view["integrity_error"]
+    else:
+        # state.json is rewritten wholesale at settle from the in-memory record, so the
+        # forged pending entry is discarded and the original declaration is judged.
+        assert view["state"] == "active"
+        record = state["rounds"][-1]
+        assert record["mechanism"]["mechanism"] == "Claims the norm got slower"
+        assert record["assessment"]["explanation"]["status"] == "contradicted"

From 1b2b4593cb72ef7b956f82f000313a9910cffbd7 Mon Sep 17 00:00:00 2001
From: BeatSeat <137616995+BeatSeat@users.noreply.github.com>
Date: Sun, 15 Nov 2026 21:43:37 +0800
Subject: [PATCH 5/5] harness: do not trust a declaration recovered after
 interruption

An interrupted attempt is settled by a new process that can only read files
the candidate could write. Settle it without the declaration, mark the record,
and leave require_explanation to keep the round unadopted.

Agent-assisted-by: Claude Code / claude-opus-5-5 (author)
Agent-assisted-by: Codex CLI / gpt-6.1-sol [effort=xhigh] (review)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: BeatSeat <137616995+BeatSeat@users.noreply.github.com>
---
 docs/optimization-workflow-unification.md |  2 +-
 src/inferrsi/harness/optimization.py      |  7 +++++
 tests/test_project_explanation.py         | 34 +++++++++++++++++++++++
 3 files changed, 42 insertions(+), 1 deletion(-)

diff --git a/docs/optimization-workflow-unification.md b/docs/optimization-workflow-unification.md
index 01ef4f58d..2e88ec5a7 100644
--- a/docs/optimization-workflow-unification.md
+++ b/docs/optimization-workflow-unification.md
@@ -218,7 +218,7 @@ inferrsi dashboard --input <campaign-dir> --output <report.html>
 
 assessment 的 `profile_evidence` 按重复配对比较每个区域和计数，`raw` 保留原始比值。声明为 `unchanged` 的区域是本轮的假设，不是结论：其他区域另外给出除以同一步骤参照区域总和后的 `normalized` 比较，`share_based_speedup` 用参照区域在步骤总时间中的占比推算整体加速。只有当某种变化让同一步骤的所有区域按同一比例缩放时，它才会被抵消；若各区域受影响程度不同（例如降频对计算受限和访存受限的区域影响不同），抵消就不成立。参照区域本身始终只做原始比较，不会互相归一化，因此整体漂移较大时它可能显示为变化，使解释判为矛盾。计数一类的工作量指标基本不随时钟和温度变化，可以说明波动范围内的小幅改进是否可重复。方向只由配对区间判断（区间整体低于或高于 1），仓库内没有设置数值容差；基线为 0 的计数没有比值，记为 `undefined`。`profile_evidence` 只描述数据，不判断机制文字本身。
 
-`explanation.status` 取值：`explained`（每条预期都被数据确认，且至少一条是变化）、`unsupported`（预期的变化没有在数据中出现）、`contradicted`（数据与预期方向相反，或声明不变的项出现明显变化）、`unverifiable`（缺少对应区域或计数、比值无定义，或少于 5 对）、`not_declared`（本轮未声明）。机制声明在执行前绑定：判定使用执行前保存的声明，并核对其摘要与 attempt 目录中的 `mechanism.json`；副本被改动时按完整性错误阻断 campaign。final 沿用被选中轮次绑定的机制重新核对。
+`explanation.status` 取值：`explained`（每条预期都被数据确认，且至少一条是变化）、`unsupported`（预期的变化没有在数据中出现）、`contradicted`（数据与预期方向相反，或声明不变的项出现明显变化）、`unverifiable`（缺少对应区域或计数、比值无定义，或少于 5 对）、`not_declared`（本轮未声明）。机制声明在执行前绑定：判定使用执行前保存的声明，并核对其摘要与 attempt 目录中的 `mechanism.json`；副本被改动时按完整性错误阻断 campaign。执行后被中断、经 `recover` 结算的轮次无法证明执行前绑定的是哪份声明，因此按未声明处理（记录 `mechanism_unverified_after_recovery`），在 `--require-explanation` 下不会被采纳。final 沿用被选中轮次绑定的机制重新核对。
 
 两个开关在 `campaign start` 时冻结，默认都不改变原有判定：
 
diff --git a/src/inferrsi/harness/optimization.py b/src/inferrsi/harness/optimization.py
index b690e7a26..6d98aa8e2 100644
--- a/src/inferrsi/harness/optimization.py
+++ b/src/inferrsi/harness/optimization.py
@@ -389,4 +389,11 @@ def recover(root: Path, assess) -> dict:
         if not path.is_file():
             raise ValueError("Execution record unavailable, inspect cleanup before recovery")
         execution = json.loads(path.read_text(encoding="utf-8"))
+        if pending.get("mechanism") is not None or pending.get("mechanism_sha256") is not None:
+            # The candidate ran with write access to every file that could hold the
+            # declaration after submission, so an interrupted attempt cannot prove which
+            # declaration was bound. Settle it without one: the explanation is reported as
+            # not declared and, under require_explanation, the round is not adopted.
+            pending = {**pending, "mechanism": None, "mechanism_sha256": None,
+                       "mechanism_unverified_after_recovery": True}
         return _settle(root, plan, state, pending, execution, assess)
diff --git a/tests/test_project_explanation.py b/tests/test_project_explanation.py
index d43d0fc76..fbd4624cb 100644
--- a/tests/test_project_explanation.py
+++ b/tests/test_project_explanation.py
@@ -222,3 +222,37 @@ def test_rewriting_the_bound_declaration_is_detected(project, capsys, target):
         record = state["rounds"][-1]
         assert record["mechanism"]["mechanism"] == "Claims the norm got slower"
         assert record["assessment"]["explanation"]["status"] == "contradicted"
+
+
+def test_recovered_attempt_does_not_trust_a_rewritable_declaration(project, capsys, monkeypatch):
+    """After an interruption the bound declaration is unprovable, so it is not adopted."""
+    from inferrsi.harness import optimization_session
+
+    original = optimization_session._session_execute
+
+    def forge_then_crash(root, directory, plan, deadline):
+        original(root, directory, plan, deadline)
+        forged = {"mechanism": "forged", "expect": [
+            {"counter": "dram_bytes", "direction": "decrease"}]}
+        from inferrsi.infra.artifacts import digest
+        state = json.loads((root / "state.json").read_text())
+        state["pending"].update(mechanism=forged, mechanism_sha256=digest(forged))
+        (root / "state.json").write_text(json.dumps(state))
+        (directory / "mechanism.json").write_text(json.dumps(forged))
+        raise KeyboardInterrupt("coordinator interrupted after the candidate ran")
+
+    assert start(project, capsys, "--require-explanation") == 0
+    monkeypatch.setattr(optimization_session, "_session_execute", forge_then_crash)
+    wrong = {"mechanism": "Claims the norm got slower", "expect": [
+        {"region": "norm", "direction": "increase"}]}
+    with pytest.raises(KeyboardInterrupt):
+        submit(project, capsys, FUSED, wrong)
+    monkeypatch.setattr(optimization_session, "_session_execute", original)
+    assert main(["campaign", "recover", "--directory", str(project[3])]) == 0
+    capsys.readouterr()
+    _, state, _ = read_records(project[3])
+    record = state["rounds"][-1]
+    assert record["mechanism_unverified_after_recovery"] is True
+    assert record["assessment"]["explanation"]["status"] == "not_declared"
+    assert record["assessment"]["eligible"] is False
+    assert state["best"]["kind"] == "baseline" and state["state"] == "active"