"""Shadow evaluation of the memory pre-screen (Jev) from DeerMem extraction records. This is the pre-``enforce`` evidence for ``TypeSafeMemoryPrescreen``: the gates documented in ``backend/docs/MEMORY_IMPROVEMENTS.md`` ("Before ``enforce``") and in ``agents/memory/AGENTS.md``. It is an operator run over **shadow records**, not a CI test: in ``mode: shadow`` the extraction call always runs, so every pre-screen verdict comes with free ground truth — that same batch's extraction outcome. The script makes no network call, reads no credential and prints none; its input is a JSONL of DeerMem ``extraction_callback`` payloads, one per line. What one line must contain (the payload the host already emits) ---------------------------------------------------------------- * ``prescreen`` — ``mode``/``verdict``/``probability``/``skip_threshold``/``model``/ ``cached``/``digest``/``signals``/``message_count``/``batch_chars``/ ``duration_ms``/``fallback_reason`` (the judge record, design §5); * ``success`` and ``token_usage`` — the extraction call's own record; * ``mutations_accepted`` — the pre-screen's "worth remembering" signal, written when the apply ran. ``> 0`` on a ``skip`` verdict means the skip would have lost a memory. Fields the collector must add (absent from the production payload today) ----------------------------------------------------------------------- * ``trivial_only`` (bool) — the batch's turns were all trivial (``filter_trivial``). The mandatory no-network baseline cannot be computed without it; coverage is reported instead of assumed. * ``reviewed`` (bool) and optionally ``review_outcome`` (``worth_remembering`` | ``none_worth_remembering``) — the manual review that §5 gate 3 counts *and* whose verdict that gate reads: the review must confirm none of the reviewed skips was worth remembering, so a reviewed skip whose outcome (or ``mutations_accepted`` counter) says otherwise fails the gate. It is also a second ground-truth witness for a skip. * ``stratum`` (string or list) / ``labels`` (list) — the §5 gate 2 subset, one of ``identity`` / ``preference`` / ``correction``. A human label, because L3 keeps signal-bearing batches out of the network population. * ``cost_usd`` (float, optional) — the record's billed cost, when known. Method, and the things it deliberately refuses to do ---------------------------------------------------- * Every record is classified into exactly one population: **network verdicts** (a request was really sent), **cache hits** (the same ``digest`` reused), **local fallbacks** (L3 / L4 / L5 / L7 / L8 or a failure fallback) and **unjudged** (no ``prescreen`` record was emitted at all: no side was enabled, or the round was forbidden to judge). Only network verdicts are classifier samples; mixing populations would let a cache hit or a local refusal vote on the model's accuracy. * ``saved_call_rate`` (a skip that kept nothing, i.e. the call was really saved) and ``miss_rate`` (a skip that lost a memory) share the scored-skip denominator and therefore sum to 1; ``saved_call_rate_over_all_verdicts`` keeps the network-verdict denominator for the "how many calls did this save" reading. * A record whose outcome is unknown (skip without ``mutations_accepted``, an extraction that failed before the apply) is **censored**: counted, reasoned about and excluded from the rates. It is never dropped to claim a pass. * The no-network heuristic baseline is run over the same scored records as the model, so both rates have one denominator: the baseline skips a batch when ``signals`` is empty and ``trivial_only`` is true. If it already captures most of the saving, a third-party egress path is not justified. §5 gate 4 needs it *and* the recorded tokens and p50/p95 verdict latency: recorded calls alone are not the evidence the gate asks for, so missing any of them is ``INSUFFICIENT`` rather than a pass. A baseline that saves as many calls as the model — which the shipped "everything trivial" sample does — is a ``FAIL``, not an enabling run, and the comparison must cover the scored skips: both sides are the records the baseline can actually score (``trivial_only`` plus a readable signals list), so a run that annotates only a subset is ``INSUFFICIENT`` instead of crediting the model with batches the heuristic never saw. * The miss-rate bound is the exact one-sided **Clopper-Pearson** interval, computed here from the regularized incomplete beta function by bisection. No normal approximation is used anywhere, and the gate reads the *bound*, not the point estimate: a sample whose observed rate is at or below the target while its upper bound is above it is ``INSUFFICIENT`` — the evidence does not establish the target yet, so the answer is to collect more skips, not to enable. * A count field that arrives negative (``mutations_accepted``, ``message_count``, ``batch_chars``, a token count) is treated as absent: a count cannot be negative, and scoring a corrupt one as a saving would tilt every rate toward a pass. * The exit code carries the verdict: ``0`` only when every gate is ``PASS``, ``1`` when any gate is ``FAIL`` or ``INSUFFICIENT`` — so an operator can gate enablement on ``python scripts/eval_memory_prescreen.py --records ... && ``. Run from the ``backend/`` directory:: PYTHONPATH=. uv run python scripts/eval_memory_prescreen.py --records /tmp/shadow.jsonl --json /tmp/prescreen-report.json """ from __future__ import annotations import argparse import json import math import statistics from collections import Counter from collections.abc import Mapping, Sequence from dataclasses import dataclass from pathlib import Path from typing import Any NETWORK = "network" CACHE_HIT = "cache_hit" LOCAL_FALLBACK = "local_fallback" UNJUDGED = "unjudged" #: Every input record lands in exactly one of these. POPULATIONS: tuple[str, ...] = (NETWORK, CACHE_HIT, LOCAL_FALLBACK, UNJUDGED) #: The three populations design §5 defines; ``UNJUDGED`` is the coverage remainder. DESIGN_POPULATIONS: tuple[str, ...] = (NETWORK, CACHE_HIT, LOCAL_FALLBACK) VERDICT_SKIP = "skip" VERDICT_EXTRACT = "extract" PASS = "PASS" FAIL = "FAIL" INSUFFICIENT = "INSUFFICIENT" REVIEW_WORTH_REMEMBERING = "worth_remembering" REVIEW_NONE_WORTH_REMEMBERING = "none_worth_remembering" REVIEW_OUTCOMES: tuple[str, ...] = (REVIEW_WORTH_REMEMBERING, REVIEW_NONE_WORTH_REMEMBERING) #: The §5 gate 2 subset. Labels are collector-supplied human labels. SENSITIVE_LABELS = frozenset({"identity", "preference", "correction"}) #: The coordinator's fallback reason -> the design lock it discharges. FALLBACK_LOCKS: dict[str, str] = { "deterministic_signals": "L3", "emergency_flush": "L4", "over_limit": "L5", "shutdown_drain": "L7", "staleness_or_consolidation": "L8", "request_failed": "failure", "no_verdict": "failure", "disabled": "off", } UNRECORDED_REASON = "(unrecorded)" CENSORED_EXTRACTION_FAILED = "extraction_failed" CENSORED_OUTCOME_NOT_RECORDED = "outcome_not_recorded" DEFAULT_MISS_TARGET = 0.01 DEFAULT_CONFIDENCE = 0.95 DEFAULT_MIN_REVIEWED_SKIPS = 200 _BETA_MAX_ITERATIONS = 300 _BETA_EPSILON = 3e-16 _BETA_TINY = 1e-300 _BETA_BISECTIONS = 200 def _as_count(value: Any) -> int | None: """A count, refusing ``bool`` and negatives. ``True`` is not a count of mutations, and a negative count is a corrupt line: the record must lose the field rather than be scored as a saving (a negative ``mutations_accepted`` used to read as "no loss", which diluted the miss-rate denominator and inflated ``saved_calls``). """ if isinstance(value, bool) or not isinstance(value, int) or value < 0: return None return value def _as_float(value: Any) -> float | None: if isinstance(value, bool) or not isinstance(value, (int, float)): return None number = float(value) return number if math.isfinite(number) else None def _as_bool(value: Any) -> bool | None: return value if isinstance(value, bool) else None def _first_count(mapping: Mapping[str, Any], keys: Sequence[str]) -> int | None: for key in keys: value = _as_count(mapping.get(key)) if value is not None: return value return None def _token_counts(usage: Any) -> tuple[int | None, int | None, int | None]: """Read input / output / total tokens from a LangChain-style ``usage_metadata``.""" if not isinstance(usage, Mapping): return (None, None, None) input_tokens = _first_count(usage, ("input_tokens", "prompt_tokens")) output_tokens = _first_count(usage, ("output_tokens", "completion_tokens")) total = _as_count(usage.get("total_tokens")) if total is None and (input_tokens is not None or output_tokens is not None): total = (input_tokens or 0) + (output_tokens or 0) return (input_tokens, output_tokens, total) def _normalize_review_outcome(value: Any) -> str | None: if not isinstance(value, str): return None text = value.strip().lower() return text if text in REVIEW_OUTCOMES else None def _sensitive_labels(raw: Mapping[str, Any]) -> frozenset[str]: """The record's ``identity`` / ``preference`` / ``correction`` labels, normalised.""" found: set[str] = set() for key in ("stratum", "labels"): value = raw.get(key) if isinstance(value, str): candidates = [value] elif isinstance(value, (list, tuple, set, frozenset)): candidates = [item for item in value if isinstance(item, str)] else: continue for item in candidates: label = item.strip().lower() if label in SENSITIVE_LABELS: found.add(label) return frozenset(found) @dataclass(frozen=True) class Prices: """Operator-supplied list price, in USD per million tokens (``None`` = unpriced).""" input_per_mtok: float | None = None output_per_mtok: float | None = None @property def configured(self) -> bool: return self.input_per_mtok is not None or self.output_per_mtok is not None def estimate(self, input_tokens: int | None, output_tokens: int | None) -> float | None: if not self.configured or (input_tokens is None and output_tokens is None): return None dollars = (input_tokens or 0) * (self.input_per_mtok or 0.0) + (output_tokens or 0) * (self.output_per_mtok or 0.0) return dollars / 1_000_000.0 def as_dict(self) -> dict[str, Any]: return {"input_per_mtok": self.input_per_mtok, "output_per_mtok": self.output_per_mtok} @dataclass(frozen=True) class Record: """One shadow record, reduced to what the evaluation reads.""" index: int population: str verdict: str | None fallback_reason: str | None mode: str | None model: str | None cached: bool | None probability: float | None skip_threshold: float | None digest: str | None duration_ms: float | None signals: frozenset[str] signals_known: bool message_count: int | None batch_chars: int | None success: bool | None mutations_accepted: int | None input_tokens: int | None output_tokens: int | None total_tokens: int | None cost_usd: float | None trivial_only: bool | None reviewed: bool review_outcome: str | None sensitive_labels: frozenset[str] censored: bool censored_reason: str | None @property def skip(self) -> bool: return self.verdict == VERDICT_SKIP @property def sensitive(self) -> bool: return bool(self.sensitive_labels) @property def scorable(self) -> bool: """Whether "was this batch worth remembering?" has an answer at all.""" return self.mutations_accepted is not None or self.review_outcome is not None @property def miss(self) -> bool | None: """Whether a skip here would have lost a memory; ``None`` when not scorable. Either witness marks a miss: the ``mutations_accepted`` counter (design §5's definition) or a human ``review_outcome``. A review saying "not worth remembering" never clears a counter that says otherwise — an enablement gate reads the more conservative of the two. """ if not self.scorable: return None if self.review_outcome == REVIEW_WORTH_REMEMBERING: return True return self.mutations_accepted is not None and self.mutations_accepted > 0 def classify_population(record: Mapping[str, Any]) -> str: """Place one raw record in exactly one population (design §5's table). A usable verdict means an answer was consumed: it is a cache hit when the adapter says so and a network verdict otherwise. No usable verdict means no request produced one — L3 / L4 / L5 / L7 / L8 or a failure fallback. No ``prescreen`` record at all means no side was enabled for this batch, or the round was forbidden to judge: an enabled side always emits its record, even when that record is a fallback. """ prescreen = record.get("prescreen") if not isinstance(prescreen, Mapping): return UNJUDGED verdict = prescreen.get("verdict") if verdict not in (VERDICT_SKIP, VERDICT_EXTRACT): return LOCAL_FALLBACK return CACHE_HIT if prescreen.get("cached") is True else NETWORK def _censored_reason(population: str, mutations_accepted: int | None, review_outcome: str | None, success: bool | None) -> str | None: """Why this sample has no outcome to score, or ``None`` when it has one.""" if population not in (NETWORK, CACHE_HIT): return None if mutations_accepted is not None or review_outcome is not None: return None if success is False: return CENSORED_EXTRACTION_FAILED return CENSORED_OUTCOME_NOT_RECORDED def parse_record(index: int, raw: Mapping[str, Any]) -> Record: """Reduce one payload to the fields the evaluation reads; never raises on shape.""" block = raw.get("prescreen") block = block if isinstance(block, Mapping) else {} population = classify_population(raw) raw_verdict = block.get("verdict") verdict = raw_verdict if raw_verdict in (VERDICT_SKIP, VERDICT_EXTRACT) else None review_outcome = _normalize_review_outcome(raw.get("review_outcome")) mutations_accepted = _as_count(raw.get("mutations_accepted")) success = _as_bool(raw.get("success")) input_tokens, output_tokens, total_tokens = _token_counts(raw.get("token_usage")) raw_signals = block.get("signals") signals_known = isinstance(raw_signals, list) signals = frozenset(str(item) for item in raw_signals) if signals_known else frozenset() digest = block.get("digest") censored_reason = _censored_reason(population, mutations_accepted, review_outcome, success) return Record( index=index, population=population, verdict=verdict, fallback_reason=block.get("fallback_reason") if isinstance(block.get("fallback_reason"), str) else None, mode=block.get("mode") if isinstance(block.get("mode"), str) else None, model=block.get("model") if isinstance(block.get("model"), str) else None, cached=_as_bool(block.get("cached")), probability=_as_float(block.get("probability")), skip_threshold=_as_float(block.get("skip_threshold")), digest=digest if isinstance(digest, str) and digest else None, duration_ms=_as_float(block.get("duration_ms")), signals=signals, signals_known=signals_known, message_count=_as_count(block.get("message_count")), batch_chars=_as_count(block.get("batch_chars")), success=success, mutations_accepted=mutations_accepted, input_tokens=input_tokens, output_tokens=output_tokens, total_tokens=total_tokens, cost_usd=_as_float(raw.get("cost_usd")), trivial_only=_as_bool(raw.get("trivial_only")), reviewed=_as_bool(raw.get("reviewed")) is True or review_outcome is not None, review_outcome=review_outcome, sensitive_labels=_sensitive_labels(raw), censored=censored_reason is not None, censored_reason=censored_reason, ) def _dig(record: Mapping[str, Any], path: str) -> Any: """Resolve a dotted ``--label`` path (``prescreen.mode``) against a raw record.""" current: Any = record for part in path.split("."): if not isinstance(current, Mapping) or part not in current: return None current = current[part] return current def record_matches(record: Mapping[str, Any], filters: Sequence[tuple[str, str]]) -> bool: """Whether every ``--label KEY=VALUE`` filter matches; strings compare as text.""" for path, expected in filters: value = _dig(record, path) if isinstance(value, str): if value != expected: return False continue try: parsed = json.loads(expected) except json.JSONDecodeError: return False if value != parsed: return False return True def parse_label_filter(text: str) -> tuple[str, str]: """``KEY=VALUE`` (dotted keys allowed) for ``--label``; argparse error otherwise.""" key, separator, value = text.partition("=") if not separator or not key.strip(): raise argparse.ArgumentTypeError(f"--label expects KEY=VALUE (e.g. prescreen.mode=shadow), got {text!r}") return key.strip(), value @dataclass(frozen=True) class LoadResult: records: tuple[Record, ...] read: int filtered_out: int malformed_lines: int blank_lines: int def load_records(path: Path, filters: Sequence[tuple[str, str]] = ()) -> LoadResult: """Read the JSONL, apply ``--label`` filtering and parse each line. A line that is not a JSON object is *counted*, not dropped: gate D fails the run rather than silently shrinking the denominator. """ records: list[Record] = [] malformed = 0 blank = 0 read = 0 filtered = 0 for number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1): if not line.strip(): blank += 1 continue try: raw = json.loads(line) except json.JSONDecodeError: malformed += 1 continue if not isinstance(raw, Mapping): malformed += 1 continue read += 1 if filters and not record_matches(raw, filters): filtered += 1 continue records.append(parse_record(number, raw)) return LoadResult(records=tuple(records), read=read, filtered_out=filtered, malformed_lines=malformed, blank_lines=blank) def _rate(numerator: int, denominator: int) -> float | None: return None if denominator <= 0 else numerator / denominator def _percentile(values: Sequence[float], fraction: float) -> float | None: """Nearest-rank quantile: the smallest sample at or above ``fraction`` of the mass.""" if not values: return None ordered = sorted(values) return ordered[max(0, math.ceil(fraction * len(ordered)) - 1)] def _latency(values: Sequence[float]) -> dict[str, Any]: if not values: return {"samples": 0, "p50": None, "p95": None} return {"samples": len(values), "p50": statistics.median(values), "p95": _percentile(values, 0.95)} def _token_totals(rows: Sequence[Record]) -> dict[str, Any]: measured = [record for record in rows if record.input_tokens is not None or record.output_tokens is not None or record.total_tokens is not None] return { "records": len(measured), "input": sum(record.input_tokens or 0 for record in measured), "output": sum(record.output_tokens or 0 for record in measured), "total": sum(record.total_tokens if record.total_tokens is not None else (record.input_tokens or 0) + (record.output_tokens or 0) for record in measured), } def _record_cost(record: Record, prices: Prices) -> float | None: if record.cost_usd is not None: return record.cost_usd return prices.estimate(record.input_tokens, record.output_tokens) def _cost_totals(rows: Sequence[Record], prices: Prices) -> dict[str, Any]: costs = [_record_cost(record, prices) for record in rows] priced = [cost for cost in costs if cost is not None] if prices.configured: pricing = "cli price per million tokens" elif priced: pricing = "record cost_usd" else: pricing = "not configured" return {"total_usd": sum(priced) if priced else None, "priced_records": len(priced), "unpriced_records": len(costs) - len(priced), "pricing": pricing} def summarize_population(records: Sequence[Record], population: str, prices: Prices = Prices()) -> dict[str, Any]: """One population's table row: counts, decision rates, latency, tokens and cost.""" rows = [record for record in records if record.population == population] skips = [record for record in rows if record.skip] extracts = [record for record in rows if record.verdict == VERDICT_EXTRACT] scored_skips = [record for record in skips if record.scorable] saved = [record for record in scored_skips if record.miss is False] misses = [record for record in scored_skips if record.miss is True] reason_rows = [record.fallback_reason or UNRECORDED_REASON for record in rows if record.population == LOCAL_FALLBACK or record.fallback_reason is not None] summary: dict[str, Any] = { "records": len(rows), "censored": sum(1 for record in rows if record.censored), "censored_reasons": dict(Counter(record.censored_reason for record in rows if record.censored_reason)), "verdicts": {"skip": len(skips), "extract": len(extracts), "none": len(rows) - len(skips) - len(extracts)}, "cached": { "true": sum(1 for record in rows if record.cached is True), "false": sum(1 for record in rows if record.cached is False), "unknown": sum(1 for record in rows if record.cached is None), }, "skip_rate": _rate(len(skips), len(rows)), "scored_skips": len(scored_skips), "saved_calls": len(saved), "misses": len(misses), "saved_call_rate": _rate(len(saved), len(scored_skips)), "miss_rate": _rate(len(misses), len(scored_skips)), "saved_call_rate_over_all_verdicts": _rate(len(saved), len(rows)), "missed_digests": [record.digest for record in misses if record.digest], "latency_ms": _latency([record.duration_ms for record in rows if record.duration_ms is not None]), "tokens": _token_totals(rows), "cost": _cost_totals(rows, prices), } if reason_rows: summary["fallback_reasons"] = dict(Counter(reason_rows)) summary["fallback_locks"] = dict(Counter(FALLBACK_LOCKS.get(reason, "unknown") for reason in reason_rows)) return summary def _beta_continued_fraction(a: float, b: float, x: float) -> float: """Lentz's continued fraction for the incomplete beta function (Numerical Recipes).""" qab, qap, qam = a + b, a + 1.0, a - 1.0 c = 1.0 d = 1.0 - qab * x / qap if abs(d) < _BETA_TINY: d = _BETA_TINY d = 1.0 / d h = d for m in range(1, _BETA_MAX_ITERATIONS + 1): m2 = 2 * m numerator = m * (b - m) * x / ((qam + m2) * (a + m2)) d = 1.0 + numerator * d if abs(d) < _BETA_TINY: d = _BETA_TINY c = 1.0 + numerator / c if abs(c) < _BETA_TINY: c = _BETA_TINY d = 1.0 / d h *= d * c numerator = -(a + m) * (qab + m) * x / ((a + m2) * (qap + m2)) d = 1.0 + numerator * d if abs(d) < _BETA_TINY: d = _BETA_TINY c = 1.0 + numerator / c if abs(c) < _BETA_TINY: c = _BETA_TINY d = 1.0 / d delta = d * c h *= delta if abs(delta - 1.0) < _BETA_EPSILON: break return h def _regularized_incomplete_beta(a: float, b: float, x: float) -> float: """``I_x(a, b)``, the regularized incomplete beta function.""" if x <= 0.0: return 0.0 if x >= 1.0: return 1.0 front = math.exp(math.lgamma(a + b) - math.lgamma(a) - math.lgamma(b) + a * math.log(x) + b * math.log1p(-x)) if x < (a + 1.0) / (a + b + 2.0): return front * _beta_continued_fraction(a, b, x) / a return 1.0 - front * _beta_continued_fraction(b, a, 1.0 - x) / b def _beta_quantile(probability: float, a: float, b: float) -> float: """Inverse of :func:`_regularized_incomplete_beta` in ``x`` (bisection; monotone).""" if probability <= 0.0: return 0.0 if probability >= 1.0: return 1.0 low, high = 0.0, 1.0 for _ in range(_BETA_BISECTIONS): middle = (low + high) / 2.0 if _regularized_incomplete_beta(a, b, middle) < probability: low = middle else: high = middle return (low + high) / 2.0 def clopper_pearson_upper(misses: int, skips: int, confidence: float = DEFAULT_CONFIDENCE) -> float | None: """Exact one-sided Clopper-Pearson upper bound for the miss rate. Solves ``P(X <= misses) = 1 - confidence`` for the proportion ``p``, i.e. ``I_{1-p}(skips - misses, misses + 1) = 1 - confidence``, by bisection on the regularized incomplete beta. This is the textbook exact bound: no normal approximation is used anywhere in this script. ``None`` when there is no sample, ``1.0`` when every skip was a miss. """ if skips <= 0: return None if not 0.0 < confidence < 1.0: raise ValueError(f"confidence must be in (0, 1), got {confidence!r}") misses = max(0, min(misses, skips)) if misses >= skips: return 1.0 bound = 1.0 - _beta_quantile(1.0 - confidence, float(skips - misses), float(misses + 1)) return min(1.0, max(0.0, bound)) def baseline_summary(records: Sequence[Record], prices: Prices = Prices()) -> dict[str, Any]: """The mandatory no-network baseline, over the same scored records as the model. A batch is a baseline skip when it carries no deterministic signal and the collector marked it ``trivial_only``. Records without ``trivial_only`` (or without a readable ``signals`` list) cannot be scored by the heuristic: the coverage is reported instead of silently shrinking the denominator. """ evaluated = [record for record in records if record.population == NETWORK] computable = [record for record in evaluated if record.trivial_only is not None and record.signals_known] scored = [record for record in computable if record.scorable] baseline_skips = [record for record in scored if record.trivial_only and not record.signals] baseline_saved = [record for record in baseline_skips if record.miss is False] baseline_misses = [record for record in baseline_skips if record.miss is True] model_skips = [record for record in scored if record.skip] model_saved = [record for record in model_skips if record.miss is False] model_misses = [record for record in model_skips if record.miss is True] baseline_saved_rate = _rate(len(baseline_saved), len(baseline_skips)) model_saved_rate = _rate(len(model_saved), len(model_skips)) return { "rule": "skip when signals is empty and trivial_only is true", "evaluated_verdicts": len(evaluated), "computable": len(computable), "coverage": _rate(len(computable), len(evaluated)), "scored": len(scored), "skips": len(baseline_skips), "saved_calls": len(baseline_saved), "misses": len(baseline_misses), "saved_call_rate": baseline_saved_rate, "miss_rate": _rate(len(baseline_misses), len(baseline_skips)), "model": { "skips": len(model_skips), "saved_calls": len(model_saved), "misses": len(model_misses), "saved_call_rate": model_saved_rate, "miss_rate": _rate(len(model_misses), len(model_skips)), }, "incremental_saved_calls": len(model_saved) - len(baseline_saved), # Difference of saved calls over the shared denominator. Subtracting the # conditional rates (saved/skips) would report zero whenever both policies # saved every batch they skipped, even if the model skipped far more. "incremental_saved_call_rate": _rate(len(model_saved) - len(baseline_saved), len(scored)), "trivial_only_missing": sum(1 for record in evaluated if record.trivial_only is None), "signals_missing": sum(1 for record in evaluated if not record.signals_known), "note": "Both sides are scored over the same scored records; a baseline that nearly matches the model means the third-party egress buys little.", "cost": _cost_totals(scored, prices), } @dataclass(frozen=True) class Gate: gate_id: str status: str detail: str def as_dict(self) -> dict[str, str]: return {"id": self.gate_id, "status": self.status, "detail": self.detail} @dataclass(frozen=True) class Evidence: """The numbers the design §5 gates read (network population only).""" scored_skips: int misses: int saved_calls: int sensitive_scored_skips: Mapping[str, int] sensitive_misses: Mapping[str, int] reviewed_skips: int reviewed_scored_skips: int reviewed_missed: int saved_skips_with_tokens: int latency_samples: int #: The no-network comparison, all three over the same records: the ones the #: baseline can score (``trivial_only`` and a readable signals list). Counting #: the model's savings over *all* scorable skips instead would credit it with #: batches the baseline never saw. baseline_computable: int baseline_paired_skips: int baseline_model_saved_calls: int baseline_heuristic_saved_calls: int @dataclass(frozen=True) class Coverage: """Accounting for the records that could not be scored.""" records: int censored: int malformed_lines: int @dataclass(frozen=True) class Thresholds: miss_target: float = DEFAULT_MISS_TARGET confidence: float = DEFAULT_CONFIDENCE min_reviewed_skips: int = DEFAULT_MIN_REVIEWED_SKIPS def evaluate_gates(evidence: Evidence, coverage: Coverage, thresholds: Thresholds = Thresholds()) -> tuple[Gate, ...]: """Every gate prints ``PASS`` / ``FAIL`` / ``INSUFFICIENT`` — never silence.""" gates: list[Gate] = [] miss_rate = _rate(evidence.misses, evidence.scored_skips) upper = clopper_pearson_upper(evidence.misses, evidence.scored_skips, thresholds.confidence) observed = f"miss_rate={evidence.misses}/{evidence.scored_skips}={miss_rate:.4f}" if miss_rate is not None else "" bound = "no bound (no sample)" if upper is None else f"one-sided {thresholds.confidence:.0%} Clopper-Pearson upper bound {upper:.4f}" sample = f"sample n={evidence.scored_skips}" if miss_rate is None: gates.append(Gate("network_miss_rate", INSUFFICIENT, "no scored skip verdict in the network population; there is no sample for the miss rate or its bound")) elif miss_rate > thresholds.miss_target: # The observation itself is over target: more samples cannot rescue it. gates.append(Gate("network_miss_rate", FAIL, f"{observed} is above the target {thresholds.miss_target:.4f}; {bound} ({sample})")) elif upper is None or upper > thresholds.miss_target: # The point estimate meets the target and the exact bound does not establish # it: the sample is underpowered, so the honest answer is more skips. gates.append(Gate("network_miss_rate", INSUFFICIENT, f"{observed} meets the target {thresholds.miss_target:.4f}, but {bound} does not ({sample}): too few skips to establish the target")) else: gates.append(Gate("network_miss_rate", PASS, f"{observed} target<={thresholds.miss_target:.4f}; {bound} ({sample})")) missed_by_label = {label: evidence.sensitive_misses.get(label, 0) for label in sorted(SENSITIVE_LABELS)} uncovered = [label for label in sorted(SENSITIVE_LABELS) if evidence.sensitive_scored_skips.get(label, 0) == 0] per_label = "; ".join(f"{label}: {missed_by_label[label]} miss(es) over {evidence.sensitive_scored_skips.get(label, 0)} scored skip(s)" for label in sorted(SENSITIVE_LABELS)) if any(count > 0 for count in missed_by_label.values()): gates.append(Gate("sensitive_stratum_zero_misses", FAIL, f"a scored skip in a required stratum was worth remembering ({per_label})")) elif uncovered: gates.append(Gate("sensitive_stratum_zero_misses", INSUFFICIENT, f"no scored skip carries the {uncovered} stratum label(s); the zero-miss check needs evidence for every required stratum {sorted(SENSITIVE_LABELS)} ({per_label})")) else: gates.append(Gate("sensitive_stratum_zero_misses", PASS, f"zero misses in every required stratum ({per_label})")) if evidence.reviewed_skips == 0: gates.append(Gate("reviewed_skips_confirm_no_loss", INSUFFICIENT, f"no skip verdict is labelled reviewed; the collector must set reviewed=true / review_outcome (target >={thresholds.min_reviewed_skips})")) elif evidence.reviewed_missed: gates.append(Gate("reviewed_skips_confirm_no_loss", FAIL, f"{evidence.reviewed_missed} reviewed skip verdict(s) were worth remembering; design §5 gate 3 is a review *confirming* none was, so the review sample contradicts the skip")) elif evidence.reviewed_skips < thresholds.min_reviewed_skips: gates.append(Gate("reviewed_skips_confirm_no_loss", FAIL, f"{evidence.reviewed_skips} reviewed skip verdict(s) of {thresholds.min_reviewed_skips} required ({evidence.reviewed_scored_skips} of them scored)")) elif evidence.reviewed_scored_skips < evidence.reviewed_skips: gates.append( Gate( "reviewed_skips_confirm_no_loss", INSUFFICIENT, f"{evidence.reviewed_skips - evidence.reviewed_scored_skips} of {evidence.reviewed_skips} reviewed skip(s) carry neither a review_outcome nor mutations_accepted, so the review confirms nothing for them", ) ) else: gates.append(Gate("reviewed_skips_confirm_no_loss", PASS, f"{evidence.reviewed_skips} reviewed skip verdict(s), all scored, none worth remembering ({thresholds.min_reviewed_skips} required)")) if coverage.records == 0: gates.append(Gate("censoring_disclosed", INSUFFICIENT, "no record was read; there is nothing to account for")) elif coverage.malformed_lines > 0: gates.append(Gate("censoring_disclosed", FAIL, f"{coverage.malformed_lines} input line(s) were not parseable JSON objects; they were counted but must not be dropped from the denominator")) else: gates.append(Gate("censoring_disclosed", PASS, f"{coverage.censored} censored sample(s) of {coverage.records} reported with their reasons; none dropped")) if evidence.scored_skips == 0: gates.append(Gate("savings_recorded", INSUFFICIENT, "no scored skip verdict, so no saved extraction call is recorded")) elif evidence.saved_calls == 0: gates.append(Gate("savings_recorded", FAIL, f"no scored skip saved an extraction call over {evidence.scored_skips} scored skip(s); the pre-screen as recorded saves nothing")) else: absent = [ name for name, count in ( ("tokens on the saved skips", evidence.saved_skips_with_tokens), ("verdict latency", evidence.latency_samples), ("the no-network baseline (collector field trivial_only)", evidence.baseline_computable), ) if count == 0 ] if absent: gates.append( Gate( "savings_recorded", INSUFFICIENT, f"{evidence.saved_calls} extraction call(s) recorded saved, but the evidence is incomplete: nothing recorded for " + "; ".join(absent) + " - design §5 gate 4 asks for recorded calls/tokens saved, their p50/p95 latency cost, and the mandatory baseline", ) ) else: recorded = f"{evidence.saved_calls} extraction call(s) recorded saved over {evidence.scored_skips} scored skip(s)" observed = f"tokens on {evidence.saved_skips_with_tokens} of them, {evidence.latency_samples} verdict latency sample(s), baseline on {evidence.baseline_computable} record(s)" gates.append(Gate("savings_recorded", PASS, f"{recorded}; {observed}")) # Both sides of this comparison are the baseline-computable records, and the gate # refuses to conclude from a subset of the scored skips: subtracting a subset's # heuristic savings from model savings over all scorable skips would count every # unannotated batch as imaginary incremental benefit. paired = evidence.baseline_paired_skips incremental = evidence.baseline_model_saved_calls - evidence.baseline_heuristic_saved_calls if paired == 0: gates.append( Gate( "baseline_incremental_savings", INSUFFICIENT, "no scored skip carries the baseline annotation (the collector must set trivial_only and signals), so there is nothing to compare the model against", ) ) elif paired < evidence.scored_skips: gates.append( Gate( "baseline_incremental_savings", INSUFFICIENT, f"the baseline can score {paired} of {evidence.scored_skips} scored skip(s); the comparison must cover the scored skips, not the subset the annotation happens to reach", ) ) elif incremental <= 0: saved_summary = f"the model saved {evidence.baseline_model_saved_calls} call(s) against the heuristic's {evidence.baseline_heuristic_saved_calls} on the same {paired} scored skip(s): {incremental:+d} incremental" gates.append(Gate("baseline_incremental_savings", FAIL, f"{saved_summary}, so the egress buys nothing over the heuristic already in the stack")) else: gates.append( Gate( "baseline_incremental_savings", PASS, f"the model saved {incremental} call(s) more than the no-network heuristic on the same {paired} scored skip(s) ({evidence.baseline_model_saved_calls} vs {evidence.baseline_heuristic_saved_calls})", ) ) return tuple(gates) def build_report( records: Sequence[Record], *, source: str, malformed_lines: int = 0, filtered_out: int = 0, read_lines: int = 0, blank_lines: int = 0, thresholds: Thresholds = Thresholds(), prices: Prices = Prices(), label_filters: Sequence[tuple[str, str]] = (), ) -> dict[str, Any]: """Assemble the full report: populations, model, baseline, stratification, gates.""" populations = {name: summarize_population(records, name, prices) for name in POPULATIONS} network = [record for record in records if record.population == NETWORK] scored_skips = [record for record in network if record.skip and record.scorable] sensitive_scored_skips = {label: sum(1 for record in scored_skips if label in record.sensitive_labels) for label in sorted(SENSITIVE_LABELS)} sensitive_misses = {label: sum(1 for record in scored_skips if label in record.sensitive_labels and record.miss is True) for label in sorted(SENSITIVE_LABELS)} reviewed_skips = [record for record in network if record.skip and record.reviewed] saved = [record for record in scored_skips if record.miss is False] baseline = baseline_summary(records, prices) evidence = Evidence( scored_skips=len(scored_skips), misses=sum(1 for record in scored_skips if record.miss is True), saved_calls=len(saved), sensitive_scored_skips=sensitive_scored_skips, sensitive_misses=sensitive_misses, reviewed_skips=len(reviewed_skips), reviewed_scored_skips=sum(1 for record in reviewed_skips if record.scorable), reviewed_missed=sum(1 for record in reviewed_skips if record.miss is True), saved_skips_with_tokens=sum(1 for record in saved if record.total_tokens is not None), latency_samples=sum(1 for record in network if record.duration_ms is not None), baseline_computable=baseline["computable"], baseline_paired_skips=baseline["model"]["skips"], baseline_model_saved_calls=baseline["model"]["saved_calls"], baseline_heuristic_saved_calls=baseline["saved_calls"], ) coverage = Coverage(records=len(records), censored=sum(1 for record in records if record.censored), malformed_lines=malformed_lines) gates = evaluate_gates(evidence, coverage, thresholds) verdicts = populations[NETWORK]["verdicts"] upper = clopper_pearson_upper(evidence.misses, evidence.scored_skips, thresholds.confidence) modes = Counter(record.mode for record in records if record.mode) notes = [ "One record is exactly one population: network verdicts are the only classifier samples; cache hits reuse another verdict and local fallbacks never asked.", "saved_call_rate and miss_rate share the scored-skip denominator (so they sum to 1); saved_call_rate_over_all_verdicts keeps the network-verdict denominator.", "The no-network baseline needs the collector's trivial_only field; its coverage is reported rather than assumed.", "This report embeds no raw record, no conversation text and no credential.", ] if modes.get("enforce"): notes.append("Records with mode=enforce carry no extraction outcome for a skip (no call ran), so such a skip is censored unless the collector records a review_outcome; a miss rate is only meaningful over shadow records.") report: dict[str, Any] = { "script": "eval_memory_prescreen", "source": source, "input": { "records": len(records), "read_lines": read_lines, "filtered_out": filtered_out, "malformed_lines": malformed_lines, "blank_lines": blank_lines, "label_filters": [f"{key}={value}" for key, value in label_filters], }, "thresholds": {"miss_target": thresholds.miss_target, "confidence": thresholds.confidence, "min_reviewed_skips": thresholds.min_reviewed_skips}, "pricing": prices.as_dict(), "populations": populations, "model": { "population": NETWORK, "scope": "network verdicts only: the classifier samples", "modes": dict(modes), "verdicts": verdicts, "skip_rate": populations[NETWORK]["skip_rate"], "scored_skips": evidence.scored_skips, "saved_calls": evidence.saved_calls, "saved_call_rate": populations[NETWORK]["saved_call_rate"], "saved_call_rate_over_all_verdicts": populations[NETWORK]["saved_call_rate_over_all_verdicts"], "misses": evidence.misses, "miss_rate": populations[NETWORK]["miss_rate"], "miss_rate_upper_bound": {"method": "clopper-pearson one-sided exact", "confidence": thresholds.confidence, "value": upper, "misses": evidence.misses, "scored_skips": evidence.scored_skips}, "missed_digests": populations[NETWORK]["missed_digests"], "latency_ms": populations[NETWORK]["latency_ms"], "tokens": populations[NETWORK]["tokens"], "cost": populations[NETWORK]["cost"], }, "baseline_heuristic": baseline, "sensitive_stratum": { "labels": sorted(SENSITIVE_LABELS), "by_label": {label: {"scored_skips": sensitive_scored_skips[label], "misses": sensitive_misses[label]} for label in sorted(SENSITIVE_LABELS)}, "note": ( "Labels come from the collector's stratum/labels field; L3 keeps signal-bearing batches out of the " "network population, so this is a human label. The gate checks every required stratum separately, " "because a pooled sample can hide a stratum with no evidence." ), }, "review": { "reviewed_skips": evidence.reviewed_skips, "reviewed_scored_skips": evidence.reviewed_scored_skips, "reviewed_missed": evidence.reviewed_missed, "review_outcomes": dict(Counter(record.review_outcome for record in network if record.review_outcome)), "note": "reviewed comes from the collector; review_outcome is a second ground-truth witness for a skip. Gate 3 requires the review to confirm none was worth remembering, so a reviewed skip found worth remembering fails it.", }, "censoring": { "records": len(records), "censored": coverage.censored, "reasons": dict(Counter(record.censored_reason for record in records if record.censored_reason)), "malformed_lines": malformed_lines, "note": "A censored sample has no outcome to score; it is counted and reported, never dropped.", }, "collector_contract": { "from the DeerMem extraction_callback payload": "prescreen.{mode,verdict,probability,skip_threshold,model,cached,digest,signals,message_count,batch_chars,duration_ms,fallback_reason}, success, token_usage, mutations_accepted", "must be added by the collector": { "trivial_only": "bool - every turn in the batch was trivial (filter_trivial); the no-network baseline cannot be computed without it", "reviewed": "bool - a human reviewed this skip verdict (design §5 gate 3)", "review_outcome": "worth_remembering | none_worth_remembering (optional; a second ground-truth witness)", "stratum": "identity | preference | correction (or a labels list); design §5 gate 2 subset", "cost_usd": "float (optional) - the record's billed cost, when the collector knows it", }, }, "notes": notes, "gates": [gate.as_dict() for gate in gates], "gate_summary": { "passed": [gate.gate_id for gate in gates if gate.status == PASS], "failed": [gate.gate_id for gate in gates if gate.status == FAIL], "insufficient": [gate.gate_id for gate in gates if gate.status == INSUFFICIENT], "enable_recommended": all(gate.status == PASS for gate in gates), }, } return report def _format_rate(value: float | None) -> str: return "n/a" if value is None else f"{value * 100:.1f}%" def _format_ms(value: float | None) -> str: return "n/a" if value is None else f"{value:.1f}ms" def _format_ratio(value: float | None) -> str: return "n/a" if value is None else f"{value:.4f}" def _format_cost(cost: Mapping[str, Any]) -> str: if cost["total_usd"] is None: return f"not configured ({cost['unpriced_records']} record(s) with no --price-in/--price-out and no record cost_usd)" return f"${cost['total_usd']:.4f} ({cost['pricing']}; {cost['priced_records']} priced / {cost['unpriced_records']} unpriced)" def print_report(report: Mapping[str, Any]) -> None: """Print the populations table, the baseline comparison and every gate verdict.""" input_summary = report["input"] filters = ", ".join(input_summary["label_filters"]) or "none" print(f"\nMemory pre-screen shadow evaluation: {report['source']}") print(f"records: {input_summary['records']} scored; {input_summary['filtered_out']} filtered out; {input_summary['malformed_lines']} malformed line(s); {input_summary['blank_lines']} blank line(s); label filters: {filters}") print("\nPopulations (mutually exclusive, design section 5; only network verdicts are classifier samples):") print(f" {'population':<15}{'n':>5}{'skip':>6}{'extract':>8}{'cached':>7}{'censored':>9}{'p50':>10}{'p95':>10}") for name in POPULATIONS: summary = report["populations"][name] latency = summary["latency_ms"] print(f" {name:<15}{summary['records']:>5}{summary['verdicts']['skip']:>6}{summary['verdicts']['extract']:>8}{summary['cached']['true']:>7}{summary['censored']:>9}{_format_ms(latency['p50']):>10}{_format_ms(latency['p95']):>10}") for name in DESIGN_POPULATIONS: reasons = report["populations"][name].get("fallback_reasons") if reasons: rendered = ", ".join(f"{reason}={count} [{FALLBACK_LOCKS.get(reason, 'unknown')}]" for reason, count in sorted(reasons.items())) print(f" {name} reasons: {rendered}") model = report["model"] verdicts = model["verdicts"] print(f"\nModel verdicts ({model['scope']}; modes {model['modes'] or 'none'}):") print(f" skip_rate {_format_rate(model['skip_rate'])} ({verdicts['skip']} skip / {verdicts['extract']} extract / {verdicts['none']} no verdict)") print(f" scored skips {model['scored_skips']}; saved calls {model['saved_calls']} = {_format_rate(model['saved_call_rate'])} of scored skips ({_format_rate(model['saved_call_rate_over_all_verdicts'])} of all network verdicts)") print(f" missed memories {model['misses']} = miss_rate {_format_rate(model['miss_rate'])} (a skip with mutations_accepted > 0)") bound = model["miss_rate_upper_bound"] print(f" miss-rate upper bound: {_format_ratio(bound['value'])} one-sided at {bound['confidence']:.0%} ({bound['method']}, misses={bound['misses']}, n={bound['scored_skips']})") latency = model["latency_ms"] tokens = model["tokens"] print(f" verdict latency p50 {_format_ms(latency['p50'])} p95 {_format_ms(latency['p95'])} (n={latency['samples']})") print(f" tokens in={tokens['input']} out={tokens['output']} total={tokens['total']} (n={tokens['records']}); cost {_format_cost(model['cost'])}") if model["missed_digests"]: print(f" missed digests: {', '.join(model['missed_digests'])}") baseline = report["baseline_heuristic"] print(f"\nNo-network heuristic baseline ({baseline['rule']}):") coverage = f"{baseline['computable']}/{baseline['evaluated_verdicts']}" print(f" computable on {coverage} network verdicts (coverage {_format_rate(baseline['coverage'])})") print(f" collector field trivial_only missing on {baseline['trivial_only_missing']} record(s); signals missing on {baseline['signals_missing']}") baseline_rates = f"saved {_format_rate(baseline['saved_call_rate'])}, miss {_format_rate(baseline['miss_rate'])}" print(f" baseline: skips {baseline['skips']}, saved calls {baseline['saved_calls']}, misses {baseline['misses']} ({baseline_rates})") model_on_set = baseline["model"] model_rates = f"saved {_format_rate(model_on_set['saved_call_rate'])}, miss {_format_rate(model_on_set['miss_rate'])}" print(f" model on the same {baseline['scored']} scored record(s): skips {model_on_set['skips']}, saved calls {model_on_set['saved_calls']}, misses {model_on_set['misses']} ({model_rates})") print(f" incremental saved calls {baseline['incremental_saved_calls']} ({_format_rate(baseline['incremental_saved_call_rate'])} of the scored records)") sensitive = report["sensitive_stratum"] review = report["review"] censoring = report["censoring"] per_label = "; ".join(f"{label}: {data['misses']} miss(es) over {data['scored_skips']} scored skip(s)" for label, data in sensitive["by_label"].items()) print(f"\nSensitive strata ({', '.join(sensitive['labels'])}): {per_label}.") print(f"Review sample: {review['reviewed_skips']} reviewed skip verdict(s) ({review['reviewed_scored_skips']} scored, {review['reviewed_missed']} found worth remembering); review outcomes: {review['review_outcomes'] or 'none'}.") print(f"Censoring: {censoring['censored']} censored of {censoring['records']} record(s) {censoring['reasons'] or ''}; malformed lines {censoring['malformed_lines']}.") print("\nGates before enforce (design section 5):") for gate in report["gates"]: print(f" [{gate['status']:<12}] {gate['id']}: {gate['detail']}") summary = report["gate_summary"] if summary["enable_recommended"]: print("\nEvaluation gates PASSED: every gate is PASS; enforce may be enabled for a controlled window.") else: print(f"\nDO NOT ENABLE: {len(summary['failed'])} gate(s) failed, {len(summary['insufficient'])} gate(s) lack evidence.") if summary["failed"]: print(f" failed: {', '.join(summary['failed'])}") if summary["insufficient"]: print(f" insufficient: {', '.join(summary['insufficient'])}") print("\nNotes:") for note in report["notes"]: print(f" - {note}") def _parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--records", type=Path, required=True, help="JSONL of shadow records: one DeerMem extraction_callback payload per line") parser.add_argument("--label", action="append", type=parse_label_filter, default=[], metavar="KEY=VALUE", help="keep only records whose (dotted) field matches; repeatable, all must match") parser.add_argument("--json", type=Path, default=None, help="write the full report as JSON to this path") parser.add_argument("--miss-target", type=float, default=DEFAULT_MISS_TARGET, help="miss-rate point estimate the gate accepts (default: %(default)s)") parser.add_argument("--confidence", type=float, default=DEFAULT_CONFIDENCE, help="confidence for the Clopper-Pearson upper bound (default: %(default)s)") parser.add_argument("--min-reviewed-skips", type=int, default=DEFAULT_MIN_REVIEWED_SKIPS, help="reviewed skip verdicts required before enforce (default: %(default)s)") parser.add_argument("--price-in", type=float, default=None, help="Jev input price in USD per million tokens (optional; enables the cost estimate)") parser.add_argument("--price-out", type=float, default=None, help="Jev output price in USD per million tokens (optional; enables the cost estimate)") return parser.parse_args(argv) def main(argv: Sequence[str] | None = None) -> int: """Run the evaluation; the exit code is the gate verdict (0 only when all PASS).""" args = _parse_args(argv) loaded = load_records(args.records, args.label) thresholds = Thresholds(miss_target=args.miss_target, confidence=args.confidence, min_reviewed_skips=args.min_reviewed_skips) prices = Prices(input_per_mtok=args.price_in, output_per_mtok=args.price_out) report = build_report( loaded.records, source=str(args.records), malformed_lines=loaded.malformed_lines, filtered_out=loaded.filtered_out, read_lines=loaded.read, blank_lines=loaded.blank_lines, thresholds=thresholds, prices=prices, label_filters=args.label, ) print_report(report) if args.json is not None: args.json.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") print(f"\nJSON report: {args.json}") return 0 if report["gate_summary"]["enable_recommended"] else 1 if __name__ == "__main__": raise SystemExit(main())