feat(gsm8k): add bounded experience flywheel for sealed practice

Introduce deterministic practice-memory infrastructure that adapts sealed
scout output into compact, retention-gated ExperienceRecords with family,
hazard, and promotion-candidate summaries. No serving, corpus, pack, or
report.json mutation — measurement-only adapter for future sprint reuse.

docs/analysis/gsm8k-experience-flywheel-pr1-lookback-2026-06-17.md

@@ -0,0 +1,148 @@
+# GSM8K Experience Flywheel PR-1 — Lookback (2026-06-17)
+
+## 1. Problem statement
+
+Capability Paradigm Sprint 5 proved that sealed practice/scout evidence can discover
+reusable reasoning organs before serving promotion (10/40/0 → 12/38/0).  The next
+layer must make that loop **systematic and programmatic** without saving garbage,
+bloating memory, or letting SPECULATIVE practice artifacts masquerade as reviewed
+knowledge.
+
+PR-1 adds a bounded, deterministic experience artifact layer — not serving promotion,
+not corpus mutation, not auto-accept.
+
+## 2. Trust boundary summary
+
+| Boundary | PR-1 behavior |
+|----------|---------------|
+| Serving path | Unchanged; wrong=0 preserved |
+| report.json | Read-only; mtime tests prove no write |
+| Sealed practice artifacts | Unchanged |
+| Teaching corpus / packs | No mutation |
+| DiscoveryCandidate / proposals | No auto-emission; bridge documented for PR-2 |
+| Contemplation findings | Remain SPECULATIVE; experience records are parallel diagnostic memory |
+| Output | Explicit `--out` only; never default-writes into repo |
+
+Experience records are **structured evidence for operators**, not active memory.
+Promotion into serving or teaching still requires reviewed gates.
+
+## 3. Artifact schema
+
+Module: `evals/gsm8k_math/train_sample/v1/experience.py`
+
+**ExperienceRecord** (pre-compaction):
+- `record_id` — SHA-256 of load-bearing fields
+- `case_id`, `serving_status`, `sealed_status`, `gold_answer`, `sealed_answer`
+- `serving_refusal_family`, `sealed_failure_family`, `candidate_family`
+- `first_missing_primitive`, `arithmetic_chain_signature`
+- `positive_evidence_refs`, `negative_evidence_refs`, `hazard_tags`
+- `recommended_action`, `promotion_status`
+- `source_run_id`, `source_report_hash`, `schema_version`
+
+**CompactedExperienceRecord** (case-level output):
+- Dedupe key over `(case_id, candidate_family, arithmetic_chain_signature, hazard_tags)`
+- `count`, `first_seen_run_id`, `last_seen_run_id`, `status_transitions`
+
+**Experience report** adds:
+- `family_summaries` — per-family lift/block counts and recommended next action
+- `hazard_summaries` — hazard tag → case_ids
+- `promotion_candidates` — families marked candidate or blocked_by_wrong_risk
+- `experience_report_hash` — self-sealing digest
+
+CLI: `scripts/gsm8k_experience_flywheel.py`
+
+## 4. Retention gates
+
+**Keep:**
+1. `lift_refused_to_correct` (refused→correct delta)
+2. `elimination_refused_to_wrong` and sealed-wrong surfaces
+3. Serving-wrong (if any)
+4. `already_served` correct (regression preservation set)
+5. `serving_conservative_win` (conservative boundary evidence)
+6. High-frequency `joint_refusal` clusters (≥3 cases share failure_family)
+
+**Drop:**
+1. Low-signal isolated `joint_refusal` (no cluster, no new family info)
+2. Duplicate signatures within a run (compacted)
+3. Raw problem text / full traces (never stored)
+
+## 5. Compaction logic
+
+Within a run and across runs (`--prior`):
+- Group by dedupe key
+- Collapse to one `CompactedExperienceRecord` with `count`, seen run IDs, status transitions
+- Latest serving/sealed status wins for the compacted row
+
+## 6. Promotion candidate rules
+
+A family is **`candidate`** only when:
+- At least one refused_to_correct record exists in the family group
+- `first_missing_primitive` and `candidate_family` are explicit
+- No `blocked_by_wrong_risk` records in the same family group
+- No unblocked `unbound_target` hazard on lift rows
+
+## 7. Blocked-by-wrong-risk rules
+
+Marked **`blocked_by_wrong_risk`** when:
+- `elimination_refused_to_wrong` or sealed_status=wrong
+- Serving-wrong delta kinds
+- Hazard tags include `sealed_elimination`, `wrong_risk`, `serving_wrong_boundary`
+- Family summary has both lift candidates and blocked records
+
+## 8. Determinism proof
+
+- `record_id`, `source_run_id`, `source_report_hash`, `experience_report_hash` are SHA-256 over canonical JSON (`formation.hashing`)
+- No clock, no randomness, no floats in hashed payloads
+- `test_live_experience_report_determinism` — identical reports on repeated live runs
+- `test_canonical_json_roundtrip` — stable serialization
+
+## 9. Mutation-boundary proof
+
+- `test_report_json_mtime_unchanged_by_experience_import`
+- No imports of `VaultStore.store`, teaching corpus writers, or pack mutators
+- Scout adapter is read-only over existing `build_scout_summary` output
+
+## 10. Tests run
+
+```bash
+git diff --check origin/main...HEAD
+pytest tests/test_gsm8k_experience_flywheel.py -q          # 18 passed
+pytest tests/test_gsm8k_sealed_attempt_scout.py -q
+pytest tests/test_contemplation_loop.py -q
+pytest tests/test_contemplation_pipeline_convergence.py -q
+pytest tests/test_architectural_invariants.py -q           # 123 total passed
+core test --suite smoke -q
+```
+
+## 11. Live artifact snapshot (train_sample, post-#815)
+
+From `build_experience_report()` on current main:
+- Serving: 12 correct / 38 refused / 0 wrong
+- Retained records: high-signal lift, sealed-wrong, promoted regression set
+- Low-signal joint refusals dropped unless clustered
+- `question_bound_product_aggregate` family appears as `promoted_in_pr` for 0003/0021
+
+## 12. Future PRs
+
+| PR | Scope |
+|----|-------|
+| PR-2 | Wire high-confidence `promotion_candidates` → `DiscoveryCandidate` / proposal draft (no auto-accept) |
+| PR-3 | Operator review workbench over experience + contemplation streams |
+| PR-4 | Sprint-to-sprint automatic candidate ranking from compacted history |
+| PR-5 | Use accepted experience records to prioritize next capability paradigm sprint |
+
+**Proposal bridge (PR-2 sketch):**
+- Map `family_summaries` with `promotion_status=candidate` to `DiscoveryCandidate` with `trigger=would_have_grounded` or a new typed trigger
+- Attach `positive_evidence_refs` as `ContemplationEvidenceRef`-compatible pointers
+- Route through existing `TeachingChainProposal` review gate only
+
+## 13. Non-goals
+
+- No serving lift required
+- No auto corpus / pack mutation
+- No auto-accept proposal
+- No broad product_bridge re-enable
+- No report.json rebaseline
+- No sealed artifact movement
+- No background daemon
+- No unbounded logging / raw trace persistence

evals/gsm8k_math/train_sample/v1/experience.py

