rea/experiments/analysis-efficiency/review-model.py

"""Check selected model facts against public producer observations, not exit status."""

import argparse
import hashlib
import json
from pathlib import Path
import re


def read(path):
    return json.loads(path.read_text())


def strings(value):
    if isinstance(value, str):
        yield value
    elif isinstance(value, list):
        for entry in value:
            yield from strings(entry)
    elif isinstance(value, dict):
        for entry in value.values():
            yield from strings(entry)


def evidence_ids(value):
    return {s for s in strings(value) if re.fullmatch(r'ev_[a-f0-9]{64}', s)}


def decode(result):
    value = result.get('structured_content', result.get('structuredContent'))
    if value is not None:
        return value
    for entry in result.get('content', []):
        if entry.get('type') == 'text':
            try:
                return json.loads(entry['text'])
            except json.JSONDecodeError:
                pass
    return None


def named_scope(value, names):
    """Accept explicit equivalent scope labels; contradictory duplicates stay a failure."""
    if not isinstance(value, dict):
        return None
    selected = {name: value[name] for name in names if name in value}
    values = list(selected.values())
    if not values:
        return None
    if all(entry == values[0] for entry in values):
        return values[0]
    return {'conflicting_reported_scopes': selected}


def scope_with_verified_counts(value, expected_scope, returned_counts):
    """Validate explicit extra row counts; never erase unknown or conflicting facts."""
    if not isinstance(value, dict):
        return value, {}
    scope = dict(value)
    checks = {}
    for name, expected in returned_counts.items():
        if name in scope:
            reported = scope[name]
            checks[name] = type(reported) is int and reported == expected
            if name not in expected_scope:
                del scope[name]
    return scope, checks


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--domain', choices=['wasm', 'evm', 'android', 'managed', 'managed-nonempty', 'firmware', 'inventory', 'javascript', 'native', 'browser'], required=True)
    parser.add_argument('--trial', type=Path, required=True)
    parser.add_argument('--reference', type=Path)
    parser.add_argument('--resource-log', type=Path, required=True)
    parser.add_argument('--output', type=Path, required=True)
    args = parser.parse_args()
    summary = read(args.trial / 'summary.json')
    settings = read(args.trial / 'trial-settings.json')
    answer_text = (args.trial / 'answer.txt').read_text().strip()
    fenced = answer_text.startswith('```json\n') and answer_text.endswith('\n```')
    answer = json.loads(answer_text[8:-4] if fenced else answer_text)
    reference = read(args.reference / 'answer.json') if args.reference else None
    observed_ids = set()
    producer_results = {}
    tool_errors = []
    for line in (args.trial / 'events.jsonl').open():
        event = json.loads(line)
        item = event.get('item', {})
        if event.get('type') != 'item.completed' or item.get('type') != 'mcp_tool_call':
            continue
        result = item.get('result') or {}
        parsed = decode(result)
        if item['status'] == 'completed':
            # Only response text contributes observed IDs, never model arguments.
            for text in strings(result):
                observed_ids.update(re.findall(r'ev_[a-f0-9]{64}', text))
            if parsed is not None:
                producer_results.setdefault(item['tool'], []).append(parsed)
        else:
            tool_errors.append({'tool': item['tool'], 'arguments': item['arguments'], 'response': parsed if parsed is not None else result})
            if parsed is not None and parsed.get('error', {}).get('code') == 'resource_constraint':
                reference_value = parsed['error'].get('details', {}).get('reported_limits', {}).get('evidence_reference')
                observed_ids.update(evidence_ids(reference_value))
    reported_ids = evidence_ids(answer.get('evidence_ids'))
    checks = {
        'reported_evidence_ids_observed_in_responses': bool(reported_ids) and reported_ids <= observed_ids,
        'temporary_profile_removed': read(args.trial / 'cleanup.json')['profile_removed'],
    }
    if summary.get('selected_artifact_sha256_before') is not None:
        checks['selected_artifact_bytes_unchanged'] = summary['selected_artifact_bytes_unchanged']
    if reference is not None:
        checks['target_digest'] = answer.get('artifact_sha256') == reference['artifact_sha256']
    differences = {}

    def check(name, actual, expected):
        checks[name] = actual == expected
        if not checks[name]:
            differences[name] = {'answer': actual, 'producer': expected}

    domain = args.domain
    if domain == 'wasm':
        for key in ['validation', 'section_count', 'import_count', 'export_count', 'exports_first4', 'wat_sha256', 'wat_code_units', 'runtime_execution', 'network_fetch']:
            check(key, answer.get(key), reference[key])
        for key in ['release', 'source_commit']:
            check('wabt_' + key, answer.get('wabt_' + key), reference['tool_profile'][key])
    elif domain == 'evm':
        n = producer_results['inspect_evm_interface'][-1]['normalized_result']
        if n.get('kind') == 'summary':
            check('evm_summary.parent_operation', n.get('parent_operation'), 'inspect_evm_interface')
            checks['evm_summary.parent_reference_observed'] = n.get('parent_evidence_id') in observed_ids
            n = n['summary']
        for key, expected in {'artifact_bytes': n['artifact']['bytes'], 'bytecode_sha256': n['bytecode']['sha256'], 'bytecode_bytes': n['bytecode']['bytes'], 'functions': n['functions'], 'evidence_kind': n['evidence_kind'], 'discovery_completeness': n['discovery_completeness'], 'runtime_execution': n['runtime_execution'], 'bytecode_kind': n['bytecode']['kind'], 'hardfork': n['bytecode']['hardfork'], 'deployment_authenticity': n['bytecode']['deployment_authenticity']}.items():
            check(key, answer.get(key), expected)
        check('evm_original_limitations', answer.get('limitations'), n['limitations'])
    elif domain == 'android':
        for key in ['package_name', 'version_name', 'version_code', 'matches']:
            check(key, answer.get(key), reference[key])
        fields = ['reported_signature', 'overload_index', 'dex_descriptor']
        checks['onCreate_fields_present_including_explicit_null'] = all(all(k in row for k in fields) for row in answer.get('onCreate', []))
        check('all_onCreate_descriptors', [{k: row.get(k) for k in fields} for row in answer.get('onCreate', [])], [{k: row[k] for k in fields} for row in reference['onCreate']])
    elif domain == 'managed':
        n = reference['artifact']
        for group, fields in [('assembly', ['name', 'version']), ('module', ['name', 'mvid'])]:
            for field in fields:
                check(group + '.' + field, answer.get(group, {}).get(field), n[group][field])
        for key in ['classification', 'target_frameworks']:
            check(key, answer.get(key), n[key])
        for key in ['module_refs', 'pinvoke_imports']:
            check(key, answer.get('native_boundaries', {}).get(key), reference['boundaries'][key])
        check('artifact_coverage', named_scope(answer.get('coverage'), ['artifact', 'managed_artifact', 'managed_metadata']), n['coverage'])
        boundary_scope, counts = scope_with_verified_counts(
            named_scope(answer.get('coverage'), ['native_boundaries', 'native_boundary_declarations']),
            reference['boundaries']['coverage'],
            {'module_ref_count': len(reference['boundaries']['module_refs']),
             'pinvoke_import_count': len(reference['boundaries']['pinvoke_imports'])})
        check('native_boundaries_coverage', boundary_scope, reference['boundaries']['coverage'])
        checks.update({'native_coverage_' + name: valid for name, valid in counts.items()})
    elif domain == 'managed-nonempty':
        n = reference['boundaries']
        check('module_mvid', answer.get('module_mvid'), n['module']['mvid'])
        check('summary', {key: answer.get('summary', {}).get(key) for key in n['summary']}, n['summary'])
        check('all_original_module_refs', answer.get('module_refs'), n['module_refs'])
        check('first3_complete_pinvoke_records', answer.get('pinvoke_first3'), n['pinvoke_imports'][:3])
        check('coverage', answer.get('coverage'), n['coverage'])
    elif domain == 'firmware':
        source = producer_results['inspect_firmware_regions'][-1]
        n = source['normalized_result']
        result = answer['result']
        check('input_size', result.get('input_size'), n['input_size'])
        check('examined_coverage', result.get('coverage', result.get('examined_coverage')), n['coverage'])
        fields = ['provider_id', 'offset', 'reported_size', 'signature', 'description', 'reported_confidence', 'size_basis']
        check('all_original_region_records', [{k: row.get(k) for k in fields} for row in result.get('regions', [])], [{k: row[k] for k in fields} for row in n['regions']])
        engine = result.get('engine', result.get('provider_provenance', {}).get('engine', {}))
        for key in ['name', 'version']:
            check('engine.' + key, engine.get(key), n['engine'][key])
        check('engine_executable_digest', engine.get('executable_sha256', engine.get('artifact_sha256')), n['engine']['executable_sha256'])
    elif domain == 'inventory':
        for key in ['manifest', 'entry_kind_counts', 'integrity_contradictions']:
            check(key, answer.get(key), reference[key])
        original = producer_results['inspect_artifact'][-1]['normalized_result']
        if original.get('kind') == 'summary':
            branches = original['summary']['inspection']['unexplored_branches']
        else:
            branches = original['unexplored_branches']
        # Caller-selected integrity policy participates in atomic Evidence and
        # branch identity. Match those IDs to this original response, not a
        # reference invocation that may use a different policy.
        check('unexplored_branches_with_original_ids', answer.get('unexplored_branches'), branches)
        coverage = answer.get('coverage', {})
        check('inspection_coverage', coverage.get('inspection', coverage), reference['coverage'])
        paths = answer.get('requested_paths')
        if isinstance(paths, dict):
            paths = [row for value in paths.values() for row in (value if isinstance(value, list) else [value]) if row is not None]
        canonical = lambda row: json.dumps(row, sort_keys=True, separators=(',', ':'))
        check('original_selected_path_records', sorted(map(canonical, paths or [])), sorted(map(canonical, reference['requested_paths'])))
    elif domain == 'javascript':
        for key in ['input_path', 'source_selection', 'statistics', 'electron']:
            check(key, answer.get(key), reference[key])
        coverage = answer['coverage']
        expected = reference['coverage']
        for key in ['application_unknowns', 'semantic_unknowns']:
            check('coverage.' + key, coverage.get(key), expected[key])
        for layer in ['application', 'semantic']:
            for key, value in expected[layer].items():
                if key in ['limits', 'families']: continue
                check('coverage.' + layer + '.' + key, coverage[layer].get(key), value)
        # Budgets vary with parent heap and belong to this actual invocation.
        # Exact optional provenance remains a separate strict check; omissions
        # must not be hidden by accepting the coarse coverage summary.
        original = producer_results['analyze_javascript_application'][-1]['normalized_result']['summary']['coverage']
        check('full_coverage_provenance', coverage, original)
        # The question names types/digests, not literal answer property names.
        # Exact node IDs select the producer's main.js records. No source or
        # Evidence values are rewritten to accept these answer labels.
        select = lambda rows: sorted([{
            'node_id': row.get('node_id'),
            'kind': row.get('kind', row.get('type')),
            'identity': row.get('identity'),
            'hashes': row.get('hashes', row.get('digests')),
        } for row in rows], key=lambda row: row['node_id'])
        check('all_selected_module_identities', select(answer.get('modules', [])), select(reference['modules']))
    elif domain == 'native':
        for key in ['entry_address', 'caller_count', 'callee_count']:
            check(key, answer.get(key), reference[key])
        prefix = answer.get('pseudocode_prefix')
        if isinstance(prefix, dict) and prefix.get('unit') == 'utf16-code-units':
            text = prefix.get('text')
            if 'requested' in prefix:
                check('pseudocode_prefix.requested', prefix['requested'], 512)
            if 'returned' in prefix:
                check('pseudocode_prefix.returned', prefix['returned'], len(reference['pseudocode_prefix'].encode('utf-16-le')) // 2)
            prefix = text
        check('pseudocode_prefix', prefix, reference['pseudocode_prefix'])
        containing = answer.get('containing_procedure')
        expected_containing = reference['containing_procedure']['procedure']
        if isinstance(containing, str):
            # Compare the explicitly reported display to the known source;
            # never parse a display name into a canonical lookup identity.
            check('containing_procedure.display', containing,
                  expected_containing['name'] + ' @ ' + expected_containing['address'])
        else:
            containing = containing if isinstance(containing, dict) else {}
            nested = containing.get('procedure')
            nested = nested if isinstance(nested, dict) else {}
            for key in ['address', 'name']:
                check('containing_procedure.' + key, containing.get(key, nested.get(key)), expected_containing[key])
        procedure = answer.get('procedure')
        procedure = procedure if isinstance(procedure, dict) else {}
        expected = reference['procedure']
        for key in ['address', 'name', 'classification', 'signature', 'locals']:
            check('procedure.' + key, procedure.get(key), expected[key])
        body = procedure.get('body')
        body = body if isinstance(body, dict) else {}
        check('procedure.body_ranges', procedure.get('body_ranges', body.get('ranges', body.get('ranges_inclusive'))), expected['body']['ranges'])
        check('procedure.total_bytes', procedure.get('total_bytes', body.get('total_bytes')), expected['body']['total_bytes'])
        if 'load_image' in reference:
            sources = [source for source in producer_results.get('inspect_native_load_image', [])
                       if isinstance(source, dict)
                       and source.get('operation') == 'inspect_native_load_image'
                       and isinstance(source.get('evidence_id'), str)
                       and source.get('evidence_id') in reported_ids
                       and isinstance(source.get('normalized_result'), dict)]
            checks['load_image_source_observed_and_referenced'] = bool(sources)
            load = sources[-1]['normalized_result'] if sources else {}
            entry = load.get('entry')
            entry = entry if isinstance(entry, dict) else {}
            check('load_image.target_sha256', load.get('target_sha256'), reference['artifact_sha256'])
            check('load_image.source_status', load.get('status'), reference['load_image']['status'])
            check('load_image.source_entry', entry.get('linear_address'), reference['entry_address'])
            status = answer.get('load_image_status')
            check('load_image_status', status.get('status') if isinstance(status, dict) else status, load.get('status'))
            if isinstance(status, dict):
                for key in ['format', 'load_segment', 'header_bytes', 'module_bytes', 'overlay_bytes']:
                    if key in status:
                        check('load_image.' + key, status[key], load.get(key))
                if 'entry_linear_address' in status:
                    check('load_image.entry_linear_address', status['entry_linear_address'], entry.get('linear_address'))
        calls = summary['completed_mcp_calls']
        checks['default_analysis_not_disabled'] = all(not call['arguments'].get('skip_auto_analysis', False) for call in calls if call['tool'] == 'open_binary')
        checks['session_close_completed'] = any(call['tool'] == 'close_binary' and call['status'] == 'completed' for call in calls)
    elif domain == 'browser':
        source = producer_results['capture_browser_scenario'][-1]
        n = source['normalized_result']
        for key in ['browser', 'completeness']:
            check(key, answer.get(key), n[key])
        check('start_origin', answer.get('start_origin'), n['scenario']['start_origin'])
        step_count = answer.get('step_count')
        check('step_count', step_count.get('captured_steps') if isinstance(step_count, dict) else step_count, len(n['steps']))
        final = n['steps'][-1]
        expected_url = final['after_url'] if isinstance(answer.get('final_url'), dict) else final['after_url']['url']
        check('final_url', answer.get('final_url'), expected_url)
        for key in ['dom', 'accessibility']:
            expected = final['artifacts'][key]
            actual = answer.get(key, {})
            if expected['state'] == 'captured':
                actual = actual.get('value', actual)
                for field in ['sha256', 'bytes']:
                    check(key + '.' + field, actual.get(field), expected['value'][field])
            else:
                check(key + '.state', actual.get('state'), expected['state'])
        checks['owned_browser_terminated'] = n['browser']['cleanup'] == 'terminated-owned-process'
        events = n['events']['items']
        page_errors = [event for event in events if event['kind'] == 'page-error']
        if isinstance(answer.get('page_errors'), dict):
            check('page_error_count', answer['page_errors'].get('count'), len(page_errors))
        network = answer.get('network', {})
        if 'observed_transactions' in network:
            check('network_transaction_count', network['observed_transactions'], len({event['transaction_id'] for event in events if event.get('transaction_id') is not None}))
        if 'observed_response_statuses' in network:
            statuses = {}
            for event in events:
                if event['kind'] == 'response':
                    status = str(event['status'])
                    statuses[status] = statuses.get(status, 0) + 1
            check('network_response_statuses', network['observed_response_statuses'], statuses)
    factual_checks = dict(checks)
    scopes = [json.loads(line) for line in args.resource_log.read_text().splitlines() if line.startswith('{')]
    scope = next(value for value in reversed(scopes) if 'resource_scope' in value)
    checks['no_resource_abort'] = scope['abort'] is None
    events = scope.get('resource_events_final')
    checks['resource_counters_verified'] = scope.get('resource_counters_verified') is True
    checks['no_observed_oom_or_task_limit'] = (
        isinstance(events, dict)
        and all(events.get('memory.events', {}).get(k) == 0 for k in ['max', 'oom', 'oom_kill'])
        and events.get('pids.events', {}).get('max') == 0
    )
    checks['no_resource_limit_event'] = (
        checks['resource_counters_verified']
        and checks['no_observed_oom_or_task_limit']
        and events['memory.events'].get('high') == 0
    )
    report = {
        'review_source_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        'domain': domain, 'checks': checks, 'all_selected_checks_pass': all(checks.values()),
        'factual_checks': factual_checks, 'all_factual_checks_pass': all(factual_checks.values()),
        'differences': differences, 'tool_errors': tool_errors, 'summary': summary,
        'fingerprints': {key: value for key, value in settings.items() if 'sha256' in key or key in ['runtime_revision', 'enabled_tools', 'authority']},
        'answer_file_sha256': hashlib.sha256((args.trial / 'answer.txt').read_bytes()).hexdigest(),
        'answer_had_json_fence': fenced, 'reported_evidence_ids': sorted(reported_ids), 'resource': scope,
        'scope': 'Fixed selected-fact review against actual public producer observations. Equivalent explicitly named answer representations are accepted; unknowns/nulls remain checked. Limitations, unsupported claims and broader behavior still require transcript review. Token usage and failed calls are retained; zero exit status is not a factual grade.',
    }
    with args.output.open('x') as output:
        output.write(json.dumps(report, indent=2, ensure_ascii=False) + '\n')
    print(json.dumps({'domain': domain, 'passed': report['all_selected_checks_pass'], 'failed_checks': [key for key, value in checks.items() if not value]}))
    raise SystemExit(0 if report['all_selected_checks_pass'] else 1)


if __name__ == '__main__':
    main()