| """Read-only exact-code conflict audit of manifest-selected TRAIN/VALID rows.""" |
| import argparse |
| import hashlib |
| import json |
| from collections import Counter, defaultdict |
| from pathlib import Path |
|
|
| import pandas as pd |
|
|
| from src.contracts.prompt import code_for_sample |
|
|
|
|
| def digest(value): |
| return hashlib.sha256(value.encode()).hexdigest() |
|
|
|
|
| def audit_selection(selection, frame): |
| groups, normalized, pairs = defaultdict(list), defaultdict(list), defaultdict(list) |
| seen = set() |
| for ref in selection: |
| ident = ref['prompt_id'] |
| if ident in seen: |
| raise ValueError('duplicate selected ID') |
| seen.add(ident) |
| row = frame.loc[ident].to_dict() |
| if int(row['target']) != ref['target'] or row['pair_id'] != ref['pair_id']: |
| raise ValueError('manifest label/pair mismatch') |
| code = code_for_sample(row) |
| item = dict(sample_id=ident, pair_id=ref['pair_id'], target=ref['target']) |
| fingerprint = digest(code) |
| groups[fingerprint].append(item) |
| normalized[digest(''.join(code.split()))].append(item) |
| pairs[ref['pair_id']].append((ref['target'], fingerprint)) |
| if not pairs or any(sorted(label for label, _ in rows) != [0, 1] for rows in pairs.values()): |
| raise ValueError('empty or incomplete pair population') |
| conflicts = {h: rows for h, rows in groups.items() if len({r['target'] for r in rows}) > 1} |
| unavoidable = sum(min(Counter(r['target'] for r in rows).values()) for rows in conflicts.values()) |
| return dict(rows=len(seen), pairs=len(pairs), unique_exact_code=len(groups), |
| exact_opposite_label_groups=len(conflicts), |
| rows_in_exact_conflicts=sum(map(len, conflicts.values())), |
| exact_identical_within_pair=sum(rows[0][1] == rows[1][1] for rows in pairs.values()), |
| deterministic_code_only_accuracy_ceiling=1-unavoidable/len(seen), |
| whitespace_removed_opposite_label_groups_screen_only=sum( |
| len({r['target'] for r in rows}) > 1 for rows in normalized.values()), |
| conflict_examples=list(conflicts.values())[:10], |
| exact_conflict_groups=conflicts), groups |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument('--manifest', type=Path, required=True) |
| parser.add_argument('--corpus', type=Path, default=Path('data/processed/corpus.parquet')) |
| parser.add_argument('--output', type=Path, required=True) |
| args = parser.parse_args() |
| if args.output.resolve() in {args.manifest.resolve(), args.corpus.resolve()}: |
| raise ValueError('output must not overwrite evidence') |
| manifest = json.loads(args.manifest.read_text()) |
| corpus_hash = hashlib.sha256(args.corpus.read_bytes()).hexdigest() |
| if corpus_hash != manifest['input_sha256']['data/processed/corpus.parquet']: |
| raise ValueError('corpus differs from frozen snapshot') |
| frame = pd.read_parquet(args.corpus).set_index('sample_id', verify_integrity=True) |
| result = dict(scope='Frozen TRAIN and VALID selections only; no TEST selection or evaluation', |
| corpus_sha256=corpus_hash, |
| manifest_sha256=hashlib.sha256(args.manifest.read_bytes()).hexdigest(), |
| dataset_snapshot_hash=manifest['dataset_snapshot_hash'], |
| limitations=[ |
| 'Exact source-code equality only; whitespace screening is not semantic equivalence.', |
| 'Does not inspect bounded/tokenized prompts or missing repository context.', |
| 'Consistency does not establish that labels are semantically correct.']) |
| groups = {} |
| for split in ('train', 'valid'): |
| result[split], groups[split] = audit_selection(manifest[split+'_selection'], frame) |
| shared = set(groups['train']) & set(groups['valid']) |
| result['cross_split_exact_code_overlap'] = dict( |
| hash_groups=len(shared), train_rows=sum(len(groups['train'][h]) for h in shared), |
| valid_rows=sum(len(groups['valid'][h]) for h in shared), |
| opposite_label_groups=sum(len({r['target'] for r in groups['train'][h]+groups['valid'][h]}) > 1 for h in shared)) |
| args.output.parent.mkdir(parents=True, exist_ok=True) |
| args.output.write_text(json.dumps(result, indent=2)+'\n') |
| print(args.output) |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
