| """CPU-only post-replay audits. Emits aggregates, never prompts, IDs or tensors.""" |
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import json |
| import math |
| from pathlib import Path |
|
|
| from .assembly_paths import AssemblyPaths |
|
|
|
|
| def sha(path: Path) -> str: |
| with path.open('rb') as f: |
| return hashlib.file_digest(f, 'sha256').hexdigest() |
|
|
|
|
| def aggregate(root: Path) -> dict[str, object]: |
| import torch |
| from safetensors import safe_open |
| manifest = json.loads((root/'MANIFEST.json').read_text()) |
| n = zeros = over_small = over_limit = 0 |
| minimum, error_max = math.inf, 0.0 |
| for split in manifest['splits'].values(): |
| for entry in split['files']: |
| path = root/entry['file'] |
| if sha(path) != entry['sha256']: |
| raise ValueError('shard hash mismatch before S25 aggregate') |
| with safe_open(path, framework='pt', device='cpu') as f: |
| lp, resid = f.get_tensor('top_lp').double(), f.get_tensor('resid').double() |
| if not torch.isfinite(lp).all() or not torch.isfinite(resid).all(): |
| raise ValueError('nonfinite teacher data') |
| if (resid < 0).any() or (lp > 0).any(): |
| raise ValueError('invalid teacher distribution') |
| error = (lp.exp().sum(-1)+resid-1).abs() |
| n += resid.numel() |
| zeros += int((resid == 0).sum()) |
| minimum = min(minimum, float(lp.min())) |
| error_max = max(error_max, float(error.max())) |
| over_small += int((error > 1e-6).sum()) |
| over_limit += int((error > 1e-4).sum()) |
| if not n or over_limit: |
| raise ValueError('empty population or probability mass error above schema tolerance') |
| return {'schema': 'd2.s25-aggregate.v1', 'classification': 'MEASURED', |
| 'synthetic': bool(manifest.get('synthetic') or manifest.get('generation', {}).get('synthetic') |
| or manifest.get('checks', {}).get('synthetic') or manifest.get('taps_source') == 'synthetic' |
| or manifest.get('checks', {}).get('taps_source') == 'synthetic'), |
| 'manifest_sha256': sha(root/'MANIFEST.json'), 'span_positions': n, |
| 'resid_eq_zero': zeros, 'min_top_lp': minimum, |
| 'abs_sum_p_plus_resid_minus_1': {'max': error_max, 'count_gt_1e-6': over_small, 'count_gt_1e-4': over_limit}} |
|
|
|
|
| def k2_complete(root: Path, work: Path, split: str) -> None: |
| manifest = json.loads((root/'MANIFEST.json').read_text()) |
| expected: dict[str, set[int]] = {} |
| for entry in manifest['splits'][split]['files']: |
| sidecar = json.loads((root/entry['sidecar']).read_text()) |
| for seq in sidecar['sequences']: |
| expected[seq['seq_id']] = set(range(seq['gen_start']-1, seq['L']-2)) |
| rows = [json.loads(line) for line in (work/(split+'.k2_drafts.jsonl')).read_text().splitlines()] |
| if len(rows) != len(expected) or {r['seq_id'] for r in rows} != expected.keys(): |
| raise ValueError('K2 deadline left incomplete sequences; resume g1 phase later under a new lease') |
| for row in rows: |
| if {int(p) for p in row['drafts']} != expected[row['seq_id']]: |
| raise ValueError('incomplete K2 block coverage') |
| if any(len(d) != 5 for d in row['drafts'].values()): |
| raise ValueError('invalid K2 draft shape') |
| print(json.dumps({'split': split, 'complete_sequences': len(expected)})) |
|
|
|
|
| def context_ready(context: Path, manifest: Path, plan: Path, *, paths: AssemblyPaths | None = None, |
| mode: str = 'prefill_replay') -> None: |
| from .sequence_audit import validate_context |
| data = json.loads(context.read_text()) |
| pm = json.loads(manifest.read_text()) |
| receipts = data.get('evidence_receipts', []) |
| verified: set[str] = set() |
| for receipt in receipts: |
| path = paths.resolve(receipt['path']) if paths else Path(receipt['path']) |
| digest = sha(path) |
| if digest != receipt['sha256']: |
| raise ValueError('context evidence receipt hash differs') |
| verified.add(digest) |
| entries = [json.loads(line) for line in plan.read_text().splitlines()] |
| if not entries: |
| raise ValueError('empty replay plan') |
| for entry in entries: |
| c = data.get('per_sequence', {}).get(entry['seq_id'], data) |
| if c.get('status', '').startswith('INCOMPLETE') or 'UNVERIFIED' in json.dumps(c): |
| raise ValueError('incomplete provenance scaffold is not a build receipt') |
| if entry['kind'] == 'acap' and entry['seq_id'] not in data.get('per_sequence', {}): |
| raise ValueError('original acap context missing') |
| validate_context(c, pm, require_chat_match=entry['kind'] != 'acap', paths=paths, mode=mode) |
| keys = ('target', 'decode_capture_target' if mode == 'decode_capture' else 'replay_target') |
| if any(c[key]['receipt_sha256'] not in verified for key in keys): |
| raise ValueError('target receipt lacks accessible hash-bound evidence_receipts') |
| if mode == 'decode_capture': |
| matches = [r for r in receipts if r['sha256'] == c['decode_capture_target']['receipt_sha256']] |
| receipt_path = paths.resolve(matches[0]['path']) if paths else Path(matches[0]['path']) |
| attestation = json.loads(receipt_path.read_text()) |
| claims = dict(c['decode_capture_target']) |
| claims.pop('receipt_sha256') |
| target_claims = dict(c['target']) |
| target_claims.pop('receipt_sha256') |
| if (attestation.get('target') != target_claims |
| or any(attestation.get('identity', {}).get(label, {}).get('sha256') != c[label + '_sha256'] |
| for label in ('tokenizer', 'encoder', 'chat_template')) |
| or attestation.get('schema') != 'd2.pass2-attestation.v1' |
| or attestation.get('decode_capture_target') != claims |
| or any(attestation.get(k) != c.get(k) for k in ('sampling', 'formatting', 'generator'))): |
| raise ValueError('decode-capture target receipt claims differ') |
| print(json.dumps({'context_sha256': sha(context), 'sequences': len(entries), 'receipts_verified': len(verified)})) |
|
|
|
|
| def g1_gate(path: Path) -> None: |
| gate = json.loads(path.read_text()) |
| if gate.get('split') != 'acap' or gate.get('g1', {}).get('pass') is not True or gate['g1'].get('binding') is not True: |
| raise ValueError('binding acap G1 did not pass; pilot preparation refused') |
|
|
|
|
| def binding(root: Path, replay: Path, gate: Path, context: Path, prompts: Path) -> None: |
| from .dataset import check_gate_file |
| manifest = json.loads((root/'MANIFEST.json').read_text()) |
| g = json.loads(gate.read_text()) |
| if check_gate_file(g, str(replay)) or g['gate']['pass'] is not True: |
| raise ValueError('foreign/failed equivalence gate') |
| if (manifest['replay']['plan_sha256'] != sha(replay/'plan.jsonl') |
| or manifest['prompt_manifest_sha256'] != sha(prompts) |
| or manifest['generation']['sequence_context_sha256'] != sha(context) |
| or manifest['checks']['equivalence_gate']['sha256'] != sha(gate) |
| or manifest['checks']['skipped_sequences'] != 0 |
| or manifest['checks']['equivalence_gate'].get('override') |
| or manifest['checks']['g1_plumbing'].get('override') |
| or manifest.get('taps_source') != 'prefill_replay' |
| or manifest['generation'].get('synthetic') is not False): |
| raise ValueError('shards are not bound to these exact replay/context/gates or contain overrides/skips') |
| print(json.dumps({'manifest_sha256': sha(root/'MANIFEST.json'), 'binding': 'verified'})) |
|
|
|
|
| def main() -> None: |
| ap = argparse.ArgumentParser(description=__doc__) |
| ap.add_argument('check', choices=('aggregate', 'k2-complete', 'context', 'g1', 'binding')) |
| ap.add_argument('--root', type=Path, required=True) |
| ap.add_argument('--work', type=Path) |
| ap.add_argument('--split') |
| ap.add_argument('--out', type=Path) |
| ap.add_argument('--manifest', type=Path) |
| ap.add_argument('--plan', type=Path) |
| ap.add_argument('--replay', type=Path) |
| ap.add_argument('--gate', type=Path) |
| ap.add_argument('--context', type=Path) |
| a = ap.parse_args() |
| if a.check == 'aggregate': |
| if a.out is None: |
| raise ValueError('--out required') |
| with a.out.open('x') as f: |
| json.dump(aggregate(a.root), f, indent=2, allow_nan=False) |
| elif a.check == 'k2-complete': |
| if a.work is None or a.split is None: |
| raise ValueError('--work and --split required') |
| k2_complete(a.root, a.work, a.split) |
| elif a.check == 'context': |
| if a.manifest is None or a.plan is None: |
| raise ValueError('--manifest and --plan required') |
| context_ready(a.root, a.manifest, a.plan) |
| elif a.check == 'binding': |
| if None in (a.replay, a.gate, a.context, a.manifest): |
| raise ValueError('--replay, --gate, --context, --manifest required') |
| binding(a.root, a.replay, a.gate, a.context, a.manifest) |
| else: |
| g1_gate(a.root) |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
