TheTree_mock/tools/check_p23_life_observation_evidence.py

#!/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 contextlib
import tempfile
import copy
import hashlib
import importlib.util
import json
import math
import os
from pathlib import Path
import re
import sys
import struct
import subprocess
import zlib

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',
)
CATEGORIES = ('room', 'queue', 'EV', 'staff', 'goods', 'business')
CONDITION_CHECKS = dict(zip(CONDITIONS, ('room_users', 'queue_overflow', 'elevator_passengers',
                                      'staff_jobs', 'carried_goods', 'route_improvement', 'distinct_pid_resume')))
# Required names, not verdicts. BLOCKERS is computed from actual execution below.
REQUIRED_CHECKS = {condition: {category: {'semantic_' + category, 'pixels_' + category,
                                       CONDITION_CHECKS[condition], 'os_guards', 'resume_recalculation'}
                              for category in CATEGORIES} for condition in CONDITIONS}


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 check_setup_budget_source(source):
    """Cheap preflight for the fixture's shared spend plan, before OS replay."""
    prepare = source.split('func prepare() -> void:', 1)[1].split('\nfunc ', 1)[0]
    categories = {
        'construction': ('budget_builds := builds.duplicate(true)',
                         'budget_builds.append([improvement[',
                         'for row: Array in budget_builds:',
                         'needed[key] += int(definition.construction_cost.get(key, 0))'),
        'signing': ('for assignment: Array in assignments:',
                    'runtime.mvp_content_catalog.get_recruitment_entry(assignment[0])',
                    'signing_total += signing_cost', 'needed[&"gold"] += signing_total'),
        'purchase': ('SupplyCatalog.get_item(SupplyCatalog.RAW_FOOD)',
                     'PurchaseQuoteV2.make(SupplyCatalog.RAW_FOOD, purchase_quantity,',
                     'PurchaseQuoteV2.eligible(runtime)', 'purchase_quote[&"effective_total"]',
                     'needed[&"gold"] += purchase_total'),
    }
    budget_end = prepare.index('\n\tvar state:')
    for category, expressions in categories.items():
        require(all(expression in prepare[:budget_end] for expression in expressions),
                'setup budget missing spend category: ' + category)
    require(prepare.count('for assignment: Array in assignments:') == 2,
            'budget and hire must share assignments')
    require('&"quantity": purchase_quantity}' in prepare,
            'budget and purchase must share quantity')
    require('deltas[key] = needed[key] - state.get_resource(key)' in prepare,
            'setup budget deltas must cover needed')
    return set(categories)


def sha(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def png_pixels(data):
    """Decode noninterlaced 8-bit RGB/RGBA PNG, validating the whole stream."""
    require(data[:8] == b'\x89PNG\r\n\x1a\n', 'PNG signature')
    pos, header, payload, ended, idat_closed = 8, None, bytearray(), False, False
    while pos < len(data):
        require(pos + 12 <= len(data), 'PNG truncated chunk')
        size, kind = struct.unpack('>I4s', data[pos:pos + 8])
        require(pos + size + 12 <= len(data), 'PNG truncated payload')
        body = data[pos + 8:pos + 8 + size]
        crc = struct.unpack('>I', data[pos + 8 + size:pos + 12 + size])[0]
        require(zlib.crc32(kind + body) == crc, 'PNG CRC mismatch')
        pos += size + 12
        require(header is not None or kind == b'IHDR', 'PNG IHDR must be first')
        if kind == b'IHDR':
            require(header is None and size == 13, 'PNG duplicate/invalid IHDR')
            w, h, depth, color, compression, filtering, interlace = struct.unpack('>IIBBBBB', body)
            require(w > 0 and h > 0 and depth == 8 and color in (2, 6)
                    and (compression, filtering, interlace) == (0, 0, 0), 'PNG unsupported IHDR')
            header = (w, h, 3 if color == 2 else 4)
        elif kind == b'IDAT':
            require(not idat_closed, 'PNG nonconsecutive IDAT')
            payload.extend(body)
        elif kind == b'IEND':
            require(size == 0 and payload and pos == len(data), 'PNG invalid IEND')
            ended = True
            break
        else:
            require(kind[0] & 32 or kind == b'PLTE', 'PNG unknown critical chunk')
            if payload:
                idat_closed = True
    require(ended and header is not None, 'PNG missing IEND')
    w, h, channels = header
    stride = w * channels
    expected = h * (stride + 1)
    try:
        decoder = zlib.decompressobj()
        raw = decoder.decompress(bytes(payload), expected + 1)
        require(decoder.eof and not decoder.unused_data and not decoder.unconsumed_tail
                and len(raw) == expected, 'PNG zlib length/end mismatch')
    except zlib.error as exc:
        raise ValueError('PNG invalid zlib') from exc
    rows, prior = [], bytearray(stride)
    for y in range(h):
        start = y * (stride + 1)
        mode = raw[start]
        require(mode in range(5), 'PNG unknown filter')
        row = bytearray(raw[start + 1:start + 1 + stride])
        for x in range(stride):
            left = row[x - channels] if x >= channels else 0
            up = prior[x]
            corner = prior[x - channels] if x >= channels else 0
            if mode == 1:
                predictor = left
            elif mode == 2:
                predictor = up
            elif mode == 3:
                predictor = (left + up) // 2
            elif mode == 4:
                p = left + up - corner
                distances = (abs(p - left), abs(p - up), abs(p - corner))
                predictor = (left, up, corner)[distances.index(min(distances))]
            else:
                predictor = 0
            row[x] = (row[x] + predictor) & 255
        rows.append([tuple(row[x:x + channels]) for x in range(0, stride, channels)])
        prior = row
    return w, h, rows


def image_kind(data):
    if data.startswith(b'\x89PNG\r\n\x1a\n'):
        w, h, _ = png_pixels(data)
        return 'png', w, h
    if data.startswith(b'\xff\xd8'):
        helper = module('p23_p08_jpeg', ROOT / 'tools/check_p08_normal_placement_evidence.py')
        w, h = helper.jpeg_dimensions(data)
        return 'jpeg', w, h
    raise ValueError('unknown screenshot byte magic')


def inspect_conversion(rec, cua_receipt, converter_sha256=None, resolve_path=Path):
    """CUA receipt and converter pin are independent of conversion assertions."""
    raw, png = resolve_path(rec['raw_path']), resolve_path(rec['png_path'])
    require(raw.resolve() != png.resolve() and not raw.is_symlink() and not png.is_symlink(),
            'conversion must preserve a separate raw file')
    require(rec['raw_sha256'] == sha(raw) == cua_receipt['raw_sha256']
            and rec['tool_call_id'] == cua_receipt['tool_call_id']
            and bool(rec['tool_call_id']), 'CUA raw hash/call mismatch')
    require(rec['raw_kind'] == 'jpeg', 'conversion raw kind')
    kind, w, h = image_kind(raw.read_bytes())
    require(kind == 'jpeg', 'conversion requires JPEG magic')
    require(rec['converter'] == '/usr/bin/sips', 'conversion executable')
    pin = converter_sha256 if converter_sha256 is not None else sha('/usr/bin/sips')
    require(rec['converter_sha256'] == pin, 'converter SHA mismatch')
    require(rec['argv'] == ['/usr/bin/sips', '-s', 'format', 'png', rec['raw_path'], '--out', rec['png_path']],
            'converter argv mismatch')
    require(type(rec['exit_code']) is int and rec['exit_code'] == 0, 'converter exit code')
    for key in ('stdout_sha256', 'stderr_sha256'):
        require(isinstance(rec[key], str) and re.fullmatch('[0-9a-f]{64}', rec[key]),
                'invalid converter output digest')
    require(sha(png) == rec['png_sha256'], 'converted PNG SHA mismatch')
    pw, ph, rows = png_pixels(png.read_bytes())
    require((w, h) == (pw, ph) == (rec['width'], rec['height']), 'conversion dimensions mismatch')
    return pw, ph, rows


def screenshot_pixels(path, conversion=None, cua_receipt=None, converter_sha256=None):
    data = Path(path).read_bytes()
    kind, _, _ = image_kind(data)
    if kind == 'jpeg':
        require(conversion is not None, 'JPEG screenshot requires conversion record')
        require(Path(conversion['raw_path']).resolve() == Path(path).resolve(), 'conversion raw path mismatch')
        require(cua_receipt is not None, 'missing CUA receipt')
        return inspect_conversion(conversion, cua_receipt, converter_sha256)
    return png_pixels(data)


def affine_point(matrix, point):
    require(len(matrix) == 6 and all(type(v) in (int, float) and math.isfinite(v) for v in matrix), 'invalid affine')
    a, b, c, d, x, y = matrix
    require(a*d-b*c != 0, 'singular affine')
    return [a*point[0]+c*point[1]+x, b*point[0]+d*point[1]+y]


def affine_rect(matrix, rect):
    x, y, w, h = rect
    points = [affine_point(matrix, p) for p in ([x,y], [x+w,y], [x,y+h], [x+w,y+h])]
    lo = [min(p[i] for p in points) for i in (0, 1)]
    return lo + [max(p[i] for p in points)-lo[i] for i in (0, 1)]


def screenshot_transform(calibration, request, image_size, clicks):
    """Content -> window screenshot pixels, cross-checked against actual clicks.

    This is independent of the renderer's content-space screen_rect. In
    particular, window_position is a desktop coordinate, not an image origin.
    No titlebar height, scale, or image extent is assumed by the checker.
    """
    scale = calibration['backing_scale']
    stretch = calibration['canvas_stretch']
    titlebar = calibration['titlebar_pixels']
    require(all(type(v) in (int, float) and math.isfinite(v) for v in
                (scale, stretch, titlebar)) and scale > 0 and stretch > 0 and titlebar >= 0,
            'invalid screenshot calibration')
    close(calibration['window_size'], request['window_size'], 'calibrated window size')
    close(calibration['window_position'], request['window_position'], 'calibrated window position')
    close(calibration['screenshot_pixels'], list(image_size), 'calibrated image size')
    close([request['screen_scale'], request['content_scale_factor']], [scale, stretch],
          'calibrated engine scale')
    close([request['window_size'][0]*scale, request['window_size'][1]*scale+titlebar],
          list(image_size), 'calibrated content extent')
    matrix = [scale, 0, 0, scale, 0, titlebar]
    matched = 0
    for receipt in clicks:
        if receipt.get('target_pid') != request['target_pid']:
            continue
        for event in receipt['events']:
            if event.get('action') != 'click':
                continue
            close(event['calibration'], calibration, 'click calibration')
            point = event['point']
            close(affine_point(point['viewport_to_screen'], point['viewport']),
                  point['screen'], 'click content affine')
            close(affine_point(matrix, point['screen']), event['coordinates'],
                  'click screenshot offset')
            matched += 1
    require(matched > 0, 'missing independent click calibration')
    return matrix


def rect_intersection(a, b):
    require(len(a) == len(b) == 4 and all(type(v) in (int, float) and math.isfinite(v)
            for v in a+b) and min(a[2:]+b[2:]) >= 0, 'invalid clipping rectangle')
    x, y = max(a[0], b[0]), max(a[1], b[1])
    return [x, y, max(0, min(a[0]+a[2], b[0]+b[2])-x),
            max(0, min(a[1]+a[3], b[1]+b[3])-y)]


def rectangle_eligibility(rect, viewport):
    clipped = rect_intersection(list(rect), [0, 0, *viewport])
    if not clipped[2] or not clipped[3]:
        return 'outside', clipped
    return ('visible' if clipped == list(rect) else 'clipped'), clipped


def paired_crop_noise(before, after, rect):
    """Measured temporal differences, never an invented JPEG RGB tolerance.

    A stable pair alone does not establish class or state identity. The result
    is diagnostic data only and must not be promoted to pixels_<category>.
    """
    require(before[:2] == after[:2], 'paired image dimensions')
    x, y, w, h = rect_intersection(list(rect), [0, 0, *before[:2]])
    errors = [max(abs(a-b) for a, b in zip(before[2][py][px][:3], after[2][py][px][:3]))
              for py in range(math.ceil(y), math.floor(y+h))
              for px in range(math.ceil(x), math.floor(x+w))]
    require(errors, 'empty noise crop')
    return dict(samples=len(errors), max_channel_delta=max(errors),
                mean_channel_delta=sum(errors)/len(errors), unchanged=sum(v == 0 for v in errors))


def drawing_rectangle_rows(report):
    """Derive identities and local positions; do not read audit verdicts."""
    if any('anchor' in b for key in ('badges', 'landing_badges', 'actor_aggregations')
           for b in report.get(key, [])):
        validate_observation_labels(report)
    expected = []
    tower = 'scripts/ui/pixel_tower_art.gd'
    people = 'scripts/ui/pixel_people.gd'
    world = 'scenes/main/world_preview.gd'
    def actor(a, category=None):
        expected.append((category or ('staff' if a['staff'] else 'guest'), a['key'],
                         a['at'] + [16*a['scale'], 18*a['scale']], people, None))
    def badge(category, identifier, b):
        expected.append((category, identifier, badge_rectangle(b['at'], b['text']), world, b['text']))
    for fid, room in report['rooms'].items():
        if 'rect' in room:
            expected.append(('room', fid, room['rect'], tower, None))
    for i, b in enumerate(report['landing_badges']):
        badge('queue', str(i), b)
    for cabin in report['cabins']:
        expected.append(('EV', cabin['shaft_id'], cabin['rect'], tower, None))
        for a in cabin['passengers']:
            actor(a)
    for a in report['actor_drawings']:
        actor(a)
        if a['staff'] and a['key'].removeprefix('staff/') in report['resting_staff']:
            actor(a, 'rest')
    for i, b in enumerate(report.get('actor_aggregations', [])):
        badge('aggregation', str(i), b)
    for i, b in enumerate(report['badges']):
        badge('goods' if b['kind'] == 'supply' else 'business', str(i), b)
    for row in report.get('traffic', []):
        if row['traffic']['current_density'] > 0 or row['traffic']['temporary_block']:
            expected.append(('traffic', ','.join(map(str, row['cell'])), row['rect'], world, None))
    return expected


PIXEL_PEOPLE_CORE_SHA256 = 'a171703083aa2d00f8be2b63e1daef9c41717c8fe59d8056878bc4b7df1e0b0c'


def actor_identity_core(actor):
    """Grid identity bounds from the pinned PixelPeople pose, including bob.

    Standard-room breathing extends the upper edge, not the lower edge.
    Arms, legs, props and the sleep symbol are outside the identity core.
    """
    key, at, scale = actor['key'], actor['at'], actor['scale']
    require(len(at) == 2 and all(type(v) in (int, float) and math.isfinite(v) for v in at)
            and type(scale) in (int, float) and math.isfinite(scale) and scale > 0,
            'invalid actor footprint: ' + key)
    require(isinstance(actor.get('mode'), str) and type(actor.get('frame')) is int
            and type(actor.get('flip')) is bool, 'missing actor pose: ' + key)
    mode, phase = actor['mode'], actor['frame'] % 8
    if mode in ('sleep', 'collapsed', 'standard_room_sleep'):
        breathing = -1 if mode == 'standard_room_sleep' and phase in (2, 3, 4) else 0
        x, y, w, h = 2, 11 + breathing, 13, 5 - breathing
    else:
        bob = -1 if mode in ('walk', 'carry_walk') and phase in (2, 6) else 0
        x, y, w, h = 6, 1 + bob, 6, 12
    if actor['flip']:
        x = 16 - x - w
    return [at[0] + x * scale, at[1] + y * scale, w * scale, h * scale]


def validate_actor_nonoverlap(report, root=ROOT, transit_keys=()):
    """Recompute identity cores from drawing inputs, including cabin passengers.

    Do not consume audit eligibility or boolean judgments: neither an off-screen
    label nor a claimed pass exempts a submitted actor from this invariant.
    Transparent margins may overlap. No asserted transit flag grants exemption.
    """
    require(hashlib.sha256((root / 'scripts/ui/pixel_people.gd').read_bytes()).hexdigest()
            == PIXEL_PEOPLE_CORE_SHA256, 'PixelPeople core source SHA256')
    actors = list(report['actor_drawings']) + [
        actor for cabin in report['cabins'] for actor in cabin['passengers']]
    rects = []
    keys = set()
    for actor in actors:
        key = actor['key']
        require(key not in keys, 'duplicate actor drawing: ' + key)
        keys.add(key)
        rect = actor_identity_core(actor)
        if key in transit_keys:
            continue
        for other_key, other in rects:
            intersects = (rect[0] < other[0] + other[2] and other[0] < rect[0] + rect[2]
                          and rect[1] < other[1] + other[3] and other[1] < rect[1] + rect[3])
            require(not intersects, 'actor-actor core overlap: ' + other_key + ' / ' + key)
        rects.append((key, rect))


def validate_drawing_rows(report, audit, root=ROOT, state=None, rest_audit=None):
    # Without raw inputs this is deliberately conservative: NO transit exemption.
    # With inputs, use the same state-derived classification and fit proof as the
    # stream checker, never a drawing's region/in_transit assertion.
    if state is None:
        validate_actor_nonoverlap(report, root)
    else:
        require(isinstance(state, dict) and isinstance(state.get('systems', {}).get('runtime'), dict),
                'malformed raw state context')
        if report['resting_staff']:
            require(rest_audit is not None, 'missing rest authority for drawing geometry')
        if rest_audit is not None:
            require(isinstance(rest_audit, dict) and isinstance(rest_audit.get('staff'), list),
                    'malformed rest authority')
            compare_rest(state, report, rest_audit, root)
        compare_cabin_geometry(state, report)
        compare_actor_geometry(state, report)
    expected = drawing_rectangle_rows(report)
    require(len(audit['rows']) == len(expected), 'offline rectangle coverage')
    for row, (category, identifier, rect, source, text) in zip(audit['rows'], expected):
        require((row['category'], row['id'], row['style_source']) == (category, identifier, source),
                'offline rectangle identity')
        rect = list(rect)
        if text is not None:
            require(row['text'] == text and type(row['font_width']) in (int, float)
                    and math.isfinite(row['font_width']) and row['font_width'] > 0, 'offline font/text')
            require(row['font_width'] <= rect[2] - 6, 'badge text exceeds reproducible rectangle')
        close(row['rect'], rect, 'offline report drawing rectangle')
        colors = list(dict.fromkeys(re.findall(r'Color\("([0-9a-fA-F]{6})"', (root/source).read_text())))
        require(row['style_colors'] == colors and colors, 'offline source palette provenance')


def encode_png(rows):
    """Lossless review crop from decoded pixels, without an imaging dependency."""
    require(rows and rows[0], 'empty crop')
    width, channels = len(rows[0]), len(rows[0][0])
    require(channels in (3, 4), 'crop channels')
    def chunk(kind, body):
        return (struct.pack('>I', len(body)) + kind + body
                + struct.pack('>I', zlib.crc32(kind + body)))
    raw = b''.join(b'\0' + bytes(v for pixel in row for v in pixel) for row in rows)
    return (b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', struct.pack(
        '>IIBBBBB', width, len(rows), 8, 2 if channels == 3 else 6, 0, 0, 0))
        + chunk(b'IDAT', zlib.compress(raw)) + chunk(b'IEND', b''))


