TheTree_mock/tools/test_p23_life_observation_evidence.py

#!/usr/bin/env python3
"""Local semantic-checker tests; synthetic examples are never game evidence."""
import copy
import ast
import unittest
import sys
import contextlib
import io
import json
import hashlib
import gzip
import struct
import zlib
import tempfile
import os
import subprocess
from pathlib import Path
from unittest import mock

sys.dont_write_bytecode = True
import check_p23_life_observation_evidence as check


def synthetic_repair28_fields(report, snapshot=None):
    """Counterfactual unit-test projection only; never rewrite raw history."""
    report['overlay_passes'] = ['traffic', 'cabins', 'actors']
    report['shaft_footprint'] = [list(c) for c in dict.fromkeys(
        (stop[0]+dx, stop[1], stop[2]) for shaft in check.elevator_report(snapshot)['shafts']
        for stop in shaft['stops'] for dx in range(max(shaft.get('width', 2), 1)))] if snapshot else []
    for cabin in report.get('cabins', []):
        if snapshot is not None:
            shaft = next(s for s in check.elevator_report(snapshot)['shafts'] if s['shaft_id'] == cabin['shaft_id'])
            cabin.update(phase=shaft['phase'], passenger_count=len(shaft['boarded']))
            count, capacity = len(shaft['boarded']), shaft['capacity']
            cabin['badge'] = ('運転 ' if shaft['operational'] else '停止 ') + ('満員 ' if capacity > 0 and count >= capacity else '') + f'{count}/{capacity}人' + (' 開 ' if cabin['doors_open'] else ' 閉 ') + shaft['phase']
        cabin['badge_rect'] = check.badge_rectangle(cabin['badge_at'], cabin['badge'])
    report['traffic_obstacles'] = check.traffic_obstacles(report)
    for row in report.get('traffic', []):
        row['segment_rects'] = check.traffic_segments(row['band_rect'], row['rect'], report['traffic_obstacles'])
    return report


class Repair28GeometryTests(unittest.TestCase):
    def test_one_screenshot_allowed_but_real_input_pair_kept(self):
        check.require_screenshot_coverage('screenshot', [{}])
        check.require_screenshot_coverage('click', [{}, {}])
        for action, shots in [('screenshot', []), ('click', [{}]), ('wheel', [{}]), ('key', None)]:
            with self.subTest(action=action), self.assertRaises(ValueError):
                check.require_screenshot_coverage(action, shots)

    def test_overview_occlusion_keeps_density_row_and_safe_order(self):
        report = LifeOverlayTests().report()
        report['actor_drawings'] = []
        report['badges'] = [dict(at=[0,y], text='cover') for y in (0,14,28,42,56,70,84)]
        synthetic_repair28_fields(report)
        self.assertEqual(report['traffic'][0]['segment_rects'], [])
        self.assertEqual(report['traffic'][0]['traffic']['current_density'], 1)
        check.validate_life_overlay(report)
        source = (check.ROOT/'scenes/main/world_preview.gd').read_text()
        broken = source.replace('canvas.draw_rect(rect, Color("34413b"))',
                                'canvas.draw_rect(rect, Color("34413b", 0.5))')
        with mock.patch.object(Path, 'read_text', return_value=broken):
            with self.assertRaisesRegex(ValueError, 'opaque later badges'):
                check.validate_life_overlay(report)

    def test_r15_empty_cabin_rejects_queue_guest_even_if_marked_riding(self):
        fixture = json.loads((check.ROOT/'tests/fixtures/p23-03/repair28_r15_negative.json').read_text())
        cabin = dict(rect=fixture['cabin_rect'], passengers=[], passenger_count=fixture['passenger_count'])
        actor = dict(fixture['actor'], riding=True, mechanical_pass=True)
        with self.assertRaisesRegex(ValueError, 'non-passenger inside cabin'):
            check.validate_cabin_intrusion(dict(cabins=[cabin], actor_drawings=[actor]))
        cabin['passengers'] = [actor]
        with self.assertRaisesRegex(ValueError, 'passenger drawn in empty cabin'):
            check.validate_cabin_intrusion(dict(cabins=[cabin], actor_drawings=[]))

    def test_r15_traffic_is_split_against_cabin_and_badge(self):
        fixture = json.loads((check.ROOT/'tests/fixtures/p23-03/repair28_r15_negative.json').read_text())
        row = fixture['traffic']
        obstacles = [fixture['cabin_rect'], check.badge_rectangle(**fixture['queue_badge'])]
        for obstacle in obstacles:
            self.assertTrue(check.intersects(row['band_rect'], obstacle))
        pieces = check.traffic_segments(row['band_rect'], row['rect'], obstacles)
        self.assertTrue(pieces)
        for piece in pieces:
            for obstacle in obstacles:
                self.assertFalse(check.intersects(piece, obstacle))

    def test_exact_subtraction_and_full_cover_fallback(self):
        self.assertEqual(check.traffic_segments([10,85,60,5], [0,0,80,96],
            [[20,0,20,90], [45,80,20,14]]), [[10,85,10,5], [40,85,5,5], [65,85,5,5]])
        self.assertEqual(check.traffic_segments([10,85,60,5], [0,0,80,96],
            [[0,80,80,16]]), [[10,0,60,5]])

    def test_claimed_safe_segments_and_obstacles_are_recomputed(self):
        report = LifeOverlayTests().report()
        report['landing_badges'] = [dict(at=[20,80], text='queue')]
        synthetic_repair28_fields(report)
        check.validate_life_overlay(report)
        for field in ('segment_rects', 'traffic_obstacles'):
            bad = copy.deepcopy(report)
            if field == 'segment_rects':
                bad['traffic'][0][field] = [bad['traffic'][0]['band_rect']]
            else:
                bad[field] = []
            with self.subTest(field=field), self.assertRaisesRegex(ValueError, 'traffic (segment|obstacle) geometry'):
                check.validate_life_overlay(bad)

    def test_width_two_walkable_interior_is_hidden_not_a_passenger(self):
        fixture = SpatialTests()
        fixture.setUp()
        fixture.test_missing_queue_overflow_rejected()
        # The old implementation assigned x+1 to g0. Preserve the typed demand
        # and assert that fabricating that old visible actor is now rejected.
        fixture.report['landing_badges'][0]['overflow'] = 3
        fixture.report['queue_positions'] = {'guest/g0': [100,82]}
        with self.assertRaisesRegex(ValueError, 'queue_positions'):
            check.compare_queues(fixture.snapshot, fixture.report)
        fixture.report['queue_positions'] = {}
        fixture.report['shaft_footprint'].pop(1)
        with self.assertRaisesRegex(ValueError, 'shaft footprint'):
            check.compare_queues(fixture.snapshot, fixture.report)

    def test_cabin_state_false_fields_rejected_on_both_layers(self):
        fixture = SpatialTests()
        fixture.setUp()
        fixture.test_cabin_geometry_and_phase()
        for layer in (0,1):
            fixture.report['layer'] = layer
            fixture.runtime['elevators']['shafts'][0]['config']['stops'][0][2] = layer
            for field, value in [('passenger_count',1), ('operational',False), ('phase','PICKUP'),
                                 ('doors_open',False), ('badge_rect',[80,58,1,14])]:
                bad = copy.deepcopy(fixture.report)
                bad['cabins'][0][field] = value
                with self.subTest(layer=layer,field=field), self.assertRaises(ValueError):
                    check.compare_cabin_geometry(fixture.snapshot, bad)


class Repair28CompositionTests(unittest.TestCase):
    """No Git subprocess, repository writes, engine or proof generation."""
    def setUp(self):
        self.s, self.m, self.t, self.q = ('1'*40, '2'*40, '3'*40, '4'*40)
        base_marker = dict(schema_version=1, rows=[dict(test='old', required=True)])
        s_marker = copy.deepcopy(base_marker)
        s_marker['rows'].append(dict(test='tests/ui/test_p23_life_observation.gd', required=True))
        m_marker = copy.deepcopy(base_marker)
        m_marker['rows'].append(dict(test='P18', required=True))
        self.documents = [base_marker, s_marker, m_marker, check.marker_join(base_marker,s_marker,m_marker)]
        names = ['project.godot','export_presets.cfg','Makefile','AGENTS.md', check.PLAN, check.CONTRACT,
            'tests/business/test_required_kitchen.gd','tools/regression/suite-policy.json',
            'scenes/main/world_preview.gd']
        base = {name: (name+' baseline').encode() for name in names}
        self.files = {r: copy.deepcopy(base) for r in (check.BASE_COMMIT,self.s,self.m,self.t)}
        for revision, document in zip(self.files, self.documents):
            self.files[revision][check.CONTRACT] = json.dumps(document).encode()
        self.files[self.s]['scenes/main/world_preview.gd'] = b'S drawing'
        self.files[self.m]['tools/regression/suite-policy.json'] = b'M only policy'
        self.files[self.m]['tests/business/test_required_kitchen.gd'] = b'M approved QA source'
        self.files[self.t].update({p:self.files[self.s][p] for p in ['scenes/main/world_preview.gd']})
        self.files[self.t].update({p:self.files[self.m][p] for p in
            ['tools/regression/suite-policy.json','tests/business/test_required_kitchen.gd']})
        self.source = {p:hashlib.sha256(v).hexdigest() for p,v in self.files[self.t].items()
                       if p not in (check.PLAN, check.CONTRACT)}
        self.inputs = {check.CONTRACT:hashlib.sha256(self.files[self.t][check.CONTRACT]).hexdigest()}
        self.roles = dict(submitted_sha=self.s, main_context_sha=self.m, tested_tree=self.t, tested_commit=self.q,
                         blob_bindings={p:self.blob(v) for p,v in self.files[self.t].items()})
        self.merge_tree = self.t
        self.calls = []

    @staticmethod
    def blob(value):
        return hashlib.sha1(b'blob '+str(len(value)).encode()+b'\0'+value).hexdigest()

    def git(self, *args):
        self.calls.append(args)
        if args[0] == 'cat-file': return b'tree\n' if args[2] == self.t else b'commit\n'
        if args[0] == 'rev-parse': return (self.q if args[1] == 'HEAD' else self.t).encode()+b'\n'
        if args[0] == 'status': return b''
        if args[0] == 'merge-tree': return self.merge_tree.encode()+b'\n'
        if args[0] == 'ls-tree':
            return b''.join(('100644 blob '+self.blob(value)+'\t'+path+'\0').encode()
                            for path,value in self.files[args[-1]].items())
        if args[0] == 'show':
            revision,path = args[1].split(':',1)
            return self.files[revision][path]
        raise AssertionError(args)

    def batch(self, oids):
        self.calls.append(('cat-file', '--batch'))
        blobs = {self.blob(v): v for files in self.files.values() for v in files.values()}
        return io.BytesIO(b''.join(oid.encode()+b' blob '+str(len(blobs[oid])).encode()+b'\n'
                                  +blobs[oid]+b'\n' for oid in oids)), 0

    def verify(self):
        original = Path.read_bytes
        def read(path):
            if path.is_relative_to(check.ROOT):
                return self.files[self.t][str(path.relative_to(check.ROOT))]
            return original(path)
        with mock.patch.object(Path, 'read_bytes', read):
            return check.verify_composition(check.ROOT,self.roles,self.source,self.inputs,self.git,
                                            read_batch=self.batch)

    def test_valid_route_b_keeps_unchanged_allowed_path_from_main(self):
        self.assertEqual(self.verify()['tested_tree'], self.t)
        self.assertIn(('merge-tree','--write-tree',self.s,self.m),self.calls)

    def test_out_of_scope_submitted_change_rejected(self):
        self.files[self.s]['autoload/simulation_runtime.gd'] = b'unauthorized'
        with self.assertRaisesRegex(ValueError,'outside allowed_paths'): self.verify()

    def test_wrong_merge_tree_rejected(self):
        self.merge_tree = '5'*40
        with self.assertRaisesRegex(ValueError,'deterministic merge-tree'): self.verify()

    def test_conflicted_merge_fails_closed(self):
        original = self.git
        def conflict(*args):
            if args[0] == 'merge-tree': raise ValueError('Git composition failed/conflicted')
            return original(*args)
        with self.assertRaisesRegex(ValueError,'conflicted'):
            check.verify_composition(check.ROOT,self.roles,self.source,self.inputs,conflict)

    def test_s_only_and_main_only_blob_provenance(self):
        for path in ('scenes/main/world_preview.gd','tools/regression/suite-policy.json',
                     'tests/business/test_required_kitchen.gd'):
            with self.subTest(path=path):
                original = self.files[self.t][path]
                self.files[self.t][path] = self.files[check.BASE_COMMIT][path]
                with self.assertRaisesRegex(ValueError,'blob/mode provenance'): self.verify()
                self.files[self.t][path] = original

    def test_structural_join_preserves_all_main_rows_and_metadata(self):
        base,s,m,t = self.documents
        for mutation in ('delete_P18','overwrite_P18','extra','type','metadata','duplicate_P23'):
            bad = copy.deepcopy(t)
            if mutation == 'delete_P18': bad['rows'].pop(1)
            if mutation == 'overwrite_P18': bad['rows'][1]['required'] = False
            if mutation == 'extra': bad['rows'].append(dict(test='unapproved'))
            if mutation == 'type': bad['schema_version'] = True
            if mutation == 'metadata': bad['extra'] = 'changed'
            if mutation == 'duplicate_P23': bad['rows'].append(copy.deepcopy(bad['rows'][-1]))
            with self.subTest(mutation=mutation), self.assertRaises(ValueError):
                check.verify_marker_join(base,s,m,bad)
        positioned = copy.deepcopy(t)
        positioned['rows'].insert(1, positioned['rows'].pop())
        check.verify_marker_join(base,s,m,positioned)
        with self.assertRaisesRegex(ValueError,'duplicate JSON key'):
            check.strict_json('{"rows":[],"rows":[]}')

    def test_binding_completeness_and_old_source_relabel_rejected(self):
        for mutation in ('missing_blob','wrong_blob','extra_blob','old_source','omitted_QA'):
            original_roles, original_source = copy.deepcopy(self.roles),dict(self.source)
            path = 'tests/business/test_required_kitchen.gd'
            if mutation == 'missing_blob': self.roles['blob_bindings'].pop(path)
            if mutation == 'wrong_blob': self.roles['blob_bindings'][path] = '0'*40
            if mutation == 'extra_blob': self.roles['blob_bindings']['unbound'] = '0'*40
            if mutation == 'old_source': self.source[path] = hashlib.sha256(self.files[self.s][path]).hexdigest()
            if mutation == 'omitted_QA': self.source.pop(path)
            with self.subTest(mutation=mutation), self.assertRaises(ValueError): self.verify()
            self.roles, self.source = original_roles, original_source


