pipeline.py

SRIC: external semantic admission

pipeline.py

Download original file

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

Read the typeset paper

Skip source text

1 from __future__ import annotations
2 
3 from typing import Any, Sequence
4 import numpy as np
5 
6 from .adapters.base import LatentReasonerAdapter, UnsupportedAdapterOperation
7 from .audit import run_causal_audit
8 from .dataset import deterministic_calibration_audit_split
9 from .metrics import covariance_from_traces
10 from .hashing import sha256_object
11 from .schemas import AuditConfig, TaskRecord
12 from .semantic import run_semantic_escalation
13 
14 
15 CAUSAL_TERMINAL = {
16     "ADAPTER_INVALID",
17     "LATENT_UNUSED",
18     "LATENT_UNORDERED",
19     "LATENT_MAGNITUDE_ONLY",
20     "SURFACE_UNSTABLE",
21     "LATENT_GENERIC",
22 }
23 
24 
25 
26 def _post_seal_explicit_cot_witness(
27     adapter: LatentReasonerAdapter,
28     records: Sequence[TaskRecord],
29 ) -> dict[str, Any]:
30     """Evaluate explicit-CoT mode only after the semantic quotient is sealed."""
31     if not adapter.supports_mode("verbalized"):
32         return {"available": False, "reason": "adapter_unsupported", "used_during_construction": False}
33     losses: list[float] = []
34     correctness: list[bool] = []
35     try:
36         for record in records:
37             observation = adapter.generate_mode(record, "verbalized")
38             losses.append(float(observation.answer_loss))
39             if observation.predicted_answer is not None:
40                 correctness.append(
41                     observation.predicted_answer.strip().lower() == record.answer.strip().lower()
42                 )
43     except UnsupportedAdapterOperation:
44         return {"available": False, "reason": "adapter_unsupported", "used_during_construction": False}
45     return {
46         "available": True,
47         "task_count": len(losses),
48         "mean_answer_loss": float(np.mean(losses)) if losses else None,
49         "accuracy": float(np.mean(correctness)) if correctness else None,
50         "used_during_construction": False,
51         "role": "post_seal_comparison_witness_only",
52     }
53 
54 def run_block24_pipeline(
55     adapter: LatentReasonerAdapter,
56     records: Sequence[TaskRecord],
57     config: AuditConfig,
58 ) -> dict[str, Any]:
59     """Run causal admission and conditionally escalate to semantic construction.
60 
61     Covariance-matched corruption is calibrated on a deterministic group-disjoint
62     support before the causal audit. The semantic constructor receives only the
63     causal-audit support.
64     """
65     causal_records = list(records)
66     covariance_reference = None
67     covariance_receipt: dict[str, Any] | None = None
68     if config.run_covariance_noise:
69         causal_split = deterministic_calibration_audit_split(
70             records,
71             seed=config.seed,
72             calibration_fraction=config.causal_calibration_fraction,
73         )
74         calibration_records = causal_split["calibration"]
75         causal_records = causal_split["audit"]
76         calibration_traces = [adapter.capture(record).trace for record in calibration_records]
77         covariance_reference = covariance_from_traces(calibration_traces)
78         covariance_receipt = {
79             "schema": "sric.real-lrm.covariance-calibration-receipt.v1",
80             "task_ids": [record.task_id for record in calibration_records],
81             "task_ids_sha256": sha256_object([record.task_id for record in calibration_records]),
82             "group_ids": sorted({record.group_id for record in calibration_records}),
83             "covariance_sha256": sha256_object(np.round(covariance_reference, 12).tolist()),
84             "final_audit_excluded": True,
85         }
86     causal = run_causal_audit(
87         adapter,
88         causal_records,
89         config,
90         covariance_reference=covariance_reference,
91         covariance_reference_receipt=covariance_receipt,
92     )
93     causal_status = str(causal["summary"]["status"])
94     if causal_status not in {"CONTENT_SPECIFIC", "PROCEDURE_SPECIFIC"}:
95         semantic: dict[str, Any] = {
96             "status": "SEMANTIC_ESCALATION_NOT_ADMISSIBLE",
97             "reason": causal_status,
98         }
99         overall = causal_status
100     elif not config.semantic_escalation:
101         semantic = {
102             "status": "SEMANTIC_ESCALATION_DISABLED",
103             "reason": "configuration",
104         }
105         overall = causal_status
106     else:
107         semantic = run_semantic_escalation(adapter, causal_records, config)
108         if semantic.get("semantic_seal") is not None:
109             semantic["post_seal_explicit_cot_witness"] = _post_seal_explicit_cot_witness(
110                 adapter, causal_records
111             )
112             if isinstance(semantic.get("chronology"), list):
113                 semantic["chronology"].append("explicit_cot_witness_revealed")
114         else:
115             semantic["post_seal_explicit_cot_witness"] = {
116                 "available": False,
117                 "reason": "no_semantic_quotient_seal",
118                 "used_during_construction": False,
119             }
120         semantic_status = str(semantic["status"])
121         overall = causal_status if semantic_status.startswith("SEMANTIC_ESCALATION_") else semantic_status
122 
123     payload = {
124         "schema": "sric.block24.external-lrm-causal-semantic-admission-result.v1",
125         "overall_status": overall,
126         "adapter": dict(adapter.model_fingerprint()),
127         "config": config.to_mapping(),
128         "causal_calibration": covariance_receipt,
129         "causal_admission": causal,
130         "semantic_escalation": semantic,
131         "claim_boundary": {
132             "real_checkpoint_evidence": not bool(adapter.model_fingerprint().get("engineered_positive_control", False)),
133             "maximum_possible_status": "CAUSALLY_VALIDATED",
134             "causal_specificity_tier": config.specificity_claim,
135             "word_for_word_hidden_thought_transcript": False,
136             "open_ended_recursive_self_improvement": False,
137         },
138     }
139     return {**payload, "result_sha256": sha256_object(payload)}

Back to source controls

Other PDFs for this work

Original file and source record

This copy matches the recorded source file. Its publication here does not change the work’s review status.

SHA-256: b06ad7f20ba0ebf0c6a55d53ce86d10581b900c967f6b86369193e64ebd89ed7

Original source on GitHub

Back to SRIC: external semantic admission