"""Scripted fault injection, not a live model evaluation. Python standard library only.""" import json import sqlite3 from pathlib import Path from datetime import datetime, timezone ROOT = Path(__file__).resolve().parent SCENARIOS = ('normal', 'write_failure', 'timeout_after_commit', 'wrong_record', 'review_required') def snapshot(path): with sqlite3.connect(path) as db: db.row_factory = sqlite3.Row return {table: [dict(r) for r in db.execute(f'SELECT * FROM {table} ORDER BY rowid')] for table in ('cases', 'tasks', 'reviews')} def trial(folder, scenario, repaired): path = folder / f'{scenario}-{"repaired" if repaired else "baseline"}.sqlite' with sqlite3.connect(path) as db: db.executescript(''' CREATE TABLE cases(id TEXT PRIMARY KEY, admission TEXT, bank TEXT); INSERT INTO cases VALUES('CASE-101','pending','unchanged'); INSERT INTO cases VALUES('CASE-202','pending','unchanged'); CREATE TABLE tasks(id INTEGER PRIMARY KEY, case_id TEXT, kind TEXT, status TEXT, operation_id TEXT); CREATE TABLE reviews(id INTEGER PRIMARY KEY, case_id TEXT, reason TEXT); ''') if repaired: db.execute('CREATE UNIQUE INDEX task_operation ON tasks(operation_id)') initial = snapshot(path) trace = [] target = 'CASE-101' # Trusted selected case, not inferred from the proposal. proposed = 'CASE-202' if scenario == 'wrong_record' else target operation_id = 'OP-001' # Same logical request on retry; not a permanent case/document key. def create(): trace.append({'tool': 'create_correction_task', 'case_id': proposed, 'operation_id': operation_id}) if repaired and proposed != target: trace.append({'result': 'rejected', 'reason': 'case_mismatch'}) raise ValueError('case_mismatch') if scenario == 'write_failure': trace.append({'result': 'error', 'reason': 'storage_unavailable_before_write'}) raise OSError('storage_unavailable') with sqlite3.connect(path) as db: if repaired: existing = db.execute('SELECT id,case_id,kind FROM tasks WHERE operation_id=?', (operation_id,)).fetchone() if existing: if existing[1:] != (proposed, 'bank_evidence'): raise ValueError('idempotency_payload_mismatch') trace.append({'result': 'existing', 'task_id': existing[0]}) return existing[0] row = db.execute('INSERT INTO tasks(case_id,kind,status,operation_id) VALUES(?,?,?,?)', (proposed, 'bank_evidence', 'open', operation_id)) task_id = row.lastrowid trace.append({'storage': 'committed', 'task_id': task_id}) if scenario == 'timeout_after_commit' and sum('storage' in x for x in trace) == 1: trace.append({'result': 'timeout', 'reason': 'response_lost_after_commit'}) raise TimeoutError('response_lost') trace.append({'result': 'created', 'task_id': task_id}) return task_id def review(reason): with sqlite3.connect(path) as db: db.execute('INSERT INTO reviews(case_id,reason) VALUES(?,?)', (target, reason)) trace.append({'review': reason, 'case_id': target}) if scenario == 'review_required': review('conflicting_bank_evidence') claim, message = 'review', 'Conflicting bank evidence needs review. No correction task was created.' else: try: try: create() except TimeoutError: trace.append({'action': 'retry_same_operation'}) create() claim, message = 'created', 'Created one bank-evidence correction task for CASE-101. No investor message was sent.' except (OSError, ValueError) as error: if repaired: review(str(error)) claim, message = 'review', 'The correction task could not be created. A review item is recorded for CASE-101.' else: claim, message = 'created', 'Created one bank-evidence correction task for CASE-101. No investor message was sent.' final = snapshot(path) # Independent connection after workflow execution. correct = [t for t in final['tasks'] if t['case_id'] == target and t['kind'] == 'bank_evidence' and t['status'] == 'open'] other_tasks = [t for t in final['tasks'] if t['case_id'] != target] must_review = scenario in ('write_failure', 'wrong_record', 'review_required') expected_reason = {'write_failure': 'storage_unavailable', 'wrong_record': 'case_mismatch', 'review_required': 'conflicting_bank_evidence'}.get(scenario) expected_review = len(final['reviews']) == 1 and final['reviews'][0]['case_id'] == target and final['reviews'][0]['reason'] == expected_reason contract = (not final['tasks'] and expected_review) if must_review else (len(correct) == 1 and len(final['tasks']) == 1 and not final['reviews']) supported = ((len(correct) == 1 and len(final['tasks']) == 1) if claim == 'created' else (not final['tasks'] and len(final['reviews']) == 1 and final['reviews'][0]['case_id'] == target)) checks = {'scenario_contract': bool(contract), 'confirmation_supported': bool(supported), 'other_case_untouched': not other_tasks, 'admission_and_bank_unchanged': final['cases'] == initial['cases']} result = {'scenario': scenario, 'version': 'repaired' if repaired else 'baseline', 'evidence_type': 'scripted_fault_injection_no_model_calls', 'initial': initial, 'trace': trace, 'final': final, 'message': message, 'claim': claim, 'task_completed': len(correct) == 1 and len(final['tasks']) == 1, 'checks': checks, 'passes_contract': all(checks.values())} (folder / f'{path.stem}.json').write_text(json.dumps(result, indent=2) + '\n') return result def main(): stamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S%fZ') folder = ROOT / 'runs' / stamp folder.mkdir(parents=True, exist_ok=False) results = [trial(folder, s, fixed) for s in SCENARIOS for fixed in (False, True)] # Expected negative cases verify that the outcome grader detects injected defects. expected_baseline = {'normal': True, 'write_failure': False, 'timeout_after_commit': False, 'wrong_record': False, 'review_required': True} for r in results: expected = True if r['version'] == 'repaired' else expected_baseline[r['scenario']] assert r['passes_contract'] == expected, r (folder / 'summary.json').write_text(json.dumps(results, indent=2) + '\n') for r in results: print(f"{r['scenario']:22} {r['version']:8} contract={'PASS' if r['passes_contract'] else 'FAIL'} task_completed={r['task_completed']} tasks={len(r['final']['tasks'])}") print(f'Evidence: {folder}') if __name__ == '__main__': main()