feat(llm): test repetition-structured supervision

llm/build_repetition_training_mix.py

@@ -0,0 +1,446 @@
+#!/usr/bin/env python3
+"""Build a fixed-user training arm with shared reply supervision."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+from collections import Counter, defaultdict
+from pathlib import Path
+from typing import Any
+
+try:
+    from .training_data import load_emotion_ids, load_examples, sha256_file
+except ImportError:
+    from training_data import load_emotion_ids, load_examples, sha256_file
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parent
+DEFAULT_EMOTIONS = ROOT / "schema" / "emotions.json"
+TOKEN_RE = re.compile(r"[a-z0-9]+(?:'[a-z0-9]+)?", re.IGNORECASE)
+REQUIRED_CONTRACT_FIELDS = (
+    "id",
+    "intent",
+    "subtype",
+    "user",
+    "emotion",
+    "reply",
+    "safety",
+)
+
+
+def stable_json(value: Any) -> str:
+    return json.dumps(value, separators=(",", ":"), sort_keys=True)
+
+
+def atomic_json(path: Path, value: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
+    temporary.replace(path)
+
+
+def atomic_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    with temporary.open("w") as output:
+        for row in rows:
+            output.write(stable_json(row) + "\n")
+    temporary.replace(path)
+
+
+def normalized_text(text: str) -> str:
+    return " ".join(text.casefold().split())
+
+
+def reply_tokens(text: str) -> frozenset[str]:
+    return frozenset(token.casefold() for token in TOKEN_RE.findall(text))
+
+
+def token_distance(first: frozenset[str], second: frozenset[str]) -> float:
+    union = first | second
+    if not union:
+        return 0.0
+    return 1.0 - len(first & second) / len(union)
+
+
+def load_emotion_profiles(path: Path) -> dict[str, dict[str, Any]]:
+    payload = json.loads(path.read_text())
+    profiles = {}
+    if not isinstance(payload, list):
+        raise ValueError(f"{path}: expected a JSON list")
+    for item in payload:
+        if not isinstance(item, dict):
+            raise ValueError(f"{path}: every emotion profile must be an object")
+        emotion = item.get("id")
+        family = item.get("family")
+        if not isinstance(emotion, str) or not emotion:
+            raise ValueError(f"{path}: every emotion profile needs an id")
+        if not isinstance(family, str) or not family:
+            raise ValueError(f"{path}: {emotion} needs a family")
+        profiles[emotion] = item
+    if not profiles:
+        raise ValueError(f"{path}: no emotion profiles")
+    return profiles
+
+
+def load_contract_rows(path: Path, emotion_ids: set[str]) -> list[dict[str, Any]]:
+    rows = []
+    seen_ids = set()
+    with path.open() as source:
+        for line_number, line in enumerate(source, 1):
+            if not line.strip():
+                continue
+            try:
+                row = json.loads(line)
+            except json.JSONDecodeError as error:
+                raise ValueError(f"{path}:{line_number}: invalid JSON: {error}") from error
+            if not isinstance(row, dict):
+                raise ValueError(f"{path}:{line_number}: expected an object")
+            missing = [field for field in REQUIRED_CONTRACT_FIELDS if field not in row]
+            if missing:
+                raise ValueError(f"{path}:{line_number}: missing fields {missing}")
+            for field in REQUIRED_CONTRACT_FIELDS[:-1]:
+                if not isinstance(row[field], str) or not row[field].strip():
+                    raise ValueError(
+                        f"{path}:{line_number}: {field} must be a non-empty string"
+                    )
+            if not isinstance(row["safety"], bool):
+                raise ValueError(f"{path}:{line_number}: safety must be boolean")
+            if row["emotion"] not in emotion_ids:
+                raise ValueError(
+                    f"{path}:{line_number}: unknown emotion {row['emotion']!r}"
+                )
+            if row["id"] in seen_ids:
+                raise ValueError(f"{path}:{line_number}: repeated id {row['id']!r}")
+            seen_ids.add(row["id"])
+            rows.append({**row, "source_index": len(rows)})
+    if not rows:
+        raise ValueError(f"{path}: no rows")
+    return rows
+
+
+def select_patterns(rows: list[dict[str, Any]], bank_size: int) -> list[dict[str, Any]]:
+    if bank_size < 1:
+        raise ValueError("bank_size must be positive")
+    if len(rows) < bank_size:
+        raise ValueError(
+            f"condition has {len(rows)} rows, fewer than bank size {bank_size}"
+        )
+    ordered = sorted(rows, key=lambda row: row["id"])
+    tokens = {row["id"]: reply_tokens(row["reply"]) for row in ordered}
+    distances: dict[tuple[str, str], float] = {}
+
+    def distance(first: dict[str, Any], second: dict[str, Any]) -> float:
+        key = tuple(sorted((first["id"], second["id"])))
+        if key not in distances:
+            distances[key] = token_distance(tokens[first["id"]], tokens[second["id"]])
+        return distances[key]
+
+    first = min(
+        ordered,
+        key=lambda candidate: (
+            sum(distance(candidate, other) for other in ordered),
+            candidate["id"],
+        ),
+    )
+    selected = [first]
+    selected_ids = {first["id"]}
+    while len(selected) < bank_size:
+        scores = [
+            (
+                min(distance(candidate, pattern) for pattern in selected),
+                candidate,
+            )
+            for candidate in ordered
+            if candidate["id"] not in selected_ids
+        ]
+        maximum = max(score for score, _ in scores)
+        chosen = min(
+            (candidate for score, candidate in scores if score == maximum),
+            key=lambda row: row["id"],
+        )
+        selected.append(chosen)
+        selected_ids.add(chosen["id"])
+    return sorted(selected, key=lambda row: row["id"])
+
+
+def assign_patterns(
+    rows: list[dict[str, Any]], patterns: list[dict[str, Any]]
+) -> dict[str, dict[str, Any]]:
+    ordered_rows = sorted(rows, key=lambda row: row["id"])
+    ordered_patterns = sorted(patterns, key=lambda row: row["id"])
+    quotient, remainder = divmod(len(ordered_rows), len(ordered_patterns))
+    capacity = {
+        pattern["id"]: quotient + (index < remainder)
+        for index, pattern in enumerate(ordered_patterns)
+    }
+    assigned: dict[str, dict[str, Any]] = {}
+    for pattern in ordered_patterns:
+        assigned[pattern["id"]] = pattern
+        capacity[pattern["id"]] -= 1
+    pairs = []
+    for row in ordered_rows:
+        if row["id"] in assigned:
+            continue
+        row_tokens = reply_tokens(row["reply"])
+        for pattern in ordered_patterns:
+            pairs.append(
+                (
+                    token_distance(row_tokens, reply_tokens(pattern["reply"])),
+                    row["id"],
+                    pattern["id"],
+                    row,
+                    pattern,
+                )
+            )
+    for _, row_id, pattern_id, row, pattern in sorted(
+        pairs, key=lambda item: item[:3]
+    ):
+        if row_id in assigned or capacity[pattern_id] == 0:
+            continue
+        assigned[row_id] = pattern
+        capacity[pattern_id] -= 1
+    if len(assigned) != len(rows) or any(capacity.values()):
+        raise RuntimeError("balanced reply-pattern assignment did not complete")
+    return assigned
+
+
+def build_repetition_mix(
+    base_path: Path,
+    contract_path: Path,
+    audit_path: Path,
+    output_path: Path,
+    transformed_path: Path,
+    patterns_path: Path,
+    manifest_path: Path,
+    emotions_path: Path,
+    bank_size: int,
+) -> dict[str, Any]:
+    for path in (output_path, transformed_path, patterns_path, manifest_path):
+        if path.exists():
+            raise RuntimeError(f"output already exists: {path}")
+    emotion_ids = set(load_emotion_ids(emotions_path))
+    profiles = load_emotion_profiles(emotions_path)
+    base = load_examples(base_path, emotion_ids)
+    contract = load_contract_rows(contract_path, emotion_ids)
+    audit = json.loads(audit_path.read_text())
+    excluded_indices = {int(match["row_index"]) for match in audit.get("near_matches", [])}
+    if any(index < 0 or index >= len(contract) for index in excluded_indices):
+        raise RuntimeError("similarity audit contains an invalid contract row index")
+    selected = [
+        row for row in contract if row["source_index"] not in excluded_indices
+    ]
+    groups: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list)
+    for row in selected:
+        groups[(row["intent"], row["subtype"])].append(row)
+
+    transformed_by_index: dict[int, dict[str, Any]] = {}
+    pattern_rows = []
+    condition_rows = []
+    changed_emotions = 0
+    for condition in sorted(groups):
+        rows = groups[condition]
+        safety_values = {row["safety"] for row in rows}
+        if len(safety_values) != 1:
+            raise RuntimeError(f"condition {condition} mixes safety roles")
+        patterns = select_patterns(rows, bank_size)
+        assignments = assign_patterns(rows, patterns)
+        uses = Counter(pattern["id"] for pattern in assignments.values())
+        for pattern in patterns:
+            pattern_rows.append(
+                {
+                    "pattern_id": pattern["id"],
+                    "intent": pattern["intent"],
+                    "subtype": pattern["subtype"],
+                    "emotion": pattern["emotion"],
+                    "emotion_family": profiles[pattern["emotion"]]["family"],
+                    "reply": pattern["reply"],
+                    "source_user": pattern["user"],
+                    "uses": uses[pattern["id"]],
+                }
+            )
+        condition_rows.append(
+            {
+                "intent": condition[0],
+                "subtype": condition[1],
+                "rows": len(rows),
+                "patterns": len(patterns),
+                "minimum_pattern_uses": min(uses.values()),
+                "maximum_pattern_uses": max(uses.values()),
+                "emotion_count": len({pattern["emotion"] for pattern in patterns}),
+                "emotion_family_count": len(
+                    {profiles[pattern["emotion"]]["family"] for pattern in patterns}
+                ),
+            }
+        )
+        for row in rows:
+            pattern = assignments[row["id"]]
+            changed_emotions += row["emotion"] != pattern["emotion"]
+            transformed_by_index[row["source_index"]] = {
+                "id": row["id"],
+                "user": row["user"],
+                "emotion": pattern["emotion"],
+                "reply": pattern["reply"],
+                "intent": row["intent"],
+                "subtype": row["subtype"],
+                "safety": row["safety"],
+                "pattern_id": pattern["id"],
+                "source_emotion": row["emotion"],
+            }
+
+    transformed = [
+        transformed_by_index[index]
+        for index in sorted(transformed_by_index)
+    ]
+    if [normalized_text(row["user"]) for row in selected] != [
+        normalized_text(row["user"]) for row in transformed
+    ]:
+        raise RuntimeError("transformation changed user messages or row order")
+    pattern_pairs = {
+        (row["pattern_id"], row["emotion"], normalized_text(row["reply"]))
+        for row in pattern_rows
+    }
+    if any(
+        (row["pattern_id"], row["emotion"], normalized_text(row["reply"]))
+        not in pattern_pairs
+        for row in transformed
+    ):
+        raise RuntimeError("transformed row does not use a registered reply pattern")
+
+    training_rows = [
+        *base,
+        *(
+            {
+                "user": row["user"],
+                "emotion": row["emotion"],
+                "reply": row["reply"],
+            }
+            for row in transformed
+        ),
+    ]
+    users = [normalized_text(row["user"]) for row in training_rows]
+    if len(users) != len(set(users)):
+        raise RuntimeError("training mix contains repeated user messages")
+
+    atomic_jsonl(transformed_path, transformed)
+    atomic_jsonl(patterns_path, pattern_rows)
+    atomic_jsonl(output_path, training_rows)
+    use_counts = Counter(row["pattern_id"] for row in transformed)
+    manifest = {
+        "schema": "microquack.repetition-training-mix.v1",
+        "dataset_role": "development_training",
+        "bank_size": bank_size,
+        "base": {
+            "path": str(base_path.resolve()),
+            "sha256": sha256_file(base_path),
+            "rows": len(base),
+        },
+        "contract": {
+            "path": str(contract_path.resolve()),
+            "sha256": sha256_file(contract_path),
+            "rows": len(contract),
+            "selected_rows": len(selected),
+        },
+        "held_out_similarity_audit": {
+            "path": str(audit_path.resolve()),
+            "sha256": sha256_file(audit_path),
+            "excluded_rows": len(excluded_indices),
+            "excluded_indices": sorted(excluded_indices),
+        },
+        "transformation": {
+            "reply_assignment": "balanced nearest selected pattern within intent/subtype",
+            "emotion_assignment": "selected reply pattern source label",
+            "conditions": len(groups),
+            "patterns": len(pattern_rows),
+            "unique_replies": len(
+                {normalized_text(row["reply"]) for row in transformed}
+            ),
+            "changed_emotion_rows": changed_emotions,
+            "minimum_pattern_uses": min(use_counts.values()),
+            "maximum_pattern_uses": max(use_counts.values()),
+            "pattern_use_counts": dict(sorted(Counter(use_counts.values()).items())),
+            "condition_summary_sha256": hashlib.sha256(
+                stable_json(condition_rows).encode()
+            ).hexdigest(),
+            "conditions_detail": condition_rows,
+        },
+        "outputs": {
+            "training": {
+                "path": str(output_path.resolve()),
+                "sha256": sha256_file(output_path),
+                "rows": len(training_rows),
+                "unique_users": len(set(users)),
+                "unique_replies": len(
+                    {normalized_text(row["reply"]) for row in training_rows}
+                ),
+                "emotion_counts": dict(
+                    sorted(Counter(row["emotion"] for row in training_rows).items())
+                ),
+            },
+            "transformed_contract": {
+                "path": str(transformed_path.resolve()),
+                "sha256": sha256_file(transformed_path),
+                "rows": len(transformed),
+            },
+            "patterns": {
+                "path": str(patterns_path.resolve()),
+                "sha256": sha256_file(patterns_path),
+                "rows": len(pattern_rows),
+            },
+        },
+        "emotions": {
+            "path": str(emotions_path.resolve()),
+            "sha256": sha256_file(emotions_path),
+            "count": len(emotion_ids),
+        },
+        "boundaries": {
+            "same_selected_user_messages": True,
+            "runtime_reply_table": False,
+            "new_teacher_generation": False,
+            "validation_in_training": False,
+            "sealed_test_used": False,
+            "chat_suites_used_for_training": False,
+        },
+    }
+    atomic_json(manifest_path, manifest)
+    return manifest
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--base", type=Path, required=True)
+    parser.add_argument("--contract", type=Path, required=True)
+    parser.add_argument("--similarity-audit", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--transformed-output", type=Path, required=True)
+    parser.add_argument("--patterns-output", type=Path, required=True)
+    parser.add_argument("--manifest", type=Path, required=True)
+    parser.add_argument("--emotions", type=Path, default=DEFAULT_EMOTIONS)
+    parser.add_argument("--bank-size", type=int, default=16)
+    args = parser.parse_args()
+    if args.bank_size < 1:
+        parser.error("--bank-size must be positive")
+    return args
+
+
+def main() -> None:
+    args = parse_args()
+    manifest = build_repetition_mix(
+        args.base.resolve(),
+        args.contract.resolve(),
+        args.similarity_audit.resolve(),
+        args.output.resolve(),
+        args.transformed_output.resolve(),
+        args.patterns_output.resolve(),
+        args.manifest.resolve(),
+        args.emotions.resolve(),
+        args.bank_size,
+    )
+    print(json.dumps(manifest["outputs"]["training"], sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

llm/compare_chat_evals.py

@@ -0,0 +1,281 @@
+#!/usr/bin/env python3
+"""Compare two symmetric chat evaluations with paired case uncertainty."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import random
+from pathlib import Path
+from typing import Any
+
+try:
+    from .training_data import sha256_file
+except ImportError:
+    from training_data import sha256_file
+
+SCHEMA = "microquack.chat-eval-comparison.v1"
+GENERATION_FIELDS = (
+    "decode_mode",
+    "seeds",
+    "reply_temperature",
+    "reply_top_k",
+    "input_token_bias",
+    "max_reply_tokens",
+    "no_repeat_bigram",
+)
+
+
+def load_report(path: Path) -> dict[str, Any]:
+    payload = json.loads(path.read_text())
+    if payload.get("schema") != "microquack.chat-eval-result.v1":
+        raise ValueError(f"{path}: unsupported chat evaluation schema")
+    return payload
+
+
+def case_outcomes(report: dict[str, Any]) -> dict[str, bool]:
+    grouped: dict[str, list[bool]] = {}
+    for row in report["predictions"]:
+        grouped.setdefault(str(row["case_id"]), []).append(bool(row["sensible"]))
+    expected_seeds = len(report["generation"]["seeds"])
+    if not grouped:
+        raise ValueError("evaluation has no predictions")
+    for case_id, outcomes in grouped.items():
+        if len(outcomes) != expected_seeds:
+            raise ValueError(
+                f"{case_id}: expected {expected_seeds} predictions, got {len(outcomes)}"
+            )
+    return {case_id: all(outcomes) for case_id, outcomes in grouped.items()}
+
+
+def percentile(values: list[float], probability: float) -> float:
+    if not values:
+        raise ValueError("cannot find a percentile of an empty sample")
+    position = probability * (len(values) - 1)
+    lower = math.floor(position)
+    upper = math.ceil(position)
+    if lower == upper:
+        return values[lower]
+    weight = position - lower
+    return values[lower] * (1 - weight) + values[upper] * weight
+
+
+def paired_summary(
+    baseline: dict[str, bool],
+    candidate: dict[str, bool],
+    case_ids: list[str],
+    *,
+    bootstrap_samples: int,
+    bootstrap_seed: int,
+    minimum_worthwhile_effect: float,
+) -> dict[str, Any]:
+    if not case_ids:
+        raise ValueError("comparison subset has no cases")
+    missing = [
+        case_id
+        for case_id in case_ids
+        if case_id not in baseline or case_id not in candidate
+    ]
+    if missing:
+        raise ValueError(f"missing case outcomes: {missing[:5]}")
+    baseline_values = [int(baseline[case_id]) for case_id in case_ids]
+    candidate_values = [int(candidate[case_id]) for case_id in case_ids]
+    deltas = [
+        candidate_value - baseline_value
+        for baseline_value, candidate_value in zip(
+            baseline_values, candidate_values, strict=True
+        )
+    ]
+    count = len(case_ids)
+    baseline_count = sum(baseline_values)
+    candidate_count = sum(candidate_values)
+    effect = sum(deltas) / count
+    generator = random.Random(bootstrap_seed)
+    bootstrap_effects = sorted(
+        sum(deltas[generator.randrange(count)] for _ in range(count)) / count
+        for _ in range(bootstrap_samples)
+    )
+    lower = percentile(bootstrap_effects, 0.025)
+    upper = percentile(bootstrap_effects, 0.975)
+    if effect >= minimum_worthwhile_effect and lower > 0:
+        decision = "clear_improvement"
+    elif upper < minimum_worthwhile_effect:
+        decision = "no_worthwhile_improvement"
+    else:
+        decision = "inconclusive"
+    return {
+        "cases": count,
+        "baseline": {
+            "robust_case_count": baseline_count,
+            "robust_case_rate": baseline_count / count,
+        },
+        "candidate": {
+            "robust_case_count": candidate_count,
+            "robust_case_rate": candidate_count / count,
+        },
+        "absolute_effect": effect,
+        "candidate_only_passes": sum(delta == 1 for delta in deltas),
+        "baseline_only_passes": sum(delta == -1 for delta in deltas),
+        "both_pass": sum(
+            baseline_value == candidate_value == 1
+            for baseline_value, candidate_value in zip(
+                baseline_values, candidate_values, strict=True
+            )
+        ),
+        "both_fail": sum(
+            baseline_value == candidate_value == 0
+            for baseline_value, candidate_value in zip(
+                baseline_values, candidate_values, strict=True
+            )
+        ),
+        "paired_bootstrap_95_interval": [lower, upper],
+        "minimum_worthwhile_effect": minimum_worthwhile_effect,
+        "decision": decision,
+    }
+
+
+def overlap_case_ids(report: dict[str, Any]) -> dict[str, list[str]] | None:
+    overlap = report.get("training_overlap")
+    if overlap is None:
+        return None
+    return {
+        name: [str(case_id) for case_id in subset["case_ids"]]
+        for name, subset in overlap["subsets"].items()
+    }
+
+
+def compare_reports(
+    baseline_path: Path,
+    candidate_path: Path,
+    *,
+    bootstrap_samples: int = 20_000,
+    bootstrap_seed: int = 73,
+    minimum_worthwhile_effect: float = 0.10,
+) -> dict[str, Any]:
+    baseline_report = load_report(baseline_path)
+    candidate_report = load_report(candidate_path)
+    if baseline_report["suite"] != candidate_report["suite"]:
+        raise ValueError("evaluation suites do not match")
+    baseline_generation = {
+        field: baseline_report["generation"][field] for field in GENERATION_FIELDS
+    }
+    candidate_generation = {
+        field: candidate_report["generation"][field] for field in GENERATION_FIELDS
+    }
+    if baseline_generation != candidate_generation:
+        raise ValueError("generation settings are not symmetric")
+
+    baseline = case_outcomes(baseline_report)
+    candidate = case_outcomes(candidate_report)
+    if set(baseline) != set(candidate):
+        raise ValueError("evaluation case sets do not match")
+    case_ids = sorted(baseline)
+    overall = paired_summary(
+        baseline,
+        candidate,
+        case_ids,
+        bootstrap_samples=bootstrap_samples,
+        bootstrap_seed=bootstrap_seed,
+        minimum_worthwhile_effect=minimum_worthwhile_effect,
+    )
+
+    baseline_overlap = overlap_case_ids(baseline_report)
+    candidate_overlap = overlap_case_ids(candidate_report)
+    paired_subsets = None
+    if baseline_overlap is not None and candidate_overlap is not None:
+        baseline_seen = set(baseline_overlap["seen"])
+        candidate_seen = set(candidate_overlap["seen"])
+        partitions = {
+            "both_seen": sorted(baseline_seen & candidate_seen),
+            "both_unseen": sorted(
+                set(case_ids) - baseline_seen - candidate_seen
+            ),
+            "baseline_seen_candidate_unseen": sorted(
+                baseline_seen - candidate_seen
+            ),
+            "baseline_unseen_candidate_seen": sorted(
+                candidate_seen - baseline_seen
+            ),
+        }
+        paired_subsets = {
+            name: paired_summary(
+                baseline,
+                candidate,
+                selected,
+                bootstrap_samples=bootstrap_samples,
+                bootstrap_seed=bootstrap_seed,
+                minimum_worthwhile_effect=minimum_worthwhile_effect,
+            )
+            for name, selected in partitions.items()
+            if selected
+        }
+
+    return {
+        "schema": SCHEMA,
+        "baseline": {
+            "report": str(baseline_path.resolve()),
+            "report_sha256": sha256_file(baseline_path),
+            "checkpoint": baseline_report["checkpoint"],
+        },
+        "candidate": {
+            "report": str(candidate_path.resolve()),
+            "report_sha256": sha256_file(candidate_path),
+            "checkpoint": candidate_report["checkpoint"],
+        },
+        "suite": baseline_report["suite"],
+        "generation": baseline_generation,
+        "metric": "robust_case_rate",
+        "uncertainty": {
+            "method": "paired case bootstrap percentile interval",
+            "confidence": 0.95,
+            "samples": bootstrap_samples,
+            "seed": bootstrap_seed,
+        },
+        "overall": overall,
+        "training_overlap": {
+            "baseline": baseline_report.get("training_overlap"),
+            "candidate": candidate_report.get("training_overlap"),
+            "paired_subsets": paired_subsets,
+        },
+    }
+
+
+def atomic_json(path: Path, payload: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    temporary.replace(path)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--baseline", type=Path, required=True)
+    parser.add_argument("--candidate", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--bootstrap-samples", type=int, default=20_000)
+    parser.add_argument("--bootstrap-seed", type=int, default=73)
+    parser.add_argument("--minimum-worthwhile-effect", type=float, default=0.10)
+    args = parser.parse_args()
+    if args.bootstrap_samples < 1:
+        parser.error("--bootstrap-samples must be positive")
+    if not 0 <= args.minimum_worthwhile_effect <= 1:
+        parser.error("--minimum-worthwhile-effect must be between zero and one")
+    return args
+
+
+def main() -> None:
+    args = parse_args()
+    result = compare_reports(
+        args.baseline.resolve(),
+        args.candidate.resolve(),
+        bootstrap_samples=args.bootstrap_samples,
+        bootstrap_seed=args.bootstrap_seed,
+        minimum_worthwhile_effect=args.minimum_worthwhile_effect,
+    )
+    atomic_json(args.output.resolve(), result)
+    print(json.dumps(result["overall"], indent=2, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

llm/evaluate_chat.py

@@ -47,12 +47,25 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--output", type=Path)
     parser.add_argument("--device", default="auto")
     parser.add_argument("--seeds", type=int, nargs="+", default=DEFAULT_SEEDS)
+    parser.add_argument(
+        "--decode-mode",
+        choices=("greedy", "sampled"),
+        default="sampled",
+        help="apply the same greedy or temperature/top-k policy to every model",
+    )
     parser.add_argument("--reply-temperature", type=float, default=0.3)
     parser.add_argument("--reply-top-k", type=int, default=5)
-    parser.add_argument("--input-token-bias", type=float, default=0.5)
+    parser.add_argument("--input-token-bias", type=float, default=0.0)
     parser.add_argument("--max-reply-tokens", type=int, default=16)
     parser.add_argument(
-        "--no-repeat-bigram", action=argparse.BooleanOptionalAction, default=True
+        "--no-repeat-bigram", action=argparse.BooleanOptionalAction, default=False
+    )
+    parser.add_argument(
+        "--training-data",
+        type=Path,
+        action="append",
+        default=[],
+        help="training JSONL used to split evaluation cases into seen and unseen prompts",
     )
     parser.add_argument("--require", action="store_true")
     args = parser.parse_args()
@@ -148,8 +161,33 @@ def has_repeated_bigram(reply: str) -> bool:
     return len(bigrams) != len(set(bigrams))
 
 
+def normalized_text(text: str) -> str:
+    return " ".join(text.casefold().split())
+
+
 def normalize_reply(reply: str) -> str:
-    return " ".join(reply.casefold().split())
+    return normalized_text(reply)
+
+
+def load_training_users(paths: list[Path]) -> set[str]:
+    users = set()
+    for path in paths:
+        with path.open() as source:
+            for line_number, line in enumerate(source, 1):
+                if not line.strip():
+                    continue
+                try:
+                    row = json.loads(line)
+                except json.JSONDecodeError as error:
+                    raise ValueError(
+                        f"{path}:{line_number}: invalid JSON: {error}"
+                    ) from error
+                if not isinstance(row, dict) or not isinstance(row.get("user"), str):
+                    raise ValueError(f"{path}:{line_number}: user must be a string")
+                user = normalized_text(row["user"])
+                if user:
+                    users.add(user)
+    return users
 
 
 def load_checkpoint(
@@ -186,17 +224,28 @@ def generate(
     device: Any,
     args: argparse.Namespace,
 ) -> list[list[tuple[str, str]]]:
+    temperature = 0.0 if args.decode_mode == "greedy" else args.reply_temperature
+    top_k = 0 if args.decode_mode == "greedy" else args.reply_top_k
     if kind == "character":
-        greedy = model.greedy_replies(prompts, device=device, batch_size=128)
-        return [greedy for _ in seeds]
+        return [
+            model.replies(
+                prompts,
+                device=device,
+                batch_size=128,
+                reply_temperature=temperature,
+                reply_top_k=top_k,
+                seed=seed,
+            )
+            for seed in seeds
+        ]
     return [
         model.replies(
             prompts,
             tokenizer,
             device=device,
             batch_size=128,
-            reply_temperature=args.reply_temperature,
-            reply_top_k=args.reply_top_k,
+            reply_temperature=temperature,
+            reply_top_k=top_k,
             seed=seed,
             no_repeat_bigram=args.no_repeat_bigram,
             max_reply_tokens=args.max_reply_tokens,
@@ -290,6 +339,47 @@ def summarize(
     }
 
 
+def summarize_training_overlap(
+    rows: list[dict[str, Any]],
+    suite: dict[str, Any],
+    seeds: list[int],
+    emotion_count: int,
+    training_users: set[str],
+) -> dict[str, Any]:
+    seen_ids = {
+        case["id"]
+        for case in suite["cases"]
+        if normalized_text(case["user"]) in training_users
+    }
+    subsets = {}
+    for name, selected_ids in (
+        ("seen", seen_ids),
+        (
+            "unseen",
+            {case["id"] for case in suite["cases"]} - seen_ids,
+        ),
+    ):
+        selected_cases = [
+            case for case in suite["cases"] if case["id"] in selected_ids
+        ]
+        if selected_cases:
+            selected_rows = [row for row in rows if row["case_id"] in selected_ids]
+            metrics = summarize(
+                selected_rows,
+                {**suite, "cases": selected_cases},
+                seeds,
+                emotion_count,
+            )
+        else:
+            metrics = None
+        subsets[name] = {
+            "cases": len(selected_cases),
+            "case_ids": [case["id"] for case in selected_cases],
+            "metrics": metrics,
+        }
+    return subsets
+
+
 def main() -> None:
     import torch
 
@@ -386,6 +476,18 @@ def main() -> None:
             )
 
     metrics = summarize(rows, suite, args.seeds, len(emotion_ids))
+    training_paths = [path.resolve() for path in args.training_data]
+    training_overlap = (
+        summarize_training_overlap(
+            rows,
+            suite,
+            args.seeds,
+            len(emotion_ids),
+            load_training_users(training_paths),
+        )
+        if training_paths
+        else None
+    )
     thresholds = suite["thresholds"]
     gates = {
         metric: metrics[metric] is not None and metrics[metric] >= threshold
@@ -415,13 +517,27 @@ def main() -> None:
         "generation": {
             "seeds": args.seeds,
             "device": str(device),
-            "reply_temperature": args.reply_temperature,
-            "reply_top_k": args.reply_top_k,
+            "decode_mode": args.decode_mode,
+            "reply_temperature": (
+                0.0 if args.decode_mode == "greedy" else args.reply_temperature
+            ),
+            "reply_top_k": 0 if args.decode_mode == "greedy" else args.reply_top_k,
             "input_token_bias": args.input_token_bias,
             "max_reply_tokens": args.max_reply_tokens,
             "no_repeat_bigram": args.no_repeat_bigram,
             "elapsed_seconds": elapsed,
         },
+        "training_overlap": (
+            {
+                "data": [
+                    {"path": str(path), "sha256": sha256_file(path)}
+                    for path in training_paths
+                ],
+                "subsets": training_overlap,
+            }
+            if training_overlap is not None
+            else None
+        ),
         "thresholds": thresholds,
         "metrics": metrics,
         "gates": gates,

llm/test_build_repetition_training_mix.py

@@ -0,0 +1,130 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from llm.build_repetition_training_mix import build_repetition_mix
+
+
+class RepetitionTrainingMixTests(unittest.TestCase):
+    def write_inputs(self, root: Path) -> tuple[Path, Path, Path, Path]:
+        emotions = root / "emotions.json"
+        base = root / "base.jsonl"
+        contract = root / "contract.jsonl"
+        audit = root / "audit.json"
+        emotions.write_text(
+            json.dumps(
+                [
+                    {"id": "calm", "family": "positive"},
+                    {"id": "happy", "family": "positive"},
+                    {"id": "worried", "family": "fear"},
+                ]
+            )
+        )
+        base.write_text(
+            json.dumps(
+                {"user": "Base user", "emotion": "calm", "reply": "Base reply"}
+            )
+            + "\n"
+        )
+        rows = []
+        for index, (emotion, reply) in enumerate(
+            (
+                ("happy", "Hello friend"),
+                ("calm", "Hello there"),
+                ("happy", "Greetings friend"),
+                ("calm", "Greetings there"),
+                ("worried", "Please stay safe"),
+                ("worried", "Move somewhere safe"),
+            )
+        ):
+            rows.append(
+                {
+                    "id": f"row-{index}",
+                    "intent": "chat",
+                    "subtype": "greeting" if index < 3 else "safety",
+                    "user": f"User message {index}",
+                    "emotion": emotion,
+                    "reply": reply,
+                    "safety": index >= 3,
+                }
+            )
+        contract.write_text("".join(json.dumps(row) + "\n" for row in rows))
+        audit.write_text(json.dumps({"near_matches": [{"row_index": 2}]}))
+        return base, contract, audit, emotions
+
+    def build(self, root: Path) -> tuple[dict, list[dict], list[dict], list[dict]]:
+        base, contract, audit, emotions = self.write_inputs(root)
+        output = root / "train.jsonl"
+        transformed = root / "transformed.jsonl"
+        patterns = root / "patterns.jsonl"
+        manifest_path = root / "manifest.json"
+        manifest = build_repetition_mix(
+            base,
+            contract,
+            audit,
+            output,
+            transformed,
+            patterns,
+            manifest_path,
+            emotions,
+            bank_size=2,
+        )
+        load = lambda path: [json.loads(line) for line in path.read_text().splitlines()]
+        return manifest, load(output), load(transformed), load(patterns)
+
+    def test_builds_balanced_reply_patterns_without_changing_users(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            manifest, training, transformed, patterns = self.build(Path(directory))
+
+        self.assertEqual(len(training), 6)
+        self.assertEqual(
+            [row["user"] for row in transformed],
+            ["User message 0", "User message 1", "User message 3", "User message 4", "User message 5"],
+        )
+        self.assertEqual(len(patterns), 4)
+        self.assertEqual(sorted(row["uses"] for row in patterns), [1, 1, 1, 2])
+        registered = {
+            (row["pattern_id"], row["emotion"], row["reply"]) for row in patterns
+        }
+        self.assertTrue(
+            all(
+                (row["pattern_id"], row["emotion"], row["reply"]) in registered
+                for row in transformed
+            )
+        )
+        self.assertEqual(manifest["held_out_similarity_audit"]["excluded_indices"], [2])
+        self.assertTrue(manifest["boundaries"]["same_selected_user_messages"])
+        self.assertFalse(manifest["boundaries"]["runtime_reply_table"])
+        self.assertLess(manifest["outputs"]["training"]["unique_replies"], 6)
+
+    def test_output_is_deterministic(self) -> None:
+        with tempfile.TemporaryDirectory() as first_directory, tempfile.TemporaryDirectory() as second_directory:
+            first = self.build(Path(first_directory))
+            second = self.build(Path(second_directory))
+        self.assertEqual(first[1:], second[1:])
+        self.assertEqual(
+            first[0]["transformation"]["condition_summary_sha256"],
+            second[0]["transformation"]["condition_summary_sha256"],
+        )
+
+    def test_rejects_a_bank_larger_than_a_condition(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            base, contract, audit, emotions = self.write_inputs(root)
+            with self.assertRaisesRegex(ValueError, "fewer than bank size"):
+                build_repetition_mix(
+                    base,
+                    contract,
+                    audit,
+                    root / "train.jsonl",
+                    root / "transformed.jsonl",
+                    root / "patterns.jsonl",
+                    root / "manifest.json",
+                    emotions,
+                    bank_size=4,
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()

llm/test_compare_chat_evals.py

@@ -0,0 +1,121 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from llm.compare_chat_evals import compare_reports
+
+
+class CompareChatEvalsTests(unittest.TestCase):
+    def write_report(
+        self,
+        path: Path,
+        passes: list[bool],
+        *,
+        decode_mode: str = "greedy",
+    ) -> None:
+        case_ids = [f"case-{index}" for index in range(len(passes))]
+        report = {
+            "schema": "microquack.chat-eval-result.v1",
+            "checkpoint": {
+                "path": str(path.with_suffix(".pt")),
+                "sha256": "checkpoint",
+                "kind": "word",
+            },
+            "suite": {"path": "suite.json", "sha256": "suite", "role": "held_out"},
+            "generation": {
+                "decode_mode": decode_mode,
+                "seeds": [7],
+                "reply_temperature": 0.0,
+                "reply_top_k": 0,
+                "input_token_bias": 0.0,
+                "max_reply_tokens": 16,
+                "no_repeat_bigram": False,
+            },
+            "training_overlap": {
+                "data": [],
+                "subsets": {
+                    "seen": {"case_ids": case_ids[:2]},
+                    "unseen": {"case_ids": case_ids[2:]},
+                },
+            },
+            "predictions": [
+                {"case_id": case_id, "seed": 7, "sensible": passed}
+                for case_id, passed in zip(case_ids, passes, strict=True)
+            ],
+        }
+        path.write_text(json.dumps(report))
+
+    def test_reports_paired_counts_effect_and_subsets(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            baseline = root / "baseline.json"
+            candidate = root / "candidate.json"
+            self.write_report(baseline, [False] * 20)
+            self.write_report(candidate, [True] * 20)
+            result = compare_reports(
+                baseline,
+                candidate,
+                bootstrap_samples=1_000,
+                bootstrap_seed=7,
+            )
+
+        overall = result["overall"]
+        self.assertEqual(overall["baseline"]["robust_case_count"], 0)
+        self.assertEqual(overall["candidate"]["robust_case_count"], 20)
+        self.assertEqual(overall["candidate_only_passes"], 20)
+        self.assertEqual(overall["absolute_effect"], 1.0)
+        self.assertEqual(overall["decision"], "clear_improvement")
+        paired = result["training_overlap"]["paired_subsets"]
+        self.assertEqual(paired["both_seen"]["cases"], 2)
+        self.assertEqual(paired["both_unseen"]["cases"], 18)
+
+    def test_equal_reports_do_not_claim_worthwhile_improvement(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            baseline = root / "baseline.json"
+            candidate = root / "candidate.json"
+            passes = [True, False] * 10
+            self.write_report(baseline, passes)
+            self.write_report(candidate, passes)
+            result = compare_reports(
+                baseline,
+                candidate,
+                bootstrap_samples=100,
+                minimum_worthwhile_effect=0.10,
+            )
+        self.assertEqual(result["overall"]["decision"], "no_worthwhile_improvement")
+
+    def test_compares_different_training_overlap_splits(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            baseline = root / "baseline.json"
+            candidate = root / "candidate.json"
+            self.write_report(baseline, [True, False, True, False])
+            self.write_report(candidate, [True, True, False, False])
+            payload = json.loads(candidate.read_text())
+            payload["training_overlap"]["subsets"] = {
+                "seen": {"case_ids": ["case-1", "case-2"]},
+                "unseen": {"case_ids": ["case-0", "case-3"]},
+            }
+            candidate.write_text(json.dumps(payload))
+            result = compare_reports(baseline, candidate, bootstrap_samples=10)
+        paired = result["training_overlap"]["paired_subsets"]
+        self.assertEqual(paired["both_seen"]["cases"], 1)
+        self.assertEqual(paired["both_unseen"]["cases"], 1)
+        self.assertEqual(paired["baseline_seen_candidate_unseen"]["cases"], 1)
+        self.assertEqual(paired["baseline_unseen_candidate_seen"]["cases"], 1)
+
+    def test_rejects_asymmetric_generation(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            baseline = root / "baseline.json"
+            candidate = root / "candidate.json"
+            self.write_report(baseline, [True, False])
+            self.write_report(candidate, [True, False], decode_mode="sampled")
+            with self.assertRaisesRegex(ValueError, "not symmetric"):
+                compare_reports(baseline, candidate, bootstrap_samples=10)
+
+
+if __name__ == "__main__":
+    unittest.main()

llm/test_evaluate_chat.py

@@ -2,9 +2,11 @@
 
 import json
 import sys
+import tempfile
 import unittest
 from collections import Counter
 from pathlib import Path
+from types import SimpleNamespace
 
 HERE = Path(__file__).resolve().parent
 ROOT = HERE.parent
@@ -12,10 +14,13 @@ sys.path.insert(0, str(HERE))
 
 from evaluate_chat import (  # noqa: E402
     contains_cue,
+    generate,
     has_repeated_bigram,
     has_repeated_word_run,
     load_cases,
+    load_training_users,
     summarize,
+    summarize_training_overlap,
 )
 from generate_common_anchors import make_examples  # noqa: E402
 
@@ -95,6 +100,86 @@ class ChatEvalTests(unittest.TestCase):
         self.assertFalse(has_repeated_word_run("I can help you."))
         self.assertFalse(has_repeated_bigram("I can help you."))
 
+    def test_generate_applies_the_same_sample_policy_to_character_models(self) -> None:
+        class FakeCharacterModel:
+            def __init__(self) -> None:
+                self.calls = []
+
+            def replies(self, prompts, **kwargs):
+                self.calls.append(kwargs)
+                return [("happy", f"seed {kwargs['seed']}") for _ in prompts]
+
+        model = FakeCharacterModel()
+        args = SimpleNamespace(
+            decode_mode="sampled",
+            reply_temperature=0.3,
+            reply_top_k=5,
+            input_token_bias=0.0,
+            no_repeat_bigram=False,
+            max_reply_tokens=16,
+        )
+        batches = generate(
+            "character", model, None, ["hello"], [7, 11], "cpu", args
+        )
+        self.assertEqual([batch[0][1] for batch in batches], ["seed 7", "seed 11"])
+        self.assertEqual([call["reply_temperature"] for call in model.calls], [0.3, 0.3])
+        self.assertEqual([call["reply_top_k"] for call in model.calls], [5, 5])
+
+        args.decode_mode = "greedy"
+        model.calls.clear()
+        generate("character", model, None, ["hello"], [7], "cpu", args)
+        self.assertEqual(model.calls[0]["reply_temperature"], 0.0)
+        self.assertEqual(model.calls[0]["reply_top_k"], 0)
+
+    def test_training_overlap_splits_seen_and_unseen_cases(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "train.jsonl"
+            path.write_text(
+                json.dumps(
+                    {"user": "  Hello   There ", "emotion": "happy", "reply": "Hi."}
+                )
+                + "\n"
+                + json.dumps({"user": "   ", "emotion": "happy", "reply": "Hi."})
+                + "\n"
+            )
+            training_users = load_training_users([path])
+            self.assertEqual(training_users, {"hello there"})
+
+            suite = {
+                "cases": [
+                    {"id": "seen", "user": "hello there", "category": "chat"},
+                    {"id": "unseen", "user": "new prompt", "category": "chat"},
+                ]
+            }
+            rows = []
+            for case_id, user in (("seen", "hello there"), ("unseen", "new prompt")):
+                rows.append(
+                    {
+                        "case_id": case_id,
+                        "category": "chat",
+                        "user": user,
+                        "seed": 7,
+                        "emotion": "happy",
+                        "reply": "Hello.",
+                        "valid_emotion": True,
+                        "well_formed_reply": True,
+                        "emotion_ok": True,
+                        "target_emotion_ok": None,
+                        "reply_cue_ok": True,
+                        "reply_forbidden_ok": True,
+                        "reply_ok": True,
+                        "sensible": True,
+                        "repeated_word_run": False,
+                        "repeated_bigram": False,
+                    }
+                )
+            subsets = summarize_training_overlap(rows, suite, [7], 73, training_users)
+            self.assertEqual(subsets["seen"]["cases"], 1)
+            self.assertEqual(subsets["seen"]["case_ids"], ["seen"])
+            self.assertEqual(subsets["unseen"]["cases"], 1)
+            self.assertEqual(subsets["unseen"]["case_ids"], ["unseen"])
+            self.assertEqual(subsets["seen"]["metrics"]["robust_case_count"], 1)
+
     def test_summary_reports_cross_case_and_seed_diversity(self) -> None:
         suite = {
             "cases": [

package.json

@@ -9,7 +9,7 @@
     "lint": "eslint .",
     "typecheck": "tsc --noEmit -p apps/web/tsconfig.json",
     "test": "vitest run && pnpm run test:teacher-data",
-    "test:teacher-data": "python3 -m unittest llm/test_generate_teacher_dataset.py llm/test_generate_teacher_corpus.py llm/test_generate_intent_pilot.py llm/test_generate_contract_corpus.py llm/test_audit_eval_similarity.py llm/test_build_intent_training_mix.py llm/test_training_data.py llm/test_evaluate_chat.py",
+    "test:teacher-data": "python3 -m unittest llm/test_generate_teacher_dataset.py llm/test_generate_teacher_corpus.py llm/test_generate_intent_pilot.py llm/test_generate_contract_corpus.py llm/test_audit_eval_similarity.py llm/test_build_intent_training_mix.py llm/test_build_repetition_training_mix.py llm/test_training_data.py llm/test_evaluate_chat.py llm/test_compare_chat_evals.py",
     "coverage": "vitest run --coverage",
     "dry": "slophammer-ts dry .",
     "slophammer": "slophammer-ts check .",