@@ -0,0 +1,907 @@
+"""GSM8K bounded experience flywheel — PR-1 practice memory layer.
+
+Deterministic, compact, append-only experience artifacts derived from sealed
+scout runs.  Measurement-only: never mutates serving, report.json, packs,
+teaching corpus, or sealed practice artifacts.
+
+Trust boundary:
+  - Reads scout summaries / rows only.
+  - Emits SPECULATIVE experience records for operator reuse.
+  - No auto-proposal, no corpus mutation, no serving promotion.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Literal
+
+from formation.hashing import canonical_json, sha256_of
+
+from evals.gsm8k_math.practice.v1.runner import classify_operation
+from evals.gsm8k_math.train_sample.v1.scout import (
+    SealedAttemptScoutRow,
+    build_scout_rows,
+    build_scout_summary,
+    classify_delta_kind,
+)
+from scripts.gsm8k_frontier_report import _classify_reason, _extract_category
+
+SCHEMA_VERSION = 1
+ADR = "experience_flywheel_pr1"
+
+Status = Literal["correct", "wrong", "refused"]
+PromotionStatus = Literal[
+    "not_promotable",
+    "candidate",
+    "blocked_by_wrong_risk",
+    "promoted_in_pr",
+    "superseded",
+]
+
+_HIGH_FREQ_JOINT_THRESHOLD = 3
+
+_HAZARD_BY_DELTA: dict[str, tuple[str, ...]] = {
+    "elimination_refused_to_wrong": ("sealed_elimination", "wrong_risk"),
+    "serving_wrong_sealed_correct": ("serving_wrong_boundary",),
+    "serving_wrong_other": ("serving_wrong_boundary",),
+    "serving_conservative_win": ("conservative_boundary",),
+}
+
+_BLOCKED_HAZARDS: frozenset[str] = frozenset(
+    {
+        "sealed_elimination",
+        "wrong_risk",
+        "serving_wrong_boundary",
+        "unblocked_ambiguity",
+        "unbound_target",
+        "unbound_unit",
+    }
+)
+
+_PRIMITIVE_BY_CATEGORY: dict[str, str] = {
+    "discrete_count_statement": "relation_hypothesis",
+    "multiplicative_aggregation": "multiplicative_aggregate",
+    "temporal_aggregation": "temporal_tariff",
+    "rate_with_currency": "rate_composition",
+    "unit_partition": "unit_partition",
+    "comparative_with_unit": "compare_multiplicative",
+}
+
+
+@dataclass(frozen=True, slots=True)
+class ExperienceRecord:
+    """One compact practice-memory record (no raw traces)."""
+
+    record_id: str
+    case_id: str
+    serving_status: Status
+    sealed_status: Status
+    gold_answer: str
+    sealed_answer: str | None
+    serving_refusal_family: str
+    sealed_failure_family: str
+    candidate_family: str | None
+    first_missing_primitive: str | None
+    arithmetic_chain_signature: str
+    positive_evidence_refs: tuple[str, ...]
+    negative_evidence_refs: tuple[str, ...]
+    hazard_tags: tuple[str, ...]
+    recommended_action: str
+    promotion_status: PromotionStatus
+    source_run_id: str
+    source_report_hash: str
+    schema_version: int = SCHEMA_VERSION
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "record_id": self.record_id,
+            "case_id": self.case_id,
+            "serving_status": self.serving_status,
+            "sealed_status": self.sealed_status,
+            "gold_answer": self.gold_answer,
+            "sealed_answer": self.sealed_answer,
+            "serving_refusal_family": self.serving_refusal_family,
+            "sealed_failure_family": self.sealed_failure_family,
+            "candidate_family": self.candidate_family,
+            "first_missing_primitive": self.first_missing_primitive,
+            "arithmetic_chain_signature": self.arithmetic_chain_signature,
+            "positive_evidence_refs": list(self.positive_evidence_refs),
+            "negative_evidence_refs": list(self.negative_evidence_refs),
+            "hazard_tags": list(self.hazard_tags),
+            "recommended_action": self.recommended_action,
+            "promotion_status": self.promotion_status,
+            "source_run_id": self.source_run_id,
+            "source_report_hash": self.source_report_hash,
+            "schema_version": self.schema_version,
+        }
+
+    @classmethod
+    def from_dict(cls, payload: dict[str, Any]) -> ExperienceRecord:
+        return cls(
+            record_id=payload["record_id"],
+            case_id=payload["case_id"],
+            serving_status=payload["serving_status"],
+            sealed_status=payload["sealed_status"],
+            gold_answer=str(payload["gold_answer"]),
+            sealed_answer=payload.get("sealed_answer"),
+            serving_refusal_family=payload["serving_refusal_family"],
+            sealed_failure_family=payload["sealed_failure_family"],
+            candidate_family=payload.get("candidate_family"),
+            first_missing_primitive=payload.get("first_missing_primitive"),
+            arithmetic_chain_signature=payload["arithmetic_chain_signature"],
+            positive_evidence_refs=tuple(payload["positive_evidence_refs"]),
+            negative_evidence_refs=tuple(payload["negative_evidence_refs"]),
+            hazard_tags=tuple(payload["hazard_tags"]),
+            recommended_action=payload["recommended_action"],
+            promotion_status=payload["promotion_status"],
+            source_run_id=payload["source_run_id"],
+            source_report_hash=payload["source_report_hash"],
+            schema_version=int(payload.get("schema_version", SCHEMA_VERSION)),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class CompactedExperienceRecord:
+    """Case-level record collapsed across duplicate signatures / runs."""
+
+    dedupe_key: str
+    record_id: str
+    case_id: str
+    serving_status: Status
+    sealed_status: Status
+    gold_answer: str
+    sealed_answer: str | None
+    serving_refusal_family: str
+    sealed_failure_family: str
+    candidate_family: str | None
+    first_missing_primitive: str | None
+    arithmetic_chain_signature: str
+    positive_evidence_refs: tuple[str, ...]
+    negative_evidence_refs: tuple[str, ...]
+    hazard_tags: tuple[str, ...]
+    recommended_action: str
+    promotion_status: PromotionStatus
+    count: int
+    first_seen_run_id: str
+    last_seen_run_id: str
+    status_transitions: tuple[str, ...]
+    source_report_hash: str
+    schema_version: int = SCHEMA_VERSION
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "dedupe_key": self.dedupe_key,
+            "record_id": self.record_id,
+            "case_id": self.case_id,
+            "serving_status": self.serving_status,
+            "sealed_status": self.sealed_status,
+            "gold_answer": self.gold_answer,
+            "sealed_answer": self.sealed_answer,
+            "serving_refusal_family": self.serving_refusal_family,
+            "sealed_failure_family": self.sealed_failure_family,
+            "candidate_family": self.candidate_family,
+            "first_missing_primitive": self.first_missing_primitive,
+            "arithmetic_chain_signature": self.arithmetic_chain_signature,
+            "positive_evidence_refs": list(self.positive_evidence_refs),
+            "negative_evidence_refs": list(self.negative_evidence_refs),
+            "hazard_tags": list(self.hazard_tags),
+            "recommended_action": self.recommended_action,
+            "promotion_status": self.promotion_status,
+            "count": self.count,
+            "first_seen_run_id": self.first_seen_run_id,
+            "last_seen_run_id": self.last_seen_run_id,
+            "status_transitions": list(self.status_transitions),
+            "source_report_hash": self.source_report_hash,
+            "schema_version": self.schema_version,
+        }
+
+
+def _record_id_payload(record: ExperienceRecord) -> dict[str, Any]:
+    return {
+        "case_id": record.case_id,
+        "serving_status": record.serving_status,
+        "sealed_status": record.sealed_status,
+        "gold_answer": record.gold_answer,
+        "sealed_answer": record.sealed_answer,
+        "serving_refusal_family": record.serving_refusal_family,
+        "sealed_failure_family": record.sealed_failure_family,
+        "candidate_family": record.candidate_family,
+        "first_missing_primitive": record.first_missing_primitive,
+        "arithmetic_chain_signature": record.arithmetic_chain_signature,
+        "hazard_tags": list(record.hazard_tags),
+        "promotion_status": record.promotion_status,
+        "schema_version": record.schema_version,
+    }
+
+
+def compute_record_id(record: ExperienceRecord) -> str:
+    return sha256_of(_record_id_payload(record))
+
+
+def compute_dedupe_key(record: ExperienceRecord) -> str:
+    payload = {
+        "case_id": record.case_id,
+        "candidate_family": record.candidate_family,
+        "arithmetic_chain_signature": record.arithmetic_chain_signature,
+        "hazard_tags": sorted(record.hazard_tags),
+    }
+    return sha256_of(payload)
+
+
+def compute_run_id(scout_summary: dict[str, Any]) -> str:
+    payload = {
+        "schema_version": scout_summary.get("schema_version"),
+        "adr": scout_summary.get("adr"),
+        "cases_source": scout_summary.get("cases_source"),
+        "sample_count": scout_summary.get("sample_count"),
+        "serving_counts": scout_summary.get("serving_counts"),
+        "sealed_counts": scout_summary.get("sealed_counts"),
+        "delta_counts": scout_summary.get("delta_counts"),
+    }
+    return sha256_of(payload)
+
+
+def compute_report_hash(scout_summary: dict[str, Any]) -> str:
+    payload = {k: v for k, v in scout_summary.items() if k != "rows"}
+    return sha256_of(payload)
+
+
+def _arithmetic_chain_signature(
+    *,
+    delta_kind: str,
+    operation_class: str,
+    first_failed_step: str | None,
+    trace_key: str,
+) -> str:
+    return "|".join(
+        [
+            delta_kind,
+            operation_class,
+            first_failed_step or "none",
+            trace_key,
+        ]
+    )
+
+
+def _infer_missing_primitive(
+    *,
+    category: str | None,
+    candidate_family: str | None,
+    failure_family: str,
+) -> str | None:
+    if category:
+        return _PRIMITIVE_BY_CATEGORY.get(category, "diagnostic_hold")
+    if candidate_family and ":" in candidate_family:
+        return candidate_family.split(":", 1)[0]
+    if failure_family.startswith("lift_skill_gap_recognized_no_injection_"):
+        parts = failure_family.split("_")
+        if parts and parts[-1] in _PRIMITIVE_BY_CATEGORY:
+            return _PRIMITIVE_BY_CATEGORY[parts[-1]]
+    return None
+
+
+def _hazard_tags(
+    *,
+    delta_kind: str,
+    served_status: Status,
+    sealed_status: Status,
+    refusal_reason: str | None,
+    failure_family: str,
+) -> tuple[str, ...]:
+    tags: list[str] = list(_HAZARD_BY_DELTA.get(delta_kind, ()))
+    reason = (refusal_reason or "").lower()
+    if "fraction" in reason or "half" in reason or "quarter" in reason:
+        tags.append("fraction_surface")
+    if "more than" in reason or "less than" in reason:
+        tags.append("comparative_surface")
+    if sealed_status == "wrong":
+        tags.append("sealed_wrong")
+    if served_status == "wrong":
+        tags.append("serving_wrong")
+    if failure_family == "joint_sealed_no_resolution":
+        tags.append("joint_no_resolution")
+    if "no admissible candidate for question" in reason:
+        tags.append("unbound_target")
+    if delta_kind == "joint_refusal" and not tags:
+        tags.append("low_signal_joint")
+    return tuple(sorted(set(tags)))
+
+
+def _recommended_action(
+    *,
+    delta_kind: str,
+    promotion_status: PromotionStatus,
+    candidate_family: str | None,
+    first_missing_primitive: str | None,
+) -> str:
+    if promotion_status == "blocked_by_wrong_risk":
+        return (
+            "blocked: sealed wrong shares recognizer surface; build confusers "
+            "before any serving promotion"
+        )
+    if promotion_status == "promoted_in_pr":
+        return "preserved: serving correct; monitor for regression"
+    if delta_kind == "lift_refused_to_correct" and first_missing_primitive:
+        return (
+            f"pursue narrow serving organ for primitive={first_missing_primitive} "
+            f"family={candidate_family or 'unclassified'} with confuser matrix"
+        )
+    if delta_kind == "elimination_refused_to_wrong":
+        return "negative evidence: sealed attempt wrong; do not promote surface"
+    if delta_kind == "joint_refusal":
+        return "diagnostic hold: joint refusal; await family cluster or new signal"
+    if delta_kind == "serving_conservative_win":
+        return "conservative boundary: serving correct where sealed did not commit"
+    return "not_promotable: insufficient lift signal"
+
+
+def _classify_promotion_status(
+    *,
+    delta_kind: str,
+    served_status: Status,
+    sealed_status: Status,
+    candidate_family: str | None,
+    first_missing_primitive: str | None,
+    hazard_tags: tuple[str, ...],
+    category: str | None,
+) -> PromotionStatus:
+    if delta_kind == "already_served" and served_status == "correct":
+        return "promoted_in_pr"
+    if delta_kind in ("elimination_refused_to_wrong", "serving_wrong_other"):
+        return "blocked_by_wrong_risk"
+    if delta_kind == "serving_wrong_sealed_correct":
+        return "blocked_by_wrong_risk"
+    if sealed_status == "wrong":
+        return "blocked_by_wrong_risk"
+    if any(tag in _BLOCKED_HAZARDS for tag in hazard_tags):
+        if delta_kind == "lift_refused_to_correct":
+            return "blocked_by_wrong_risk"
+    if delta_kind == "lift_refused_to_correct":
+        if not candidate_family or not first_missing_primitive:
+            return "not_promotable"
+        if category is None and "unbound_target" in hazard_tags:
+            return "blocked_by_wrong_risk"
+        return "candidate"
+    return "not_promotable"
+
+
+def _positive_evidence_refs(
+    *,
+    case_id: str,
+    trace_key: str,
+    candidate_family: str | None,
+    delta_kind: str,
+) -> tuple[str, ...]:
+    refs = [f"scout:case_id={case_id}", f"scout:trace_key={trace_key}"]
+    if candidate_family:
+        refs.append(f"scout:candidate_family={candidate_family}")
+    if delta_kind == "lift_refused_to_correct":
+        refs.append("scout:delta=lift_refused_to_correct")
+    return tuple(refs)
+
+
+def _negative_evidence_refs(
+    *,
+    case_id: str,
+    delta_kind: str,
+    sealed_status: Status,
+    sealed_answer: str | None,
+    gold_answer: str,
+) -> tuple[str, ...]:
+    refs: list[str] = []
+    if delta_kind == "elimination_refused_to_wrong" or sealed_status == "wrong":
+        refs.append(f"scout:sealed_wrong:case_id={case_id}")
+        if sealed_answer is not None:
+            refs.append(f"scout:sealed_answer={sealed_answer}:gold={gold_answer}")
+    return tuple(refs)
+
+
+def _high_frequency_joint_families(rows: tuple[SealedAttemptScoutRow, ...]) -> set[str]:
+    counts: dict[str, int] = {}
+    for row in rows:
+        delta = classify_delta_kind(row.served_status, row.aggressive_status)
+        if delta == "joint_refusal":
+            counts[row.failure_family] = counts.get(row.failure_family, 0) + 1
+    return {fam for fam, n in counts.items() if n >= _HIGH_FREQ_JOINT_THRESHOLD}
+
+
+def should_retain_row(
+    row: SealedAttemptScoutRow,
+    *,
+    delta_kind: str,
+    high_freq_joint_families: set[str],
+) -> bool:
+    if delta_kind in (
+        "lift_refused_to_correct",
+        "elimination_refused_to_wrong",
+        "serving_wrong_sealed_correct",
+        "serving_wrong_other",
+    ):
+        return True
+    if delta_kind == "already_served" and row.served_status == "correct":
+        return True
+    if delta_kind == "serving_conservative_win":
+        return True
+    if delta_kind == "joint_refusal":
+        return row.failure_family in high_freq_joint_families
+    return False
+
+
+def scout_row_to_experience_record(
+    row: SealedAttemptScoutRow,
+    *,
+    source_run_id: str,
+    source_report_hash: str,
+    operation_class: str,
+    category: str | None,
+    high_freq_joint_families: set[str],
+) -> ExperienceRecord | None:
+    delta_kind = classify_delta_kind(row.served_status, row.aggressive_status)
+    if not should_retain_row(
+        row, delta_kind=delta_kind, high_freq_joint_families=high_freq_joint_families
+    ):
+        return None
+
+    chain_sig = _arithmetic_chain_signature(
+        delta_kind=delta_kind,
+        operation_class=operation_class,
+        first_failed_step=row.first_failed_step,
+        trace_key=row.trace_key,
+    )
+    hazards = _hazard_tags(
+        delta_kind=delta_kind,
+        served_status=row.served_status,
+        sealed_status=row.aggressive_status,
+        refusal_reason=row.refusal_reason,
+        failure_family=row.failure_family,
+    )
+    missing = _infer_missing_primitive(
+        category=category,
+        candidate_family=row.candidate_lift_family,
+        failure_family=row.failure_family,
+    )
+    promotion = _classify_promotion_status(
+        delta_kind=delta_kind,
+        served_status=row.served_status,
+        sealed_status=row.aggressive_status,
+        candidate_family=row.candidate_lift_family,
+        first_missing_primitive=missing,
+        hazard_tags=hazards,
+        category=category,
+    )
+    serving_family = row.failure_family if row.served_status == "refused" else "n/a"
+    record = ExperienceRecord(
+        record_id="",
+        case_id=row.case_id,
+        serving_status=row.served_status,
+        sealed_status=row.aggressive_status,
+        gold_answer=row.gold_answer,
+        sealed_answer=row.aggressive_answer,
+        serving_refusal_family=serving_family,
+        sealed_failure_family=row.failure_family,
+        candidate_family=row.candidate_lift_family,
+        first_missing_primitive=missing,
+        arithmetic_chain_signature=chain_sig,
+        positive_evidence_refs=_positive_evidence_refs(
+            case_id=row.case_id,
+            trace_key=row.trace_key,
+            candidate_family=row.candidate_lift_family,
+            delta_kind=delta_kind,
+        ),
+        negative_evidence_refs=_negative_evidence_refs(
+            case_id=row.case_id,
+            delta_kind=delta_kind,
+            sealed_status=row.aggressive_status,
+            sealed_answer=row.aggressive_answer,
+            gold_answer=row.gold_answer,
+        ),
+        hazard_tags=hazards,
+        recommended_action=_recommended_action(
+            delta_kind=delta_kind,
+            promotion_status=promotion,
+            candidate_family=row.candidate_lift_family,
+            first_missing_primitive=missing,
+        ),
+        promotion_status=promotion,
+        source_run_id=source_run_id,
+        source_report_hash=source_report_hash,
+    )
+    rid = compute_record_id(record)
+    return ExperienceRecord(
+        record_id=rid,
+        case_id=record.case_id,
+        serving_status=record.serving_status,
+        sealed_status=record.sealed_status,
+        gold_answer=record.gold_answer,
+        sealed_answer=record.sealed_answer,
+        serving_refusal_family=record.serving_refusal_family,
+        sealed_failure_family=record.sealed_failure_family,
+        candidate_family=record.candidate_family,
+        first_missing_primitive=record.first_missing_primitive,
+        arithmetic_chain_signature=record.arithmetic_chain_signature,
+        positive_evidence_refs=record.positive_evidence_refs,
+        negative_evidence_refs=record.negative_evidence_refs,
+        hazard_tags=record.hazard_tags,
+        recommended_action=record.recommended_action,
+        promotion_status=record.promotion_status,
+        source_run_id=record.source_run_id,
+        source_report_hash=record.source_report_hash,
+    )
+
+
+def records_from_scout_summary(
+    scout_summary: dict[str, Any],
+    cases_by_id: dict[str, dict[str, Any]] | None = None,
+) -> tuple[ExperienceRecord, ...]:
+    rows_data = scout_summary.get("rows")
+    if rows_data is None:
+        raise ValueError("scout_summary must include rows for experience extraction")
+    rows = tuple(
+        SealedAttemptScoutRow(
+            case_id=r["case_id"],
+            served_status=r["served_status"],
+            aggressive_status=r["aggressive_status"],
+            aggressive_answer=r.get("aggressive_answer"),
+            gold_answer=str(r["gold_answer"]),
+            refusal_reason=r.get("refusal_reason"),
+            failure_family=r["failure_family"],
+            candidate_lift_family=r.get("candidate_lift_family"),
+            first_failed_step=r.get("first_failed_step"),
+            trace_key=r["trace_key"],
+        )
+        for r in rows_data
+    )
+    return records_from_scout_rows(
+        rows,
+        scout_summary=scout_summary,
+        cases_by_id=cases_by_id,
+    )
+
+
+def records_from_scout_rows(
+    rows: tuple[SealedAttemptScoutRow, ...],
+    *,
+    scout_summary: dict[str, Any],
+    cases_by_id: dict[str, dict[str, Any]] | None = None,
+) -> tuple[ExperienceRecord, ...]:
+    run_id = compute_run_id(scout_summary)
+    report_hash = compute_report_hash(scout_summary)
+    high_freq = _high_frequency_joint_families(rows)
+    out: list[ExperienceRecord] = []
+    for row in rows:
+        raw_case = (cases_by_id or {}).get(row.case_id, {})
+        op_class = classify_operation(raw_case.get("answer_expression", ""))
+        category = (
+            _extract_category(row.refusal_reason or "")
+            if row.refusal_reason
+            else None
+        )
+        rec = scout_row_to_experience_record(
+            row,
+            source_run_id=run_id,
+            source_report_hash=report_hash,
+            operation_class=op_class,
+            category=category,
+            high_freq_joint_families=high_freq,
+        )
+        if rec is not None:
+            out.append(rec)
+    return tuple(sorted(out, key=lambda r: (r.case_id, r.record_id)))
+
+
+def compact_records(
+    records: tuple[ExperienceRecord, ...],
+) -> tuple[CompactedExperienceRecord, ...]:
+    groups: dict[str, list[ExperienceRecord]] = {}
+    for rec in records:
+        key = compute_dedupe_key(rec)
+        groups.setdefault(key, []).append(rec)
+
+    compacted: list[CompactedExperienceRecord] = []
+    for dedupe_key, group in sorted(groups.items()):
+        group = sorted(group, key=lambda r: (r.source_run_id, r.record_id))
+        first = group[0]
+        last = group[-1]
+        transitions: list[str] = []
+        for rec in group:
+            transition = f"{rec.serving_status}/{rec.sealed_status}:{rec.promotion_status}"
+            if not transitions or transitions[-1] != transition:
+                transitions.append(transition)
+        compacted.append(
+            CompactedExperienceRecord(
+                dedupe_key=dedupe_key,
+                record_id=first.record_id,
+                case_id=first.case_id,
+                serving_status=last.serving_status,
+                sealed_status=last.sealed_status,
+                gold_answer=last.gold_answer,
+                sealed_answer=last.sealed_answer,
+                serving_refusal_family=last.serving_refusal_family,
+                sealed_failure_family=last.sealed_failure_family,
+                candidate_family=last.candidate_family,
+                first_missing_primitive=last.first_missing_primitive,
+                arithmetic_chain_signature=last.arithmetic_chain_signature,
+                positive_evidence_refs=last.positive_evidence_refs,
+                negative_evidence_refs=last.negative_evidence_refs,
+                hazard_tags=last.hazard_tags,
+                recommended_action=last.recommended_action,
+                promotion_status=last.promotion_status,
+                count=len(group),
+                first_seen_run_id=first.source_run_id,
+                last_seen_run_id=last.source_run_id,
+                status_transitions=tuple(transitions),
+                source_report_hash=last.source_report_hash,
+            )
+        )
+    return tuple(sorted(compacted, key=lambda c: (c.case_id, c.dedupe_key)))
+
+
+def merge_compacted_runs(
+    prior: tuple[CompactedExperienceRecord, ...],
+    new_records: tuple[ExperienceRecord, ...],
+) -> tuple[CompactedExperienceRecord, ...]:
+    """Merge prior compacted state with records from a new scout run."""
+    revived = [
+        ExperienceRecord(
+            record_id=c.record_id,
+            case_id=c.case_id,
+            serving_status=c.serving_status,
+            sealed_status=c.sealed_status,
+            gold_answer=c.gold_answer,
+            sealed_answer=c.sealed_answer,
+            serving_refusal_family=c.serving_refusal_family,
+            sealed_failure_family=c.sealed_failure_family,
+            candidate_family=c.candidate_family,
+            first_missing_primitive=c.first_missing_primitive,
+            arithmetic_chain_signature=c.arithmetic_chain_signature,
+            positive_evidence_refs=c.positive_evidence_refs,
+            negative_evidence_refs=c.negative_evidence_refs,
+            hazard_tags=c.hazard_tags,
+            recommended_action=c.recommended_action,
+            promotion_status=c.promotion_status,
+            source_run_id=c.last_seen_run_id,
+            source_report_hash=c.source_report_hash,
+        )
+        for c in prior
+        for _ in range(c.count)
+    ]
+    combined = tuple(revived) + new_records
+    return compact_records(combined)
+
+
+def build_family_summaries(
+    compacted: tuple[CompactedExperienceRecord, ...],
+) -> tuple[dict[str, Any], ...]:
+    families: dict[str, list[CompactedExperienceRecord]] = {}
+    for rec in compacted:
+        fam = rec.candidate_family or rec.sealed_failure_family
+        families.setdefault(fam, []).append(rec)
+
+    summaries: list[dict[str, Any]] = []
+    for family, group in sorted(families.items()):
+        refused_to_correct = sum(
+            1
+            for r in group
+            if r.promotion_status == "candidate"
+            and r.serving_status == "refused"
+            and r.sealed_status == "correct"
+        )
+        sealed_wrong = sum(
+            1 for r in group if "sealed_wrong" in r.hazard_tags
+        )
+        joint_refusal = sum(
+            1 for r in group if "low_signal_joint" in r.hazard_tags or "joint_no_resolution" in r.hazard_tags
+        )
+        promoted = sum(1 for r in group if r.promotion_status == "promoted_in_pr")
+        blocked = sum(1 for r in group if r.promotion_status == "blocked_by_wrong_risk")
+        primitives: dict[str, int] = {}
+        for r in group:
+            if r.first_missing_primitive:
+                primitives[r.first_missing_primitive] = (
+                    primitives.get(r.first_missing_primitive, 0) + r.count
+                )
+        top_primitives = [
+            p for p, _ in sorted(primitives.items(), key=lambda x: (-x[1], x[0]))
+        ][:3]
+        promotion_status = "not_promotable"
+        if blocked and refused_to_correct:
+            promotion_status = "blocked_by_wrong_risk"
+        elif refused_to_correct and not blocked:
+            promotion_status = "candidate"
+        elif blocked:
+            promotion_status = "blocked_by_wrong_risk"
+        summaries.append(
+            {
+                "family": family,
+                "case_ids": sorted({r.case_id for r in group}),
+                "refused_to_correct_count": refused_to_correct,
+                "sealed_wrong_count": sealed_wrong,
+                "joint_refusal_count": joint_refusal,
+                "promoted_count": promoted,
+                "blocked_count": blocked,
+                "top_missing_primitives": top_primitives,
+                "promotion_status": promotion_status,
+                "recommended_next_action": _family_next_action(
+                    family=family,
+                    promotion_status=promotion_status,
+                    refused_to_correct=refused_to_correct,
+                    blocked=blocked,
+                ),
+            }
+        )
+    return tuple(summaries)
+
+
+def _family_next_action(
+    *,
+    family: str,
+    promotion_status: str,
+    refused_to_correct: int,
+    blocked: int,
+) -> str:
+    if promotion_status == "candidate":
+        return (
+            f"design narrow serving organ for family={family} "
+            f"({refused_to_correct} refused_to_correct) with confuser matrix"
+        )
+    if promotion_status == "blocked_by_wrong_risk":
+        return (
+            f"blocked: family={family} has {blocked} wrong-risk records; "
+            "strengthen confusers before promotion"
+        )
+    return f"diagnostic hold: family={family} lacks promotable lift signal"
+
+
+def build_hazard_summaries(
+    compacted: tuple[CompactedExperienceRecord, ...],
+) -> tuple[dict[str, Any], ...]:
+    hazards: dict[str, list[str]] = {}
+    for rec in compacted:
+        for tag in rec.hazard_tags:
+            hazards.setdefault(tag, []).append(rec.case_id)
+    return tuple(
+        {
+            "hazard": tag,
+            "case_ids": sorted(set(case_ids)),
+            "count": len(set(case_ids)),
+        }
+        for tag, case_ids in sorted(hazards.items())
+    )
+
+
+def build_promotion_candidate_summary(
+    family_summaries: tuple[dict[str, Any], ...],
+) -> tuple[dict[str, Any], ...]:
+    return tuple(
+        s
+        for s in family_summaries
+        if s["promotion_status"] in ("candidate", "blocked_by_wrong_risk")
+    )
+
+
+def build_experience_report(
+    scout_summary: dict[str, Any] | None = None,
+    *,
+    cases: list[dict[str, Any]] | None = None,
+    prior_compacted: tuple[CompactedExperienceRecord, ...] | None = None,
+    include_raw_records: bool = False,
+) -> dict[str, Any]:
+    if scout_summary is None:
+        scout_summary = build_scout_summary(cases, include_rows=True)
+    elif "rows" not in scout_summary:
+        raise ValueError("scout_summary must include rows")
+
+    cases_by_id = {c["case_id"]: c for c in (cases or [])}
+    if not cases_by_id and scout_summary.get("rows"):
+        for row in scout_summary["rows"]:
+            cases_by_id.setdefault(row["case_id"], {})
+
+    records = records_from_scout_summary(scout_summary, cases_by_id)
+    if prior_compacted:
+        compacted = merge_compacted_runs(prior_compacted, records)
+    else:
+        compacted = compact_records(records)
+
+    family_summaries = build_family_summaries(compacted)
+    hazard_summaries = build_hazard_summaries(compacted)
+    promotion_summary = build_promotion_candidate_summary(family_summaries)
+    run_id = compute_run_id(scout_summary)
+    report_hash = compute_report_hash(scout_summary)
+
+    body: dict[str, Any] = {
+        "schema_version": SCHEMA_VERSION,
+        "adr": ADR,
+        "regime": "gsm8k_experience_flywheel",
+        "source_run_id": run_id,
+        "source_report_hash": report_hash,
+        "scout_serving_counts": scout_summary.get("serving_counts"),
+        "scout_sealed_counts": scout_summary.get("sealed_counts"),
+        "retained_record_count": len(records),
+        "compacted_record_count": len(compacted),
+        "case_records": [c.as_dict() for c in compacted],
+        "family_summaries": list(family_summaries),
+        "hazard_summaries": list(hazard_summaries),
+        "promotion_candidates": list(promotion_summary),
+    }
+    if include_raw_records:
+        body["raw_records"] = [r.as_dict() for r in records]
+    body["experience_report_hash"] = sha256_of(
+        {k: v for k, v in body.items() if k != "experience_report_hash"}
+    )
+    return body
+
+
+def write_experience_jsonl(
+    report: dict[str, Any],
+    path: Path,
+    *,
+    records_key: str = "case_records",
+) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8") as fh:
+        for row in report.get(records_key, []):
+            fh.write(canonical_json(row).decode("utf-8") + "\n")
+
+
+def write_experience_json(report: dict[str, Any], path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(canonical_json(report) + b"\n")
+
+
+def load_compacted_from_report(payload: dict[str, Any]) -> tuple[CompactedExperienceRecord, ...]:
+    return tuple(
+        CompactedExperienceRecord(
+            dedupe_key=c["dedupe_key"],
+            record_id=c["record_id"],
+            case_id=c["case_id"],
+            serving_status=c["serving_status"],
+            sealed_status=c["sealed_status"],
+            gold_answer=str(c["gold_answer"]),
+            sealed_answer=c.get("sealed_answer"),
+            serving_refusal_family=c["serving_refusal_family"],
+            sealed_failure_family=c["sealed_failure_family"],
+            candidate_family=c.get("candidate_family"),
+            first_missing_primitive=c.get("first_missing_primitive"),
+            arithmetic_chain_signature=c["arithmetic_chain_signature"],
+            positive_evidence_refs=tuple(c["positive_evidence_refs"]),
+            negative_evidence_refs=tuple(c["negative_evidence_refs"]),
+            hazard_tags=tuple(c["hazard_tags"]),
+            recommended_action=c["recommended_action"],
+            promotion_status=c["promotion_status"],
+            count=int(c["count"]),
+            first_seen_run_id=c["first_seen_run_id"],
+            last_seen_run_id=c["last_seen_run_id"],
+            status_transitions=tuple(c["status_transitions"]),
+            source_report_hash=c["source_report_hash"],
+            schema_version=int(c.get("schema_version", SCHEMA_VERSION)),
+        )
+        for c in payload.get("case_records", [])
+    )
+
+
+__all__ = [
+    "ADR",
+    "CompactedExperienceRecord",
+    "ExperienceRecord",
+    "PromotionStatus",
+    "SCHEMA_VERSION",
+    "build_experience_report",
+    "build_family_summaries",
+    "build_hazard_summaries",
+    "build_promotion_candidate_summary",
+    "compact_records",
+    "compute_dedupe_key",
+    "compute_record_id",
+    "compute_report_hash",
+    "compute_run_id",
+    "load_compacted_from_report",
+    "merge_compacted_runs",
+    "records_from_scout_rows",
+    "records_from_scout_summary",
+    "scout_row_to_experience_record",
+    "should_retain_row",
+    "write_experience_json",
+    "write_experience_jsonl",
+]

scripts/gsm8k_experience_flywheel.py

@@ -0,0 +1,109 @@
+#!/usr/bin/env python3
+"""GSM8K bounded experience flywheel — deterministic practice memory builder.
+
+Reads sealed scout evidence and emits compact experience artifacts.  Never
+mutates serving, report.json, packs, teaching corpus, or sealed practice lanes
+unless an explicit --out path is provided by the operator.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+_REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(_REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(_REPO_ROOT))
+
+from evals.gsm8k_math.train_sample.v1.experience import (
+    build_experience_report,
+    load_compacted_from_report,
+    write_experience_json,
+    write_experience_jsonl,
+)
+from evals.gsm8k_math.train_sample.v1.runner import _CASES_PATH, _load_cases
+from evals.gsm8k_math.train_sample.v1.scout import build_scout_summary
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="Build bounded GSM8K experience flywheel artifact from scout"
+    )
+    parser.add_argument(
+        "--cases",
+        type=Path,
+        default=_CASES_PATH,
+        help="Path to cases.jsonl (default: train_sample)",
+    )
+    parser.add_argument(
+        "--limit",
+        type=int,
+        default=None,
+        help="Score only the first N cases (sorted by case_id)",
+    )
+    parser.add_argument(
+        "--prior",
+        type=Path,
+        default=None,
+        help="Optional prior experience report JSON for cross-run compaction",
+    )
+    parser.add_argument(
+        "--out",
+        type=Path,
+        default=None,
+        help="Optional JSON output path (never writes repo artifacts by default)",
+    )
+    parser.add_argument(
+        "--jsonl-out",
+        type=Path,
+        default=None,
+        help="Optional JSONL output path for compacted case records",
+    )
+    parser.add_argument(
+        "--include-raw",
+        action="store_true",
+        help="Include pre-compaction raw records in JSON output",
+    )
+    args = parser.parse_args(argv)
+
+    if not args.cases.exists():
+        print(f"ERROR: cases file not found: {args.cases}", file=sys.stderr)
+        return 1
+
+    cases = _load_cases(args.cases)
+    if args.limit is not None:
+        cases = sorted(cases, key=lambda c: c["case_id"])[: args.limit]
+
+    scout_summary = build_scout_summary(
+        cases,
+        cases_source=str(args.cases),
+        include_rows=True,
+    )
+
+    prior_compacted = None
+    if args.prior is not None:
+        if not args.prior.exists():
+            print(f"ERROR: prior report not found: {args.prior}", file=sys.stderr)
+            return 1
+        prior_payload = json.loads(args.prior.read_text(encoding="utf-8"))
+        prior_compacted = load_compacted_from_report(prior_payload)
+
+    report = build_experience_report(
+        scout_summary,
+        cases=cases,
+        prior_compacted=prior_compacted,
+        include_raw_records=args.include_raw,
+    )
+    print(json.dumps(report, indent=2, sort_keys=True))
+
+    if args.out is not None:
+        write_experience_json(report, args.out)
+    if args.jsonl_out is not None:
+        write_experience_jsonl(report, args.jsonl_out)
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_gsm8k_experience_flywheel.py