def frame_rows(report, audit, request, receipt, images, clicks, root=ROOT, state=None, rest_audit=None):
    """One mechanical path for frozen diagnostics and actual OS frames.

    Errors are retained as failures, including incomplete frozen audits. Every
    supplied row is accounted for; no palette value is an image oracle.
    """
    errors, rows = [], []
    transit_authority = ('raw state, track, time and pause' if state is not None
                         else 'absent: fail closed, no transit exemptions')
    try:
        validate_drawing_rows(report, audit, root, state, rest_audit)
    except (ValueError, KeyError, TypeError) as exc:
        errors.append(str(exc))
    calibration = screenshot_transform(receipt['calibration'], request, images[0][:2], clicks)
    for image in images:
        require(image[:2] == images[0][:2], 'frame image dimensions')
    for point in ([0, 0], [1, 0], [0, 1]):
        close(affine_point(request['viewport_to_screen'], affine_point(request['local_to_viewport'], point)),
              affine_point(request['local_to_screen'], point), 'frame affine composition')
    for index, row in enumerate(audit['rows']):
        failures = list(errors)
        close(row['viewport_rect'], affine_rect(request['local_to_viewport'], row['rect']),
              'frame viewport rectangle')
        close(row['screen_rect'], affine_rect(request['local_to_screen'], row['rect']),
              'frame content rectangle')
        status, clipped = rectangle_eligibility(row['rect'], report['viewport'])
        screenshot_rect = affine_rect(calibration, affine_rect(request['local_to_screen'], clipped))
        room = report['rooms'].get(row['id']) if row['category'] == 'room' else None
        if row['category'] in ('business', 'goods'):
            room = report['rooms'].get(report['badges'][int(row['id'])]['facility'])
        operation, state = None, None
        if room is not None:
            states = enum_names(root, 'scripts/facility/facility_instance.gd', 'State')
            operations = enum_names(root, 'scripts/facility/facility_instance.gd', 'OperationState')
            require(type(room['state']) is int and 0 <= room['state'] < len(states), 'frame room enum')
            state = states[room['state']]
            require(state == room['state_name'], 'frame room state name')
            require(type(room['operation']) is int and 0 <= room['operation'] < len(operations), 'frame operation enum')
            operation = operations[room['operation']]
        actors = list(report['actor_drawings']) + [a for c in report['cabins'] for a in c['passengers']]
        actor = next((a for a in actors if a['key'] == row['id']), None)
        if actor is not None:
            actor_states = enum_names(root, 'scripts/staff/staff_runtime.gd' if actor['staff']
                                     else 'scripts/guest/guest_runtime.gd', 'State')
            require(type(actor['state']) is int and 0 <= actor['state'] < len(actor_states),
                    'frame actor state enum')
            state = actor_states[actor['state']]
        if row['category'] == 'queue':
            badge = report['landing_badges'][int(row['id'])]
            state = {k: badge[k] for k in ('members', 'visible_count', 'overflow')}
        if row['category'] == 'aggregation':
            badge = report['actor_aggregations'][int(row['id'])]
            state = {k: badge[k] for k in ('members', 'drawn_keys')}
        if row['category'] == 'EV':
            cabin = next(c for c in report['cabins'] if c['shaft_id'] == row['id'])
            state = {k: cabin[k] for k in ('operational', 'doors_open')}
        if row['category'] == 'traffic':
            traffic = next(r['traffic'] for r in report.get('traffic', [])
                           if ','.join(map(str, r['cell'])) == row['id'])
            state = {k: traffic[k] for k in ('current_density', 'temporary_block')}
        result = dict(row_index=index, category=row['category'], id=row['id'], layer=report['layer'],
                      state=state, state_name=state if isinstance(state, str) else None, operation=operation,
                      dimmed=row['category'] == 'room' and operation != 'OPERATING',
                      text=row.get('text'), local_rect=row['rect'], screenshot_rect=screenshot_rect,
                      style_source=row['style_source'], style_source_sha256=sha(root/row['style_source']),
                      style_colors=row['style_colors'], font_width=row.get('font_width'),
                      eligibility=status, reason={'visible': 'whole rectangle in viewport',
                          'clipped': 'rectangle intersects viewport', 'outside': 'no viewport intersection'}[status],
                      crops=[])
        if status != 'outside':
            for number, (width, height, pixels) in enumerate(images):
                x, y, w, h = rect_intersection(screenshot_rect, [0, 0, width, height])
                crop = [line[math.ceil(x):math.floor(x+w)]
                        for line in pixels[math.ceil(y):math.floor(y+h)]]
                if not crop or not crop[0]:
                    failures.append('eligible crop outside image')
                    continue
                if len({pixel for line in crop for pixel in line}) == 1:
                    failures.append('blank crop')
                data = encode_png(crop)
                result['crops'].append(dict(image_index=number, sha256=hashlib.sha256(data).hexdigest(),
                                            png_bytes=data))
        result['mechanical_pass'] = not failures
        result['failures'] = failures
        rows.append(result)
    return dict(rows=rows, errors=errors, mechanical_pass=not errors and all(r['mechanical_pass'] for r in rows),
                expected_row_count=len(drawing_rectangle_rows(report)),
                transit_authority=transit_authority)


def write_frame_review(directory, frame, provenance):
    directory = Path(directory)
    directory.mkdir(parents=True, exist_ok=True)
    manifest = copy.deepcopy(frame)
    manifest['images'] = provenance
    for row in manifest['rows']:
        for crop in row['crops']:
            name = 'row-%04d-image-%d.png' % (row['row_index'], crop['image_index'])
            (directory/name).write_bytes(crop.pop('png_bytes'))
            crop['path'] = name
    data = (json.dumps(manifest, sort_keys=True, ensure_ascii=False, indent=2) + '\n').encode()
    (directory/'manifest.json').write_bytes(data)
    digest = hashlib.sha256(data).hexdigest()
    (directory/'manifest.sha256').write_text(digest + '\n')
    return manifest, digest


def reviewed_pixel_categories(manifest, digest, verdict_path, *, require_judgments=False):
    """The worker never writes verdicts; absent independent review earns nothing."""
    if not Path(verdict_path).is_file():
        return set()
    verdict = json.loads(Path(verdict_path).read_bytes())
    require(verdict['manifest_sha256'] == digest, 'review manifest SHA mismatch')
    require(verdict['reviewer'] == 'independent-opus-direct-image', 'independent image reviewer required')
    if require_judgments:
        require(isinstance(verdict.get('judgments'), list), 'explicit indexed judgments required')
        verdict = dict(verdict, rows=verdict['judgments'])
    eligible = [r for r in manifest['rows'] if r['eligibility'] != 'outside']
    require([v['row_index'] for v in verdict['rows']] == [r['row_index'] for r in eligible],
            'review row coverage')
    for row, decision in zip(eligible, verdict['rows']):
        require(all(decision[k] is True for k in ('category', 'state', 'position')), 'review row rejected')
    if not manifest['mechanical_pass']:
        return set()
    return {r['category'] for r in eligible if r['mechanical_pass']}


def mirrored_inventory(directory):
    """Read every indexed byte; preserve original receipt paths as authority."""
    directory = Path(directory).resolve()
    index = json.loads((directory/'MIRROR_INDEX.json').read_bytes())
    require(Path(index['mirror']).resolve() == directory, 'mirror root mismatch')
    paths, mapping = set(), []
    source = Path(index['source'])
    for row in index['files']:
        name = Path(row['path'])
        require(not name.is_absolute() and '..' not in name.parts and str(name) not in paths,
                'invalid/duplicate mirror path')
        paths.add(str(name))
        local = directory/name
        require(local.resolve().is_relative_to(directory) and not local.is_symlink(), 'mirror escape')
        data = local.read_bytes()
        digest = hashlib.sha256(data).hexdigest()
        require(digest == row['sha256'] and len(data) == row['bytes'], 'mirror hash/size mismatch')
        mapping.append(dict(original=str(source/name), mirror=str(local), sha256=digest))
    prior = index['previous_inventory']
    prior_name = Path(prior['path'])
    require(not prior_name.is_absolute() and '..' not in prior_name.parts
            and (directory/prior_name).resolve().is_relative_to(directory)
            and not (directory/prior_name).is_symlink(), 'previous inventory escape')
    require(sha(directory/prior['path']) == prior['sha256'], 'previous mirror inventory hash')
    old = json.loads((directory/prior['path']).read_bytes())
    require(all(row in index['files'] for row in old['files']), 'previous mirror entries changed')
    return index, mapping


