SRIC: external semantic admission

Download original file

This is the original source text. Reading it does not run or verify the program.

Read the typeset paper

1 from __future__ import annotations
2 
3 from dataclasses import asdict
4 from typing import Any, Mapping, Sequence
5 import numpy as np
6 
7 from .adapters.base import LatentReasonerAdapter, UnsupportedAdapterOperation
8 from .dataset import semantic_twin, validate_semantic_twin
9 from .donors import (
10     DonorAssignment,
11     build_different_answer_far_proof_plan,
12     build_same_answer_far_proof_plan,
13     validate_donor_plan,
14 )
15 from .hashing import sha256_object
16 from .interventions import (
17     covariance_matched_noise,
18     norm_matched_noise,
19     permute_trace,
20     truncate_trace,
21     zero_trace,
22 )
23 from .metrics import covariance_from_traces, summarize_effect
24 from .schemas import AuditConfig, TaskRecord, TraceObservation
25 
26 
27 def _max_logit_error(first: TraceObservation, second: TraceObservation) -> float:
28     left = np.asarray(first.answer_logits, dtype=np.float64)
29     right = np.asarray(second.answer_logits, dtype=np.float64)
30     if left.shape != right.shape:
31         return float("inf")
32     return float(np.max(np.abs(left - right))) if left.size else 0.0
33 
34 
35 def _score(adapter: LatentReasonerAdapter, task: TaskRecord, trace: Any) -> TraceObservation:
36     observation = adapter.score_with_trace(task, trace)
37     if not np.isfinite(observation.answer_loss):
38         raise ValueError(f"Non-finite answer loss for {task.task_id}")
39     return observation
40 
41 
42 def _accuracy(observations: Sequence[tuple[TaskRecord, TraceObservation]]) -> float | None:
43     usable = [
44         (task, observation)
45         for task, observation in observations
46         if observation.predicted_answer is not None
47     ]
48     if not usable:
49         return None
50     return float(
51         np.mean(
52             [
53                 observation.predicted_answer.strip().lower() == task.answer.strip().lower()
54                 for task, observation in usable
55             ]
56         )
57     )
58 
59 
60 def classify(summary: Mapping[str, Any], config: AuditConfig) -> tuple[str, dict[str, bool]]:
61     thresholds = config.thresholds
62     replay = summary["replay_max_abs"] <= thresholds.replay_max_abs
63     necessity = (
64         summary["necessity"]["mean"] >= thresholds.necessity_mean
65         and summary["necessity"]["positive_fraction"] >= thresholds.necessity_positive_fraction
66     )
67     ordered = summary["order"]["mean"] >= thresholds.order_mean
68     norm_structured = summary["norm_structure"]["mean"] >= thresholds.structure_mean
69     covariance_structured = (
70         True
71         if summary.get("covariance_structure") is None
72         else summary["covariance_structure"]["mean"] >= thresholds.covariance_structure_mean
73     )
74     structured = norm_structured and covariance_structured
75     procedure = (
76         summary["procedure_transplant"]["mean"] >= thresholds.procedure_mean
77         and summary["procedure_transplant"]["positive_fraction"] >= thresholds.procedure_positive_fraction
78     )
79     twin_stable = (
80         abs(summary["twin_transplant"]["mean"]) <= thresholds.twin_abs_mean_max
81         and summary["twin_max_abs"] <= thresholds.twin_absolute_ceiling
82     )
83     specificity_label = f"{config.specificity_claim}_specific"
84     gates = {
85         "replay": replay,
86         "necessity": necessity,
87         "ordered": ordered,
88         "norm_structured": norm_structured,
89         "covariance_structured": covariance_structured,
90         "specificity": procedure,
91         specificity_label: procedure,
92         "semantic_twin_stable": twin_stable,
93     }
94     if not replay:
95         status = "ADAPTER_INVALID"
96     elif not necessity:
97         status = "LATENT_UNUSED"
98     elif not ordered:
99         status = "LATENT_UNORDERED"
100     elif not structured:
101         status = "LATENT_MAGNITUDE_ONLY"
102     elif not twin_stable:
103         status = "SURFACE_UNSTABLE"
104     elif not procedure:
105         status = "LATENT_GENERIC"
106     else:
107         status = "PROCEDURE_SPECIFIC" if config.specificity_claim == "procedure" else "CONTENT_SPECIFIC"
108     return status, gates
109 
110 
111 def run_causal_audit(
112     adapter: LatentReasonerAdapter,
113     records: Sequence[TaskRecord],
114     config: AuditConfig,
115     donor_plan: Sequence[DonorAssignment] | None = None,
116     different_answer_plan: Sequence[DonorAssignment] | None = None,
117     covariance_reference: np.ndarray | None = None,
118     covariance_reference_receipt: Mapping[str, Any] | None = None,
119 ) -> dict[str, Any]:
120     if not records:
121         raise ValueError("Audit requires at least one task")
122     donor_plan = list(donor_plan or build_same_answer_far_proof_plan(records))
123     different_answer_plan = list(different_answer_plan or build_different_answer_far_proof_plan(records))
124     validate_donor_plan(records, donor_plan, expected_relation="same")
125     validate_donor_plan(records, different_answer_plan, expected_relation="different")
126     by_id = {record.task_id: record for record in records}
127     donor_by_recipient = {assignment.recipient_id: assignment for assignment in donor_plan}
128     different_by_recipient = {assignment.recipient_id: assignment for assignment in different_answer_plan}
129 
130     clean: dict[str, TraceObservation] = {}
131     twins: dict[str, TaskRecord] = {}
132     twin_clean: dict[str, TraceObservation] = {}
133     for record in records:
134         clean[record.task_id] = adapter.capture(record)
135         twin = semantic_twin(record, 0)
136         validate_semantic_twin(record, twin)
137         twins[record.task_id] = twin
138         twin_clean[record.task_id] = adapter.capture(twin)
139 
140     if config.run_covariance_noise:
141         covariance = (
142             np.asarray(covariance_reference, dtype=np.float64)
143             if covariance_reference is not None
144             else covariance_from_traces(
145                 np.asarray(observation.trace, dtype=np.float64) for observation in clean.values()
146             )
147         )
148         covariance_source = "calibration_support" if covariance_reference is not None else "audit_support_fallback"
149     else:
150         covariance = None
151         covariance_source = "disabled"
152 
153     rows: list[dict[str, Any]] = []
154     for record in records:
155         ordinary = clean[record.task_id]
156         trace = np.asarray(ordinary.trace, dtype=np.float64)
157         assignment = donor_by_recipient[record.task_id]
158         different_assignment = different_by_recipient[record.task_id]
159         donor_trace = clean[assignment.donor_id].trace
160         different_trace = clean[different_assignment.donor_id].trace
161         replay = _score(adapter, record, trace)
162         zero = _score(adapter, record, zero_trace(trace))
163         permuted = _score(adapter, record, permute_trace(trace, record.task_id, config.seed))
164         norm_noise = _score(adapter, record, norm_matched_noise(trace, record.task_id, config.seed))
165         covariance_noise = None
166         if covariance is not None:
167             covariance_noise = _score(
168                 adapter,
169                 record,
170                 covariance_matched_noise(trace, covariance, record.task_id, config.seed),
171             )
172         donor = _score(adapter, record, donor_trace)
173         different_donor = _score(adapter, record, different_trace)
174         twin = _score(adapter, record, twin_clean[record.task_id].trace)
175 
176         truncation_losses: list[float] = []
177         first_near_clean_step: int | None = None
178         if config.truncation:
179             truncation_losses = [
180                 _score(adapter, record, truncate_trace(trace, retained)).answer_loss
181                 for retained in range(trace.shape[0] + 1)
182             ]
183             tolerance = max(0.005, 0.05 * abs(float(ordinary.answer_loss)))
184             for retained, loss in enumerate(truncation_losses):
185                 if loss <= float(ordinary.answer_loss) + tolerance:
186                     first_near_clean_step = retained
187                     break
188 
189         row = {
190             "task_id": record.task_id,
191             "group_id": record.group_id,
192             "orbit_id": record.orbit_id,
193             "answer": record.answer,
194             "proof_depth": record.proof_depth,
195             "donor_id": assignment.donor_id,
196             "different_answer_donor_id": different_assignment.donor_id,
197             "donor_signature_distance": assignment.signature_distance,
198             "different_donor_signature_distance": different_assignment.signature_distance,
199             "clean_loss": ordinary.answer_loss,
200             "replay_loss": replay.answer_loss,
201             "zero_loss": zero.answer_loss,
202             "permuted_loss": permuted.answer_loss,
203             "norm_noise_loss": norm_noise.answer_loss,
204             "covariance_noise_loss": None if covariance_noise is None else covariance_noise.answer_loss,
205             "procedure_donor_loss": donor.answer_loss,
206             "different_answer_donor_loss": different_donor.answer_loss,
207             "twin_donor_loss": twin.answer_loss,
208             "replay_max_abs": _max_logit_error(ordinary, replay),
209             "necessity_effect": zero.answer_loss - ordinary.answer_loss,
210             "order_effect": permuted.answer_loss - ordinary.answer_loss,
211             "norm_structure_effect": norm_noise.answer_loss - ordinary.answer_loss,
212             "covariance_structure_effect": None if covariance_noise is None else covariance_noise.answer_loss - ordinary.answer_loss,
213             "procedure_transplant_effect": donor.answer_loss - ordinary.answer_loss,
214             "different_answer_transplant_effect": different_donor.answer_loss - ordinary.answer_loss,
215             "twin_transplant_effect": twin.answer_loss - ordinary.answer_loss,
216             "truncation_losses": truncation_losses,
217             "first_near_clean_step": first_near_clean_step,
218             "predicted_answer": ordinary.predicted_answer,
219             "generated_correct": ordinary.generated_correct,
220         }
221         rows.append(row)
222 
223     summary: dict[str, Any] = {
224         "task_count": len(rows),
225         "specificity_claim": config.specificity_claim,
226         "covariance_source": covariance_source,
227         "covariance_reference_receipt": None if covariance_reference_receipt is None else dict(covariance_reference_receipt),
228         "replay_max_abs": max(float(row["replay_max_abs"]) for row in rows),
229         "replay_loss_max_abs": max(abs(float(row["replay_loss"]) - float(row["clean_loss"])) for row in rows),
230         "twin_max_abs": max(abs(float(row["twin_transplant_effect"])) for row in rows),
231         "latent_accuracy": _accuracy([(record, clean[record.task_id]) for record in records]),
232         "necessity": summarize_effect(rows, "necessity_effect", replicates=config.bootstrap_replicates, seed=config.seed + 1),
233         "order": summarize_effect(rows, "order_effect", replicates=config.bootstrap_replicates, seed=config.seed + 2),
234         "norm_structure": summarize_effect(rows, "norm_structure_effect", replicates=config.bootstrap_replicates, seed=config.seed + 3),
235         "covariance_structure": None,
236         "procedure_transplant": summarize_effect(
237             rows, "procedure_transplant_effect", replicates=config.bootstrap_replicates, seed=config.seed + 4
238         ),
239         "different_answer_transplant": summarize_effect(
240             rows, "different_answer_transplant_effect", replicates=config.bootstrap_replicates, seed=config.seed + 5
241         ),
242         "twin_transplant": summarize_effect(
243             rows, "twin_transplant_effect", replicates=config.bootstrap_replicates, seed=config.seed + 6
244         ),
245         "mean_first_near_clean_step": None,
246     }
247     summary["specificity_transplant"] = dict(summary["procedure_transplant"])
248     if covariance is not None:
249         summary["covariance_structure"] = summarize_effect(
250             rows,
251             "covariance_structure_effect",
252             replicates=config.bootstrap_replicates,
253             seed=config.seed + 7,
254         )
255     retained = [row["first_near_clean_step"] for row in rows if row["first_near_clean_step"] is not None]
256     if retained:
257         summary["mean_first_near_clean_step"] = float(np.mean(retained))
258 
259     mode_controls: dict[str, Any] = {}
260     if config.run_mode_controls:
261         # Explicit CoT uses the gold proof steps and is therefore reserved as a
262         # post-seal witness. Only the direct-answer control is admissible here.
263         for mode in ("direct",):
264             if not adapter.supports_mode(mode):
265                 mode_controls[mode] = {"available": False}
266                 continue
267             observations: list[tuple[TaskRecord, TraceObservation]] = []
268             losses: list[float] = []
269             try:
270                 for record in records:
271                     observation = adapter.capture_mode(record, mode)
272                     observations.append((record, observation))
273                     losses.append(float(observation.answer_loss))
274                 mode_controls[mode] = {
275                     "available": True,
276                     "mean_answer_loss": float(np.mean(losses)),
277                     "accuracy": _accuracy(observations),
278                 }
279             except UnsupportedAdapterOperation:
280                 mode_controls[mode] = {"available": False}
281         mode_controls["verbalized"] = {
282             "available": False,
283             "reason": "reserved_for_post_seal_witness",
284             "used_before_semantic_seal": False,
285         }
286     summary["mode_controls"] = mode_controls
287 
288     status, gates = classify(summary, config)
289     summary["status"] = status
290     summary["gates"] = gates
291 
292     result_without_digest = {
293         "schema": "sric.real-lrm.causal-audit-result.v2",
294         "claim_level": "external-latent-causal-admission",
295         "adapter": dict(adapter.model_fingerprint()),
296         "config": config.to_mapping(),
297         "same_answer_donor_plan": [asdict(item) for item in donor_plan],
298         "different_answer_donor_plan": [asdict(item) for item in different_answer_plan],
299         "summary": summary,
300         "rows": rows,
301         "claim_boundary": {
302             "real_model_semantics_established": False,
303             "specificity_claim": config.specificity_claim,
304             "prontoqa_procedure_algorithm_claim": False,
305             "endogenous_ontology_established": False,
306             "word_for_word_thought_transcript": False,
307         },
308     }
309     return {**result_without_digest, "result_sha256": sha256_object(result_without_digest)}

Back to source controls

Back to SRIC: external semantic admission