chickenbstrips/TheTree_mock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

P23-03: 生活の観察

Browse Source
This commit is contained in:
Agent Control Runner
2026-10-16 19:18:11 +09:00
parent 26f60095bf
commit 059c2f743b
6 changed files with 285 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/check_p23_life_observation_evidence.py
+74 -3
View File
@@ -310,6 +310,9 @@ def paired_crop_noise(before, after, rect):
def drawing_rectangle_rows(report):
"""Derive identities and local positions; do not read audit verdicts."""
if any('anchor' in b for key in ('badges', 'landing_badges', 'actor_aggregations')
for b in report.get(key, [])):
validate_observation_labels(report)
expected = []
tower = 'scripts/ui/pixel_tower_art.gd'
people = 'scripts/ui/pixel_people.gd'
@@ -549,13 +552,16 @@ def write_frame_review(directory, frame, provenance):
return manifest, digest
def reviewed_pixel_categories(manifest, digest, verdict_path):
def reviewed_pixel_categories(manifest, digest, verdict_path, *, require_judgments=False):
"""The worker never writes verdicts; absent independent review earns nothing."""
if not Path(verdict_path).is_file():
return set()
verdict = json.loads(Path(verdict_path).read_bytes())
require(verdict['manifest_sha256'] == digest, 'review manifest SHA mismatch')
require(verdict['reviewer'] == 'independent-opus-direct-image', 'independent image reviewer required')
if require_judgments:
require(isinstance(verdict.get('judgments'), list), 'explicit indexed judgments required')
verdict = dict(verdict, rows=verdict['judgments'])
eligible = [r for r in manifest['rows'] if r['eligibility'] != 'outside']
require([v['row_index'] for v in verdict['rows']] == [r['row_index'] for r in eligible],
'review row coverage')
@@ -651,7 +657,7 @@ def inspect_offline_mirror(directory):
receipt_sha256=sha(folder/'cua-receipt.json')))
review_dir = ROOT/'docs/plans/original-completion-2026-10-02/execution/p23-03/offline-r07-review'/name
manifest, digest = write_frame_review(review_dir, frame, provenance)
credited = reviewed_pixel_categories(manifest, digest, review_dir/'independent-verdict.json')
credited = reviewed_pixel_categories(manifest, digest, review_dir/'independent-verdict.json', require_judgments=True)
present = {r['category'] for r in frame['rows'] if r['eligibility'] != 'outside'}
overlays = [dict(cell=r['cell'], rect=r['rect']) for r in report['traffic']
if (r['traffic']['current_density'] > 0 or r['traffic']['temporary_block'])
@@ -803,6 +809,61 @@ def badge_rectangle(at, text):
return list(at) + [len(text) * 10 + 6, 14]
def observation_label_layout(report):
"""Independent ordered rectangle placement; never consume reported at/reasons."""
placed, result = [], []
for i, cabin in enumerate(report['cabins']):
placed.extend([(f'cabin/{i}/rect', cabin['rect']),
(f'cabin/{i}/badge', badge_rectangle(cabin['badge_at'], cabin['badge']))])
for kind, collection in [('business', 'badges'), ('supply', 'badges'),
('landing', 'landing_badges'), ('aggregation', 'actor_aggregations')]:
for i, row in enumerate(report.get(collection, [])):
if collection == 'badges' and row['kind'] != kind:
continue
anchor = list(row.get('anchor', row['at']))
at, reasons = anchor[:], []
while True:
conflict = next(((identity, rect) for identity, rect in placed
if intersects(badge_rectangle(at, row['text']), rect)), None)
if conflict is None:
break
identity, rect = conflict
reasons.append(identity)
at[0] = rect[0] + rect[2]
result.append((collection, i, dict(anchor=anchor, at=at, displaced_by=reasons)))
placed.append((f'{kind}/{i}', badge_rectangle(at, row['text'])))
return result
def observation_anchor_report(report):
"""Projection for existing raw-state semantic recomputation, not evidence."""
import copy
raw = copy.deepcopy(report)
for collection in ('badges', 'landing_badges', 'actor_aggregations'):
for row in raw.get(collection, []):
if 'anchor' in row:
row['at'] = row.pop('anchor')
row.pop('displaced_by', None)
return raw
def validate_observation_labels(report):
require(all(key in report for key in ('cabins', 'badges', 'landing_badges', 'actor_aggregations')),
'missing observation label collections')
rectangles = [badge_rectangle(c['badge_at'], c['badge']) for c in report['cabins']]
rectangles += [badge_rectangle(b['at'], b['text'])
for key in ('badges', 'landing_badges', 'actor_aggregations')
for b in report.get(key, [])]
for i, rect in enumerate(rectangles):
require(not any(intersects(rect, other) for other in rectangles[:i]),
'observation label intersection')
for collection, index, wanted in observation_label_layout(report):
row = report[collection][index]
for key, value in wanted.items():
require(key in row, 'missing label ' + key)
close(row[key], value, 'observation label/' + key)
def intersects(a, b):
cut = rect_intersection(a, b)
return cut[2] > 0 and cut[3] > 0
@@ -965,6 +1026,9 @@ def validate_life_overlay(report, root=ROOT, snapshot=None):
def compare_queues(snapshot, report):
if any('anchor' in b for b in report.get('landing_badges', [])):
validate_observation_labels(report)
report = observation_anchor_report(report)
runtime = snapshot['systems']['runtime']
transform = Transform(runtime, report)
facilities = codec.indexed(runtime['facilities']['instances'], 'instance_id')
@@ -1222,6 +1286,9 @@ def compare_goods(snapshot, report, root=ROOT):
def compare_badges(state, report):
if any('anchor' in b for b in report.get('badges', [])):
validate_observation_labels(report)
report = observation_anchor_report(report)
runtime = state['systems']['runtime']
transform = Transform(runtime, report)
expected = []
@@ -1490,6 +1557,9 @@ def landing_actor_region(state, report, transform, cell, region, transit):
def compare_actor_geometry(state, report):
if any('anchor' in b for b in report.get('actor_aggregations', [])):
validate_observation_labels(report)
report = observation_anchor_report(report)
runtime = state['systems']['runtime']
compare_queues(state, report)
compare_room_services(state, report)
@@ -1800,6 +1870,7 @@ def inspect_stream(directory, root=ROOT):
report = read_object(directory, row['report'])
fields = compare_actor_rows(state, report, root)
require(report['elevators'] == elevator_report(state), 'typed elevator report mismatch')
validate_observation_labels(report)
compare_room_services(state, report, root)
compare_rest(state, report, read_object(directory, row['rest_audit']), root)
compare_queues(state, report)
@@ -2240,7 +2311,7 @@ def inspect_os_requests(directory, layout, role, pid, root=ROOT):
require(frame['mechanical_pass'], 'frame mechanical checks: ' + str(frame['errors'])
+ str([r['failures'] for r in frame['rows'] if not r['mechanical_pass']]))
pixel_categories.update(reviewed_pixel_categories(
manifest, manifest_sha, review_dir/'independent-verdict.json'))
manifest, manifest_sha, review_dir/'independent-verdict.json', require_judgments=True))
records.append(dict(request=request, ack=ack, observed=observed, cua=cua, request_sha256=digest))
require(len(seen) == len(bindings), 'unattributed screenshot bindings')
return records, pixel_categories
tools/test_p23_life_observation_evidence.py
+81
View File
@@ -2293,5 +2293,86 @@ class FrameReviewTests(unittest.TestCase):
self.assertEqual(check.reviewed_pixel_categories(manifest, digest, verdict_path), set())
class Repair36LabelTests(unittest.TestCase):
inputs = check.ROOT.parent / '.runtime/P23-03/build/audit-support/badge-repair36-exact-inputs'
def recorded(self):
request = json.loads((self.inputs/'bad-native-request/request.json').read_bytes())
stream = self.inputs.parent.parent/'p23-normal-os-26f-r17/os-pair/B/writer-stream'
capture = next(json.loads(line) for line in (stream/'manifest.jsonl').read_text().splitlines()
if json.loads(line).get('label') == 'writer-request-0021')
return request, check.read_object(stream, capture['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.inputs/'bad-native-request/review-image-0/manifest.json').read_bytes())
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()