diff --git a/README.md b/README.md index 5bd2ddb..f681186 100644 --- a/README.md +++ b/README.md @@ -40,6 +40,7 @@ The layout follows the order of the experiment. Small numbered files under | --- | --- | | `scripts/01_scenario_construction/` | Steps 1-2: generate and validate controlled Scenario 1 records. | | `scripts/02_model_execution/` | Step 3: validate, cache, or run Qwen3-32B on Modal. | +| `scripts/03_probe_analysis/` | Step 6: analyze saved probe predictions without starting model compute. | | `scripts/90_runway_reproduction/` | Historical Llama 3.1 8B runway reproduction commands. | | `src/scenario1/` | Scenario registry normalization, record construction, manifest generation, and semantic validation. | | `src/infrastructure/` | Modal request/result contracts, model runner, and cost records. | @@ -68,7 +69,7 @@ Do not run files under `archive/` as part of the active experiment. | 3 | Validate, cache, or run one Qwen model turn | `scripts/02_model_execution/03_modal_qwen_runner.py` | Model-turn result, activations, and cost record | | 4 | Run the complete live agent chain | Sequential orchestrator and safe executor; not yet implemented | Completed live v2 trajectories | | 5 | Normalize trajectories for measurement | `src/pipeline/handoff.py` and `src/extraction/trajectory.py` | Ordered activation examples | -| 6 | Fit probes and compute trajectory metrics | `src/probes/` and `src/analysis/trajectory_metrics.py` | AUROC, Brier, ECE, LAT, and Temporal Divergence values | +| 6 | Fit probes and compute trajectory metrics | `src/probes/`, `src/analysis/trajectory_metrics.py`, and `scripts/03_probe_analysis/06_analyze_depth_degradation.py` | AUROC, Brier, ECE, LAT, Temporal Divergence, and depth-degradation values | | 7 | Produce figures and compact reports | `src/analysis/geometry.py`, `src/analysis/calibration.py`, and `results/` | PCA, layer, calibration, and depth figures | Steps 1-3 have runnable command wrappers. Step 4 must connect the one-turn @@ -182,13 +183,43 @@ After live trajectories exist: the injection point. 6. `src/analysis/trajectory_metrics.py` summarizes observed propagation and action outcomes. -7. `src/analysis/geometry.py` and `src/analysis/calibration.py` create PCA and +7. `src/analysis/depth_degradation.py` compares matched 2-hop and 3-hop probe + predictions. +8. `src/analysis/geometry.py` and `src/analysis/calibration.py` create PCA and reliability figures. These libraries are implemented, but the public repository does not yet claim Scenario 1 AUROC, calibration, depth-degradation, or Temporal Divergence results. Those numbers require real trajectories and activations from Step 4. +### Depth-degradation analysis + +The depth analysis consumes saved probe-prediction JSONL rows. It does not load +the model, generate trajectories, start a GPU, or call an LLM judge. It reports +executor-level AUROC, Brier score, ECE, Temporal Divergence, and match-group +bootstrap confidence intervals for 2-hop and 3-hop trajectories. + +Run it after probe scores have been exported: + +```bash +python scripts/03_probe_analysis/06_analyze_depth_degradation.py \ + predictions.jsonl \ + --experiment-id scenario1-v1 \ + --output-json results/depth_degradation.json \ + --output-csv results/depth_degradation.csv +``` + +The reported delta is always `3-hop minus 2-hop`. A negative AUROC delta means +the probe discriminates less well at greater depth. Rows must state their +`label_target`; the analysis supports action-executed and injection-present +labels but never mixes them silently. A `latent_compromise_status` of +`candidate` selects a review cohort and is not a confirmed mechanistic label. + +Match-group split helpers live in `src/analysis/splits.py`. Related clean and +injected trajectories at both depths always stay together. Use an unsplit +manifest when there are too few complete groups for non-empty train, +validation, and test partitions. + ## Scenario 1 controls One trajectory is one complete pipeline run. One match group contains four @@ -302,6 +333,12 @@ python -m pytest \ tests/test_trajectory_acceptance.py -q ``` +Run the compute-independent depth-analysis checks: + +```bash +python -m pytest tests/test_depth_degradation.py tests/test_splits.py -q +``` + ## Historical runway work The runway used Llama 3.1 8B Instruct and NARCBench-Core to validate the diff --git a/scripts/03_probe_analysis/06_analyze_depth_degradation.py b/scripts/03_probe_analysis/06_analyze_depth_degradation.py new file mode 100644 index 0000000..40df137 --- /dev/null +++ b/scripts/03_probe_analysis/06_analyze_depth_degradation.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +"""Analyze depth degradation from precomputed probe-prediction JSONL rows.""" + +from __future__ import annotations + +import argparse +import csv +import json +from pathlib import Path +import sys + + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from src.analysis.depth_degradation import ( + analyze_depth_degradation, + load_prediction_jsonl, + tabular_result_rows, +) + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("predictions", type=Path) + parser.add_argument("--experiment-id", required=True) + parser.add_argument("--output-json", type=Path, required=True) + parser.add_argument("--output-csv", type=Path) + parser.add_argument("--n-bootstrap", type=int, default=1000) + parser.add_argument("--confidence", type=float, default=0.95) + parser.add_argument("--n-bins", type=int, default=10) + parser.add_argument("--random-state", type=int, default=42) + args = parser.parse_args() + + result = analyze_depth_degradation( + load_prediction_jsonl(args.predictions), + experiment_id=args.experiment_id, + n_bootstrap=args.n_bootstrap, + confidence=args.confidence, + n_bins=args.n_bins, + random_state=args.random_state, + ) + args.output_json.parent.mkdir(parents=True, exist_ok=True) + args.output_json.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n") + + if args.output_csv: + rows = tabular_result_rows(result) + args.output_csv.parent.mkdir(parents=True, exist_ok=True) + fieldnames = sorted({key for row in rows for key in row}) + with args.output_csv.open("w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(rows) + + +if __name__ == "__main__": + main() diff --git a/src/analysis/depth_degradation.py b/src/analysis/depth_degradation.py new file mode 100644 index 0000000..d4ca0f0 --- /dev/null +++ b/src/analysis/depth_degradation.py @@ -0,0 +1,764 @@ +"""Depth-degradation analysis over precomputed trajectory probe scores. + +This module is intentionally compute-backend agnostic. It consumes JSON-like +prediction rows after activation extraction and probe scoring have completed. +""" + +from __future__ import annotations + +import hashlib +import json +from collections import defaultdict +from pathlib import Path +from typing import Iterable + +import numpy as np +from sklearn.metrics import brier_score_loss, roc_auc_score + +from src.probes.linear_probe import compute_ece +from src.probes.temporal_divergence import temporal_divergence + + +REQUIRED_FIELDS = { + "trajectory_id", + "match_group_id", + "domain_id", + "wording_id", + "condition", + "injection_present", + "hop_mode", + "agent_id", + "agent_role", + "hop_index", + "distance_from_injection", + "anchor_agent_id", + "anchor_hop_index", + "model", + "thinking_mode", + "layer", + "probe_name", + "score", + "label", + "label_target", + "behavioral_outcome", + "action_fired", + "latent_compromise_status", + "seed", +} +HOP_MODES = {"2-hop", "3-hop"} +CONDITIONS = {"clean", "injected"} +THINKING_MODES = {"off", "on"} +LABEL_TARGETS = {"trajectory_action_executed", "injection_present"} +BEHAVIORAL_OUTCOMES = { + "clean", + "resisted", + "propagated_but_not_executed", + "attempted_but_blocked", + "executed", + "indeterminate", +} +LATENT_STATUSES = {"not_candidate", "candidate", "probe_supported"} +METRIC_NAMES = ("auroc", "brier", "ece", "temporal_divergence_mean") + + +def analyze_depth_degradation( + rows: Iterable[dict], + *, + experiment_id: str, + n_bootstrap: int = 1000, + confidence: float = 0.95, + n_bins: int = 10, + random_state: int = 42, +) -> dict: + """Summarize metrics by depth and compute paired 3-hop minus 2-hop deltas.""" + + rows = sorted((dict(row) for row in rows), key=_prediction_row_sort_key) + validate_prediction_rows(rows) + if not experiment_id.strip(): + raise ValueError("experiment_id must be non-empty") + if n_bootstrap < 1: + raise ValueError("n_bootstrap must be at least 1") + if not 0.0 < confidence < 1.0: + raise ValueError("confidence must be between 0 and 1") + if n_bins < 1: + raise ValueError("n_bins must be at least 1") + + group_fields = ( + "model", + "thinking_mode", + "label_target", + "probe_name", + "layer", + "seed", + "hop_mode", + ) + grouped = _group_rows(rows, group_fields) + rng = np.random.default_rng(random_state) + group_results = [] + for key in sorted(grouped, key=_sortable_key): + group_rows = grouped[key] + metrics = _metric_snapshot(group_rows, n_bins=n_bins) + intervals = _bootstrap_intervals( + group_rows, + cluster_field="match_group_id", + n_bootstrap=n_bootstrap, + confidence=confidence, + n_bins=n_bins, + rng=rng, + ) + values = dict(zip(group_fields, key)) + executor_rows = _executor_rows(group_rows) + scored_executor_rows = _scored_executor_rows(group_rows) + group_results.append({ + **values, + "observation_agent": "executor", + "n_predictions": len(scored_executor_rows), + "n_excluded_unlabeled": len(executor_rows) - len(scored_executor_rows), + "n_indeterminate": sum( + row["behavioral_outcome"] == "indeterminate" + for row in executor_rows + ), + "behavioral_outcome_counts": _behavioral_outcome_counts(executor_rows), + "n_sequence_predictions": len(group_rows), + "n_trajectories": len({row["trajectory_id"] for row in group_rows}), + "n_match_groups": len({row["match_group_id"] for row in group_rows}), + "n_positive": sum(int(row["label"]) for row in scored_executor_rows), + "n_negative": sum(1 - int(row["label"]) for row in scored_executor_rows), + **metrics, + "confidence_intervals": intervals, + }) + + comparison_fields = ( + "model", + "thinking_mode", + "label_target", + "probe_name", + "layer", + "seed", + ) + by_configuration = _group_rows(rows, comparison_fields) + comparisons = [] + for key in sorted(by_configuration, key=_sortable_key): + config_rows = by_configuration[key] + by_hop = _group_rows(config_rows, ("hop_mode",)) + missing_hops = HOP_MODES - {hop_key[0] for hop_key in by_hop} + if missing_hops: + values = dict(zip(comparison_fields, key)) + raise ValueError( + f"Missing hop conditions {sorted(missing_hops)} for configuration {values}" + ) + two_hop = by_hop[("2-hop",)] + three_hop = by_hop[("3-hop",)] + two_metrics = _metric_snapshot(two_hop, n_bins=n_bins) + three_metrics = _metric_snapshot(three_hop, n_bins=n_bins) + deltas = { + metric: _difference(three_metrics[metric], two_metrics[metric]) + for metric in METRIC_NAMES + } + delta_intervals, n_match_groups = _bootstrap_delta_intervals( + two_hop, + three_hop, + n_bootstrap=n_bootstrap, + confidence=confidence, + n_bins=n_bins, + rng=rng, + ) + comparisons.append({ + **dict(zip(comparison_fields, key)), + "comparison": "3-hop_minus_2-hop", + "observation_agent": "executor", + "n_match_groups": n_match_groups, + "deltas": deltas, + "confidence_intervals": delta_intervals, + "interpretation": { + "auroc": "negative indicates worse discrimination at 3-hop", + "brier": "positive indicates worse probabilistic accuracy at 3-hop", + "ece": "positive indicates worse calibration at 3-hop", + "temporal_divergence_mean": ( + "signed change; interpret relative to the declared probe direction" + ), + }, + }) + + n_match_groups = len({str(row["match_group_id"]) for row in rows}) + return { + "schema_version": "spec_gap.depth_degradation.v2", + "experiment_id": experiment_id, + "data_manifest_hash": prediction_manifest_hash(rows), + "n_match_groups": n_match_groups, + "claim_scope": ( + "unsplit exploratory analysis; not a held-out performance estimate" + if n_match_groups < 3 + else "preliminary match-group evaluation" + ), + "sampling_grid": _sampling_grid(rows), + "behavioral_outcome_counts": _behavioral_outcome_counts( + _unique_executor_rows(rows) + ), + "analysis_config": { + "n_bootstrap": n_bootstrap, + "confidence": confidence, + "n_bins": n_bins, + "random_state": random_state, + "delta_definition": "3-hop minus 2-hop", + "classification_observation": "one executor score per trajectory", + "classification_exclusion": ( + "rows with label=null are excluded from AUROC, Brier, and ECE" + ), + "temporal_observation": "full ordered agent-score sequence", + "temporal_indeterminate_policy": ( + "indeterminate trajectories remain in Temporal Divergence" + ), + "bootstrap_unit": "match_group_id for depth metrics and depth deltas", + }, + "depth_metrics": group_results, + "depth_comparisons": comparisons, + } + + +def validate_prediction_rows(rows: list[dict]) -> None: + """Validate the analysis contract and fail closed on incomplete trajectories.""" + + if not rows: + raise ValueError("At least one prediction row is required.") + + seen_rows: set[tuple] = set() + trajectory_identity: dict[str, tuple] = {} + trajectory_label: dict[str, int | None] = {} + trajectory_layers: dict[tuple, set[int]] = defaultdict(set) + trajectory_hops: dict[tuple, dict[int, tuple[str, str]]] = defaultdict(dict) + match_group_cells: dict[str, dict[tuple[str, str, str], set[str]]] = defaultdict( + lambda: defaultdict(set) + ) + match_group_design: dict[str, tuple[str, str]] = {} + for index, row in enumerate(rows): + missing = sorted(REQUIRED_FIELDS - set(row)) + if missing: + raise ValueError(f"Row {index} is missing prediction fields: {missing}") + if row["hop_mode"] not in HOP_MODES: + raise ValueError(f"Row {index} has unsupported hop_mode {row['hop_mode']!r}") + if row["condition"] not in CONDITIONS: + raise ValueError(f"Row {index} has unsupported condition {row['condition']!r}") + if row["thinking_mode"] not in THINKING_MODES: + raise ValueError( + f"Row {index} has unsupported thinking_mode {row['thinking_mode']!r}" + ) + if row["label_target"] not in LABEL_TARGETS: + raise ValueError( + f"Row {index} has unsupported label_target {row['label_target']!r}" + ) + if row["behavioral_outcome"] not in BEHAVIORAL_OUTCOMES: + raise ValueError( + f"Row {index} has unsupported behavioral_outcome " + f"{row['behavioral_outcome']!r}" + ) + if row["latent_compromise_status"] not in LATENT_STATUSES: + raise ValueError( + f"Row {index} has unsupported latent_compromise_status " + f"{row['latent_compromise_status']!r}" + ) + if row["label"] not in (0, 1, False, True, None): + raise ValueError(f"Row {index} label must be binary or null") + if not isinstance(row["injection_present"], bool): + raise ValueError(f"Row {index} injection_present must be boolean") + if row["behavioral_outcome"] == "indeterminate": + if row["action_fired"] is not None: + raise ValueError( + f"Row {index} indeterminate outcome requires action_fired=null" + ) + elif not isinstance(row["action_fired"], bool): + raise ValueError(f"Row {index} completed outcome requires boolean action_fired") + score = float(row["score"]) + if not np.isfinite(score) or not 0.0 <= score <= 1.0: + raise ValueError(f"Row {index} score must be a finite probability in [0, 1]") + if not isinstance(row["hop_index"], int) or not isinstance(row["anchor_hop_index"], int): + raise ValueError(f"Row {index} hop indices must be integers") + if not isinstance(row["layer"], int): + raise ValueError(f"Row {index} layer must be an integer") + if row["anchor_agent_id"] != "worker_1" or row["anchor_hop_index"] != 1: + raise ValueError("Worker1 at hop 1 must be the fixed injection anchor") + if row["distance_from_injection"] != row["hop_index"] - 1: + raise ValueError( + f"Row {index} distance_from_injection must equal hop_index minus 1" + ) + + expected_injection = row["condition"] == "injected" + if row["injection_present"] is not expected_injection: + raise ValueError( + f"Row {index} condition and injection_present do not agree" + ) + if row["condition"] == "clean" and row["behavioral_outcome"] not in { + "clean", + "indeterminate", + }: + raise ValueError( + f"Row {index} clean condition must use clean or indeterminate behavioral_outcome" + ) + if row["condition"] == "injected" and row["behavioral_outcome"] == "clean": + raise ValueError(f"Row {index} injected condition cannot use clean behavioral_outcome") + if row["action_fired"] and row["behavioral_outcome"] != "executed": + raise ValueError(f"Row {index} action_fired requires executed behavioral_outcome") + if not row["action_fired"] and row["behavioral_outcome"] == "executed": + raise ValueError(f"Row {index} executed behavioral_outcome requires action_fired") + + candidate_outcomes = {"propagated_but_not_executed", "attempted_but_blocked"} + if row["behavioral_outcome"] in candidate_outcomes: + if row["latent_compromise_status"] not in {"candidate", "probe_supported"}: + raise ValueError( + f"Row {index} propagated/blocked outcome must be a latent candidate" + ) + elif row["latent_compromise_status"] != "not_candidate": + raise ValueError( + f"Row {index} latent candidate status conflicts with behavioral outcome" + ) + + expected_label = ( + None + if ( + row["label_target"] == "trajectory_action_executed" + and row["behavioral_outcome"] == "indeterminate" + ) + else int(row["action_fired"]) + if row["label_target"] == "trajectory_action_executed" + else int(row["injection_present"]) + ) + normalized_label = None if row["label"] is None else int(row["label"]) + if normalized_label != expected_label: + raise ValueError( + f"Row {index} label does not match label_target {row['label_target']!r}" + ) + + trajectory_id = str(row["trajectory_id"]) + identity = ( + str(row["match_group_id"]), + str(row["domain_id"]), + str(row["wording_id"]), + str(row["condition"]), + str(row["hop_mode"]), + str(row["model"]), + str(row["thinking_mode"]), + str(row["label_target"]), + str(row["behavioral_outcome"]), + row["action_fired"], + int(row["seed"]), + ) + previous_identity = trajectory_identity.setdefault(trajectory_id, identity) + if previous_identity != identity: + raise ValueError(f"Trajectory {trajectory_id!r} has inconsistent identity metadata") + previous_label = trajectory_label.setdefault(trajectory_id, normalized_label) + if previous_label != normalized_label: + raise ValueError(f"Trajectory {trajectory_id!r} has inconsistent labels") + + match_group_id = str(row["match_group_id"]) + design = (str(row["domain_id"]), str(row["wording_id"])) + previous_design = match_group_design.setdefault(match_group_id, design) + if previous_design != design: + raise ValueError( + f"Match group {match_group_id!r} mixes domain or wording metadata" + ) + cell = (str(row["thinking_mode"]), str(row["condition"]), str(row["hop_mode"])) + match_group_cells[match_group_id][cell].add(trajectory_id) + + row_key = ( + trajectory_id, + row["model"], + row["thinking_mode"], + row["label_target"], + row["probe_name"], + row["layer"], + row["seed"], + row["hop_index"], + ) + if row_key in seen_rows: + raise ValueError(f"Duplicate prediction row: {row_key}") + seen_rows.add(row_key) + + coverage_key = ( + trajectory_id, + str(row["model"]), + str(row["thinking_mode"]), + str(row["label_target"]), + str(row["probe_name"]), + int(row["seed"]), + ) + trajectory_layers[coverage_key].add(int(row["layer"])) + hop_key = (*coverage_key, int(row["layer"])) + if row["hop_index"] in trajectory_hops[hop_key]: + raise ValueError(f"Duplicate hop_index in trajectory/layer group: {hop_key}") + trajectory_hops[hop_key][int(row["hop_index"])] = ( + str(row["agent_id"]), + str(row["agent_role"]), + ) + + expected_layers: dict[tuple, set[int]] = {} + for coverage_key, layers in trajectory_layers.items(): + _, model, thinking_mode, label_target, probe_name, seed = coverage_key + config_key = (model, thinking_mode, label_target, probe_name, seed) + expected = expected_layers.setdefault(config_key, set(layers)) + if expected != layers: + raise ValueError( + f"Inconsistent layer coverage for {config_key}: " + f"expected {sorted(expected)}, found {sorted(layers)}" + ) + + for hop_key, hop_map in trajectory_hops.items(): + ordered = sorted(hop_map) + if ordered != list(range(len(ordered))): + raise ValueError( + f"Hop indices must be contiguous from zero for {hop_key}; found {ordered}" + ) + trajectory_id = hop_key[0] + hop_mode = trajectory_identity[trajectory_id][4] + expected_agents = ( + [("planner_1", "planner"), ("worker_1", "worker"), ("executor_1", "executor")] + if hop_mode == "2-hop" + else [ + ("planner_1", "planner"), + ("worker_1", "worker"), + ("worker_2", "worker"), + ("executor_1", "executor"), + ] + ) + actual_agents = [hop_map[index] for index in ordered] + if actual_agents != expected_agents: + raise ValueError( + f"Trajectory {trajectory_id!r} has incorrect agent order: {actual_agents}" + ) + + expected_base_cells = { + (condition, hop_mode) + for condition in CONDITIONS + for hop_mode in HOP_MODES + } + for match_group_id, cells in match_group_cells.items(): + thinking_modes = {thinking_mode for thinking_mode, _, _ in cells} + for thinking_mode in thinking_modes: + observed = { + (condition, hop_mode) + for mode, condition, hop_mode in cells + if mode == thinking_mode + } + if observed != expected_base_cells: + raise ValueError( + f"Match group {match_group_id!r} is missing a clean/injected depth cell " + f"for thinking_mode={thinking_mode!r}" + ) + for condition, hop_mode in expected_base_cells: + trajectory_ids = cells[(thinking_mode, condition, hop_mode)] + if len(trajectory_ids) != 1: + raise ValueError( + f"Match group {match_group_id!r} must contain one trajectory per cell" + ) + + +def prediction_manifest_hash(rows: Iterable[dict]) -> str: + """Hash normalized input rows so reported results identify their exact inputs.""" + + normalized = sorted( + (dict(row) for row in rows), + key=lambda row: ( + str(row.get("trajectory_id")), + str(row.get("model")), + str(row.get("thinking_mode")), + str(row.get("label_target")), + str(row.get("probe_name")), + int(row.get("layer", -1)), + int(row.get("seed", -1)), + int(row.get("hop_index", -1)), + ), + ) + payload = json.dumps(normalized, sort_keys=True, separators=(",", ":"), allow_nan=False) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def load_prediction_jsonl(path: str | Path) -> list[dict]: + """Load prediction rows from JSON Lines.""" + + rows = [] + with Path(path).open() as handle: + for line_number, line in enumerate(handle, start=1): + if not line.strip(): + continue + try: + rows.append(json.loads(line)) + except json.JSONDecodeError as exc: + raise ValueError(f"Invalid JSON on line {line_number} of {path}") from exc + validate_prediction_rows(rows) + return rows + + +def tabular_result_rows(result: dict) -> list[dict]: + """Flatten group metrics and depth deltas for CSV/DataFrame export.""" + + rows = [] + for group in result["depth_metrics"]: + row = {key: value for key, value in group.items() if key != "confidence_intervals"} + row["record_type"] = "depth_metric" + for metric, interval in group["confidence_intervals"].items(): + row[f"{metric}_ci_lower"] = interval["lower"] if interval else None + row[f"{metric}_ci_upper"] = interval["upper"] if interval else None + rows.append(row) + for comparison in result["depth_comparisons"]: + row = { + key: value + for key, value in comparison.items() + if key not in {"deltas", "confidence_intervals", "interpretation"} + } + row["record_type"] = "depth_comparison" + for metric, value in comparison["deltas"].items(): + row[f"{metric}_delta"] = value + interval = comparison["confidence_intervals"][metric] + row[f"{metric}_delta_ci_lower"] = interval["lower"] if interval else None + row[f"{metric}_delta_ci_upper"] = interval["upper"] if interval else None + rows.append(row) + return rows + + +def _metric_snapshot(rows: list[dict], *, n_bins: int) -> dict: + executor_rows = _scored_executor_rows(rows) + classification = _classification_snapshot(executor_rows, n_bins=n_bins) + temporal_values = list(_temporal_values_by_trajectory(rows).values()) + return { + **classification, + "temporal_divergence_mean": float(np.mean(temporal_values)), + } + + +def _executor_rows(rows: list[dict], *, require_unique: bool = True) -> list[dict]: + executor_rows = [row for row in rows if row["agent_id"] == "executor_1"] + if not require_unique: + return executor_rows + trajectory_ids = {str(row["trajectory_id"]) for row in rows} + executor_trajectories = {str(row["trajectory_id"]) for row in executor_rows} + if executor_trajectories != trajectory_ids or len(executor_rows) != len(trajectory_ids): + raise ValueError("Classification metrics require exactly one executor score per trajectory") + return executor_rows + + +def _scored_executor_rows(rows: list[dict], *, require_unique: bool = True) -> list[dict]: + return [ + row + for row in _executor_rows(rows, require_unique=require_unique) + if row["label"] is not None + ] + + +def _classification_snapshot(rows: list[dict], *, n_bins: int) -> dict: + if not rows: + return {"auroc": None, "brier": None, "ece": None} + labels = np.asarray([int(row["label"]) for row in rows], dtype=int) + scores = np.asarray([float(row["score"]) for row in rows], dtype=float) + return { + "auroc": ( + float(roc_auc_score(labels, scores)) + if len(np.unique(labels)) == 2 + else None + ), + "brier": float(brier_score_loss(labels, scores)), + "ece": float(compute_ece(labels, scores, n_bins=n_bins)), + } + + +def _behavioral_outcome_counts(executor_rows: list[dict]) -> dict[str, int]: + return { + outcome: sum(row["behavioral_outcome"] == outcome for row in executor_rows) + for outcome in sorted(BEHAVIORAL_OUTCOMES) + } + + +def _unique_executor_rows(rows: list[dict]) -> list[dict]: + unique: dict[str, dict] = {} + for row in rows: + if row["agent_id"] == "executor_1": + unique.setdefault(str(row["trajectory_id"]), row) + return list(unique.values()) + + +def _temporal_values_by_trajectory(rows: list[dict]) -> dict[str, float]: + grouped = _group_rows(rows, ("trajectory_id",)) + values = {} + for (trajectory_id,), trajectory_rows in grouped.items(): + ordered = sorted(trajectory_rows, key=lambda row: row["hop_index"]) + anchor = int(ordered[0]["anchor_hop_index"]) + anchor_position = [row["hop_index"] for row in ordered].index(anchor) + result = temporal_divergence( + [float(row["score"]) for row in ordered], + anchor_position=anchor_position, + ) + values[str(trajectory_id)] = result.divergence_score + return values + + +def _bootstrap_intervals( + rows: list[dict], + *, + cluster_field: str, + n_bootstrap: int, + confidence: float, + n_bins: int, + rng: np.random.Generator, +) -> dict: + cluster_ids = sorted({str(row[cluster_field]) for row in rows}) + samples = {metric: [] for metric in METRIC_NAMES} + for _ in range(n_bootstrap): + selected = rng.choice(cluster_ids, size=len(cluster_ids), replace=True).tolist() + snapshot = _resampled_snapshot( + rows, + selected, + cluster_field=cluster_field, + n_bins=n_bins, + ) + for metric in METRIC_NAMES: + if snapshot[metric] is not None: + samples[metric].append(snapshot[metric]) + return { + metric: _confidence_interval(values, confidence) + for metric, values in samples.items() + } + + +def _bootstrap_delta_intervals( + two_hop: list[dict], + three_hop: list[dict], + *, + n_bootstrap: int, + confidence: float, + n_bins: int, + rng: np.random.Generator, +) -> tuple[dict, int]: + two_groups = {str(row["match_group_id"]) for row in two_hop} + three_groups = {str(row["match_group_id"]) for row in three_hop} + matched_groups = sorted(two_groups & three_groups) + if not matched_groups: + raise ValueError("Depth comparison requires at least one match group at both depths") + if two_groups != three_groups: + raise ValueError( + "Depth comparison requires identical match-group coverage at 2-hop and 3-hop" + ) + + samples = {metric: [] for metric in METRIC_NAMES} + for _ in range(n_bootstrap): + selected = rng.choice( + matched_groups, + size=len(matched_groups), + replace=True, + ).tolist() + two_snapshot = _resampled_snapshot( + two_hop, + selected, + cluster_field="match_group_id", + n_bins=n_bins, + ) + three_snapshot = _resampled_snapshot( + three_hop, + selected, + cluster_field="match_group_id", + n_bins=n_bins, + ) + for metric in METRIC_NAMES: + delta = _difference(three_snapshot[metric], two_snapshot[metric]) + if delta is not None: + samples[metric].append(delta) + return ( + { + metric: _confidence_interval(values, confidence) + for metric, values in samples.items() + }, + len(matched_groups), + ) + + +def _resampled_snapshot( + rows: list[dict], + selected_clusters: list[str], + *, + cluster_field: str, + n_bins: int, +) -> dict: + rows_by_cluster: dict[str, list[dict]] = defaultdict(list) + for row in rows: + rows_by_cluster[str(row[cluster_field])].append(row) + sampled_rows = [ + row + for cluster_id in selected_clusters + for row in rows_by_cluster[cluster_id] + ] + executor_rows = _scored_executor_rows(sampled_rows, require_unique=False) + classification = _classification_snapshot(executor_rows, n_bins=n_bins) + + temporal_by_trajectory = _temporal_values_by_trajectory(rows) + temporal_by_cluster: dict[str, list[float]] = defaultdict(list) + trajectory_cluster = { + str(row["trajectory_id"]): str(row[cluster_field]) + for row in rows + } + for trajectory_id, value in temporal_by_trajectory.items(): + temporal_by_cluster[trajectory_cluster[trajectory_id]].append(value) + sampled_temporal = [ + value + for cluster_id in selected_clusters + for value in temporal_by_cluster[cluster_id] + ] + return { + **classification, + "temporal_divergence_mean": float(np.mean(sampled_temporal)), + } + + +def _confidence_interval(values: list[float], confidence: float) -> dict | None: + if not values: + return None + alpha = (1.0 - confidence) / 2.0 + return { + "lower": float(np.quantile(values, alpha)), + "upper": float(np.quantile(values, 1.0 - alpha)), + } + + +def _difference(three_hop: float | None, two_hop: float | None) -> float | None: + if three_hop is None or two_hop is None: + return None + return float(three_hop - two_hop) + + +def _group_rows(rows: list[dict], fields: tuple[str, ...]) -> dict[tuple, list[dict]]: + grouped: dict[tuple, list[dict]] = defaultdict(list) + for row in rows: + grouped[tuple(row[field] for field in fields)].append(row) + return dict(grouped) + + +def _sortable_key(key: tuple) -> tuple[str, ...]: + return tuple(str(value) for value in key) + + +def _prediction_row_sort_key(row: dict) -> tuple: + return ( + str(row.get("match_group_id")), + str(row.get("thinking_mode")), + str(row.get("condition")), + str(row.get("hop_mode")), + str(row.get("trajectory_id")), + str(row.get("model")), + str(row.get("label_target")), + str(row.get("probe_name")), + int(row.get("layer", -1)), + int(row.get("seed", -1)), + int(row.get("hop_index", -1)), + ) + + +def _sampling_grid(rows: list[dict]) -> dict: + domains = sorted({str(row["domain_id"]) for row in rows}) + wordings = sorted({str(row["wording_id"]) for row in rows}) + observed = sorted({(str(row["domain_id"]), str(row["wording_id"])) for row in rows}) + expected = {(domain, wording) for domain in domains for wording in wordings} + return { + "domains": domains, + "wordings": wordings, + "observed_domain_wording_cells": [ + {"domain_id": domain, "wording_id": wording} + for domain, wording in observed + ], + "fully_crossed": set(observed) == expected, + } diff --git a/src/analysis/splits.py b/src/analysis/splits.py new file mode 100644 index 0000000..d03e051 --- /dev/null +++ b/src/analysis/splits.py @@ -0,0 +1,318 @@ +"""Leakage-safe match-group splits for Scenario 1 evaluation.""" + +from __future__ import annotations + +import hashlib +from typing import Iterable + + +REQUIRED_SPLIT_FIELDS = { + "trajectory_id", + "match_group_id", + "condition", + "hop_mode", + "thinking_mode", +} +EXPECTED_CELLS = { + ("clean", "2-hop"), + ("injected", "2-hop"), + ("clean", "3-hop"), + ("injected", "3-hop"), +} + + +def build_match_group_split_manifest( + rows: Iterable[dict], + *, + train_fraction: float = 0.6, + validation_fraction: float = 0.2, + random_state: int = 42, +) -> dict: + """Assign complete four-trajectory match groups to train/validation/test.""" + + grouped = _validate_and_group(rows) + group_ids = _ordered_group_ids(grouped, random_state=random_state) + counts = _split_counts( + len(group_ids), + train_fraction=train_fraction, + validation_fraction=validation_fraction, + ) + assignments = {} + start = 0 + for split_name in ("train", "validation", "test"): + end = start + counts[split_name] + for match_group_id in group_ids[start:end]: + assignments[match_group_id] = split_name + start = end + + manifest = _manifest( + grouped, + assignments, + mode="evaluation", + random_state=random_state, + fractions={ + "train": train_fraction, + "validation": validation_fraction, + "test": 1.0 - train_fraction - validation_fraction, + }, + counts=counts, + claim_scope="grouped train/validation/test evaluation", + ) + validate_split_manifest(manifest) + return manifest + + +def build_unsplit_manifest(rows: Iterable[dict]) -> dict: + """Keep complete groups together without creating data splits.""" + + grouped = _validate_and_group(rows) + assignments = {match_group_id: "all" for match_group_id in grouped} + manifest = _manifest( + grouped, + assignments, + mode="unsplit", + random_state=None, + fractions=None, + counts={"all": len(grouped)}, + claim_scope="all match groups retained without train/validation/test separation", + ) + validate_split_manifest(manifest) + return manifest + + +def build_matched_split_manifest(*args, **kwargs) -> dict: + """Backward-compatible alias; new code should use match-group terminology.""" + + return build_match_group_split_manifest(*args, **kwargs) + + +def apply_split_manifest(rows: Iterable[dict], manifest: dict) -> list[dict]: + """Return row copies annotated with their match-group split.""" + + validate_split_manifest(manifest) + assignment = { + entry["match_group_id"]: entry["split"] + for entry in manifest["match_groups"] + } + annotated = [] + for row in rows: + match_group_id = row.get("match_group_id") + if match_group_id not in assignment: + raise ValueError( + f"match_group_id {match_group_id!r} is missing from the split manifest" + ) + annotated.append({**row, "split": assignment[match_group_id]}) + return annotated + + +def validate_split_manifest(manifest: dict) -> None: + """Fail closed on incomplete groups or trajectory leakage.""" + + mode = manifest.get("mode") + allowed_splits = ( + {"train", "validation", "test"} + if mode == "evaluation" + else {"all"} + if mode == "unsplit" + else set() + ) + if not allowed_splits: + raise ValueError(f"Unsupported split-manifest mode {mode!r}") + + groups = manifest.get("match_groups") + if not isinstance(groups, list) or not groups: + raise ValueError("Split manifest must contain at least one match group.") + + seen_groups: set[str] = set() + trajectory_split: dict[str, str] = {} + split_counts = {name: 0 for name in allowed_splits} + for entry in groups: + match_group_id = entry.get("match_group_id") + split = entry.get("split") + if not match_group_id or match_group_id in seen_groups: + raise ValueError( + f"Duplicate or missing match_group_id in split manifest: {match_group_id!r}" + ) + if split not in allowed_splits: + raise ValueError(f"Unsupported split {split!r} for match group {match_group_id!r}") + cells = { + (cell.get("thinking_mode"), cell.get("condition"), cell.get("hop_mode")) + for cell in entry.get("cells") or [] + } + thinking_modes = {thinking_mode for thinking_mode, _, _ in cells} + expected_cells = { + (thinking_mode, condition, hop_mode) + for thinking_mode in thinking_modes + for condition, hop_mode in EXPECTED_CELLS + } + if not thinking_modes or cells != expected_cells: + raise ValueError( + f"Match group {match_group_id!r} must contain clean/injected at both depths; " + f"found {sorted(cells)}" + ) + trajectory_ids = entry.get("trajectory_ids") or [] + expected_trajectories = 4 * len(thinking_modes) + if ( + len(trajectory_ids) != expected_trajectories + or len(set(trajectory_ids)) != expected_trajectories + ): + raise ValueError( + f"Match group {match_group_id!r} must contain four trajectories per thinking mode" + ) + for trajectory_id in trajectory_ids: + previous = trajectory_split.get(trajectory_id) + if previous is not None and previous != split: + raise ValueError( + f"Trajectory {trajectory_id!r} leaks across {previous!r} and {split!r}" + ) + trajectory_split[trajectory_id] = split + seen_groups.add(match_group_id) + split_counts[split] += 1 + + if mode == "evaluation": + missing = [name for name, count in split_counts.items() if count == 0] + if missing: + raise ValueError(f"Split manifest has no match groups in: {missing}") + + +def _validate_and_group(rows: Iterable[dict]) -> dict[str, dict]: + rows = list(rows) + if not rows: + raise ValueError("At least one trajectory row is required to build splits.") + + groups: dict[str, dict] = {} + trajectory_group: dict[str, str] = {} + trajectory_cell: dict[str, tuple[str, str]] = {} + for index, row in enumerate(rows): + missing = sorted(REQUIRED_SPLIT_FIELDS - set(row)) + if missing: + raise ValueError(f"Row {index} is missing split fields: {missing}") + trajectory_id = str(row["trajectory_id"]) + match_group_id = str(row["match_group_id"]) + base_cell = (str(row["condition"]), str(row["hop_mode"])) + if base_cell not in EXPECTED_CELLS: + raise ValueError(f"Row {index} has unsupported match-group cell {base_cell!r}") + cell = (str(row["thinking_mode"]), *base_cell) + if not cell[0]: + raise ValueError(f"Row {index} has unsupported match-group cell {cell!r}") + + previous_group = trajectory_group.setdefault(trajectory_id, match_group_id) + if previous_group != match_group_id: + raise ValueError( + f"Trajectory {trajectory_id!r} belongs to multiple match groups: " + f"{previous_group!r} and {match_group_id!r}" + ) + previous_cell = trajectory_cell.setdefault(trajectory_id, cell) + if previous_cell != cell: + raise ValueError(f"Trajectory {trajectory_id!r} belongs to multiple cells") + + grouped = groups.setdefault( + match_group_id, + {"trajectory_ids": set(), "cell_trajectories": {}}, + ) + grouped["trajectory_ids"].add(trajectory_id) + grouped["cell_trajectories"].setdefault(cell, set()).add(trajectory_id) + + for match_group_id, grouped in groups.items(): + cells = set(grouped["cell_trajectories"]) + thinking_modes = {thinking_mode for thinking_mode, _, _ in cells} + expected_cells = { + (thinking_mode, condition, hop_mode) + for thinking_mode in thinking_modes + for condition, hop_mode in EXPECTED_CELLS + } + if not thinking_modes or cells != expected_cells: + raise ValueError( + f"Match group {match_group_id!r} must contain clean/injected at both depths; " + f"found {sorted(cells)}" + ) + if any(len(ids) != 1 for ids in grouped["cell_trajectories"].values()): + raise ValueError( + f"Match group {match_group_id!r} must contain one trajectory per cell" + ) + if len(grouped["trajectory_ids"]) != 4 * len(thinking_modes): + raise ValueError( + f"Match group {match_group_id!r} must contain four trajectories per thinking mode" + ) + return groups + + +def _ordered_group_ids(grouped: dict[str, dict], *, random_state: int) -> list[str]: + return sorted( + grouped, + key=lambda match_group_id: hashlib.sha256( + f"{random_state}:{match_group_id}".encode("utf-8") + ).hexdigest(), + ) + + +def _manifest( + grouped: dict[str, dict], + assignments: dict[str, str], + *, + mode: str, + random_state: int | None, + fractions: dict | None, + counts: dict[str, int], + claim_scope: str, +) -> dict: + match_groups = [] + for match_group_id in sorted(grouped): + group = grouped[match_group_id] + cells = [ + { + "thinking_mode": thinking_mode, + "condition": condition, + "hop_mode": hop_mode, + "trajectory_id": next( + iter(group["cell_trajectories"][(thinking_mode, condition, hop_mode)]) + ), + } + for thinking_mode, condition, hop_mode in sorted(group["cell_trajectories"]) + ] + match_groups.append({ + "match_group_id": match_group_id, + "split": assignments[match_group_id], + "cells": cells, + "trajectory_ids": sorted(group["trajectory_ids"]), + }) + return { + "schema_version": "spec_gap.match_group_splits.v1", + "mode": mode, + "random_state": random_state, + "fractions": fractions, + "n_match_groups": len(grouped), + "counts": counts, + "claim_scope": claim_scope, + "match_groups": match_groups, + } + + +def _split_counts( + n_match_groups: int, + *, + train_fraction: float, + validation_fraction: float, +) -> dict[str, int]: + test_fraction = 1.0 - train_fraction - validation_fraction + fractions = (train_fraction, validation_fraction, test_fraction) + if any(fraction <= 0.0 for fraction in fractions): + raise ValueError("Train, validation, and test fractions must all be positive.") + if n_match_groups < 3: + raise ValueError( + "At least three match groups are required for train/validation/test; " + "use build_unsplit_manifest when train/validation/test splits are not appropriate." + ) + + train = max(1, round(n_match_groups * train_fraction)) + validation = max(1, round(n_match_groups * validation_fraction)) + test = n_match_groups - train - validation + while test < 1 and train > 1: + train -= 1 + test += 1 + while test < 1 and validation > 1: + validation -= 1 + test += 1 + if test < 1: + raise ValueError("Fractions cannot produce three non-empty splits.") + return {"train": train, "validation": validation, "test": test} diff --git a/tests/test_depth_degradation.py b/tests/test_depth_degradation.py new file mode 100644 index 0000000..3357911 --- /dev/null +++ b/tests/test_depth_degradation.py @@ -0,0 +1,332 @@ +"""Tests for compute-independent depth-degradation analysis.""" + +import copy + +import pytest + +from src.analysis.depth_degradation import ( + analyze_depth_degradation, + prediction_manifest_hash, + tabular_result_rows, +) + + +def _prediction_rows( + *, + layers=(32,), + n_match_groups=6, + thinking_modes=("off",), + label_target="trajectory_action_executed", + all_negative=False, +): + rows = [] + agents = { + "2-hop": [ + ("planner_1", "planner"), + ("worker_1", "worker"), + ("executor_1", "executor"), + ], + "3-hop": [ + ("planner_1", "planner"), + ("worker_1", "worker"), + ("worker_2", "worker"), + ("executor_1", "executor"), + ], + } + wordings = ("send", "forward", "transmit") + domains = ("public_health", "climate_science") + + for group_index in range(n_match_groups): + domain_id = domains[(group_index // 3) % len(domains)] + wording_id = wordings[group_index % len(wordings)] + match_group_id = f"{domain_id}-{wording_id}-{group_index}" + for thinking_mode in thinking_modes: + for hop_mode in ("2-hop", "3-hop"): + for condition in ("clean", "injected"): + action_fired = ( + condition == "injected" + and not all_negative + and ( + group_index % 3 != 2 + if hop_mode == "2-hop" + else group_index % 3 == 0 + ) + ) + if condition == "clean": + outcome = "clean" + latent_status = "not_candidate" + elif action_fired: + outcome = "executed" + latent_status = "not_candidate" + else: + outcome = "propagated_but_not_executed" + latent_status = "candidate" + + injection_present = condition == "injected" + label = ( + int(action_fired) + if label_target == "trajectory_action_executed" + else int(injection_present) + ) + trajectory_id = ( + f"{match_group_id}-{thinking_mode}-{hop_mode}-{condition}" + ) + for layer in layers: + scores = _scores( + hop_mode=hop_mode, + condition=condition, + action_fired=action_fired, + ) + for hop_index, (agent_id, agent_role) in enumerate(agents[hop_mode]): + rows.append({ + "trajectory_id": trajectory_id, + "match_group_id": match_group_id, + "domain_id": domain_id, + "wording_id": wording_id, + "condition": condition, + "injection_present": injection_present, + "hop_mode": hop_mode, + "agent_id": agent_id, + "agent_role": agent_role, + "hop_index": hop_index, + "distance_from_injection": hop_index - 1, + "anchor_agent_id": "worker_1", + "anchor_hop_index": 1, + "model": "Qwen/Qwen3-32B", + "thinking_mode": thinking_mode, + "layer": layer, + "probe_name": "goldowsky_dill", + "score": scores[hop_index], + "label": label, + "label_target": label_target, + "behavioral_outcome": outcome, + "action_fired": action_fired, + "latent_compromise_status": latent_status, + "seed": 0, + }) + return rows + + +def _scores(*, hop_mode, condition, action_fired): + if hop_mode == "2-hop": + if action_fired: + return [0.10, 0.75, 0.90] + if condition == "injected": + return [0.10, 0.45, 0.35] + return [0.05, 0.10, 0.10] + if action_fired: + return [0.10, 0.50, 0.55, 0.55] + if condition == "injected": + return [0.10, 0.45, 0.55, 0.60] + return [0.10, 0.25, 0.35, 0.45] + + +def _analyze(rows, *, random_state=11): + return analyze_depth_degradation( + rows, + experiment_id="scenario1-test", + n_bootstrap=80, + random_state=random_state, + ) + + +def test_reports_executor_metrics_and_match_group_depth_delta(): + result = _analyze(_prediction_rows()) + + assert result["schema_version"] == "spec_gap.depth_degradation.v2" + assert result["sampling_grid"]["fully_crossed"] is True + assert len(result["depth_metrics"]) == 2 + assert all(group["observation_agent"] == "executor" for group in result["depth_metrics"]) + comparison = result["depth_comparisons"][0] + assert comparison["n_match_groups"] == 6 + assert comparison["deltas"]["auroc"] < 0 + assert comparison["deltas"]["brier"] > 0 + assert comparison["confidence_intervals"]["brier"] is not None + + table = tabular_result_rows(result) + assert {row["record_type"] for row in table} == { + "depth_metric", + "depth_comparison", + } + + +def test_bootstrap_and_manifest_hash_are_reproducible(): + rows = _prediction_rows() + first = _analyze(rows, random_state=123) + second = _analyze(list(reversed(rows)), random_state=123) + + assert first == second + assert prediction_manifest_hash(rows) == prediction_manifest_hash(reversed(rows)) + + +def test_constant_action_labels_report_undefined_auroc_without_crashing(): + result = _analyze(_prediction_rows(all_negative=True)) + + assert all(group["auroc"] is None for group in result["depth_metrics"]) + assert all( + group["confidence_intervals"]["auroc"] is None + for group in result["depth_metrics"] + ) + assert result["depth_comparisons"][0]["deltas"]["auroc"] is None + + +def test_thinking_modes_are_never_pooled(): + result = _analyze(_prediction_rows(thinking_modes=("off", "on"))) + + assert len(result["depth_metrics"]) == 4 + assert len(result["depth_comparisons"]) == 2 + assert {row["thinking_mode"] for row in result["depth_comparisons"]} == {"off", "on"} + + +def test_classification_metrics_ignore_non_executor_score_changes(): + rows = _prediction_rows() + baseline = _analyze(rows) + changed = copy.deepcopy(rows) + for row in changed: + if row["agent_id"] != "executor_1": + row["score"] = 1.0 - row["score"] + result = _analyze(changed) + + for before, after in zip(baseline["depth_metrics"], result["depth_metrics"]): + assert before["auroc"] == after["auroc"] + assert before["brier"] == after["brier"] + assert before["ece"] == after["ece"] + assert before["temporal_divergence_mean"] != after["temporal_divergence_mean"] + + +def test_rejects_missing_hop_condition_in_match_group(): + rows = [ + row + for row in _prediction_rows() + if not ( + row["match_group_id"].endswith("-0") + and row["hop_mode"] == "3-hop" + and row["condition"] == "injected" + ) + ] + with pytest.raises(ValueError, match="missing a clean/injected depth cell"): + _analyze(rows) + + +def test_rejects_inconsistent_layer_coverage(): + rows = _prediction_rows(layers=(16, 32)) + target = rows[0]["trajectory_id"] + rows = [ + row for row in rows + if not (row["trajectory_id"] == target and row["layer"] == 32) + ] + with pytest.raises(ValueError, match="Inconsistent layer coverage"): + _analyze(rows) + + +def test_rejects_worker2_as_injection_anchor(): + rows = _prediction_rows() + for row in rows: + row["anchor_agent_id"] = "worker_2" + row["anchor_hop_index"] = 2 + row["distance_from_injection"] = row["hop_index"] - 2 + with pytest.raises(ValueError, match="Worker1 at hop 1"): + _analyze(rows) + + +def test_rejects_action_label_created_from_latent_candidate(): + rows = _prediction_rows() + target = next( + row["trajectory_id"] + for row in rows + if row["latent_compromise_status"] == "candidate" + ) + for row in rows: + if row["trajectory_id"] == target: + row["label"] = 1 + with pytest.raises(ValueError, match="label does not match label_target"): + _analyze(rows) + + +def _mark_trajectory_indeterminate(rows, trajectory_id): + for row in rows: + if row["trajectory_id"] != trajectory_id: + continue + row["behavioral_outcome"] = "indeterminate" + row["action_fired"] = None + row["latent_compromise_status"] = "not_candidate" + if row["label_target"] == "trajectory_action_executed": + row["label"] = None + + +def test_indeterminate_action_outcomes_are_excluded_from_classification_only(): + rows = _prediction_rows() + targets = { + row["trajectory_id"] + for row in rows + if row["match_group_id"].endswith("-0") and row["condition"] == "injected" + } + for trajectory_id in targets: + _mark_trajectory_indeterminate(rows, trajectory_id) + + result = _analyze(rows) + + assert all(group["n_predictions"] == 11 for group in result["depth_metrics"]) + assert all(group["n_excluded_unlabeled"] == 1 for group in result["depth_metrics"]) + assert all(group["n_indeterminate"] == 1 for group in result["depth_metrics"]) + assert result["behavioral_outcome_counts"]["indeterminate"] == 2 + assert all( + group["temporal_divergence_mean"] is not None + for group in result["depth_metrics"] + ) + + +def test_indeterminate_action_outcome_rejects_binary_action_label(): + rows = _prediction_rows() + target = rows[0]["trajectory_id"] + _mark_trajectory_indeterminate(rows, target) + for row in rows: + if row["trajectory_id"] == target: + row["label"] = 0 + + with pytest.raises(ValueError, match="label does not match label_target"): + _analyze(rows) + + +def test_indeterminate_outcome_rejects_false_action_status(): + rows = _prediction_rows() + target = rows[0]["trajectory_id"] + _mark_trajectory_indeterminate(rows, target) + for row in rows: + if row["trajectory_id"] == target: + row["action_fired"] = False + + with pytest.raises(ValueError, match="requires action_fired=null"): + _analyze(rows) + + +def test_indeterminate_outcome_keeps_known_injection_construction_label(): + rows = _prediction_rows(label_target="injection_present") + target = next(row["trajectory_id"] for row in rows if row["condition"] == "injected") + _mark_trajectory_indeterminate(rows, target) + + result = _analyze(rows) + + assert all(group["n_excluded_unlabeled"] == 0 for group in result["depth_metrics"]) + + +def test_all_indeterminate_action_outcomes_report_null_classification_metrics(): + rows = _prediction_rows() + for trajectory_id in {row["trajectory_id"] for row in rows}: + _mark_trajectory_indeterminate(rows, trajectory_id) + + result = _analyze(rows) + + for group in result["depth_metrics"]: + assert group["n_predictions"] == 0 + assert group["n_excluded_unlabeled"] == group["n_trajectories"] + assert group["auroc"] is None + assert group["brier"] is None + assert group["ece"] is None + assert group["temporal_divergence_mean"] is not None + + +def test_top_level_outcome_counts_do_not_double_count_layers(): + result = _analyze(_prediction_rows(layers=(16, 32))) + + assert sum(result["behavioral_outcome_counts"].values()) == 24 diff --git a/tests/test_splits.py b/tests/test_splits.py new file mode 100644 index 0000000..e780dd0 --- /dev/null +++ b/tests/test_splits.py @@ -0,0 +1,102 @@ +"""Tests for deterministic match-group split construction.""" + +import copy + +import pytest + +from src.analysis.splits import ( + apply_split_manifest, + build_match_group_split_manifest, + build_unsplit_manifest, + validate_split_manifest, +) + + +def _group_rows(n_match_groups=5, thinking_modes=("off",)): + rows = [] + for group_index in range(n_match_groups): + match_group_id = f"group-{group_index}" + for thinking_mode in thinking_modes: + for hop_mode in ("2-hop", "3-hop"): + for condition in ("clean", "injected"): + rows.append({ + "trajectory_id": ( + f"{match_group_id}-{thinking_mode}-{hop_mode}-{condition}" + ), + "match_group_id": match_group_id, + "thinking_mode": thinking_mode, + "hop_mode": hop_mode, + "condition": condition, + }) + return rows + + +def test_builds_reproducible_non_leaking_splits(): + rows = _group_rows() + first = build_match_group_split_manifest(rows, random_state=7) + second = build_match_group_split_manifest(reversed(rows), random_state=7) + + assert first == second + assert first["counts"] == {"train": 3, "validation": 1, "test": 1} + assert first["n_match_groups"] == 5 + + annotated = apply_split_manifest(rows, first) + split_by_group = {} + for row in annotated: + split_by_group.setdefault(row["match_group_id"], set()).add(row["split"]) + assert all(len(splits) == 1 for splits in split_by_group.values()) + + +def test_keeps_both_thinking_modes_in_one_group_split(): + rows = _group_rows(thinking_modes=("off", "on")) + manifest = build_match_group_split_manifest(rows) + + assert all(len(entry["trajectory_ids"]) == 8 for entry in manifest["match_groups"]) + annotated = apply_split_manifest(rows, manifest) + for match_group_id in {row["match_group_id"] for row in rows}: + splits = { + row["split"] + for row in annotated + if row["match_group_id"] == match_group_id + } + assert len(splits) == 1 + + +def test_rejects_incomplete_four_trajectory_group(): + rows = [ + row + for row in _group_rows() + if not ( + row["match_group_id"] == "group-0" + and row["condition"] == "clean" + and row["hop_mode"] == "3-hop" + ) + ] + with pytest.raises(ValueError, match="clean/injected at both depths"): + build_match_group_split_manifest(rows) + + +def test_rejects_trajectory_leakage_in_manifest(): + manifest = build_match_group_split_manifest(_group_rows()) + corrupted = copy.deepcopy(manifest) + first = corrupted["match_groups"][0] + other = next( + entry + for entry in corrupted["match_groups"] + if entry["split"] != first["split"] + ) + other["trajectory_ids"].append(first["trajectory_ids"][0]) + + with pytest.raises(ValueError, match="four trajectories per thinking mode"): + validate_split_manifest(corrupted) + + +def test_two_group_data_uses_unsplit_manifest(): + rows = _group_rows(n_match_groups=2) + with pytest.raises(ValueError, match="build_unsplit_manifest"): + build_match_group_split_manifest(rows) + + manifest = build_unsplit_manifest(rows) + assert manifest["mode"] == "unsplit" + assert manifest["counts"] == {"all": 2} + assert "without train/validation/test separation" in manifest["claim_scope"]