diff --git a/README.md b/README.md
index 8db242d06..3e77efbbb 100644
--- a/README.md
+++ b/README.md
@@ -44,8 +44,12 @@ including limitations from failed attempts.
 Illustrative target workflow; no measured results. [Watch the MP4](docs/assets/inferrsi-loop.mp4).
 
 Knowledge retrieval is read-only; candidate storage is optional and separately
-authorized. Automatic search and automatic knowledge promotion remain planned. Missing
-or incomplete evidence stays `INCONCLUSIVE`.
+authorized. Automatic model-backed search and automatic knowledge promotion remain
+planned. Missing or incomplete evidence stays `INCONCLUSIVE`.
+
+The **offline harness-improvement MVP** adds a separate Skill Eval path for supplied
+declarative candidates. Its fixed-output experiments validate the controller;
+real-model gains remain unmeasured. [Flow, limits and reproduction →](docs/harness-improvement.md)
 
 ## Choose your starting path
 
@@ -149,6 +153,13 @@ not implemented. Process-cleanup behavior still needs platform-specific validati
 See the [source audit and proposed remedies](docs/reviews/source-audit-ac70272.md)
 for the 34 review points, qualifications and suggested acceptance checks.
 
+Alongside the engine loop, the [offline improvement controller](docs/harness-improvement.md)
+composes the existing Skill Eval comparator and independent verifier. Its selection
+policy lives in `harness/improvement.py`; the tool adapter freezes snapshots and
+carries the total budget across probes, rerun and holdout. The engine diagrams above
+continue to describe the engine path; this additional lane does not provide engine
+performance evidence or an autonomous model-backed optimizer.
+
 <details>
 <summary><strong>See the implemented modules</strong></summary>
 
@@ -175,6 +186,7 @@ for the 34 review points, qualifications and suggested acceptance checks.
 | [Quickstart](docs/quickstart.md) | Installation, real A/B experiments, and reports |
 | [Operation guide](docs/operations.md) | Every CLI/Python operation: entry point, effect class, boundary and example |
 | [Bounded skill evaluation](docs/skill-evaluation.md) | Real-client skill/task comparisons, independent outcomes and evidence limits |
+| [Offline harness improvement](docs/harness-improvement.md) | One INV/DOM step, quality/cost selection, held-out gate and reproducible controller experiments |
 | [Qwen B300](docs/qwen-image-2.1-turbo-b300.md) | Reproduction, profiling, optimization ledger and exact image checks |
 | [Qwen Ascend](docs/qwen-image-2.1-turbo-ascend.md) | NPU results, rejected rounds and verification scope |
 | [Knowledge base](src/inferrsi/data/knowledge_packs/README.md) | Three learning tiers: contribution, dependency impact and peer techniques |
diff --git a/docs/architecture.md b/docs/architecture.md
index 759177364..60a3260af 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -67,6 +67,32 @@ Local executor commands are noninteractive and receive EOF on standard input.
 Supply their inputs through command arguments or files rather than the caller's
 input stream.
 
+### Offline harness-improvement lane
+
+The [single-step offline controller](harness-improvement.md) adds a bounded policy
+above the existing Skill Eval comparator. Supplied declarative INV/DOM edits are
+materialized as immutable Git snapshots. Development grades select a successor;
+it must run again before freezing for one independent held-out accept/reject gate.
+The controller accounts for the entire loop; candidate execution cost is reported
+separately. It reuses `run_trial` and the local executor, including their inherited
+absolute deadline, rather than creating another execution backend.
+
+```mermaid
+flowchart LR
+    I["Frozen tasks, edits, policy and total budget"] --> C["Offline improvement controller"]
+    C --> S["Existing Skill Eval comparator / run_trial"]
+    S --> V["Independent grader + verified receipts"]
+    V --> C
+    C --> H["Freeze final candidate → one holdout gate"]
+    C --> X["Development-only experience; no Store write"]
+```
+
+The feedback arrow represents development evidence only. Holdout cannot select a
+different candidate or cause another edit. Real-client execution, autonomous
+proposal generation, external sandbox enforcement and Store/CAS reuse are outside
+this MVP. [PR skill evaluation](skill-eval-pr.md) is another existing consumer of
+the comparator; the engine AB/BA path retains its independent acceptance contract.
+
 ## Directory map
 
 ![Implemented modules mapped to directories](assets/modular-directory-map-en.svg)
@@ -80,6 +106,8 @@ src/inferrsi/
   knowledge_operations.py   Normalized reads and bounded Collect (KnowledgeReader)
   knowledge_store.py        Authorized local store Create/Update (KnowledgeStore)
   harness/
+    improvement.py          Offline quality-first selection and usage accounting
+    skill_evaluation.py     Skill trial/client interface and independent grading
     runner.py               One AB/BA loop; execute/snapshot adapter hooks
     evaluation.py           Performance/quality/source-integrity adjudication
     reporting.py            HTML evidence report