def inspect_offline_mirror(directory):
    """Preparation diagnostics, deliberately separate from the acceptance gate.

    Frozen incomplete draw audits cannot acquire missing rows by relabelling
    source-derived rectangles as observed engine output.
    """
    directory = Path(directory).resolve()
    index, mapping = mirrored_inventory(directory)
    by_original = {row['original']: row for row in mapping}
    def resolve(original):
        require(original in by_original, 'unindexed original path')
        return Path(by_original[original]['mirror'])
    pair = directory/'os-pair/A'
    clicks = [json.loads((pair/name/'cua-receipt.json').read_bytes())
              for name in ('writer-request-0001', 'writer-request-0003')]
    results, inspected = [], []
    for name in ('writer-request-0002', 'writer-request-0004'):
        folder = pair/name
        request = json.loads((folder/'request.json').read_bytes())
        receipt = json.loads((folder/'cua-receipt.json').read_bytes())
        require(sha(folder/'request.json') == receipt['request_sha256'], 'offline request binding')
        require(receipt['target_pid'] == request['target_pid'], 'offline screenshot PID')
        report, audit = decode(request['report']), decode(request['rectangles'])
        images = []
        for number, suffix in enumerate(('', '-after')):
            conversion_path = folder/('root-early-conversion'+suffix+'.json')
            conversion = json.loads(conversion_path.read_bytes())
            shot = receipt['screenshots'][number]
            require(resolve(conversion['raw_path']) == directory/shot['path'], 'offline raw path binding')
            image = inspect_conversion(conversion, shot, resolve_path=resolve)
            for stream in ('stdout', 'stderr'):
                require(sha(Path(str(conversion_path)+'.'+stream)) == conversion[stream+'_sha256'],
                        'offline converter log hash')
            images.append(image)
            for original in (str(Path(index['source'])/conversion_path.relative_to(directory)),
                             conversion['png_path']):
                inspected.append(by_original[original])
        stream = pair/'writer-stream'
        captures = [row for row in map(json.loads, (stream/'manifest.jsonl').read_text().splitlines())
                    if row.get('label') == name and 'report' in row]
        require(len(captures) == 1, 'offline raw-state capture coverage')
        capture = captures[0]
        state, rest_audit = drawing_context(stream, capture, report)
        frame = frame_rows(report, audit, request, receipt, images, clicks, ROOT,
                           state, rest_audit)
        provenance = []
        for number, suffix in enumerate(('', '-after')):
            conversion_path = folder/('root-early-conversion'+suffix+'.json')
            conversion = json.loads(conversion_path.read_bytes())
            provenance.append(dict(whole_image_sha256=conversion['png_sha256'],
                                   raw_sha256=conversion['raw_sha256'],
                                   conversion_sha256=sha(conversion_path),
                                   request_sha256=sha(folder/'request.json'),
                                   receipt_sha256=sha(folder/'cua-receipt.json')))
        review_dir = ROOT/'docs/plans/original-completion-2026-10-02/execution/p23-03/offline-r07-review'/name
        manifest, digest = write_frame_review(review_dir, frame, provenance)
        credited = reviewed_pixel_categories(manifest, digest, review_dir/'independent-verdict.json', require_judgments=True)
        present = {r['category'] for r in frame['rows'] if r['eligibility'] != 'outside'}
        overlays = [dict(cell=r['cell'], rect=r['rect']) for r in report['traffic']
                    if (r['traffic']['current_density'] > 0 or r['traffic']['temporary_block'])
                    and not any(a['category'] == 'traffic' and a['id'] == ','.join(map(str, r['cell']))
                                for a in audit['rows'])]
        results.append(dict(request=name, layer=report['layer'], manifest_sha256=digest,
                            manifest_path=str(review_dir/'manifest.json'), rows=manifest['rows'],
                            mechanical_pass=frame['mechanical_pass'], errors=frame['errors'],
                            unaudited_traffic_overlays=overlays,
                            pending_categories=[c for c in CATEGORIES if c not in present],
                            pixel_categories=sorted(credited), independent_review='pending'))
    return dict(task_id='P23-03', acceptance='pending', status='blocked',
                inventory_count=len(mapping), path_map=mapping, inspected_eight=inspected,
                frames=results, root_native_os_full_regression_independent_review='pending')


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']
    # ElevatorService.restore_snapshot validates this exact trip shape (412-438).
    # QUEUED means no cabin seat yet, not an absent shaft assignment.
    configs = {shaft['config']['shaft_id']: shaft['config'] for shaft in saved['shafts']}
    for trip in saved['trips']:
        keys = {'actor_id', 'request_id', 'shaft_id', 'connector_id', 'from', 'to',
                'enqueued_at', 'state'}
        if 'companion_id' in trip:
            keys.add('companion_id')
            require(isinstance(trip['companion_id'], str) and trip['companion_id'],
                    'invalid elevator companion identity')
        require(set(trip) == keys, 'exact elevator trip keys')
        require(trip['shaft_id'] in configs, 'elevator trip shaft attribution')
        config = configs[trip['shaft_id']]
        require(trip['from'] != trip['to'] and trip['from'] in config['stops']
                and trip['to'] in config['stops'], 'elevator trip stops')
        low, high = sorted((trip['from'][1], trip['to'][1]))
        require(trip['connector_id'] == f"{trip['shaft_id']}.link.{low}.{high}",
                'elevator trip connector attribution')
        require(trip['state'] in ('QUEUED', 'BOARDED'), 'elevator trip state')
    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']),
                   # get_report() merges config before counting (345-359).
                   waiting_count=sum(t['shaft_id'] == row['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 badge_rectangle(at, text):
    return list(at) + [len(text) * 10 + 6, 14]


def observation_label_layout(report):
    """Independent ordered rectangle placement; never consume reported at/reasons."""
    placed, result = [], []
    for i, cabin in enumerate(report['cabins']):
        placed.extend([(f'cabin/{i}/rect', cabin['rect']),
                       (f'cabin/{i}/badge', badge_rectangle(cabin['badge_at'], cabin['badge']))])
    for kind, collection in [('business', 'badges'), ('supply', 'badges'),
                             ('landing', 'landing_badges'), ('aggregation', 'actor_aggregations')]:
        for i, row in enumerate(report.get(collection, [])):
            if collection == 'badges' and row['kind'] != kind:
                continue
            anchor = list(row.get('anchor', row['at']))
            at, reasons = anchor[:], []
            while True:
                conflict = next(((identity, rect) for identity, rect in placed
                                 if intersects(badge_rectangle(at, row['text']), rect)), None)
                if conflict is None:
                    break
                identity, rect = conflict
                reasons.append(identity)
                at[0] = rect[0] + rect[2]
            result.append((collection, i, dict(anchor=anchor, at=at, displaced_by=reasons)))
            placed.append((f'{kind}/{i}', badge_rectangle(at, row['text'])))
    return result


def observation_anchor_report(report):
    """Projection for existing raw-state semantic recomputation, not evidence."""
    import copy
    raw = copy.deepcopy(report)
    for collection in ('badges', 'landing_badges', 'actor_aggregations'):
        for row in raw.get(collection, []):
            if 'anchor' in row:
                row['at'] = row.pop('anchor')
                row.pop('displaced_by', None)
    return raw


def validate_observation_labels(report):
    require(all(key in report for key in ('cabins', 'badges', 'landing_badges', 'actor_aggregations')),
            'missing observation label collections')
    rectangles = [badge_rectangle(c['badge_at'], c['badge']) for c in report['cabins']]
    rectangles += [badge_rectangle(b['at'], b['text'])
                   for key in ('badges', 'landing_badges', 'actor_aggregations')
                   for b in report.get(key, [])]
    for i, rect in enumerate(rectangles):
        require(not any(intersects(rect, other) for other in rectangles[:i]),
                'observation label intersection')
    for collection, index, wanted in observation_label_layout(report):
        row = report[collection][index]
        for key, value in wanted.items():
            require(key in row, 'missing label ' + key)
            close(row[key], value, 'observation label/' + key)


def intersects(a, b):
    cut = rect_intersection(a, b)
    return cut[2] > 0 and cut[3] > 0


def validate_cabin_intrusion(report):
    for cabin in report['cabins']:
        require(len(cabin['passengers']) <= cabin['passenger_count'], 'passenger drawn in empty cabin')
        for actor in report.get('actor_drawings', []):
            require(not intersects(actor_identity_core(actor), cabin['rect']),
                    'non-passenger inside cabin: ' + actor['key'])


def traffic_obstacles(report):
    obstacles = []
    for cabin in report['cabins']:
        rect = badge_rectangle(cabin['badge_at'], cabin['badge'])
        close(cabin['badge_rect'], rect, 'cabin badge rectangle')
        obstacles.extend([cabin['rect'], rect])
    for collection in ('badges', 'landing_badges', 'actor_aggregations'):
        obstacles.extend(badge_rectangle(b['at'], b['text']) for b in report.get(collection, []))
    return obstacles


def traffic_segments(band, cell, obstacles):
    # Independent rectangle-set subtraction, including the deterministic fallback
    # for a band entirely behind a cabin. Never accept self-asserted no-overlap.
    for y in (band[1], cell[1], cell[1] + cell[3] - band[3]):
        pieces = [[band[0], y, band[2], band[3]]]
        for obstacle in obstacles:
            remaining = []
            for x, top, w, h in pieces:
                cut = rect_intersection([x, top, w, h], obstacle)
                if not cut[2] or not cut[3]:
                    remaining.append([x, top, w, h])
                    continue
                left, upper, cw, ch = cut
                remaining.extend(r for r in ([x, top, w, upper-top],
                    [x, upper+ch, w, top+h-upper-ch], [x, upper, left-x, ch],
                    [left+cw, upper, x+w-left-cw, ch]) if r[2] > 0 and r[3] > 0)
            pieces = remaining
        if pieces:
            return pieces
    return []


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)
        rect = [x + 2, y + 2, w - 4, h - 4]
        thickness = max(2, min(5, (h - 4) * .16))
        band = [rect[0] + rect[2] * .12, y + h - 2 * (w / 16) - thickness / 2,
                rect[2] * .76, thickness]
        rows.append(dict(cell=cell, traffic={k: v for k, v in saved.items() if k != 'cell'},
                         rect=rect, band_rect=band,
                         segment_rects=traffic_segments(band, rect, traffic_obstacles(report))))
    close(report['traffic'], rows, 'traffic and cell rect')
    validate_life_overlay(report, snapshot=snapshot)


def validate_life_overlay(report, root=ROOT, snapshot=None):
    """Check consumed frame and source wiring, not merely reported order flags."""
    serial = report.get('observation_serial')
    require(type(serial) is int and serial > 0
            and type(report.get('overlay_consumed_serial')) is int
            and report['overlay_consumed_serial'] == serial, 'stale LifeOverlay serial')
    require(report.get('overlay_passes') == ['traffic', 'cabins', 'actors'], 'LifeOverlay pass order')
    indices = report.get('overlay_child_indices', [])
    require(len(indices) == 3 and all(type(i) is int and i >= 0 for i in indices)
            and indices[0] < indices[1] < indices[2], 'LifeOverlay child order')
    source = (root / 'scenes/main/world_preview.gd').read_text()
    def body(name):
        match = re.search(r'^func ' + name + r'\([^\n]*\n(.*?)(?=^func |\Z)', source, re.M | re.S)
        require(match is not None, 'missing overlay source function: ' + name)
        return match[1]
    ready, draw, overlay = body('_ready'), body('_draw_contents'), body('_draw_life_overlay')
    ordered = [ready.find('add_child(' + node + ')')
               for node in ('_inactive_layer_art', '_life_overlay', '_layer_lens')]
    require(all(i >= 0 for i in ordered) and ordered == sorted(set(ordered)), 'LifeOverlay source child order')
    require('_life_overlay.draw.connect(_draw_life_overlay)' in ready
            and '_life_overlay.queue_redraw()' in draw, 'LifeOverlay source connection')
    require('_draw_congestion_overlay(' not in draw and '_draw_actors(' not in draw
            and '_draw_observed_actor(' not in body('_draw_elevators'), 'actor or band drawn in parent')
    require(0 <= overlay.find('_draw_congestion_overlay(_life_overlay)')
            < overlay.find('_draw_elevators(_life_overlay)')
            < overlay.find('_draw_life_badges(_life_overlay)')
            < overlay.find('_draw_observed_actor(actor, _life_overlay)')
            < overlay.find('_draw_actors(_life_overlay)'), 'LifeOverlay source pass order')
    require('canvas.draw_rect(rect, Color("34413b"))' in body('_draw_readable_badge')
            and 'canvas.draw_rect(row[&"band_rect"], Color(color, 0.90))' in body('_draw_congestion_overlay'),
            'occluded traffic requires opaque later badges and an actual band draw')
    require('text.length() * 10 + 6' in body('_readable_badge_rect')
            and '_readable_badge_rect(at, text)' in body('_draw_readable_badge')
            and 'PixelArt.elevator_cabin(canvas,' in body('_draw_elevators')
            and 'PixelArt.elevator_gate(canvas,' in body('_draw_elevators')
            and '_draw_elevators(' not in draw, 'cabin/badge draw geometry or active-layer canvas differs')
    actors = list(report['actor_drawings']) + [a for c in report['cabins'] for a in c['passengers']]
    # Projection authority comes from raw state, never riding/region flags.
    cabin_actors, carried_keys = set(), set()
    if snapshot is not None:
        if report['cabins']:
            compare_cabin_geometry(snapshot, report)
            shafts = {s['shaft_id']: s for s in elevator_report(snapshot)['shafts']}
            for cabin in report['cabins']:
                members = {key.replace('staff:', 'staff/', 1).replace('guest:', 'guest/', 1)
                           for key in shafts[cabin['shaft_id']]['boarded']}
                cabin_actors.update(id(a) for a in cabin['passengers'] if a['key'] in members)
        tasks = snapshot['systems']['runtime']['staff']['tasks']
        carried_keys = {'staff/' + t['target_entity_id'] for t in tasks
                        if t['task_type'] == 'rescue_transport'
                        and t['state'] not in ('WORKING', 'COMPLETED', 'CANCELLED', 'FAILED')
                        and t['assigned_staff_id']}
        carried_keys.difference_update(report.get('boarded', {}), report.get('queue_positions', {}))
        if carried_keys:
            # Reuse the original raw binding, track and raised-region fit checks.
            compare_actor_geometry(snapshot, report)
    obstacles = traffic_obstacles(report)
    close(report['traffic_obstacles'], obstacles, 'traffic obstacle geometry')
    validate_cabin_intrusion(report)
    for row in report.get('traffic', []):
        x, y, w, h = row['rect']
        thickness = max(2, min(5, h * .16))
        expected = [x + w * .12, y + h + 2 - 2 * ((w + 4) / 16) - thickness / 2,
                    w * .76, thickness]
        close(row['band_rect'], expected, 'traffic band rectangle')
        segments = traffic_segments(expected, row['rect'], obstacles)
        close(row['segment_rects'], segments, 'traffic segment geometry')
        require(all(not intersects(part, obstacle) for part in segments for obstacle in obstacles),
                'traffic intersects cabin or badge')
        if row['traffic']['current_density'] <= 0 and not row['traffic']['temporary_block']:
            continue
        intervals = []
        for actor in actors:
            core = actor_identity_core(actor)
            overlaps = (core[0] < expected[0] + expected[2] and expected[0] < core[0] + core[2]
                          and core[1] < expected[1] + expected[3] and expected[1] < core[1] + core[3])
            require(not overlaps or id(actor) in cabin_actors
                    or (actor in report['actor_drawings'] and actor['key'] in carried_keys),
                    'floor actor core over traffic band: ' + actor['key'])
            # The relocated fragments also cannot cross an ordinary floor core.
            require(not any(intersects(core, part) for part in segments)
                    or id(actor) in cabin_actors or actor['key'] in carried_keys,
                    'floor actor core over traffic segment: ' + actor['key'])
            if overlaps:
                intervals.append((max(core[0], expected[0]),
                                  min(core[0] + core[2], expected[0] + expected[2])))
        end = expected[0]
        for left, right in sorted(intervals):
            if left > end:
                break
            end = max(end, right)
        require(end < expected[0] + expected[2],
                'traffic band fully occluded: ' + str(row.get('cell')))


def compare_queues(snapshot, report):
    if any('anchor' in b for b in report.get('landing_badges', [])):
        validate_observation_labels(report)
        report = observation_anchor_report(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'])
    footprint = list(dict.fromkeys((cell[0] + dx, cell[1], cell[2])
                     for shaft in elevators['shafts'] for cell in shaft['stops']
                     for dx in range(max(int(shaft.get('width', 2)), 1))))
    close(report['shaft_footprint'], [list(c) for c in footprint], 'shaft footprint')
    lift_cells = set(footprint)
    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 in lift_cells or 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 room_slots(rect, pixel, definition='staff_waiting_room'):
    x, y, w, h = rect
    step = 20 if definition in ('cafeteria', 'tavern') else 16
    columns = max(int(w / pixel / step), 1)
    rows = max(int((h / pixel - (30 if definition == 'chapel' else 10)) / 20), 1)
    if definition == 'chapel':
        columns = min(columns, 4)
    anchors = [[x + (col + .5) * w / columns, y + h - (4 + row * 20) * pixel]
               for row in range(rows) for col in range(columns)]
    return anchors[:8] if definition == 'chapel' else anchors


def compare_rest(state, report, audit, root=ROOT):
    """Recompute selection, not shortest paths. Shortest-path authority is the
    exact pinned PathService source producing the recorded public API paths.
    Edge reports must be nonempty, including for temporarily closed cells.
    """
    require(audit['path_service_sha256'] == sha(root / 'scripts/world/path_service.gd'),
            'rest PathService source pin')
    runtime = state['systems']['runtime']
    members = runtime['staff']['staff']
    facilities = runtime['facilities']['instances']
    require([r['staff_id'] for r in audit['staff']] == [r['staff_id'] for r in members],
            'rest staff order/coverage')
    passable = {tuple(r['cell']) for r in runtime['path']['cell_traffic']}
    resting = {}
    selections = {}
    for member, row in zip(members, audit['staff']):
        sid = member['staff_id']
        for key in ('state', 'health', 'current_task_id', 'current_cell'):
            close(row[key], member[key], 'rest input/' + key)
        memberships = [dict(raid_id=raid['raid_id'], state=raid['state'], side=c['side'])
                       for raid in runtime['defense']['raids'] for c in raid['combatants']
                       if c['actor_id'] == sid]
        close(row['defense_memberships'], memberships, 'rest defense membership')
        locked = sid in runtime['events']['actor_locks']
        transit = any(t['actor_id'] == 'staff:' + sid for t in runtime['elevators']['trips'])
        close(row['actor_lock'], locked, 'rest actor lock')
        close(row['in_transit'], transit, 'rest transit')
        defense = any(r['state'] not in ('DEFENDED', 'BREACHED', 'DEFENDERS_RETREATED', 'CANCELLED')
                      and r['side'] == 'DEFENDER' for r in memberships)
        eligible = member['state'] == 'OFF_DUTY' and member['health'] > 0 and not member['current_task_id'] and not locked and not defense
        selected = dict(reachable=False, resting=False, label='休憩先なし（稼働中の待機室・宿舎と経路を確認）') if eligible else {}
        best_cost = 2147483647
        require([f['instance_id'] for f in row['facilities']] == [f['instance_id'] for f in facilities],
                'rest facility order/coverage')
        for facility, observed in zip(facilities, row['facilities']):
            for key in ('instance_id', 'definition_id', 'origin', 'occupied_cells'):
                close(observed[key], facility[key], 'rest facility/' + key)
            operation = 'SUSPENDED' if facility['suspension_reason'] else ('OPERATING' if facility['state'] in ('OPEN', 'DEGRADED') else 'INACTIVE')
            close(observed['operation_state'], operation, 'rest operation')
            occupied = facility['occupied_cells'] or [facility['origin']]
            probes = [dict(cell=[c[0] + dx, c[1], c[2]], has_cell=(c[0] + dx, c[1], c[2]) in passable)
                      for c in occupied for dx in (0, -1, 1)]
            close(observed['probes'], probes, 'rest candidate probes')
            ordered = sorted({tuple(p['cell']) for p in probes if p['has_cell']}, key=lambda c: (c[2], c[1], c[0]))
            close(observed['ordered'], [list(c) for c in ordered], 'rest candidate order')
            require([r['cell'] for r in observed['routes']] == [list(c) for c in ordered], 'rest route coverage/order')
            best = None
            for route in observed['routes']:
                path, edges = route['path'], route['edges']
                require(len(edges) == max(len(path) - 1, 0), 'rest edge coverage')
                if not path:
                    require(route['calculate_path_cost'] == -1, 'rest empty route cost')
                    continue
                require(path[0] == member['current_cell'] and path[-1] == route['cell'], 'rest path endpoints')
                require(all(tuple(c) in passable for c in path), 'rest path cell absent')
                cost = 0
                for a, b, edge in zip(path, path[1:], edges):
                    require(edge and edge['from'] == a and edge['to'] == b, 'rest empty or discontinuous edge report')
                    require(type(edge['total_cost']) is int and edge['total_cost'] >= 0, 'rest edge cost')
                    cost += edge['total_cost']
                require(cost == route['calculate_path_cost'], 'rest edge sum differs from calculate_path_cost')
                if best is None or cost < best[0]:
                    best = (cost, path[-1])
            if not eligible or operation != 'OPERATING' or facility['definition_id'] not in ('staff_waiting_room', 'staff_dormitory') or best is None or best[0] >= best_cost:
                continue
            best_cost = best[0]
            cell = member['current_cell']
            at = any(c[2] == cell[2] and c[1] == cell[1] and abs(c[0] - cell[0]) <= 1 for c in occupied) and not transit
            selected = dict(reachable=True, resting=at, cell=best[1], facility_id=facility['instance_id'],
                            label=('宿舎で休憩中' if facility['definition_id'] == 'staff_dormitory' else '待機室で休憩中') if at else '休憩先へ移動中')
        close(row['comparison_only'], selected, 'rest selection comparison')
        selections[sid] = selected
        if selected.get('resting'):
            resting[sid] = selected['facility_id']
    close(report['resting_staff'], resting, 'draw resting_staff')
    # Fitting is cross-kind, and queue precedence applies to resting staff too.
    # Reconstruct the complete group and independently prove p=1 no-fit before
    # allowing an identity to appear only in an aggregation's member list.
    if any(r['cells'] for r in runtime['world']['occupancies']
           if r['entity_id'] in resting.values()):
        compare_actor_geometry(state, report)
    return selections


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_badges(state, report):
    if any('anchor' in b for b in report.get('badges', [])):
        validate_observation_labels(report)
        report = observation_anchor_report(report)
    runtime = state['systems']['runtime']
    transform = Transform(runtime, report)
    expected = []
    labels = dict(BROKEN='破損', NO_ACCESS='通路なし', NO_POWER='魔力なし', NO_STAFF='職員不足',
                  CLOSED='営業停止', DEGRADED='低下運転', OPEN='営業中')
    for facility in runtime['facilities']['instances']:
        fid = facility['instance_id']
        if facility['origin'][2] != report['layer']: continue
        cells = next((r['cells'] for r in runtime['world']['occupancies'] if r['entity_id'] == fid), [])
        if not cells: continue
        x,y,w,h = transform.room(cells)
        room = report['rooms'][fid]
        count, limit = room['count'], facility['customer_capacity']
        if facility['suspension_reason']:
            business = '休止・摘発' if facility['suspension_reason'] == 'RAIDED' else '休止・破壊'
            label = business
        else:
            business = labels.get(facility['state'], facility['state'])
            if facility['state'] not in ('OPEN', 'DEGRADED'):
                label = '未稼働'
            else:
                label = f'利用{count}/{limit}人' if limit >= 0 else f'利用{count}人'
                if limit > 0 and count >= limit: label = '満員 ' + label
                visible = len(room_slots([x,y,w,h], transform.dim[0]/16, facility['definition_id']))
                if count > visible: label += f'（他{count-visible}人）'
        expected.append(dict(facility=fid, kind='business', at=[x+2,y+2], text=business+' '+label))
        requests = [r for r in report['goods']['requests'] if fid in (r['source_facility_id'], r['destination_facility_id'])]
        if requests:
            states = {}
            for r in requests: states[r['state']] = states.get(r['state'], 0)+1
            text = '物資依頼 ' + ' / '.join(f'{k}:{states[k]}' for k in sorted(states))
            expected.append(dict(facility=fid, kind='supply', at=[x+2,y+16], requests=requests, text=text))
    close(report['badges'], expected, 'business/supply badges')


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
        else:
            badge = '運転 ' + badge
        doors = shaft['operational'] and shaft['phase'] in ('IDLE', 'BOARDING', 'ALIGHTING')
        badge += (' 開 ' if doors else ' 閉 ') + shaft['phase']
        expected.append(dict(shaft_id=shaft['shaft_id'], rect=rect, pixel=p, operational=shaft['operational'],
                             phase=shaft['phase'], passenger_count=count, doors_open=doors,
                             badge_rect=badge_rectangle([rect[0], rect[1]-14], badge),
                             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)
    validate_cabin_intrusion(report)


def compare_motion(state, report, audit):
    frames = audit['frames']
    require(len(frames) == 2, 'two motion history frames required')
    previous, current = frames
    require(type(previous['frame']) is int and current['frame'] > previous['frame'], 'motion frame order')
    require(current['delta'] >= 0 and current['scaled_delta'] >= 0, 'motion delta')
    before, after = previous['display'], current['display']
    for key, value in after.items():
        close(report[key], value, 'motion report binding/' + key)
    close(after['animation_time'], before['animation_time'] + current['scaled_delta'], 'motion animation delta')
    if before['paused'] and after['paused']:
        close(current['scaled_delta'], 0, 'paused animation moved')
    now = after['animation_time']
    runtime = state['systems']['runtime']
    cells = {'guest/' + r['guest_id']: r['current_cell'] for r in runtime['guests']['guests']}
    cells.update({'staff/' + r['staff_id']: r['current_cell'] for r in runtime['staff']['staff']})
    def interpolated(track, endpoint, paused, transform=None):
        t = 1 if paused else max(0, min(1, (now-track['at'])/.42))
        if transform:
            a, b = transform.center(track['from']), transform.center(endpoint)
            return [a[i] + (b[i]-a[i])*t for i in (0, 1)]
        return track['from']+(endpoint-track['from'])*t
    for key, track in after['actor_tracks'].items():
        require(key in cells and track['cell'] == cells[key], 'motion actor endpoint')
        old = before['actor_tracks'].get(key)
        if old is None:
            wanted = dict(cell=cells[key], **{'from': cells[key]}, at=now, direction=1.0)
        elif old['cell'] == cells[key]:
            wanted = old
        else:
            a, b = old['cell'], cells[key]
            origin = a if a[2] == b[2] and abs(a[0]-b[0])+abs(a[1]-b[1]) == 1 else b
            wanted = dict(cell=b, **{'from': origin}, at=now,
                          direction=(1.0 if b[0] > a[0] else -1.0) if a[0] != b[0] else old['direction'])
        close(track, wanted, 'actor interpolation origin/history')
        # Ordinary walking centers; queue/room/rescue/working projections have
        # independent semantic placement checks and do not use this center.
        drawing = next((a for a in after['actor_drawings'] if a['key'] == key), None)
        if drawing and drawing['mode'] in ('walk', 'carry_walk') and key not in report['queue_positions']:
            position = interpolated(track, cells[key], after['paused'], Transform(runtime, report))
            # Same-cell crowd offsets are handled by the complete actor check.
            require(all(math.isfinite(v) for v in position), 'nonfinite interpolation')
    floors = {s['shaft_id']: s['visual_floor'] for s in elevator_report(state)['shafts']}
    for key, track in after['elevator_tracks'].items():
        require(key in floors, 'unknown elevator motion')
        old = before['elevator_tracks'].get(key)
        if old is None:
            wanted = dict(floor=floors[key], **{'from': floors[key]}, at=now)
        elif math.isclose(old['floor'], floors[key], abs_tol=.00001, rel_tol=.00001):
            wanted = old
        else:
            wanted = dict(floor=floors[key], **{'from': old['floor']}, at=now)
        close(track, wanted, 'elevator interpolation origin/history')
    if before['paused'] and after['paused'] and all(before[k] == after[k] for k in ('layer', 'zoom', 'pan', 'viewport', 'actor_tracks', 'elevator_tracks')):
        prior_actors = {r['key']: r for r in before['actor_drawings']}
        for row in after['actor_drawings']:
            old = prior_actors.get(row['key'])
            if old and all(old[k] == row[k] for k in ('state', 'mode', 'activity', 'receiving', 'carrying')):
                close(row['at'], old['at'], 'paused actor moved')
    return ['motion_history', 'interpolation', 'pause']


def fitted_actor_layout(members, world_pixel, height):
    """Independent interval solver. Inputs are reconstructed preferred rows/bounds."""
    def scaled(member, scale):
        row = copy.deepcopy(member['row'])
        row['scale'] = min(row['scale'], scale)
        s = row['scale']
        row['at'] = [math.copysign(math.floor(abs(v)+.5), v) for v in
                     (row['center'][0]-8*s, row['center'][1]+height/2-3*world_pixel-17*s)]
        return row
    def pack(group, scale):
        trial = [scaled(m, scale) for m in group]
        right = -math.inf
        for member, row in zip(group, trial):
            x, y, w, h = actor_identity_core(row)
            left = max(x, member['bounds'][0], right)
            row['at'][0] += left-x
            right = left+w
        left = math.inf
        for member, row in reversed(list(zip(group, trial))):
            x, y, w, h = actor_identity_core(row)
            bx, by, bw, bh = member['bounds']
            target = min(x, bx+bw-w, left-w)
            row['at'][0] += target-x
            left = target
            if target < bx or y < by or y+h > by+bh:
                return None
        return trial
    active = copy.deepcopy([m for m in members if not m['transit']])
    # A smaller sprite keeps its feet, so its vertical interval moves. Rebuild
    # components after EVERY trial, including unsuccessful trials. Scale ceilings
    # only decrease, making termination independent of an arbitrary iteration cap.
    while True:
        groups, bottom = [], -math.inf
        for member in active:
            member['row'] = scaled(member, member['row']['scale'])
        for member in sorted(active, key=lambda m: (actor_identity_core(m['row'])[1], m['row']['key'])):
            x, y, w, h = actor_identity_core(member['row'])
            if y >= bottom:
                groups.append([])
                bottom = y+h
            else:
                bottom = max(bottom, y+h)
            groups[-1].append(member)
        reduced = False
        for group in groups:
            group.sort(key=lambda m: (m['row']['center'][0], m['row']['key']))
            ceiling = max(1, *(m['row']['scale'] for m in group))
            if pack(group, ceiling) is None and ceiling > 1:
                for member in group:
                    member['row'] = scaled(member, max(1, int(ceiling)-1))
                reduced = True
        if not reduced:
            break
    result = {m['row']['key']: copy.deepcopy(m['row']) for m in members if m['transit']}
    aggregations = []
    for group in groups:
        group.sort(key=lambda m: (m['row']['center'][0], m['row']['key']))
        base = max(1, *(m['row']['scale'] for m in group))
        fitted = pack(group, base)
        if fitted is None:
            retained, fitted, identities = [], [], []
            for member in group:
                row = member['row']
                identities.append(dict(key=row['key'], id=row['key'].split('/', 1)[1],
                                       state=row['state'], cell=row['cell'], mode=row['mode']))
                attempt = pack(retained+[member], 1)
                if attempt is not None:
                    retained.append(member)
                    fitted = attempt
            aggregations.append(dict(members=identities, drawn_keys=[r['key'] for r in fitted],
                                     at=group[0]['bounds'][:2], text='+'+str(len(group)-len(fitted))))
        result.update({row['key']: row for row in fitted})
    return result, aggregations


def actor_track_transit(report, key, cell):
    """Track endpoint is bound to raw state by the caller, not to a draw flag."""
    require(type(report['paused']) is bool, 'actor pause must be boolean')
    track = report['actor_tracks'][key]
    close(track['cell'], cell, 'actor authoritative cell')
    origin = track['from']
    require(isinstance(origin, list) and len(origin) == 3
            and all(type(v) is int for v in origin)
            and origin[2] == cell[2]
            and abs(origin[0]-cell[0])+abs(origin[1]-cell[1]) <= 1,
            'actor track is not a source-valid adjacent step')
    now, started = report['animation_time'], track['at']
    require(all(type(v) in (int, float) and math.isfinite(v) for v in (now, started))
            and now >= started, 'invalid or future actor track time')
    return not report['paused'] and origin != cell and now-started < .42


def landing_actor_region(state, report, transform, cell, region, transit):
    """Derive corridor landings from raw topology, never from actor flags."""
    if not any(intersects(region, cabin['rect']) for cabin in report['cabins']):
        return dict(bounds=region, transit=transit)
    runtime = state['systems']['runtime']
    footprint = {(stop[0]+dx, stop[1], stop[2])
                 for shaft in elevator_report(state)['shafts'] for stop in shaft['stops']
                 for dx in range(max(int(shaft.get('width', 2)), 1))}
    walkable = {tuple(c) for c in runtime['world']['walkable_cells']}
    facilities = codec.indexed(runtime['facilities']['instances'], 'instance_id')
    owners = {tuple(c): r['entity_id'] for r in runtime['world']['occupancies'] for c in r['cells']}
    distance = lambda c: (abs(c[0]-cell[0]), c[0])
    origins = sorted((c for c in footprint if c[1:] == tuple(cell[1:])), key=distance)
    lane = set()
    for origin in origins:
        for direction in (-1, 1):
            for step in range(1, 9):
                target = (origin[0]+direction*step, origin[1], origin[2])
                facility = facilities.get(owners.get(target))
                if (target in footprint or target not in walkable or
                        (facility and facility['definition_id'] not in ('guest_corridor', 'service_corridor'))):
                    break
                lane.add(target)
    lane = sorted(lane, key=distance)
    if not lane:
        rect = transform.rect(cell)
        badges = [b for b in report['landing_badges'] if b['cell'][1:] == list(cell[1:])]
        anchor = min(badges, key=lambda b: distance(b['cell']))['at'] if badges else [rect[0]-80, rect[1]-16]
        return dict(bounds=region, transit=False, no_lane=True, badge_at=anchor)
    reserved = list(report['queue_positions'].values())
    selected = next((c for c in lane if transform.center(c) not in reserved), lane[0])
    return dict(bounds=transform.rect(selected), transit=False, center=transform.center(selected))


def compare_actor_geometry(state, report):
    if any('anchor' in b for b in report.get('actor_aggregations', [])):
        validate_observation_labels(report)
        report = observation_anchor_report(report)
    runtime = state['systems']['runtime']
    compare_queues(state, report)
    compare_room_services(state, report)
    boarded = {actor.replace('staff:', 'staff/', 1).replace('guest:', 'guest/', 1): True
               for shaft in elevator_report(state)['shafts'] for actor in shaft['boarded']}
    close(report['boarded'], boarded, 'boarded actor coverage')
    transform = Transform(runtime, report)
    p, height = transform.dim[0]/16, transform.dim[1]
    facilities = codec.indexed(runtime['facilities']['instances'], 'instance_id')
    staff = codec.indexed(runtime['staff']['staff'], 'staff_id')
    occupancy = {r['entity_id']: r['cells'] for r in runtime['world']['occupancies']}
    # Cabin rectangles and badges are independently recomputed before use below.
    if report['cabins']:
        compare_cabin_geometry(state, report)
    counts, expected = {}, []
    def track_position(key, cell):
        track = report['actor_tracks'][key]
        actor_track_transit(report, key, cell)
        t = 1 if report['paused'] else max(0, min(1, (report['animation_time']-track['at'])/.42))
        a, b = transform.center(track['from']), transform.center(cell)
        return [a[i]+(b[i]-a[i])*t for i in (0, 1)]
    def offset(cell, center):
        key = tuple(cell)
        n = counts.get(key, 0)
        counts[key] = n+1
        return [center[0]+[0,4,-4,2,-2,6,-6][n%7]*p, center[1]-(n%2)*p]
    def slot_position(fid, identifier, definition, fallback, dy, missing_fallback=False):
        cells = occupancy.get(fid, [])
        if not cells:
            return fallback
        users = report['rooms'].get(fid, {}).get('people', [])
        if identifier not in users:
            return fallback if missing_fallback else None
        anchors = room_slots(transform.room(cells), p, definition)
        index = users.index(identifier)
        return [anchors[index][0], anchors[index][1]+dy] if index < len(anchors) else None
    people = {'guest/'+r['guest_id']: r for r in runtime['guests']['guests']}
    people.update({'staff/'+sid: r for sid, r in staff.items()})
    regions = {}
    def actor_region(key):
        if key in regions:
            return regions[key]
        person = people[key]
        cell = person['current_cell']
        rect = transform.rect(cell)
        transit = actor_track_transit(report, key, cell)
        if transit:
            other = transform.rect(report['actor_tracks'][key]['from'])
            x, y = min(rect[0], other[0]), min(rect[1], other[1])
            rect = [x, y, max(rect[0]+rect[2], other[0]+other[2])-x,
                    max(rect[1]+rect[3], other[1]+other[3])-y]
        result = dict(bounds=rect, transit=transit)
        fid = ''
        if key in report['queue_positions']:
            cx, cy = report['queue_positions'][key]
            result = dict(bounds=[cx-transform.dim[0]/2, cy-height/2, transform.dim[0], height], transit=False)
        elif key.startswith('guest/'):
            target = person['target_facility_id']
            if person['current_service_id'] and person['guest_id'] in report['rooms'].get(target, {}).get('people', []):
                fid = target
        else:
            sid = person['staff_id']
            if sid in report['carried_patients']:
                result = copy.deepcopy(actor_region('staff/'+report['carried_patients'][sid]))
                result['bounds'][1] -= 12*p
                if 'center' in result:
                    result['center'][1] -= 12*p
            elif sid in report['resting_staff']:
                fid = report['resting_staff'][sid]
            elif sid in report['treated_patients']:
                fid = report['treated_patients'][sid]
            else:
                target = person['assigned_facility_id']
                cells = occupancy.get(target, [])
                workers = report['rooms'].get(target, {}).get('staff', [])
                if person['state'] in ('WORKING', 'AVAILABLE') and cell in cells and sid in workers:
                    rect = transform.room(cells)
                    columns = max(int(rect[2]/(16*p)), 1)
                    slot = workers.index(sid)
                    if rect[3] >= 40*p and slot < columns:
                        result = dict(bounds=[rect[0]+slot*rect[2]/columns, rect[1], rect[2]/columns, rect[3]], transit=False)
        if fid and occupancy.get(fid):
            result = dict(bounds=transform.room(occupancy[fid]), transit=False)
        if key not in report['queue_positions'] and 'center' not in result and not result.get('no_lane'):
            result = landing_actor_region(state, report, transform, cell, result['bounds'], result['transit'])
        regions[key] = result
        return result
    for guest in runtime['guests']['guests']:
        key, cell = 'guest/'+guest['guest_id'], guest['current_cell']
        if key in report['boarded'] or key in report['queue_hidden'] or cell[2] != report['layer']:
            continue
        center = actor_region(key).get('center', track_position(key, cell))[:]
        if key in report['queue_positions']:
            center = report['queue_positions'][key]
        elif guest['current_service_id']:
            fid = guest['target_facility_id']
            definition = facilities.get(fid, {}).get('definition_id', '')
            cells = occupancy.get(fid, [])
            if cells and cells[0][2] == report['layer']:
                center = slot_position(fid, guest['guest_id'], definition, center,
                                       3*p-height*.5-(5*p if definition in ('standard_room', 'premium_room') else 0), True)
        else:
            center = offset(cell, center)
        if center is not None:
            expected.append((key, center, False, bool(guest['current_service_id']), False, False))
    for sid, member in staff.items():
        key, cell = 'staff/'+sid, member['current_cell']
        if key in report['boarded'] or key in report['queue_hidden'] or cell[2] != report['layer']:
            continue
        center = actor_region(key).get('center', track_position(key, cell))[:]
        special = True
        if key in report['queue_positions']:
            special = False
            center = report['queue_positions'][key]
        elif sid in report['carried_patients']:
            carrier = staff.get(report['carried_patients'][sid])
            if carrier:
                center = track_position('staff/'+carrier['staff_id'], carrier['current_cell'])
                center[1] -= 12*p
                center = actor_region(key).get('center', center)[:]
        elif sid in report['resting_staff']:
            center = slot_position(report['resting_staff'][sid], sid, 'staff_waiting_room', center, -height*.5)
        elif sid in report['treated_patients']:
            center = slot_position(report['treated_patients'][sid], sid, 'infirmary', center, -2*p-height*.5)
        else:
            special = False
            center = offset(cell, center)
            fid = member['assigned_facility_id']
            cells = occupancy.get(fid, [])
            workers = report['rooms'].get(fid, {}).get('staff', [])
            if member['state'] in ('WORKING', 'AVAILABLE') and fid in facilities and cell in cells and sid in workers:
                rect = transform.room(cells)
                columns = max(int(rect[2]/(16*p)), 1)
                index = workers.index(sid)
                if rect[3] >= 40*p and index < columns:
                    center = [rect[0]+(index+.5)*rect[2]/columns, rect[1]+26*p-height*.5]
            center = report['queue_positions'].get(key, center)
        if center is not None:
            expected.append((key, center, True, False, bool(member['carried_inventory_id']) and not special, False))
    floor_keys = [r[0] for r in expected]
    cabin_bounds = {}
    shafts = {s['shaft_id']: s for s in elevator_report(state)['shafts']}
    for cabin in report['cabins']:
        shaft = shafts[cabin['shaft_id']]
        x,y,w,h = cabin['rect']
        columns = max(int((w/p-4)/8), 1)
        passengers = shaft['boarded']
        for i, actor_id in enumerate(passengers):
            key = actor_id.replace('staff:', 'staff/', 1).replace('guest:', 'guest/', 1)
            is_staff = key.startswith('staff/')
            center = [x+(2+(i%columns+.5)*8)*p, y+h-(2+((i//columns)%2)*9)*p+3*p-height*.5]
            carried = bool(staff.get(key.removeprefix('staff/'), {}).get('carried_inventory_id')) if is_staff else False
            expected.append((key, center, is_staff, False, carried, True))
            cabin_bounds[key] = cabin['rect']
    members, stranded = [], []
    for key, center, is_staff, receiving, carrying, riding in expected:
        person = people[key]
        cell = person['current_cell']
        track = report['actor_tracks'].get(key, {})
        relocation = {} if riding else actor_region(key)
        if riding:
            region, transit = cabin_bounds[key], False
        else:
            region, transit = relocation['bounds'], relocation['transit']
        scale = .5 if riding else .75
        scale = max(math.floor(p*scale), 1) if p >= 2 else p*scale
        at = [center[0]-8*scale, center[1]+height*.5-3*p-17*scale]
        rounded = [math.copysign(math.floor(abs(v)+.5), v) for v in at]
        hashed = 5381
        for char in key:
            hashed = (hashed*33+ord(char)) & 0xffffffff
        variant = hashed % 15
        frame = int(report['animation_time']*12)+variant
        activity = facilities.get(person.get('target_facility_id'), {}).get('definition_id', '') if not is_staff else 'work'
        if is_staff and person['role_id'] == 'cleaner':
            activity = 'idle'
            task = next((t for t in runtime['staff']['tasks'] if t['task_id'] == person['current_task_id']), None)
            room = facilities.get(person['assigned_facility_id'], {})
            if (person['state'] == 'WORKING' and person['health'] > 0 and not person['is_retiring']
                    and room.get('definition_id') == 'cleaning_room' and room.get('state') in ('OPEN', 'DEGRADED')
                    and not room.get('suspension_reason') and task and task['task_type'] == 'facility_clean'
                    and task['state'] == 'WORKING' and task['assigned_staff_id'] == person['staff_id']
                    and cell == task['target_cell'] and any(r['cell'] == cell for r in runtime['path']['cell_traffic'])
                    and not any(t['actor_id'] == 'staff:'+person['staff_id'] for t in runtime['elevators']['trips'])
                    and task['target_entity_id'] in facilities
                    and any(r['id'] == task['target_entity_id'] for r in runtime['facility_cleanliness']['facilities'])):
                activity = 'clean'
        moving = (not receiving and key not in report['queue_positions'] and not riding
                  and not report['paused'] and track.get('cell') != track.get('from')
                  and report['animation_time']-track.get('at', 0) < .42)
        mode = 'idle'
        if is_staff and person['state'] == 'INCAPACITATED':
            mode = 'collapsed'
        elif receiving and activity in ('standard_room', 'premium_room'):
            mode = 'standard_room_sleep' if activity == 'standard_room' else 'sleep'
        elif moving:
            mode = 'carry_walk' if carrying else 'walk'
        elif carrying:
            mode = 'carry'
        elif is_staff and person['state'] == 'WORKING' and not riding:
            mode = activity if activity in ('clean', 'idle', 'work') else 'work'
        elif receiving and activity in ('cafeteria', 'tavern'):
            mode = 'eat'
        elif key in report['queue_positions'] or (not is_staff and person['state'] == 'QUEUING'):
            mode = 'wait'
        states = enum_names(ROOT, 'scripts/staff/staff_runtime.gd' if is_staff else 'scripts/guest/guest_runtime.gd', 'State')
        row = dict(key=key, center=center, at=rounded, scale=scale, cell=cell, mode=mode,
                   frame=frame, variant=variant, flip=track.get('direction', 1) < 0,
                   staff=is_staff, state=states.index(person['state']), receiving=receiving,
                   carrying=carrying, riding=riding)
        if relocation.get('no_lane'):
            aggregate = next((a for a in stranded if a['at'] == relocation['badge_at']), None)
            if aggregate is None:
                aggregate = dict(members=[], drawn_keys=[], at=relocation['badge_at'], text='')
                stranded.append(aggregate)
            aggregate['members'].append(dict(key=key, id=key.split('/', 1)[1], state=row['state'], cell=cell, mode=mode))
            aggregate['text'] = '+'+str(len(aggregate['members']))
        else:
            members.append(dict(row=row, bounds=[region[0]+p, region[1]+p, region[2]-2*p, region[3]-2*p], transit=transit))
    fitted, aggregations = fitted_actor_layout(members, p, height)
    aggregations = stranded + aggregations
    close(report.get('actor_aggregations'), aggregations, 'actor aggregation/no-fit')
    require([r['key'] for r in report['actor_drawings']] == [k for k in floor_keys if k in fitted], 'actor drawing coverage/order')
    actual = list(report['actor_drawings']) + [r for c in report['cabins'] for r in c['passengers']]
    require(len(actual) == len(fitted) and {r['key'] for r in actual} == set(fitted), 'missing or duplicate actor')
    for row in actual:
        require('region' not in row and 'boarded' not in row, 'unsupported actor projection flags')
        for field, value in fitted[row['key']].items():
            close(row[field], value, 'actor fitted '+field)
    exemptions = {m['row']['key'] for m in members if m['transit']}
    validate_actor_nonoverlap(report, transit_keys=exemptions)
    return ['actor_geometry', 'passenger_geometry']


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 drawing_context(stream, capture, report):
    """Shared OS/offline binding: a missing stream never grants exemptions."""
    require(isinstance(capture, dict) and all(k in capture for k in ('report', 'state', 'rest_audit')),
            'missing raw drawing context')
    close(read_object(stream, capture['report']), report, 'raw-state report binding')
    state = read_object(stream, capture['state'])
    rest = read_object(stream, capture['rest_audit'])
    require(isinstance(state, dict) and isinstance(state.get('systems', {}).get('runtime'), dict),
            'malformed raw state context')
    require(isinstance(rest, dict) and isinstance(rest.get('staff'), list)
            and 'path_service_sha256' in rest, 'malformed rest authority')
    return state, rest


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'])
            close(read_object(directory, row['state_after']), state, 'read-only audit snapshot identity')
            report = read_object(directory, row['report'])
            fields = compare_actor_rows(state, report, root)
            require(report['elevators'] == elevator_report(state), 'typed elevator report mismatch')
            validate_observation_labels(report)
            compare_room_services(state, report, root)
            compare_rest(state, report, read_object(directory, row['rest_audit']), root)
            compare_queues(state, report)
            compare_traffic(state, report)
            compare_goods(state, report, root)
            compare_badges(state, report)
            compare_cabin_geometry(state, report)
            compare_motion(state, report, read_object(directory, row['motion']))
            compare_actor_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(executed=None):
    executed = executed or {}
    result = []
    for condition in CONDITIONS:
        categories = {}
        for category in CATEGORIES:
            actual = set(executed.get(condition, {}).get(category, []))
            required = REQUIRED_CHECKS[condition][category]
            missing = sorted(required - actual)
            categories[category] = dict(checks=sorted(actual), missing=missing,
                                        status='passed' if actual and not missing else 'pending')
        result.append(dict(condition=condition, categories=categories,
                           status='passed' if all(c['status'] == 'passed' for c in categories.values()) else 'pending'))
    return result


def BLOCKERS(executed=None):
    return {r['condition']: {c: d['missing'] for c, d in r['categories'].items() if d['missing']}
            for r in condition_matrix(executed) if r['status'] != 'passed'}


CONTRACT = 'docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json'
PLAN = 'docs/plans/original-completion-2026-10-02/IMPLEMENTATION_PLAN.md'
BASE_COMMIT = '860136f7099f5e3aee99da5798a627a77f2d8eea'
ALLOWED_PATHS = ('scenes/main/world_preview.gd', 'scripts/ui/room_presentation.gd',
    'tests/ui/test_p23_life_observation.gd', 'tests/ui/test_p23_life_observation.gd.uid',
    'tests/fixtures/p23-03/**', 'tools/run_p23_life_observation.py',
    'tools/check_p23_life_observation_evidence.py', 'tools/test_p23_life_observation_evidence.py',
    'tools/regression/suite-policy.json',
    'docs/plans/original-completion-2026-10-02/execution/p23-03/**', CONTRACT)


def strict_json(data):
    def pairs(items):
        result = {}
        for key, value in items:
            require(key not in result, 'duplicate JSON key: ' + key)
            result[key] = value
        return result
    return json.loads(data, object_pairs_hook=pairs,
                      parse_constant=lambda value: require(False, 'nonfinite JSON: ' + value))


def marker_join(base, submitted, main):
    """Preserve M verbatim structurally, adding exactly the S P23 row."""
    test = 'tests/ui/test_p23_life_observation.gd'
    for document in (base, submitted, main):
        rows = document['rows']
        require(len({row['test'] for row in rows}) == len(rows), 'duplicate marker test')
    additions = [row for row in submitted['rows'] if row['test'] == test]
    require(len(additions) == 1 and not any(row['test'] == test for row in base['rows'] + main['rows']),
            'marker must add exactly one new P23 row')
    stripped = copy.deepcopy(submitted)
    stripped['rows'] = [row for row in stripped['rows'] if row['test'] != test]
    require(structural_equal(stripped, base), 'submitted marker changes non-P23 structure')
    joined = copy.deepcopy(main)
    joined['rows'].append(copy.deepcopy(additions[0]))
    return joined


def structural_equal(a, b):
    return json.dumps(a, sort_keys=True, ensure_ascii=False, allow_nan=False) == json.dumps(
        b, sort_keys=True, ensure_ascii=False, allow_nan=False)


def verify_marker_join(base, submitted, main, tested):
    expected = marker_join(base, submitted, main)
    p23 = expected['rows'][-1]
    require(structural_equal([row for row in tested['rows'] if row['test'] == p23['test']], [p23]),
            'joined marker P23 row changed or duplicated')
    # merge-tree determines the insertion position; deleting the one addition
    # must leave M's entire ordered structure, including all P18 rows, intact.
    stripped = copy.deepcopy(tested)
    stripped['rows'] = [row for row in stripped['rows'] if row['test'] != p23['test']]
    require(structural_equal(stripped, main), 'completion marker is not the strict structural join')


def git_bytes(root, *arguments):
    """Root-only adapter. Worker tests inject an in-memory object database."""
    process = subprocess.run(['git', '-C', str(root), *arguments], capture_output=True, check=False)
    require(process.returncode == 0, 'Git composition failed/conflicted: ' + ' '.join(arguments))
    return process.stdout


_VERIFIED_VIEWS = {}


def blob_oid(content):
    return hashlib.sha1(b'blob %d\0' % len(content) + content).hexdigest()


def git_blob_stream(root, oids, read_batch=None):
    """Pace requests to one owned child; adapters return (binary stream, status)."""
    requested = list(oids)
    require(all(isinstance(oid, str) and re.fullmatch('[0-9a-f]{40}', oid)
                for oid in requested), 'invalid blob oid')
    process = stream = None
    try:
        if read_batch is None:
            process = subprocess.Popen(['git', '-C', str(root), 'cat-file', '--batch'],
                stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
            stream = process.stdout
        else:
            stream, status = read_batch(requested)
        count = 0
        for oid in requested:
            if process is not None:
                process.stdin.write((oid + '\n').encode())
                process.stdin.flush()
            header = stream.readline(128)
            match = re.fullmatch(rb'([0-9a-f]{40}) blob (0|[1-9][0-9]*)\n', header)
            require(match is not None and match[1].decode() == oid, 'invalid/out-of-order blob header')
            size = int(match[2])
            content = stream.read(size)
            require(len(content) == size, 'short blob read')
            require(stream.read(1) == b'\n', 'missing blob LF')
            require(blob_oid(content) == oid, 'blob content SHA-1 mismatch')
            count += 1
            yield oid, content
        if process is not None:
            process.stdin.close()
        require(stream.read(1) == b'', 'extra trailing blob output')
        require((process.wait() if process is not None else status) == 0, 'cat-file failed')
        require(count == len(requested), 'blob output count mismatch')
    finally:
        if process is not None:
            if process.poll() is None:
                process.kill()
            process.wait()
            process.stdin.close()
            process.stdout.close()
        elif stream is not None:
            stream.close()


def verify_composition(root, roles, source_map, inputs, read_git=None, view=None, read_batch=None):
    require(isinstance(roles, dict), 'missing externally bound S/M/T/Q')
    names = ('submitted_sha', 'main_context_sha', 'tested_tree', 'tested_commit')
    require(all(isinstance(roles.get(k), str) and re.fullmatch('[0-9a-f]{40}', roles[k]) for k in names),
            'invalid externally bound S/M/T/Q')
    s, m, t, q = (roles[k] for k in names)
    git = read_git or (lambda *args: git_bytes(root, *args))
    for commit in (BASE_COMMIT, s, m, q):
        require(git('cat-file', '-t', commit).strip() == b'commit', 'role is not a commit')
    require(git('cat-file', '-t', t).strip() == b'tree', 'T is not a tree')
    require(git('rev-parse', q+'^{tree}').decode().strip() == t, 'Q does not contain T')
    head = git('rev-parse', 'HEAD').decode().strip()
    if head not in (s, q):
        require(git('rev-parse', 'HEAD^{tree}').decode().strip() == t, 'C tree differs from T')
        require(git('rev-list', '--parents', '-n', '1', 'HEAD').decode().split() == [head, m, s],
                'C parents must be exactly [M,S]')
    require(not git('status', '--porcelain'), 'tested checkout is dirty')
    merged = git('merge-tree', '--write-tree', s, m).decode().splitlines()
    require(merged and merged[0] == t, 'T differs from deterministic merge-tree')
    def tree(revision):
        result = {}
        for record in git('ls-tree', '-r', '-z', revision).split(b'\0'):
            if not record:
                continue
            metadata, path = record.split(b'\t', 1)
            mode, kind, oid = metadata.decode().split()
            require(path.decode() not in result, 'duplicate T path')
            result[path.decode()] = (mode, kind, oid)
        return result
    base, submitted, main, tested = map(tree, (BASE_COMMIT, s, m, t))
    changed_s = {p for p in base.keys() | submitted.keys() if base.get(p) != submitted.get(p)}
    changed_m = {p for p in base.keys() | main.keys() if base.get(p) != main.get(p)}
    def allowed(path):
        return any(path.startswith(rule[:-2]) if rule.endswith('/**') else path == rule
                   for rule in ALLOWED_PATHS)
    require(all(allowed(p) for p in changed_s), 'S changed a path outside allowed_paths')
    for paths, authority, label in ((changed_s-changed_m, submitted, 'S-only'),
                                    (changed_m-changed_s, main, 'M-only')):
        for path in paths:
            require(tested.get(path) == authority.get(path), label + ' blob/mode provenance: ' + path)
    def document(revision):
        return strict_json(git('show', revision + ':' + CONTRACT))
    verify_marker_join(document(BASE_COMMIT), document(s), document(m), document(t))
    runner = module('p23_composition_source_filter', ROOT/'tools/run_all_tests.py')
    if view is not None:
        registered = _VERIFIED_VIEWS.get(Path(view).resolve())
        require(registered is not None, 'unregistered verified view')
        require(registered[0] == t, 'verified view tree mismatch')
        require(registered[1] == {p: e for p, e in tested.items() if e[1] == 'blob'},
                'verified view manifest mismatch')
    with (contextlib.nullcontext(Path(view)) if view is not None else
          tested_checkout(root, roles, git, read_batch)) as source_view:
        require(source_map == {p: row['sha256'] for p, row in runner.source_refs(source_view).items()},
                'incomplete tested-tree source/settings/QA map')
    # Source refs include all QA suites and settings, not just S-changed files.
    required = set(source_map) | {p.split('#', 1)[0] for p in inputs} | {PLAN, CONTRACT,
        'project.godot', 'tests/business/test_required_kitchen.gd', 'tools/regression/suite-policy.json'}
    bindings = roles.get('blob_bindings')
    require(isinstance(bindings, dict) and set(bindings) == required, 'incomplete/extraneous T blob bindings')
    paths = sorted(required)
    for path in paths:
        entry = tested.get(path)
        require(entry is not None and entry[1] == 'blob' and bindings[path] == entry[2], 'T blob binding: ' + path)
    contents = (((tested[p][2], (Path(view)/p).read_bytes()) for p in paths) if view is not None
                else git_blob_stream(root, [tested[p][2] for p in paths], read_batch))
    with contextlib.closing(contents):
        for number, (oid, content) in enumerate(contents):
            path = paths[number]
            require(blob_oid(content) == tested[path][2] == oid, 'binding blob SHA-1 mismatch: ' + path)
            if head != s or path in changed_s-changed_m:
                require((root/path).read_bytes() == content, 'checkout bytes differ from T: ' + path)
            digest = hashlib.sha256(content).hexdigest()
            if path in source_map:
                require(source_map[path] == digest, 'evidence source is not T: ' + path)
            if path in inputs:
                require(inputs[path] == digest, 'evidence input is not T: ' + path)
    return {k: roles[k] for k in names}


@contextlib.contextmanager
def tested_checkout(root, roles, read_git=None, read_batch=None):
    """Disposable Git-blob view shared by the existing filters and readers."""
    require(isinstance(roles, dict), 'missing externally bound S/M/T/Q')
    tree = roles.get('tested_tree', '')
    require(isinstance(tree, str) and re.fullmatch('[0-9a-f]{40}', tree), 'invalid T')
    git = read_git or (lambda *args: git_bytes(root, *args))
    with tempfile.TemporaryDirectory(prefix='p23-tested-tree-') as temporary:
        view = Path(temporary)
        manifest = {}
        for record in git('ls-tree', '-r', '-z', tree).split(b'\0'):
            if not record:
                continue
            metadata, raw_path = record.split(b'\t', 1)
            mode, kind, oid = metadata.decode().split()
            path = Path(raw_path.decode())
            require(not path.is_absolute() and '..' not in path.parts, 'unsafe T path')
            require(kind == 'blob' and mode in ('100644', '100755'), 'T source aliases refused')
            require(path.as_posix() not in manifest, 'duplicate T path')
            manifest[path.as_posix()] = (mode, kind, oid)
        paths = list(manifest)
        contents = git_blob_stream(root, [manifest[p][2] for p in paths], read_batch)
        with contextlib.closing(contents):
            for number, (_, content) in enumerate(contents):
                target = view/paths[number]
                target.parent.mkdir(parents=True, exist_ok=True)
                target.write_bytes(content)
        _VERIFIED_VIEWS[view.resolve()] = (tree, manifest)
        try:
            yield view
        finally:
            del _VERIFIED_VIEWS[view.resolve()]


def root_context(directory, index, explicit=None):
    path = directory/'root-composition.json'
    require(path.is_file() and not path.is_symlink()
            and path.resolve().is_relative_to(directory.resolve()), 'missing/unsafe Root composition context')
    data = path.read_bytes()
    require(index.get('root_composition_sha256') == hashlib.sha256(data).hexdigest(),
            'Root composition context SHA mismatch')
    roles = strict_json(data)
    require(explicit is None or structural_equal(explicit, roles), 'explicit Root context disagrees')
    return roles


def normative_plan_text(root):
    """Requirement text is an input; explicitly labelled progress is metadata."""
    text = (root / PLAN).read_text()
    section = re.search(r'^### P23[^\n]*\n(.*?)(?=^### |\Z)', text, re.M | re.S)
    require(section is not None, 'missing normative #p23 plan section')
    lines = []
    for line in section.group(0).splitlines():
        if line.startswith('現在地:'):
            continue
        line = line.split(' 状態:', 1)[0].rstrip()
        if line:
            lines.append(line)
    require(any(line.startswith('- P23-03 ') and '受入:' in line for line in lines),
            'missing P23-03 work/acceptance plan text')
    return '\n'.join(lines) + '\n'


def input_map(root):
    names = [SPEC, CONTRACT, 'AGENTS.md', 'docs/traceability/original/BASELINE.txt']
    names += [p.relative_to(root).as_posix() for p in (root / 'tests/fixtures/p23-03').rglob('*')
              if p.is_file()]
    pins = {name: sha(root / name) for name in sorted(names)}
    pins[PLAN + '#p23-normative'] = hashlib.sha256(normative_plan_text(root).encode()).hexdigest()
    return pins


def input_provenance(root):
    # Full-plan progress changes deliberately do not participate in input equality.
    return {'full_plan_sha256': sha(root / PLAN)}


def inspect_os_receipt(receipt, source_map, pins=None):
    require(receipt.get('engine_ended') is True, 'engine_ended must be true')
    require(receipt.get('passed') is True and receipt.get('exit_code') == 0, 'unsuccessful OS process')
    require(not receipt.get('error') and receipt.get('guard_errors') == []
            and receipt.get('cleanup_errors') == [], 'OS cleanup/guard failure')
    require(receipt.get('default_before') == receipt.get('default_after')
            and isinstance(receipt.get('default_before'), dict), 'default profile changed')
    require(receipt.get('stage_before') == receipt.get('stage_after')
            and isinstance(receipt.get('stage_before'), dict), 'stage changed')
    require(receipt.get('source_map') == source_map, 'OS source pin differs')
    after = receipt.get('source_after', {})
    require({name: pin['sha256'] for name, pin in after.items()} == source_map, 'source changed')
    cleanup = receipt.get('owned_cleanup')
    require(isinstance(cleanup, list) and cleanup
            and all(row.get('engine_ended') is True for row in cleanup), 'missing owned cleanup')
    require(bool(receipt.get('ordinary_owner_proof')), 'missing ordinary audio owner proof')
    for field in ('engine_sha256', 'settings_sha256', 'environment_sha256'):
        require(isinstance(receipt.get(field), str) and re.fullmatch('[0-9a-f]{64}', receipt[field])
                and receipt[field] == receipt.get(field + '_after'), 'OS pin changed: ' + field)
    require(receipt.get('spec_sha256') == SPEC_SHA, 'OS immutable spec')
    require(receipt.get('inputs') and receipt['inputs'] == receipt.get('inputs_after'), 'OS inputs changed')
    if pins is not None:
        require(receipt['inputs'] == pins, 'OS input applicability')
    require(receipt.get('environment') and receipt['environment'] == receipt.get('environment_after'), 'OS environment changed')
    require(type(receipt.get('actual_pid')) is int and receipt['actual_pid'] > 0, 'OS actual PID')
    require(receipt.get('profile') and receipt.get('profile_name') and isinstance(receipt.get('profile_after'), dict), 'OS private profile provenance')
    before = receipt.get('profile_before', {})
    require(type(before.get('existed')) is bool and isinstance(before.get('files'), dict), 'OS profile before')
    for member, pin in before['files'].items():
        require(receipt['profile_after'].get(member) == pin, 'OS profile input changed: ' + member)


def inspect_resume(writer, reader, saved, requests, formal_members):
    require(type(writer['actual_pid']) is int and type(reader['actual_pid']) is int
            and writer['actual_pid'] > 0 and reader['actual_pid'] > 0
            and writer['actual_pid'] != reader['actual_pid'], 'resume requires distinct actual PIDs')
    require(saved['pid'] == writer['actual_pid'], 'formal writer PID')
    require(set(formal_members) == {'world.bin', 'manifest.json', 'thumbnail.png', 'journal.log'}, 'formal members coverage')
    close(saved['members'], formal_members, 'formal member hashes')
    require({k: v['sha256'] for k, v in reader['formal_input'].items()} == formal_members, 'reader formal input hashes')
    loads = [row for row in requests if row['request']['action'] == 'load_menu']
    require(len(loads) == 1, 'missing or duplicate normal load ACK')
    item = loads[0]
    request, ack, observed, cua = [item[k] for k in ('request', 'ack', 'observed', 'cua')]
    require(request['target_pid'] == ack['target_pid'] == observed['pid'] == cua['target_pid'] == reader['actual_pid'], 'load ACK PID')
    require(request['sequence'] == ack['sequence'] == observed['sequence'], 'load ACK request ID')
    require(ack['request_sha256'] == item['request_sha256'] == observed['request_sha256'], 'load ACK hash')
    require(0 < request['requested_ticks_usec'] < observed['ack_observed_ticks_usec'], 'load ACK time order')
    require(cua['method'] == 'native_os' and cua['tool_call_id'] and cua['timestamp_ns'] > 0, 'load CUA provenance')
    require(request['requested_unix_ns'] <= cua['timestamp_ns'] <= ack['timestamp_ns'] <= observed['ack_observed_unix_ns'], 'load external ACK time order')
    require(observed['status'] == '手動セーブを読み込みました。', 'normal load result')
    return ['distinct_pid', 'formal_members', 'load_ack']


def require_screenshot_coverage(action, screenshots):
    require(isinstance(screenshots, list) and len(screenshots) >= (1 if action == 'screenshot' else 2),
            'screenshot-only needs one image; real input needs before/after images')


def inspect_os_requests(directory, layout, role, pid, root=ROOT):
    pair = directory/'os-pair'/layout
    def path(name):
        result = directory/name
        require(result.resolve().is_relative_to(directory.resolve()) and not result.is_symlink(), 'OS artifact escape')
        return result
    def document(name):
        return json.loads(path(name).read_bytes())
    listing = json.loads((pair/(role+'-requests.json')).read_bytes())
    require(listing['pid'] == pid and listing['requests'] and len(set(listing['requests'])) == len(listing['requests']), 'OS request list')
    bindings = json.loads((pair/(role+'-screenshots.json')).read_bytes())
    seen, records, pixel_categories = set(), [], set()
    stream = pair/(role+'-stream')
    captures = {row['label']: row for row in map(json.loads, (stream/'manifest.jsonl').read_text().splitlines()) if 'report' in row}
    prior_tick = -1
    for sequence, ack_name in enumerate(listing['requests'], 1):
        ack_path = path(ack_name)
        request_path = ack_path.with_name('request.json')
        observed = json.loads(ack_path.with_name('observed.json').read_bytes())
        request, ack = json.loads(request_path.read_bytes()), json.loads(ack_path.read_bytes())
        cua = document(ack['input_receipt'])
        require_screenshot_coverage(request['action'], cua.get('screenshots'))
        require(request['sequence'] == ack['sequence'] == observed['sequence'] == sequence, 'OS request sequence')
        require(request['target_pid'] == ack['target_pid'] == observed['pid'] == cua['target_pid'] == pid, 'OS request PID')
        digest = sha(request_path)
        require(ack['request_sha256'] == observed['request_sha256'] == digest, 'OS ACK hash')
        require(prior_tick < request['requested_ticks_usec'] < observed['ack_observed_ticks_usec'], 'OS ACK order')
        prior_tick = observed['ack_observed_ticks_usec']
        require(request['requested_unix_ns'] <= cua['timestamp_ns'] <= ack['timestamp_ns'] <= observed['ack_observed_unix_ns'], 'OS external ACK time order')
        require(cua['method'] == 'native_os' and cua['tool_name'] and cua['tool_call_id']
                and cua['timestamp_ns'] > 0 and cua['collector_pid'] > 0 and cua['collector_pid'] != pid, 'OS CUA provenance')
        require('arguments' in cua and isinstance(cua['arguments'], dict), 'full CUA arguments required')
        if request['action'] != 'screenshot':
            require(cua['events'] and observed['events'], 'OS input events missing')
        if request['action'] == 'wheel':
            require(request['ctrl'] is False and any(e.get('ctrl') is False and e.get('pressed') for e in observed['events']), 'ordinary wheel evidence')
        report = decode(request['report'])
        capture = captures[request_path.parent.name]
        state, rest_audit = drawing_context(stream, capture, report)
        for number, shot in enumerate(cua['screenshots']):
            matches = [b for b in bindings if b['request'] == str(request_path.relative_to(directory)) and b['screenshot_index'] == number]
            require(len(matches) == 1, 'screenshot binding coverage')
            binding = matches[0]
            identity = (binding['request'], number)
            require(identity not in seen, 'duplicate screenshot binding')
            seen.add(identity)
            require(binding['request_sha256'] == digest and binding['target_pid'] == pid and binding['sequence'] == sequence,
                    'screenshot request attribution')
            require(binding['input_receipt'] == ack['input_receipt'] and binding['input_receipt_sha256'] == sha(path(ack['input_receipt'])), 'screenshot CUA receipt hash')
            require(binding['raw_path'] == shot['path'] and binding['raw_sha256'] == shot['raw_sha256'] == sha(path(shot['path']))
                    and binding['tool_call_id'] == shot['tool_call_id'] and shot['tool_call_id'], 'screenshot raw provenance')
            conversion = document(binding['conversion']) if binding['conversion'] else None
            image = screenshot_pixels(path(shot['path']), conversion, shot)
            require(binding['png_sha256'] == sha(path(binding['png_path'])) and (binding['width'], binding['height']) == image[:2], 'decoded screenshot binding')
            if conversion:
                require(Path(conversion['png_path']).resolve() == path(binding['png_path']).resolve(), 'converted output binding')
                for suffix in ('stdout', 'stderr'):
                    require(sha(path(binding['conversion']+'.'+suffix)) == conversion[suffix+'_sha256'], 'converter log hash')
            else:
                require(binding['png_path'] == binding['raw_path'], 'PNG must bind original bytes')
            # Screenshot actions observe a stable report; input before/after
            # screenshots are provenance only and never relabelled as this frame.
            if request['action'] == 'screenshot':
                clicks = [document(json.loads(path(name).read_bytes())['input_receipt'])
                          for name in listing['requests']]
                frame = frame_rows(report, decode(request['rectangles']), request, cua, [image], clicks, root,
                                   state, rest_audit)
                review_dir = request_path.parent/('review-image-' + str(number))
                manifest, manifest_sha = write_frame_review(review_dir, frame, [dict(
                    whole_image_sha256=binding['png_sha256'], raw_sha256=shot['raw_sha256'],
                    conversion_sha256=sha(path(binding['conversion'])) if conversion else None,
                    request_sha256=digest, receipt_sha256=sha(path(ack['input_receipt'])))])
                require(frame['mechanical_pass'], 'frame mechanical checks: ' + str(frame['errors'])
                        + str([r['failures'] for r in frame['rows'] if not r['mechanical_pass']]))
                pixel_categories.update(reviewed_pixel_categories(
                    manifest, manifest_sha, review_dir/'independent-verdict.json', require_judgments=True))
        records.append(dict(request=request, ack=ack, observed=observed, cua=cua, request_sha256=digest))
    require(len(seen) == len(bindings), 'unattributed screenshot bindings')
    return records, pixel_categories


def inspect_native_phases(directory, index, root=ROOT):
    runner = module('p23_existing_runner_checks', root/'tools/run_all_tests.py')
    expected = ['runs/'+name+'/'+name+'-receipt.json' for name in ('import', 'parse', 'layout-a', 'layout-b')]
    require(index['phases'] == expected, 'native phase coverage/order')
    for name in expected:
        path = directory/name
        receipt = json.loads(path.read_bytes())
        pins = json.loads(path.with_name('p23-inputs.json').read_bytes())
        require(pins['receipt_sha256'] == sha(path), 'native receipt pin')
        require(pins['source_map'] == index['source_map'] and {k: v['sha256'] for k, v in pins['source_after'].items()} == index['source_map'], 'native sources changed')
        require(pins['inputs'] == pins['inputs_after'] == input_map(root) and pins['spec_sha256'] == SPEC_SHA, 'native input applicability')
        require(pins['default_before'] == pins['default_after'] and pins['stage_before'] == pins['stage_after'], 'native profile/stage changed')
        require(pins['settings_sha256'] == sha(root/'project.godot'), 'native settings changed')
        require(receipt['passed'] is True and receipt['failed'] is False and receipt['engine_ended'] is True
                and receipt['exit_code'] == 0 and not receipt.get('validation_error') and not receipt.get('guard_error'), 'native process failed')
        capture = receipt['outer_actual_capture']
        require(capture['passed'] is True and capture['all_captured_ended'] is True, 'native owned cleanup')
        require(receipt['whole_startup_profile_after']['fresh_before_engine'] is True, 'native profile protection')
        require(receipt['default_player_entire_profile_unchanged'] is True, 'native default profile changed')
        require(isinstance(receipt['controlled_environment'], dict), 'native environment absent')
        diagnostics = receipt['finite_original_diagnostic_review']
        require(runner.process_gate(receipt['native_process'], diagnostics['allowed_e31_error_lines']), 'native process gate')
        if receipt['test']:
            require(receipt['source_sha256'] == receipt['source_sha256_end'] == sha(root/receipt['test'])
                    and receipt['completion']['passed'] is True and receipt['reported_checks'] > 0 and receipt['reported_failures'] == 0, 'native strict marker')


def check(root, directory, root_roles=None):
    try:
        index_path = directory/'index.json'
        if not index_path.is_file():
            return ['missing evidence index: ' + str(index_path)]
        index = strict_json(index_path.read_bytes())
        roles = root_context(directory, index, root_roles)
        with tested_checkout(root, roles) as view:
            return check_tested(root, directory, roles, view)
    except (OSError, ValueError, KeyError, TypeError, AttributeError) as exc:
        return ['invalid evidence index: ' + str(exc)]


def check_tested(checkout, directory, root_roles, root):
    """Never infer completion of unimplemented gates from an index assertion."""
    deficits = []
    executed = {c: {category: set() for category in CATEGORIES} for c in CONDITIONS}
    semantic, pixel_categories, anchors = set(), set(), set()
    index_path = directory / 'index.json'
    if not index_path.is_file():
        return ['missing evidence index: ' + str(index_path)]
    try:
        index = json.loads(index_path.read_bytes())
        require(index.get('schema') == 'p23-life-observation-v2',
                'historical v1 evidence has diagnostic applicability only')
        runner = module('p23_source_applicability', root/'tools/run_all_tests.py')
        require(index.get('source_map') == {n: v['sha256'] for n, v in runner.source_refs(root).items()}, 'exact frozen source map')
        roles = verify_composition(checkout, root_roles, index['source_map'], input_map(root), view=root)
        if git_bytes(checkout, 'rev-parse', 'HEAD').decode().strip() != roles['submitted_sha']:
            require(index['source_map'] == {n: v['sha256'] for n, v in runner.source_refs(checkout).items()},
                    'exact checkout source map')
        require(index.get('composition') == roles and index.get('commit') == roles['tested_commit'],
                'evidence role binding differs from Root S/M/T/Q; historical v1 is not T evidence')
        if index.get('input_map') != input_map(root):
            deficits.append('normative input_map differs from current immutable inputs')
        if index.get('spec_sha256') != SPEC_SHA:
            deficits.append('immutable spec SHA missing or different')
        if index.get('marker_contract_sha256') != sha(root / CONTRACT):
            deficits.append('marker contract SHA missing or different')
        artifacts = index.get('artifacts', {})
        if not artifacts:
            deficits.append('missing indexed evidence artifacts')
        for name, digest in artifacts.items():
            path = directory / name
            require(path.resolve().is_relative_to(directory.resolve()) and not path.is_symlink(),
                    'artifact escapes evidence directory')
            require(path.is_file() and sha(path) == digest, 'artifact SHA mismatch: ' + name)
        try:
            inspect_native_phases(directory, index, root)
        except (OSError, ValueError, KeyError, TypeError, AttributeError) as exc:
            deficits.append('native evidence deficit: ' + str(exc))
        for layout in ('A', 'B'):
            layout_semantic, layout_pixels = set(), set()
            stream = directory / ('layout-' + layout)
            if (stream / 'manifest.jsonl').is_file():
                inspect_stream(stream, root)
            else:
                deficits.append('missing native layout-' + layout + ' typed diagnostic stream')
            for role in ('writer', 'reader'):
                stream = directory / 'os-pair' / layout / (role + '-stream')
                if (stream / 'manifest.jsonl').is_file():
                    inspect_stream(stream, root)
                    # Successful independent recalculation, with nonempty
                    # observed rows required before a category earns coverage.
                    for line in (stream/'manifest.jsonl').read_text().splitlines():
                        capture = json.loads(line)
                        if 'report' not in capture: continue
                        report = read_object(stream, capture['report'])
                        presence = {'room': report['rooms'], 'queue': report['landing_badges'],
                                    'EV': report['cabins'], 'staff': report['staff'],
                                    'goods': report['goods']['requests'], 'business': report['badges']}
                        semantic.update('semantic_'+c for c, rows in presence.items() if rows)
                        layout_semantic.update(c for c, rows in presence.items() if rows)
                        if any(r['people'] for r in report['rooms'].values()): anchors.add('room_users')
                        if any(r['overflow'] > 0 for r in report['landing_badges']): anchors.add('queue_overflow')
                        if any(r['passengers'] for r in report['cabins']): anchors.add('elevator_passengers')
                        if report['staff'] and report['tasks']: anchors.add('staff_jobs')
                        if any(r['carried_inventory_id'] for r in report['staff']) and report['goods']['requests']: anchors.add('carried_goods')
                    if role == 'reader': anchors.add('resume_recalculation')
                else:
                    deficits.append('missing OS ' + layout + '-' + role + ' typed diagnostic stream')
                ended = directory / 'os-runs' / (layout + '-' + role) / 'ended.json'
                if ended.is_file():
                    receipt = json.loads(ended.read_bytes())
                    inspect_os_receipt(receipt, index.get('source_map'), input_map(root))
                    records, categories = inspect_os_requests(directory, layout, role, receipt['actual_pid'], root)
                    pixel_categories.update('pixels_'+c for c in categories if c in CATEGORIES)
                    layout_pixels.update(categories)
                    if role == 'reader':
                        writer = json.loads((directory/'os-runs'/(layout+'-writer')/'ended.json').read_bytes())
                        pair = directory/'os-pair'/layout
                        members = {p.name: sha(p) for p in (pair/'formal').iterdir() if p.is_file()}
                        inspect_resume(writer, receipt, json.loads((pair/'save.json').read_bytes()), records, members)
                        require((pair/'writer-formal-model.var').read_bytes() == (pair/'reader-formal-model.var').read_bytes(), 'resumed whole formal model')
                        anchors.add('distinct_pid_resume')
                else:
                    deficits.append('missing OS ' + layout + '-' + role + ' process receipt')
            for category in CATEGORIES:
                if category not in layout_semantic or category not in layout_pixels:
                    deficits.append('missing actual OS ' + layout + ' semantic/pixel coverage: ' + category)
        if all((directory / ('layout-' + layout) / 'manifest.jsonl').is_file() for layout in ('A', 'B')):
            compare_layouts(comparison_records(directory / 'layout-A'), comparison_records(directory / 'layout-B'))
            anchors.add('route_improvement')
        if not deficits:
            anchors.add('os_guards')
        for condition in CONDITIONS:
            for category in CATEGORIES:
                executed[condition][category].update((semantic | pixel_categories | anchors) & REQUIRED_CHECKS[condition][category])
        deficits.extend(condition + ': ' + json.dumps(missing, ensure_ascii=False) for condition, missing in BLOCKERS(executed).items())
    except (OSError, ValueError, KeyError, TypeError, AttributeError) as exc:
        deficits.append('invalid evidence index: ' + str(exc))
    return deficits


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--verify-only', action='store_true')
    parser.add_argument('--project-root', type=Path, default=ROOT)
    parser.add_argument('--diagnostics', type=Path)
    parser.add_argument('--offline-mirror', type=Path)
    parser.add_argument('--layout-a', type=Path)
    parser.add_argument('--layout-b', type=Path)
    parser.add_argument('--root-composition', type=Path)
    args = parser.parse_args(argv)
    check_setup_budget_source((args.project_root / 'tests/fixtures/p23-03/observation.gd').read_text())
    if args.offline_mirror:
        require(not any((args.verify_only, args.diagnostics, args.layout_a, args.layout_b)),
                'offline preparation must not be combined with acceptance checks')
        print(json.dumps(inspect_offline_mirror(args.offline_mirror), ensure_ascii=False))
        return 2
    if args.verify_only or (os.environ.get('BUILD_OUTPUT_DIR') and not any(
            (args.diagnostics, args.layout_a, args.layout_b))):
        build = os.environ.get('BUILD_OUTPUT_DIR')
        if not build:
            print('BUILD_OUTPUT_DIR is required', file=sys.stderr)
            return 2
        try:
            root_roles = strict_json(args.root_composition.read_bytes()) if args.root_composition else None
            deficits = check(args.project_root, Path(build) / 'p23-life-observation-evidence', root_roles)
        except (OSError, ValueError) as exc:
            deficits = ['invalid Root composition context: ' + str(exc)]
        print(json.dumps(dict(task_id='P23-03', integrity='fail' if deficits else 'pass',
                              deficits=deficits, acceptance='pending'), ensure_ascii=False))
        return 1 if deficits else 0
    # 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())