#!/usr/bin/env python3 """Deterministic, preservation-first pre-ingestion gate for the VPL R05 intake. Standard-library only. No model, network, scoring, semantic matching or source edits. Exact payload equality is NOT proof of shared authorship or independent execution. Original IDs/rows/provenance remain in every emitted ledger. Nothing is deleted. """ from __future__ import annotations import argparse import copy import hashlib import json import re import sys from collections import defaultdict from pathlib import Path from typing import Any POLICY_ID = 'ACTION4_EXACT_PAYLOAD_PREINGEST_R01' SLOTS = frozenset({'CLAUDE', 'GEMINI', 'GROK', 'DEEPSEEK', 'GLM', 'KIMI'}) COLLECTIONS = { 'new_findings': 'NEW_FINDING', 'prior_finding_dispositions': 'PRIOR_DISPOSITION', 'prior_dispositions': 'PRIOR_DISPOSITION', 'prior_n_rows': 'PRIOR_DISPOSITION', 'independently_confirmed_peer_limitations': 'PEER_CONFIRMATION', } CATEGORIES = ('NEW_FINDING', 'PRIOR_DISPOSITION', 'PEER_CONFIRMATION') # For NEW findings only, the identifier is excluded from grouping. Prior/peer # reference identities remain part of the group key. In particular: retain # severity, disposition, evidence, location, stage, all prose and list ordering. ID_FIELDS = frozenset({'id', 'finding_id'}) class AdmissionError(ValueError): """Malformed/ambiguous input must not create a publishable derived ledger.""" def require(condition: bool, message: str) -> None: if not condition: raise AdmissionError(message) def canonical(value: Any) -> bytes: """This project's JSON serialization, not a claim of universal JSON canonicalization.""" try: return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(',', ':'), allow_nan=False).encode('utf-8') except (TypeError, ValueError, UnicodeError) as exc: raise AdmissionError(f'NON_CANONICAL_JSON: {exc}') from exc def sha(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def strict_json(data: bytes) -> Any: def pairs(items): result = {} for key, val in items: require(key not in result, f'DUPLICATE_JSON_KEY: {key}') result[key] = val return result def bad_constant(value): raise AdmissionError(f'NON_FINITE_JSON: {value}') try: return json.loads(data.decode('utf-8'), object_pairs_hook=pairs, parse_constant=bad_constant) except (UnicodeError, json.JSONDecodeError) as exc: raise AdmissionError(f'INVALID_JSON: {exc}') from exc def nonblank(value: Any, field: str) -> str: require(isinstance(value, str) and bool(value.strip()), f'INVALID_{field}') return value def row_id(row: dict) -> str: present = [row[k] for k in ID_FIELDS if k in row] require(bool(present), 'MISSING_FINDING_ID') for value in present: nonblank(value, 'FINDING_ID') require(len(set(present)) == 1, 'CONFLICTING_ID_ALIASES') return present[0] def source_context(intake: dict) -> dict: source = intake.get('source') require(isinstance(source, dict), 'MISSING_SOURCE_CONTEXT') repo = nonblank(source.get('repository'), 'REPOSITORY') commit = source.get('commit') require(isinstance(commit, str) and re.fullmatch('[0-9a-f]{40}', commit) is not None, 'INVALID_SOURCE_COMMIT') return {'repository': repo, 'commit': commit} def collect_rows(intake: dict) -> tuple[list[dict], list[dict], dict]: require(isinstance(intake, dict), 'INTAKE_NOT_OBJECT') require(intake.get('kind') == 'V5_FRESH_R03_CONSOLIDATED_INTAKE', 'WRONG_INTAKE_KIND') require(isinstance(intake.get('reviewer_records'), dict), 'MISSING_REVIEWER_RECORDS') require(set(intake['reviewer_records']) == SLOTS, 'REVIEWER_COVERAGE_MISMATCH') context = source_context(intake) raw_records, coverage = [], [] for slot in sorted(SLOTS): record = intake['reviewer_records'][slot] require(isinstance(record, dict), f'INVALID_REVIEWER_RECORD: {slot}') require('submitted_ledger' in record, f'MISSING_LEDGER_FIELD: {slot}') ledger = record['submitted_ledger'] if ledger is None: reason = nonblank(record.get('ledger_availability'), 'MISSING_LEDGER_REASON') coverage.append({'slot': slot, 'ledger_state': 'NOT_AVAILABLE', 'reason': reason, 'rows': None, 'not_a_clean_review': True}) continue require(isinstance(ledger, dict), f'INVALID_LEDGER: {slot}') require(ledger.get('reviewer_slot') == slot, f'REVIEWER_NAMESPACE_MISMATCH: {slot}') require('new_findings' in ledger, f'MISSING_NEW_FINDINGS: {slot}') target = ledger.get('target') if target is not None: require(isinstance(target, dict), f'INVALID_TARGET: {slot}') if 'source_commit' in target: require(target['source_commit'] == context['commit'], f'MIXED_SOURCE_COMMIT: {slot}') if 'repository' in target: require(target['repository'] == context['repository'], f'MIXED_REPOSITORY: {slot}') # Reject unrecognized finding/disposition row arrays, rather than ignore them. for name, value in ledger.items(): if isinstance(value, list) and any(t in name.lower() for t in ('finding', 'disposition')): require(name in COLLECTIONS, f'UNSUPPORTED_ROW_COLLECTION: {slot}/{name}') counts, seen_identity = defaultdict(int), {} for collection, category in COLLECTIONS.items(): if collection not in ledger: continue array = ledger[collection] require(isinstance(array, list), f'INVALID_ARRAY: {slot}/{collection}') for index, row in enumerate(array): require(isinstance(row, dict), f'INVALID_ROW: {slot}/{collection}/{index}') identifier = row_id(row) payload = {k: v for k, v in row.items() if k not in ID_FIELDS} require(bool(payload), f'EMPTY_FINDING_PAYLOAD: {slot}/{identifier}') if category == 'NEW_FINDING': require(any(isinstance(row.get(k), str) and row[k].strip() for k in ('title', 'finding', 'description')), f'MISSING_FINDING_CLAIM: {slot}/{identifier}') payload_bytes = canonical(payload) identity = (category, identifier) if identity in seen_identity: require(seen_identity[identity] == payload_bytes, f'CONFLICTING_ID_WITHIN_REVIEWER: {slot}/{identifier}') seen_identity[identity] = payload_bytes pointer = f'/reviewer_records/{slot}/submitted_ledger/{collection}/{index}' raw_records.append({ 'occurrence_ref': pointer, 'slot': slot, 'category': category, 'collection': collection, 'original_id': identifier, 'original_row': copy.deepcopy(row), 'original_row_canonical_sha256': sha(canonical(row)), 'payload_sha256': sha(payload_bytes), 'source_context': context.copy(), 'provenance': { 'round': record.get('round'), 'source_review_commit': record.get('source_review_commit'), 'stage1_commit': record.get('stage1_commit'), 'declared_session': ledger.get('session', ledger.get('audit_run_id')), 'remote_ledger_sha256_as_recorded': record.get('ledger_sha256', record.get('ledger_git_sha256')), 'embedded_ledger_canonical_sha256': sha(canonical(ledger)), 'assurance': 'BOUND_TO_INPUT_SNAPSHOT_NOT_NEW_PROVIDER_ATTESTATION', }, }) counts[category] += 1 coverage.append({'slot': slot, 'ledger_state': 'PRESENT', 'rows': sum(counts.values()), 'by_category': dict(sorted(counts.items())), 'intake_status_as_recorded': record.get('intake_status')}) require(bool(raw_records), 'NO_ROWS_TO_INGEST') return raw_records, coverage, context def deduplicate_before_ingestion(raw_records: list[dict]) -> tuple[list[dict], list[dict]]: """Group before counting. Compare actual bytes as well as their digest. Never merge by title/topic, reviewer name, source ID alone, supplied duplicate flags, or inferred meaning. A different evidence reference changes the payload. """ buckets = defaultdict(list) for item in raw_records: payload = {k: v for k, v in item['original_row'].items() if k not in ID_FIELDS} # A generic 'OPEN' disposition on N01 is not the disposition on N02. # Ignore ID only for substantive NEW_FINDING payload copies, never for # references/dispositions about an identified pre-existing finding. key = canonical({'category': item['category'], 'source': item['source_context'], 'referenced_id': None if item['category'] == 'NEW_FINDING' else item['original_id'], 'payload': payload}) buckets[key].append(item) groups, crosswalk = [], [] for key in sorted(buckets): members = sorted(buckets[key], key=lambda x: x['occurrence_ref']) group_id = 'A4-' + sha(key) # Representative is a storage choice, explicitly not a human authorship claim. rep = members[0] ids = sorted({x['original_id'] for x in members}) group = { 'group_id': group_id, 'category': rep['category'], 'payload_sha256': rep['payload_sha256'], 'payload': {k: v for k, v in rep['original_row'].items() if k not in ID_FIELDS}, 'representative_ref': rep['occurrence_ref'], 'representative_is_not_authorship_claim': True, 'occurrences': [x['occurrence_ref'] for x in members], 'source_ids': ids, 'reviewer_slots': sorted({x['slot'] for x in members}), 'duplicate_occurrences_excluded_from_new_content_count': len(members) - 1, 'additional_independent_votes_from_repetition': 0, 'independent_reproduction_status': 'NOT_ESTABLISHED_BY_DEDUPLICATION', 'content_admission_only': True, } groups.append(group) for i, row in enumerate(members): crosswalk.append({ 'occurrence_ref': row['occurrence_ref'], 'slot': row['slot'], 'original_id': row['original_id'], 'category': row['category'], 'group_id': group_id, 'content_count_contribution': 1 if i == 0 else 0, 'classification': 'REPRESENTATIVE_CONTENT' if i == 0 else 'EXACT_PAYLOAD_REPEAT', 'duplicate_additional_independent_vote_credit': 0, 'raw_record_retained': True, }) return sorted(groups, key=lambda x: x['group_id']), sorted(crosswalk, key=lambda x: x['occurrence_ref']) def build_ledger(intake: dict, input_sha256: str) -> dict: require(isinstance(input_sha256, str) and re.fullmatch('[0-9a-f]{64}', input_sha256) is not None, 'INVALID_INPUT_HASH') # Validate the entire preserved JSON tree, including otherwise uncounted metadata. canonical(intake) raw, coverage, context = collect_rows(intake) groups, crosswalk = deduplicate_before_ingestion(raw) # No tally is accepted before the gate: counts come ONLY from admitted groups. counts = {} for category in CATEGORIES: rows_n = sum(r['category'] == category for r in raw) groups_n = sum(g['category'] == category for g in groups) counts[category] = {'submitted_rows': rows_n, 'distinct_payload_groups': groups_n, 'repeat_occurrences_not_added_to_content_count': rows_n - groups_n} require(len(crosswalk) == len(raw), 'CROSSWALK_COVERAGE_ERROR') require(sum(x['content_count_contribution'] for x in crosswalk) == len(groups), 'COUNT_ERROR') # Same IDs with different conclusions across reviewers are retained as variants. variants = defaultdict(list) for item in raw: variants[(item['category'], item['original_id'])].append(item) id_variants = [] for (category, identifier), members in sorted(variants.items()): if len({x['payload_sha256'] for x in members}) > 1: id_variants.append({'category': category, 'source_id': identifier, 'status': 'DISTINCT_PAYLOADS_PRESERVED_NOT_ADJUDICATED', 'occurrences': sorted(x['occurrence_ref'] for x in members)}) return { 'kind': 'VPL_ACTION4_PREINGESTED_LEDGER', 'policy_id': POLICY_ID, 'input_sha256': input_sha256, 'input_revision': intake.get('revision'), 'source_context': context, 'scope': 'Reprocess historical FRESH_R03 inputs that fed HF03; no findings re-reviewed or closed.', 'gate': 'EXACT_PAYLOAD_DEDUPLICATION_BEFORE_DERIVED_LEDGER_COUNTS', 'rules': { 'new_finding_id_fields_excluded_from_payload_comparison': sorted(ID_FIELDS), 'prior_and_peer_reference_ids_included_in_group_key': True, 'canonicalization': 'JSON sorted object keys, UTF-8, compact separators; no prose/list normalization', 'all_other_payload_fields_compared': True, 'supplied_duplicate_flags_used_as_proof': False, 'similarity_or_id_only_merge': False, 'source_truth_or_authorship_verified': False, 'automatic_independent_corroboration_award': False, }, 'counts': counts, 'reviewer_coverage': coverage, 'raw_records': raw, 'admitted_groups': groups, 'duplicate_crosswalk': crosswalk, 'same_id_distinct_payload_variants': id_variants, 'original_intake': copy.deepcopy(intake), 'status': 'DERIVED_CONTENT_LEDGER_NOT_REVIEW_VERDICT_OR_SCORE', 'limitations': [ 'Exact payload comparison does not detect paraphrase/semantic duplicates.', 'Distinct payloads are not necessarily distinct underlying defects.', 'Distinct evidence references remain separate; their truth/independence requires evidence review.', 'Repeated identical payloads supply zero additional independent votes automatically.', 'The bound consolidated intake is the input; no fresh Git/remote reviewer revalidation is implied.', 'Missing ledger is recorded as unavailable, never as a clean review.', 'No frozen candidate, historical disposition, assessment, or live pipeline is overwritten.', ], } def serialize(value: Any) -> bytes: return (json.dumps(value, sort_keys=True, ensure_ascii=False, indent=2, allow_nan=False) + '\n').encode('utf-8') def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--input', type=Path, required=True) parser.add_argument('--expected-input-sha256', required=True) parser.add_argument('--output', type=Path, required=True, help='New output file only; existing files are never overwritten.') args = parser.parse_args(argv) try: require(not args.input.is_symlink(), 'INPUT_SYMLINK_NOT_ALLOWED') data = args.input.read_bytes() require(len(data) <= 20_000_000, 'INPUT_TOO_LARGE') require(sha(data) == args.expected_input_sha256, 'INPUT_HASH_MISMATCH') document = build_ledger(strict_json(data), sha(data)) require(args.output.resolve() != args.input.resolve(), 'OUTPUT_IS_INPUT') require(args.output.parent.is_dir(), 'OUTPUT_PARENT_MISSING') encoded = serialize(document) # finish serialization before creating any output # Exclusive create prevents accidental replacement of prior evidence. with args.output.open('xb') as stream: stream.write(encoded) print(json.dumps({'status': 'DERIVED_LEDGER_WRITTEN', 'counts': document['counts'], 'output_sha256': sha(args.output.read_bytes()), 'policy': POLICY_ID}, sort_keys=True)) return 0 except (AdmissionError, OSError, RecursionError) as exc: print(f'ADMISSION_STOP: {exc}', file=sys.stderr) return 2 if __name__ == '__main__': raise SystemExit(main())