diff --git a/docs/examples/harness-improvement/summary.json b/docs/examples/harness-improvement/summary.json
new file mode 100644
index 000000000..b7128d3b1
--- /dev/null
+++ b/docs/examples/harness-improvement/summary.json
@@ -0,0 +1,79 @@
+{
+  "kind": "scripted_controller_experiment",
+  "limits": {
+    "repeats": 1,
+    "max_task_calls": 24,
+    "timeout_seconds": 10,
+    "campaign_seconds": 300,
+    "token_stop_new_trials": 2400
+  },
+  "results": {
+    "bounded_loop": {
+      "selected": "composed",
+      "accepted": true,
+      "successes": 3,
+      "trials": 3,
+      "total_calls": 24,
+      "total_tokens": 2010,
+      "search_tokens": 1530,
+      "selected_holdout_tokens": 180,
+      "wall_seconds": 1.162965984083712
+    },
+    "worse": {
+      "selected": "h0",
+      "accepted": false,
+      "successes": 1,
+      "trials": 3,
+      "total_calls": 24,
+      "total_tokens": 1950,
+      "search_tokens": 1350,
+      "selected_holdout_tokens": 300,
+      "wall_seconds": 1.016869701910764
+    },
+    "compose_regression": {
+      "selected": "h0",
+      "accepted": false,
+      "successes": 1,
+      "trials": 3,
+      "total_calls": 24,
+      "total_tokens": 2130,
+      "search_tokens": 1530,
+      "selected_holdout_tokens": 300,
+      "wall_seconds": 1.035327467136085
+    },
+    "dominated": {
+      "selected": "h0",
+      "accepted": false,
+      "successes": 1,
+      "trials": 3,
+      "total_calls": 24,
+      "total_tokens": 2220,
+      "search_tokens": 1620,
+      "selected_holdout_tokens": 300,
+      "wall_seconds": 1.0334522619377822
+    },
+    "holdout_regression": {
+      "selected": "h0",
+      "accepted": false,
+      "successes": 0,
+      "trials": 3,
+      "total_calls": 24,
+      "total_tokens": 2010,
+      "search_tokens": 1530,
+      "selected_holdout_tokens": 300,
+      "wall_seconds": 1.0665274921339005
+    },
+    "h0": {
+      "successes": 1,
+      "trials": 3,
+      "total_calls": 6,
+      "total_tokens": 600
+    },
+    "repeated_h0": {
+      "successes": 4,
+      "trials": 12,
+      "total_calls": 24,
+      "total_tokens": 2400
+    }
+  }
+}
diff --git a/docs/harness-improvement.md b/docs/harness-improvement.md
new file mode 100644
index 000000000..9e72e2b80
--- /dev/null
+++ b/docs/harness-improvement.md
@@ -0,0 +1,180 @@
+# Bounded harness improvement (offline MVP)
+
+The [issue #93 proposal](https://github.com/InferMatrixLab/InferRSI/issues/93)
+now has a single-step controller for trusted fixed-output performance-audit
+experiments. It reuses the configurable Skill Eval comparison, trial executor,
+independent grader and retained-evidence checks. It does not call a proposal
+model, launch a real model, install runtime hooks or promote skills.
+
+## Harness² and HarnessPro: what is combined
+
+| Design | Source contribution | This implementation |
+| --- | --- | --- |
+| [Harness²](https://github.com/google-research/harness2/tree/1cfe254e8d25b15f37048c3730134686abe60335) | Task-local INV/DOM proposals, matched probes, composition and successor execution; experience for later tasks | At most one supplied general and one domain edit, matched development probes, conflict-aware composition and mandatory rerun |
+| [HarnessPro](https://github.com/pzs19/HarnessPro/tree/8ae622eb4a4923dbb4b515be7ded614f88fd4f38) | Trace-driven diagnosis and quality/cost selection among reusable harnesses | Caller-declared quality floor and regression limit; quality first, then known input + output tokens; full search and execution accounting |
+
+Harness² adapts during a task; HarnessPro searches for reusable operating points
+offline. This MVP deliberately uses independent development grades for selection
+(*verifier-guided search*), rather than reproducing either paper's label-free
+feedback. It selects by a fixed lexicographic quality/token rule, which chooses
+a Pareto member, without implementing population search or weighted profiles.
+Higher quality may justify higher execution cost; at equal quality the candidate
+must consume fewer known tokens. Ties retain H0. Cost here is a token proxy,
+not dollars; cached input is a subset of input and is never added twice.
+
+The last three proposal comments are the implementation requirements:
+[comparison](https://github.com/InferMatrixLab/InferRSI/issues/93#issuecomment-6364484589),
+[updated English plan](https://github.com/InferMatrixLab/InferRSI/issues/93#issuecomment-6364589831),
+and [updated Chinese plan](https://github.com/InferMatrixLab/InferRSI/issues/93#issuecomment-6364595822).
+Their historical statement that [PR #118](https://github.com/InferMatrixLab/InferRSI/pull/118)
+was a draft is superseded: its [PR evaluation integration](skill-eval-pr.md) is
+on main. That integration and this offline controller both reuse Skill Eval.
+Neither paper's benchmark numbers establish an InferRSI improvement.
+
+## Implemented flow
+
+```mermaid
+flowchart TB
+    F["Freeze H0, edits, splits, grader, policy and total limits"] --> P["INV / DOM matched development probes"]
+    P --> C["Quality then tokens; compose, choose one or keep H0"]
+    C --> R["Mandatory successor rerun; reject regressions"]
+    R --> G["Freeze candidate; one held-out comparison"]
+    G --> A["Accept bounded candidate or retain H0"]
+    R --> X["Development-only experience; no promotion"]
+    E["Engine RunSpec → AB/BA → quality/performance evidence"]
+```
+
+`src/inferrsi/harness/improvement.py` owns selection and usage accounting.
+`tools/skill_eval_improve.py` owns freezing, declarative snapshot materialization
+and orchestration through `tools/skill_eval_compare.py`. There is no second
+subprocess runner. The engine lane and its acceptance contract remain separate.
+
+1. **Freeze before probing.** `prepare(config, repo, output)` validates disjoint
+   nonempty development/holdout IDs and split labels, common model/settings,
+   default audit support, selection thresholds and the full scheduled call
+   allowance. It freezes controller/grader hashes and all candidate inputs.
+   Review task content for semantic overlap: distinct IDs alone do not establish
+   independent held-out tasks.
+2. **Materialize supplied edits.** `proposals` contains at most `inv` and `dom`
+   maps of path to replacement text. `allowed_paths` may name only existing
+   UTF-8 `.md`/`.txt` resources in the selected baseline bundle. Metadata,
+   executable entries, symlinks, new files and out-of-bundle paths are rejected.
+   Skill front matter remains fixed. Git object plumbing uses a temporary index;
+   source branches, checkouts and the caller's index are unchanged. Candidate
+   commits preserve the real H0 parent and original bytes outside the edits.
+3. **Probe and select.** Each candidate is paired with H0 under balanced AB/BA
+   ordering. Missing/inconclusive evidence, unknown token usage or inconsistent
+   baseline grades retain H0. Candidates must pass the declared quality floor,
+   meet or exceed H0 successes and obey the regression limit before token
+   comparisons. Compatible eligible edits compose; conflicting whole-file edits
+   select one candidate instead of silently merging. No partial-file synthesis.
+4. **Rerun.** The selected successor, including H0, runs again in fresh trial
+   workspaces. Composition must independently qualify here. A failed rerun
+   retains H0 without trying another candidate. Rerun grades must preserve H0's
+   development outcomes and cannot rank below an eligible probe under the frozen
+   quality/token rule; a worse composition is rejected even if it beats H0.
+   If budget/execution stops prevent
+   rerunning, the report records the missing stage and retains H0.
+5. **Freeze and gate once.** Write `final.json` and development-only
+   `experience.json` before evaluating holdout. Holdout may only accept/reject
+   the frozen candidate. Rejection retains H0; it never chooses the next edit.
+   At equal held-out quality, token cost may tie H0 but may not increase; higher
+   quality can justify higher cost under the same declared quality-first rule.
+   An exclusive results directory consumes attempts, including interrupted ones;
+   no retry or resume exists in this controller. A new directory is not proof
+   of a new holdout: the operator must enforce campaign-wide non-reuse.
+
+Every stage rechecks its authorized plan digest and reads grades through the existing
+receipt/output/trace verification. Earlier probe evidence is re-read before
+selection and all evidence is checked again before reporting acceptance.
+An invalid local artifact cannot remain a passing candidate. These hashes are
+local integrity checks, not signatures against coordinated rewriting.
+
+## Budget and report interpretation
+
+The total call allowance includes both arms of every probe, the successor rerun
+and the held-out comparison. The same absolute deadline and remaining observed
+token threshold carry across stages, including comparison preparation time.
+Unknown token usage stops further task calls and stages. The token threshold stops **new**
+trials; one running trial can cross it. It is not a provider token or monetary
+cap. Unusable evidence or exhausted limits retain H0.
+Each client execution is also bounded by the frozen `timeout_seconds`; an
+injected offline client may use a shorter limit, but cannot extend that bound.
+
+The controller keeps a matched H0 in each comparison. With two proposals this
+means three development comparisons and one held-out comparison, not the four
+executor rollouts of the paper's single-task method. This conservative reuse of
+the existing two-arm evaluator costs additional H0 executions; the report counts
+them all. It does not infer that a composition works from its component probes.
+
+`report.json` distinguishes:
+
+- `usage.search`: both arms of probes and the mandatory development rerun.
+- `usage.total`: all executed trials, including both held-out arms.
+- `usage.selected_holdout`: held-out execution of the accepted candidate, or
+  H0's matched held-out arm on rejection; unavailable if holdout never ran.
+- Input/output/cached tokens, recorded command tool calls and executor elapsed
+  seconds. Command counts cover the existing trace's `command_execution` items,
+  not every possible tool type. `wall_seconds` covers the orchestration run;
+  initial proposal materialization/preparation occurs before it.
+- Zero proposal/composition **model** calls: edits are supplied data and
+  composition is deterministic. Human authoring effort is not measured.
+  Monetary cost remains unknown, rather than fabricated as zero.
+- Verifier failures in pair grades, separate from runner/input/evidence errors.
+
+Malformed usage containers (including strings and lists) and invalid token
+fields remain unknown in both live parsing and retained-report normalization;
+raw trace bytes are preserved. The improvement loop stops before the next task
+call when input/output usage is unknown and retains H0. Observed token counters
+are only lower bounds; they do not turn missing usage into zero total cost.
+
+The saved `development` record describes the attempted successor even if the
+held-out gate later rejects it. `holdout.successes` scores the frozen candidate;
+`selected` records the ultimate decision. A rejected candidate's score must not
+be presented as the fallback H0 score.
+
+## Reproduce the offline experiment
+
+Requires Python 3.10+, dev dependencies and Git 2.45+ as in
+[configured comparisons](skill-eval-comparison.md). From a development checkout:
+
+```bash
+python -m tests.fixtures.harness_improvement_demo --output <new-offline-output>
+python -m pytest tests/test_harness_improvement.py tests/test_skill_eval_compare.py \
+  tests/test_skill_eval_pipeline.py tests/test_skill_eval_verifier.py tests/test_skill_evaluation.py
+```
+
+The demo runs a successful composition, a cheaper degraded candidate, composition
+regression and held-out regression, plus single H0 and equal-allowance repeated
+H0 controls. It writes `summary.json`, full comparison reports, frozen inputs,
+raw traces, independent grading outputs and receipts. Review raw bundles before
+sharing; subprocess records contain local paths.
+The [recorded example summary](examples/harness-improvement/summary.json) contains
+only the scripted controller outcomes and locally observed CPU wall times; full
+trial evidence is generated by the command above, not embedded in that excerpt.
+
+Answers and token/tool observations are **scripted inputs** selected from the
+actual candidate resource bytes. The unchanged audit verifier grades them in
+real CPU subprocesses. This demonstrates controller behavior, not learned skill
+efficacy, token savings from a model, transfer to new tasks, or engine speedup.
+The repeated-H0 arm has the same total call/token/time *allowance* as the loop;
+actual consumption can differ and is reported. Fixed repeat order cannot measure
+model sampling variation. Actual CPU wall time includes subprocess overhead and
+must not be interpreted as agent/model latency.
+
+For embedding, the Python API is `prepare(...)`, then
+`run_offline(repo, output, digest, client_factory=...)`. The factory is trusted
+test code and receives only stage/candidate labels. `None` and `CodexClient`
+are rejected. There is no generic real-execution CLI. A custom Python factory
+can execute arbitrary code and is not confined by this API; only trusted offline
+fixtures are supported.
+
+## Remaining proposal scope
+
+Real-model diagnosis/proposal, external process/filesystem/network isolation,
+hard provider spending limits, synthesized diagnostic tests, multi-round search,
+Store/CAS admission and library-on/off transfer evaluation remain future slices.
+`experience.json` is a development record, not Store integration or a reusable
+validated skill. The next task starts from an explicitly chosen H0. No canonical
+catalog, knowledge store, model settings, engine RunSpec or permission policy is
+automatically updated. Issue #93 acceptance remains separate from software tests.
diff --git a/docs/operations.md b/docs/operations.md
index 78a8ad6c5..7151beb1c 100644
--- a/docs/operations.md
+++ b/docs/operations.md
@@ -29,6 +29,7 @@ Effect classes used below:
 | Adapt existing Flare delivery evidence | [Offline qualification adapter](flare-qualification.md), `python -m inferrsi.model.validators.flare` | Read-only | Existing raw packets and a separate receipt; one HTTP delivery metric, no model/server/device execution |
 | Evaluate selected skill revisions | [Bounded skill-evaluation driver](skill-evaluation.md) | Experiment-executing | Explicitly authorized real Codex calls, independent task verdicts and private evidence; separate from numeric engine-performance scoring |
 | Compare skill additions or updates | [Configured skill comparison](skill-eval-comparison.md), `python -m tools.skill_eval_compare` | Read-only preparation / experiment-executing / report-generating | Fixed Git snapshots and one shared exam; explicit plan authorization for execution; JSON + Markdown regeneration makes no model calls |
+| Validate one harness-improvement step | [Offline improvement](harness-improvement.md), Python `tools.skill_eval_improve.prepare` / `run_offline` | Snapshot-writing / offline experiment-executing | Supplied declarative INV/DOM edits, shared total limits, rerun and one holdout gate; trusted fixed-output clients only; no real-model or Store execution |
 | Plan and report skill PR evaluations | [Skill PR integration](skill-eval-pr.md), `python -m tools.skill_eval_pr` | Read-only planning / authorized experiment-executing / PR-publishing | Trusted mapping, exact operator approval, consumed execution identity and credential-separated publication from retained evidence; no automatic paid trials |
 | Recover goal progress and the next proposal | [Offline goal checkpoints](goal-checkpoints.md), `python -m tools.goal_checkpoint` | Read-only | Explicit scope/criteria, pinned owner decisions and existing comparison receipts; a disposable view, never execution or acceptance authority |
 | Summarize campaign gains | `tools/campaign_summary_chart.py`; `report`/`run` figure | Figure-generating | Renders recorded values into one gain figure plus a matching table; validates the specification and computes nothing new |
diff --git a/docs/skill-eval-comparison.md b/docs/skill-eval-comparison.md
index ccff0d8ee..c83529baa 100644
--- a/docs/skill-eval-comparison.md
+++ b/docs/skill-eval-comparison.md
@@ -7,6 +7,8 @@ generates JSON and Markdown reports. It reuses the existing reader,
 
 The [PR integration](skill-eval-pr.md) adds automatic planning, explicit operator
 authorization and retained-result publication around these interfaces.
+The [offline improvement MVP](harness-improvement.md) reuses the same interfaces
+for bounded INV/DOM probes, successor rerun and a one-shot held-out gate.
 
 ```mermaid
 flowchart LR
diff --git a/src/inferrsi/harness/improvement.py b/src/inferrsi/harness/improvement.py
new file mode 100644
index 000000000..8852e2e90
--- /dev/null
+++ b/src/inferrsi/harness/improvement.py
@@ -0,0 +1,117 @@
+"""Bounded, verifier-guided harness selection; no execution or model access."""
+
+import math
+from dataclasses import dataclass
+
+from .skill_evaluation import normalize_usage
+
+
+@dataclass(frozen=True)
+class SelectionPolicy:
+    min_success_rate: float
+    max_regressions: int
+
+    def __post_init__(self):
+        if (type(self.min_success_rate) not in (int, float)
+                or not math.isfinite(self.min_success_rate)
+                or not 0 <= self.min_success_rate <= 1):
+            raise ValueError("min_success_rate must be finite and within [0, 1]")
+        if type(self.max_regressions) is not int or self.max_regressions < 0:
+            raise ValueError("max_regressions must be a nonnegative integer")
+
+
+def usage_totals(trials: list[dict]) -> dict:
+    """Unknown and invalid metrics stay unknown; cached input is not added twice."""
+    result = {"calls": len(trials), "cost": None}
+    usage = [normalize_usage(t["observations"].get("usage")) for t in trials]
+    for field in ("input_tokens", "output_tokens", "cached_input_tokens"):
+        values = [(u or {}).get(field) for u in usage]
+        result[field] = sum(values) if values and all(
+            type(value) is int and value >= 0 for value in values) else None
+    for field, values in (
+        ("tool_calls", [t["observations"].get("tool_calls") for t in trials]),
+        ("elapsed_seconds", [t.get("elapsed_seconds") for t in trials]),
+    ):
+        result[field] = sum(values) if values and all(
+            type(value) in (int, float) and math.isfinite(value) and value >= 0
+            for value in values) else None
+    tokens = [result["input_tokens"], result["output_tokens"]]
+    result["total_tokens"] = sum(tokens) if None not in tokens else None
+    return result
+
+
+def assessment(report: dict, policy: SelectionPolicy) -> dict:
+    """Only complete, receipt-verified matched evidence can qualify a successor."""
+    pairs = report["pairs"]
+    base = [p["baseline"] for p in pairs]
+    candidate = [p["candidate"] for p in pairs]
+    costs = {"baseline": usage_totals(base), "candidate": usage_totals(candidate)}
+    passed = sum(t["verdict"] == "PASS" for t in candidate)
+    base_passed = sum(t["verdict"] == "PASS" for t in base)
+    regressions = sum(p["outcome"] == "regression" for p in pairs)
+    complete = (bool(pairs) and report["status"]["execution_state"] == "completed"
+                and all(p["outcome"] not in ("missing_pair", "inconclusive") for p in pairs))
+    known = all(costs[role]["total_tokens"] is not None for role in costs)
+    reason = "eligible"
+    if not complete:
+        reason = "incomplete_evidence"
+    elif not known:
+        reason = "unknown_usage"
+    elif passed < base_passed or passed / len(pairs) < policy.min_success_rate:
+        reason = "quality_floor"
+    elif regressions > policy.max_regressions:
+        reason = "regression_limit"
+    elif passed == base_passed and costs["candidate"]["total_tokens"] >= (
+            costs["baseline"]["total_tokens"]):
+        reason = "no_improvement"
+    return {"eligible": reason == "eligible", "reason": reason, "complete": complete,
+            "successes": passed, "baseline_successes": base_passed,
+            "trials": len(pairs), "regressions": regressions, "usage": costs}
+
+
+def choose(probes: dict[str, dict], policy: SelectionPolicy, *, composable: bool) -> dict:
+    """Quality first, then tokens; compose only two individually eligible edits.
+
+    This lexicographic rule selects a Pareto member without arbitrary weights.
+    Composition is only a proposal: its mandatory rerun may reject it.
+    """
+    scores = {name: assessment(report, policy) for name, report in probes.items()}
+    if any(s["reason"] in ("incomplete_evidence", "unknown_usage") for s in scores.values()):
+        return {"successor": "h0", "reason": "uncertain_probe", "probes": scores}
+    signatures = {
+        tuple((p["case_id"], p["repeat"], p["baseline"]["verdict"]) for p in r["pairs"])
+        for r in probes.values()
+    }
+    if len(signatures) > 1:
+        return {"successor": "h0", "reason": "unstable_baseline", "probes": scores}
+    eligible = [name for name, score in scores.items() if score["eligible"]]
+    if len(eligible) == 2 and composable:
+        return {"successor": "composed", "reason": "compose_eligible", "probes": scores}
+    winner = min(eligible, key=lambda name: (
+        -scores[name]["successes"], scores[name]["usage"]["candidate"]["total_tokens"], name
+    )) if eligible else "h0"
+    return {"successor": winner, "reason": "quality_then_tokens" if eligible else "retain_h0",
+            "probes": scores}
+
+
+def holdout_accepts(report: dict, policy: SelectionPolicy) -> bool:
+    score = assessment(report, policy)
+    # Holdout checks acceptance, not a new search/ranking opportunity. Equal quality/cost is OK.
+    return score["reason"] == "eligible" or (score["reason"] == "no_improvement" and
+        score["usage"]["candidate"]["total_tokens"] == score["usage"]["baseline"]["total_tokens"])
+
+
+def rerun_assessment(report: dict, probes: dict[str, dict], policy: SelectionPolicy) -> dict:
+    score = assessment(report, policy)
+    if not score["eligible"]:
+        return score
+    if choose({**probes, "rerun": report}, policy, composable=False)["reason"] == (
+            "unstable_baseline"):
+        return {**score, "eligible": False, "reason": "unstable_baseline"}
+    rank = (-score["successes"], score["usage"]["candidate"]["total_tokens"])
+    for probe in probes.values():
+        other = assessment(probe, policy)
+        if other["eligible"] and (
+                -other["successes"], other["usage"]["candidate"]["total_tokens"]) < rank:
+            return {**score, "eligible": False, "reason": "worse_than_eligible_probe"}
+    return score
diff --git a/src/inferrsi/harness/skill_evaluation.py b/src/inferrsi/harness/skill_evaluation.py
index eb87c04fe..155844382 100644
--- a/src/inferrsi/harness/skill_evaluation.py
+++ b/src/inferrsi/harness/skill_evaluation.py
@@ -75,6 +75,19 @@ def command(self, workspace: Path, final: Path, schema: Path, prompt: str) -> tu
         return (*argv, prompt)
 
 
+def normalize_usage(usage) -> dict | None:
+    """Keep valid token observations; malformed containers/fields remain unknown.
+
+    The original provider event remains in the trace for diagnosis. This view is
+    shared by live parsing and retained-result reporting, including older records.
+    """
+    if not isinstance(usage, dict):
+        return None
+    return {key: value if type(value) is int and value >= 0 else None
+            for key, value in usage.items()
+            if key in ("input_tokens", "output_tokens", "cached_input_tokens")}
+
+
 def trace_metrics(path: Path) -> dict:
     # Missing usage stays unknown; cached_input is a subset of input.
     usage, commands, errors, completed = None, [], [], 0
@@ -82,7 +95,7 @@ def trace_metrics(path: Path) -> dict:
         event = json.loads(line)
         if event["type"] == "turn.completed":
             completed += 1
-            usage = event.get("usage")
+            usage = normalize_usage(event.get("usage"))
         elif event["type"] in ("turn.failed", "error"):
             errors.append(event)
         elif event["type"] == "item.completed":
@@ -115,7 +128,7 @@ def summarize(trials: list[dict]) -> dict:
     groups = {}
     for condition in sorted({t["condition"] for t in trials}):
         selected = [t for t in trials if t["condition"] == condition]
-        usage = [t["observations"]["usage"] for t in selected]
+        usage = [normalize_usage(t["observations"]["usage"]) for t in selected]
         groups[condition] = {
             "trials": len(selected),
             "verdicts": {
@@ -172,17 +185,23 @@ def run_trial(
     verifier: Path,
     prompt: str,
     deadline: float,
+    *,
+    timeout_seconds: float | None = None,
 ) -> dict:
     trial_dir.mkdir(parents=True)
     final = trial_dir / "answer.json"
     before = {
         p.relative_to(workspace).as_posix(): sha256(p) for p in workspace.rglob("*") if p.is_file()
     }
+    command = client.command(workspace, final, schema, prompt)
+    # An injected client may tighten the frozen limit, but cannot extend it.
+    timeout = client.timeout_seconds if timeout_seconds is None else min(
+        client.timeout_seconds, timeout_seconds)
     execution = execute_command(
-        client.command(workspace, final, schema, prompt),
+        command,
         workspace,
         trial_dir / "agent",
-        min(deadline, time.monotonic() + client.timeout_seconds),
+        min(deadline, time.monotonic() + timeout),
     )
     atomic_write_json(trial_dir / "execution.json", execution)
     try:
diff --git a/tests/fixtures/harness_improvement_client.py b/tests/fixtures/harness_improvement_client.py
new file mode 100644
index 000000000..57f8cb8b4
--- /dev/null
+++ b/tests/fixtures/harness_improvement_client.py
@@ -0,0 +1,26 @@
+"""Scripted answers/usage for controller tests, never model or efficacy measurements."""
+
+import json
+import sys
+from pathlib import Path
+
+answer_path, final_path, correct, tokens, fault = sys.argv[1:]
+answer = json.loads(Path(answer_path).read_text(encoding="utf-8"))
+if correct == "no":
+    answer["claim_scope"] = "production"
+Path(final_path).write_text(json.dumps(answer), encoding="utf-8")
+if fault == "hidden_write":
+    Path("task.json").write_text("{}", encoding="utf-8")
+for i in range(3 if int(tokens) == 100 else 1):
+    print(json.dumps({"type": "item.completed", "item": {
+        "type": "command_execution", "command": "scripted-audit", "exit_code": 0}}))
+event = {"type": "turn.completed"}
+if fault != "unknown_usage":
+    event["usage"] = {"input_tokens": int(tokens) - 20, "output_tokens": 20,
+                      "cached_input_tokens": 10}
+if fault == "string_usage":
+    event["usage"] = "malformed"
+elif fault == "list_usage":
+    event["usage"] = [1]
+print(json.dumps(event))
+raise SystemExit(2 if fault == "runner_error" else 0)
diff --git a/tests/fixtures/harness_improvement_demo.py b/tests/fixtures/harness_improvement_demo.py
new file mode 100644
index 000000000..49b3b8cd5
--- /dev/null
+++ b/tests/fixtures/harness_improvement_demo.py
@@ -0,0 +1,134 @@
+"""CPU-only reproducible single-step experiment; all answer/usage choices are scripted."""
+
+import argparse
+import copy
+import json
+import sys
+import tempfile
+from pathlib import Path
+
+from inferrsi.infra.artifacts import write_json
+from tests.fixtures.skill_eval_compare_support import DATA, ROOT, source_repo
+from tools import skill_eval_compare as compare
+from tools import skill_eval_improve as improve
+
+
+def configuration(root):
+    repo, baseline, _ = source_repo(root)
+    tasks = {}
+    answers = {}
+    # Different audit cases in development and holdout, with independent fixed answers.
+    for split, indices in (("development", (0, 1, 2)), ("holdout", (3, 4, 5))):
+        tasks[split] = []
+        for category, index in zip(("control", "general", "domain"), indices):
+            case = DATA["cases"][index]
+            task = copy.deepcopy(DATA["task"])
+            task.update(id=f"{split}-{category}", split=split)
+            for key, value in case["changes"].items():
+                target = task
+                fields = key.split(".")
+                for field in fields[:-1]:
+                    target = target[field]
+                target[fields[-1]] = value
+            tasks[split].append(task)
+            answers[task["id"]] = case["answer"]
+    write_json(root / "answers.json", answers)
+    config = {"skill": "selected", "skill_root": "library", "baseline": baseline,
+              "allowed_paths": ["selected/SKILL.md", "selected/resource.txt"],
+              "proposals": {
+                  "inv": {"selected/SKILL.md": "---\nname: selected\n"
+                          "description: offline snapshot fixture\n---\nINV enabled\n"},
+                  "dom": {"selected/resource.txt": "DOM enabled\n"}},
+              **tasks, "model": {"name": "offline-fixture", "effort": "test"},
+              "selection": {"min_success_rate": 0.5, "max_regressions": 0},
+              "limits": {"repeats": 1, "max_task_calls": 24, "timeout_seconds": 10,
+                         "campaign_seconds": 300, "token_stop_new_trials": 2400}}
+    return config, repo
+
+
+class ScriptedClient:
+    timeout_seconds = 10
+
+    def __init__(self, root, phase, fault=None):
+        self.root, self.phase, self.fault = root, phase, fault
+        self.calls = []
+
+    def command(self, workspace, final, schema, prompt):
+        task = compare.read_json(workspace / "task.json")
+        inv = "INV enabled" in (workspace / "knowledge/selected/SKILL.md").read_text()
+        dom = "DOM enabled" in (workspace / "knowledge/selected/resource.txt").read_text()
+        category = task["id"].split("-")[-1]
+        correct = category == "control" or (category == "general" and inv) or (
+            category == "domain" and dom)
+        tokens = 60 if inv and dom else 80 if inv else 70 if dom else 100
+        fault = "none"
+        if self.fault == "worse" and (inv or dom):
+            correct, tokens = False, 25
+        elif self.fault == "compose_regression" and inv and dom:
+            correct = False
+        elif self.fault == "dominated" and inv and dom:
+            correct, tokens = category != "domain", 90
+        elif self.fault == "holdout_regression" and self.phase == "holdout" and (inv or dom):
+            correct = False
+        elif self.fault in ("unknown_usage", "string_usage", "list_usage",
+                            "runner_error", "hidden_write") and inv:
+            fault = self.fault
+        answers = compare.read_json(self.root / "answers.json")
+        answer = final.parent.parent / (final.parent.name + "-fixed.json")
+        write_json(answer, answers[task["id"]])
+        self.calls.append({"task_id": task["id"], "phase": self.phase,
+                           "files": sorted(p.relative_to(workspace).as_posix()
+                                           for p in workspace.rglob("*") if p.is_file())})
+        return (sys.executable, str(ROOT / "tests/fixtures/harness_improvement_client.py"),
+                str(answer), str(final), "yes" if correct else "no", str(tokens), fault)
+
+
+def run_demo(output):
+    output.mkdir(parents=True, exist_ok=False)
+    with tempfile.TemporaryDirectory(prefix="improvement-demo-") as temporary:
+        root = Path(temporary)
+        config, repo = configuration(root)
+        results = {}
+        for scenario, fault in (("bounded_loop", None), ("worse", "worse"),
+                                ("compose_regression", "compose_regression"),
+                                ("dominated", "dominated"),
+                                ("holdout_regression", "holdout_regression")):
+            directory = output / scenario
+            digest = improve.prepare(config, repo, directory)
+            results[scenario] = improve.run_offline(
+                repo, directory, digest, client_factory=lambda phase, candidate, fault=fault:
+                ScriptedClient(root, phase, fault))
+        for name, repeats in (("h0", 1), ("repeated_h0", 4)):
+            comparison = {key: config[key] for key in ("skill", "skill_root", "model")}
+            comparison.update(mode="update", cases=f"{name}-tasks.json", snapshots={
+                role: {"git_revision": config["baseline"]} for role in compare.ROLES},
+                limits={**config["limits"], "repeats": repeats})
+            write_json(output / f"{name}-tasks.json", config["holdout"])
+            write_json(output / f"{name}-config.json", comparison)
+            digest = compare.prepare(output / f"{name}-config.json", repo, output / name)
+            results[name] = compare.execute(output / name, digest,
+                                           client=ScriptedClient(root, name))
+        summary = {name: {
+            "selected": r["selected"], "accepted": r["accepted"],
+            "successes": r["holdout"]["successes"], "trials": r["holdout"]["trials"],
+            "total_calls": r["usage"]["total"]["calls"],
+            "total_tokens": r["usage"]["total"]["total_tokens"],
+            "search_tokens": r["usage"]["search"]["total_tokens"],
+            "selected_holdout_tokens": r["usage"]["selected_holdout"]["total_tokens"],
+            "wall_seconds": r["wall_seconds"],
+        } for name, r in results.items() if name not in ("h0", "repeated_h0")}
+        for name in ("h0", "repeated_h0"):
+            r = results[name]
+            summary[name] = {"successes": r["conditions"]["candidate"]["verdicts"]["PASS"],
+                             "trials": r["conditions"]["candidate"]["trials"],
+                             "total_calls": r["status"]["recorded_trials"],
+                             "total_tokens": r["status"]["observed_tokens"]}
+        write_json(output / "summary.json", {"kind": "scripted_controller_experiment",
+                                              "limits": config["limits"], "results": summary})
+        return summary
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--output", type=Path, required=True)
+    print(json.dumps(run_demo(parser.parse_args().output.resolve()), indent=2))
diff --git a/tests/test_harness_improvement.py b/tests/test_harness_improvement.py
new file mode 100644
index 000000000..ea7dd52ef
--- /dev/null
+++ b/tests/test_harness_improvement.py
@@ -0,0 +1,354 @@
+import copy
+import subprocess
+import sys
+from unittest.mock import patch
+
+import pytest
+
+from inferrsi.harness.improvement import (
+    SelectionPolicy,
+    assessment,
+    choose,
+    holdout_accepts,
+    usage_totals,
+)
+from inferrsi.harness.skill_evaluation import CodexClient
+from tests.fixtures.harness_improvement_demo import ScriptedClient, configuration
+from tests.fixtures.skill_eval_compare_support import git
+from tools import skill_eval_compare as compare
+from tools import skill_eval_improve as improve
+
+
+def fixture(tmp_path):
+    try:
+        compare.require_git_version(subprocess.check_output(("git", "--version"), text=True))
+    except ValueError as error:
+        pytest.skip(str(error))
+    return configuration(tmp_path)
+
+
+def run(tmp_path, fault=None, change=None):
+    config, repo = fixture(tmp_path)
+    if change:
+        change(config)
+    output = tmp_path / "run"
+    digest = improve.prepare(config, repo, output)
+    clients = []
+
+    def factory(phase, candidate):
+        client = ScriptedClient(tmp_path, phase, fault)
+        clients.append(client)
+        return client
+
+    result = improve.run_offline(repo, output, digest, client_factory=factory)
+    return result, clients, output, repo, digest
+
+
+def test_successful_loop_materializes_edits_and_accounts_for_all_stages(tmp_path):
+    result, clients, output, repo, digest = run(tmp_path)
+    assert result["selected"] == "composed" and result["accepted"]
+    assert list(result["phases"]) == ["inv", "dom", "rerun", "holdout"]
+    assert result["holdout"]["successes"] == 3
+    assert result["holdout"]["baseline_successes"] == 1
+    assert result["usage"]["total"]["total_tokens"] == 2010
+    assert result["usage"]["total"]["calls"] == 24
+    assert result["usage"]["search"]["total_tokens"] == 1530
+    assert result["usage"]["selected_holdout"]["total_tokens"] == 180
+    assert result["usage"]["total"]["cached_input_tokens"] == 240
+    assert result["usage"]["total"]["tool_calls"] == 48
+    assert result["usage"]["total"]["cost"] is None
+    assert sum(len(client.calls) for client in clients) == 24
+    assert not git(repo, "status", "--porcelain")
+    plan = improve.load_freeze(output, digest)
+    changed = git(repo, "diff", "--name-only", plan["revisions"]["h0"],
+                  plan["revisions"]["composed"]).splitlines()
+    assert changed == ["library/selected/SKILL.md", "library/selected/resource.txt"]
+    # Independently checked answers, not the fixture's proposed correctness flag, drive verdicts.
+    report = compare.read_report(output / "results/holdout")
+    assert all(p["candidate"]["checks"]["passed"] for p in report["pairs"])
+    assert "holdout" not in (output / "results/experience.json").read_text()
+    assert all(not any("answers" in name or "verify.py" in name for name in call["files"])
+               for client in clients for call in client.calls)
+    assert all(call["task_id"].startswith("development-") for client in clients[:3]
+               for call in client.calls)
+    assert all(call["task_id"].startswith("holdout-") for call in clients[-1].calls)
+    with pytest.raises(FileExistsError):
+        improve.run_offline(repo, output, digest,
+                            client_factory=lambda *args: pytest.fail("must not repeat holdout"))
+
+
+@pytest.mark.parametrize("fault,successor,frozen", [
+    ("worse", "h0", "h0"),
+    ("compose_regression", "composed", "h0"),
+    ("dominated", "composed", "h0"),
+    ("holdout_regression", "composed", "composed"),
+])
+def test_degraded_candidates_rerun_and_holdout_retain_baseline(tmp_path, fault, successor, frozen):
+    result, clients, output, _, _ = run(tmp_path, fault)
+    assert result["selected"] == "h0" and not result["accepted"]
+    assert result["development"]["attempted_successor"] == successor
+    assert result["frozen_candidate"] == frozen
+    assert [client.phase for client in clients] == ["inv", "dom", "rerun", "holdout"]
+    assert result["usage"]["selected_holdout"]["total_tokens"] == 300
+    if fault == "holdout_regression":
+        assert result["holdout"]["successes"] == 0
+        # No holdout result contaminates improvement experience or causes another search.
+        experience = compare.read_json(output / "results/experience.json")
+        assert experience["frozen_candidate"] == "composed"
+        assert experience["rerun"]["successes"] == 3
+    if fault == "dominated":
+        assert result["development"]["rerun"]["reason"] == "worse_than_eligible_probe"
+
+
+@pytest.mark.parametrize("fault", ["unknown_usage", "runner_error", "hidden_write"])
+def test_unknown_usage_runner_failures_and_input_tampering_cannot_win(tmp_path, fault):
+    result, _, _, _, _ = run(tmp_path, fault)
+    assert not result["accepted"] and result["selected"] == "h0"
+    assert result["decision"]["successor"] == "h0"
+    if fault == "unknown_usage":
+        assert result["usage"]["total"]["total_tokens"] is None
+        assert result["stop_reason"] == "unknown_usage"
+    if fault == "runner_error":
+        assert result["stop_reason"] == "incomplete_execution"
+        assert "rerun" not in result["phases"]  # Cannot invent an execution after stopping.
+
+
+def test_conflicting_edits_select_one_and_still_rerun(tmp_path):
+    def conflict(config):
+        config["proposals"]["dom"] = {"selected/SKILL.md":
+            config["proposals"]["inv"]["selected/SKILL.md"] + "A different INV policy\n"}
+
+    result, _, _, _, _ = run(tmp_path, change=conflict)
+    assert result["development"]["conflicts"] == ["selected/SKILL.md"]
+    assert result["decision"]["successor"] in ("inv", "dom")
+    assert result["phases"]["rerun"]["candidate"] == result["decision"]["successor"]
+
+
+def test_no_proposals_still_reruns_h0(tmp_path):
+    result, clients, _, _, _ = run(tmp_path, change=lambda c: c.update(proposals={}))
+    assert result["selected"] == "h0"
+    assert [client.phase for client in clients] == ["rerun", "holdout"]
+
+
+def test_total_token_stop_applies_across_comparisons(tmp_path):
+    result, _, _, _, _ = run(
+        tmp_path, change=lambda c: c["limits"].update(token_stop_new_trials=600))
+    assert result["selected"] == "h0" and result["stop_reason"] == "incomplete_execution"
+    # First probe costs 540, second may cross 600 once, then no new trials/stages.
+    assert result["usage"]["total"]["calls"] == 7
+    assert result["usage"]["total"]["total_tokens"] == 640
+    assert "rerun" not in result["phases"] and "holdout" not in result["phases"]
+
+
+def test_expired_absolute_deadline_prevents_phase_calls(tmp_path):
+    config, repo = fixture(tmp_path)
+    output = tmp_path / "run"
+    digest = improve.prepare(config, repo, output)
+
+    def factory(phase, candidate):
+        # Exercise shared executor's inherited absolute deadline, with no blocking wait.
+        class LateClient(ScriptedClient):
+            def command(self, *args):
+                pytest.fail("expired executor must not call client")
+        return LateClient(tmp_path, phase)
+
+    original = compare.execute
+
+    def expired(*args, **kwargs):
+        kwargs["campaign_deadline"] = 0
+        return original(*args, **kwargs)
+
+    with patch.object(compare, "execute", expired):
+        result = improve.run_offline(repo, output, digest, client_factory=factory)
+    assert result["selected"] == "h0" and result["usage"]["total"]["calls"] == 0
+
+
+@pytest.mark.parametrize("configured_timeout,client_timeout", [(1, 10), (10, 0.1)])
+def test_frozen_and_shorter_client_timeouts_both_bound_offline_calls(
+        tmp_path, configured_timeout, client_timeout):
+    config, repo = fixture(tmp_path)
+    config["limits"]["timeout_seconds"] = configured_timeout
+    output = tmp_path / "run"
+    digest = improve.prepare(config, repo, output)
+    calls = []
+
+    class DelayedClient:
+        timeout_seconds = client_timeout
+
+        def command(self, workspace, final, schema, prompt):
+            calls.append(final)
+            return (sys.executable, "-c", "import time; time.sleep(1.3)")
+
+    result = improve.run_offline(repo, output, digest,
+                                client_factory=lambda *args: DelayedClient())
+    assert len(calls) == 1
+    record = compare.read_json(calls[0].parent / "result.json")
+    assert record["execution_state"] == "timeout"
+    report = compare.read_report(output / "results/inv")
+    assert report["status"]["stop_reason"] == "client_timeout"
+    assert result["selected"] == "h0" and not result["accepted"]
+    assert list(result["phases"]) == ["inv"]
+
+
+@pytest.mark.parametrize("mutation", ["overlap", "mixed_split", "calls", "threshold", "extra"])
+def test_reject_invalid_control_plan_before_execution(tmp_path, mutation):
+    config, repo = fixture(tmp_path)
+    if mutation == "overlap":
+        config["holdout"][0]["id"] = config["development"][0]["id"]
+    elif mutation == "mixed_split":
+        config["development"][0]["split"] = "holdout"
+    elif mutation == "calls":
+        config["limits"]["max_task_calls"] = 12
+    elif mutation == "threshold":
+        config["selection"]["min_success_rate"] = float("nan")
+    else:
+        config["proposals"]["third"] = config["proposals"]["inv"]
+    with pytest.raises(ValueError):
+        improve.prepare(config, repo, tmp_path / "run")
+    assert not (tmp_path / "run").exists()
+
+
+@pytest.mark.parametrize("path", ["../answers.json", "selected/metadata.json", "exam.txt",
+                                  "selected/tool.py", "/task.json"])
+def test_proposals_cannot_change_hidden_answers_metadata_or_executables(tmp_path, path):
+    config, repo = fixture(tmp_path)
+    config["allowed_paths"].append(path)
+    config["proposals"]["inv"] = {path: "untrusted"}
+    with pytest.raises(ValueError):
+        improve.prepare(config, repo, tmp_path / "run")
+
+
+@pytest.mark.parametrize("client", [None, CodexClient("unused", "test", 1)])
+def test_plan_mutation_and_real_client_rejected(tmp_path, client):
+    config, repo = fixture(tmp_path)
+    output = tmp_path / "run"
+    digest = improve.prepare(config, repo, output)
+    original = (output / "freeze.json").read_bytes()
+    (output / "freeze.json").write_bytes(original + b" ")
+    with pytest.raises(ValueError, match="plan changed"):
+        improve.run_offline(repo, output, digest, client_factory=None)
+    (output / "freeze.json").write_bytes(original)
+    with pytest.raises(ValueError, match="offline client"):
+        improve.run_offline(repo, output, digest, client_factory=lambda *args: client)
+
+
+@pytest.mark.parametrize("artifact", ["answer.json", "result.json", "agent/stdout.log"])
+def test_late_probe_artifact_tampering_rolls_back(tmp_path, artifact):
+    config, repo = fixture(tmp_path)
+    output = tmp_path / "run"
+    digest = improve.prepare(config, repo, output)
+
+    def factory(phase, candidate):
+        if phase == "dom":
+            path = output / "results/inv/results/development-control-candidate-1" / artifact
+            path.write_bytes(path.read_bytes() + b" ")
+        return ScriptedClient(tmp_path, phase)
+
+    result = improve.run_offline(repo, output, digest, client_factory=factory)
+    assert result["selected"] == "h0" and result["decision"]["successor"] == "h0"
+    assert result["stop_reason"] == "invalid_retained_evidence"
+
+
+def test_phase_schedule_cannot_hide_failed_pair_after_authorization(tmp_path):
+    config, repo = fixture(tmp_path)
+    output = tmp_path / "run"
+    digest = improve.prepare(config, repo, output)
+
+    def factory(phase, candidate):
+        if phase == "holdout":
+            path = output / "results/inv/plan/freeze.json"
+            plan = compare.read_json(path)
+            plan["schedule"] = [e for e in plan["schedule"]
+                                if e["task_id"] != "development-domain"]
+            compare.write_json(path, plan)
+        return ScriptedClient(tmp_path, phase)
+
+    with pytest.raises(ValueError, match="plan changed after authorization"):
+        improve.run_offline(repo, output, digest, client_factory=factory)
+    assert not (output / "report.json").exists()
+
+
+def test_rerun_with_unstable_baseline_is_rejected(tmp_path):
+    from inferrsi.harness.improvement import rerun_assessment
+
+    _, _, output, _, _ = run(tmp_path)
+    probe = compare.read_report(output / "results/inv")
+    rerun = compare.read_report(output / "results/rerun")
+    rerun["pairs"][1]["baseline"]["verdict"] = "PASS"
+    result = rerun_assessment(rerun, {"inv": probe}, SelectionPolicy(0.5, 0))
+    assert not result["eligible"] and result["reason"] == "unstable_baseline"
+
+
+@pytest.mark.parametrize("newline,description", [("\r\n", "fixture"),
+                                                ("\n", "fixture --- embedded delimiter")])
+def test_metadata_prefix_cannot_be_changed_via_crlf_or_embedded_delimiter(
+        tmp_path, newline, description):
+    config, repo = fixture(tmp_path)
+    path = repo / "library/selected/SKILL.md"
+    source = newline.join(("---", "name: selected", f"description: {description}",
+                           "license: UNKNOWN", "---", "Old body", ""))
+    path.write_bytes(source.encode())
+    git(repo, "add", "library/selected/SKILL.md")
+    git(repo, "commit", "--quiet", "-m", "Fixture metadata prefix")
+    config["baseline"] = git(repo, "rev-parse", "HEAD")
+    # Body-only changes remain valid with both supported front-matter encodings.
+    config["proposals"]["inv"] = {"selected/SKILL.md": source.replace("Old body", "INV enabled")}
+    improve.prepare(config, repo, tmp_path / "valid")
+    config["proposals"]["inv"] = {"selected/SKILL.md": source.replace("UNKNOWN", "MIT")}
+    with pytest.raises(ValueError, match="metadata must remain unchanged"):
+        improve.prepare(config, repo, tmp_path / "invalid")
+
+
+def test_policy_unknown_invalid_usage_and_quality_before_cost():
+    trials = [{"observations": {"usage": {"input_tokens": 10, "output_tokens": 2,
+                                          "cached_input_tokens": 4}, "tool_calls": 1},
+               "elapsed_seconds": 0.1}]
+    assert usage_totals(trials)["total_tokens"] == 12
+    for value in (-1, True, float("nan"), None):
+        bad = copy.deepcopy(trials)
+        bad[0]["observations"]["usage"]["input_tokens"] = value
+        assert usage_totals(bad)["total_tokens"] is None
+
+
+def test_cost_selection_ignores_unknown_and_preserves_quality_floor(tmp_path):
+    _, _, output, _, _ = run(tmp_path)
+    report = compare.read_report(output / "results/inv")
+    policy = SelectionPolicy(min_success_rate=0.5, max_regressions=0)
+    assert assessment(report, policy)["eligible"]
+    # More successes can justify greater cost; same quality needs lower known cost.
+    assert choose({"inv": report}, policy, composable=False)["successor"] == "inv"
+    report["pairs"][0]["candidate"]["observations"]["usage"] = None
+    assert choose({"inv": report}, policy, composable=False)["successor"] == "h0"
+
+
+@pytest.mark.parametrize("usage", ["malformed", [1]])
+def test_malformed_usage_container_stays_unknown(usage):
+    trial = {"observations": {"usage": usage}, "elapsed_seconds": 1}
+    totals = usage_totals([trial])
+    assert totals["total_tokens"] is None
+    assert totals["input_tokens"] is None and totals["output_tokens"] is None
+
+
+@pytest.mark.parametrize("fault", ["string_usage", "list_usage"])
+def test_malformed_trace_usage_stops_calls_and_retains_h0(tmp_path, fault):
+    result, clients, output, _, _ = run(tmp_path, fault)
+    assert result["selected"] == "h0" and not result["accepted"]
+    assert result["usage"]["total"]["total_tokens"] is None
+    assert result["stop_reason"] == "unknown_usage"
+    assert sum(len(client.calls) for client in clients) == 2
+    assert list(result["phases"]) == ["inv"]
+    # Reporting retained bytes must also work without another client invocation.
+    report = compare.read_report(output / "results/inv")
+    trial = report["pairs"][0]["candidate"]
+    assert trial["observations"]["usage"] is None
+    assert report["conditions"]["candidate"]["input_tokens"] is None
+    assert report["conditions"]["candidate"]["output_tokens"] is None
+
+
+def test_holdout_allows_exact_tie_but_rejects_equal_quality_at_higher_cost(tmp_path):
+    _, _, output, _, _ = run(tmp_path, "worse")
+    report = compare.read_report(output / "results/holdout")  # H0 vs H0.
+    policy = SelectionPolicy(0, 0)
+    assert holdout_accepts(report, policy)
+    report["pairs"][0]["candidate"]["observations"]["usage"]["input_tokens"] += 1
+    assert not holdout_accepts(report, policy)
diff --git a/tests/test_skill_eval_compare.py b/tests/test_skill_eval_compare.py
index 2f55f5304..a331bd5f4 100644
--- a/tests/test_skill_eval_compare.py
+++ b/tests/test_skill_eval_compare.py
@@ -125,6 +125,29 @@ def test_budget_stops_leave_missing_sides_and_unknown_usage(tmp_path):
     assert result["pairs"][1]["baseline"]["missing"] is True
 
 
+@pytest.mark.parametrize("usage", ["malformed", [1]])
+def test_legacy_malformed_usage_report_stays_unknown_without_rewriting_evidence(tmp_path, usage):
+    config, repo = configuration(tmp_path, ids=("one",))
+    output = tmp_path / "run"
+    authorization = compare.prepare(config, repo, output)
+    client = FixedOutputClient(tmp_path)
+    original = compare.execute(output, authorization, client=client)
+    path = output / original["pairs"][0]["candidate"]["evidence"]["result.json"]
+    record = compare.read_json(path)
+    # Model a legacy controller that retained raw provider usage with a valid receipt.
+    record["observations"]["usage"] = usage
+    write_json(path, record)
+    receipt = path.parent / "receipt.json"
+    write_json(receipt, {"result_sha256": sha256(path)})
+    saved = path.read_bytes(), receipt.read_bytes()
+    result = compare.report(output)
+    assert result["pairs"][0]["candidate"]["observations"]["usage"] is None
+    assert result["conditions"]["candidate"]["input_tokens"] is None
+    assert result["conditions"]["candidate"]["output_tokens"] is None
+    assert len(client.calls) == 2
+    assert saved == (path.read_bytes(), receipt.read_bytes())
+
+
 def test_run_status_receipt_rejects_changed_completion_and_budget_fields(tmp_path):
     config, repo = configuration(tmp_path, ids=("one",), token_stop_new_trials=1)
     output = tmp_path / "run"
diff --git a/tests/test_skill_evaluation.py b/tests/test_skill_evaluation.py
index c8edafa60..88ce6f026 100644
--- a/tests/test_skill_evaluation.py
+++ b/tests/test_skill_evaluation.py
@@ -47,6 +47,33 @@ def test_trace_retains_unknown_usage_and_errors(tmp_path):
     assert trace["completed_turns"] == 0 and trace["errors"]
 
 
+@pytest.mark.parametrize("usage", ["malformed", [1]])
+def test_malformed_usage_is_unknown_in_trace_and_legacy_summary(tmp_path, usage):
+    path = tmp_path / "events.jsonl"
+    raw = json.dumps({"type": "turn.completed", "usage": usage}).encode() + b"\n"
+    path.write_bytes(raw)
+    trace = trace_metrics(path)
+    assert trace["usage"] is None and trace["completed_turns"] == 1
+    assert path.read_bytes() == raw
+    legacy = {"task_id": "one", "repeat": 1, "condition": "N", "verdict": "PASS",
+              "elapsed_seconds": 1, "observations": {"usage": usage}}
+    group = summarize([legacy])["conditions"]["N"]
+    assert group["input_tokens"] is None and group["output_tokens"] is None
+    # This is only the observed lower bound, never a claim of complete zero cost.
+    assert load_tool("skill_eval_pilot").observed_tokens(usage) == 0
+    assert legacy["observations"]["usage"] == usage
+
+
+@pytest.mark.parametrize("invalid", ["17", [17], True, -17, 1.5])
+def test_invalid_token_fields_keep_valid_partial_observations(tmp_path, invalid):
+    path = tmp_path / "events.jsonl"
+    path.write_text(json.dumps({"type": "turn.completed", "usage": {
+        "input_tokens": invalid, "output_tokens": 17, "cached_input_tokens": invalid}}))
+    usage = trace_metrics(path)["usage"]
+    assert usage == {"input_tokens": None, "output_tokens": 17, "cached_input_tokens": None}
+    assert load_tool("skill_eval_pilot").observed_tokens(usage) == 17
+
+
 @pytest.mark.parametrize("status", ["timeout", "error", "cancelled", "failed"])
 def test_incomplete_execution_never_passes(status):
     assert (
diff --git a/tools/skill_eval_compare.py b/tools/skill_eval_compare.py
index 1f924d034..904f45983 100644
--- a/tools/skill_eval_compare.py
+++ b/tools/skill_eval_compare.py
@@ -9,7 +9,13 @@
 import time
 from pathlib import Path
 
-from inferrsi.harness.skill_evaluation import CodexClient, export_bundle, run_trial, summarize
+from inferrsi.harness.skill_evaluation import (
+    CodexClient,
+    export_bundle,
+    normalize_usage,
+    run_trial,
+    summarize,
+)
 from inferrsi.infra.artifacts import atomic_write_json, sha256, write_json
 from inferrsi.knowledge_operations import KnowledgeReader, safe_path
 from tools.skill_eval_pilot import PROMPT, SCHEMA, observed_tokens
@@ -384,7 +390,9 @@ def receipt_error(raw_record: bytes, receipt_path: Path, label: str) -> str | No
     return None
 
 
-def execute(output: Path, authorization: str, *, client=None) -> dict:
+def execute(output: Path, authorization: str, *, client=None,
+            campaign_deadline: float | None = None,
+            stop_on_unknown_usage: bool = False) -> dict:
     if authorization != sha256(output / "plan/freeze.json"):
         raise ValueError("Execution requires authorization for this exact plan SHA-256")
     plan = load_plan(output)
@@ -398,6 +406,8 @@ def execute(output: Path, authorization: str, *, client=None) -> dict:
                {"started_at": time.time(), "client_runtime": runtime})
     tasks = {task["id"]: task for task in read_json(output / "plan/tasks.json")}
     deadline = time.monotonic() + plan["limits"]["campaign_seconds"]
+    if campaign_deadline is not None:
+        deadline = min(deadline, campaign_deadline)
     tokens, completed, stop_reason = 0, 0, None
     for entry in plan["schedule"]:
         if time.monotonic() >= deadline or tokens >= plan["limits"]["token_stop_new_trials"]:
@@ -418,7 +428,8 @@ def execute(output: Path, authorization: str, *, client=None) -> dict:
             directory = trial_directory(output, entry)
             result = run_trial(client, task, role, entry["repeat"], workspace,
                                directory, output / "plan/schema.json",
-                               output / "validator/verify.py", prompt, deadline)
+                               output / "validator/verify.py", prompt, deadline,
+                               timeout_seconds=plan["limits"]["timeout_seconds"])
             result["evidence_sha256"] = {name: sha256(directory / name) for name in EVIDENCE
                                          if (directory / name).is_file()}
             atomic_write_json(directory / "result.json", result)
@@ -429,6 +440,11 @@ def execute(output: Path, authorization: str, *, client=None) -> dict:
         if result["execution_state"] != "completed":
             stop_reason = "client_" + result["execution_state"]
             break
+        usage = normalize_usage(result["observations"]["usage"])
+        if stop_on_unknown_usage and (usage is None or any(
+                usage.get(key) is None for key in ("input_tokens", "output_tokens"))):
+            stop_reason = "unknown_usage"
+            break
     status = {"client_runtime": runtime,
               "recorded_trials": completed, "expected_trials": len(plan["schedule"]),
               "observed_tokens": tokens, "stop_reason": stop_reason,
@@ -473,7 +489,11 @@ def retained_trial(output: Path, entry: dict) -> dict:
     # Select report fields; raw commands and private workspace paths stay in local artifacts.
     selected = {key: record[key] for key in ("task_id", "condition", "repeat", "split",
                 "verdict", "checks", "elapsed_seconds", "observations", "inputs_unchanged")}
-    selected["observations"] = {"usage": record["observations"]["usage"]}
+    selected["observations"] = {
+        "usage": normalize_usage(record["observations"]["usage"]),
+        "tool_calls": len(record["observations"]["commands"])
+        if isinstance(record["observations"].get("commands"), list) else None,
+    }
     error = record.get("evaluation_error")
     if data is None:
         error = error or "missing_retained_output"
@@ -513,7 +533,7 @@ def condition_summary(trials: list[dict]) -> dict:
         group["expected_trials"] = len(selected)
         group["missing_trials"] = sum(t["missing"] for t in selected)
         for metric in ("input_tokens", "output_tokens"):
-            usage = [t["observations"]["usage"] for t in selected]
+            usage = [normalize_usage(t["observations"]["usage"]) for t in selected]
             group[metric] = sum(u[metric] for u in usage) if all(
                 u is not None and u.get(metric) is not None for u in usage
             ) and usage else None
diff --git a/tools/skill_eval_improve.py b/tools/skill_eval_improve.py
new file mode 100644
index 000000000..5f3455fb8
--- /dev/null
+++ b/tools/skill_eval_improve.py
@@ -0,0 +1,277 @@
+"""Offline single-step harness policy above skill_eval_compare.
+
+The Python adapter accepts trusted fixed-output clients only. There is deliberately
+no real-client execution CLI: external process isolation is not implemented here.
+"""
+
+import copy
+import io
+import os
+import subprocess
+import tempfile
+import time
+from pathlib import Path
+
+from inferrsi.harness.improvement import (
+    SelectionPolicy,
+    assessment,
+    choose,
+    holdout_accepts,
+    rerun_assessment,
+    usage_totals,
+)
+from inferrsi.harness.skill_evaluation import CodexClient
+from inferrsi.infra.artifacts import sha256, write_json
+from inferrsi.knowledge_operations import KnowledgeReader, safe_path
+from tools import skill_eval_compare as compare
+
+CONTROLLERS = (*compare.CONTROLLERS, "tools/skill_eval_improve.py",
+               "src/inferrsi/harness/improvement.py")
+
+
+def frontmatter_prefix(data: bytes) -> bytes:
+    # Match the catalog's delimiter-line rule, including CRLF and surrounding whitespace.
+    stream = io.BytesIO(data)
+    prefix = stream.readline()
+    if prefix.strip() != b"---":
+        raise ValueError("Skill must start with front matter")
+    for line in stream:
+        prefix += line
+        if line.strip() == b"---":
+            return prefix
+    raise ValueError("Unclosed skill front matter")
+
+
+def materialize(repo: Path, baseline: str, skill_root: str, skill: str,
+                proposals: dict, allowed_paths: list[str]) -> tuple[dict, list[str]]:
+    """Write immutable Git objects using a private index; never change a checkout/ref."""
+    if set(proposals) - {"inv", "dom"}:
+        raise ValueError("At most one inv and one dom proposal are supported")
+    if len(set(allowed_paths)) != len(allowed_paths):
+        raise ValueError("Repeated allowlist path")
+    with tempfile.TemporaryDirectory(prefix="improve-snapshot-") as temporary:
+        root = Path(temporary)
+        library = compare.snapshot(repo, baseline, skill_root, root / "source")
+        bundle = KnowledgeReader(skill_root=library).read("checkout:" + skill)
+        for name in allowed_paths:
+            safe_path(name)
+            if name not in bundle.content or Path(name).suffix not in (".md", ".txt"):
+                raise ValueError("Only existing bundled Markdown/text resources may be edited")
+            bundle.content[name].decode("utf-8")
+        for edits in proposals.values():
+            if not isinstance(edits, dict) or set(edits) - set(allowed_paths):
+                raise ValueError("Candidate edit outside frozen allowlist")
+            for name, text in edits.items():
+                if not isinstance(text, str) or "\0" in text:
+                    raise ValueError("Candidate resources must be UTF-8 text")
+                if Path(name).name == "SKILL.md" and frontmatter_prefix(
+                        text.encode("utf-8")) != frontmatter_prefix(bundle.content[name]):
+                    raise ValueError("Skill metadata must remain unchanged")
+                if text.encode("utf-8") == bundle.content[name]:
+                    raise ValueError("Proposal must change resource bytes")
+        inv, dom = proposals.get("inv", {}), proposals.get("dom", {})
+        conflicts = sorted(name for name in inv.keys() & dom.keys() if inv[name] != dom[name])
+        candidates = {name: edits for name, edits in proposals.items() if edits}
+        if inv and dom and not conflicts:
+            candidates["composed"] = {**inv, **dom}
+        revisions = {"h0": baseline}
+        # Object plumbing avoids hooks, checkout filters and branch/index mutations.
+        env = {**os.environ, "GIT_INDEX_FILE": str(root / "index"),
+               "GIT_AUTHOR_NAME": "InferRSI Snapshot", "GIT_COMMITTER_NAME": "InferRSI Snapshot",
+               "GIT_AUTHOR_EMAIL": "snapshot@example.invalid",
+               "GIT_COMMITTER_EMAIL": "snapshot@example.invalid",
+               "GIT_AUTHOR_DATE": "2000-01-01T00:00:00Z",
+               "GIT_COMMITTER_DATE": "2000-01-01T00:00:00Z"}
+
+        def git(*args, data=None):
+            return subprocess.run([*compare.GIT_READ, *args], cwd=repo, env=env,
+                                  input=data, capture_output=True, check=True).stdout.strip()
+
+        for name, edits in candidates.items():
+            git("read-tree", baseline)
+            for path, text in sorted(edits.items()):
+                blob = git("hash-object", "-w", "--stdin", data=text.encode("utf-8"))
+                entry = b"100644 " + blob + b"\t" + (skill_root + "/" + path).encode() + b"\0"
+                git("update-index", "-z", "--index-info", data=entry)
+            tree = git("write-tree").decode("ascii")
+            revisions[name] = git("-c", "commit.gpgsign=false", "commit-tree", tree,
+                                  "-p", baseline, data=b"Declarative harness snapshot\n").decode()
+        return revisions, conflicts
+
+
+def prepare(config: dict, repo: Path, output: Path) -> str:
+    """Freeze tasks, supplied proposals and selection/budget policy without model calls."""
+    config = copy.deepcopy(config)
+    if set(config) != {"skill", "skill_root", "baseline", "proposals", "allowed_paths",
+                       "development", "holdout", "model", "limits", "selection"}:
+        raise ValueError("Unknown or missing improvement configuration field")
+    SelectionPolicy(**config["selection"])
+    compare.require_git_version(subprocess.check_output(("git", "--version"), text=True))
+    cases = config["development"] + config["holdout"]
+    for split in ("development", "holdout"):
+        if not config[split] or any(t.get("split") != split for t in config[split]):
+            raise ValueError("Nonempty, disjoint development and holdout splits are required")
+    comparison = {key: config[key] for key in ("skill", "skill_root", "model", "limits")}
+    comparison.update(mode="update", cases="tasks.json", snapshots={
+        role: {"git_revision": config["baseline"]} for role in compare.ROLES})
+    compare.validate(comparison, cases)
+    proposals = sum(bool(edits) for edits in config["proposals"].values())
+    required_calls = 2 * config["limits"]["repeats"] * (
+        (proposals + 1) * len(config["development"]) + len(config["holdout"]))
+    if required_calls > config["limits"]["max_task_calls"]:
+        raise ValueError("Whole loop exceeds max_task_calls (including rerun and holdout)")
+    revisions, conflicts = materialize(
+        repo, config["baseline"], config["skill_root"], config["skill"],
+        config["proposals"], config["allowed_paths"])
+    output.mkdir(parents=True, exist_ok=False)
+    write_json(output / "freeze.json", {
+        "format_version": 1, "config": config, "revisions": revisions,
+        "conflicts": conflicts, "max_scheduled_calls": required_calls,
+        "controllers": {name: sha256(compare.ROOT / name) for name in CONTROLLERS},
+        "grader_sha256": sha256(compare.ROOT / "tools/skill_eval_verifier.py"),
+        "execution": "trusted_offline_clients_only", "promotion": "none",
+    })
+    return sha256(output / "freeze.json")
+
+
+def load_freeze(output: Path, digest: str) -> dict:
+    if sha256(output / "freeze.json") != digest:
+        raise ValueError("Improvement plan changed after freeze")
+    plan = compare.read_json(output / "freeze.json")
+    for name, expected in plan["controllers"].items():
+        if sha256(compare.ROOT / name) != expected:
+            raise ValueError("Improvement controller changed after freeze")
+    if sha256(compare.ROOT / "tools/skill_eval_verifier.py") != plan["grader_sha256"]:
+        raise ValueError("Improvement grader changed after freeze")
+    return plan
+
+
+def run_offline(repo: Path, output: Path, digest: str, *, client_factory) -> dict:
+    """One exclusive attempt. The factory is trusted test code, never a candidate hook.
+
+    No implicit client, retries, resume, proposal agent or holdout feedback loop.
+    Hashes detect local mutation; they are not an OS sandbox or signatures.
+    """
+    plan = load_freeze(output, digest)
+    config = plan["config"]
+    policy = SelectionPolicy(**config["selection"])
+    results = output / "results"
+    results.mkdir()  # Exclusive claim also consumes interrupted attempts.
+    start = time.monotonic()
+    deadline = start + config["limits"]["campaign_seconds"]
+    phases, reports, trials = {}, {}, []
+    stopped = None
+
+    def read_phase(phase):
+        directory = results / phase
+        if sha256(directory / "plan/freeze.json") != phases[phase]["plan_sha256"]:
+            raise ValueError("Comparison plan changed after authorization")
+        return compare.read_report(directory)
+
+    def evaluate(phase, candidate, split):
+        nonlocal stopped
+        load_freeze(output, digest)
+        if stopped:
+            return None
+        usage = usage_totals(trials)
+        remaining_time = int(deadline - time.monotonic())
+        remaining_calls = config["limits"]["max_task_calls"] - len(trials)
+        remaining_tokens = config["limits"]["token_stop_new_trials"] - (
+            usage["total_tokens"] or 0)
+        required = 2 * len(config[split]) * config["limits"]["repeats"]
+        if trials and usage["total_tokens"] is None:
+            stopped = "unknown_usage"
+        elif remaining_time <= 0 or remaining_calls < required or remaining_tokens <= 0:
+            stopped = "total_budget"
+        if stopped:
+            return None
+        phase_config = {key: config[key] for key in ("skill", "skill_root", "model")}
+        phase_config.update(mode="update", cases=f"{phase}-tasks.json", snapshots={
+            "baseline": {"git_revision": plan["revisions"]["h0"]},
+            "candidate": {"git_revision": plan["revisions"][candidate]}}, limits={
+                **config["limits"], "max_task_calls": remaining_calls,
+                "campaign_seconds": remaining_time, "token_stop_new_trials": remaining_tokens})
+        write_json(results / f"{phase}-tasks.json", config[split])
+        write_json(results / f"{phase}-config.json", phase_config)
+        directory = results / phase
+        authorization = compare.prepare(results / f"{phase}-config.json", repo, directory)
+        # Factory receives only phase and candidate names, never tasks, answers or gate results.
+        client = client_factory(phase, candidate)
+        if client is None or isinstance(client, CodexClient):
+            raise ValueError("Only an explicitly injected trusted offline client is supported")
+        compare.execute(directory, authorization, client=client, campaign_deadline=deadline,
+                        stop_on_unknown_usage=True)
+        phases[phase] = {"candidate": candidate, "plan_sha256": authorization,
+                         "report": f"results/{phase}/report.json"}
+        report = read_phase(phase)  # Revalidate retained bytes before selection.
+        reports[phase] = report
+        trials.extend(t for p in report["pairs"] for t in (p["baseline"], p["candidate"])
+                      if not t["missing"])
+        if report["status"].get("stop_reason") == "unknown_usage":
+            stopped = "unknown_usage"
+        elif report["status"]["execution_state"] != "completed":
+            stopped = "incomplete_execution"
+        elif time.monotonic() >= deadline:
+            stopped = "total_budget"
+        else:
+            total_tokens = usage_totals(trials)["total_tokens"]
+            if total_tokens is None:
+                stopped = "unknown_usage"
+            elif total_tokens > config["limits"]["token_stop_new_trials"]:
+                stopped = "total_budget"
+        return report
+
+    for name in ("inv", "dom"):
+        if name in plan["revisions"]:
+            evaluate(name, name, "development")
+    # Reload earlier probe evidence too: a later phase cannot invalidate it silently.
+    probes = {name: read_phase(name) for name in ("inv", "dom")
+              if name in reports}
+    decision = choose(probes, policy, composable="composed" in plan["revisions"])
+    successor = "h0" if stopped else decision["successor"]
+    rerun = evaluate("rerun", successor, "development")
+    frozen = successor if successor != "h0" and rerun and rerun_assessment(rerun, probes, policy)[
+        "eligible"] and not stopped else "h0"
+    # Improver-facing record is closed before holdout; contains no holdout inputs or outcomes.
+    experience = {"parent": plan["revisions"]["h0"], "decision": decision,
+                  "attempted_successor": successor, "frozen_candidate": frozen,
+                  "rerun": rerun_assessment(rerun, probes, policy) if rerun else None,
+                  "conflicts": plan["conflicts"], "applicability": "offline performance audit",
+                  "promotion": "none", "transfer": "not_evaluated"}
+    write_json(results / "experience.json", experience)
+    write_json(results / "final.json", {"candidate": frozen,
+                                        "git_revision": plan["revisions"][frozen]})
+    final_digest = sha256(results / "final.json")
+    gate = evaluate("holdout", frozen, "holdout")
+    load_freeze(output, digest)
+    if sha256(results / "final.json") != final_digest:
+        raise ValueError("Final candidate changed during holdout")
+    accepted = frozen != "h0" and gate is not None and not stopped and holdout_accepts(gate, policy)
+    # Report refreshed evidence, including any late mutation, without another trial.
+    for phase in reports:
+        reports[phase] = read_phase(phase)
+    if any(not assessment(r, policy)["complete"] for r in reports.values()):
+        accepted = False
+        stopped = stopped or "invalid_retained_evidence"
+    all_trials = [t for r in reports.values() for p in r["pairs"]
+                  for t in (p["baseline"], p["candidate"]) if not t["missing"]]
+    search_trials = [t for phase, r in reports.items() if phase != "holdout"
+                     for p in r["pairs"] for t in (p["baseline"], p["candidate"])
+                     if not t["missing"]]
+    selected_role = "candidate" if accepted else "baseline"
+    selected_trials = [p[selected_role] for p in reports.get("holdout", {}).get("pairs", [])
+                       if not p[selected_role]["missing"]]
+    result = {"format_version": 1, "plan_sha256": digest, "final_sha256": final_digest,
+              "decision": decision, "frozen_candidate": frozen,
+              "selected": frozen if accepted else "h0", "accepted": accepted,
+              "stop_reason": stopped, "phases": phases,
+              "development": experience,
+              "holdout": assessment(reports["holdout"], policy) if gate else None,
+              "usage": {"search": usage_totals(search_trials), "total": usage_totals(all_trials),
+                        "selected_holdout": usage_totals(selected_trials),
+                        "proposal_model_calls": 0, "composition_model_calls": 0},
+              "wall_seconds": time.monotonic() - start,
+              "claim": "Fixed-output controller validation only; model efficacy and engine "
+                       "performance are unmeasured. Monetary cost is unknown."}
+    write_json(output / "report.json", result)
+    return result
diff --git a/tools/skill_eval_pilot.py b/tools/skill_eval_pilot.py
index d52dc7e2f..d07bab23a 100644
--- a/tools/skill_eval_pilot.py
+++ b/tools/skill_eval_pilot.py
@@ -7,7 +7,13 @@
 import time
 from pathlib import Path
 
-from inferrsi.harness.skill_evaluation import CodexClient, export_bundle, run_trial, summarize
+from inferrsi.harness.skill_evaluation import (
+    CodexClient,
+    export_bundle,
+    normalize_usage,
+    run_trial,
+    summarize,
+)
 from inferrsi.infra.artifacts import sha256, write_json
 from inferrsi.knowledge_operations import KnowledgeReader
 
@@ -282,6 +288,7 @@ def recover_attempts(output: Path, schedule: list[dict]) -> list[dict]:
 
 def observed_tokens(usage: dict | None) -> int:
     # Count observed token fields only; report completeness separately in status and summary.
+    usage = normalize_usage(usage)
     if usage is None:
         return 0
     return sum(
@@ -422,12 +429,9 @@ def execute(output: Path, max_new_trials: int | None = None):
             "recorded_trials": len(trials),
             "observed_tokens": tokens,
             "token_usage_complete": all(
-                trial["observations"]["usage"] is not None
-                and all(
-                    trial["observations"]["usage"].get(key) is not None
-                    for key in ("input_tokens", "output_tokens")
-                )
-                for trial in trials
+                usage is not None
+                and all(usage.get(key) is not None for key in ("input_tokens", "output_tokens"))
+                for usage in (normalize_usage(t["observations"]["usage"]) for t in trials)
             ),
             "execution_state": "completed" if len(trials) == len(plan["schedule"]) else "stopped",
             "stop_reason": stop_reason,