class Repair31StreamTests(unittest.TestCase):
    def test_owned_child_cleanup_on_parse_failure_and_early_close(self):
        oid = check.blob_oid(b'x')
        for malformed in (True, False):
            child = mock.Mock()
            child.stdin = io.BytesIO()
            child.stdout = io.BytesIO(b'bad\n' if malformed else oid.encode()+b' blob 1\nx\n')
            child.poll.return_value = None
            child.wait.return_value = -9
            with mock.patch.object(subprocess, 'Popen', return_value=child):
                stream = check.git_blob_stream(check.ROOT, [oid])
                if malformed:
                    with self.assertRaises(ValueError): next(stream)
                else:
                    self.assertEqual(next(stream), (oid, b'x'))
                    stream.close()
            child.kill.assert_called_once_with()
            child.wait.assert_called_once_with()
            self.assertTrue(child.stdin.closed)
            self.assertTrue(child.stdout.closed)

    def test_strict_stream_negatives_and_repeated_oids(self):
        content = b'abc'
        oid = check.blob_oid(content)
        other = check.blob_oid(b'def')
        record = oid.encode()+b' blob 3\nabc\n'
        def read(data, oids=(oid,), status=0):
            return list(check.git_blob_stream(check.ROOT, oids,
                        lambda requested: (io.BytesIO(data), status)))
        self.assertEqual(read(record*2, (oid, oid)), [(oid, content)]*2)
        invalid = [record[:-2], record[:-1], record.replace(b'blob 3', b'blob 2'),
                   record.replace(b'blob 3', b'blob 4'), record.replace(b'blob', b'tree'),
                   oid.encode()+b' missing\n', oid.encode()+b' ambiguous\n',
                   record+record, record.replace(b'abc', b'def'), b'',
                   record.replace(oid.encode(), other.encode())]
        for data in invalid:
            with self.subTest(data=data), self.assertRaises(ValueError):
                read(data)
        with self.assertRaises(ValueError):
            read(record, status=1)
        with self.assertRaises(ValueError):
            read(record*2, (oid, other))

    def test_checkout_paths_and_modes(self):
        oid = check.blob_oid(b'x')
        roles = dict(tested_tree='1'*40)
        for path, mode, kind, duplicate in [('a', '100644', 'blob', True),
                ('a','120000','blob',False), ('a','160000','commit',False),
                ('/a','100644','blob',False), ('a/../b','100644','blob',False)]:
            record = f'{mode} {kind} {oid}\t{path}\0'.encode()
            with self.subTest(path=path, mode=mode), self.assertRaises(ValueError):
                with check.tested_checkout(check.ROOT, roles,
                        lambda *args: record*(2 if duplicate else 1),
                        lambda oids: (io.BytesIO(oid.encode()+b' blob 1\nx\n'),0)):
                    self.fail('unsafe checkout yielded')

    def test_view_authority_and_binding_recheck(self):
        fixture = Repair28CompositionTests()
        fixture.setUp()
        def verify(view):
            return check.verify_composition(check.ROOT, fixture.roles, fixture.source,
                fixture.inputs, fixture.git, view=view, read_batch=fixture.batch)
        with tempfile.TemporaryDirectory() as temporary:
            with self.assertRaisesRegex(ValueError, 'unregistered'):
                verify(Path(temporary))
        with check.tested_checkout(check.ROOT, fixture.roles, fixture.git, fixture.batch) as view:
            registered = check._VERIFIED_VIEWS[view.resolve()]
            check._VERIFIED_VIEWS[view.resolve()] = ('0'*40, registered[1])
            with self.assertRaisesRegex(ValueError, 'tree mismatch'): verify(view)
            check._VERIFIED_VIEWS[view.resolve()] = (registered[0], {})
            with self.assertRaisesRegex(ValueError, 'manifest mismatch'): verify(view)
            check._VERIFIED_VIEWS[view.resolve()] = registered
            # A normative document is outside source_refs; its bytes must still be checked.
            (view/check.PLAN).write_bytes(b'tampered')
            with mock.patch.object(Path, 'read_bytes', autospec=True) as reader:
                # Preserve real view reads and synthesize only the source checkout comparison.
                original = io.open
                def read(path):
                    if path.is_relative_to(view):
                        with original(path, 'rb') as stream: return stream.read()
                    return fixture.files[fixture.t][str(path.relative_to(check.ROOT))]
                reader.side_effect = read
                with self.assertRaisesRegex(ValueError, 'SHA-1 mismatch'): verify(view)
        self.assertNotIn(view.resolve(), check._VERIFIED_VIEWS)

    def test_constant_calls_full_composition_with_5000_paths(self):
        fixture = Repair28CompositionTests()
        fixture.setUp()
        additions = {f'assets/bulk/{i:05}.txt': b'repeated small blob' for i in range(5000)}
        for files in fixture.files.values(): files.update(additions)
        fixture.source.update({p: hashlib.sha256(v).hexdigest() for p,v in additions.items()})
        fixture.roles['blob_bindings'].update({p: check.blob_oid(v) for p,v in additions.items()})
        original = Path.read_bytes
        verified = set()
        def read(path):
            if path.is_relative_to(check.ROOT):
                return fixture.files[fixture.t][str(path.relative_to(check.ROOT))]
            data = original(path)
            if 'bulk' in path.parts:
                self.assertEqual(data, b'repeated small blob')
                verified.add(path.name)
            return data
        with mock.patch.object(Path, 'read_bytes', read):
            with check.tested_checkout(check.ROOT, fixture.roles, fixture.git, fixture.batch) as view:
                check.verify_composition(check.ROOT, fixture.roles, fixture.source, fixture.inputs,
                    fixture.git, view=view, read_batch=fixture.batch)
        self.assertEqual(len(verified), 5000)
        self.assertEqual(fixture.calls.count(('cat-file','--batch')), 1)
        self.assertEqual(len([c for c in fixture.calls if c[0] == 'show']), 4)

    def test_real_pipe_overflow_requests_and_repeated_paths(self):
        runtime = os.environ.get('TMPDIR')
        self.assertTrue(runtime, 'assigned TMPDIR required')
        with tempfile.TemporaryDirectory(dir=runtime, prefix='p23-repair31-') as temporary:
            repo = Path(temporary)
            def git(*args, data=None):
                return subprocess.run(['git','-C',str(repo),*args], input=data,
                    capture_output=True, check=True, timeout=30).stdout
            git('init','-q')
            content = b'pipe pacing verified\n'
            oid = git('hash-object','-w','--stdin', data=content).decode().strip()
            count = 5000
            tree = git('mktree', data=b''.join(
                f'100644 blob {oid}\tpath-{i:05}\n'.encode() for i in range(count))).decode().strip()
            # A watchdog bounds a deadlock regression; elapsed time is not a pass metric.
            program = '''import sys
from pathlib import Path
sys.path.insert(0, sys.argv[1])
import check_p23_life_observation_evidence as c
repo, tree = Path(sys.argv[2]), sys.argv[3]
with c.tested_checkout(repo, {'tested_tree': tree}) as view:
    paths = list(view.iterdir())
    assert len(paths) == 5000
    assert all(p.read_bytes() == b'pipe pacing verified\\n' for p in paths)
'''
            subprocess.run([sys.executable, '-B', '-c', program, str(check.ROOT/'tools'),
                            str(repo), tree], check=True, timeout=60)


class Repair30EntryTests(unittest.TestCase):
    """Synthetic labelled Git history; zero native or OS credit."""
    def setUp(self):
        runtime = os.environ.get('TMPDIR')
        self.assertTrue(runtime, 'assigned TMPDIR required')
        self.temporary = tempfile.TemporaryDirectory(dir=runtime, prefix='p23-repair30-')
        self.addCleanup(self.temporary.cleanup)
        self.repo = Path(self.temporary.name)/'repo'
        self.repo.mkdir()
        self.build = Path(self.temporary.name)/'build'
        self.evidence = self.build/'p23-life-observation-evidence'
        self.evidence.mkdir(parents=True)
        self.git('init', '-q')
        self.git('config', 'user.name', 'Synthetic fixture')
        self.git('config', 'user.email', 'fixture@example.invalid')
        for path in ('tools/run_all_tests.py', 'tests/fixtures/p23-03/observation.gd',
                     check.PLAN, check.SPEC, 'docs/traceability/original/BASELINE.txt'):
            self.write(path, (check.ROOT/path).read_bytes())
        for path in ('project.godot', 'export_presets.cfg', 'Makefile', 'AGENTS.md',
                     'scenes/main/world_preview.gd', 'tests/business/test_required_kitchen.gd',
                     'tools/regression/suite-policy.json'):
            self.write(path, 'synthetic baseline\n')
        for i in range(220):
            self.write(f'assets/repair31/{i:04}.txt', f'verified path {i}\n')
        self.marker = dict(schema_version=1, rows=[dict(test='history-'+str(i), required=True) for i in range(6)])
        self.write_marker(self.marker)
        self.base = self.commit('synthetic base')
        submitted = copy.deepcopy(self.marker)
        submitted['rows'].insert(0, dict(test='tests/ui/test_p23_life_observation.gd', required=True))
        self.write_marker(submitted)
        self.write('scenes/main/world_preview.gd', 'synthetic S\n')
        self.s = self.commit('synthetic S')
        self.git('checkout', '-q', '--detach', self.base)
        main = copy.deepcopy(self.marker)
        main['rows'].append(dict(test='P18-and-kitchen-QA', required=True))
        self.write_marker(main)
        self.write('tests/business/test_required_kitchen.gd', 'synthetic M QA\n')
        self.m = self.commit('synthetic M')
        self.t = self.git('merge-tree', '--write-tree', self.s, self.m).splitlines()[0]
        self.q = self.git('commit-tree', self.t, '-p', self.s, '-m', 'synthetic Q, no evidence')
        self.c = self.git('commit-tree', self.t, '-p', self.m, '-p', self.s, '-m', 'synthetic C')
        self.git('checkout', '-q', '--detach', self.q)
        runner = check.module('p23_synthetic_source_filter', check.ROOT/'tools/run_all_tests.py')
        sources = {p: row['sha256'] for p, row in runner.source_refs(self.repo).items()}
        inputs = check.input_map(self.repo)
        required = set(sources) | {p.split('#')[0] for p in inputs} | {check.PLAN, check.CONTRACT,
            'project.godot', 'tests/business/test_required_kitchen.gd', 'tools/regression/suite-policy.json'}
        self.roles = dict(submitted_sha=self.s, main_context_sha=self.m, tested_tree=self.t,
                          tested_commit=self.q, blob_bindings={p:self.git('rev-parse', self.t+':'+p) for p in required})
        self.index = dict(schema='p23-life-observation-v2', source_map=sources, input_map=inputs,
                          composition={k:self.roles[k] for k in ('submitted_sha','main_context_sha','tested_tree','tested_commit')},
                          commit=self.q, spec_sha256=check.SPEC_SHA,
                          marker_contract_sha256=check.sha(self.repo/check.CONTRACT), artifacts={}, phases=[])
        self.publish()
        self.git('checkout', '-q', '--detach', self.s)
        self.patch = mock.patch.object(check, 'BASE_COMMIT', self.base)
        self.patch.start()
        self.addCleanup(self.patch.stop)

    def git(self, *args):
        result = subprocess.run(['git', '-C', str(self.repo), *args], capture_output=True, check=True)
        return result.stdout.decode().strip()

    def write(self, path, content):
        target = self.repo/path
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(content.encode() if isinstance(content, str) else content)

    def write_marker(self, value):
        self.write(check.CONTRACT, json.dumps(value, indent=2)+'\n')

    def commit(self, message):
        self.git('add', '.')
        self.git('commit', '-q', '-m', message)
        return self.git('rev-parse', 'HEAD')

    def publish(self):
        context = self.evidence/'root-composition.json'
        context.write_text(json.dumps(self.roles))
        self.index['root_composition_sha256'] = check.sha(context)
        (self.evidence/'index.json').write_text(json.dumps(self.index))

    def cli(self, verify=True, extra_env=None, extra_args=()):
        env = dict(BUILD_OUTPUT_DIR=str(self.build))
        env.update(extra_env or {})
        with mock.patch.dict(os.environ, env, clear=True), contextlib.redirect_stdout(io.StringIO()) as output:
            code = check.main((['--verify-only'] if verify else []) + ['--project-root', str(self.repo)] + list(extra_args))
        self.assertEqual(code, 1)
        result = json.loads(output.getvalue())
        self.assertEqual(result['acceptance'], 'pending')
        return result['deficits']

    def assert_entry(self, verify=True):
        deficits = self.cli(verify)
        self.assertTrue(any('native evidence deficit' in d for d in deficits), deficits)
        self.assertTrue(any('missing OS A-writer process receipt' in d for d in deficits), deficits)
        for condition in check.CONDITIONS:
            self.assertTrue(any(d.startswith(condition+':') for d in deficits), deficits)
        self.assertFalse(any('invalid evidence index' in d for d in deficits), deficits)

    def test_normal_and_verify_only_S_Q_C_keep_all_seven_blockers(self):
        for head in (self.s, self.q, self.c):
            self.git('checkout', '-q', '--detach', head)
            for verify in (False, True):
                with self.subTest(head=head, verify=verify):
                    self.assert_entry(verify)

    def test_whole_entry_has_constant_git_calls_and_verifies_every_path(self):
        spawn = subprocess.Popen
        calls, read_paths = [], set()
        original = Path.read_bytes
        def counted(*args, **kwargs):
            calls.append(args[0])
            return spawn(*args, **kwargs)
        def read(path):
            data = original(path)
            if 'repair31' in path.parts:
                self.assertEqual(data, f'verified path {int(path.stem)}\n'.encode())
                read_paths.add(path.name)
            return data
        with mock.patch.object(subprocess, 'Popen', counted), mock.patch.object(Path, 'read_bytes', read):
            self.assert_entry()
        self.assertEqual(len(read_paths), 220)
        self.assertEqual(sum(command[-2:] == ['cat-file','--batch'] for command in calls), 1)
        self.assertEqual(sum('show' in command for command in calls), 4)

    def test_wrong_roles_and_missing_Q(self):
        for key, value in (('submitted_sha',self.base), ('main_context_sha',self.base),
                           ('tested_tree',self.git('rev-parse', self.base+'^{tree}')),
                           ('tested_commit','0'*40)):
            original = self.roles[key]
            self.roles[key] = value
            self.publish()
            with self.subTest(key=key):
                self.assertTrue(any('invalid evidence index' in d for d in self.cli()))
            self.roles[key] = original

    def test_dirty_roles_and_source_QA_rollback(self):
        for head in (self.s,self.q,self.c):
            self.git('checkout', '-q', '--detach', head)
            for path in ('scenes/main/world_preview.gd','tests/business/test_required_kitchen.gd'):
                original = (self.repo/path).read_bytes()
                self.write(path, 'synthetic rollback\n')
                with self.subTest(head=head,path=path):
                    self.assertTrue(any('dirty' in d for d in self.cli()))
                self.write(path, original)

    def test_unexpected_C_parents_and_tree(self):
        wrong_tree = self.git('rev-parse',self.base+'^{tree}')
        for tree, parents in ((self.t,[self.s,self.m]), (self.t,[self.m,self.s,self.base]),
                               (self.t,[self.m]), (wrong_tree,[self.m,self.s])):
            args = ['commit-tree', tree, '-m', 'synthetic invalid C']
            for parent in parents:
                args.extend(['-p',parent])
            head = self.git(*args)
            self.git('checkout','-q','--detach',head)
            with self.subTest(tree=tree,parents=parents):
                self.assertTrue(any('C parents' in d or 'C tree' in d for d in self.cli()))

    def test_context_missing_mismatch_symlink_and_environment_ignored(self):
        context = self.evidence/'root-composition.json'
        original = context.read_bytes()
        context.unlink()
        self.assertTrue(any('context' in d for d in self.cli(extra_env={'P23_ROOT_COMPOSITION':str(self.repo/'unused')})))
        context.write_bytes(original+b' ')
        self.assertTrue(any('SHA mismatch' in d for d in self.cli()))
        context.unlink()
        target = Path(self.temporary.name)/'context.json'
        target.write_bytes(original)
        context.symlink_to(target)
        self.assertTrue(any('unsafe' in d for d in self.cli()))
        context.unlink()
        context.write_bytes(original)
        alternate = Path(self.temporary.name)/'alternate.json'
        alternate.write_text('{}')
        self.assertTrue(any('disagrees' in d for d in self.cli(extra_args=['--root-composition',str(alternate)])))
        self.assertTrue(any('native evidence deficit' in d for d in self.cli(extra_env={'P23_ROOT_COMPOSITION':str(alternate)})))
        self.index.pop('root_composition_sha256')
        (self.evidence/'index.json').write_text(json.dumps(self.index))
        self.assertTrue(any('SHA mismatch' in d for d in self.cli()))