@@ -0,0 +1,394 @@
+"""Tests for GSM8K bounded experience flywheel (PR-1)."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from evals.gsm8k_math.runner import CaseOutcome
+from evals.gsm8k_math.train_sample.v1.experience import (
+    build_experience_report,
+    compact_records,
+    compute_dedupe_key,
+    compute_record_id,
+    compute_report_hash,
+    compute_run_id,
+    load_compacted_from_report,
+    merge_compacted_runs,
+    records_from_scout_rows,
+    scout_row_to_experience_record,
+    should_retain_row,
+    write_experience_json,
+)
+from evals.gsm8k_math.train_sample.v1.scout import (
+    SealedAttemptScoutRow,
+    build_scout_row,
+    build_scout_summary,
+    classify_delta_kind,
+)
+from formation.hashing import canonical_json
+
+_REPO_ROOT = Path(__file__).resolve().parents[1]
+_REPORT = _REPO_ROOT / "evals/gsm8k_math/train_sample/v1/report.json"
+_FIXTURE_CASES = _REPO_ROOT / "tests/fixtures/gsm8k_experience_flywheel_cases.jsonl"
+
+
+def _outcome(
+    *,
+    case_id: str,
+    outcome: str,
+    reason: str = "",
+    actual: float | None = None,
+    expected: float = 0.0,
+) -> CaseOutcome:
+    return CaseOutcome(
+        case_id=case_id,
+        outcome=outcome,  # type: ignore[arg-type]
+        reason=reason,
+        expected_answer=expected,
+        expected_unit="",
+        actual_answer=actual,
+        actual_unit=None,
+        trace_hash=None,
+        realized_prose=None,
+    )
+
+
+def _lift_row(case_id: str = "gsm8k-train-sample-v1-0003") -> SealedAttemptScoutRow:
+    raw = {
+        "case_id": case_id,
+        "question": "Revenue question",
+        "answer_numeric": 864,
+        "answer_expression": "#### 864",
+    }
+    served = _outcome(
+        case_id=case_id,
+        outcome="refused",
+        reason=(
+            "candidate_graph: recognizer matched but produced no injection "
+            "(category=discrete_count_statement)"
+        ),
+        expected=864.0,
+    )
+    sealed = _outcome(
+        case_id=case_id,
+        outcome="correct",
+        reason="resolve_pooled",
+        actual=864.0,
+        expected=864.0,
+    )
+    return build_scout_row(raw, served, sealed)
+
+
+def _sealed_wrong_row(case_id: str = "gsm8k-train-sample-v1-0011") -> SealedAttemptScoutRow:
+    raw = {
+        "case_id": case_id,
+        "question": "Elimination hazard",
+        "answer_numeric": 50,
+        "answer_expression": "#### 50",
+    }
+    served = _outcome(
+        case_id=case_id,
+        outcome="refused",
+        reason="candidate_graph: no admissible candidate for statement",
+        expected=50.0,
+    )
+    sealed = _outcome(
+        case_id=case_id,
+        outcome="wrong",
+        reason="resolve_pooled",
+        actual=3200.0,
+        expected=50.0,
+    )
+    return build_scout_row(raw, served, sealed)
+
+
+def _joint_refusal_row(
+    case_id: str,
+    failure_family: str = "joint_skill_gap_no_admissible_statement",
+) -> SealedAttemptScoutRow:
+    raw = {
+        "case_id": case_id,
+        "question": "Joint refusal",
+        "answer_numeric": 10,
+        "answer_expression": "#### 10",
+    }
+    served = _outcome(
+        case_id=case_id,
+        outcome="refused",
+        reason="candidate_graph: no admissible candidate for statement",
+        expected=10.0,
+    )
+    sealed = _outcome(
+        case_id=case_id,
+        outcome="refused",
+        reason="resolve_pooled: no resolution",
+        expected=10.0,
+    )
+    row = build_scout_row(raw, served, sealed)
+    return SealedAttemptScoutRow(
+        case_id=row.case_id,
+        served_status=row.served_status,
+        aggressive_status=row.aggressive_status,
+        aggressive_answer=row.aggressive_answer,
+        gold_answer=row.gold_answer,
+        refusal_reason=row.refusal_reason,
+        failure_family=failure_family,
+        candidate_lift_family=row.candidate_lift_family,
+        first_failed_step=row.first_failed_step,
+        trace_key=row.trace_key,
+    )
+
+
+def _scout_summary_from_rows(rows: tuple[SealedAttemptScoutRow, ...]) -> dict:
+    return {
+        "schema_version": 1,
+        "adr": "0175",
+        "regime": "sealed_attempt_scout",
+        "cases_source": "fixture",
+        "sample_count": len(rows),
+        "serving_counts": {"correct": 0, "wrong": 0, "refused": len(rows)},
+        "sealed_counts": {"correct": 0, "wrong": 0, "refused": len(rows)},
+        "delta_counts": {"joint_refusal": len(rows)},
+        "lift_recommendations": [],
+        "rows": [r.as_dict() for r in rows],
+    }
+
+
+def test_record_id_is_deterministic():
+    row = _lift_row()
+    scout = _scout_summary_from_rows((row,))
+    recs = records_from_scout_rows((row,), scout_summary=scout, cases_by_id={})
+    assert len(recs) == 1
+    a = compute_record_id(recs[0])
+    b = compute_record_id(recs[0])
+    assert a == b
+    assert recs[0].record_id == a
+
+
+def test_run_id_and_report_hash_deterministic():
+    row = _lift_row()
+    scout = _scout_summary_from_rows((row,))
+    assert compute_run_id(scout) == compute_run_id(scout)
+    assert compute_report_hash(scout) == compute_report_hash(scout)
+
+
+def test_refused_to_correct_retained_as_candidate():
+    row = _lift_row()
+    scout = _scout_summary_from_rows((row,))
+    recs = records_from_scout_rows((row,), scout_summary=scout)
+    assert len(recs) == 1
+    assert recs[0].promotion_status == "candidate"
+    assert recs[0].candidate_family is not None
+    assert recs[0].first_missing_primitive == "relation_hypothesis"
+
+
+def test_sealed_wrong_retained_as_blocked():
+    row = _sealed_wrong_row()
+    scout = _scout_summary_from_rows((row,))
+    recs = records_from_scout_rows((row,), scout_summary=scout)
+    assert len(recs) == 1
+    assert recs[0].promotion_status == "blocked_by_wrong_risk"
+    assert "sealed_wrong" in recs[0].hazard_tags
+    assert recs[0].negative_evidence_refs
+
+
+def test_low_signal_joint_refusal_dropped():
+    row = _joint_refusal_row("gsm8k-train-sample-v1-9001")
+    delta = classify_delta_kind(row.served_status, row.aggressive_status)
+    assert delta == "joint_refusal"
+    assert not should_retain_row(row, delta_kind=delta, high_freq_joint_families=set())
+
+
+def test_high_frequency_joint_refusal_retained():
+    fam = "joint_skill_gap_no_admissible_statement"
+    rows = tuple(_joint_refusal_row(f"gsm8k-train-sample-v1-90{i:02d}", fam) for i in range(3))
+    scout = _scout_summary_from_rows(rows)
+    recs = records_from_scout_rows(rows, scout_summary=scout)
+    assert len(recs) == 3
+
+
+def test_duplicate_compaction_collapses_count():
+    row = _lift_row()
+    scout = _scout_summary_from_rows((row,))
+    recs = records_from_scout_rows((row, row), scout_summary=scout)
+    compacted = compact_records(recs)
+    assert len(compacted) == 1
+    assert compacted[0].count == 2
+    assert compacted[0].first_seen_run_id == compacted[0].last_seen_run_id
+
+
+def test_merge_compacted_runs_increments_count():
+    row = _lift_row()
+    scout = _scout_summary_from_rows((row,))
+    first = compact_records(records_from_scout_rows((row,), scout_summary=scout))
+    second_recs = records_from_scout_rows((row,), scout_summary=scout)
+    merged = merge_compacted_runs(first, second_recs)
+    assert len(merged) == 1
+    assert merged[0].count == 2
+
+
+def test_blocked_family_cannot_be_candidate_in_summary():
+    rows = (_lift_row("gsm8k-train-sample-v1-0003"), _sealed_wrong_row())
+    scout = _scout_summary_from_rows(rows)
+    report = build_experience_report(scout, include_raw_records=False)
+    families = {f["family"]: f for f in report["family_summaries"]}
+    blocked_fams = [
+        f for f in report["family_summaries"] if f["promotion_status"] == "candidate"
+    ]
+    for fam in blocked_fams:
+        assert fam["blocked_count"] == 0
+    assert any(f["promotion_status"] == "blocked_by_wrong_risk" for f in families.values())
+
+
+def test_experience_report_hash_stable():
+    row = _lift_row()
+    scout = _scout_summary_from_rows((row,))
+    a = build_experience_report(scout)
+    b = build_experience_report(scout)
+    assert a["experience_report_hash"] == b["experience_report_hash"]
+
+
+def test_canonical_json_roundtrip(tmp_path: Path):
+    row = _lift_row()
+    scout = _scout_summary_from_rows((row,))
+    report = build_experience_report(scout)
+    out = tmp_path / "experience.json"
+    write_experience_json(report, out)
+    loaded = json.loads(out.read_text(encoding="utf-8"))
+    assert loaded["experience_report_hash"] == report["experience_report_hash"]
+    compacted = load_compacted_from_report(loaded)
+    assert len(compacted) == 1
+
+
+def test_report_json_mtime_unchanged_by_experience_import():
+    before = _REPORT.stat().st_mtime_ns
+    _ = compute_record_id
+    after = _REPORT.stat().st_mtime_ns
+    assert before == after
+
+
+def test_live_experience_report_determinism():
+    a = build_experience_report()
+    b = build_experience_report()
+    assert json.dumps(a, sort_keys=True) == json.dumps(b, sort_keys=True)
+
+
+def test_live_serving_wrong_remains_zero_in_experience():
+    report = build_experience_report()
+    assert report["scout_serving_counts"]["wrong"] == 0
+
+
+def test_no_floats_in_hashed_payloads():
+    row = _lift_row()
+    scout = _scout_summary_from_rows((row,))
+    recs = records_from_scout_rows((row,), scout_summary=scout)
+    for rec in recs:
+        canonical_json(rec.as_dict())
+
+
+def test_promoted_in_pr_for_served_correct():
+    raw = {
+        "case_id": "gsm8k-train-sample-v1-0002",
+        "question": "Already served",
+        "answer_numeric": 18,
+        "answer_expression": "#### 18",
+    }
+    served = _outcome(case_id=raw["case_id"], outcome="correct", actual=18.0, expected=18.0)
+    sealed = _outcome(case_id=raw["case_id"], outcome="correct", actual=18.0, expected=18.0)
+    row = build_scout_row(raw, served, sealed)
+    scout = _scout_summary_from_rows((row,))
+    recs = records_from_scout_rows((row,), scout_summary=scout)
+    assert len(recs) == 1
+    assert recs[0].promotion_status == "promoted_in_pr"
+
+
+def test_dedupe_key_ignores_run_id():
+    row = _lift_row()
+    scout = _scout_summary_from_rows((row,))
+    cases_by_id = {
+        row.case_id: {
+            "case_id": row.case_id,
+            "answer_expression": "#### 864",
+        }
+    }
+    recs = records_from_scout_rows(
+        (row,), scout_summary=scout, cases_by_id=cases_by_id
+    )
+    key_a = compute_dedupe_key(recs[0])
+    op_class = recs[0].arithmetic_chain_signature.split("|")[1]
+    rec_b = scout_row_to_experience_record(
+        row,
+        source_run_id="different-run",
+        source_report_hash="different-hash",
+        operation_class=op_class,
+        category="discrete_count_statement",
+        high_freq_joint_families=set(),
+    )
+    assert rec_b is not None
+    assert compute_dedupe_key(rec_b) == key_a
+
+
+@pytest.fixture
+def injected_scout_summary():
+    cases = [
+        {
+            "case_id": "gsm8k-train-sample-v1-0003",
+            "question": "Q",
+            "answer_numeric": 864,
+            "answer_expression": "#### 864",
+        },
+        {
+            "case_id": "gsm8k-train-sample-v1-0011",
+            "question": "Q2",
+            "answer_numeric": 50,
+            "answer_expression": "#### 50",
+        },
+    ]
+
+    def serving(adapted: dict) -> CaseOutcome:
+        if "0003" in adapted["id"]:
+            return _outcome(
+                case_id=adapted["id"],
+                outcome="refused",
+                reason=(
+                    "candidate_graph: recognizer matched but produced no injection "
+                    "(category=discrete_count_statement)"
+                ),
+                expected=864.0,
+            )
+        return _outcome(
+            case_id=adapted["id"],
+            outcome="refused",
+            reason="candidate_graph: no admissible candidate for statement",
+            expected=50.0,
+        )
+
+    def sealed(adapted: dict) -> CaseOutcome:
+        if "0003" in adapted["id"]:
+            return _outcome(
+                case_id=adapted["id"],
+                outcome="correct",
+                actual=864.0,
+                expected=864.0,
+            )
+        return _outcome(
+            case_id=adapted["id"],
+            outcome="wrong",
+            actual=3200.0,
+            expected=50.0,
+        )
+
+    return build_scout_summary(
+        cases, cases_source="fixture", serving_scorer=serving, sealed_scorer=sealed
+    )
+
+
+def test_injected_scout_adapter_produces_retained_records(injected_scout_summary):
+    report = build_experience_report(injected_scout_summary)
+    assert report["retained_record_count"] >= 2
+    statuses = {r["promotion_status"] for r in report["case_records"]}
+    assert "candidate" in statuses
+    assert "blocked_by_wrong_risk" in statuses