305 lines
19 KiB
Python
305 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""Read-only P04 evidence preflight. Never grants original acceptance.
|
|
|
|
Required conditions and authority bindings are fixed here, never supplied by an index.
|
|
See execution/p04-01/CHECKER_PROTOCOL.md for the supported input subset.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SPEC = ROOT / 'docs/spec/acceptance/P04-01.json'
|
|
SPEC_HASH = '9ec722e7dfb0bee178d291624a026420c08412574e238bba1fde012c7a6f7881'
|
|
DOMAINS = {'閉店', '停電', '破壊', '退出', '課金', 'シフト交代', '00:00決算'}
|
|
OWNER_HASH = '4583ee9aad45c4db183d72df4420df48aa8e0897e3b5efda13016f0b5cdbd68e'
|
|
# Condition -> (domain, authority refs, required engine measurements).
|
|
# Measurements describe safety/counts, not the still-pending attribution policy.
|
|
REQUIRED = {
|
|
'power_stop': ('停電', ('BASELINE:183',), {'charges': 0}),
|
|
'destruction_stop': ('破壊', ('BASELINE:183',), {'charges': 0}),
|
|
'dependency_stop': ('閉店', ('D05-P04-DEPENDENCY',), {'charges': 0, 'progress_delta': 0, 'reservation_ids_preserved': True, 'service_id_preserved': True}),
|
|
'shift_completion': ('シフト交代', ('BASELINE:64', 'BASELINE:67', 'D05-P04-SHIFT-1800'), {'minute': 1080, 'completions': 1, 'charges': 1, 'completion_before_shift': True}),
|
|
'recovery_completion': ('課金', ('D05-P04-DEPENDENCY',), {'completions': 1, 'charges': 1}),
|
|
'same_minute_replay': ('課金', ('D05-P04-DEPENDENCY', 'D05-P04-SHIFT-1800'), {'completions': 1, 'charges': 1, 'duplicate_generations': 0, 'same_minute_replayed': True}),
|
|
'closure_release': ('閉店', ('D-034',), {'orphan_reservations': 0, 'released_once': True, 'started_service_continues': True}),
|
|
'exit_release': ('退出', ('D-034',), {'orphan_reservations': 0, 'released_once': True}),
|
|
'patience_release': ('退出', ('D-034',), {'orphan_reservations': 0, 'released_once': True}),
|
|
'paused_midnight': ('00:00決算', ('D05-P04-DEPENDENCY', 'BASELINE:71', 'BASELINE:72'), {'orphan_reservations': 0, 'crossed_midnight_while_paused': True, 'stopped_charges': 0, 'completions': 1, 'charges': 1}),
|
|
'midnight_settlement': ('00:00決算', ('BASELINE:71', 'BASELINE:72', 'A01'), {'settlements': 1, 'sales_payments': 1, 'payroll_payments': 1, 'upkeep_payments': 1, 'duty_payments': 1, 'nonzero_payroll': True}),
|
|
'midnight_replay': ('00:00決算', ('BASELINE:71', 'BASELINE:72', 'A01'), {'settlements': 1, 'same_minute_replayed': True, 'duplicate_payments': 0}),
|
|
'save_restore': ('00:00決算', ('BASELINE:71', 'BASELINE:72', 'A01'), {'settlements': 1, 'duplicate_payments': 0, 'formal_save_restored': True}),
|
|
'night_staff_absent': ('シフト交代', ('BASELINE:67', 'D05-P04-SHIFT-1800'), {'minute': 1081, 'state': 'WAITING_FOR_STAFF', 'night_staff_count': 0}),
|
|
'overtime_task': ('シフト交代', ('BASELINE:67',), {'interrupt_level': 0, 'is_overtime': True, 'duty_recorded_once': True}),
|
|
}
|
|
REACH_REFS = {
|
|
'EV': [('autoload/simulation_runtime.gd', 6733, 6742),
|
|
('scripts/staff/task_dispatcher.gd', 256, 261),
|
|
('autoload/simulation_runtime.gd', 9141, 9170),
|
|
('autoload/simulation_runtime.gd', 8986, 8987),
|
|
('scripts/transport/elevator_service.gd', 279, 284)],
|
|
'transport': [('autoload/simulation_runtime.gd', 12023, 12030),
|
|
('autoload/simulation_runtime.gd', 12976, 13050)],
|
|
}
|
|
|
|
|
|
def require(condition, message):
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def sha(path):
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def unique(pairs):
|
|
result = {}
|
|
for key, value in pairs:
|
|
require(key not in result, 'duplicate JSON key: ' + key)
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def load(path):
|
|
return json.loads(path.read_text(encoding='utf-8'), object_pairs_hook=unique,
|
|
parse_constant=lambda x: require(False, 'nonfinite JSON: ' + x))
|
|
|
|
|
|
def safe(base, relative):
|
|
require(isinstance(relative, str) and relative, 'invalid path')
|
|
p = Path(relative)
|
|
require(not p.is_absolute() and all(x not in {'.', '..'} for x in relative.split('/')),
|
|
'unsafe path: ' + relative)
|
|
current = base
|
|
for part in p.parts:
|
|
current = current / part
|
|
require(not current.is_symlink(), 'symlink: ' + relative)
|
|
require(current.resolve().is_relative_to(base.resolve()) and current.is_file(),
|
|
'missing/outside file: ' + relative)
|
|
return current
|
|
|
|
|
|
def pins(mapping, base):
|
|
require(isinstance(mapping, dict) and mapping, 'missing pin map')
|
|
for name, digest in mapping.items():
|
|
require(isinstance(digest, str) and re.fullmatch('[0-9a-f]{64}', digest), 'invalid hash: ' + name)
|
|
require(sha(safe(base, name)) == digest, 'stale pin: ' + name)
|
|
|
|
|
|
def pointer(value, path):
|
|
require(isinstance(path, str) and (path == '' or path.startswith('/')), 'invalid JSON pointer')
|
|
for token in path.split('/')[1:]:
|
|
require(not re.search('~(?![01])', token), 'invalid pointer escape')
|
|
token = token.replace('~1', '/').replace('~0', '~')
|
|
if isinstance(value, list):
|
|
require(re.fullmatch('0|[1-9][0-9]*', token) is not None, 'invalid array index')
|
|
value = value[int(token)]
|
|
else:
|
|
require(isinstance(value, dict) and token in value, 'missing pointer: ' + path)
|
|
value = value[token]
|
|
return value
|
|
|
|
|
|
def literal(a, b):
|
|
return type(a) is type(b) and json.dumps(a, sort_keys=True) == json.dumps(b, sort_keys=True)
|
|
|
|
|
|
def coverage(index, raw, source):
|
|
authority = index['authority']
|
|
for ref in {r for _, refs, _ in REQUIRED.values() for r in refs}:
|
|
entry = authority.get(ref, {})
|
|
require(entry.get('status') in {'original', 'approved'}, 'pending/missing authority: ' + ref)
|
|
aid = entry.get('artifact_id')
|
|
require(aid in raw and index['artifacts'][aid]['type'] == 'authority', 'raw authority: ' + ref)
|
|
if ref.startswith('D05-P04-'):
|
|
require(sha(raw[aid]) == OWNER_HASH, 'owner approval hash mismatch')
|
|
owner = load(raw[aid])
|
|
require(owner.get('origin') == 'direct_user_reply' and owner.get('formal_acceptance') is False,
|
|
'owner approval scope')
|
|
require(ref in {v['id'] for v in owner['approved_choices']}, 'owner approval condition')
|
|
else:
|
|
path = ('docs/traceability/original/BASELINE.txt' if ref.startswith('BASELINE:') else
|
|
'docs/DESIGN_DECISIONS.md' if ref == 'D-034' else
|
|
'docs/plans/original-completion-2026-10-02/execution/p04-01-input-r01/ORDER_CONTRACT_DRAFT.md')
|
|
require(sha(raw[aid]) == sha(safe(ROOT, path)), 'authority document mismatch: ' + ref)
|
|
conditions = index.get('conditions')
|
|
require(isinstance(conditions, dict) and set(conditions) == set(REQUIRED), 'mandatory conditions missing/unknown')
|
|
for cid, (domain_name, refs, expected) in REQUIRED.items():
|
|
condition = conditions[cid]
|
|
require(condition.get('authority_refs') == list(refs), 'condition authority: ' + cid)
|
|
require(condition.get('nativeWindows') == 'pending', 'nativeWindows condition: ' + cid)
|
|
domain = index['domains'][domain_name]
|
|
require(set(refs) <= set(domain['authority_refs']), 'domain authority: ' + cid)
|
|
cases = [c for c in domain['cases'] if c['id'] == condition.get('case_id')]
|
|
require(len(cases) == 1, 'mandatory engine case: ' + cid)
|
|
case = cases[0]
|
|
require(case.get('raw_origin') == 'engine', 'case origin: ' + cid)
|
|
observation = load(raw[case['after']])
|
|
measurements = observation.get('conditions', {}).get(cid)
|
|
require(isinstance(measurements, dict), 'missing engine measurements: ' + cid)
|
|
for metric, value in expected.items():
|
|
require(literal(measurements.get(metric), value), cid + ': measurement ' + metric)
|
|
require(any(p.get('case_id') == case['id'] and p.get('artifact_id') == case['after']
|
|
and p.get('json_pointer') == '/conditions/' + cid + '/' + metric
|
|
and p.get('operator') == 'eq' and literal(p.get('expected'), value)
|
|
and p.get('authority_ref') in refs for p in domain['invariants']),
|
|
cid + ': missing mandatory invariant ' + metric)
|
|
if cid == 'save_restore':
|
|
writer = load(raw[case['receipt']])
|
|
reader_id = condition.get('restore_receipt')
|
|
require(reader_id in raw, 'missing restore receipt')
|
|
reader = load(raw[reader_id])
|
|
require(type(reader.get('start_pid')) is int and reader['start_pid'] > 0
|
|
and reader['start_pid'] != writer['start_pid']
|
|
and reader.get('end_pid') == reader['start_pid']
|
|
and type(reader.get('exit_code')) is int and reader['exit_code'] == 0
|
|
and reader.get('engine_ended') is True, 'separate PID restore receipt')
|
|
require(reader.get('source_map') == source and reader.get('environment') == index['environment'], 'restore applicability')
|
|
require(reader.get('case_id') == case['id'] and
|
|
reader.get('artifact_hashes') == writer.get('artifact_hashes'), 'restore case artifact binding')
|
|
save_id = condition.get('save_artifact')
|
|
require(save_id in raw and index['artifacts'][save_id]['type'] == 'save', 'formal save artifact')
|
|
require(writer.get('saved_sha256') == sha(raw[save_id]) == reader.get('restored_sha256'), 'save/restore identity')
|
|
require(reader.get('stdout') in raw and reader.get('native_log') in raw, 'restore logs')
|
|
reachability = index.get('reachability')
|
|
require(isinstance(reachability, dict) and set(reachability) == {'EV', 'transport'}, 'reachability required')
|
|
for name, decision in reachability.items():
|
|
refs = decision.get('source_refs')
|
|
require(isinstance(refs, list) and refs and isinstance(decision.get('reason'), str)
|
|
and decision['reason'].strip(), 'unsupported reach=false: ' + name)
|
|
for ref in refs:
|
|
require(ref.get('path') in source and ref.get('sha256') == source[ref['path']], 'reachability source pin')
|
|
lines = safe(ROOT, ref['path']).read_text().splitlines()
|
|
start, end = ref.get('start'), ref.get('end')
|
|
require(type(start) is int and type(end) is int and 1 <= start <= end <= len(lines), 'reachability lines')
|
|
require(ref.get('text') == '\n'.join(lines[start - 1:end]), 'reachability source text')
|
|
require(set(REACH_REFS[name]) <= {(r['path'], r['start'], r['end']) for r in refs}, 'missing static reachability chain: ' + name)
|
|
# Staff dispatch reserves task items before travel. A guest-only exclusion
|
|
# must not hide this reachable EV reservation path.
|
|
if name == 'EV':
|
|
require(decision.get('reach') is True, 'EV staff reservation path is reachable')
|
|
domain = index['domains']['停電']
|
|
cases = [c for c in domain['cases'] if c['id'] == decision.get('case_id')]
|
|
require(len(cases) == 1 and cases[0].get('raw_origin') == 'engine', 'EV engine case')
|
|
case = cases[0]
|
|
ev = load(raw[case['after']]).get('ev', {})
|
|
require(ev.get('reserved_before_travel') is True and ev.get('power_stopped') is True,
|
|
'EV reserved staff power-stop scenario')
|
|
require(any(p.get('case_id') == case['id'] and p.get('artifact_id') == case['after']
|
|
and p.get('json_pointer') == '/ev/orphan_reservations' and p.get('operator') == 'eq'
|
|
and literal(p.get('expected'), 0) for p in domain['invariants']), 'EV orphan invariant')
|
|
else:
|
|
require(decision.get('reach') is False, 'transport reservation/payment exclusion')
|
|
|
|
|
|
def verify(directory):
|
|
# Contract hashes the supplied spec string, while the checked-in JSON has
|
|
# one final LF. Accept only that exact representation difference.
|
|
spec_bytes = SPEC.read_bytes()
|
|
require(sha(SPEC) == SPEC_HASH or (spec_bytes.endswith(b'\n') and
|
|
hashlib.sha256(spec_bytes[:-1]).hexdigest() == SPEC_HASH), 'dispatch spec pin mismatch')
|
|
spec = load(SPEC)
|
|
evidence = directory / 'p04-evidence'
|
|
require(not evidence.is_symlink(), 'symlink evidence directory')
|
|
index = load(safe(evidence, 'index.json'))
|
|
require(type(index.get('schema_version')) is int and index['schema_version'] == 1, 'schema_version')
|
|
require(index.get('task_id') == 'P04-01' and index.get('formal_acceptance') is False, 'task/acceptance')
|
|
head = subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=ROOT, text=True).strip()
|
|
require(index.get('candidate_commit') == head, 'candidate_commit mismatch')
|
|
source = index.get('source_map')
|
|
pins(source, ROOT)
|
|
for name, digest in spec['current_source_pins'].items():
|
|
require(source.get(name) == digest, 'immutable source pin: ' + name)
|
|
oldmap = load(ROOT / 'docs/plans/original-completion-2026-10-02/execution/p04-01-input-r01/historical-source-map.json')
|
|
required = {p for p in oldmap if p.endswith('.gd') and p.startswith(('autoload/', 'scripts/'))}
|
|
require(required <= source.keys(), 'missing runtime/service dependency source pins')
|
|
environment = index.get('environment')
|
|
require(isinstance(environment, dict) and isinstance(environment.get('os'), str) and environment['os'], 'environment OS')
|
|
for kind in ('engine', 'configuration', 'inputs'):
|
|
pins(environment.get(kind), evidence)
|
|
artifacts = index.get('artifacts')
|
|
require(isinstance(artifacts, dict) and artifacts, 'missing artifacts')
|
|
paths, raw = set(), {}
|
|
for aid, item in artifacts.items():
|
|
require(isinstance(aid, str) and aid and isinstance(item, dict), 'invalid artifact')
|
|
path = safe(evidence, item['path'])
|
|
require(path not in paths, 'duplicate artifact path')
|
|
paths.add(path)
|
|
require(type(item.get('bytes')) is int and item['bytes'] == path.stat().st_size, 'artifact size: ' + aid)
|
|
require(sha(path) == item.get('sha256'), 'artifact hash: ' + aid)
|
|
require(item.get('type') in {'observation', 'state', 'receipt', 'stdout', 'native_log', 'save', 'authority'}, 'artifact type')
|
|
raw[aid] = path
|
|
authority = index.get('authority')
|
|
require(isinstance(authority, dict) and authority, 'missing authority')
|
|
domains = index.get('domains')
|
|
require(isinstance(domains, dict) and set(domains) == DOMAINS, 'required seven domains')
|
|
for name, domain in domains.items():
|
|
require(isinstance(domain, dict), 'invalid domain: ' + name)
|
|
for field in ('source_refs', 'authority_refs', 'cases', 'invariants'):
|
|
require(isinstance(domain.get(field), list) and domain[field], name + ': missing ' + field)
|
|
require(domain.get('applicability') == {'source_map': source, 'environment': environment}, name + ': applicability mismatch')
|
|
for ref in domain['source_refs']:
|
|
require(isinstance(ref, dict) and ref.get('path') in source and isinstance(ref.get('function'), str) and ref['function'], 'invalid source ref')
|
|
require(ref.get('sha256') == source[ref['path']], 'source ref hash')
|
|
for ref in domain['authority_refs']:
|
|
require(ref in authority and authority[ref].get('status') in {'original', 'approved'}, 'pending/missing authority')
|
|
require(authority[ref].get('artifact_id') in raw, 'missing raw authority')
|
|
case_ids = set()
|
|
for case in domain['cases']:
|
|
require(isinstance(case, dict) and isinstance(case.get('id'), str) and case['id'] not in case_ids, 'duplicate/invalid case')
|
|
case_ids.add(case['id'])
|
|
for field in ('input', 'before', 'operation', 'after', 'receipt'):
|
|
require(case.get(field) in raw, name + ': missing case artifact ' + field)
|
|
receipt = load(raw[case['receipt']])
|
|
pid = receipt.get('start_pid')
|
|
require(type(pid) is int and pid > 0 and receipt.get('end_pid') == pid and type(receipt.get('exit_code')) is int and receipt['exit_code'] == 0 and receipt.get('engine_ended') is True, 'unexecuted/incomplete receipt')
|
|
require(receipt.get('source_map') == source and receipt.get('environment') == environment, 'receipt applicability')
|
|
for field in ('stdout', 'native_log'):
|
|
require(receipt.get(field) in raw, 'receipt missing raw ' + field)
|
|
require(receipt.get('case_id') == case['id'], 'receipt case identity')
|
|
require(receipt.get('artifact_hashes') == {field: sha(raw[case[field]]) for field in ('input', 'before', 'operation', 'after')}, 'receipt artifact binding')
|
|
for pred in domain['invariants']:
|
|
require(isinstance(pred, dict) and pred.get('artifact_id') in raw, 'missing predicate artifact')
|
|
require(pred.get('authority_ref') in domain['authority_refs'], 'predicate authority')
|
|
require(pred.get('raw_origin') == 'engine', 'unsupported derivation/origin')
|
|
value = pointer(load(raw[pred['artifact_id']]), pred['json_pointer'])
|
|
op = pred.get('operator')
|
|
if op in {'same_literal', 'distinct'}:
|
|
require(pred.get('paired_artifact_id') in raw, 'missing paired artifact')
|
|
other = pointer(load(raw[pred['paired_artifact_id']]), pred['paired_json_pointer'])
|
|
result = literal(value, other) if op == 'same_literal' else not literal(value, other)
|
|
elif op == 'eq':
|
|
result = literal(value, pred['expected'])
|
|
elif op == 'nonnegative':
|
|
result = type(value) in (int, float) and value >= 0
|
|
elif op == 'count_eq':
|
|
result = isinstance(value, (list, dict)) and type(pred.get('expected')) is int and len(value) == pred['expected']
|
|
else:
|
|
raise ValueError('unsupported predicate: ' + str(op))
|
|
require(result, name + ': predicate failed')
|
|
require(index.get('review_status') in {'pending_review', 'approved'}, 'review status')
|
|
coverage(index, raw, source)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--verify-only', action='store_true', required=True)
|
|
parser.parse_args()
|
|
try:
|
|
require(bool(os.environ.get('BUILD_OUTPUT_DIR')), 'BUILD_OUTPUT_DIR required')
|
|
verify(Path(os.environ['BUILD_OUTPUT_DIR']))
|
|
except (ValueError, OSError, KeyError, TypeError, IndexError, AttributeError, subprocess.SubprocessError) as exc:
|
|
print(json.dumps({'task_id': 'P04-01', 'machine_verified': False, 'formal_acceptance': False, 'error': str(exc)}, ensure_ascii=False))
|
|
return 1
|
|
print(json.dumps({'task_id': 'P04-01', 'machine_verified': True, 'formal_acceptance': False}))
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|