class ActorNonoverlapTests(unittest.TestCase):
    def report(self, second_at):
        return dict(actor_drawings=[
            dict(key='guest/a', at=[0, 0], scale=1, mode='idle', frame=0, flip=False),
            dict(key='staff/b', at=second_at, scale=1, mode='idle', frame=0, flip=False)], cabins=[])

    def test_edge_contact_is_not_overlap(self):
        for at in ([6, 0], [0, 12], [-6, 0], [0, -12]):
            with self.subTest(at=at):
                check.validate_actor_nonoverlap(self.report(at))

    def test_overlap_ignores_asserted_verdicts_and_offscreen_labels(self):
        report = self.report([5, 11])
        for actor in report['actor_drawings']:
            actor.update(mechanical_pass=True, eligibility='offscreen', rect=[999, 999, 1, 1])
        with self.assertRaisesRegex(ValueError, 'actor-actor core overlap: guest/a / staff/b'):
            check.validate_actor_nonoverlap(report)

    def test_cabin_passenger_is_checked_against_floor_actor(self):
        report = self.report([0, 0])
        report['cabins'] = [dict(passengers=[report['actor_drawings'].pop()])]
        with self.assertRaisesRegex(ValueError, 'actor-actor core overlap'):
            check.validate_actor_nonoverlap(report)

    def test_invalid_footprints_are_not_silently_skipped(self):
        for scale in (0, -1, float('nan'), float('inf'), True):
            report = self.report([100, 100])
            report['actor_drawings'][0]['scale'] = scale
            with self.subTest(scale=scale), self.assertRaisesRegex(ValueError, 'invalid actor footprint'):
                check.validate_actor_nonoverlap(report)

    def test_r13_false_row_11_overlaps_staff_row_29(self):
        fixture = json.loads((check.ROOT / 'tests/fixtures/p23-03/r13_actor_overlap.json').read_text())
        self.assertEqual(fixture['independent_judgment'], [11, False, False, False])
        with self.assertRaisesRegex(ValueError,
                'actor-actor core overlap: guest/guest-auto-00000041 / staff/staff_gina'):
            check.validate_actor_nonoverlap(fixture['report_inputs'])
        # Exercise the actual offline evidence gate, before any audit assertions.
        with self.assertRaisesRegex(ValueError, 'actor-actor core overlap'):
            check.validate_drawing_rows(fixture['report_inputs'], {'rows': []})

    def test_overlapping_full_boxes_with_disjoint_cores_pass(self):
        # Both 16x18 boxes overlap by 9x18, but their six-pixel cores do not.
        check.validate_actor_nonoverlap(self.report([7, 0]))

    def test_walk_bob_is_part_of_core(self):
        report = self.report([0, 12])
        report['actor_drawings'][1].update(mode='walk', frame=2)
        with self.assertRaisesRegex(ValueError, 'actor-actor core overlap'):
            check.validate_actor_nonoverlap(report)
        report['actor_drawings'][1]['frame'] = 3
        check.validate_actor_nonoverlap(report)

    def test_flipped_core_and_lying_breathing_bounds(self):
        actor = self.report([0, 0])['actor_drawings'][0]
        actor.update(flip=True, scale=3)
        self.assertEqual(check.actor_identity_core(actor), [12, 3, 18, 36])
        actor.update(mode='standard_room_sleep', frame=2)
        self.assertEqual(check.actor_identity_core(actor), [3, 30, 39, 18])
        actor['frame'] = 5
        self.assertEqual(check.actor_identity_core(actor), [3, 33, 39, 15])
        actor.update(mode='collapsed', flip=False)
        self.assertEqual(check.actor_identity_core(actor), [6, 33, 39, 15])

    def test_missing_pose_is_rejected(self):
        report = self.report([100, 0])
        del report['actor_drawings'][0]['frame']
        with self.assertRaisesRegex(ValueError, 'missing actor pose'):
            check.validate_actor_nonoverlap(report)

    def test_duplicate_actor_is_rejected(self):
        report = self.report([100, 0])
        report['actor_drawings'][1]['key'] = 'guest/a'
        with self.assertRaisesRegex(ValueError, 'duplicate actor drawing'):
            check.validate_actor_nonoverlap(report)

    def test_forged_transit_flag_does_not_exempt_overlap(self):
        report = self.report([0, 0])
        report['actor_drawings'][1]['in_transit'] = True
        with self.assertRaisesRegex(ValueError, 'actor-actor core overlap'):
            check.validate_actor_nonoverlap(report)

    def test_core_table_requires_exact_art_source(self):
        with mock.patch.object(check, 'PIXEL_PEOPLE_CORE_SHA256', '0' * 64):
            with self.assertRaisesRegex(ValueError, 'PixelPeople core source SHA256'):
                check.validate_actor_nonoverlap(self.report([100, 0]))

    def test_recorded_service_region_and_logical_cell_have_no_common_area(self):
        # Diagnostic of the unresolved simultaneous-bounds requirement, not
        # permission to hide these occupants or bypass containment checks.
        fixture = json.loads((check.ROOT / 'tests/fixtures/p23-03/r13_service_region_gap.json').read_text())
        for guest in fixture['guests']:
            traffic = next(r for r in fixture['traffic'] if r['cell'] == guest['cell'])
            x, y, w, h = traffic['rect']
            cell = [x - 2, y - 2, w + 4, h + 4]  # get_cell_rect(...).grow(-2)
            pixel = cell[2] / 16
            x, y, w, h = fixture['room_rects'][guest['target']]
            region = [x + pixel, y + pixel, w - 2 * pixel, h - 2 * pixel]
            with self.subTest(guest=guest['id']):
                self.assertLessEqual(min(cell[0] + cell[2], region[0] + region[2])
                                     - max(cell[0], region[0]), 0)


class LifeOverlayTests(unittest.TestCase):
    def report(self):
        return synthetic_repair28_fields(dict(observation_serial=7, overlay_consumed_serial=7,
                    overlay_passes=['traffic', 'actors'], overlay_child_indices=[0, 1, 2],
                    actor_drawings=[dict(key='guest/a', at=[20, 30], scale=3,
                                         mode='idle', frame=0, flip=False)], cabins=[],
                    traffic=[dict(rect=[2, 2, 60, 92], band_rect=[9.2, 85.5, 45.6, 5],
                                  traffic=dict(current_density=1, temporary_block=False))]))

    def test_front_and_back_valid_consumed_band(self):
        for layer in (0, 1):
            report = self.report()
            report['layer'] = layer
            check.validate_life_overlay(report)

    def test_stale_serial_fails(self):
        report = self.report()
        report['overlay_consumed_serial'] = 6
        with self.assertRaisesRegex(ValueError, 'stale LifeOverlay serial'):
            check.validate_life_overlay(report)

    def test_band_after_actors_fails(self):
        report = self.report()
        report['overlay_passes'].reverse()
        with self.assertRaisesRegex(ValueError, 'LifeOverlay pass order'):
            check.validate_life_overlay(report)

    def test_overlay_below_inactive_art_fails(self):
        report = self.report()
        report['overlay_child_indices'] = [1, 0, 2]
        with self.assertRaisesRegex(ValueError, 'LifeOverlay child order'):
            check.validate_life_overlay(report)

    def test_core_crossed_by_actual_band_fails(self):
        report = self.report()
        report['actor_drawings'][0]['at'][1] = 60
        with self.assertRaisesRegex(ValueError, 'floor actor core over traffic band'):
            check.validate_life_overlay(report)

    def test_forged_band_geometry_fails(self):
        report = self.report()
        report['traffic'][0]['band_rect'][1] = 50
        with self.assertRaisesRegex(ValueError, 'traffic band rectangle'):
            check.validate_life_overlay(report)

    def test_report_cannot_excuse_wrong_source_order(self):
        source = (check.ROOT / 'scenes/main/world_preview.gd').read_text()
        # Claimed pass order stays valid; actual callback source is inverted.
        broken = source.replace('_draw_congestion_overlay(_life_overlay)', 'SWAP')
        broken = broken.replace('_draw_actors(_life_overlay)', '_draw_congestion_overlay(_life_overlay)')
        broken = broken.replace('SWAP', '_draw_actors(_life_overlay)')
        with mock.patch.object(Path, 'read_text', return_value=broken):
            with self.assertRaisesRegex(ValueError, 'LifeOverlay source pass order'):
                check.validate_life_overlay(self.report())


class ProjectionBandTests(unittest.TestCase):
    def setUp(self):
        fixture = json.loads((check.ROOT / 'tests/fixtures/p23-03/repair28_frame180_projection.json').read_text())
        self.report, self.snapshot = fixture['report'], fixture['snapshot']
        synthetic_repair28_fields(self.report, self.snapshot)

    def occupied(self):
        for row in self.report['traffic']:
            if row['cell'] in ([0, 1, 0], [1, 1, 0]):
                self.assertEqual(row['traffic']['current_density'], 0)
                row['traffic']['current_density'] = 1

    def test_counterfactual_actual_frame180_density_one(self):
        """COUNTERFACTUAL density substitution; never occupied-band engine evidence."""
        self.occupied()
        for key in ('guest/p23-018', 'guest/p23-019'):
            self.assertTrue(any(a['key'] == key for c in self.report['cabins'] for a in c['passengers']))
        check.validate_life_overlay(self.report, snapshot=self.snapshot)

    def test_cabin_passenger_union_fully_occludes_band(self):
        self.occupied()
        row = next(r for r in self.report['traffic'] if r['cell'] == [0, 1, 0])
        x, y, w, h = row['band_rect']
        # Each core covers only part; their touching intervals cover the union.
        for i, actor in enumerate(self.report['cabins'][0]['passengers']):
            actor.update(scale=2, flip=False, mode='idle', at=[x+i*12-12, y-2])
        with self.assertRaisesRegex(ValueError, 'traffic band fully occluded:'):
            check.validate_life_overlay(self.report, snapshot=self.snapshot)

    def test_spoofed_cabin_rectangle_fails_raw_recomputation(self):
        self.occupied()
        self.report['cabins'][0]['rect'][1] += 1
        with self.assertRaisesRegex(ValueError, 'cabin/rect'):
            check.validate_life_overlay(self.report, snapshot=self.snapshot)

    def test_flag_only_cabin_projection_without_raw_membership_fails(self):
        self.occupied()
        for cabin in self.report['cabins']:
            for actor in cabin['passengers']:
                actor['key'] = 'guest/spoof-' + actor['key']
                actor['riding'] = True
        with self.assertRaisesRegex(ValueError, 'floor actor core over traffic band'):
            check.validate_life_overlay(self.report, snapshot=self.snapshot)

    def test_no_raw_state_never_exempts_cabin_projection(self):
        self.occupied()
        with self.assertRaisesRegex(ValueError, 'floor actor core over traffic band'):
            check.validate_life_overlay(self.report)

    def test_floor_copy_of_raw_cabin_member_keeps_zero_intersection(self):
        self.occupied()
        self.report['actor_drawings'].append(copy.deepcopy(self.report['cabins'][0]['passengers'][0]))
        with self.assertRaisesRegex(ValueError, 'non-passenger inside cabin'):
            check.validate_life_overlay(self.report, snapshot=self.snapshot)

    def test_blocked_band_uses_same_projection_and_legibility_rules(self):
        row = next(r for r in self.report['traffic'] if r['cell'] == [0, 1, 0])
        row['traffic']['temporary_block'] = True
        check.validate_life_overlay(self.report, snapshot=self.snapshot)
        x, y, w, h = row['band_rect']
        for i, actor in enumerate(self.report['cabins'][0]['passengers']):
            actor.update(scale=2, flip=False, mode='idle', at=[x+i*12-12, y-2])
        with self.assertRaisesRegex(ValueError, 'traffic band fully occluded:'):
            check.validate_life_overlay(self.report, snapshot=self.snapshot)

    def test_flag_only_carried_projection_without_binding_fails(self):
        self.occupied()
        actor = copy.deepcopy(self.report['cabins'][0]['passengers'][0])
        actor.update(key='staff/spoof', staff=True, riding=False, region='carried')
        self.report['actor_drawings'].append(actor)
        self.report['carried_patients'] = {'spoof': 'carrier'}
        with self.assertRaisesRegex(ValueError, 'non-passenger inside cabin: staff/spoof'):
            check.validate_life_overlay(self.report, snapshot=self.snapshot)

    def test_carried_projection_requires_original_binding_tracks_and_fit(self):
        fixture = AnchoredGeometryTests()
        fixture.setUp()
        fixture.person('staff/carrier', [0, 0, 0], 'AVAILABLE')
        fixture.person('staff/patient', [0, 0, 0], 'INCAPACITATED')
        fixture.runtime['staff']['tasks'] = [dict(task_type='rescue_transport', state='MOVING',
            assigned_staff_id='carrier', target_entity_id='patient', rescue_facility_id='')]
        fixture.report['carried_patients'] = {'patient': 'carrier'}
        fixture.drawing('staff/carrier', [108, 214])
        fixture.drawing('staff/patient', [108, 166], 'collapsed')
        for key, value in LifeOverlayTests().report().items():
            if key.startswith('overlay_') or key == 'observation_serial':
                fixture.report[key] = value
        # Synthetic band crosses the raised patient, with uncovered span.
        fixture.report['traffic'] = [dict(cell=[0, 1, 0], rect=[78, 100.5, 60, 92],
            band_rect=[85.2, 184, 45.6, 5], traffic=dict(current_density=1, temporary_block=False))]
        synthetic_repair28_fields(fixture.report)
        check.validate_life_overlay(fixture.report, snapshot=fixture.snapshot)
        for mutation in ('binding', 'track', 'position'):
            report, snapshot = copy.deepcopy(fixture.report), copy.deepcopy(fixture.snapshot)
            if mutation == 'binding': snapshot['systems']['runtime']['staff']['tasks'] = []
            if mutation == 'track': report['actor_tracks']['staff/carrier']['cell'] = [1, 0, 0]
            if mutation == 'position': report['actor_drawings'][1]['at'][1] += 1
            with self.subTest(mutation=mutation), self.assertRaises(ValueError):
                check.validate_life_overlay(report, snapshot=snapshot)


class SetupBudgetTests(unittest.TestCase):
    def setUp(self):
        self.source = (check.ROOT / 'tests/fixtures/p23-03/observation.gd').read_text()

    def test_all_planned_spends_enter_common_needed_and_deltas(self):
        self.assertEqual(check.check_setup_budget_source(self.source),
                         {'construction', 'signing', 'purchase'})

    def test_missing_spend_category_fails_before_os_replay(self):
        expressions = {
            'construction': 'needed[key] += int(definition.construction_cost.get(key, 0))',
            'signing': 'needed[&"gold"] += signing_total',
            'purchase': 'needed[&"gold"] += purchase_total',
        }
        for category, expression in expressions.items():
            with self.subTest(category=category), self.assertRaisesRegex(ValueError, category):
                check.check_setup_budget_source(self.source.replace(expression, 'pass'))

    def test_missing_delta_coverage_fails(self):
        with self.assertRaisesRegex(ValueError, 'deltas'):
            check.check_setup_budget_source(self.source.replace(
                'deltas[key] = needed[key] - state.get_resource(key)', 'pass'))


class QueuedElevatorTests(unittest.TestCase):
    def setUp(self):
        fixture = json.loads((check.ROOT / 'tests/fixtures/p23-03/queued_elevator.json').read_text())
        self.state = {'systems': {'runtime': {'elevators': fixture['saved']}}}

    def test_queued_trip_without_cabin_seat_uses_config_shaft(self):
        saved = self.state['systems']['runtime']['elevators']
        self.assertNotIn('shaft_id', saved['shafts'][0])
        self.assertEqual(saved['shafts'][0]['boarded'], [])
        report = check.elevator_report(self.state)
        self.assertEqual(report['shafts'][0]['waiting_count'], 1)
        self.assertEqual(report['shafts'][0]['passenger_count'], 0)
        self.assertAlmostEqual(report['shafts'][0]['visual_floor'], .9)
        self.assertEqual(report['trips'], saved['trips'])

    def test_tampered_shaft_attribution_fails(self):
        self.state['systems']['runtime']['elevators']['trips'][0]['shaft_id'] = 'wrong.shaft'
        with self.assertRaisesRegex(ValueError, 'shaft attribution'):
            check.elevator_report(self.state)

    def test_wrong_waiting_count_fails_whole_report_comparison(self):
        expected = check.elevator_report(self.state)
        tampered = copy.deepcopy(expected)
        tampered['shafts'][0]['waiting_count'] = 0
        directory = mock.MagicMock()
        manifest = dict(label='queued-fixture', pid=123, state='state',
                        state_after='state', report='report')
        directory.__truediv__.return_value.open.return_value = io.StringIO(json.dumps(manifest)+'\n')
        # Exercise the actual stream comparison; isolate unrelated actor fields.
        with mock.patch.object(check, 'read_object', side_effect=[self.state, self.state,
                                                                {'elevators': tampered}]), \
             mock.patch.object(check, 'compare_actor_rows', return_value=[]):
            with self.assertRaisesRegex(ValueError, 'typed elevator report mismatch'):
                check.inspect_stream(directory)

    def test_extra_and_missing_trip_keys_fail(self):
        for mutation in ('extra', 'missing'):
            with self.subTest(mutation=mutation):
                state = copy.deepcopy(self.state)
                trip = state['systems']['runtime']['elevators']['trips'][0]
                if mutation == 'extra':
                    trip['unexpected'] = True
                else:
                    del trip['shaft_id']
                with self.assertRaisesRegex(ValueError, 'exact elevator trip keys'):
                    check.elevator_report(state)


class SnapshotCellTests(unittest.TestCase):
    def test_layout_arrival_compares_serialized_cell_arrays(self):
        source = (check.ROOT / 'tests/fixtures/p23-03/layout_comparison.gd').read_text()
        serializer = (check.ROOT / 'scripts/guest/guest_runtime.gd').read_text()
        self.assertIn('&"current_cell": _cell_to_array(current_cell)', serializer)
        goal = check.re.search(r'const GOAL := Vector3i\(([^)]+)\)', source)
        self.assertIsNotNone(goal)
        coordinates = ast.literal_eval('[' + goal.group(1) + ']')
        comparison = check.re.search(r'if guest\[&"current_cell"\] != (.+?) or finished\.has', source)
        self.assertIsNotNone(comparison)
        expression = comparison.group(1)
        self.assertEqual(expression, '[GOAL.x, GOAL.y, GOAL.z]')
        for axis, value in zip('xyz', coordinates):
            expression = expression.replace('GOAL.' + axis, str(value))
        target = ast.literal_eval(expression)
        self.assertEqual(json.loads('[2, 1, 0]'), target)
        for cell in ([2, 0, 0], [3, 1, 0], [2, 1, 1]):
            self.assertNotEqual(cell, target)


