core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(evals): add GSM8K sealed attempt scout (#812)

Browse source 
* feat(evals): add GSM8K sealed attempt scout

Deterministic train_sample dual-scorer (serving vs resolve_pooled) that
classifies refusal families and ranks lift targets for Capability Strike.
Measurement-only: no serving mutation, no report.json writes by default.

* chore(analysis): normalize sealed scout lookback EOF
This commit is contained in:
Shay 2026-06-17 19:55:13 -07:00 committed by GitHub
parent 708f27a240
commit 0215c30bbe
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 842 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

85
docs/analysis/gsm8k-sealed-attempt-scout-s1-lookback-2026-06-17.md Normal file
View file

@ -0,0 +1,85 @@
# GSM8K Sealed Attempt Scout S1 — Lookback (2026-06-17)
## Purpose
First practical bridge from ADR-0175 practice/contemplation design into the
Capability Strike lift workflow. Dual-scores **train_sample** cases with:
1. **Serving** — conservative `_score_one_candidate_graph` (wrong=0 path)
2. **Sealed** — aggressive `resolve_pooled_scorer` (may be wrong; practice-only)
## Boundaries (enforced)
- **No serving mutation** — scout imports scorers read-only; no runtime edits.
- **No `report.json` rebaseline** — default CLI prints to stdout only.
- **No sealed-lane movement** — does not call `regenerate_practice_artifacts()`.
- **No autonomous promotion** — recommendations are diagnostic/SPECULATIVE only.
## Runner
```bash
uv run python scripts/gsm8k_sealed_attempt_scout.py
uv run python scripts/gsm8k_sealed_attempt_scout.py --out /tmp/scout.jsonl
```
Core logic: `evals/gsm8k_math/train_sample/v1/scout.py`
## Schema (`SealedAttemptScoutRow`)
| Field | Description |
|-------|-------------|
| `case_id` | Train-sample case id |
| `served_status` | correct / wrong / refused |
| `aggressive_status` | correct / wrong / refused |
| `aggressive_answer` | Sealed numeric answer if any |
| `gold_answer` | Dataset gold |
| `refusal_reason` | Serving refusal when refused |
| `failure_family` | Conservative taxonomy |
| `candidate_lift_family` | Primitive hint when lift candidate |
| `first_failed_step` | question_parse / injection / completeness / … |
| `trace_key` | Deterministic SHA-256 prefix |
## Baseline observed (full train_sample, #811 main)
Ephemeral serving (live code): **8 correct / 42 refused / 0 wrong**.
Scout full pass (serving arm matches live): `serving_counts.wrong == 0`.
Typical cross-regime pattern on refused cases:
- `lift_refused_to_correct` — sealed commits, serving refuses (primary lift map)
- `joint_refusal` — both arms refuse (substrate gap)
- `elimination_refused_to_wrong` — sealed wrong (not a lift target)
## Top recommended lift families (scout ranking)
On full 50-case pass, top groups cluster on:
1. `recognized_no_injection` + `discrete_count_statement` → `relation_hypothesis`
2. `recognized_no_injection` + `multiplicative_aggregation` → `multiplicative_aggregate`
3. `no_admissible_question` → `question_binding` families (peer/conditional/yield)
**Note:** Track A Batch 3 landed `peer_partition_question` independently; scout
would have surfaced 0025-style `no_admissible_question` refusals on the #811
baseline.
## Usage alongside Capability Strike
1. Run scout after a merge to rank refused cases where sealed already commits.
2. Pick the highest-count **family** with confuser review (not case-id chasing).
3. Implement narrow injector lift; re-run ephemeral `build_report()` for proof.
4. Never wire `resolve_pooled` wholesale to serving.
## Limitations (S1)
- Failure taxonomy is conservative; unknown → `unclassified`.
- No per-step operation-chain extraction beyond `first_failed_step` heuristic.
- Train_sample only (50 cases); practice lane (150) is a future `--cases` extension.
- No timestamps in golden outputs; order fixed by `case_id`.
## Non-goals
- No serving guesses.
- No pack/policy/identity mutation.
- No accepted runtime proposal emission.
- No `determine()` / `FrameVerdict` / `CLOSE`.

474
evals/gsm8k_math/train_sample/v1/scout.py Normal file
View file

@ -0,0 +1,474 @@
"""GSM8K train-sample sealed attempt scout — measurement-only (ADR-0175 S1).
Dual-scores each train_sample case with the conservative serving scorer and the
sealed ``resolve_pooled`` aggressive scorer. Emits deterministic lift-target
evidence without mutating serving, report.json, or practice artifacts.
"""
from __future__ import annotations
import hashlib
import json
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Literal
from evals.gsm8k_math.practice.v1.propose_runner import resolve_pooled_scorer
from evals.gsm8k_math.practice.v1.runner import classify_operation, diagnose_refusal
from evals.gsm8k_math.runner import CaseOutcome, _score_one_candidate_graph
from evals.gsm8k_math.train_sample.v1.runner import _adapt, _load_cases, _CASES_PATH
from scripts.gsm8k_frontier_report import _classify_reason, _extract_category
Status = Literal["correct", "wrong", "refused"]
_DELTA_KINDS: tuple[str, ...] = (
"already_served",
"serving_conservative_win",
"serving_wrong_sealed_correct",
"serving_wrong_other",
"lift_refused_to_correct",
"elimination_refused_to_wrong",
"joint_refusal",
)
_BUCKET_PRIORITY: dict[str, int] = {
"recognized_no_injection": 0,
"no_admissible_statement": 1,
"no_admissible_question": 2,
"no_solvable_branch": 3,
"incomplete_reading": 4,
"other_refused": 5,
"other": 6,
}
_PRIMITIVE_BY_NO_INJ_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",
}
_EVIDENCE_SNIPPET_RE = re.compile(r"\d|half|quarter|third|twice|each|per|every", re.I)
@dataclass(frozen=True, slots=True)
class SealedAttemptScoutRow:
case_id: str
served_status: Status
aggressive_status: Status
aggressive_answer: str | None
gold_answer: str
refusal_reason: str | None
failure_family: str
candidate_lift_family: str | None
first_failed_step: str | None
trace_key: str
def as_dict(self) -> dict[str, Any]:
return {
"case_id": self.case_id,
"served_status": self.served_status,
"aggressive_status": self.aggressive_status,
"aggressive_answer": self.aggressive_answer,
"gold_answer": self.gold_answer,
"refusal_reason": self.refusal_reason,
"failure_family": self.failure_family,
"candidate_lift_family": self.candidate_lift_family,
"first_failed_step": self.first_failed_step,
"trace_key": self.trace_key,
}
@dataclass(frozen=True, slots=True)
class LiftRecommendation:
rank: int
failure_family: str
serving_bucket: str
serving_no_injection_category: str | None
operation_class: str
lift_count: int
case_ids: tuple[str, ...]
candidate_primitive: str
expected_movement: str
safe_next_action: str
def as_dict(self) -> dict[str, Any]:
return {
"rank": self.rank,
"failure_family": self.failure_family,
"serving_bucket": self.serving_bucket,
"serving_no_injection_category": self.serving_no_injection_category,
"operation_class": self.operation_class,
"lift_count": self.lift_count,
"case_ids": self.case_ids,
"candidate_primitive": self.candidate_primitive,
"expected_movement": self.expected_movement,
"safe_next_action": self.safe_next_action,
}
def adapt_train_sample_case(raw: dict[str, Any]) -> dict[str, Any]:
return _adapt(raw)
def _evidence_snippet(question: str, *, limit: int = 96) -> str:
text = (question or "").strip()
if len(text) <= limit:
return text
m = _EVIDENCE_SNIPPET_RE.search(text)
if m is None:
return text[:limit]
start = max(0, m.start() - 20)
return text[start : start + limit].strip()
def _trace_key(case_id: str, served_reason: str, sealed_reason: str) -> str:
payload = f"{case_id}|{served_reason}|{sealed_reason}"
return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]
def classify_delta_kind(served: Status, aggressive: Status) -> str:
if served == "correct" and aggressive == "correct":
return "already_served"
if served == "correct" and aggressive != "correct":
return "serving_conservative_win"
if served == "wrong" and aggressive == "correct":
return "serving_wrong_sealed_correct"
if served == "wrong":
return "serving_wrong_other"
if served == "refused" and aggressive == "correct":
return "lift_refused_to_correct"
if served == "refused" and aggressive == "wrong":
return "elimination_refused_to_wrong"
return "joint_refusal"
def _candidate_lift_family(
*,
delta_kind: str,
serving_bucket: str,
category: str | None,
) -> str | None:
if delta_kind != "lift_refused_to_correct":
return None
if category:
primitive = _PRIMITIVE_BY_NO_INJ_CATEGORY.get(category, "diagnostic_hold")
return f"{primitive}:{category}"
if serving_bucket == "no_admissible_question":
return "question_binding:peer_or_conditional"
if serving_bucket == "incomplete_reading":
return "completeness:unconsumed_quantity"
return "unclassified"
def classify_failure_family(
*,
delta_kind: str,
served_status: Status,
served_reason: str,
served_bucket: str,
served_category: str | None,
sealed_reason: str,
) -> str:
diagnosis = (
diagnose_refusal(served_reason) if served_status == "refused" else "n/a"
)
if delta_kind == "already_served":
return "already_served"
if delta_kind == "serving_conservative_win":
return "conservative_boundary"
if delta_kind in ("serving_wrong_sealed_correct", "serving_wrong_other"):
return "serving_wrong_boundary"
if delta_kind == "elimination_refused_to_wrong":
return "sealed_elimination"
if delta_kind == "lift_refused_to_correct":
parts = ["lift", diagnosis, served_bucket]
if served_category:
parts.append(served_category)
return "_".join(parts)
if delta_kind == "joint_refusal":
if "no resolution" in (sealed_reason or "").lower():
return "joint_sealed_no_resolution"
parts = ["joint", diagnosis, served_bucket]
if served_category:
parts.append(served_category)
return "_".join(parts)
return "unclassified"
def _first_failed_step(served_status: Status, served_reason: str) -> str | None:
if served_status != "refused":
return None
low = (served_reason or "").lower()
if "no admissible candidate for question" in low:
return "question_parse"
if "no admissible candidate for statement" in low:
return "statement_parse"
if "produced no injection" in low:
return "recognizer_injection"
if "no branch produced a solvable" in low:
return "graph_solve"
if "incomplete reading" in low:
return "completeness_guard"
return "unclassified"
def score_case_dual(
raw: dict[str, Any],
*,
serving_scorer: Callable[[dict[str, Any]], CaseOutcome] = _score_one_candidate_graph,
sealed_scorer: Callable[[dict[str, Any]], CaseOutcome] = resolve_pooled_scorer,
) -> tuple[CaseOutcome, CaseOutcome]:
adapted = adapt_train_sample_case(raw)
return serving_scorer(adapted), sealed_scorer(adapted)
def build_scout_row(
raw: dict[str, Any],
served: CaseOutcome,
sealed: CaseOutcome,
) -> SealedAttemptScoutRow:
served_status: Status = served.outcome # type: ignore[assignment]
aggressive_status: Status = sealed.outcome # type: ignore[assignment]
served_reason = served.reason or ""
sealed_reason = sealed.reason or ""
served_bucket = _classify_reason(served_reason)
served_category = _extract_category(served_reason)
delta_kind = classify_delta_kind(served_status, aggressive_status)
failure_family = classify_failure_family(
delta_kind=delta_kind,
served_status=served_status,
served_reason=served_reason,
served_bucket=served_bucket,
served_category=served_category,
sealed_reason=sealed_reason,
)
aggressive_answer = (
None
if sealed.actual_answer is None
else str(sealed.actual_answer)
)
return SealedAttemptScoutRow(
case_id=raw["case_id"],
served_status=served_status,
aggressive_status=aggressive_status,
aggressive_answer=aggressive_answer,
gold_answer=str(raw["answer_numeric"]),
refusal_reason=served_reason if served_status == "refused" else None,
failure_family=failure_family,
candidate_lift_family=_candidate_lift_family(
delta_kind=delta_kind,
serving_bucket=served_bucket,
category=served_category,
),
first_failed_step=_first_failed_step(served_status, served_reason),
trace_key=_trace_key(raw["case_id"], served_reason, sealed_reason),
)
def build_scout_rows(
cases: list[dict[str, Any]],
*,
serving_scorer: Callable[[dict[str, Any]], CaseOutcome] | None = None,
sealed_scorer: Callable[[dict[str, Any]], CaseOutcome] | None = None,
) -> tuple[SealedAttemptScoutRow, ...]:
serving = serving_scorer or _score_one_candidate_graph
sealed = sealed_scorer or resolve_pooled_scorer
rows: list[SealedAttemptScoutRow] = []
for raw in sorted(cases, key=lambda c: c["case_id"]):
served, aggressive = score_case_dual(
raw, serving_scorer=serving, sealed_scorer=sealed
)
rows.append(build_scout_row(raw, served, aggressive))
return tuple(rows)
def _aggregate_counts(rows: tuple[SealedAttemptScoutRow, ...]) -> dict[str, Any]:
serving_counts = {"correct": 0, "wrong": 0, "refused": 0}
sealed_counts = {"correct": 0, "wrong": 0, "refused": 0}
delta_counts: dict[str, int] = {k: 0 for k in _DELTA_KINDS}
failure_family_counts: dict[str, int] = {}
diagnosis_counts: dict[str, int] = {}
for row in rows:
serving_counts[row.served_status] += 1
sealed_counts[row.aggressive_status] += 1
delta_kind = classify_delta_kind(row.served_status, row.aggressive_status)
delta_counts[delta_kind] = delta_counts.get(delta_kind, 0) + 1
failure_family_counts[row.failure_family] = (
failure_family_counts.get(row.failure_family, 0) + 1
)
if row.served_status == "refused":
diag = diagnose_refusal(row.refusal_reason or "")
diagnosis_counts[diag] = diagnosis_counts.get(diag, 0) + 1
return {
"serving_counts": serving_counts,
"sealed_counts": sealed_counts,
"delta_counts": dict(sorted(delta_counts.items())),
"failure_family_counts": dict(sorted(failure_family_counts.items())),
"diagnosis_counts": dict(sorted(diagnosis_counts.items())),
}
def build_lift_recommendations(
rows: tuple[SealedAttemptScoutRow, ...],
cases_by_id: dict[str, dict[str, Any]],
*,
top: int | None = None,
) -> tuple[LiftRecommendation, ...]:
lift_rows = [
r
for r in rows
if r.served_status == "refused" and r.aggressive_status == "correct"
]
groups: dict[tuple[str, str, str | None, str], list[SealedAttemptScoutRow]] = {}
for row in lift_rows:
raw = cases_by_id[row.case_id]
op_class = classify_operation(raw.get("answer_expression", ""))
served_bucket = _classify_reason(row.refusal_reason or "")
category = _extract_category(row.refusal_reason or "")
key = (row.failure_family, served_bucket, category, op_class)
groups.setdefault(key, []).append(row)
recs: list[LiftRecommendation] = []
for (failure_family, bucket, category, op_class), grouped in groups.items():
case_ids = tuple(sorted(r.case_id for r in grouped))
primitive = (
_PRIMITIVE_BY_NO_INJ_CATEGORY.get(category or "", "diagnostic_hold")
if category
else "diagnostic_hold"
)
movement = (
"downstream_reclassification"
if bucket == "recognized_no_injection" and category
else "diagnostic_only"
)
action = (
f"Injector/recognizer gap for category={category}: sealed resolve_pooled "
f"commits correctly on {len(grouped)} train_sample cases; pursue targeted "
f"injector lift — never wire resolve_pooled wholesale to serving."
if category
else (
f"Serving refused but sealed correct on {len(grouped)} cases "
f"({failure_family}); pursue narrow family lift with confusers."
)
)
recs.append(
LiftRecommendation(
rank=0,
failure_family=failure_family,
serving_bucket=bucket,
serving_no_injection_category=category,
operation_class=op_class,
lift_count=len(grouped),
case_ids=case_ids,
candidate_primitive=primitive,
expected_movement=movement,
safe_next_action=action,
)
)
recs.sort(
key=lambda rec: (
-rec.lift_count,
_BUCKET_PRIORITY.get(rec.serving_bucket, 99),
rec.failure_family,
rec.serving_no_injection_category or "",
rec.operation_class,
)
)
ranked = tuple(
LiftRecommendation(
rank=idx,
failure_family=rec.failure_family,
serving_bucket=rec.serving_bucket,
serving_no_injection_category=rec.serving_no_injection_category,
operation_class=rec.operation_class,
lift_count=rec.lift_count,
case_ids=rec.case_ids,
candidate_primitive=rec.candidate_primitive,
expected_movement=rec.expected_movement,
safe_next_action=rec.safe_next_action,
)
for idx, rec in enumerate(recs, start=1)
)
if top is not None:
return ranked[:top]
return ranked
def build_scout_summary(
cases: list[dict[str, Any]] | None = None,
*,
cases_source: str = "evals/gsm8k_math/train_sample/v1/cases.jsonl",
serving_scorer: Callable[[dict[str, Any]], CaseOutcome] | None = None,
sealed_scorer: Callable[[dict[str, Any]], CaseOutcome] | None = None,
include_rows: bool = True,
top_recommendations: int | None = None,
) -> dict[str, Any]:
loaded = cases if cases is not None else _load_cases(_CASES_PATH)
rows = build_scout_rows(
loaded, serving_scorer=serving_scorer, sealed_scorer=sealed_scorer
)
cases_by_id = {c["case_id"]: c for c in loaded}
aggregates = _aggregate_counts(rows)
recommendations = build_lift_recommendations(
rows, cases_by_id, top=top_recommendations
)
summary: dict[str, Any] = {
"schema_version": 1,
"adr": "0175",
"regime": "sealed_attempt_scout",
"cases_source": cases_source,
"sample_count": len(loaded),
**aggregates,
"lift_recommendations": [r.as_dict() for r in recommendations],
}
if include_rows:
summary["rows"] = [r.as_dict() for r in rows]
return summary
def render_markdown(summary: dict[str, Any]) -> str:
lines: list[str] = [
"# GSM8K sealed attempt scout (deterministic report)",
"",
f"- sample: {summary['sample_count']} ({summary['cases_source']})",
]
sc = summary["serving_counts"]
ac = summary["sealed_counts"]
lines.append(
f"- serving: correct={sc['correct']} wrong={sc['wrong']} refused={sc['refused']}"
)
lines.append(
f"- sealed (resolve_pooled): correct={ac['correct']} wrong={ac['wrong']} "
f"refused={ac['refused']}"
)
lines.append("")
lines.append("## Cross-regime deltas")
for key, val in summary["delta_counts"].items():
if val:
lines.append(f"- {key}: {val}")
lines.append("")
lines.append("## Top lift recommendations")
for rec in summary.get("lift_recommendations", [])[:5]:
lines.append(
f"- #{rec['rank']} {rec['failure_family']} (n={rec['lift_count']}, "
f"primitive={rec['candidate_primitive']})"
)
lines.append("")
lines.append(
"safe_action: targeted injector lift only; resolve_pooled remains sealed."
)
return "\n".join(lines)
def write_jsonl(rows: list[dict[str, Any]], path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as fh:
for row in rows:
fh.write(json.dumps(row, sort_keys=True) + "\n")

91
scripts/gsm8k_sealed_attempt_scout.py Normal file
View file

@ -0,0 +1,91 @@
#!/usr/bin/env python3
"""Deterministic GSM8K train-sample sealed attempt scout (ADR-0175 S1).
Dual-scores train_sample cases with serving vs sealed resolve_pooled scorers.
Measurement-only — never writes report.json unless caller passes --out.
"""
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.runner import _CASES_PATH, _load_cases
from evals.gsm8k_math.train_sample.v1.scout import (
build_scout_summary,
render_markdown,
write_jsonl,
)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="GSM8K sealed attempt 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(
"--out",
type=Path,
default=None,
help="Optional JSONL output path (never writes repo artifacts by default)",
)
parser.add_argument(
"--json-only",
action="store_true",
help="Skip markdown summary block",
)
parser.add_argument(
"--no-rows",
action="store_true",
help="Omit per-case rows from JSON output",
)
parser.add_argument(
"--top",
type=int,
default=None,
help="Emit only top N lift recommendations",
)
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]
summary = build_scout_summary(
cases,
cases_source=str(args.cases),
include_rows=not args.no_rows,
top_recommendations=args.top,
)
print(json.dumps(summary, indent=2, sort_keys=True))
if not args.json_only:
print("\n---\n")
print(render_markdown(summary))
if args.out is not None:
rows = summary.get("rows", [])
write_jsonl(rows, args.out)
return 0
if __name__ == "__main__":
raise SystemExit(main())

