From 0ca48cc9a3577e31f2ae554a6e4b27cbd9619007 Mon Sep 17 00:00:00 2001 From: Shay Date: Wed, 17 Jun 2026 21:05:02 -0700 Subject: [PATCH] feat(gsm8k): add bounded experience flywheel for sealed practice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- ...rience-flywheel-pr1-lookback-2026-06-17.md | 148 +++ .../gsm8k_math/train_sample/v1/experience.py | 907 ++++++++++++++++++ scripts/gsm8k_experience_flywheel.py | 109 +++ tests/test_gsm8k_experience_flywheel.py | 394 ++++++++ 4 files changed, 1558 insertions(+) create mode 100644 docs/analysis/gsm8k-experience-flywheel-pr1-lookback-2026-06-17.md create mode 100644 evals/gsm8k_math/train_sample/v1/experience.py create mode 100644 scripts/gsm8k_experience_flywheel.py create mode 100644 tests/test_gsm8k_experience_flywheel.py diff --git a/docs/analysis/gsm8k-experience-flywheel-pr1-lookback-2026-06-17.md b/docs/analysis/gsm8k-experience-flywheel-pr1-lookback-2026-06-17.md new file mode 100644 index 00000000..5fe9dbfb --- /dev/null +++ b/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 \ No newline at end of file diff --git a/evals/gsm8k_math/train_sample/v1/experience.py b/evals/gsm8k_math/train_sample/v1/experience.py new file mode 100644 index 00000000..13ecc1d5 --- /dev/null +++ b/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", +] \ No newline at end of file diff --git a/scripts/gsm8k_experience_flywheel.py b/scripts/gsm8k_experience_flywheel.py new file mode 100644 index 00000000..bffefa5f --- /dev/null +++ b/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()) \ No newline at end of file diff --git a/tests/test_gsm8k_experience_flywheel.py b/tests/test_gsm8k_experience_flywheel.py new file mode 100644 index 00000000..f9a076c5 --- /dev/null +++ b/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 \ No newline at end of file