class ImageTests(unittest.TestCase):
    def test_collector_conversion_is_bound_to_each_cua_screenshot(self):
        runner = check.module('p23_unit_collector', check.ROOT/'tools/run_p23_life_observation.py')
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            pair = root/'os-pair/A'
            request_dir = pair/'writer-request-0001'
            request_dir.mkdir(parents=True)
            request = request_dir/'request.json'
            request.write_text(json.dumps(dict(sequence=1, target_pid=123)))
            raw = request_dir/'raw.jpg'
            raw.write_bytes(b'\xff\xd8unit-only\xff\xd9')
            shot = dict(path=str(raw.relative_to(root)), raw_sha256=check.sha(raw), tool_call_id='unit-shot-call')
            receipt = request_dir/'cua.json'
            receipt.write_text(json.dumps(dict(screenshots=[shot])))
            ack = request_dir/'ack.json'
            ack.write_text(json.dumps(dict(request_sha256=check.sha(request), input_receipt=str(receipt.relative_to(root)))))
            (pair/'writer-requests.json').write_text(json.dumps(dict(requests=[str(ack.relative_to(root))])))
            def convert(raw_path, png_path, cua, receipt_path):
                self.assertEqual(cua, shot)
                self.assertEqual(Path(raw_path).read_bytes(), b'\xff\xd8unit-only\xff\xd9')
                Path(png_path).write_bytes(self.png())
                Path(receipt_path).write_text('{}')
            with mock.patch.object(runner, 'module', return_value=check), \
                 mock.patch.object(check, 'image_kind', return_value=('jpeg', 2, 2)), \
                 mock.patch.object(runner, 'convert_screenshot', side_effect=convert) as converter:
                output = runner.collect_screenshots(root, pair, 'writer')
                converter.assert_called_once()
            rows = json.loads((root/output).read_text())
            self.assertEqual(rows[0]['tool_call_id'], shot['tool_call_id'])
            self.assertEqual(rows[0]['raw_sha256'], check.sha(raw))
            self.assertEqual(rows[0]['png_sha256'], check.sha(root/rows[0]['png_path']))
            self.assertTrue(rows[0]['conversion'])

    @staticmethod
    def png(mode=0, channels=3):
        def chunk(kind, body):
            return struct.pack('>I', len(body)) + kind + body + struct.pack('>I', zlib.crc32(kind + body))
        rows = [bytes([10, 20, 30, 40][:channels] * 2),
                bytes([50, 60, 70, 80][:channels] * 2)]
        filtered = bytearray()
        prior = bytes(2 * channels)
        for row in rows:
            filtered.append(mode)
            for x, value in enumerate(row):
                left, up, corner = (row[x-channels] if x >= channels else 0,
                                    prior[x], prior[x-channels] if x >= channels else 0)
                p = left + up - corner
                distances = [abs(p-left), abs(p-up), abs(p-corner)]
                predictors = [0, left, up, (left+up)//2,
                              [left, up, corner][distances.index(min(distances))]]
                filtered.append((value-predictors[mode]) % 256)
            prior = row
        header = struct.pack('>IIBBBBB', 2, 2, 8, 2 if channels == 3 else 6, 0, 0, 0)
        return (b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', header)
                + chunk(b'IDAT', zlib.compress(filtered)) + chunk(b'IEND', b''))

    def test_all_png_filters_rgb_and_rgba(self):
        for mode in range(5):
            for channels in (3, 4):
                with self.subTest(mode=mode, channels=channels):
                    w, h, rows = check.png_pixels(self.png(mode, channels))
                    self.assertEqual((w, h), (2, 2))
                    self.assertEqual(rows[0][0], tuple([10, 20, 30, 40][:channels]))
                    self.assertEqual(rows[1][1], tuple([50, 60, 70, 80][:channels]))

    def test_crc_and_trailing_bytes_rejected(self):
        data = bytearray(self.png())
        data[29] ^= 1
        with self.assertRaisesRegex(ValueError, 'CRC mismatch'):
            check.png_pixels(bytes(data))
        with self.assertRaisesRegex(ValueError, 'invalid IEND'):
            check.png_pixels(self.png() + b'extra')

    def conversion(self):
        raw, png = b'\xff\xd8synthetic-unit-test\xff\xd9', self.png()
        digest = lambda b: hashlib.sha256(b).hexdigest()
        rec = dict(raw_path='/raw.png', raw_sha256=digest(raw), raw_kind='jpeg',
                   tool_call_id='unit-test-call', converter='/usr/bin/sips', converter_sha256='a'*64,
                   argv=['/usr/bin/sips', '-s', 'format', 'png', '/raw.png', '--out', '/decoded.png'],
                   exit_code=0, stdout_sha256=digest(b''), stderr_sha256=digest(b''),
                   png_path='/decoded.png', png_sha256=digest(png), width=2, height=2)
        cua = dict(raw_sha256=digest(raw), tool_call_id='unit-test-call')
        return rec, cua, {'/raw.png': raw, '/decoded.png': png}

    def test_conversion_chain_and_specific_forgery_deficits(self):
        rec, cua, files = self.conversion()
        helper = mock.Mock()
        helper.jpeg_dimensions.return_value = (2, 2)
        with mock.patch.object(Path, 'read_bytes', lambda p: files[str(p)]), \
             mock.patch.object(check, 'module', return_value=helper):
            self.assertEqual(check.inspect_conversion(rec, cua, 'a'*64)[:2], (2, 2))
            for key, value, deficit in [('converter_sha256', 'b'*64, 'converter SHA mismatch'),
                                         ('argv', [], 'converter argv mismatch'),
                                         ('raw_sha256', 'b'*64, 'CUA raw hash/call mismatch'),
                                         ('width', 3, 'conversion dimensions mismatch')]:
                with self.subTest(key=key):
                    bad = dict(rec, **{key: value})
                    with self.assertRaisesRegex(ValueError, deficit):
                        check.inspect_conversion(bad, cua, 'a'*64)
            with self.assertRaisesRegex(ValueError, 'JPEG screenshot requires conversion record'):
                check.screenshot_pixels('/raw.png')


class RestAuditTests(unittest.TestCase):
    def fixture(self):
        member = dict(staff_id='s', state='OFF_DUTY', health=100, current_task_id='', current_cell=[0, 0, 0])
        facilities = [dict(instance_id=fid, definition_id='staff_waiting_room', origin=[0, 0, 0],
                           occupied_cells=[], state='OPEN', suspension_reason='') for fid in ('a', 'b')]
        runtime = dict(staff=dict(staff=[member]), facilities=dict(instances=facilities),
                       defense=dict(raids=[]), events=dict(actor_locks={}), elevators=dict(trips=[]),
                       path=dict(cell_traffic=[dict(cell=[x, 0, 0]) for x in (-1, 0, 1)]),
                       world=dict(bounds=dict(min_x=-1, max_x=1, min_y=0, max_y=0), occupancies=[]))
        routes = [dict(cell=[x, 0, 0], path=[[0, 0, 0], [x, 0, 0]] if x else [[0, 0, 0]],
                       edges=[dict(**{'from': [0, 0, 0], 'to': [x, 0, 0]}, total_cost=0)] if x else [],
                       calculate_path_cost=0) for x in (-1, 0, 1)]
        observed = []
        for f in facilities:
            observed.append({k: f[k] for k in ('instance_id', 'definition_id', 'origin', 'occupied_cells')})
            observed[-1].update(operation_state='OPERATING', probes=[dict(cell=[x, 0, 0], has_cell=True) for x in (0, -1, 1)],
                                ordered=[[x, 0, 0] for x in (-1, 0, 1)], routes=copy.deepcopy(routes))
        selected = dict(reachable=True, resting=True, cell=[-1, 0, 0], facility_id='a', label='待機室で休憩中')
        audit = dict(path_service_sha256=check.sha(check.ROOT/'scripts/world/path_service.gd'),
                     staff=[dict(member, defense_memberships=[], actor_lock=False, in_transit=False,
                                 facilities=observed, comparison_only=selected)])
        report = dict(resting_staff={'s': 'a'}, actor_drawings=[], layer=0, carried_patients={}, rooms={},
                      viewport=[1440, 810], zoom=1, pan=[0, 0])
        return dict(systems=dict(runtime=runtime)), report, audit

    def test_selection_ties_and_nonmutation(self):
        state, report, audit = self.fixture()
        before = copy.deepcopy((state, report, audit))
        self.assertEqual(check.compare_rest(state, report, audit)['s']['cell'], [-1, 0, 0])
        self.assertEqual((state, report, audit), before)

    def test_rest_forgeries(self):
        mutations = {
            'selection': lambda a: a['staff'][0]['comparison_only'].update(facility_id='b'),
            'tie': lambda a: a['staff'][0]['comparison_only'].update(cell=[0, 0, 0]),
            'lock': lambda a: a['staff'][0].update(actor_lock=True),
            'transit': lambda a: a['staff'][0].update(in_transit=True),
            'label': lambda a: a['staff'][0]['comparison_only'].update(label='宿舎で休憩中'),
            'cost': lambda a: a['staff'][0]['facilities'][0]['routes'][0]['edges'][0].update(total_cost=9),
            'facility_order': lambda a: a['staff'][0]['facilities'].reverse(),
            'candidate_order': lambda a: a['staff'][0]['facilities'][0]['ordered'].reverse(),
            'empty_edge': lambda a: a['staff'][0]['facilities'][0]['routes'][0]['edges'][0].clear(),
        }
        for name, mutate in mutations.items():
            with self.subTest(name=name):
                state, report, audit = self.fixture()
                mutate(audit)
                with self.assertRaises(ValueError):
                    check.compare_rest(state, report, audit)

    def test_ineligible_and_transit_recomputed(self):
        state, report, audit = self.fixture()
        state['systems']['runtime']['events']['actor_locks']['s'] = 'event'
        audit['staff'][0].update(actor_lock=True, comparison_only={})
        report['resting_staff'] = {}
        self.assertEqual(check.compare_rest(state, report, audit), {'s': {}})
        state, report, audit = self.fixture()
        state['systems']['runtime']['elevators']['trips'] = [dict(actor_id='staff:s')]
        audit['staff'][0]['in_transit'] = True
        audit['staff'][0]['comparison_only'].update(resting=False, label='休憩先へ移動中')
        report['resting_staff'] = {}
        self.assertFalse(check.compare_rest(state, report, audit)['s']['resting'])


class InfrastructureTests(unittest.TestCase):
    def test_matrix_requires_each_condition_and_each_category(self):
        executed = {c: {k: set(v) for k, v in categories.items()} for c, categories in check.REQUIRED_CHECKS.items()}
        self.assertTrue(all(r['status'] == 'passed' for r in check.condition_matrix(executed)))
        for condition in check.CONDITIONS:
            for category in check.CATEGORIES:
                with self.subTest(condition=condition, category=category):
                    broken = copy.deepcopy(executed)
                    broken[condition][category].clear()
                    matrix = check.condition_matrix(broken)
                    self.assertEqual(next(r for r in matrix if r['condition'] == condition)['status'], 'pending')
                    self.assertIn(category, check.BLOCKERS(broken)[condition])

    def test_distinct_pid_ack_and_formal_hash_negatives(self):
        members = {name: 'a'*64 for name in ('world.bin', 'manifest.json', 'thumbnail.png', 'journal.log')}
        writer = dict(actual_pid=1)
        reader = dict(actual_pid=2, formal_input={k: dict(sha256=v) for k, v in members.items()})
        saved = dict(pid=1, members=members)
        records = [dict(request=dict(action='load_menu', target_pid=2, sequence=1, requested_ticks_usec=10, requested_unix_ns=90),
                        ack=dict(target_pid=2, sequence=1, request_sha256='b'*64, timestamp_ns=110),
                        observed=dict(pid=2, sequence=1, request_sha256='b'*64, ack_observed_ticks_usec=20, ack_observed_unix_ns=120, status='手動セーブを読み込みました。'),
                        cua=dict(target_pid=2, method='native_os', tool_call_id='actual-call', timestamp_ns=100),
                        request_sha256='b'*64)]
        check.inspect_resume(writer, reader, saved, records, members)
        for change in ('pid', 'ack', 'time', 'hash', 'id'):
            with self.subTest(change=change):
                r, q = copy.deepcopy(reader), copy.deepcopy(records)
                if change == 'pid': r['actual_pid'] = 1
                if change == 'ack': q.clear()
                if change == 'time': q[0]['observed']['ack_observed_ticks_usec'] = 9
                if change == 'hash': r['formal_input']['world.bin']['sha256'] = 'c'*64
                if change == 'id': q[0]['ack']['sequence'] = 2
                with self.assertRaises(ValueError): check.inspect_resume(writer, r, saved, q, members)

    def test_interpolation_history_tampering(self):
        state = dict(systems=dict(runtime=dict(guests=dict(guests=[]), staff=dict(staff=[]),
                     elevators=dict(shafts=[], trips=[], current_minute=0))))
        report = dict(animation_time=0.0, paused=True, layer=0, zoom=1, pan=[0,0], viewport=[100,100],
                      actor_tracks={}, elevator_tracks={}, actor_drawings=[], cabins=[])
        audit = dict(frames=[dict(frame=1, delta=.01, scaled_delta=0, display=copy.deepcopy(report)),
                             dict(frame=2, delta=.01, scaled_delta=0, display=copy.deepcopy(report))])
        check.compare_motion(state, report, audit)
        for change in ('delta', 'frame', 'decreasing_frame', 'no_frames', 'one_frame', 'display', 'report_binding'):
            with self.subTest(change=change):
                bad = copy.deepcopy(audit)
                if change == 'delta': bad['frames'][1]['scaled_delta'] = 1
                if change == 'frame': bad['frames'][1]['frame'] = 1
                if change == 'decreasing_frame': bad['frames'][1]['frame'] = 0
                if change == 'no_frames': bad['frames'].clear()
                if change == 'one_frame': bad['frames'].pop()
                if change == 'report_binding': bad['frames'][1]['display']['zoom'] = 2
                if change == 'display': bad['frames'][1]['display']['animation_time'] = 1
                with self.assertRaises(ValueError): check.compare_motion(state, report, bad)

    def test_verify_only_missing_index_is_a_concrete_deficit(self):
        with mock.patch.dict(check.os.environ, {'BUILD_OUTPUT_DIR': '/missing-p23-proof'}, clear=True):
            with contextlib.redirect_stdout(io.StringIO()) as output:
                self.assertEqual(check.main(['--verify-only']), 1)
        result = json.loads(output.getvalue())
        self.assertEqual(result['integrity'], 'fail')
        self.assertEqual(result['acceptance'], 'pending')
        self.assertIn('index.json', result['deficits'][0])
        self.assertNotIn('blocked', result['deficits'][0])

    def test_verify_only_requires_build_output(self):
        with mock.patch.dict(check.os.environ, {}, clear=True):
            with contextlib.redirect_stderr(io.StringIO()) as output:
                self.assertEqual(check.main(['--verify-only']), 2)
        self.assertIn('BUILD_OUTPUT_DIR is required', output.getvalue())

    def test_normative_plan_change_invalidates_but_progress_does_not(self):
        root = check.ROOT
        text = (root / check.PLAN).read_text()
        original = check.input_map(root)
        progress = text.splitlines(keepends=True)
        progress[2] = 'Progress-only test mutation\n'
        with mock.patch.object(Path, 'read_text', return_value=''.join(progress)):
            self.assertEqual(check.input_map(root), original)
        changed = text.replace('同需要の動線改善を観察できる。', 'altered normative acceptance')
        self.assertNotEqual(changed, text)
        with mock.patch.object(Path, 'read_text', return_value=changed):
            self.assertNotEqual(check.input_map(root), original)
        self.assertNotIn(check.PLAN, original)
        self.assertIn('full_plan_sha256', check.input_provenance(root))
        for path in (root / 'tests/fixtures/p23-03').rglob('*'):
            if path.is_file():
                self.assertIn(path.relative_to(root).as_posix(), original)

    def test_os_process_guards_reject_engine_profile_and_source_drift(self):
        source = {'source.gd': 'a' * 64}
        receipt = dict(engine_ended=True, passed=True, exit_code=0, error=None,
                       guard_errors=[], cleanup_errors=[], default_before={}, default_after={},
                       stage_before={}, stage_after={}, source_map=source,
                       source_after={'source.gd': {'sha256': 'a' * 64}},
                       owned_cleanup=[{'engine_ended': True}], ordinary_owner_proof={'present': True})
        receipt.update(spec_sha256=check.SPEC_SHA, inputs={'spec': 'a'*64}, inputs_after={'spec': 'a'*64},
                       environment={'P23_LAYOUT': 'A'}, environment_after={'P23_LAYOUT': 'A'}, actual_pid=123,
                       profile='/private/test', profile_name='test', profile_before={'existed': False, 'files': {}}, profile_after={})
        for field in ('engine_sha256', 'settings_sha256', 'environment_sha256'):
            receipt[field] = receipt[field+'_after'] = 'a'*64
        check.inspect_os_receipt(receipt, source)
        for field, value in [('engine_ended', False), ('default_after', {'changed': True}),
                             ('source_after', {'source.gd': {'sha256': 'b' * 64}}),
                             ('stage_after', {'changed': True}), ('owned_cleanup', []),
                             ('environment_after', {}), ('profile_before', {}),
                             ('engine_sha256_after', 'b'*64), ('spec_sha256', 'b'*64)]:
            with self.subTest(field=field):
                altered = copy.deepcopy(receipt)
                altered[field] = value
                with self.assertRaises(ValueError):
                    check.inspect_os_receipt(altered, source)