192
tests/test_gsm8k_sealed_attempt_scout.py Normal file
View file

@ -0,0 +1,192 @@
"""Tests for GSM8K train-sample sealed attempt scout (ADR-0175 S1)."""
from __future__ import annotations
import json
from pathlib import Path
from evals.gsm8k_math.runner import CaseOutcome
from evals.gsm8k_math.train_sample.v1.scout import (
SealedAttemptScoutRow,
build_scout_row,
build_scout_summary,
classify_delta_kind,
classify_failure_family,
render_markdown,
score_case_dual,
)
_REPO_ROOT = Path(__file__).resolve().parents[1]
_REPORT = _REPO_ROOT / "evals/gsm8k_math/train_sample/v1/report.json"
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 test_delta_kind_partition():
assert classify_delta_kind("correct", "correct") == "already_served"
assert classify_delta_kind("correct", "refused") == "serving_conservative_win"
assert classify_delta_kind("wrong", "correct") == "serving_wrong_sealed_correct"
assert classify_delta_kind("wrong", "wrong") == "serving_wrong_other"
assert classify_delta_kind("refused", "correct") == "lift_refused_to_correct"
assert classify_delta_kind("refused", "wrong") == "elimination_refused_to_wrong"
assert classify_delta_kind("refused", "refused") == "joint_refusal"
def test_failure_family_conservative_defaults():
family = classify_failure_family(
delta_kind="joint_refusal",
served_status="refused",
served_reason="candidate_graph: no admissible candidate for statement",
served_bucket="no_admissible_statement",
served_category=None,
sealed_reason="resolve_pooled: no resolution",
)
assert "joint" in family
def test_lift_candidate_row_fields():
raw = {
"case_id": "gsm8k-train-sample-v1-0001",
"question": "How many apples?",
"answer_numeric": 5,
"answer_expression": "#### 5",
}
served = _outcome(
case_id=raw["case_id"],
outcome="refused",
reason="candidate_graph: recognizer matched but produced no injection (category=discrete_count_statement)",
expected=5.0,
)
sealed = _outcome(
case_id=raw["case_id"],
outcome="correct",
reason="resolve_pooled",
actual=5.0,
expected=5.0,
)
row = build_scout_row(raw, served, sealed)
assert row.served_status == "refused"
assert row.aggressive_status == "correct"
assert row.candidate_lift_family is not None
assert row.trace_key
def test_serving_wrong_boundary_family():
family = classify_failure_family(
delta_kind="serving_wrong_sealed_correct",
served_status="wrong",
served_reason="wrong answer",
served_bucket="wrong",
served_category=None,
sealed_reason="resolve_pooled",
)
assert family == "serving_wrong_boundary"
def test_scout_summary_determinism_small_fixture():
cases = [
{
"case_id": "gsm8k-train-sample-v1-9001",
"question": "Tom has 2 apples. How many apples does Tom have?",
"answer_numeric": 2,
"answer_expression": "#### 2",
}
]
def serving(_adapted: dict) -> CaseOutcome:
return _outcome(
case_id=_adapted["id"],
outcome="refused",
reason="no admissible candidate for question",
expected=float(_adapted["expected_answer"]),
)
def sealed(_adapted: dict) -> CaseOutcome:
return _outcome(
case_id=_adapted["id"],
outcome="refused",
reason="resolve_pooled: no resolution",
expected=float(_adapted["expected_answer"]),
)
a = build_scout_summary(cases, serving_scorer=serving, sealed_scorer=sealed)
b = build_scout_summary(cases, serving_scorer=serving, sealed_scorer=sealed)
assert json.dumps(a, sort_keys=True) == json.dumps(b, sort_keys=True)
def test_markdown_render_is_stable():
summary = {
"sample_count": 1,
"cases_source": "fixture",
"serving_counts": {"correct": 0, "wrong": 0, "refused": 1},
"sealed_counts": {"correct": 0, "wrong": 0, "refused": 1},
"delta_counts": {"joint_refusal": 1},
"lift_recommendations": [],
}
assert render_markdown(summary) == render_markdown(summary)
def test_live_train_sample_serving_wrong_is_zero():
summary = build_scout_summary()
assert summary["serving_counts"]["wrong"] == 0
assert summary["sample_count"] == 50
def test_live_scout_summary_determinism():
a = build_scout_summary(include_rows=False)
b = build_scout_summary(include_rows=False)
assert json.dumps(a, sort_keys=True) == json.dumps(b, sort_keys=True)
def test_report_json_mtime_unchanged_by_scout_import():
before = _REPORT.stat().st_mtime_ns
_ = SealedAttemptScoutRow
after = _REPORT.stat().st_mtime_ns
assert before == after
def test_injected_scorers_without_heavy_reader():
cases = [
{
"case_id": "gsm8k-train-sample-v1-9002",
"question": "A",
"answer_numeric": 10,
"answer_expression": "#### 10",
}
]
def serving(_adapted: dict) -> CaseOutcome:
return _outcome(case_id=_adapted["id"], outcome="refused", expected=10.0)
def sealed(_adapted: dict) -> CaseOutcome:
return _outcome(
case_id=_adapted["id"],
outcome="correct",
actual=10.0,
expected=10.0,
)
served, sealed_out = score_case_dual(
cases[0], serving_scorer=serving, sealed_scorer=sealed
)
assert served.outcome == "refused"
assert sealed_out.outcome == "correct"