"""Export reviewed decision payloads without changing the captured model view.""" from __future__ import annotations import argparse import hashlib import json from pathlib import Path import sys ROOT = Path(__file__).resolve().parents[1] sys.path[:0] = [str(ROOT/'eval'), str(ROOT/'eval/swe'), str(ROOT/'swefactory_self/results'), str(ROOT/'training/swe_world')] from evalkit.request_view import request_view, response_assistant, text_content from verify_native_alignment import load_response, following_results from run_instance_v3 import load_task_row from to_sft_qwen import Renderer, build_decision from tokenized_sft import validate_structure from native_routes import RouteAudit, validate_teacher_policy from native_task_identity import task_sha256 VERDICTS = frozenset({'keep', 'mask', 'quarantine_source'}) def validate_review(review: dict, call_requests: dict, request_count: int) -> dict: """A call belongs to one captured response, never to a reviewer supplied index.""" decision = review.get('decision') if decision not in {'keep', 'keep_with_masks', 'quarantine'}: raise ValueError('missing or unsupported review decision') reviews = review.get('tool_reviews') if not isinstance(reviews, list): raise ValueError('missing tool_reviews') seen, bad, contaminated = set(), set(), False reviewed_requests = set() for item in reviews: call_id, index = item.get('call_id'), item.get('request_index') if call_id in seen or call_id not in call_requests: raise ValueError('duplicate or unknown tool review') if type(index) is not int or call_requests[call_id] != index: raise ValueError('wrong review request_index') verdict = item.get('verdict') if verdict not in VERDICTS or not item.get('reason'): raise ValueError('missing/unknown verdict or reason') seen.add(call_id) reviewed_requests.add(index) if verdict == 'mask': bad.add(index) contaminated |= verdict == 'quarantine_source' if seen != set(call_requests): raise ValueError('incomplete tool review') masks = review.get('mask_requests', []) if (not isinstance(masks, list) or len(set(masks)) != len(masks) or any(type(i) is not int or not 1 <= i <= request_count for i in masks)): raise ValueError('invalid mask_requests') masks = set(masks) if contaminated and decision != 'quarantine': raise ValueError('quarantine_source contradicts top-level decision') if decision != 'quarantine' and not bad <= masks: raise ValueError('bad call missing from mask_requests') if (decision == 'keep' and masks) or (decision == 'keep_with_masks' and not masks): raise ValueError('mask_requests contradict top-level decision') return {**review, 'mask_requests': sorted(masks | bad)} def load_public_task(task_file: Path, instance_id: str) -> dict: """Use the solver's public JSONL contract without rewriting its bytes.""" if task_file is None: raise ValueError('Explicit public task fixture required') try: task = load_task_row(instance_id, Path(task_file)) except SystemExit as error: raise ValueError(str(error)) from error if any(not isinstance(task.get(k), str) or not task[k].strip() for k in ('instance_id', 'repo', 'base_commit', 'problem_statement', 'image')): raise ValueError('Incomplete public task fixture') return task def request_effort(request: dict) -> str: declared = [] if 'reasoning_effort' in request: declared.append(request['reasoning_effort']) reasoning = request.get('reasoning') if isinstance(reasoning, dict) and 'effort' in reasoning: declared.append(reasoning['effort']) if not declared: return 'not_applicable' if any(not isinstance(e, str) or not e.strip() or e != declared[0] for e in declared): raise ValueError('Missing or conflicting wire reasoning effort') return declared[0] def response_identity(path: Path, captured: dict) -> tuple[str, str]: events = [json.loads(line[5:]) for line in path.read_text(encoding='utf-8').splitlines() if line.startswith('data:') and line[5:].strip() != '[DONE]'] if captured['protocol'] == 'responses': records = [event['response'] for event in events if isinstance(event.get('response'), dict)] terminal = captured['response'] if any(not isinstance(terminal.get(k), str) or not terminal[k].strip() for k in ('id', 'model')): raise ValueError('Missing actual Responses ID/model identity') else: records = events identities = {} for key in ('id', 'model'): values = [record[key] for record in records if key in record] if (not values or any(not isinstance(value, str) or not value.strip() for value in values) or len(set(values)) != 1): raise ValueError(f'Missing or inconsistent actual response {key} identity') identities[key] = values[0] return identities['id'], identities['model'] def apply_review(sample: dict, review: dict, request_index: int) -> dict: result = dict(sample) if review['decision'] == 'quarantine': reason = 'REVIEW_QUARANTINE' elif request_index in review.get('mask_requests', []): reason = 'REVIEW_BAD_ACTION' elif review['decision'] in {'keep', 'keep_with_masks'}: reason = None else: raise ValueError('missing or unsupported review decision') if reason: result.update(mask=[0] * len(sample['ids']), keep_msgs=[], spans={}, masked_response_reason=reason) result['training_eligible'] = False return result def decision_rows(root: Path): for line in (root/'decisions-reviewed.jsonl').open(encoding='utf-8'): sample = json.loads(line) yield {'sample_id': sample['sample_id'], 'arm': sample['arm'], 'request_index': sample['request_index'], 'seq_tokens': len(sample['ids']), 'supervised_tokens': sum(sample['mask']), 'reason': sample['masked_response_reason'], 'training_eligible': False} def export(root: Path) -> dict: output = root/'decisions-reviewed.jsonl' if output.exists(): raise FileExistsError(output) review_path = root/'review.json' review = json.loads(review_path.read_text(encoding='utf-8')) selection = json.loads((root/'selection.json').read_text(encoding='utf-8')) alignment = json.loads((root/'token-alignment.json').read_text(encoding='utf-8')) if not alignment['passed']: raise ValueError('request/token alignment has not passed') renderer = Renderer() summary = {'training_eligible': False, 'review_sha256': hashlib.sha256(review_path.read_bytes()).hexdigest(), 'arms': {}} with output.open('w', encoding='utf-8', newline='\n') as handle: for arm in ('production', 'eval'): directory = root/arm events_path = directory/'session-events.json' arm_review = review['arms'][arm] if arm_review['source_events_sha256'] != hashlib.sha256(events_path.read_bytes()).hexdigest(): raise ValueError('review source events hash mismatch') events = json.loads(events_path.read_text(encoding='utf-8')) paths = sorted((directory/'wire').glob('*.request.json')) calls = {} for index, path in enumerate(paths, 1): captured = load_response(path.with_name(path.name.replace('.request.json', '.response.sse'))) assistant = captured['assistant'] if captured['protocol'] == 'chat' else response_assistant(captured['response']) for call in assistant.get('tool_calls') or []: if call['id'] in calls: raise ValueError('duplicate captured call ID') calls[call['id']] = index event_calls = [e['data']['callId'] for e in events if e['type'] == 'tool/call'] if len(event_calls) != len(set(event_calls)) or set(event_calls) != set(calls): raise ValueError('session events differ from captured calls') arm_review = validate_review(arm_review, calls, len(paths)) counts = {'decisions': 0, 'kept_decisions': 0, 'supervised_tokens': 0, 'max_sequence_tokens': 0, 'quarantined': arm_review['decision'] == 'quarantine'} for index, path in enumerate(paths, 1): view = request_view(json.loads(path.read_text(encoding='utf-8'))) captured = load_response(path.with_name(path.name.replace('.request.json', '.response.sse'))) assistant = captured['assistant'] if captured['protocol'] == 'chat' else response_assistant(captured['response']) status = captured['status'] if captured['protocol'] == 'chat' else captured['response']['status'] next_view = request_view(json.loads(paths[index].read_text(encoding='utf-8'))) if index < len(paths) else None sample = build_decision(renderer, view, assistant, following_results(assistant, next_view), sample_id=f'{root.name}:{arm}:{index}', response_status=status) reviewed = apply_review(sample, arm_review, index) if reviewed['ids'] != sample['ids'] or any(reviewed['mask'][:sample['prompt_tokens']]): raise ValueError('review changed context or supervised the prefix') if reviewed['keep_msgs']: validate_structure(reviewed, 65536) reviewed.update(arm=arm, request_index=index, request_sha256=hashlib.sha256(path.read_bytes()).hexdigest(), instance_id=selection['instance_id'], family=selection['family'], pool='A', generation_kind='acceptance_smoke', review_sha256=summary['review_sha256']) handle.write(json.dumps(reviewed, ensure_ascii=False) + '\n') counts['decisions'] += 1 counts['kept_decisions'] += bool(reviewed['keep_msgs']) counts['supervised_tokens'] += sum(reviewed['mask']) counts['max_sequence_tokens'] = max(counts['max_sequence_tokens'], len(reviewed['ids'])) summary['arms'][arm] = counts summary['payload_sha256'] = hashlib.sha256(output.read_bytes()).hexdigest() (root/'decision-export.json').write_text(json.dumps(summary, indent=2) + '\n', encoding='utf-8', newline='\n') return summary PROBE_IDENTITY_KEYS = ('protocol', 'freeze_sha256', 'script_sha256', 'case_list_sha256', 'case_count', 'max_feedback', 'source_policy_sha256', 'module_sha256') def check_probe_feedback(root: Path, protocol: dict, prediction: dict): """public-probe-feedback-v1 runs must carry the pre-call freeze and a passing terminal check.""" from native_release import read_json, file_hash frozen = protocol.get('public_probe') if frozen is None: if '-probe' in protocol.get('profile_name', ''): raise ValueError('Public probe protocol missing') return if (any(key not in frozen for key in PROBE_IDENTITY_KEYS) or frozen.get('protocol') != 'public-probe-feedback-v1' or frozen.get('max_feedback') != 1): raise ValueError('Public probe identity incomplete or downgraded') freeze = Path(frozen.get('freeze_path', '')) if (not freeze.is_file() or root.resolve() in freeze.resolve().parents or file_hash(freeze) != frozen.get('freeze_sha256')): raise ValueError('Public probe freeze missing, inside the run, or changed') receipt = read_json(root/'public-probe-feedback.json') runs = receipt.get('runs') issued = [run for run in runs or [] if run.get('feedback_issued')] terminal = receipt.get('terminal_run_index') from public_probe import sha_bytes patch_sha = sha_bytes(prediction.get('model_patch', '').encode()) requests = [request_view(json.loads(path.read_text(encoding='utf-8'))) for path in sorted((root/'wire').glob('*.request.json'))] def delivered(run): boundary = run.get('after_request_index') if type(boundary) is not int or boundary < 1: return False texts = [] for view in requests[boundary:]: for message in view['messages']: if message.get('role') != 'user': continue content = message.get('content') if isinstance(content, str): texts.append(content) elif isinstance(content, list): texts.extend(part['text'] for part in content if isinstance(part, dict) and isinstance(part.get('text'), str)) return any(text.startswith('A public behaviour check derived from the problem statement') and sha_bytes(text.encode()) == run.get('feedback_sha256') for text in texts) if (not all(delivered(run) for run in issued) or receipt.get('model_patch_sha256') != patch_sha or not runs or runs[-1].get('model_patch_sha256') != patch_sha): raise ValueError('Public probe wire delivery or submitted model_patch binding failed') if (any(receipt.get(key) != frozen.get(key) for key in PROBE_IDENTITY_KEYS) or not isinstance(runs, list) or not runs or receipt.get('feedback_count') != len(issued) or len(issued) > frozen.get('max_feedback', 1) or receipt.get('feedback_delivered') != [True]*len(issued) or type(terminal) is not int or terminal != runs[-1].get('index') or runs[-1].get('status') != 'passed' or runs[-1].get('workspace_sha256') != receipt.get('final_workspace_sha256') or receipt.get('final_status') != 'passed' or receipt.get('passed') is not True or prediction.get('public_probe_final', {}).get('passed') is not True): raise ValueError('Public probe feedback receipt does not prove one bounded feedback and a passing terminal check') def reviewed_student_policy(root, review): """Opt-in policy snapshot; absent metadata retains historical strict admission.""" from native_release import read_json, file_hash selected = review.get('student_admission_policy') if selected is None: return None canonical = ROOT/'storage/config/student-admission-v2.json' policy = read_json(canonical) expected = {'policy_id': policy['policy_id'], 'sha256': file_hash(canonical)} if selected != expected or file_hash(root/'student-admission-policy.json') != expected['sha256']: raise ValueError('Student admission policy snapshot differs from registered policy') return policy def check_reviewed_recovery(review, policy, affected, decoded, states): """Keep useful repairs while failed actions and redundant repetitions stay masked.""" from takeover import TEST_COMMAND note = review.get('student_recovery_review', {}) assessments = note.get('decisions', []) if (not isinstance(assessments, list) or len(assessments) != len(affected) or {x.get('request_index') for x in assessments} != set(affected) or len(affected) > policy['max_reviewed_symptom_decisions']): raise ValueError('Recovered student symptoms need bounded, complete review') masks = set(review['mask_requests']) kept_rechecks = 0 for assessment in assessments: index = assessment['request_index'] disposition = assessment.get('disposition') if not isinstance(assessment.get('reason'), str) or not assessment['reason'].strip(): raise ValueError('Recovered student action lacks a concrete review reason') triggers = affected[index] contract_error = any(t['kind'] == 'W' and t['evidence']['reason'] == 'tool_contract_error' for t in triggers) repeated = [t for t in triggers if t['kind'] == 'L'] if disposition == 'mask' and index in masks: continue if contract_error or index in masks: raise ValueError('Failed student tool actions must remain masked') if disposition == 'reverification' and repeated: calls = decoded[index-1][1].get('tool_calls', []) if any(t['evidence']['tool'] != 'bash' for t in repeated) or not any( TEST_COMMAND.search((json.loads(c['function']['arguments']).get('command', ''))) for c in calls if c['function']['name'] == 'bash'): raise ValueError('Useful reverification must reference an actual test command') kept_rechecks += 1 elif disposition != 'failed_test_observation' or repeated or not all( t['kind'] == 'W' and t['evidence']['reason'] == 'visible_test_failure_after_modification' for t in triggers): raise ValueError('Unsupported student recovery assessment') if kept_rechecks > policy['max_kept_reverification_decisions']: raise ValueError('Too many kept repeated verifications') core = note.get('retained_core_edit_requests', []) if (not isinstance(core, list) or not core or len(set(core)) != len(core) or any(type(i) is not int or i in masks or i not in review.get('core_edit_request_indices', []) or i not in states or i+1 not in states or states[i] == states[i+1] for i in core)): raise ValueError('Recovered student source needs a retained, observed core edit') finish = note.get('clean_finish_request') if (finish != len(decoded) or finish in masks or affected and finish <= max(affected)): raise ValueError('Recovered student source needs a clean supervised finish') def check_student_execution(root, protocol, decoded, handoff, review=None): """Recompute symptoms from captured history and the recorded workspace boundaries.""" from native_release import read_json, read_jsonl from takeover import decisions, detect kind = protocol['trajectory_kind'] if kind == 'P1': return [] policy = reviewed_student_policy(root, review or {}) if kind == 'P3' else None affected = {} boundary = handoff['boundary_request_index'] if handoff else len(decoded) path = root/'takeover-evaluations.jsonl' if not path.is_file(): raise ValueError('Missing student boundary evaluations') rows = read_jsonl(path) if [r.get('request_index') for r in rows] != list(range(1, boundary+1)): raise ValueError('Incomplete student boundary evaluations') states = {} for index, (row, (view, _, _, _)) in enumerate(zip(rows, decoded), 1): state = row.get('workspace_state', {}).get('sha256') history = decisions(view['messages']) if (not state or len(history) != index-1 or any(set(d['results']) != {c['id'] for c in d['calls']} for d in history)): raise ValueError('Student boundary lacks complete tool history or workspace state') states[index] = state triggers = detect(history, states, index) action = 'handoff' if handoff and index == boundary else 'student' if (row.get('triggers') != triggers or row.get('action') != action or row.get('forced_smoke') is not False or row.get('prefix_limit') != 15000 or type(row.get('prefix_tokens')) is not int or not 0 < row['prefix_tokens'] <= 15000): raise ValueError('Student boundary differs from raw history or frozen limits') if kind == 'P3' and triggers: if policy is None or any(t['kind'] not in policy['allowed_recovered_symptoms'] for t in triggers): raise ValueError('P3 student has an agreed L/S/W symptom') for trigger in triggers: evidence = trigger['evidence'] bad_index = evidence['second']['request_index'] if trigger['kind'] == 'L' else evidence['request_index'] affected.setdefault(bad_index, []).append(trigger) if kind == 'P2' and bool(triggers) != (index == boundary): raise ValueError('P2 trigger must be the first natural student symptom') if handoff and (handoff['trigger'] != rows[-1]['triggers'][0] or handoff['workspace_state'] != rows[-1]['workspace_state']): raise ValueError('P2 handoff trigger or workspace differs from student boundary') state = read_json(root/'takeover-state.json') terminal = read_json(root/'terminal.json') _, last, status, origin = decoded[-1] if (state.get('trajectory_kind') != kind or state.get('phase') != origin['phase'] or state.get('stop') is not None or state.get('terminal') is not None or state.get('forced_takeover_at') is not None or protocol.get('forced_takeover_at') is not None or state.get('handoff_request_index') != (boundary if handoff else None) or (terminal.get('terminal') or {}).get('reason', {}).get('kind') != 'completed' or not terminal.get('session_id') or status != 'completed' or origin.get('finish_reason') != 'stop' or last.get('tool_calls')): raise ValueError(kind + ' requires a natural stop, not a budget or infrastructure termination') if handoff and terminal['session_id'] != handoff['dsh_session_id']: raise ValueError('P2 handoff DSH session changed') if policy is not None: check_reviewed_recovery(review, policy, affected, decoded, states) if kind == 'P3' and protocol.get('public_probe') is not None: probe = read_json(root/'public-probe-feedback.json') if (probe.get('feedback_count') != 0 or probe.get('feedback_delivered') != [] or any(r.get('feedback_issued') for r in probe['runs']) or any(text_content(m.get('content')).startswith( 'A public behaviour check derived from the problem statement') for view, _, _, _ in decoded for m in view['messages'] if m.get('role') == 'user')): raise ValueError('P3 public probe feedback disqualifies unassisted student success') return rows def candidate_rows(root: Path, task_file=None, *, allow_synthetic=False) -> list[dict]: """Reconcile one v2 trajectory. No default review and no eligibility inference.""" from native_release import read_json, read_jsonl, file_hash, object_hash protocol = read_json(root/'protocol.json') if (protocol.get('schema_version') != 2 or protocol.get('generation_kind') != 'production_candidate' or protocol.get('trajectory_kind') not in {'P1', 'P2', 'P3'} or not protocol.get('profile_name') or protocol.get('student_prefix_limit') != 15000): raise ValueError('Native v2 production_candidate protocol required; v1/smoke is diagnostic only') route_audit = RouteAudit(root, protocol) synthetic = protocol.get('synthetic', False) if type(synthetic) is not bool or synthetic and not allow_synthetic: raise ValueError('Synthetic fixture is not a production source') task = load_public_task(task_file, protocol['instance_id']) if file_hash(task_file) != protocol.get('task_fixture_sha256'): raise ValueError('Task/protocol identity mismatch') prediction = read_json(root/'prediction.json') judge = read_json(root/'judge.json') review = read_json(root/'review.json') alignment = read_json(root/'token-alignment.json') environment = read_json(root/'environment.json') cleanup = read_json(root/'cleanup.json') read_json(root/'budget-final.json') if (prediction.get('instance_id') != task['instance_id'] or prediction.get('terminal_status') != 'completed' or not prediction.get('model_patch', '').strip()): raise ValueError('Incomplete task or empty patch') if (judge.get('schema_version') != 2 or judge.get('prediction_sha256') != file_hash(root/'prediction.json') or judge.get('instance_id') != task['instance_id'] or judge.get('result', {}).get('accepted') is not True or judge.get('independent') is not True or judge.get('fresh_before') is not True): raise ValueError('Missing independent fresh-before accepted judge receipt') before, judge_environment = judge.get('before'), judge.get('environment') if (protocol.get('base_commit') != task['base_commit'] or environment.get('base') != task['base_commit'] or not isinstance(before, dict) or before.get('base_commit') != task['base_commit']): raise ValueError('Judge, solver and task base identity mismatch') image_id = protocol.get('image_id') if (protocol.get('image') != task['image'] or not isinstance(image_id, str) or not image_id.strip() or environment.get('image_id') != image_id or not isinstance(judge_environment, dict) or judge_environment.get('image_id') != image_id): raise ValueError('Judge, solver and task image identity mismatch') if environment.get('passed') is not True or cleanup.get('verified') is not True: raise ValueError('Environment/cleanup has not passed') for receipt in (review, alignment, judge): if (receipt.get('protocol_sha256') != file_hash(root/'protocol.json') or receipt.get('provenance_sha256') != file_hash(root/'request-provenance.jsonl')): raise ValueError('Review/alignment/judge not bound to protocol and provenance') if any(r.get('synthetic', False) is not synthetic for r in (review, alignment, judge, environment, cleanup)): raise ValueError('Synthetic receipt cannot be promoted by changing protocol flags') if (review.get('source_events_sha256') != file_hash(root/'session-events.json') or review.get('prediction_sha256') != file_hash(root/'prediction.json') or review.get('public_regression_passed') is not True or review.get('boundary_passed') is not True): raise ValueError('Missing process, public regression or boundary review') probe_path = root/'public-probe.json' probe = read_json(probe_path) if (review.get('public_probe_sha256') != file_hash(probe_path) or probe.get('prediction_sha256') != file_hash(root/'prediction.json')): raise ValueError('Public probe not bound to review and prediction') if probe.get('synthetic', False) is not synthetic: raise ValueError('Synthetic public probe differs from protocol') cases = probe.get('cases') if (probe.get('passed') is not True or not isinstance(cases, list) or not cases or any(not isinstance(case, dict) or case.get('passed') is not True for case in cases)): raise ValueError('Public probe requires nonempty passing case results') check_probe_feedback(root, protocol, prediction) if alignment.get('passed') is not True: raise ValueError('Request/token alignment has not passed') provenance = read_jsonl(root/'request-provenance.jsonl') paths = sorted((root/'wire').glob('*.request.json')) if not paths or len(paths) != len(provenance) or prediction.get('request_count') != len(paths): raise ValueError('Request/provenance/prediction count mismatch') decoded, calls, response_ids = [], {}, set() for index, (path, origin) in enumerate(zip(paths, provenance), 1): response_path = path.with_name(path.name.replace('.request.json', '.response.sse')) if (path.name != f'{index:04d}.request.json' or type(origin.get('request_index')) is not int or origin['request_index'] != index or origin.get('request_sha256') != file_hash(path) or origin.get('response_sha256') != file_hash(response_path)): raise ValueError('Wire/provenance request index or hash mismatch') required = ('provider', 'requested_model', 'served_model', 'reasoning_effort', 'response_id') if (any(not isinstance(origin.get(k), str) or not origin[k] for k in required) or not isinstance(origin.get('attempts'), list) or not origin['attempts'] or origin.get('phase') not in {'teacher', 'student'}): raise ValueError('Incomplete actual per-request identity/attempts') request = read_json(path) if request.get('model') != origin['requested_model']: raise ValueError('Requested model differs from wire') if request_effort(request) != origin['reasoning_effort']: raise ValueError('Reasoning effort differs from wire; absent effort is not_applicable') allowed_models = route_audit.validate(index, origin, request) captured = load_response(response_path) assistant = captured['assistant'] if captured['protocol'] == 'chat' else response_assistant(captured['response']) status = captured['status'] if captured['protocol'] == 'chat' else captured['response']['status'] response_id, observed_model = response_identity(response_path, captured) if observed_model != origin['served_model'] or observed_model not in allowed_models: raise ValueError('Actual response model differs from provenance or frozen served aliases') if response_id != origin['response_id']: raise ValueError('Actual response ID differs from provenance') if response_id in response_ids: raise ValueError('Duplicate response ID') response_ids.add(response_id) for call in assistant.get('tool_calls') or []: if not call.get('id') or call['id'] in calls: raise ValueError('Missing/duplicate actual call ID') calls[call['id']] = index decoded.append((request_view(request), assistant, status, origin)) events = read_json(root/'session-events.json') event_calls = [e['data']['callId'] for e in events if e['type'] == 'tool/call'] if len(event_calls) != len(set(event_calls)) or set(event_calls) != set(calls): raise ValueError('Actual session calls differ from wire') review = validate_review(review, calls, len(paths)) if review['decision'] == 'quarantine': raise ValueError('Quarantined source cannot be published') kind = protocol['trajectory_kind'] phases = [r['phase'] for _, _, _, r in decoded] if kind == 'P1' and set(phases) != {'teacher'} or kind == 'P3' and set(phases) != {'student'}: raise ValueError('P1/P3 phase does not match protocol source') handoff = None if kind == 'P2': handoff = read_json(root/'handoff.json') boundary = handoff.get('boundary_request_index') trigger = handoff.get('trigger') container = handoff.get('container') if (handoff.get('schema_version') != 1 or handoff.get('trajectory_kind') != 'P2' or type(boundary) is not int or not 1 < boundary <= len(paths) or phases != ['student']*(boundary-1) + ['teacher']*(len(paths)-boundary+1) or not isinstance(trigger, dict) or trigger.get('kind') not in {'L','S','W'} or trigger.get('natural') is not True or handoff.get('student_decisions') != boundary-1 or type(handoff.get('prefix_tokens')) is not int or not 0 < handoff['prefix_tokens'] <= 15000 or not all(handoff.get(k) for k in ('prefix_ids_sha256','prefix_view_sha256','workspace_state_sha256','dsh_session_id')) or not isinstance(container, dict) or not container.get('id') or not container.get('name') or handoff.get('workspace_state', {}).get('sha256') != handoff['workspace_state_sha256']): raise ValueError('P2 requires one genuine bounded student-to-teacher handoff') if container['id'] != environment.get('container_id'): raise ValueError('P2 handoff environment changed') student_boundaries = check_student_execution(root, protocol, decoded, handoff, review) alignment_rows = alignment.get('decisions', []) if [r.get('request_index') for r in alignment_rows] != list(range(1, len(paths)+1)): raise ValueError('Missing/duplicate alignment decisions') renderer = Renderer() source_id = object_hash({'protocol': file_hash(root/'protocol.json'), 'events': file_hash(root/'session-events.json'), 'provenance': file_hash(root/'request-provenance.jsonl')}) result = [] for index, (view, assistant, status, origin) in enumerate(decoded, 1): next_view = decoded[index][0] if index < len(decoded) else None sample = build_decision(renderer, view, assistant, following_results(assistant, next_view), sample_id=f'{source_id}:{index}', response_status=status) aligned = alignment_rows[index-1] prefix_hash = object_hash(sample['ids'][:sample['prompt_tokens']]) if index <= len(student_boundaries) and student_boundaries[index-1]['prefix_tokens'] != sample['prompt_tokens']: raise ValueError('Student boundary token count differs from reconstructed full prefix') if (aligned.get('passed') is not True or aligned.get('prefix_ids_sha256') != prefix_hash or aligned.get('request_sha256') != origin['request_sha256']): raise ValueError(f'Reconstructed prefix IDs differ from independent alignment at request {index}') if handoff and index == handoff['boundary_request_index']: if (handoff['prefix_ids_sha256'] != prefix_hash or handoff['prefix_tokens'] != sample['prompt_tokens'] or handoff['prefix_view_sha256'] != object_hash(view)): raise ValueError('P2 student prefix differs from handoff') reviewed = apply_review(sample, review, index) if kind == 'P2' and origin['phase'] == 'student': reviewed.update(mask=[0]*len(sample['ids']), keep_msgs=[], spans={}, masked_response_reason='STUDENT_PREFIX') if any(reviewed['mask']): validate_structure(reviewed, 65536) reviewed.update({k: origin[k] for k in ('request_index', 'request_sha256', 'response_sha256', 'phase', 'provider', 'requested_model', 'served_model', 'reasoning_effort')}) reviewed.update(source_id=source_id, instance_id=task['instance_id'], family=task['repo'].strip().casefold(), component_id=review.get('component_id'), pool='B' if kind == 'P2' else 'A', trajectory_kind=kind, generation_kind='production_candidate', synthetic=synthetic, quarantined=False, task_sha256=task_sha256(task), protocol_sha256=file_hash(root/'protocol.json'), provenance_sha256=file_hash(root/'request-provenance.jsonl'), judge_sha256=file_hash(root/'judge.json'), review_sha256=file_hash(root/'review.json'), mask_sha256=object_hash(reviewed['mask']), prefix_ids_sha256=prefix_hash) if review.get('student_admission_policy') is not None: reviewed['student_admission_policy_id'] = review['student_admission_policy']['policy_id'] reviewed['student_admission_policy_sha256'] = review['student_admission_policy']['sha256'] result.append(reviewed) return result def export_candidate(root: Path, output: Path, task_file: Path, *, synthetic=False): from native_release import object_hash, write_json rows = candidate_rows(root, task_file, allow_synthetic=synthetic) evidence = candidate_evidence(root, task_file) value = {'schema_version': 2, 'run_root': str(root.resolve()), 'rows_sha256': object_hash(rows), 'evidence': evidence} write_json(output, value) return output def candidate_evidence(root, task_file): from native_release import read_json, reference, verify_reference names = ('protocol.json', 'prediction.json', 'request-provenance.jsonl', 'review.json', 'judge.json', 'token-alignment.json', 'session-events.json', 'environment.json', 'budget-final.json', 'cleanup.json', 'public-probe.json') evidence = {name: reference(root/name) for name in names} evidence['task'] = reference(task_file) protocol = read_json(root/'protocol.json') if protocol['trajectory_kind'] == 'P2': evidence['handoff'] = reference(root/'handoff.json') if protocol['trajectory_kind'] in {'P2', 'P3'}: for name in ('terminal.json', 'takeover-state.json', 'takeover-evaluations.jsonl'): evidence[name] = reference(root/name) review = read_json(root/'review.json') if review.get('path_style_policy') is not None: evidence['path_style_policy'] = reference(verify_reference(review['path_style_policy'])) if review.get('student_admission_policy') is not None: evidence['student-admission-policy.json'] = reference(root/'student-admission-policy.json') evidence['review_rubric'] = reference(verify_reference(review['rubric'])) if (root/'source-lineage.json').exists(): evidence['source-lineage.json'] = reference(root/'source-lineage.json') if (root/'original-review.json').exists(): evidence['original-review.json'] = reference(root/'original-review.json') if protocol.get('public_probe') is not None: evidence['public-probe-feedback.json'] = reference(root/'public-probe-feedback.json') evidence['public-probe-freeze'] = reference(Path(protocol['public_probe']['freeze_path'])) for directory in ('wire', 'upstream-attempts'): for path in sorted((root/directory).glob('*')): if path.is_file(): evidence[directory+'/'+path.name] = reference(path) return evidence if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('run_root', type=Path) args = parser.parse_args() print(json.dumps(export(args.run_root.resolve())))