P23-03: 生活の観察

2026-10-16 14:17:14 +09:00
parent 9a26f9ab8c
commit 6c13940cf4
10 changed files with 724 additions and 50 deletions
@@ -18,6 +18,233 @@ 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 verify(self):
+        with mock.patch.object(Path, 'read_bytes', lambda p:self.files[self.t][str(p.relative_to(check.ROOT))]):
+            return check.verify_composition(check.ROOT,self.roles,self.source,self.inputs,self.git)
+
+    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 ActorNonoverlapTests(unittest.TestCase):
    def report(self, second_at):
        return dict(actor_drawings=[
@@ -123,12 +350,12 @@ class ActorNonoverlapTests(unittest.TestCase):

 class LifeOverlayTests(unittest.TestCase):
    def report(self):
-        return dict(observation_serial=7, overlay_consumed_serial=7,
+        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))])
+                                  traffic=dict(current_density=1, temporary_block=False))]))

    def test_front_and_back_valid_consumed_band(self):
        for layer in (0, 1):
@@ -181,6 +408,7 @@ 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']:
@@ -228,7 +456,7 @@ class ProjectionBandTests(unittest.TestCase):
    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, 'floor actor core over traffic band'):
+        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):
@@ -247,7 +475,7 @@ class ProjectionBandTests(unittest.TestCase):
        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, 'floor actor core over traffic band: staff/spoof'):
+        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):
@@ -266,6 +494,7 @@ class ProjectionBandTests(unittest.TestCase):
        # 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)
@@ -807,6 +1036,7 @@ class SpatialTests(unittest.TestCase):
                           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]])
@@ -988,10 +1218,10 @@ class SpatialTests(unittest.TestCase):
        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(queue_positions={'guest/g0': [100, 82]},
-                           queue_hidden={'guest/g1': True, 'guest/g2': True}, landing_badges=[dict(at=[-4, 54],
-                           text='待ち3人（他2人）', cell=[0, 0, 0],
-                           members=['guest/g0', 'guest/g1', 'guest/g2'], visible_count=1, overflow=2)])
+        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):
@@ -1031,6 +1261,7 @@ class SpatialTests(unittest.TestCase):
        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)
@@ -1169,6 +1400,7 @@ class AnchoredGeometryTests(unittest.TestCase):
        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])
@@ -1233,6 +1465,7 @@ class AnchoredGeometryTests(unittest.TestCase):
        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'])
@@ -1241,9 +1474,8 @@ class AnchoredGeometryTests(unittest.TestCase):

    def test_passenger_and_floor_actor_front_and_back(self):
        cabin = self.cabin(1)
-        self.person('guest/floor', [0,0,0])
-        self.drawing('guest/floor', [108,214])
-        cabin['passengers'][0]['at'][0] = 108
+        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
@@ -1251,6 +1483,7 @@ class AnchoredGeometryTests(unittest.TestCase):
            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)