#!/usr/bin/env python3 """Partial read-only semantic diagnostics. This version cannot accept P23-03. Whole typed inputs are preserved. The existing typed codec is reused verbatim. Unimplemented visual/OS checks are explicit blockers, never successful checks. """ import argparse import copy import hashlib import importlib.util import json import math from pathlib import Path import re import sys sys.dont_write_bytecode = True ROOT = Path(__file__).resolve().parent.parent SPEC = 'docs/spec/acceptance/P23-03.json' SPEC_SHA = '8568457df849ef587c731f072d77244ef5420fc4e1190c5069cfd6cfeea9c89f' CONDITIONS = ( 'room_occupancy_and_operation', 'queues_and_overflow', 'elevator_state_and_passengers', 'staff_state_and_jobs', 'supplies_and_carried_goods', 'same_demand_route_improvement', 'normal_os_and_distinct_pid_resume', ) BLOCKERS = { CONDITIONS[0]: 'Service/rescue occupancy, reason fields and room rectangles checked; rest route selection and badge pixels unfinished.', CONDITIONS[1]: 'Queue lane allocation, overflow and positions checked; screenshot pixel provenance unfinished.', CONDITIONS[2]: 'Shaft/trip and cabin geometry checked; prior-frame interpolation history and passenger geometry unfinished.', CONDITIONS[3]: 'Typed staff/task projection implemented; motion/rest/rescue/cleaning drawing recomputation unfinished.', CONDITIONS[4]: 'Carried inventory ID and logistics report checked; supply badge pixels unfinished.', CONDITIONS[5]: 'Public-command A/B fixture and typed arrival metric added; native execution, definition pins and full command/path provenance checks pending.', CONDITIONS[6]: 'P23 passive request/ACK collector and independent CUA/formal-resume gates unfinished; Root evidence missing.', } def module(name, path): spec = importlib.util.spec_from_file_location(name, path) result = importlib.util.module_from_spec(spec) spec.loader.exec_module(result) return result codec = module('p23_existing_typed_codec', ROOT / 'tools/check_p08_migration_evidence.py') def decode(node): # Keep P08 scalar semantics, adding the renderer's Rect2 and vector keys. if node['type'] == 'Dictionary': result = {} for key, value in node['entries']: key = decode(key) if isinstance(key, list): key = tuple(key) require(key not in result, 'duplicate typed dictionary key') result[key] = decode(value) return result if node['type'] == 'Array': return [decode(value) for value in node['items']] if node['type'] == 'Rect2': match = re.fullmatch(r'Rect2\(([^()]*)\)', node['text']) require(match is not None, 'invalid typed Rect2') values = [float(v.strip()) for v in match[1].split(',')] require(len(values) == 4 and all(math.isfinite(v) for v in values), 'invalid rectangle') return values return codec.decode(node) def require(value, message): if not value: raise ValueError(message) def sha(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def enum_names(root, relative, name): source = (root / relative).read_text() match = re.search(r'\benum\s+' + re.escape(name) + r'\s*\{([^}]+)\}', source) require(match is not None, 'missing source enum ' + name) names = [x.strip() for x in match[1].split(',') if x.strip()] require(all(re.fullmatch(r'[A-Z][A-Z_0-9]*', x) for x in names), 'unsupported enum schema') require(len(set(names)) == len(names), 'duplicate source enum') return names def projection(rows, mapping, states): result = [] ids = set() for row in rows: value = {target: copy.deepcopy(row[source]) for target, source in mapping.items()} require(value['id'] not in ids, 'duplicate actor/task ID') ids.add(value['id']) require(value['state'] in states, 'unknown typed state') result.append(value) return result def compare_actor_rows(snapshot, report, root=ROOT): runtime = snapshot['systems']['runtime'] guest_states = enum_names(root, 'scripts/guest/guest_runtime.gd', 'State') staff_states = enum_names(root, 'scripts/staff/staff_runtime.gd', 'State') task_states = enum_names(root, 'scripts/staff/staff_task.gd', 'State') mappings = ( ('guests', runtime['guests']['guests'], guest_states, dict(id='guest_id', state='state', cell='current_cell', target='target_facility_id', service='current_service_id', queue='queue_request_id')), ('staff', runtime['staff']['staff'], staff_states, dict(id='staff_id', state='state', cell='current_cell', task='current_task_id', facility='assigned_facility_id', role='role_id', carried_inventory_id='carried_inventory_id')), ('tasks', runtime['staff']['tasks'], task_states, dict(id='task_id', state='state', type='task_type', staff='assigned_staff_id', target='target_entity_id', rescue_facility='rescue_facility_id')), ) for key, rows, states, mapping in mappings: expected = projection(rows, mapping, states) require(report[key] == expected, 'typed projection mismatch: ' + key) return [key for key, *_ in mappings] def elevator_report(snapshot): saved = snapshot['systems']['runtime']['elevators'] rows = [] for shaft in saved['shafts']: row = copy.deepcopy(shaft) row.update(shaft['config']) floor = float(shaft['current_floor']) if shaft['phase'] in ('PICKUP', 'TRAVEL'): destination = shaft['boarding_floor'] if shaft['phase'] == 'PICKUP' else shaft['target_floor'] progress = 1.0 - shaft['remaining_minutes'] / max(float(shaft['phase_total_minutes']), 1.0) floor = shaft['phase_origin_floor'] + (destination - shaft['phase_origin_floor']) * progress row.update(visual_floor=floor, passenger_count=len(shaft['boarded']), passenger_ids=copy.deepcopy(shaft['boarded']), waiting_count=sum(t['shaft_id'] == shaft['shaft_id'] and t['state'] == 'QUEUED' for t in saved['trips'])) rows.append(row) # WorldPreview sorts its report's trip copy for queue placement. trips = sorted(copy.deepcopy(saved['trips']), key=lambda t: (t['enqueued_at'], t['actor_id'])) return dict(current_minute=saved['current_minute'], shafts=rows, trips=trips) def close(actual, expected, label): if isinstance(expected, (float, int)) and not isinstance(expected, bool): require(type(actual) in (int, float) and math.isfinite(actual) and math.isclose(actual, expected, abs_tol=0.0001, rel_tol=0.000001), label) elif isinstance(expected, dict): require(isinstance(actual, dict) and actual.keys() == expected.keys(), label + ' keys') for key in expected: close(actual[key], expected[key], label + '/' + str(key)) elif isinstance(expected, list): require(isinstance(actual, list) and len(actual) == len(expected), label + ' length') for a, e in zip(actual, expected): close(a, e, label) else: require(type(actual) is type(expected) and actual == expected, label) class Transform: """Independent port of the shared cell/map transform; no report rect input.""" def __init__(self, runtime, report): self.bounds = runtime['world']['bounds'] b = self.bounds columns, rows = b['max_x'] - b['min_x'] + 1, b['max_y'] - b['min_y'] + 1 width, height = report['viewport'] require(width > 0 and height > 0 and report['zoom'] > 0, 'camera dimensions') fit = min(max(width - 152, 1) / (columns * 16), max(height - 140, 1) / (rows * 24)) pixel = max(math.floor(fit), 1) if fit >= 1 else max(fit, .25) self.dim = [max(math.floor(v * pixel * report['zoom'] + .5), minimum) for v, minimum in ((16, 4), (24, 6))] self.origin = [math.floor((extent - count * dim) / 2 + pan) for extent, count, dim, pan in zip((width, height), (columns, rows), self.dim, report['pan'])] def rect(self, cell): b = self.bounds return [self.origin[0] + (cell[0] - b['min_x']) * self.dim[0], self.origin[1] + (b['max_y'] - cell[1]) * self.dim[1], *self.dim] def center(self, cell): x, y, w, h = self.rect(cell) return [x + w / 2, y + h / 2] def room(self, cells): rects = [self.rect(c) for c in cells] x, y = min(r[0] for r in rects), min(r[1] for r in rects) return [x, y, max(r[0] + r[2] for r in rects) - x, max(r[1] + r[3] for r in rects) - y] def compare_traffic(snapshot, report): runtime = snapshot['systems']['runtime'] transform = Transform(runtime, report) rows = [] for saved in sorted(runtime['path']['cell_traffic'], key=lambda r: (r['cell'][1], r['cell'][0])): cell = saved['cell'] if cell[2] != report['layer']: continue x, y, w, h = transform.rect(cell) rows.append(dict(cell=cell, traffic={k: v for k, v in saved.items() if k != 'cell'}, rect=[x + 2, y + 2, w - 4, h - 4])) close(report['traffic'], rows, 'traffic and cell rect') def compare_queues(snapshot, report): runtime = snapshot['systems']['runtime'] transform = Transform(runtime, report) facilities = codec.indexed(runtime['facilities']['instances'], 'instance_id') requests = codec.indexed(runtime['facilities']['requests'], 'request_id') guests = codec.indexed(runtime['guests']['guests'], 'guest_id') groups = {} def add(key, cell): if cell[2] != report['layer']: return members = groups.setdefault(tuple(cell), []) if key not in members: members.append(key) for facility in facilities.values(): for rid in facility['queue_entries']: request = requests.get(rid) if request is None: continue for gid in request['guest_ids']: guest = guests.get(gid) if guest and guest['state'] == 'QUEUING': add('guest/' + gid, guest['current_cell']) elevators = elevator_report(snapshot) for trip in elevators['trips']: if trip['state'] == 'QUEUED': add(trip['actor_id'].replace('guest:', 'guest/', 1).replace('staff:', 'staff/', 1), trip['from']) lift_cells = {tuple(cell) for shaft in elevators['shafts'] for cell in shaft['stops']} walkable = {tuple(cell) for cell in runtime['world']['walkable_cells']} owners = {tuple(cell): row['entity_id'] for row in runtime['world']['occupancies'] for cell in row['cells']} reserved, positions, hidden, badges = set(), {}, {}, [] for cell in sorted(groups, key=lambda c: (c[1], c[0])): members = groups[cell] lane = [] if cell in lift_cells else [cell] for direction in (-1, 1): for distance in range(1, 9): target = (cell[0] + direction * distance, cell[1], cell[2]) facility = facilities.get(owners.get(target)) if target not in walkable or (facility and facility['definition_id'] not in ('guest_corridor', 'service_corridor')): break lane.append(target) index = 0 for target in lane: if index >= len(members): break if target in reserved: continue reserved.add(target) positions[members[index]] = transform.center(target) index += 1 hidden.update({member: True for member in members[index:]}) overflow = len(members) - index rect = transform.rect(cell) badges.append(dict(at=[rect[0] + (-80 if cell in lift_cells else 0), rect[1] - 16], text=f'待ち{len(members)}人' + (f'（他{overflow}人）' if overflow else ''), cell=list(cell), members=members, visible_count=index, overflow=overflow)) for key, expected in [('queue_positions', positions), ('queue_hidden', hidden), ('landing_badges', badges)]: close(report[key], expected, key) def compare_room_services(snapshot, report, root=ROOT): """Check authoritative service occupants and reasons; rest selection remains pending. Extra rest/rescue occupants are checked separately below, never silently counted as service users. This does not claim the missing path-cost proof. """ runtime = snapshot['systems']['runtime'] states = enum_names(root, 'scripts/facility/facility_instance.gd', 'State') operations = enum_names(root, 'scripts/facility/facility_instance.gd', 'OperationState') rooms = [r for r in runtime['facilities']['instances'] if r['origin'][2] == report['layer']] treated, carried = {}, {} for task in runtime['staff']['tasks']: if task['task_type'] != 'rescue_transport' or task['state'] in ('COMPLETED', 'CANCELLED', 'FAILED') or not task['assigned_staff_id']: continue if task['state'] == 'WORKING': treated[task['target_entity_id']] = task['rescue_facility_id'] else: carried[task['target_entity_id']] = task['assigned_staff_id'] close(report['treated_patients'], treated, 'rescue treatment') close(report['carried_patients'], carried, 'rescue carrier') staff = codec.indexed(runtime['staff']['staff'], 'staff_id') all_rooms = codec.indexed(runtime['facilities']['instances'], 'instance_id') for sid, fid in report['resting_staff'].items(): require(sid in staff and fid in all_rooms, 'rest reference') member, room = staff[sid], all_rooms[fid] require(member['state'] == 'OFF_DUTY' and member['health'] > 0 and not member['current_task_id'], 'resting staff eligibility') require(room['definition_id'] in ('staff_waiting_room', 'staff_dormitory') and room['state'] in ('OPEN', 'DEGRADED') and not room['suspension_reason'], 'operating rest room') require(set(report['rooms']) == {r['instance_id'] for r in rooms}, 'room identity') for room in rooms: observed = report['rooms'][room['instance_id']] users = sorted({gid for service in room['active_services'] for gid in service['guest_ids']}) users += [sid for sid in staff if report['resting_staff'].get(sid) == room['instance_id']] users += [task['target_entity_id'] for task in runtime['staff']['tasks'] if task['task_type'] == 'rescue_transport' and task['state'] == 'WORKING' and task['assigned_staff_id'] and task['rescue_facility_id'] == room['instance_id']] require(observed['people'] == users, 'service/rest/rescue occupant list') require(observed['count'] == len(observed['people']), 'room count') operation = 'SUSPENDED' if room['suspension_reason'] else ('OPERATING' if room['state'] in ('OPEN', 'DEGRADED') else 'INACTIVE') expected = dict(state=states.index(room['state']), state_name=room['state'], reasons=room['state_reasons'], operation=operations.index(operation), limit=room['customer_capacity'], staff=room['active_staff_ids'], suspension=room['suspension_reason']) for key, value in expected.items(): close(observed[key], value, 'room/' + key) cells = next((r['cells'] for r in runtime['world']['occupancies'] if r['entity_id'] == room['instance_id']), []) if cells: close(observed['rect'], Transform(runtime, report).room(cells), 'room rect') require(observed['definition_id'] == room['definition_id'], 'room definition') def comparison_records(directory): records = {} for line in (directory / 'manifest.jsonl').read_text().splitlines(): row = json.loads(line) if 'object' not in row: continue value = read_object(directory, row['object']) if row['label'] == 'event': continue require(row['label'] not in records, 'duplicate experiment record') records[row['label']] = value return records def compare_goods(snapshot, report, root=ROOT): runtime = snapshot['systems']['runtime'] states = enum_names(root, 'scripts/inventory/logistics_request.gd', 'State') candidates = codec.indexed(json.loads((root / 'data/characters/mvp_staff_candidates.json').read_text())['entries'], 'id') staff = codec.indexed(runtime['staff']['staff'], 'staff_id') facilities = runtime['facilities']['instances'] def owner(inventory): return next((f['instance_id'] for f in facilities if f['local_inventory_id'] == inventory), '') rows, active, recent = [], [], [] for request in sorted(runtime['logistics_requests'], key=lambda r: r['request_id']): require(request['state'] in states, 'logistics enum') task = next((t for t in runtime['staff']['tasks'] if t['target_entity_id'] == request['request_id'] and t['task_type'] in ('supply_pickup', 'supply_delivery', 'supply_return') and t['state'] not in ('COMPLETED', 'CANCELLED', 'FAILED')), None) carrier = request['assigned_carrier_id'] row = dict(source_facility_id=owner(request['source_inventory_id']), destination_facility_id=owner(request['destination_inventory_id']), carrier_name=candidates.get(carrier, {}).get('display_name', carrier) if carrier in staff else '', reason_codes=[request['state_reason']] if request['state_reason'] else [], request_id=request['request_id'], state=request['state'], reason=request['state_reason'], carrier_id=carrier, source_inventory_id=request['source_inventory_id'], destination_inventory_id=request['destination_inventory_id'], item_requirements=request['item_requirements'], weight=request['total_weight'], retry_count=request['retry_count'], task_id=task['task_id'] if task else '') rows.append(row) if request['state'] in ('COMPLETED', 'CANCELLED'): recent.append((request['state_changed_at'], request['request_id'], row)) else: active.append(row) recent.sort(key=lambda r: (r[0], r[1]), reverse=True) close(report['goods'], dict(requests=rows, deliveries=active + [r[2] for r in recent[:10]]), 'goods') def compare_cabin_geometry(snapshot, report): runtime = snapshot['systems']['runtime'] transform = Transform(runtime, report) expected = [] p = transform.dim[0] / 16 for shaft in elevator_report(snapshot)['shafts']: if not shaft['stops'] or shaft['stops'][0][2] != report['layer']: continue floor = shaft['visual_floor'] track = report['elevator_tracks'][shaft['shaft_id']] close(track['floor'], floor, 'authoritative elevator track endpoint') require(track['at'] <= report['animation_time'], 'elevator track time') if not report['paused']: # The prior-frame origin still needs a chained visual-history proof. t = max(0, min(1, (report['animation_time'] - track['at']) / .42)) floor = track['from'] + (floor - track['from']) * t x, y, w, h = transform.rect(shaft['stops'][0]) baseline = y + h - (floor - shaft['stops'][0][1]) * transform.dim[1] - 3 * p def rounded(v): return math.copysign(math.floor(abs(v) + .5), v) rect = [rounded(x + 4*p), rounded(baseline - 19*p), max(shaft.get('width', 2), 1)*w - 8*p, 19*p] if rect[0] >= report['viewport'][0] or rect[1] >= report['viewport'][1] or rect[0]+rect[2] <= 0 or rect[1]+rect[3] <= 0: continue count, capacity = len(shaft['boarded']), shaft['capacity'] badge = f'{count}/{capacity}人' if capacity > 0 and count >= capacity: badge = '満員 ' + badge if not shaft['operational']: badge = '停止 ' + badge expected.append(dict(shaft_id=shaft['shaft_id'], rect=rect, pixel=p, operational=shaft['operational'], doors_open=shaft['operational'] and shaft['phase'] in ('IDLE', 'BOARDING', 'ALIGHTING'), badge_at=[rect[0], rect[1]-14], badge=badge)) require(len(report['cabins']) == len(expected), 'cabin count') for row, wanted in zip(report['cabins'], expected): for key, value in wanted.items(): close(row[key], value, 'cabin/' + key) def compare_layouts(a, b): require(a['declaration'] == b['declaration'], 'identical predeclared inputs') declaration = a['declaration'] require(declaration['metric'] == 'sum_destination_arrival_minutes_entire_cohort', 'predeclared metric') require(a['initial'] == b['initial'], 'identical seed budget clock definitions and initial state') arrivals = {r['id']: r['offset'] for r in declaration['arrivals']} require(len(arrivals) == len(declaration['arrivals']) and arrivals, 'whole nonempty cohort') builds = [] metrics = [] for records in (a, b): commands = [records[k] for k in sorted(records) if k.startswith('build-')] require(all(r['result']['code'] == 'OK' for r in commands), 'paid construction succeeded') builds.append([r['payload'] for r in commands]) require({k.removeprefix('arrival-') for k in records if k.startswith('arrival-')} == set(arrivals), 'arrival cohort identity') for gid, offset in arrivals.items(): row = records['arrival-' + gid] require(row['offset'] == offset and row['result']['code'] == 'OK', 'actual accepted arrival') require(row['payload'] == a['arrival-' + gid]['payload'], 'same actual arrival command') completed = {} first_minute = records['tick-000']['minute'] for offset in range(declaration['window'] + 1): tick = records['tick-%03d' % offset] require(tick['minute'] == first_minute + offset, 'contiguous actual clock') for guest in tick['guests']['guests']: gid = guest['guest_id'] require(gid in arrivals and offset >= arrivals[gid], 'actual cohort membership') if guest['current_cell'] == declaration['to']: completed.setdefault(gid, offset - arrivals[gid]) require(set(completed) == set(arrivals), 'no unfinished or omitted journeys') metrics.append(sum(completed.values())) require(builds[1] == builds[0] + [declaration['improvement']], 'only predeclared layout improvement') require(b['tick-000']['minute'] == a['tick-000']['minute'], 'same observation start') require(metrics[1] < metrics[0], 'strict whole-cohort improvement') return dict(metric=declaration['metric'], A=metrics[0], B=metrics[1]) def read_object(directory, digest): require(isinstance(digest, str) and re.fullmatch('[0-9a-f]{64}', digest), 'invalid object digest') path = directory / 'objects' / (digest + '.json') require(not path.is_symlink() and sha(path) == digest, 'whole typed object hash') typed = json.loads(path.read_bytes()) require(typed.get('type') == 'Dictionary', 'whole dictionary required') return decode(typed) def inspect_stream(directory, root=ROOT): labels = set() checked = [] with (directory / 'manifest.jsonl').open() as stream: for line in stream: row = json.loads(line) if 'object' in row: read_object(directory, row['object']) continue require(row['label'] not in labels, 'duplicate streamed label') labels.add(row['label']) require(type(row['pid']) is int and row['pid'] > 0, 'actual diagnostic PID required') state = read_object(directory, row['state']) report = read_object(directory, row['report']) fields = compare_actor_rows(state, report, root) require(report['elevators'] == elevator_report(state), 'typed elevator report mismatch') compare_room_services(state, report, root) compare_queues(state, report) compare_traffic(state, report) compare_goods(state, report, root) compare_cabin_geometry(state, report) checked.append(dict(label=row['label'], fields=fields + ['elevators', 'room_service_fields', 'queues', 'traffic', 'goods', 'cabin_geometry'])) require(checked, 'empty diagnostic stream') return checked def condition_matrix(): return [dict(condition=name, status='blocked', reason=BLOCKERS[name]) for name in CONDITIONS] def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--diagnostics', type=Path) parser.add_argument('--layout-a', type=Path) parser.add_argument('--layout-b', type=Path) args = parser.parse_args(argv) # Registration hashes the immutable JSON after trimming outer whitespace. require(hashlib.sha256((ROOT / SPEC).read_text().strip().encode()).hexdigest() == SPEC_SHA, 'immutable spec mismatch') checked = inspect_stream(args.diagnostics.resolve()) if args.diagnostics else [] require(bool(args.layout_a) == bool(args.layout_b), 'both layout streams required') comparison = compare_layouts(comparison_records(args.layout_a), comparison_records(args.layout_b)) if args.layout_a else None print(json.dumps(dict(task_id='P23-03', status='blocked', semantic_diagnostics=checked, layout_comparison=comparison, condition_matrix=condition_matrix(), root_evidence='pending'), ensure_ascii=False)) return 2 if __name__ == '__main__': raise SystemExit(main())