P08-03: 配置の通常操作受入

This commit is contained in:
Agent Control Runner
2026-10-15 00:32:39 +09:00
parent f5f85ae08c
commit d369d1c399
5 changed files with 175 additions and 12 deletions
+23 -8
View File
@@ -116,6 +116,9 @@ def expected_operation(formal, result, event_type):
recent['items'] = recent['items'][-256:]
return expected
def r09_reuse(index):
return index.get('commit') == R09_COMMIT
def png(data):
require(data.startswith(b'\x89PNG\r\n\x1a\n'), 'missing OS screenshot')
offset = 8; chunks = []; payload = b''
@@ -135,6 +138,7 @@ def png(data):
require(bool(zlib.decompress(payload)), 'empty OS screenshot')
except zlib.error as exc:
raise ValueError('invalid OS screenshot pixels') from exc
return struct.unpack('>II', data[16:24])
def input_map(root):
names = [SPEC, CONTRACT, 'AGENTS.md', RECORDER, RECORDER+'.uid',
@@ -633,21 +637,25 @@ def check(project_root, evidence_dir):
require(index['schema'] == 'p08-normal-placement-v1' and index['acceptance'] == 'pending', 'schema/acceptance')
require(index['collection_error'] is None, 'collection failed')
require(index['conditions'] == CONDITIONS, 'missing condition')
actual = {n: v['sha256'] for n, v in runner.source_refs(root).items()}
original = evidence
pins_path = root/R09_DOCS/'ROOT_NATIVE_R09_COLLECTION_PINS.json'
if pins_path.exists():
reuse = r09_reuse(index)
dimensions_path = root/R09_DOCS/'ROOT_R09_READONLY_IMAGE_DIMENSIONS.json'
if reuse:
require(pins_path.is_file() and dimensions_path.is_file(), 'R09 reuse candidate')
r09_applicability(root, index)
r09_runner_binding(evidence, index, load(pins_path))
original = r09_collection(evidence, index, load(pins_path))
r09_images(evidence, index, load(root/R09_DOCS/'ROOT_R09_READONLY_IMAGE_DIMENSIONS.json'), load(pins_path))
r09_images(evidence, index, load(dimensions_path), load(pins_path))
actual = index['source_map']
else:
actual = {n: v['sha256'] for n, v in runner.source_refs(root).items()}
require(index['source_map'] == actual, 'stale source')
require(index['inputs'] == input_map(root), 'stale input')
require(index['spec_sha256'] == SPEC_SHA == hashlib.sha256((root/SPEC).read_text().strip().encode()).hexdigest(), 'immutable spec')
require(index['marker_contract_sha256'] == sha(root/CONTRACT), 'stale marker')
require(re.fullmatch('[0-9a-f]{40}', index['commit']) is not None, 'commit')
if not pins_path.exists():
if not reuse:
require(index['commit'] == subprocess.check_output(['git','-C',str(root),'rev-parse','HEAD'],text=True).strip(), 'stale commit')
def artifact(name):
require(isinstance(name, str) and name and not Path(name).is_absolute() and '..' not in Path(name).parts, 'unsafe/missing artifact')
@@ -969,13 +977,20 @@ def check(project_root, evidence_dir):
and raw['tool_name'] and raw['tool_call_id'], 'fake OS provenance')
require(raw['events'] and observed['events'], 'missing OS events')
require(raw['collector_pid'] > 0 and raw['collector_pid'] not in os_pids, 'fake OS collector')
screen_dimensions = []
for key in ('screen_before', 'screen_after'):
require(ack[key] == folder+key.replace('_', '-')+'.png', 'R09 screenshot request binding')
data = artifact(ack[key]).read_bytes()
if pins_path.exists():
require(ack[key] == folder+key.replace('_', '-')+'.png', 'R09 screenshot request binding')
jpeg_dimensions(data)
if reuse:
dimensions = jpeg_dimensions(data)
elif data.startswith(b'\x89PNG\r\n\x1a\n'):
dimensions = png(data)
elif data.startswith(b'\xff\xd8'):
dimensions = jpeg_dimensions(data)
else:
png(data)
raise ValueError('missing OS screenshot')
screen_dimensions.append(dimensions)
require(screen_dimensions[0] == screen_dimensions[1] and all(v > 0 for v in screen_dimensions[0]), 'OS screenshot dimension mismatch')
if request['action'] == 'cell' and 'position' in row:
dispatch_gate(row['position'], observed['events'])
if row['id'] == 'pan_zoom':
+87 -1
View File
@@ -598,7 +598,7 @@ class EvidenceTests(unittest.TestCase):
self.reject('operation kind mismatch')
def test_png_signature_is_not_a_screenshot(self):
(self.out/'os-pair/screen.png').write_bytes(b'\x89PNG\r\n\x1a\nFAKE')
next((self.out/'os-pair').rglob('screen-before.png')).write_bytes(b'\x89PNG\r\n\x1a\nFAKE')
F.repin(self.out)
self.reject('invalid OS screenshot')
@@ -722,6 +722,92 @@ def refresh_images(out, index, measurement):
measurement['original_index_sha256'] = hashlib.sha256((out/'index.json').read_bytes()).hexdigest()
class FreshRoutingTests(unittest.TestCase):
setUp = EvidenceTests.setUp
edit = EvidenceTests.edit
reject = EvidenceTests.reject
# Reuse the whole synthetic proof, not a boolean checker substitute.
def historical(self):
F.put(self.root/C.R09_DOCS/'ROOT_NATIVE_R09_COLLECTION_PINS.json', {'synthetic': True})
def fresh_check(self):
commit = C.load(self.out/'index.json')['commit']
original_load = C.load
def guarded_load(path):
self.assertNotEqual(Path(path).name, 'ROOT_R09_READONLY_IMAGE_DIMENSIONS.json')
return original_load(path)
with contextlib.ExitStack() as stack:
stack.enter_context(patch.object(C.subprocess, 'check_output', return_value=commit+'\n'))
stack.enter_context(patch.object(C, 'load', side_effect=guarded_load))
for name in ('r09_applicability', 'r09_runner_binding', 'r09_collection', 'r09_images'):
stack.enter_context(patch.object(C, name, side_effect=AssertionError('fresh used '+name)))
return C.check(self.root, self.out)
def test_historical_pins_fresh_positive(self):
self.historical()
self.assertEqual([], self.fresh_check())
def test_historical_pins_stale_commit(self):
self.historical()
self.edit('index.json', lambda d: d.update(commit='0'*40))
with patch.object(C.subprocess, 'check_output', return_value='1'*40+'\n'):
self.reject('stale commit')
def test_historical_pins_stale_source(self):
self.historical()
self.edit('index.json', lambda d: d['source_map'].update({'project.godot':'0'*64}))
self.assertIn('stale source', ' '.join(self.fresh_check()))
def test_historical_pins_stale_input(self):
self.historical()
self.edit('index.json', lambda d: d['inputs'].update({'AGENTS.md':'0'*64}))
self.assertIn('stale input', ' '.join(self.fresh_check()))
def test_reuse_missing_pins(self):
self.edit('index.json', lambda d: d.update(commit=C.R09_COMMIT))
self.reject('R09 reuse candidate')
def test_reuse_wrong_applicability(self):
self.historical()
F.put(self.root/C.R09_DOCS/'ROOT_R09_READONLY_IMAGE_DIMENSIONS.json', {'synthetic':True})
self.edit('index.json', lambda d: d.update(commit=C.R09_COMMIT))
with patch.object(C.subprocess, 'check_output', return_value=C.R09_COMMIT+'\n'), patch.object(
C.subprocess, 'run', return_value=SimpleNamespace(returncode=1)):
self.reject('R09_COMMIT not an ancestor')
def screenshots(self, data):
for path in self.out.rglob('screen-*.png'):
path.write_bytes(data)
F.repin(self.out)
def test_fresh_jpeg_positive(self):
self.historical()
self.screenshots(negative_jpeg())
self.assertEqual([], self.fresh_check())
def test_fresh_truncated_jpeg(self):
self.screenshots(negative_jpeg()[:-2])
self.assertIn('invalid JPEG SOI/EOI', ' '.join(self.fresh_check()))
def test_fresh_garbage(self):
self.screenshots(b'garbage')
self.assertIn('missing OS screenshot', ' '.join(self.fresh_check()))
def test_fresh_dimension_mismatch(self):
self.screenshots(negative_jpeg())
next(self.out.rglob('screen-after.png')).write_bytes(negative_jpeg(width=640))
F.repin(self.out)
self.assertIn('OS screenshot dimension mismatch', ' '.join(self.fresh_check()))
def test_fresh_unbound_path(self):
ack = next(self.out.rglob('ack.json'))
data = C.load(ack)
data['screen_before'] = data['screen_after']
F.put(ack, data)
F.repin(self.out)
self.assertIn('R09 screenshot request binding', ' '.join(self.fresh_check()))
class R09ImageTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory(dir=ROOT/'tests/fixtures/p08-03/.scratch')