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-15 09:36:36 +09:00
parent 03100beabe
commit 4c8a18aa75
11 changed files with 2207 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/test_p23_life_observation_evidence.py
+307
View File
@@ -3,11 +3,318 @@
import copy
import unittest
import sys
import contextlib
import io
import json
import hashlib
import struct
import zlib
import tempfile
from pathlib import Path
from unittest import mock
sys.dont_write_bytecode = True
import check_p23_life_observation_evidence as check
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(dir=check.ROOT/'tests/fixtures/p23-03') 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_rectangle_pixel_and_transform_negatives(self):
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())))
report = dict(rooms={}, landing_badges=[], cabins=[], resting_staff={}, badges=[],
actor_drawings=[dict(key='staff/s', staff=True, at=[1, 1], scale=1)])
row = 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)
audit = dict(rows=[row])
matrix = [1, 0, 0, 1, 0, 0]
image = (24, 24, [[(113, 157, 148)]*24 for _ in range(24)])
self.assertEqual(check.compare_rectangles(report, audit, matrix, matrix, matrix, image), ['staff/staff/s'])
for change in ('rectangle', 'pixels', 'transform', 'coverage', 'palette'):
with self.subTest(change=change):
a, m, img = copy.deepcopy(audit), matrix[:], image
if change == 'rectangle': a['rows'][0]['rect'][0] += 1
if change == 'pixels': img = (24, 24, [[(0, 0, 0)]*24 for _ in range(24)])
if change == 'transform': m[4] = 10
if change == 'coverage': a['rows'].clear()
if change == 'palette': a['rows'][0]['style_colors'] = ['000000']
with self.assertRaises(ValueError): check.compare_rectangles(report, a, matrix, m, matrix, img)
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', 'display'):
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 == '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],