P23-03: 生活の観察

tools/check_p23_life_observation_evidence.py

@@ -812,10 +812,10 @@ def compare_traffic(snapshot, report):
         rows.append(dict(cell=cell, traffic={k: v for k, v in saved.items() if k != 'cell'},
                          rect=rect, band_rect=band))
     close(report['traffic'], rows, 'traffic and cell rect')
-    validate_life_overlay(report)
+    validate_life_overlay(report, snapshot=snapshot)
 
 
-def validate_life_overlay(report, root=ROOT):
+def validate_life_overlay(report, root=ROOT, snapshot=None):
     """Check consumed frame and source wiring, not merely reported order flags."""
     serial = report.get('observation_serial')
     require(type(serial) is int and serial > 0
@@ -842,6 +842,25 @@ def validate_life_overlay(report, root=ROOT):
             < overlay.find('_draw_observed_actor(actor, _life_overlay)')
             < overlay.find('_draw_actors(_life_overlay)'), 'LifeOverlay source pass order')
     actors = list(report['actor_drawings']) + [a for c in report['cabins'] for a in c['passengers']]
+    # Projection authority comes from raw state, never riding/region flags.
+    cabin_actors, carried_keys = set(), set()
+    if snapshot is not None:
+        if report['cabins']:
+            compare_cabin_geometry(snapshot, report)
+            shafts = {s['shaft_id']: s for s in elevator_report(snapshot)['shafts']}
+            for cabin in report['cabins']:
+                members = {key.replace('staff:', 'staff/', 1).replace('guest:', 'guest/', 1)
+                           for key in shafts[cabin['shaft_id']]['boarded']}
+                cabin_actors.update(id(a) for a in cabin['passengers'] if a['key'] in members)
+        tasks = snapshot['systems']['runtime']['staff']['tasks']
+        carried_keys = {'staff/' + t['target_entity_id'] for t in tasks
+                        if t['task_type'] == 'rescue_transport'
+                        and t['state'] not in ('WORKING', 'COMPLETED', 'CANCELLED', 'FAILED')
+                        and t['assigned_staff_id']}
+        carried_keys.difference_update(report.get('boarded', {}), report.get('queue_positions', {}))
+        if carried_keys:
+            # Reuse the original raw binding, track and raised-region fit checks.
+            compare_actor_geometry(snapshot, report)
     for row in report.get('traffic', []):
         x, y, w, h = row['rect']
         thickness = max(2, min(5, h * .16))
@@ -850,11 +869,24 @@ def validate_life_overlay(report, root=ROOT):
         close(row['band_rect'], expected, 'traffic band rectangle')
         if row['traffic']['current_density'] <= 0 and not row['traffic']['temporary_block']:
             continue
+        intervals = []
         for actor in actors:
             core = actor_identity_core(actor)
             intersects = (core[0] < expected[0] + expected[2] and expected[0] < core[0] + core[2]
                           and core[1] < expected[1] + expected[3] and expected[1] < core[1] + core[3])
-            require(not intersects, 'traffic band covers actor core: ' + actor['key'])
+            require(not intersects or id(actor) in cabin_actors
+                    or (actor in report['actor_drawings'] and actor['key'] in carried_keys),
+                    'floor actor core over traffic band: ' + actor['key'])
+            if intersects:
+                intervals.append((max(core[0], expected[0]),
+                                  min(core[0] + core[2], expected[0] + expected[2])))
+        end = expected[0]
+        for left, right in sorted(intervals):
+            if left > end:
+                break
+            end = max(end, right)
+        require(end < expected[0] + expected[2],
+                'traffic band fully occluded: ' + str(row.get('cell')))
 
 
 def compare_queues(snapshot, report):

tools/test_p23_life_observation_evidence.py

@@ -157,7 +157,7 @@ class LifeOverlayTests(unittest.TestCase):
     def test_core_crossed_by_actual_band_fails(self):
         report = self.report()
         report['actor_drawings'][0]['at'][1] = 60
-        with self.assertRaisesRegex(ValueError, 'traffic band covers actor core'):
+        with self.assertRaisesRegex(ValueError, 'floor actor core over traffic band'):
             check.validate_life_overlay(report)
 
     def test_forged_band_geometry_fails(self):
@@ -177,6 +177,105 @@ class LifeOverlayTests(unittest.TestCase):
                 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']
+
+    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, 'floor actor core over traffic band'):
+            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, 'floor actor core over traffic band: 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))]
+        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()