class SemanticTests(unittest.TestCase):
    def setUp(self):
        self.guest = dict(guest_id='g', state='QUEUING', current_cell=[1, 0, 0],
                          target_facility_id='f', current_service_id='', queue_request_id='q')
        self.staff = dict(staff_id='s', state='WORKING', current_cell=[2, 0, 1],
                          current_task_id='t', assigned_facility_id='f', role_id='porter',
                          carried_inventory_id='cargo')
        self.task = dict(task_id='t', state='WORKING', task_type='delivery',
                         assigned_staff_id='s', target_entity_id='request', rescue_facility_id='')
        self.snapshot = {'systems': {'runtime': {'guests': {'guests': [self.guest]},
                         'staff': {'staff': [self.staff], 'tasks': [self.task]}}}}
        self.report = dict(guests=[dict(id='g', state='QUEUING', cell=[1, 0, 0], target='f', service='', queue='q')],
                           staff=[dict(id='s', state='WORKING', cell=[2, 0, 1], task='t', facility='f',
                                       role='porter', carried_inventory_id='cargo')],
                           tasks=[dict(id='t', state='WORKING', type='delivery', staff='s',
                                       target='request', rescue_facility='')])

    def test_real_schema_projection(self):
        self.assertEqual(check.compare_actor_rows(self.snapshot, self.report), ['guests', 'staff', 'tasks'])

    def test_changed_state_position_target_task_and_goods_are_rejected(self):
        for category, field, replacement in [('guests', 'cell', [8, 0, 0]), ('guests', 'target', 'other'),
                                             ('guests', 'state', 'MOVING'), ('staff', 'task', ''),
                                             ('staff', 'carried_inventory_id', ''), ('tasks', 'staff', 'other')]:
            with self.subTest(category=category, field=field):
                report = copy.deepcopy(self.report)
                report[category][0][field] = replacement
                with self.assertRaises(ValueError):
                    check.compare_actor_rows(self.snapshot, report)

    def test_missing_extra_and_boolean_only_rows_are_rejected(self):
        for bad in [[], self.report['guests'] * 2, [True]]:
            report = copy.deepcopy(self.report)
            report['guests'] = bad
            with self.assertRaises(ValueError):
                check.compare_actor_rows(self.snapshot, report)

    def test_unknown_enum_and_duplicate_id_are_rejected(self):
        self.guest['state'] = 'FICTIONAL'
        with self.assertRaises(ValueError):
            check.compare_actor_rows(self.snapshot, self.report)
        self.guest['state'] = 'QUEUING'
        self.snapshot['systems']['runtime']['guests']['guests'].append(copy.deepcopy(self.guest))
        with self.assertRaises(ValueError):
            check.compare_actor_rows(self.snapshot, self.report)

    def test_typed_rectangle_and_duplicate_vector_key(self):
        self.assertEqual(check.decode(dict(type='Rect2', text='Rect2(1, 2, 16, 24)')), [1, 2, 16, 24])
        with self.assertRaises(ValueError):
            check.decode(dict(type='Rect2', text='Rect2(1, nan, 16, 24)'))
        key = dict(type='Vector3i', text='Vector3i(1, 2, 0)')
        value = dict(type='int', text='1')
        with self.assertRaises(ValueError):
            check.decode(dict(type='Dictionary', entries=[[key, value], [key, value]]))

    def test_immutable_typed_input_is_not_modified(self):
        before = copy.deepcopy(self.snapshot)
        check.compare_actor_rows(self.snapshot, self.report)
        self.assertEqual(before, self.snapshot)


class PackingTests(unittest.TestCase):
    def test_scale_trial_rebuilds_vertical_groups(self):
        members = self.members(2)
        for member, y in zip(members, (118, 144)):
            member['row'].update(center=[108, y], at=[84, y-15])
            member['bounds'] = [80, 74, 20, 130]
        # At p=3 the vertical intervals intersect; at p=2 they separate.
        # Keeping the stale component would incorrectly force both down to p=1.
        rows, badges = check.fitted_actor_layout(members, 4, 96)
        self.assertEqual(badges, [])
        self.assertEqual([rows['guest/'+str(i)]['scale'] for i in range(2)], [2, 2])
        self.assertEqual([rows['guest/'+str(i)]['at'] for i in range(2)], [[76, 120], [76, 146]])
        check.validate_actor_nonoverlap(dict(actor_drawings=list(rows.values()), cabins=[]))

    def members(self, count=4):
        # R13 cell dimensions and the actual _actor_offset sequence; no transit.
        offsets = [0, 16, -16, 8, -8, 24, -24]
        return [dict(row=dict(key='guest/'+str(i), center=[108+offsets[i % 7], 118-4*(i % 2)],
                         at=[84+offsets[i % 7], 103-4*(i % 2)], scale=3,
                         mode='idle', frame=0, flip=False, state=3, cell=[1, 1, 0]),
                     bounds=[80, 74, 56, 88], transit=False) for i in range(count)]

    def test_four_movers_fit_at_two_not_three(self):
        rows, badges = check.fitted_actor_layout(self.members(), 4, 96)
        self.assertEqual(badges, [])
        self.assertEqual([rows['guest/'+str(i)]['scale'] for i in range(4)], [2]*4)
        self.assertEqual([rows['guest/'+str(i)]['at'] for i in range(4)],
                         [[88, 120], [112, 116], [76, 120], [100, 116]])
        check.validate_actor_nonoverlap(dict(actor_drawings=list(rows.values()), cabins=[]))

    def test_cross_kind_group_and_passenger(self):
        members = self.members()
        members[1]['row']['key'] = 'staff/worker'
        members[2]['row']['key'] = 'guest/passenger'
        members[2]['row']['scale'] = 2
        members[2]['row']['at'] = [76, 120]
        rows, badges = check.fitted_actor_layout(members, 4, 96)
        self.assertEqual(badges, [])
        self.assertEqual(len(rows), 4)
        check.validate_actor_nonoverlap(dict(actor_drawings=list(rows.values()), cabins=[]))

    def test_saturation_keeps_all_identities(self):
        rows, badges = check.fitted_actor_layout(self.members(12), 4, 96)
        self.assertEqual(len(rows), 9)
        self.assertEqual(len(badges), 1)
        self.assertEqual(len(badges[0]['members']), 12)
        self.assertEqual(badges[0]['text'], '+3')
        self.assertEqual(set(badges[0]['drawn_keys']), set(rows))
        self.assertTrue(all(r['scale'] == 1 for r in rows.values()))
        check.validate_actor_nonoverlap(dict(actor_drawings=list(rows.values()), cabins=[]))

    def test_each_actor_keeps_own_bounds(self):
        members = self.members(2)
        members[0]['bounds'] = [80, 74, 12, 88]
        members[1]['bounds'] = [124, 74, 12, 88]
        rows, badges = check.fitted_actor_layout(members, 4, 96)
        self.assertEqual(badges, [])
        self.assertEqual(check.actor_identity_core(rows['guest/0'])[0:3:2], [80, 12])
        self.assertEqual(check.actor_identity_core(rows['guest/1'])[0:3:2], [124, 12])

    def test_receiving_cafeteria_uses_r13_room_not_access_cell(self):
        fixture = json.loads((check.ROOT/'tests/fixtures/p23-03/r13_service_region_gap.json').read_text())
        guest = fixture['guests'][0]
        rect = fixture['room_rects'][guest['target']]
        anchor = check.room_slots(rect, 4, 'cafeteria')[0]
        center = [anchor[0], anchor[1]+12-48]
        member = dict(row=dict(key='guest/'+guest['id'], center=center,
                              at=[center[0]-24, center[1]-15], scale=3,
                              mode='eat', frame=0, flip=False, state=5, cell=guest['cell']),
                      bounds=[rect[0]+4, rect[1]+4, rect[2]-8, rect[3]-8], transit=False)
        rows, badges = check.fitted_actor_layout([member], 4, 96)
        self.assertFalse(badges)
        core = check.actor_identity_core(next(iter(rows.values())))
        self.assertGreaterEqual(core[0], 1031)
        self.assertEqual(next(iter(rows.values()))['cell'], [5, 0, 0])


class SpatialTests(unittest.TestCase):
    def setUp(self):
        self.runtime = dict(world=dict(bounds=dict(min_x=0, max_x=1, min_y=0, max_y=0),
                            walkable_cells=[[0, 0, 0], [1, 0, 0]], occupancies=[]),
                            path=dict(cell_traffic=[dict(cell=[0, 0, 0], base_capacity=2,
                                      current_density=3, dynamic_cost=0, temporary_block=False)]),
                            facilities=dict(instances=[], requests=[]), guests=dict(guests=[]),
                            staff=dict(staff=[], tasks=[]), logistics_requests=[],
                            elevators=dict(current_minute=0, shafts=[], trips=[]))
        self.snapshot = dict(systems=dict(runtime=self.runtime))
        self.report = dict(viewport=[184, 164], zoom=1, pan=[0, 0], layer=0,
                           traffic=[dict(cell=[0, 0, 0], rect=[78, 72, 12, 20],
                                    band_rect=[79.44, 90.4, 9.12, 3.2],
                                    traffic=dict(base_capacity=2, current_density=3,
                                                 dynamic_cost=0, temporary_block=False))],
                           queue_positions={}, queue_hidden={}, landing_badges=[], rooms={},
                           treated_patients={}, carried_patients={}, resting_staff={},
                           goods=dict(requests=[], deliveries=[]), actor_drawings=[], cabins=[],
                           observation_serial=1, overlay_consumed_serial=1,
                           overlay_passes=['traffic', 'actors'], overlay_child_indices=[0, 1, 2])
        synthetic_repair28_fields(self.report)

    def four_movers(self):
        self.runtime['world'].update(bounds=dict(min_x=1, max_x=1, min_y=1, max_y=1), walkable_cells=[[1, 1, 0]])
        self.report.update(viewport=[216, 236], paused=True, animation_time=0,
                           actor_tracks={}, boarded={}, actor_aggregations=[])
        for i, (center, at) in enumerate(zip([[108,118], [124,114], [92,118], [116,114]],
                                             [[88,120], [112,116], [76,120], [100,116]])):
            key = 'guest/'+str(i)
            self.runtime['guests']['guests'].append(dict(guest_id=str(i), current_cell=[1,1,0],
                         current_service_id='', target_facility_id='', state='MOVING'))
            self.report['actor_tracks'][key] = dict(cell=[1,1,0], **{'from':[1,1,0]}, at=0, direction=1)
            hashed = 5381
            for char in key:
                hashed = (hashed*33+ord(char)) & 0xffffffff
            self.report['actor_drawings'].append(dict(key=key, center=center, at=at, scale=2,
                 cell=[1,1,0], mode='idle', frame=hashed%15, variant=hashed%15, flip=False,
                 staff=False, state=3, receiving=False, carrying=False, riding=False))

    def test_four_movers_complete_geometry(self):
        self.four_movers()
        check.compare_actor_geometry(self.snapshot, self.report)

    def moving_four(self):
        self.four_movers()
        self.report.update(paused=False, animation_time=.21, badges=[])
        for row in self.report['actor_drawings']:
            self.report['actor_tracks'][row['key']]['from'] = [0, 1, 0]
            row['center'][0] -= 32
            row.update(scale=3, mode='walk', frame=row['variant']+2)
            row['at'] = [row['center'][0]-24, row['center'][1]-15]

    def drawing_audit(self):
        rows = []
        for category, identifier, rect, source, text in check.drawing_rectangle_rows(self.report):
            self.assertIsNone(text)
            colors = list(dict.fromkeys(check.re.findall(r'Color\("([0-9a-fA-F]{6})"',
                                                         (check.ROOT/source).read_text())))
            rows.append(dict(category=category, id=identifier, rect=rect,
                             style_source=source, style_colors=colors))
        return dict(rows=rows)

    def test_transit_raw_state_in_full_and_standalone_paths(self):
        self.moving_four()
        check.compare_actor_geometry(self.snapshot, self.report)
        check.validate_drawing_rows(self.report, self.drawing_audit(), state=self.snapshot)
        # Without authoritative context the overlapping moving cores fail closed.
        with self.assertRaisesRegex(ValueError, 'core overlap'):
            check.validate_drawing_rows(self.report, self.drawing_audit())

    def test_transit_cannot_be_claimed_by_flags_or_invalid_tracks(self):
        self.moving_four()
        mutations = {
            'paused': lambda r: r.update(paused=True),
            'expired': lambda r: r.update(animation_time=.42),
            'nonboolean_pause': lambda r: r.update(paused=0),
            'future': lambda r: r['actor_tracks']['guest/0'].update(at=1),
            'nonfinite': lambda r: r.update(animation_time=float('nan')),
            'wrong_endpoint': lambda r: r['actor_tracks']['guest/0'].update(cell=[0, 1, 0]),
            'teleport_origin': lambda r: r['actor_tracks']['guest/0'].update({'from': [-9, 1, 0]}),
        }
        for name, mutate in mutations.items():
            report = copy.deepcopy(self.report)
            mutate(report)
            for row in report['actor_drawings']:
                row.update(in_transit=True, region='transit')
            with self.subTest(name=name), self.assertRaises(ValueError):
                check.validate_drawing_rows(report, self.drawing_audit(), state=self.snapshot)

    def test_staff_and_guests_share_complete_geometry_front_and_back(self):
        self.four_movers()
        self.runtime['guests']['guests'].pop()
        self.runtime['staff']['staff'] = [dict(staff_id='3', current_cell=[1, 1, 0],
            state='AVAILABLE', assigned_facility_id='', carried_inventory_id='', role_id='guard')]
        self.report['actor_tracks']['staff/3'] = self.report['actor_tracks'].pop('guest/3')
        row = self.report['actor_drawings'][-1]
        hashed = 5381
        for char in 'staff/3':
            hashed = (hashed*33+ord(char)) & 0xffffffff
        row.update(key='staff/3', staff=True, state=2, variant=hashed%15, frame=hashed%15)
        for layer in (0, 1):
            self.report['layer'] = layer
            for member in self.runtime['guests']['guests'] + self.runtime['staff']['staff']:
                member['current_cell'][2] = layer
            for track in self.report['actor_tracks'].values():
                track['cell'][2] = layer
                track['from'][2] = layer
            for drawing in self.report['actor_drawings']:
                drawing['cell'][2] = layer
            with self.subTest(layer=layer):
                check.compare_actor_geometry(self.snapshot, self.report)

    def test_spoofed_region_does_not_allow_core_outside_own_cell(self):
        self.four_movers()
        self.report['actor_drawings'][0].update(region='facility', in_transit=True, at=[500, 120])
        with self.assertRaises(ValueError):
            check.compare_actor_geometry(self.snapshot, self.report)

    def test_aggregation_badge_is_required_in_drawing_audit(self):
        self.four_movers()
        self.report['badges'] = []
        audit = self.drawing_audit()
        self.report['actor_aggregations'] = [dict(at=[80, 74], text='+1',
                                                members=[], drawn_keys=[])]
        # This low-level test covers badge provenance only. Full geometry already
        # rejects the unnecessary aggregate in test_fitted_geometry_rejects_tampering.
        with self.assertRaisesRegex(ValueError, 'offline rectangle coverage'):
            check.validate_drawing_rows(self.report, audit)

    def test_low_world_pixel_complete_geometry_keeps_fractional_base(self):
        self.four_movers()
        self.runtime['guests']['guests'] = self.runtime['guests']['guests'][:1]
        self.report.update(viewport=[156, 146], actor_drawings=self.report['actor_drawings'][:1])
        self.report['actor_drawings'][0].update(center=[78, 73], at=[77, 72], scale=.1875)
        check.compare_actor_geometry(self.snapshot, self.report)

    def test_saturated_complete_geometry_keeps_every_identity(self):
        self.four_movers()
        prototype = copy.deepcopy(self.report['actor_drawings'][0])
        self.runtime['guests']['guests'] = []
        self.report['actor_drawings'] = []
        # 10 upright cores need 60px at p=1, but the interval is only 56px.
        # The first nine in preferred-x/key order occupy x82..136; guest/5
        # cannot be retained. This expected layout is independent of the solver.
        order = [6, 2, 9, 4, 0, 7, 3, 1, 8, 5]
        identities = {}
        for i in range(10):
            key = 'guest/'+str(i)
            cell = [1, 1, 0]
            self.runtime['guests']['guests'].append(dict(guest_id=str(i), current_cell=cell,
                current_service_id='', target_facility_id='', state='MOVING'))
            self.report['actor_tracks'][key] = dict(cell=cell, **{'from': cell}, at=0, direction=1)
            identities[i] = dict(key=key, id=str(i), cell=cell, state=3, mode='idle')
            if i == 5:
                continue
            hashed = 5381
            for char in key:
                hashed = (hashed*33+ord(char)) & 0xffffffff
            center = [108+[0,16,-16,8,-8,24,-24][i%7], 118-4*(i%2)]
            self.report['actor_drawings'].append(dict(prototype, key=key, center=center,
                at=[76+6*order.index(i), center[1]+19], scale=1, frame=hashed%15, variant=hashed%15))
        self.report['actor_aggregations'] = [dict(members=[identities[i] for i in order],
            drawn_keys=['guest/'+str(i) for i in order[:-1]], at=[80,74], text='+1')]
        check.compare_actor_geometry(self.snapshot, self.report)
        self.report['actor_aggregations'][0]['members'].pop()
        with self.assertRaisesRegex(ValueError, 'aggregation/no-fit'):
            check.compare_actor_geometry(self.snapshot, self.report)

    def test_fitted_geometry_rejects_tampering(self):
        self.four_movers()
        for field, value in [('cell', [0,1,0]), ('state', 4), ('scale', 3), ('at', [999,120])]:
            report = copy.deepcopy(self.report)
            report['actor_drawings'][0][field] = value
            with self.subTest(field=field), self.assertRaises(ValueError):
                check.compare_actor_geometry(self.snapshot, report)
        for mutation in ('missing', 'duplicate', 'aggregate'):
            report = copy.deepcopy(self.report)
            if mutation == 'missing':
                report['actor_drawings'].pop()
            elif mutation == 'duplicate':
                report['actor_drawings'].append(copy.deepcopy(report['actor_drawings'][0]))
            else:
                report['actor_aggregations'] = [dict(drawn_keys=[], text='+4')]
            with self.subTest(mutation=mutation), self.assertRaises(ValueError):
                check.compare_actor_geometry(self.snapshot, report)

    def test_density_and_pan_are_independently_recomputed(self):
        check.compare_traffic(self.snapshot, self.report)
        for key, value in [('current_density', 0), ('base_capacity', 9)]:
            changed = copy.deepcopy(self.report)
            changed['traffic'][0]['traffic'][key] = value
            with self.assertRaises(ValueError):
                check.compare_traffic(self.snapshot, changed)
        self.report['pan'][0] = 20
        with self.assertRaises(ValueError):
            check.compare_traffic(self.snapshot, self.report)

    def test_missing_queue_overflow_rejected(self):
        saved = json.loads((check.ROOT / 'tests/fixtures/p23-03/queued_elevator.json').read_text())['saved']
        self.runtime['elevators'] = saved
        saved['trips'] = [dict(copy.deepcopy(saved['trips'][0]), actor_id='guest:g'+str(i),
                              request_id='trip'+str(i), enqueued_at=i) for i in range(3)]
        self.report.update(shaft_footprint=[[0,0,0],[1,0,0],[0,1,0],[1,1,0]], queue_positions={},
                           queue_hidden={'guest/g0': True, 'guest/g1': True, 'guest/g2': True}, landing_badges=[dict(at=[-4, 54],
                           text='待ち3人（他3人）', cell=[0, 0, 0],
                           members=['guest/g0', 'guest/g1', 'guest/g2'], visible_count=0, overflow=3)])
        check.compare_queues(self.snapshot, self.report)
        self.report['landing_badges'][0]['overflow'] = 0
        with self.assertRaises(ValueError):
            check.compare_queues(self.snapshot, self.report)

    def test_fictional_goods_and_rooms_rejected(self):
        check.compare_goods(self.snapshot, self.report)
        check.compare_room_services(self.snapshot, self.report)
        self.report['goods']['requests'].append(dict(stock=100))
        with self.assertRaises(ValueError):
            check.compare_goods(self.snapshot, self.report)
        self.report['rooms']['invented'] = dict(count=0)
        with self.assertRaises(ValueError):
            check.compare_room_services(self.snapshot, self.report)

    def test_room_reason_and_rescue_occupancy(self):
        room = dict(instance_id='clinic', origin=[0, 0, 0], definition_id='infirmary',
                    state='NO_POWER', state_reasons=['NO_POWER'], suspension_reason='',
                    customer_capacity=2, active_staff_ids=[], active_services=[], occupied_cells=[])
        self.runtime['facilities']['instances'] = [room]
        self.runtime['staff']['tasks'] = [dict(task_id='rescue', task_type='rescue_transport',
             state='WORKING', assigned_staff_id='carrier', target_entity_id='patient', rescue_facility_id='clinic')]
        self.report['treated_patients'] = {'patient': 'clinic'}
        self.report['rooms'] = {'clinic': dict(state=2, state_name='NO_POWER', reasons=['NO_POWER'],
             operation=0, people=['patient'], count=1, limit=2, staff=[], suspension='')}
        check.compare_room_services(self.snapshot, self.report)
        for field, wrong in [('state', 6), ('operation', 1), ('people', []), ('count', 0)]:
            changed = copy.deepcopy(self.report)
            changed['rooms']['clinic'][field] = wrong
            with self.assertRaises(ValueError):
                check.compare_room_services(self.snapshot, changed)

    def test_cabin_geometry_and_phase(self):
        shaft = dict(config=dict(shaft_id='lift', stops=[[0, 0, 0]], width=2, capacity=4),
                     current_floor=0, phase='IDLE', operational=True, boarded=[])
        self.runtime['elevators']['shafts'] = [shaft]
        self.report.update(paused=True, animation_time=0, elevator_tracks={'lift': dict(floor=0, at=0)},
             cabins=[dict(shaft_id='lift', rect=[80, 72, 24, 19], pixel=1, operational=True,
                          doors_open=True, badge_at=[80, 58], badge='0/4人', passengers=[])])
        synthetic_repair28_fields(self.report, self.snapshot)
        check.compare_cabin_geometry(self.snapshot, self.report)
        for field, value in [('rect', [80, 70, 24, 19]), ('doors_open', False), ('badge', '4/4人')]:
            changed = copy.deepcopy(self.report)
            changed['cabins'][0][field] = value
            with self.assertRaises(ValueError):
                check.compare_cabin_geometry(self.snapshot, changed)


