P23-03: 生活の観察

This commit is contained in:
Agent Control Runner
2026-10-16 14:24:10 +09:00
parent 6c13940cf4
commit 447f77a13b
2 changed files with 247 additions and 16 deletions
+77 -14
View File
@@ -5,6 +5,8 @@ Whole typed inputs are preserved. The existing typed codec is reused verbatim.
Unimplemented visual/OS checks are explicit blockers, never successful checks. Unimplemented visual/OS checks are explicit blockers, never successful checks.
""" """
import argparse import argparse
import contextlib
import tempfile
import copy import copy
import hashlib import hashlib
import importlib.util import importlib.util
@@ -1848,7 +1850,11 @@ def verify_composition(root, roles, source_map, inputs, read_git=None):
require(git('cat-file', '-t', commit).strip() == b'commit', 'role is not a commit') require(git('cat-file', '-t', commit).strip() == b'commit', 'role is not a commit')
require(git('cat-file', '-t', t).strip() == b'tree', 'T is not a tree') require(git('cat-file', '-t', t).strip() == b'tree', 'T is not a tree')
require(git('rev-parse', q+'^{tree}').decode().strip() == t, 'Q does not contain T') require(git('rev-parse', q+'^{tree}').decode().strip() == t, 'Q does not contain T')
require(git('rev-parse', 'HEAD').decode().strip() == q, 'checkout is not Q') head = git('rev-parse', 'HEAD').decode().strip()
if head not in (s, q):
require(git('rev-parse', 'HEAD^{tree}').decode().strip() == t, 'C tree differs from T')
require(git('rev-list', '--parents', '-n', '1', 'HEAD').decode().split() == [head, m, s],
'C parents must be exactly [M,S]')
require(not git('status', '--porcelain'), 'tested checkout is dirty') require(not git('status', '--porcelain'), 'tested checkout is dirty')
merged = git('merge-tree', '--write-tree', s, m).decode().splitlines() merged = git('merge-tree', '--write-tree', s, m).decode().splitlines()
require(merged and merged[0] == t, 'T differs from deterministic merge-tree') require(merged and merged[0] == t, 'T differs from deterministic merge-tree')
@@ -1875,11 +1881,10 @@ def verify_composition(root, roles, source_map, inputs, read_git=None):
def document(revision): def document(revision):
return strict_json(git('show', revision + ':' + CONTRACT)) return strict_json(git('show', revision + ':' + CONTRACT))
verify_marker_join(document(BASE_COMMIT), document(s), document(m), document(t)) verify_marker_join(document(BASE_COMMIT), document(s), document(m), document(t))
source_names = {p for p in tested if p.split('/', 1)[0] in runner = module('p23_composition_source_filter', ROOT/'tools/run_all_tests.py')
('autoload', 'scripts', 'scenes', 'data', 'tests', 'assets', 'tools') with tested_checkout(root, roles, git) as view:
and Path(p).name != '.DS_Store' and '__pycache__' not in Path(p).parts} require(source_map == {p: row['sha256'] for p, row in runner.source_refs(view).items()},
source_names.update(('project.godot', 'export_presets.cfg', 'Makefile', 'AGENTS.md')) 'incomplete tested-tree source/settings/QA map')
require(set(source_map) == source_names, 'incomplete tested-tree source/settings/QA map')
# Source refs include all QA suites and settings, not just S-changed files. # Source refs include all QA suites and settings, not just S-changed files.
required = set(source_map) | {p.split('#', 1)[0] for p in inputs} | {PLAN, CONTRACT, required = set(source_map) | {p.split('#', 1)[0] for p in inputs} | {PLAN, CONTRACT,
'project.godot', 'tests/business/test_required_kitchen.gd', 'tools/regression/suite-policy.json'} 'project.godot', 'tests/business/test_required_kitchen.gd', 'tools/regression/suite-policy.json'}
@@ -1889,7 +1894,8 @@ def verify_composition(root, roles, source_map, inputs, read_git=None):
entry = tested.get(path) entry = tested.get(path)
require(entry is not None and entry[1] == 'blob' and bindings[path] == entry[2], 'T blob binding: ' + path) require(entry is not None and entry[1] == 'blob' and bindings[path] == entry[2], 'T blob binding: ' + path)
content = git('show', t + ':' + path) content = git('show', t + ':' + path)
require((root/path).read_bytes() == content, 'checkout bytes differ from T: ' + path) if head != s or path in changed_s-changed_m:
require((root/path).read_bytes() == content, 'checkout bytes differ from T: ' + path)
digest = hashlib.sha256(content).hexdigest() digest = hashlib.sha256(content).hexdigest()
if path in source_map: if path in source_map:
require(source_map[path] == digest, 'evidence source is not T: ' + path) require(source_map[path] == digest, 'evidence source is not T: ' + path)
@@ -1898,6 +1904,41 @@ def verify_composition(root, roles, source_map, inputs, read_git=None):
return {k: roles[k] for k in names} return {k: roles[k] for k in names}
@contextlib.contextmanager
def tested_checkout(root, roles, read_git=None):
"""Disposable Git-blob view shared by the existing filters and readers."""
require(isinstance(roles, dict), 'missing externally bound S/M/T/Q')
tree = roles.get('tested_tree', '')
require(isinstance(tree, str) and re.fullmatch('[0-9a-f]{40}', tree), 'invalid T')
git = read_git or (lambda *args: git_bytes(root, *args))
with tempfile.TemporaryDirectory(prefix='p23-tested-tree-') as temporary:
view = Path(temporary)
for record in git('ls-tree', '-r', '-z', tree).split(b'\0'):
if not record:
continue
metadata, raw_path = record.split(b'\t', 1)
mode, kind, oid = metadata.decode().split()
path = Path(raw_path.decode())
require(not path.is_absolute() and '..' not in path.parts, 'unsafe T path')
require(kind == 'blob' and mode in ('100644', '100755'), 'T source aliases refused')
target = view/path
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(git('show', tree + ':' + path.as_posix()))
yield view
def root_context(directory, index, explicit=None):
path = directory/'root-composition.json'
require(path.is_file() and not path.is_symlink()
and path.resolve().is_relative_to(directory.resolve()), 'missing/unsafe Root composition context')
data = path.read_bytes()
require(index.get('root_composition_sha256') == hashlib.sha256(data).hexdigest(),
'Root composition context SHA mismatch')
roles = strict_json(data)
require(explicit is None or structural_equal(explicit, roles), 'explicit Root context disagrees')
return roles
def normative_plan_text(root): def normative_plan_text(root):
"""Requirement text is an input; explicitly labelled progress is metadata.""" """Requirement text is an input; explicitly labelled progress is metadata."""
text = (root / PLAN).read_text() text = (root / PLAN).read_text()
@@ -2097,6 +2138,19 @@ def inspect_native_phases(directory, index, root=ROOT):
def check(root, directory, root_roles=None): def check(root, directory, root_roles=None):
try:
index_path = directory/'index.json'
if not index_path.is_file():
return ['missing evidence index: ' + str(index_path)]
index = strict_json(index_path.read_bytes())
roles = root_context(directory, index, root_roles)
with tested_checkout(root, roles) as view:
return check_tested(root, directory, roles, view)
except (OSError, ValueError, KeyError, TypeError, AttributeError) as exc:
return ['invalid evidence index: ' + str(exc)]
def check_tested(checkout, directory, root_roles, root):
"""Never infer completion of unimplemented gates from an index assertion.""" """Never infer completion of unimplemented gates from an index assertion."""
deficits = [] deficits = []
executed = {c: {category: set() for category in CATEGORIES} for c in CONDITIONS} executed = {c: {category: set() for category in CATEGORIES} for c in CONDITIONS}
@@ -2110,7 +2164,10 @@ def check(root, directory, root_roles=None):
'historical v1 evidence has diagnostic applicability only') 'historical v1 evidence has diagnostic applicability only')
runner = module('p23_source_applicability', root/'tools/run_all_tests.py') runner = module('p23_source_applicability', root/'tools/run_all_tests.py')
require(index.get('source_map') == {n: v['sha256'] for n, v in runner.source_refs(root).items()}, 'exact frozen source map') require(index.get('source_map') == {n: v['sha256'] for n, v in runner.source_refs(root).items()}, 'exact frozen source map')
roles = verify_composition(root, root_roles, index['source_map'], input_map(root)) roles = verify_composition(checkout, root_roles, index['source_map'], input_map(root))
if git_bytes(checkout, 'rev-parse', 'HEAD').decode().strip() != roles['submitted_sha']:
require(index['source_map'] == {n: v['sha256'] for n, v in runner.source_refs(checkout).items()},
'exact checkout source map')
require(index.get('composition') == roles and index.get('commit') == roles['tested_commit'], require(index.get('composition') == roles and index.get('commit') == roles['tested_commit'],
'evidence role binding differs from Root S/M/T/Q; historical v1 is not T evidence') 'evidence role binding differs from Root S/M/T/Q; historical v1 is not T evidence')
if index.get('input_map') != input_map(root): if index.get('input_map') != input_map(root):
@@ -2127,7 +2184,10 @@ def check(root, directory, root_roles=None):
require(path.resolve().is_relative_to(directory.resolve()) and not path.is_symlink(), require(path.resolve().is_relative_to(directory.resolve()) and not path.is_symlink(),
'artifact escapes evidence directory') 'artifact escapes evidence directory')
require(path.is_file() and sha(path) == digest, 'artifact SHA mismatch: ' + name) require(path.is_file() and sha(path) == digest, 'artifact SHA mismatch: ' + name)
inspect_native_phases(directory, index, root) try:
inspect_native_phases(directory, index, root)
except (OSError, ValueError, KeyError, TypeError, AttributeError) as exc:
deficits.append('native evidence deficit: ' + str(exc))
for layout in ('A', 'B'): for layout in ('A', 'B'):
layout_semantic, layout_pixels = set(), set() layout_semantic, layout_pixels = set(), set()
stream = directory / ('layout-' + layout) stream = directory / ('layout-' + layout)
@@ -2199,8 +2259,7 @@ def main(argv=None):
parser.add_argument('--offline-mirror', type=Path) parser.add_argument('--offline-mirror', type=Path)
parser.add_argument('--layout-a', type=Path) parser.add_argument('--layout-a', type=Path)
parser.add_argument('--layout-b', type=Path) parser.add_argument('--layout-b', type=Path)
parser.add_argument('--root-composition', type=Path, parser.add_argument('--root-composition', type=Path)
default=os.environ.get('P23_ROOT_COMPOSITION'))
args = parser.parse_args(argv) args = parser.parse_args(argv)
check_setup_budget_source((args.project_root / 'tests/fixtures/p23-03/observation.gd').read_text()) check_setup_budget_source((args.project_root / 'tests/fixtures/p23-03/observation.gd').read_text())
if args.offline_mirror: if args.offline_mirror:
@@ -2208,13 +2267,17 @@ def main(argv=None):
'offline preparation must not be combined with acceptance checks') 'offline preparation must not be combined with acceptance checks')
print(json.dumps(inspect_offline_mirror(args.offline_mirror), ensure_ascii=False)) print(json.dumps(inspect_offline_mirror(args.offline_mirror), ensure_ascii=False))
return 2 return 2
if args.verify_only: if args.verify_only or (os.environ.get('BUILD_OUTPUT_DIR') and not any(
(args.diagnostics, args.layout_a, args.layout_b))):
build = os.environ.get('BUILD_OUTPUT_DIR') build = os.environ.get('BUILD_OUTPUT_DIR')
if not build: if not build:
print('BUILD_OUTPUT_DIR is required', file=sys.stderr) print('BUILD_OUTPUT_DIR is required', file=sys.stderr)
return 2 return 2
root_roles = strict_json(args.root_composition.read_bytes()) if args.root_composition else None try:
deficits = check(args.project_root, Path(build) / 'p23-life-observation-evidence', root_roles) root_roles = strict_json(args.root_composition.read_bytes()) if args.root_composition else None
deficits = check(args.project_root, Path(build) / 'p23-life-observation-evidence', root_roles)
except (OSError, ValueError) as exc:
deficits = ['invalid Root composition context: ' + str(exc)]
print(json.dumps(dict(task_id='P23-03', integrity='fail' if deficits else 'pass', print(json.dumps(dict(task_id='P23-03', integrity='fail' if deficits else 'pass',
deficits=deficits, acceptance='pending'), ensure_ascii=False)) deficits=deficits, acceptance='pending'), ensure_ascii=False))
return 1 if deficits else 0 return 1 if deficits else 0
+170 -2
View File
@@ -11,6 +11,8 @@ import hashlib
import struct import struct
import zlib import zlib
import tempfile import tempfile
import os
import subprocess
from pathlib import Path from pathlib import Path
from unittest import mock from unittest import mock
@@ -182,7 +184,12 @@ class Repair28CompositionTests(unittest.TestCase):
raise AssertionError(args) raise AssertionError(args)
def verify(self): def verify(self):
with mock.patch.object(Path, 'read_bytes', lambda p:self.files[self.t][str(p.relative_to(check.ROOT))]): original = Path.read_bytes
def read(path):
if path.is_relative_to(check.ROOT):
return self.files[self.t][str(path.relative_to(check.ROOT))]
return original(path)
with mock.patch.object(Path, 'read_bytes', read):
return check.verify_composition(check.ROOT,self.roles,self.source,self.inputs,self.git) 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): def test_valid_route_b_keeps_unchanged_allowed_path_from_main(self):
@@ -245,6 +252,167 @@ class Repair28CompositionTests(unittest.TestCase):
self.roles, self.source = original_roles, original_source self.roles, self.source = original_roles, original_source
class Repair30EntryTests(unittest.TestCase):
"""Synthetic labelled Git history; zero native or OS credit."""
def setUp(self):
runtime = os.environ.get('TMPDIR')
self.assertTrue(runtime, 'assigned TMPDIR required')
self.temporary = tempfile.TemporaryDirectory(dir=runtime, prefix='p23-repair30-')
self.addCleanup(self.temporary.cleanup)
self.repo = Path(self.temporary.name)/'repo'
self.repo.mkdir()
self.build = Path(self.temporary.name)/'build'
self.evidence = self.build/'p23-life-observation-evidence'
self.evidence.mkdir(parents=True)
self.git('init', '-q')
self.git('config', 'user.name', 'Synthetic fixture')
self.git('config', 'user.email', 'fixture@example.invalid')
for path in ('tools/run_all_tests.py', 'tests/fixtures/p23-03/observation.gd',
check.PLAN, check.SPEC, 'docs/traceability/original/BASELINE.txt'):
self.write(path, (check.ROOT/path).read_bytes())
for path in ('project.godot', 'export_presets.cfg', 'Makefile', 'AGENTS.md',
'scenes/main/world_preview.gd', 'tests/business/test_required_kitchen.gd',
'tools/regression/suite-policy.json'):
self.write(path, 'synthetic baseline\n')
self.marker = dict(schema_version=1, rows=[dict(test='history-'+str(i), required=True) for i in range(6)])
self.write_marker(self.marker)
self.base = self.commit('synthetic base')
submitted = copy.deepcopy(self.marker)
submitted['rows'].insert(0, dict(test='tests/ui/test_p23_life_observation.gd', required=True))
self.write_marker(submitted)
self.write('scenes/main/world_preview.gd', 'synthetic S\n')
self.s = self.commit('synthetic S')
self.git('checkout', '-q', '--detach', self.base)
main = copy.deepcopy(self.marker)
main['rows'].append(dict(test='P18-and-kitchen-QA', required=True))
self.write_marker(main)
self.write('tests/business/test_required_kitchen.gd', 'synthetic M QA\n')
self.m = self.commit('synthetic M')
self.t = self.git('merge-tree', '--write-tree', self.s, self.m).splitlines()[0]
self.q = self.git('commit-tree', self.t, '-p', self.s, '-m', 'synthetic Q, no evidence')
self.c = self.git('commit-tree', self.t, '-p', self.m, '-p', self.s, '-m', 'synthetic C')
self.git('checkout', '-q', '--detach', self.q)
runner = check.module('p23_synthetic_source_filter', check.ROOT/'tools/run_all_tests.py')
sources = {p: row['sha256'] for p, row in runner.source_refs(self.repo).items()}
inputs = check.input_map(self.repo)
required = set(sources) | {p.split('#')[0] for p in inputs} | {check.PLAN, check.CONTRACT,
'project.godot', 'tests/business/test_required_kitchen.gd', 'tools/regression/suite-policy.json'}
self.roles = dict(submitted_sha=self.s, main_context_sha=self.m, tested_tree=self.t,
tested_commit=self.q, blob_bindings={p:self.git('rev-parse', self.t+':'+p) for p in required})
self.index = dict(schema='p23-life-observation-v2', source_map=sources, input_map=inputs,
composition={k:self.roles[k] for k in ('submitted_sha','main_context_sha','tested_tree','tested_commit')},
commit=self.q, spec_sha256=check.SPEC_SHA,
marker_contract_sha256=check.sha(self.repo/check.CONTRACT), artifacts={}, phases=[])
self.publish()
self.git('checkout', '-q', '--detach', self.s)
self.patch = mock.patch.object(check, 'BASE_COMMIT', self.base)
self.patch.start()
self.addCleanup(self.patch.stop)
def git(self, *args):
result = subprocess.run(['git', '-C', str(self.repo), *args], capture_output=True, check=True)
return result.stdout.decode().strip()
def write(self, path, content):
target = self.repo/path
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(content.encode() if isinstance(content, str) else content)
def write_marker(self, value):
self.write(check.CONTRACT, json.dumps(value, indent=2)+'\n')
def commit(self, message):
self.git('add', '.')
self.git('commit', '-q', '-m', message)
return self.git('rev-parse', 'HEAD')
def publish(self):
context = self.evidence/'root-composition.json'
context.write_text(json.dumps(self.roles))
self.index['root_composition_sha256'] = check.sha(context)
(self.evidence/'index.json').write_text(json.dumps(self.index))
def cli(self, verify=True, extra_env=None, extra_args=()):
env = dict(BUILD_OUTPUT_DIR=str(self.build))
env.update(extra_env or {})
with mock.patch.dict(os.environ, env, clear=True), contextlib.redirect_stdout(io.StringIO()) as output:
code = check.main((['--verify-only'] if verify else []) + ['--project-root', str(self.repo)] + list(extra_args))
self.assertEqual(code, 1)
result = json.loads(output.getvalue())
self.assertEqual(result['acceptance'], 'pending')
return result['deficits']
def assert_entry(self, verify=True):
deficits = self.cli(verify)
self.assertTrue(any('native evidence deficit' in d for d in deficits), deficits)
self.assertTrue(any('missing OS A-writer process receipt' in d for d in deficits), deficits)
for condition in check.CONDITIONS:
self.assertTrue(any(d.startswith(condition+':') for d in deficits), deficits)
self.assertFalse(any('invalid evidence index' in d for d in deficits), deficits)
def test_normal_and_verify_only_S_Q_C_keep_all_seven_blockers(self):
for head in (self.s, self.q, self.c):
self.git('checkout', '-q', '--detach', head)
for verify in (False, True):
with self.subTest(head=head, verify=verify):
self.assert_entry(verify)
def test_wrong_roles_and_missing_Q(self):
for key, value in (('submitted_sha',self.base), ('main_context_sha',self.base),
('tested_tree',self.git('rev-parse', self.base+'^{tree}')),
('tested_commit','0'*40)):
original = self.roles[key]
self.roles[key] = value
self.publish()
with self.subTest(key=key):
self.assertTrue(any('invalid evidence index' in d for d in self.cli()))
self.roles[key] = original
def test_dirty_roles_and_source_QA_rollback(self):
for head in (self.s,self.q,self.c):
self.git('checkout', '-q', '--detach', head)
for path in ('scenes/main/world_preview.gd','tests/business/test_required_kitchen.gd'):
original = (self.repo/path).read_bytes()
self.write(path, 'synthetic rollback\n')
with self.subTest(head=head,path=path):
self.assertTrue(any('dirty' in d for d in self.cli()))
self.write(path, original)
def test_unexpected_C_parents_and_tree(self):
wrong_tree = self.git('rev-parse',self.base+'^{tree}')
for tree, parents in ((self.t,[self.s,self.m]), (self.t,[self.m,self.s,self.base]),
(self.t,[self.m]), (wrong_tree,[self.m,self.s])):
args = ['commit-tree', tree, '-m', 'synthetic invalid C']
for parent in parents:
args.extend(['-p',parent])
head = self.git(*args)
self.git('checkout','-q','--detach',head)
with self.subTest(tree=tree,parents=parents):
self.assertTrue(any('C parents' in d or 'C tree' in d for d in self.cli()))
def test_context_missing_mismatch_symlink_and_environment_ignored(self):
context = self.evidence/'root-composition.json'
original = context.read_bytes()
context.unlink()
self.assertTrue(any('context' in d for d in self.cli(extra_env={'P23_ROOT_COMPOSITION':str(self.repo/'unused')})))
context.write_bytes(original+b' ')
self.assertTrue(any('SHA mismatch' in d for d in self.cli()))
context.unlink()
target = Path(self.temporary.name)/'context.json'
target.write_bytes(original)
context.symlink_to(target)
self.assertTrue(any('unsafe' in d for d in self.cli()))
context.unlink()
context.write_bytes(original)
alternate = Path(self.temporary.name)/'alternate.json'
alternate.write_text('{}')
self.assertTrue(any('disagrees' in d for d in self.cli(extra_args=['--root-composition',str(alternate)])))
self.assertTrue(any('native evidence deficit' in d for d in self.cli(extra_env={'P23_ROOT_COMPOSITION':str(alternate)})))
self.index.pop('root_composition_sha256')
(self.evidence/'index.json').write_text(json.dumps(self.index))
self.assertTrue(any('SHA mismatch' in d for d in self.cli()))
class ActorNonoverlapTests(unittest.TestCase): class ActorNonoverlapTests(unittest.TestCase):
def report(self, second_at): def report(self, second_at):
return dict(actor_drawings=[ return dict(actor_drawings=[
@@ -600,7 +768,7 @@ class SnapshotCellTests(unittest.TestCase):
class ImageTests(unittest.TestCase): class ImageTests(unittest.TestCase):
def test_collector_conversion_is_bound_to_each_cua_screenshot(self): 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') 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: with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary) root = Path(temporary)
pair = root/'os-pair/A' pair = root/'os-pair/A'
request_dir = pair/'writer-request-0001' request_dir = pair/'writer-request-0001'