c7105d9723
Preserve reproducible complete versus summary/page PCAP measurements on two official HTTP recordings and a fixed metadata/URI question. Summary/page uses more than twice the SDK JSON and about 51% more total model input in both real-model pairs, so existing complete defaults and production APIs/cache remain unchanged. These results apply to these captures and question; no general speedup is claimed. Refs #2016.
Authenticate experiment SDK envelopes with the pinned runtime's existing Evidence parser and check coverage, target/view/operation ownership and retained-parent identity. Separate factual and exact-workflow grading, retaining diagnostics and partial facts for malformed replies. Register six comparator CLI scenarios in the discovered Vitest evaluation project and remove the manual suite. The final change is limited to nine experiment files and one discovered evaluation test; production behavior is unchanged.
Validation includes eight actual SDK workflows and four actual Codex CLI tasks, independent raw-frame and CLI/MCP checks, authentication of all 18 original receipts, 34 context/grading regressions and six comparator scenarios. All four stored model answers pass factual and workflow regrading without model reruns. Original measured runtime remains 3ae826ceba6f25b08bce50bcb51001209bc16b57; original measurements and private raw artifacts are unchanged. Serial local checks use one CPU with 3-GiB early-stop/4-GiB hard/no-swap limits. Unknown terminal counters remain explicitly unknown.
Integrate upstream shared fixture owners and Windows verification, removing both superseded local differences. All final experiment/comparator files remain byte-identical to the reviewed sources and the final PR scope is 10 files. Full bounded preflight passes in 73.953 seconds with ~2.23 GiB peak; payload cleanup is verified, zero phase OOM/limit events are observed, and terminal counters remain unknown. Current-head automatic CI 38492231165 passes all 35 selected jobs, including every compiled shard, source/static/build/docs, actual Windows/native/package and real application lanes, Portable reports/inventory, reverse-tools coverage/mutation and CI required. Greptile passes and all review threads are resolved. Actual CI checkout is 4eb2262359cb01f2fb1314db0bc18d3dc9092792, integrated with main 41a2e84647. Passing retries do not establish the root cause or a fix of the earlier main worker cleanup failure; its original evidence is preserved. No release.
Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
Co-authored-by: Kaoru0822 <208281322+kaoru0822-kitauji@users.noreply.github.com>
Co-authored-by: morluto <76467478+morluto@users.noreply.github.com>
599 lines
38 KiB
Python
599 lines
38 KiB
Python
"""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 pcap_observation_fields(value):
|
|
"""Equivalent explicit labels are accepted; conflicting duplicates remain failures."""
|
|
if not isinstance(value, dict):
|
|
return value
|
|
result = dict(value)
|
|
for label, canonical in {
|
|
'outer_limitations': 'limitations', 'normalized_kind': 'kind',
|
|
'normalized_view': 'view', 'normalized_parent_evidence_id': 'parent_evidence_id',
|
|
'normalized_coverage': 'coverage', 'normalized_unknowns': 'unknowns',
|
|
'normalized_source_coverage': 'source_coverage',
|
|
}.items():
|
|
if label not in result:
|
|
continue
|
|
if canonical in result and json.dumps(result[label], sort_keys=True) != json.dumps(result[canonical], sort_keys=True):
|
|
return {**result, 'conflicting_fields': [label, canonical]}
|
|
result[canonical] = result.pop(label)
|
|
return result
|
|
|
|
|
|
def pcap_complete_capture(value):
|
|
"""Project the selected metadata/counts from the cited complete receipt."""
|
|
inventories = ['endpoints', 'dns_queries', 'tls_sni_values', 'protocol_distribution',
|
|
'http_request_uris', 'unattributed_dns_addresses']
|
|
if not isinstance(value, dict) or not all(isinstance(value.get(key), list) for key in inventories):
|
|
return None
|
|
capture = {key: item for key, item in value.items() if key not in inventories + ['diagnostics']}
|
|
capture['retained_counts'] = {key: len(value[key]) for key in inventories}
|
|
diagnostics = value.get('diagnostics')
|
|
if 'diagnostics' not in value:
|
|
capture['diagnostics_code_units'] = None
|
|
elif isinstance(diagnostics, dict) and all(isinstance(diagnostics.get(key + '_stderr'), str) for key in ['tshark', 'capinfos']):
|
|
capture['diagnostics_code_units'] = {key: len(diagnostics[key + '_stderr'].encode('utf-16-le', errors='surrogatepass')) // 2
|
|
for key in ['tshark', 'capinfos']}
|
|
else:
|
|
return None
|
|
return capture
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--domain', choices=['wasm', 'evm', 'android', 'flutter-aot', 'sqlite', 'pcap', '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 = {}
|
|
pcap_calls = []
|
|
pcap_attempts = []
|
|
workflow_checks = {'pcap_exact_call_sequence': False} if args.domain == 'pcap' else {}
|
|
tool_errors = []
|
|
with (args.trial / 'events.jsonl').open() as event_lines:
|
|
for line in event_lines:
|
|
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 args.domain == 'pcap':
|
|
pcap_attempts.append({'tool': item.get('tool'), 'arguments': item.get('arguments'),
|
|
'status': item.get('status'),
|
|
'evidence_id': parsed.get('evidence_id') if isinstance(parsed, dict) else None,
|
|
'readable_evidence': isinstance(parsed, dict)
|
|
and isinstance(parsed.get('normalized_result'), dict)
|
|
and isinstance(parsed.get('evidence_id'), str)
|
|
and re.fullmatch(r'ev_[a-f0-9]{64}', parsed['evidence_id']) is not None})
|
|
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:
|
|
if args.domain == 'pcap':
|
|
pcap_calls.append((item['tool'], item.get('arguments'), parsed))
|
|
else:
|
|
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})
|
|
error = parsed.get('error') if isinstance(parsed, dict) else None
|
|
if isinstance(error, dict) and error.get('code') == 'resource_constraint':
|
|
details = error.get('details')
|
|
limits = details.get('reported_limits') if isinstance(details, dict) else None
|
|
if isinstance(limits, dict):
|
|
observed_ids.update(evidence_ids(limits.get('evidence_reference')))
|
|
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 == 'sqlite':
|
|
for key in ['artifact', 'wal', 'engine', 'database_encoding', 'schema', 'rows', 'producer_limitations', 'limitations']:
|
|
# Missing is not SQL/JSON null; booleans are not numeric counters.
|
|
check(key, (key in answer, json.dumps(answer.get(key), sort_keys=True)),
|
|
(True, json.dumps(reference[key], sort_keys=True)))
|
|
elif domain == 'pcap':
|
|
check('capture', ('capture' in answer, json.dumps(answer.get('capture'), sort_keys=True)),
|
|
(True, json.dumps(reference['capture'], sort_keys=True)))
|
|
prefix = answer.get('http_request_uris_first2')
|
|
records = {}
|
|
for tool, arguments, value in pcap_calls:
|
|
if not isinstance(value, dict) or not isinstance(value.get('evidence_id'), str):
|
|
continue
|
|
identity = value['evidence_id']
|
|
normalized = value.get('normalized_result')
|
|
supported = (isinstance(normalized, dict) and isinstance(value.get('limitations'), list)
|
|
and all(isinstance(item, str) for item in value['limitations']))
|
|
if supported:
|
|
supported = all(key not in normalized or (isinstance(normalized[key], list)
|
|
and all(isinstance(item, str) for item in normalized[key]))
|
|
for key in ['limitations', 'unknowns'])
|
|
if supported and ('summary' in normalized or normalized.get('kind') == 'summary'):
|
|
supported = isinstance(normalized.get('summary'), dict)
|
|
if supported and ('page' in normalized or normalized.get('kind') == 'pcap'):
|
|
supported = isinstance(normalized.get('page'), dict)
|
|
entry = (tool, arguments, value) if supported else None
|
|
if identity not in records:
|
|
records[identity] = entry
|
|
elif json.dumps(records[identity], sort_keys=True) != json.dumps(entry, sort_keys=True):
|
|
# An ID with conflicting producers or envelopes is ambiguous.
|
|
records[identity] = None
|
|
observations = answer.get('delivery_observations')
|
|
canonical_observations = [pcap_observation_fields(value) for value in observations] if isinstance(observations, list) else observations
|
|
if isinstance(prefix, dict):
|
|
checks['pcap_prefix_fields_supported'] = set(prefix) <= {'items', 'coverage', 'source_coverage'} and 'items' in prefix
|
|
if 'source_coverage' in prefix:
|
|
check('pcap_prefix_source_coverage', prefix['source_coverage'], reference['capture']['coverage'])
|
|
if 'coverage' in prefix:
|
|
last = canonical_observations[-1] if isinstance(canonical_observations, list) and canonical_observations else None
|
|
identity = last.get('evidence_id') if isinstance(last, dict) else None
|
|
entry = records.get(identity) if isinstance(identity, str) else None
|
|
record = entry[2] if entry is not None else None
|
|
page = record['normalized_result'] if record is not None else {}
|
|
checks['pcap_prefix_coverage_observed'] = (
|
|
page.get('kind') == 'pcap'
|
|
and page.get('page', {}).get('facet') == 'http-uris'
|
|
and json.dumps(page['page'].get('items'), sort_keys=True) == json.dumps(prefix.get('items'), sort_keys=True)
|
|
and 'coverage' in page
|
|
and json.dumps(prefix['coverage'], sort_keys=True) == json.dumps(page['coverage'], sort_keys=True)
|
|
)
|
|
prefix = prefix.get('items')
|
|
check('http_request_uris_first2', ('http_request_uris_first2' in answer, json.dumps(prefix, sort_keys=True)),
|
|
(True, json.dumps(reference['http_request_uris_first2'], sort_keys=True)))
|
|
expected_observations = []
|
|
if isinstance(canonical_observations, list):
|
|
for observation in canonical_observations:
|
|
identity = observation.get('evidence_id') if isinstance(observation, dict) else None
|
|
entry = records.get(identity) if isinstance(identity, str) else None
|
|
record = entry[2] if entry is not None else None
|
|
if record is None:
|
|
expected_observations.append(None)
|
|
continue
|
|
value = record['normalized_result']
|
|
expected = {'evidence_id': record['evidence_id'], 'limitations': record['limitations']}
|
|
for key in ['kind', 'view', 'parent_evidence_id', 'coverage', 'unknowns']:
|
|
if key in value:
|
|
expected[key] = value[key]
|
|
if 'limitations' in value:
|
|
expected['normalized_limitations'] = value['limitations']
|
|
if 'source_coverage' in observation and 'source_coverage' in value:
|
|
expected['source_coverage'] = value['source_coverage']
|
|
expected_observations.append(expected)
|
|
checks['pcap_delivery_observations_nonempty'] = isinstance(observations, list) and bool(observations)
|
|
check('pcap_original_delivery_observations', json.dumps(canonical_observations, sort_keys=True),
|
|
json.dumps(expected_observations, sort_keys=True))
|
|
if isinstance(canonical_observations, list) and canonical_observations and all(isinstance(x, dict) for x in canonical_observations):
|
|
check('pcap_requested_delivery_chain', [x.get('kind', 'complete') for x in canonical_observations],
|
|
[x.get('kind', 'complete') for x in reference['delivery_observations']])
|
|
check('pcap_requested_views', json.dumps([x.get('view') for x in canonical_observations], sort_keys=True),
|
|
json.dumps([x.get('view') for x in reference['delivery_observations']], sort_keys=True))
|
|
if len(reference['delivery_observations']) == 2:
|
|
check('pcap_summary_coverage', json.dumps(canonical_observations[0].get('coverage'), sort_keys=True),
|
|
json.dumps({'status': 'complete-within-view', 'examined': 1, 'total': 1,
|
|
'next_offset': None, 'exhausted': True}, sort_keys=True))
|
|
identities = [x.get('evidence_id') for x in canonical_observations]
|
|
checks['pcap_observations_have_reported_ids'] = (
|
|
isinstance(answer.get('evidence_ids'), list)
|
|
and answer['evidence_ids'] == identities
|
|
and all(isinstance(value, str) and re.fullmatch(r'ev_[a-f0-9]{64}', value) for value in identities)
|
|
and len(set(identities)) == len(identities)
|
|
)
|
|
entries = [records.get(identity) if isinstance(identity, str) else None for identity in identities]
|
|
cited = [entry[2] if entry is not None else None for entry in entries]
|
|
expected_tools = ['inspect_pcap_capture'] + (['inspect_analysis_view'] if len(reference['delivery_observations']) == 2 else [])
|
|
check('pcap_cited_producer_tools', [entry[0] if entry is not None else None for entry in entries], expected_tools)
|
|
# Summary delivery delegates to a retained view; its envelope operation
|
|
# is inspect_analysis_view even though the requested tool is inspect_pcap_capture.
|
|
expected_operations = ['inspect_pcap_capture'] if len(expected_tools) == 1 else ['inspect_analysis_view'] * 2
|
|
check('pcap_cited_operations', [record.get('operation') if isinstance(record, dict) else None for record in cited], expected_operations)
|
|
path = reference['capture']['target']['path']
|
|
expected_arguments = [{'path': path}]
|
|
expected_parameters = [{'path': path}]
|
|
if len(expected_tools) == 2:
|
|
parent = cited[0]['normalized_result'].get('parent_evidence_id') if isinstance(cited[0], dict) else None
|
|
views = [observation.get('view') for observation in reference['delivery_observations']]
|
|
expected_arguments = [{'path': path, 'detail': 'summary'},
|
|
{'source': {'kind': 'retained-evidence', 'evidence_id': parent}, 'view': views[-1]}]
|
|
expected_parameters = [{'parent_evidence_id': parent, 'view': view} for view in views]
|
|
check('pcap_cited_call_arguments', json.dumps([entry[1] if entry is not None else None for entry in entries], sort_keys=True),
|
|
json.dumps(expected_arguments, sort_keys=True))
|
|
workflow_checks['pcap_exact_call_sequence'] = (
|
|
len(identities) == len(expected_tools)
|
|
and json.dumps(pcap_attempts, sort_keys=True) == json.dumps([
|
|
{'tool': tool, 'arguments': arguments, 'status': 'completed',
|
|
'evidence_id': identity, 'readable_evidence': True}
|
|
for tool, arguments, identity in zip(expected_tools, expected_arguments, identities)
|
|
], sort_keys=True)
|
|
)
|
|
check('pcap_cited_parameters', json.dumps([record.get('parameters') if isinstance(record, dict) else None for record in cited], sort_keys=True),
|
|
json.dumps(expected_parameters, sort_keys=True))
|
|
checks['pcap_cited_subjects_match_artifact'] = bool(cited) and all(
|
|
isinstance(record, dict)
|
|
and isinstance(record.get('subject'), dict)
|
|
and isinstance(record['subject'].get('digest'), dict)
|
|
and record['subject']['digest'].get('sha256') == reference['artifact_sha256']
|
|
and record['subject'].get('local_path') == reference['capture']['target']['path']
|
|
for record in cited
|
|
)
|
|
checks['pcap_cited_parents_linked'] = bool(cited) and all(isinstance(record, dict) for record in cited) and (
|
|
len(cited) == 1 or (len(cited) == 2 and all(
|
|
isinstance(record['normalized_result'].get('parent_evidence_id'), str)
|
|
and re.fullmatch(r'ev_[a-f0-9]{64}', record['normalized_result']['parent_evidence_id'])
|
|
and record['normalized_result'].get('parent_digest') == record['normalized_result']['parent_evidence_id'][3:]
|
|
and record['normalized_result']['parent_evidence_id'] != record['evidence_id']
|
|
and isinstance(record.get('evidence_links'), list)
|
|
and record['normalized_result']['parent_evidence_id'] in record['evidence_links']
|
|
for record in cited
|
|
))
|
|
)
|
|
checks['pcap_cited_source_supports_uri_prefix'] = False
|
|
checks['pcap_cited_capture'] = False
|
|
if all(isinstance(record, dict) for record in cited):
|
|
if len(cited) == 1:
|
|
source = cited[0]['normalized_result']
|
|
check('pcap_cited_capture', json.dumps(answer.get('capture'), sort_keys=True),
|
|
json.dumps(pcap_complete_capture(source), sort_keys=True))
|
|
checks['pcap_cited_source_supports_uri_prefix'] = (
|
|
source.get('target') == reference['capture']['target']
|
|
and isinstance(source.get('http_request_uris'), list)
|
|
and json.dumps(source['http_request_uris'][:2], sort_keys=True) == json.dumps(prefix, sort_keys=True)
|
|
)
|
|
elif len(cited) == 2:
|
|
source, page = [record['normalized_result'] for record in cited]
|
|
capture = source.get('summary', {})
|
|
check('pcap_cited_capture', json.dumps(answer.get('capture'), sort_keys=True),
|
|
json.dumps(capture, sort_keys=True))
|
|
selected_page = page.get('page', {})
|
|
checks['pcap_cited_source_supports_uri_prefix'] = (
|
|
source.get('kind') == 'summary'
|
|
and source.get('parent_operation') == 'inspect_pcap_capture'
|
|
and capture == reference['capture']
|
|
and source.get('artifact') == {key: capture['target'][key] for key in ['path', 'sha256']}
|
|
and isinstance(source.get('parent_evidence_id'), str)
|
|
and page.get('parent_evidence_id') == source['parent_evidence_id']
|
|
and page.get('parent_operation') == 'inspect_pcap_capture'
|
|
and json.dumps(page.get('view'), sort_keys=True) == json.dumps({'kind': 'pcap', 'facet': 'http-uris', 'offset': 0, 'limit': 2}, sort_keys=True)
|
|
and selected_page.get('facet') == 'http-uris'
|
|
and json.dumps(selected_page.get('items'), sort_keys=True) == json.dumps(prefix, sort_keys=True)
|
|
and page.get('source_coverage') == capture.get('coverage')
|
|
and json.dumps(page.get('coverage'), sort_keys=True) == json.dumps(reference['delivery_observations'][-1].get('coverage'), sort_keys=True)
|
|
and page.get('artifact') == {key: capture['target'][key] for key in ['path', 'sha256']}
|
|
)
|
|
else:
|
|
checks['pcap_requested_delivery_chain'] = False
|
|
if len(reference['delivery_observations']) == 2:
|
|
checks['pcap_summary_coverage'] = False
|
|
checks['pcap_observations_have_reported_ids'] = False
|
|
checks['pcap_requested_views'] = False
|
|
checks['pcap_cited_source_supports_uri_prefix'] = False
|
|
checks['pcap_cited_capture'] = False
|
|
checks['pcap_cited_subjects_match_artifact'] = False
|
|
checks['pcap_cited_parents_linked'] = False
|
|
checks['pcap_cited_producer_tools'] = False
|
|
checks['pcap_cited_operations'] = False
|
|
checks['pcap_cited_call_arguments'] = False
|
|
checks['pcap_cited_parameters'] = False
|
|
elif domain == 'flutter-aot':
|
|
for key in ['target', 'abi', 'libapp', 'string_pool', 'coverage', 'limitations']:
|
|
check(key, answer.get(key), reference[key])
|
|
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)
|
|
checks.update(workflow_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()),
|
|
**({'workflow_checks': workflow_checks, 'all_workflow_checks_pass': all(workflow_checks.values())}
|
|
if domain == 'pcap' else {}),
|
|
'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()
|