class AnchoredGeometryTests(unittest.TestCase):
    """Whole state classification fixtures, with independently chosen draw rows."""
    def setUp(self):
        SpatialTests.setUp(self)
        self.runtime['world'].update(bounds=dict(min_x=0, max_x=3, min_y=0, max_y=1),
            walkable_cells=[[x, y, 0] for y in range(2) for x in range(4)])
        self.report.update(viewport=[408, 332], paused=True, animation_time=0,
            actor_tracks={}, boarded={}, actor_aggregations=[], badges=[], traffic=[])
        self.runtime.update(defense=dict(raids=[]), events=dict(actor_locks={}))

    def person(self, key, cell, state='MOVING', **fields):
        kind, identifier = key.split('/')
        if kind == 'guest':
            person = dict(guest_id=identifier, current_cell=cell, state=state,
                          current_service_id='', target_facility_id='')
            self.runtime['guests']['guests'].append(person)
        else:
            person = dict(staff_id=identifier, current_cell=cell, state=state, health=100,
                          assigned_facility_id='', current_task_id='', carried_inventory_id='', role_id='guard')
            self.runtime['staff']['staff'].append(person)
        person.update(fields)
        self.report['actor_tracks'][key] = dict(cell=cell[:], **{'from': cell[:]}, at=0, direction=1)
        return person

    def drawing(self, key, center, mode='idle', scale=3, receiving=False, riding=False, at=None):
        kind, identifier = key.split('/')
        source = self.runtime['staff']['staff'] if kind == 'staff' else self.runtime['guests']['guests']
        person = next(r for r in source if r[kind+'_id'] == identifier)
        hashed = 5381
        for char in key:
            hashed = (hashed*33+ord(char)) & 0xffffffff
        row = dict(key=key, center=center, at=at or [center[0]-8*scale, center[1]+36-17*scale],
            scale=scale, mode=mode, frame=hashed%15, variant=hashed%15, flip=False,
            staff=kind == 'staff', state=check.enum_names(check.ROOT,
                'scripts/'+kind+'/'+kind+'_runtime.gd', 'State').index(person['state']),
            cell=person['current_cell'][:], receiving=receiving, carrying=False, riding=riding)
        if not riding:
            self.report['actor_drawings'].append(row)
        return row

    def room(self, fid, definition, cells, users=(), workers=()):
        facility = dict(instance_id=fid, definition_id=definition, origin=cells[0], occupied_cells=cells,
            state='OPEN', state_reasons=[], suspension_reason='', customer_capacity=10,
            active_services=[dict(guest_ids=list(users))] if users else [], active_staff_ids=list(workers), queue_entries=[])
        self.runtime['facilities']['instances'].append(facility)
        self.runtime['world']['occupancies'].append(dict(entity_id=fid, cells=cells))
        self.report['rooms'][fid] = dict(people=list(users), count=len(users), staff=list(workers),
            state=check.enum_names(check.ROOT, 'scripts/facility/facility_instance.gd', 'State').index('OPEN'),
            state_name='OPEN', reasons=[], operation=1, limit=10, suspension='', definition_id=definition,
            rect=check.Transform(self.runtime, self.report).room(cells))
        return facility

    def queue_work(self):
        facility = self.room('work', 'cafeteria', [[x,y,0] for y in (0,1) for x in (2,3)], workers=['w'])
        for key in ('a', 'b'):
            self.person('guest/'+key, [1,1,0], 'QUEUING', current_service_id='service', target_facility_id='work')
        # Queue takes precedence even over a nonempty current_service_id.
        facility['active_services'] = [dict(guest_ids=['a','b'])]
        self.report['rooms']['work'].update(people=['a','b'], count=2)
        facility['queue_entries'] = ['request']
        self.runtime['facilities']['requests'] = [dict(request_id='request', guest_ids=['a','b'])]
        self.person('staff/w', [2,1,0], 'WORKING', assigned_facility_id='work')
        self.report['queue_positions'] = {'guest/a':[172,118], 'guest/b':[108,118]}
        self.report['landing_badges'] = [dict(at=[140,54], text='待ち2人', cell=[1,1,0],
            members=['guest/a','guest/b'], visible_count=2, overflow=0)]
        self.drawing('guest/a', [172,118], 'eat', receiving=True)
        self.drawing('guest/b', [108,118], 'eat', receiving=True)
        self.drawing('staff/w', [236,126], 'work')

    def test_queue_lane_and_work_staff(self):
        self.queue_work()
        check.compare_actor_geometry(self.snapshot, self.report)

    def test_queue_precedence_order_and_category_forgeries(self):
        self.queue_work()
        for mutation in ('precedence', 'order', 'category', 'feet', 'pose'):
            report = copy.deepcopy(self.report)
            if mutation == 'precedence': report['actor_drawings'][0].update(center=[268,210], at=[244,195])
            if mutation == 'order':
                report['queue_positions']['guest/a'], report['queue_positions']['guest/b'] = report['queue_positions']['guest/b'], report['queue_positions']['guest/a']
            if mutation == 'category': report['actor_drawings'][0].update(staff=True, region='work', in_transit=True)
            if mutation == 'feet': report['actor_drawings'][0]['at'][1] -= 17
            if mutation == 'pose': report['actor_drawings'][0]['mode'] = 'wait'
            with self.subTest(mutation=mutation), self.assertRaises(ValueError):
                check.compare_actor_geometry(self.snapshot, report)

    def test_receiving_r13_guest_cannot_be_forced_into_access_cell(self):
        fixture = json.loads((check.ROOT/'tests/fixtures/p23-03/r13_service_region_gap.json').read_text())
        guest = fixture['guests'][0]
        self.runtime['world']['bounds'].update(max_x=9)
        self.report.update(viewport=[792,332], pan=[567,318])
        fid = guest['target']
        self.room(fid, 'cafeteria', [[x,y,0] for y in (0,1) for x in range(6,10)], users=[guest['id']])
        self.person('guest/'+guest['id'], guest['cell'], 'RECEIVING_SERVICE', current_service_id='service', target_facility_id=fid)
        # Exact historical room x1027..1283; access cell x963..1027.
        self.assertEqual(self.report['rooms'][fid]['rect'], fixture['room_rects'][fid])
        row = self.drawing('guest/'+guest['id'], [1069+2/3,528], 'eat', receiving=True, at=[1046,513])
        check.compare_actor_geometry(self.snapshot, self.report)
        row.update(center=[995,532], at=[971,517], region='floor')
        with self.assertRaises(ValueError): check.compare_actor_geometry(self.snapshot, self.report)

    def test_carried_patient_and_treated_staff(self):
        self.person('staff/carrier', [0,0,0], 'AVAILABLE')
        self.person('staff/patient', [0,0,0], 'INCAPACITATED')
        self.person('staff/treated', [2,0,0], 'INCAPACITATED')
        self.room('clinic', 'infirmary', [[2,0,0]])
        self.runtime['staff']['tasks'] = [dict(task_type='rescue_transport', state=phase,
            assigned_staff_id='carrier', target_entity_id=patient, rescue_facility_id='clinic')
            for patient, phase in [('patient','MOVING'), ('treated','WORKING')]]
        self.report.update(carried_patients={'patient':'carrier'}, treated_patients={'treated':'clinic'})
        self.report['rooms']['clinic'].update(people=['treated'], count=1)
        self.drawing('staff/carrier', [108,214])
        self.drawing('staff/patient', [108,166], 'collapsed')
        row = self.drawing('staff/treated', [236,190], 'collapsed')
        check.compare_actor_geometry(self.snapshot, self.report)
        self.report['actor_drawings'][1]['center'][1] += 4
        self.report['actor_drawings'][1]['at'][1] += 4
        with self.assertRaises(ValueError): check.compare_actor_geometry(self.snapshot, self.report)

    def cabin(self, count):
        shaft = json.loads((check.ROOT/'tests/fixtures/p23-03/queued_elevator.json').read_text())['saved']['shafts'][0]
        shaft.update(current_floor=0, phase='IDLE', boarded=['guest:p%02d'%i for i in range(count)])
        # Synthetic saved configuration, not a production definition or capacity edit.
        shaft['config']['capacity'] = max(4, count)
        self.runtime['elevators']['shafts'] = [shaft]
        sid = shaft['config']['shaft_id']
        self.report['elevator_tracks'] = {sid: dict(floor=0, **{'from':0}, at=0)}
        cabin = dict(shaft_id=sid, rect=[92,174,96,76], pixel=4, operational=True, doors_open=True,
                     badge_at=[92,160], badge=('満員 ' if count >= 4 else '')+f'{count}/{max(4,count)}人', passengers=[])
        self.report['cabins'] = [cabin]
        synthetic_repair28_fields(self.report, self.snapshot)
        for i in range(count):
            key = 'guest/p%02d'%i
            self.person(key, [0,0,0])
            self.report['boarded'][key] = True
            cabin['passengers'].append(self.drawing(key, [116+32*(i%2),206-36*((i//2)%2)], scale=2, riding=True))
        return cabin

    def occupied_rest(self):
        state, _, audit = RestAuditTests().fixture()
        old = state['systems']['runtime']
        self.runtime['path'] = old['path']
        self.person('staff/s', [0,0,0], 'OFF_DUTY')
        self.room('a', 'staff_waiting_room', [[0,0,0]])
        self.room('b', 'staff_waiting_room', [[0,0,0]])
        for facility in self.runtime['facilities']['instances']:
            facility['occupied_cells'] = []  # Public selector falls back to origin.
        self.report['resting_staff'] = {'s':'a'}
        self.report['rooms']['a'].update(people=['s'], count=1)
        self.drawing('staff/s', [108,198])
        return audit

    def test_compare_rest_with_actual_occupied_facility(self):
        audit = self.occupied_rest()
        self.assertTrue(check.compare_rest(self.snapshot, self.report, audit)['s']['resting'])
        self.report['actor_drawings'][0]['at'][1] += 1
        with self.assertRaises(ValueError): check.compare_rest(self.snapshot, self.report, audit)

    def test_compare_rest_independently_proves_no_fit(self):
        audit = self.occupied_rest()
        rest = self.report['actor_drawings'].pop()
        rest.update(scale=1, at=[100,217])
        order = [6,2,9,4,0,7,3,1,8,5]
        identities = {}
        for i in range(10):
            key = 'guest/'+str(i)
            self.person(key, [0,0,0])
            center = [108+[0,16,-16,8,-8,24,-24][i%7],214-4*(i%2)]
            row = self.drawing(key, center, scale=1, at=[76+6*order.index(i),center[1]+19])
            identities[i] = dict(key=key, id=str(i), state=3, cell=[0,0,0], mode='idle')
        self.report['actor_drawings'] = [r for r in self.report['actor_drawings'] if r['key'] != 'guest/5']+[rest]
        self.report['actor_aggregations'] = [dict(members=[identities[i] for i in order],
            drawn_keys=['guest/'+str(i) for i in order[:9]], at=[80,170], text='+1')]
        check.compare_rest(self.snapshot, self.report, audit)
        stale = copy.deepcopy(self.report)
        stale['actor_aggregations'][0]['members'].append(dict(key='staff/s', id='s', state=0, cell=[0,0,0], mode='idle'))
        with self.assertRaises(ValueError): check.compare_rest(self.snapshot, stale, audit)
        self.report['actor_aggregations'][0]['members'].pop()
        with self.assertRaises(ValueError): check.compare_rest(self.snapshot, self.report, audit)

    def test_compare_rest_with_queued_staff(self):
        audit = self.occupied_rest()
        self.person('staff/q', [0,0,0], 'OFF_DUTY')
        saved = json.loads((check.ROOT/'tests/fixtures/p23-03/queued_elevator.json').read_text())['saved']
        saved['trips'][0]['actor_id'] = 'staff:q'
        self.runtime['elevators'] = saved
        self.runtime['world']['walkable_cells'].append([-1,0,0])
        self.report['queue_positions'] = {'staff/q':[44,214]}
        self.report['landing_badges'] = [dict(at=[-4,150], text='待ち1人', cell=[0,0,0],
            members=['staff/q'], visible_count=1, overflow=0)]
        queued = copy.deepcopy(audit['staff'][0])
        queued.update(staff_id='q', in_transit=True)
        queued['comparison_only'].update(resting=False, label='休憩先へ移動中')
        audit['staff'].append(queued)
        self.drawing('staff/q', [44,214], 'wait')
        synthetic_repair28_fields(self.report, self.snapshot)
        result = check.compare_rest(self.snapshot, self.report, audit)
        self.assertTrue(result['s']['resting'])
        self.assertFalse(result['q']['resting'])
        self.report['actor_drawings'][-1].update(center=[108,198], at=[84,183])
        with self.assertRaises(ValueError): check.compare_rest(self.snapshot, self.report, audit)

    def test_passenger_and_floor_actor_front_and_back(self):
        cabin = self.cabin(1)
        self.person('guest/floor', [2,0,0])
        self.drawing('guest/floor', [236,214])
        for layer in (0,1):
            self.report['layer'] = layer
            for guest in self.runtime['guests']['guests']: guest['current_cell'][2] = layer
            for cell in self.runtime['elevators']['shafts'][0]['config']['stops']: cell[2] = layer
            for row in self.report['actor_drawings']+cabin['passengers']: row['cell'][2] = layer
            for track in self.report['actor_tracks'].values():
                track['cell'][2] = track['from'][2] = layer
            synthetic_repair28_fields(self.report, self.snapshot)
            with self.subTest(layer=layer):
                check.compare_cabin_geometry(self.snapshot, self.report)
                check.compare_actor_geometry(self.snapshot, self.report)

    def test_cabin_aggregation_retains_full_boarded_membership(self):
        cabin = self.cabin(40)
        originals = copy.deepcopy(cabin['passengers'])
        retained = {}
        for y in (170,206):
            group = sorted([r for r in originals if r['center'][1] == y], key=lambda r:(r['center'][0],r['key']))
            drawn = group[:14]  # 88px holds fourteen 6px cores, never fifteen.
            for i, row in enumerate(drawn):
                row.update(scale=1, at=[94+6*i,y+19])
                retained[row['key']] = row
            self.report['actor_aggregations'].append(dict(
                members=[dict(key=r['key'], id=r['key'][6:], state=r['state'], cell=r['cell'], mode=r['mode']) for r in group],
                drawn_keys=[r['key'] for r in drawn], at=[96,178], text='+6'))
        cabin['passengers'] = [retained[r['key']] for r in originals if r['key'] in retained]
        check.compare_cabin_geometry(self.snapshot, self.report)
        check.compare_actor_geometry(self.snapshot, self.report)
        self.assertEqual(sum(len(a['members']) for a in self.report['actor_aggregations']), 40)
        for mutation in ('lost_identity','badge'):
            report = copy.deepcopy(self.report)
            if mutation == 'lost_identity': report['actor_aggregations'][0]['members'].pop()
            else: report['actor_aggregations'][0]['at'][0] += 1
            with self.subTest(mutation=mutation), self.assertRaises(ValueError):
                check.compare_actor_geometry(self.snapshot, report)


class CabinLandingTests(unittest.TestCase):
    """Constructed raw topology; explicit expected draws, never OS evidence."""
    setUp = AnchoredGeometryTests.setUp
    person = AnchoredGeometryTests.person
    drawing = AnchoredGeometryTests.drawing

    def lift(self, floor=0):
        self.runtime['elevators']['shafts'] = [dict(
            config=dict(shaft_id='lift', stops=[[1,0,0], [1,1,0]], width=2, capacity=4),
            current_floor=0, phase='PICKUP', boarding_floor=1, phase_origin_floor=0,
            remaining_minutes=10*(1-floor), phase_total_minutes=10, operational=True, boarded=[])]
        self.report['elevator_tracks'] = {'lift': dict(floor=floor, **{'from': floor}, at=0)}
        # The transform is 64x96; the full two-cell cabin is 96x76.
        y = check.math.copysign(check.math.floor(abs(174-96*floor)+.5), 174-96*floor)
        self.report['cabins'] = [dict(shaft_id='lift', rect=[156,y,96,76], pixel=4,
            operational=True, doors_open=False, badge_at=[156,y-14], passengers=[])]
        synthetic_repair28_fields(self.report, self.snapshot)

    def test_moving_guest_same_floor_and_old_position_negative(self):
        self.lift()
        self.person('guest/g', [1,0,0])
        self.drawing('guest/g', [108,214])
        check.compare_actor_geometry(self.snapshot, self.report)
        bad = copy.deepcopy(self.report)
        bad['actor_drawings'][0].update(center=[172,214], at=[148,199])
        with self.assertRaisesRegex(ValueError, 'non-passenger inside cabin'):
            check.compare_actor_geometry(self.snapshot, bad)

    def test_transit_from_lift_to_corridor_is_relocated_before_jitter(self):
        self.lift()
        self.person('guest/g', [0,0,0])
        self.report.update(paused=False, animation_time=.21)
        self.report['actor_tracks']['guest/g']['from'] = [1,0,0]
        row = self.drawing('guest/g', [108,214], 'walk')
        row['frame'] += 2
        check.compare_actor_geometry(self.snapshot, self.report)
        bad = copy.deepcopy(self.report)
        bad['actor_drawings'][0]['center'][0] += 1
        with self.assertRaisesRegex(ValueError, 'actor fitted center'):
            check.compare_actor_geometry(self.snapshot, bad)

    def test_carrier_and_raised_patient_follow_relocated_region(self):
        self.lift()
        self.person('staff/carrier', [1,0,0], 'AVAILABLE')
        self.person('staff/patient', [1,0,0], 'INCAPACITATED')
        self.runtime['staff']['tasks'] = [dict(task_type='rescue_transport', state='MOVING',
            assigned_staff_id='carrier', target_entity_id='patient', rescue_facility_id='')]
        self.report['carried_patients'] = {'patient': 'carrier'}
        self.drawing('staff/carrier', [108,214])
        self.drawing('staff/patient', [108,166], 'collapsed')
        check.compare_actor_geometry(self.snapshot, self.report)
        self.assertTrue(all(not check.intersects(check.actor_identity_core(r), self.report['cabins'][0]['rect'])
                            for r in self.report['actor_drawings']))

    def test_fractional_moving_cabin_overlaps_floor_above(self):
        self.lift(.8)
        self.person('guest/g', [1,1,0])
        self.drawing('guest/g', [108,118])
        check.compare_actor_geometry(self.snapshot, self.report)

    def test_no_lane_aggregates_identity_and_never_drops_silently(self):
        self.lift()
        self.runtime['world']['walkable_cells'] = [[1,0,0], [2,0,0]]
        self.person('guest/g', [1,0,0])
        self.report['actor_aggregations'] = [dict(members=[dict(key='guest/g', id='g', state=3,
            cell=[1,0,0], mode='idle')], drawn_keys=[], at=[60,150], text='+1')]
        check.compare_actor_geometry(self.snapshot, self.report)
        for mutation in ('badge', 'identity', 'hidden'):
            bad = copy.deepcopy(self.report)
            if mutation == 'badge': bad['actor_aggregations'][0]['at'][0] += 1
            if mutation == 'identity': bad['actor_aggregations'][0]['members'] = []
            if mutation == 'hidden':
                bad['actor_aggregations'] = []
                bad['queue_hidden'] = {'guest/g': True}
            with self.subTest(mutation=mutation), self.assertRaises(ValueError):
                check.compare_actor_geometry(self.snapshot, bad)
        self.person('guest/h', [1,0,0])
        aggregate = self.report['actor_aggregations'][0]
        aggregate['members'].append(dict(key='guest/h', id='h', state=3, cell=[1,0,0], mode='idle'))
        aggregate['text'] = '+2'
        check.compare_actor_geometry(self.snapshot, self.report)

    def test_reserved_lane_prefers_free_then_shares_with_queue(self):
        self.lift()
        self.person('guest/q', [1,0,0], 'QUEUING')
        self.person('guest/g', [1,0,0])
        self.runtime['elevators']['trips'] = [dict(request_id='queue', actor_id='guest:q', state='QUEUED',
            shaft_id='lift', connector_id='lift.link.0.1', enqueued_at=0,
            **{'from': [1,0,0], 'to': [1,1,0]})]
        self.report.update(queue_positions={'guest/q':[108,214]},
            landing_badges=[dict(at=[60,150], text='待ち1人', cell=[1,0,0],
                                members=['guest/q'], visible_count=1, overflow=0)])
        self.drawing('guest/q', [108,214], 'wait')
        self.drawing('guest/g', [300,214])
        check.compare_actor_geometry(self.snapshot, self.report)
        self.runtime['world']['walkable_cells'].remove([3,0,0])
        self.report['actor_drawings'][0].update(at=[100,199])
        self.report['actor_drawings'][1].update(center=[108,214], at=[82,199])
        check.compare_actor_geometry(self.snapshot, self.report)

    def test_flags_drop_and_unnecessary_relocation_fail(self):
        self.lift()
        self.person('guest/g', [1,0,0])
        self.drawing('guest/g', [108,214])
        check.compare_actor_geometry(self.snapshot, self.report)
        for mutation in ('riding', 'region', 'boarded', 'drop'):
            bad = copy.deepcopy(self.report)
            if mutation == 'riding': bad['actor_drawings'][0]['riding'] = True
            if mutation == 'region':
                bad['actor_drawings'][0]['region'] = 'landing'
            if mutation == 'boarded': bad['boarded']['guest/g'] = True
            if mutation == 'drop': bad['actor_drawings'] = []
            with self.subTest(mutation=mutation), self.assertRaises(ValueError):
                check.compare_actor_geometry(self.snapshot, bad)
        # A free corridor actor must remain at its actual cell, even beside a lift.
        self.runtime['guests']['guests'][0]['current_cell'] = [3,0,0]
        self.report['actor_tracks']['guest/g'].update(cell=[3,0,0], **{'from': [3,0,0]})
        with self.assertRaisesRegex(ValueError, 'actor fitted'):
            check.compare_actor_geometry(self.snapshot, self.report)


class ComparisonTests(unittest.TestCase):
    def pair(self):
        declaration = dict(metric='sum_destination_arrival_minutes_entire_cohort',
                           arrivals=[dict(id='g', offset=0)], window=2, to=[2, 1, 0],
                           improvement=dict(definition_id='guest_elevator', origin=[4, 0, 0]))
        a = dict(declaration=declaration, initial=dict(seed=1))
        a['arrival-g'] = dict(offset=0, payload=dict(guest_id='g'), result=dict(code='OK'))
        for minute in range(3):
            a['tick-%03d' % minute] = dict(minute=minute, guests=dict(guests=[dict(guest_id='g',
                current_cell=[2, 1, 0] if minute == 2 else [2, 0, 0])]))
        b = copy.deepcopy(a)
        b['build-0'] = dict(payload=declaration['improvement'], result=dict(code='OK'))
        b['tick-001']['guests']['guests'][0]['current_cell'] = [2, 1, 0]
        return a, b

    def test_strict_metric_from_typed_cells(self):
        a, b = self.pair()
        self.assertEqual(check.compare_layouts(a, b)['B'], 1)
        for mutation in ('demand', 'incomplete', 'no_improvement', 'extra_build'):
            with self.subTest(mutation=mutation):
                a, b = self.pair()
                if mutation == 'demand':
                    b['arrival-g']['payload']['guest_id'] = 'other'
                elif mutation == 'incomplete':
                    for minute in (1, 2):
                        b['tick-%03d' % minute]['guests']['guests'][0]['current_cell'] = [2, 0, 0]
                elif mutation == 'no_improvement':
                    b['tick-001'] = copy.deepcopy(a['tick-001'])
                else:
                    b['build-1'] = copy.deepcopy(b['build-0'])
                with self.assertRaises(ValueError):
                    check.compare_layouts(a, b)


class OfflineGeometryTests(unittest.TestCase):
    """Synthetic calibration examples; no OS or acceptance credit."""

    def setUp(self):
        self.calibration = dict(window_size=[100, 80], window_position=[240, 150],
                                screenshot_pixels=[200, 190], backing_scale=2,
                                canvas_stretch=1, titlebar_pixels=30)
        self.request = dict(window_size=[100, 80], window_position=[240, 150],
                            screen_scale=2, content_scale_factor=1, target_pid=123)
        self.clicks = [dict(target_pid=123, events=[dict(action='click',
            coordinates=[20, 70], calibration=copy.deepcopy(self.calibration),
            point=dict(screen=[10, 20], viewport=[10, 20],
                       viewport_to_screen=[1, 0, 0, 1, 0, 0]))])]

    def transform(self):
        return check.screenshot_transform(self.calibration, self.request, [200, 190], self.clicks)

    def test_uses_measured_scale_origin_not_desktop_position(self):
        self.assertEqual(check.affine_rect(self.transform(), [10, 20, 5, 7]), [20, 70, 10, 14])

    def test_shifted_titlebar_fails_independent_click(self):
        self.calibration['titlebar_pixels'] += 1
        self.calibration['screenshot_pixels'][1] += 1
        self.clicks[0]['events'][0]['calibration'] = copy.deepcopy(self.calibration)
        with self.assertRaises(ValueError):
            check.screenshot_transform(self.calibration, self.request, [200, 191], self.clicks)

    def test_wrong_click_offset_fails(self):
        self.clicks[0]['events'][0]['coordinates'][1] -= 30
        with self.assertRaisesRegex(ValueError, 'click screenshot offset'):
            self.transform()

    def test_wrong_pid_cannot_calibrate(self):
        self.clicks[0]['target_pid'] += 1
        with self.assertRaisesRegex(ValueError, 'missing independent click'):
            self.transform()

    def test_nonfinite_and_false_scale_rejected(self):
        for scale in (False, float('nan'), 0, -1):
            self.calibration['backing_scale'] = scale
            with self.assertRaisesRegex(ValueError, 'invalid screenshot calibration'):
                self.transform()

    def test_all_visibility_cases_are_accounted_for(self):
        self.assertEqual(check.rectangle_eligibility([2, 3, 4, 5], [10, 10]),
                         ('visible', [2, 3, 4, 5]))
        self.assertEqual(check.rectangle_eligibility([-2, 3, 4, 5], [10, 10]),
                         ('clipped', [0, 3, 2, 5]))
        self.assertEqual(check.rectangle_eligibility([10, 3, 4, 5], [10, 10])[0], 'outside')
        self.assertEqual(check.rectangle_eligibility([2, 3, 0, 5], [10, 10])[0], 'outside')

    def test_invalid_rectangles_rejected(self):
        for rect in ([0, 0, -1, 4], [float('nan'), 0, 1, 4]):
            with self.assertRaisesRegex(ValueError, 'invalid clipping rectangle'):
                check.rectangle_eligibility(rect, [10, 10])

    def test_noise_is_measured_and_does_not_assert_class(self):
        before = (2, 2, [[(10, 20, 30)]*2 for _ in range(2)])
        after = copy.deepcopy(before)
        after[2][1][1] = (10, 27, 30)
        noise = check.paired_crop_noise(before, after, [0, 0, 2, 2])
        self.assertEqual(noise, dict(samples=4, max_channel_delta=7,
                                     mean_channel_delta=1.75, unchanged=3))
        self.assertNotIn('pass', noise)
        self.assertNotIn('state', noise)

    def test_wrong_dimensions_and_empty_crop_rejected(self):
        image = (1, 1, [[(0, 0, 0)]])
        with self.assertRaisesRegex(ValueError, 'paired image dimensions'):
            check.paired_crop_noise(image, (2, 1, []), [0, 0, 1, 1])
        with self.assertRaisesRegex(ValueError, 'empty noise crop'):
            check.paired_crop_noise(image, image, [1, 0, 1, 1])

    def test_report_positions_and_source_provenance_are_recomputed(self):
        source = 'scripts/ui/pixel_people.gd'
        report = dict(rooms={}, landing_badges=[], cabins=[], resting_staff={}, badges=[],
                      actor_drawings=[dict(key='staff/s', staff=True, at=[1, 1], scale=1,
                                           mode='idle', frame=0, flip=False)])
        colors = list(dict.fromkeys(check.re.findall(r'Color\("([0-9a-fA-F]{6})"',
                                                   (check.ROOT/source).read_text())))
        audit = dict(rows=[dict(category='staff', id='staff/s', rect=[1, 1, 16, 18],
                                style_source=source, style_colors=colors)])
        check.validate_drawing_rows(report, audit)
        for change in ('shift', 'identity', 'missing', 'palette'):
            tampered = copy.deepcopy(audit)
            if change == 'shift': tampered['rows'][0]['rect'][0] += 1
            if change == 'identity': tampered['rows'][0]['category'] = 'guest'
            if change == 'missing': tampered['rows'] = []
            if change == 'palette': tampered['rows'][0]['style_colors'] = ['000000']
            with self.subTest(change=change), self.assertRaises(ValueError):
                check.validate_drawing_rows(report, tampered)


class FrameReviewTests(unittest.TestCase):
    def test_frame_raw_context_valid_malformed_and_missing_rest(self):
        fixture = SpatialTests()
        fixture.setUp()
        fixture.moving_four()
        fixture.report.update(viewport=[216,236], traffic=[])
        audit = fixture.drawing_audit()
        for row in audit['rows']:
            row.update(viewport_rect=row['rect'][:], screen_rect=row['rect'][:])
        for context, rest, passes in ((fixture.snapshot,None,True), ({},None,False)):
            result = check.frame_rows(fixture.report, audit, self.request, dict(calibration=self.calibration),
                                     [self.image], self.clicks, state=context, rest_audit=rest)
            self.assertEqual(not result['errors'], passes)
        fixture.report['resting_staff'] = {'s':'a'}
        for rest in (None, [], {}):
            result = check.frame_rows(fixture.report, audit, self.request, dict(calibration=self.calibration),
                                     [self.image], self.clicks, state=fixture.snapshot, rest_audit=rest)
            self.assertFalse(result['mechanical_pass'])
        fixture.report['resting_staff'] = {}
        result = check.frame_rows(fixture.report, audit, self.request, dict(calibration=self.calibration),
                                 [self.image], self.clicks)
        self.assertFalse(result['mechanical_pass'])
        self.assertIn('core overlap', str(result['errors']))

    def test_frame_with_valid_occupied_rest_authority(self):
        fixture = AnchoredGeometryTests()
        fixture.setUp()
        rest = fixture.occupied_rest()
        rows = []
        for category, identifier, rect, source, text in check.drawing_rectangle_rows(fixture.report):
            self.assertIsNone(text)
            colors = list(dict.fromkeys(check.re.findall(r'Color\("([0-9a-fA-F]{6})"', (check.ROOT/source).read_text())))
            rows.append(dict(category=category, id=identifier, rect=rect, viewport_rect=rect[:],
                screen_rect=rect[:], style_source=source, style_colors=colors))
        result = check.frame_rows(fixture.report, dict(rows=rows), self.request, dict(calibration=self.calibration),
                                 [self.image], self.clicks, state=fixture.snapshot, rest_audit=rest)
        self.assertEqual(result['errors'], [])
        rest['path_service_sha256'] = '0'*64
        result = check.frame_rows(fixture.report, dict(rows=rows), self.request, dict(calibration=self.calibration),
                                 [self.image], self.clicks, state=fixture.snapshot, rest_audit=rest)
        self.assertFalse(result['mechanical_pass'])
        self.assertIn('rest PathService source pin', result['errors'])

    def test_os_offline_raw_context_binding_and_hashes(self):
        def typed(value):
            if isinstance(value, dict): return dict(type='Dictionary', entries=[[typed(k),typed(v)] for k,v in value.items()])
            if isinstance(value, list): return dict(type='Array', items=[typed(v) for v in value])
            return dict(type={str:'String',int:'int'}[type(value)], text=json.dumps(value))
        with tempfile.TemporaryDirectory() as temporary:
            stream = Path(temporary)
            (stream/'objects').mkdir()
            def put(value):
                raw = json.dumps(typed(value)).encode()
                digest = hashlib.sha256(raw).hexdigest()
                (stream/'objects'/(digest+'.json')).write_bytes(raw)
                return digest
            report = {'actor_drawings':[]}
            state = {'systems':{'runtime':{}}}
            rest = {'staff':[], 'path_service_sha256':check.sha(check.ROOT/'scripts/world/path_service.gd')}
            capture = dict(report=put(report), state=put(state), rest_audit=put(rest))
            self.assertEqual(check.drawing_context(stream, capture, report), (state,rest))
            for field in ('state','rest_audit','report'):
                missing = dict(capture)
                del missing[field]
                with self.assertRaises(ValueError): check.drawing_context(stream, missing, report)
            for field in ('state','rest_audit'):
                bad = dict(capture, **{field:put({})})
                with self.assertRaises(ValueError): check.drawing_context(stream, bad, report)
            with self.assertRaises(ValueError): check.drawing_context(stream, capture, {'wrong':1})
            with self.assertRaises(FileNotFoundError):
                check.drawing_context(stream, dict(capture, state='0'*64), report)
            (stream/'objects'/(capture['state']+'.json')).write_text('{}')
            with self.assertRaisesRegex(ValueError, 'object hash'): check.drawing_context(stream, capture, report)

    def setUp(self):
        OfflineGeometryTests.setUp(self)
        self.request.update(local_to_viewport=[1, 0, 0, 1, 0, 0],
                            viewport_to_screen=[1, 0, 0, 1, 0, 0],
                            local_to_screen=[1, 0, 0, 1, 0, 0])
        self.report = dict(layer=0, viewport=[100, 80], rooms={}, cabins=[],
                           landing_badges=[], badges=[], resting_staff={}, traffic=[],
                           actor_drawings=[dict(key='staff/s', staff=True, at=[1, 1], scale=1, state=0,
                                                mode='idle', frame=0, flip=False)])
        source = 'scripts/ui/pixel_people.gd'
        colors = list(dict.fromkeys(check.re.findall(r'Color\("([0-9a-fA-F]{6})"',
                                                   (check.ROOT/source).read_text())))
        self.audit = dict(rows=[dict(category='staff', id='staff/s', rect=[1, 1, 16, 18],
                         viewport_rect=[1, 1, 16, 18], screen_rect=[1, 1, 16, 18],
                         style_source=source, style_colors=colors)])
        self.image = (200, 190, [[(x % 256, y % 256, 0) for x in range(200)] for y in range(190)])

    def frame(self):
        return check.frame_rows(self.report, self.audit, self.request,
                                dict(calibration=self.calibration), [self.image], self.clicks)

    def test_frame_without_raw_state_explicitly_denies_transit_exemptions(self):
        self.report['actor_drawings'][0]['in_transit'] = True
        self.assertEqual(self.frame()['transit_authority'],
                         'absent: fail closed, no transit exemptions')

    def test_blank_crop_fails_without_threshold(self):
        self.image = (200, 190, [[(0, 0, 0)] * 200 for _ in range(190)])
        result = self.frame()
        self.assertFalse(result['mechanical_pass'])
        self.assertIn('blank crop', result['rows'][0]['failures'])

    def test_missing_extra_and_unaudited_traffic_fail(self):
        for change in ('missing', 'extra', 'traffic'):
            with self.subTest(change=change):
                original = copy.deepcopy(self.audit)
                if change == 'missing': self.audit['rows'].clear()
                if change == 'extra': self.audit['rows'] *= 2
                if change == 'traffic':
                    self.report['traffic'] = [dict(cell=[-6, 2, 0], rect=[1, 1, 5, 5],
                        traffic=dict(current_density=1, temporary_block=False))]
                result = self.frame()
                self.assertFalse(result['mechanical_pass'])
                self.assertEqual(len(result['rows']), len(self.audit['rows']))
                self.assertIn('offline rectangle coverage', result['errors'])
                self.audit = original

    def test_dense_or_blocked_traffic_row_closes_audit_gap(self):
        source = 'scenes/main/world_preview.gd'
        colors = list(dict.fromkeys(check.re.findall(r'Color\("([0-9a-fA-F]{6})"',
                                                   (check.ROOT/source).read_text())))
        for density, blocked in ((1, False), (0, True), (0, False)):
            with self.subTest(density=density, blocked=blocked):
                self.report['traffic'] = [dict(cell=[-6, 2, 0], rect=[1, 1, 5, 5],
                    traffic=dict(current_density=density, temporary_block=blocked))]
                active = density > 0 or blocked
                missing = self.frame()
                self.assertEqual(missing['mechanical_pass'], not active)
                if active:
                    self.assertIn('offline rectangle coverage', missing['errors'])
                    self.audit['rows'].append(dict(category='traffic', id='-6,2,0',
                        rect=[1, 1, 5, 5], viewport_rect=[1, 1, 5, 5],
                        screen_rect=[1, 1, 5, 5], style_source=source, style_colors=colors))
                    complete = self.frame()
                    self.assertTrue(complete['mechanical_pass'], complete['errors'])
                    self.audit['rows'].pop()

    def test_wrong_state_enum_and_name_fail(self):
        source = 'scripts/ui/pixel_tower_art.gd'
        self.report['actor_drawings'] = []
        room = dict(rect=[1, 1, 16, 18], state=3, state_name='NO_STAFF', operation=0)
        self.report['rooms'] = {'r': room}
        self.audit['rows'][0].update(category='room', id='r', style_source=source,
            style_colors=list(dict.fromkeys(check.re.findall(r'Color\("([0-9a-fA-F]{6})"',
                                                             (check.ROOT/source).read_text()))))
        self.assertEqual(self.frame()['rows'][0]['operation'], 'INACTIVE')
        for state, name in ((99, 'NO_STAFF'), (3, 'OPEN'), (True, 'NO_ACCESS')):
            room.update(state=state, state_name=name)
            with self.assertRaises(ValueError): self.frame()

    def test_frame_rejects_affine_and_calibration_shift(self):
        self.request['local_to_screen'][4] += 1
        with self.assertRaisesRegex(ValueError, 'affine composition'): self.frame()
        self.request['local_to_screen'][4] -= 1
        self.clicks[0]['events'][0]['coordinates'][1] += 1
        with self.assertRaisesRegex(ValueError, 'click screenshot offset'): self.frame()

    def test_eligible_crop_fully_outside_image_fails(self):
        for key in ('viewport_to_screen', 'local_to_screen'):
            self.request[key][4] = 300
        self.audit['rows'][0]['screen_rect'][0] += 300
        result = self.frame()
        self.assertFalse(result['mechanical_pass'])
        self.assertIn('eligible crop outside image', result['rows'][0]['failures'])

    def test_manifest_crop_and_independent_review_binding(self):
        with tempfile.TemporaryDirectory() as temporary:
            directory = Path(temporary)
            manifest, digest = check.write_frame_review(directory, self.frame(),
                [dict(whole_image_sha256='a'*64, raw_sha256='b'*64, conversion_sha256='c'*64)])
            self.assertEqual(check.sha(directory/'manifest.json'), digest)
            crop = manifest['rows'][0]['crops'][0]
            self.assertEqual(check.sha(directory/crop['path']), crop['sha256'])
            self.assertEqual(check.png_pixels((directory/crop['path']).read_bytes())[:2], (32, 36))
            verdict_path = directory/'independent-verdict.json'
            self.assertEqual(check.reviewed_pixel_categories(manifest, digest, verdict_path), set())
            verdict = dict(manifest_sha256=digest, reviewer='independent-opus-direct-image',
                           rows=[dict(row_index=0, category=True, state=True, position=True)])
            for change in ('wrong_sha', 'missing_row', 'extra_row', 'rejected_state'):
                bad = copy.deepcopy(verdict)
                if change == 'wrong_sha': bad['manifest_sha256'] = '0'*64
                if change == 'missing_row': bad['rows'] = []
                if change == 'extra_row': bad['rows'] *= 2
                if change == 'rejected_state': bad['rows'][0]['state'] = False
                verdict_path.write_text(json.dumps(bad))
                with self.subTest(change=change), self.assertRaises(ValueError):
                    check.reviewed_pixel_categories(manifest, digest, verdict_path)
            verdict_path.write_text(json.dumps(verdict))
            self.assertEqual(check.reviewed_pixel_categories(manifest, digest, verdict_path), {'staff'})
            manifest['mechanical_pass'] = False
            self.assertEqual(check.reviewed_pixel_categories(manifest, digest, verdict_path), set())


class Repair36LabelTests(unittest.TestCase):
    inputs = check.ROOT / 'tests/fixtures/p23-03/recorded-badge-negative.json.gz'

    def _recorded_payloads(self):
        bundle = json.loads(gzip.decompress(self.inputs.read_bytes()))
        expected = {
            'request': (655285, 'f9cd2d5ff9dd63d9cf7ae4a295ccef8eea16f75ac784696dc40906703731e1d8'),
            'state': (1867011, '724d8fe362ba3e05eddd83c4b7c54da101899df6cd91ae88250feea3d4feff82'),
            'manifest': (75726, 'a98782ec1263aba400d542862dd084d2f1dd4cf7a0851cf7f0679867e545a338'),
        }
        payloads = {}
        for name, (length, digest) in expected.items():
            payload = bundle['files'][name]['raw_utf8'].encode('utf-8')
            check.require(len(payload) == length and hashlib.sha256(payload).hexdigest() == digest,
                          'recorded fixture payload mismatch: ' + name)
            payloads[name] = payload
        return payloads

    def recorded(self):
        payloads = self._recorded_payloads()
        request = json.loads(payloads['request'])
        typed_state = json.loads(payloads['state'])
        check.require(typed_state.get('type') == 'Dictionary', 'whole dictionary required')
        return request, check.decode(typed_state), check.decode(request['report'])

    def resolved(self, report):
        result = copy.deepcopy(report)
        for collection, index, fields in check.observation_label_layout(report):
            result[collection][index].update(fields)
        return result

    def test_recorded_negative_and_positive_with_raw_semantics(self):
        request, state, old = self.recorded()
        rectangles = check.drawing_rectangle_rows(old)
        self.assertTrue(check.intersects(rectangles[5][2], rectangles[26][2]))
        with self.assertRaises(ValueError):
            check.validate_observation_labels(old)
        report = self.resolved(old)
        check.validate_observation_labels(report)
        check.compare_queues(state, report)
        check.compare_badges(state, report)
        check.compare_actor_geometry(state, report)
        queue = next(b for b in report['landing_badges'] if b['visible_count'] == 0)
        manifest = json.loads(self._recorded_payloads()['manifest'])
        original = manifest['rows'][5]
        self.assertEqual(queue['anchor'], original['local_rect'][:2])
        self.assertEqual(original['screenshot_rect'][:2], [642,469])
        self.assertEqual(queue['members'], next(b['members'] for b in old['landing_badges'] if b['visible_count'] == 0))

    def test_recorded_forgery_controls(self):
        _, state, old = self.recorded()
        good = self.resolved(old)
        for attack in ('at', 'displaced_by', 'count', 'hidden', 'anchor'):
            report = copy.deepcopy(good)
            row = report['actor_aggregations'][0]
            if attack == 'at': row['at'][0] += 1
            elif attack == 'displaced_by': row['displaced_by'] = []
            elif attack == 'count': row['text'] = '+'
            elif attack == 'hidden': report['actor_aggregations'] = []
            else: row['anchor'][0] += 1
            with self.subTest(attack=attack), self.assertRaises(ValueError):
                check.validate_observation_labels(report)
                check.compare_queues(state, report)
                check.compare_actor_geometry(state, report)

    def test_generic_order_and_edge_contact(self):
        for kinds in [('business', 'aggregation'), ('supply', 'landing'),
                      ('business', 'supply', 'landing', 'aggregation')]:
            report = dict(cabins=[], badges=[], landing_badges=[], actor_aggregations=[])
            for kind in kinds:
                key = 'badges' if kind in ('business', 'supply') else 'landing_badges' if kind == 'landing' else 'actor_aggregations'
                report[key].append(dict(kind=kind, at=[10,20], text='abc'))
            result = self.resolved(report)
            check.validate_observation_labels(result)
            positions = [fields['at'] for _, _, fields in check.observation_label_layout(result)]
            self.assertEqual(positions, [[10+36*i,20] for i in range(len(kinds))])
        report = dict(cabins=[dict(rect=[0,50,20,20], badge_at=[10,20], badge='abc')],
                      badges=[], landing_badges=[dict(at=[10,20], text='x')], actor_aggregations=[])
        self.assertEqual(self.resolved(report)['landing_badges'][0]['at'], [46,20])
        report['landing_badges'][0]['at'] = [46,20]
        self.assertEqual(self.resolved(report)['landing_badges'][0]['displaced_by'], [])

    def test_future_review_requires_indexed_judgments(self):
        manifest = dict(rows=[dict(row_index=7, category='queue', eligibility='visible', mechanical_pass=True)], mechanical_pass=True)
        verdict = dict(manifest_sha256='digest', reviewer='independent-opus-direct-image',
                       state_bits='1', position_bits='1', category_bits='1')
        with tempfile.TemporaryDirectory() as folder:
            path = Path(folder)/'verdict.json'
            path.write_text(json.dumps(verdict))
            with self.assertRaisesRegex(ValueError, 'indexed judgments'):
                check.reviewed_pixel_categories(manifest, 'digest', path, require_judgments=True)
            verdict['judgments'] = [dict(row_index=7, category=True, state=True, position=True)]
            path.write_text(json.dumps(verdict))
            self.assertEqual(check.reviewed_pixel_categories(manifest, 'digest', path, require_judgments=True), {'queue'})


if __name__ == '__main__':
    unittest.main()