P04-01: 同時発生の更新規約

This commit is contained in:
Agent Control Runner
2026-10-18 18:11:13 +09:00
parent e7f2271164
commit 2cd734062a
16 changed files with 1773 additions and 17 deletions
@@ -1,3 +1,24 @@
## RUN47 current admission (supersedes earlier execution instructions)
Current S4 check-only and native call the same engine-free `admission` function.
The real selected source map determines the sorted unique res:// `.gd` roster;
the whole list must match, including membership and order. Product is exactly
selected minus `tools/check_p04_order_evidence.py`; no other extra is allowed.
There is no current fixed product/script count. The supplied M copy happens to
select 1368 product files and 329 scripts; fresh Q additionally selects the checker.
Plans now carry `parse_paths`, and FreshSeries checks plan/actual/source equality.
Before import is attempted, cleanup requires the complete original inventory and
no generated UID. After import starts, the existing exact planned UID seal applies.
Failure stays failure; no previous native receipt is adopted.
The older relocation, counts, runs and receipts below are historical context only.
The historical r28 wrapper and data retain 1338/316. `receipt_lifecycle_fixtures.py`
now explicitly loads that pinned historical wrapper. Current-derived finite cases
are in `admission_controls.py`, using the retained verified-M copy and BUILD scratch.
Root must freeze final Q and run six clean check-only plans, then all fresh native
boundaries and ten S5 cases once in one heavy lease. See REPAIR47_PLAN.json and
REPAIR47_STATUS.md; worker does not execute that native sequence.
# S4 finite Root collection r28 — NOT RUN
This adds only the existing `power_stop` evidence obligation. It adds no counted
@@ -141,7 +162,7 @@ the descriptor it just appended and the index of that exact observation.
| `write_bytes` to `bootstrap/<name>` | 1 per pinned bootstrap key | The reviewed map names `profile_guard.gd` and `parse_full_dependencies.gd`; setup only |
| `/bin/cp -c` to `stage/.godot/<name>` | 1 per cache inventory key | Optional eligible-cache branch only; fresh mode never accesses cache |
| `cache-reuse.json` | 0 or 1 | Only after all optional cache clones and guards |
| `parse-paths.json` | 1 | Setup after 316-script guard |
| `parse-paths.json` | 1 | Setup after source-derived roster admission (historical r28 used 316) |
| `contracts.json` | 1 | Setup after unchanged suite-policy guard |
| `runner-input.json` | 1 | Setup; no final read/rewrite; absent if earlier setup fails |
| `engine-identity-before-<phase>.json` | 1 each for import, parse, writer, reader | Immediately after that phase's repin; failure stops loop, never resumes it |
@@ -0,0 +1,164 @@
#!/usr/bin/env python3
"""Finite engine-free admission controls. Synthetic plans are never evidence."""
import argparse
import ast
from contextlib import redirect_stdout
import hashlib
import io
import json
from pathlib import Path
import sys
from types import SimpleNamespace
from unittest.mock import patch
sys.dont_write_bytecode = True
import run_s4_native as s4
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--project-root', required=True, type=Path)
parser.add_argument('--scratch', required=True, type=Path)
args = parser.parse_args()
root = args.project_root.resolve(strict=True)
expected_root = Path(__file__).resolve().parents[6] / 'build/owner-handoff/p04-after-p26-closure-r01/verified-m-engine-free-r45'
assert root == expected_root.resolve(strict=True), 'retained M only; never Root Q'
scratch = args.scratch.resolve()
assert scratch.is_relative_to(s4.BUILD.resolve()) and not scratch.exists()
scratch.mkdir(parents=True)
r = s4.module('admission_real_selection', root / 'tools/run_all_tests.py')
selected = r.source_refs(root)
product = {n: p for n, p in selected.items() if n != s4.CHECKER}
paths = ['res://' + n for n in sorted(product) if n.endswith('.gd')]
results = []
def check(label, action, reject=False):
try:
action()
except (ValueError, AssertionError):
assert reject, label
results.append(dict(case=label, result='rejected'))
else:
assert not reject, label
results.append(dict(case=label, result='pass'))
assert len(paths) == 329, 'diagnostic pin for supplied M, not production admission'
check('real-329', lambda: s4.admission(selected, product, paths))
check('selected-checker-only-extra', lambda: s4.admission(
dict(product, **{s4.CHECKER: {'sha256': 'synthetic-checker'}}), product, paths))
other = dict(product, **{'scripts/another_valid.gd': {'sha256': 'synthetic'}})
other_paths = sorted(paths + ['res://scripts/another_valid.gd'])
check('another-derived-roster', lambda: s4.admission(other, other, other_paths))
mutations = {
'missing': paths[:-1],
'extra': sorted(paths + ['res://scripts/extra.gd']),
'same-count-substitution': sorted(paths[:-1] + ['res://scripts/substitute.gd']),
'duplicate': sorted(paths + [paths[0]]),
'non-gd': sorted(paths + ['res://project.godot']),
'stale-literal-count': paths[:316],
}
for label, roster in mutations.items():
check(label, lambda roster=roster: s4.admission(selected, product, roster), True)
check('checker-as-product', lambda: s4.admission(
dict(selected, **{s4.CHECKER: {}}), dict(product, **{s4.CHECKER: {}}), paths), True)
check('missing-product', lambda: s4.admission(selected, dict(list(product.items())[1:]), paths), True)
check('extra-selected-input', lambda: s4.admission(dict(selected, unexpected={}), product, paths), True)
# Execute the real check-only main with explicit historical template/engine
# stubs. Source selection and admission are real; no fresh-Q claim is made.
real_admission = s4.admission
calls = []
def observed(a, b, c):
calls.append((a, b, c))
return real_admission(a, b, c)
argv = ['run_s4_native.py', '--historical-r17', '--project-root', str(root),
'--output', str(scratch / 'never-native'), '--plan', str(scratch / 'synthetic-plan.json'), '--check-only']
with patch.object(s4, 'engine_pin', return_value={'synthetic': True}), \
patch.object(s4, 'template_guard'), patch.object(s4, 'MATERIALS', []), \
patch.object(s4, 'CONTRACTS', s4.HERE.parent.parent / 'completion_marker_contracts.json'), \
patch.object(s4, 'admission', side_effect=observed), \
patch.object(sys, 'argv', argv), redirect_stdout(io.StringIO()):
assert s4.main() == 0
assert len(calls) == 1 and calls[0][2] == paths
assert not (scratch / 'never-native').exists()
assert r.read(scratch / 'synthetic-plan.json')['parse_paths'] == paths
results.append(dict(case='real-check-only-shared-admission', result='pass'))
for label, roster in mutations.items():
def mutated(a, b, c):
return real_admission(a, b, roster)
bad_plan = scratch / ('rejected-plan-' + label + '.json')
bad_argv = list(argv)
bad_argv[bad_argv.index('--plan') + 1] = str(bad_plan)
with patch.object(s4, 'engine_pin', return_value={'synthetic': True}), \
patch.object(s4, 'template_guard'), patch.object(s4, 'admission', side_effect=mutated), \
patch.object(sys, 'argv', bad_argv):
check('check-only-' + label, s4.main, True)
assert not bad_plan.exists(), 'admission must precede plan write'
# Historical admission executes the original pinned r28 AST unchanged.
series = s4.module('admission_series', s4.HERE.parent / 'p04_native_series.py')
series_tree = ast.parse((s4.HERE.parent / 'p04_native_series.py').read_text())
fresh = next(n for n in series_tree.body if isinstance(n, ast.ClassDef) and n.name == 'FreshSeries')
roster_block = next(n for n in ast.walk(fresh) if isinstance(n, ast.If)
and "'S4 plan/actual parse roster'" in ast.unparse(n))
roster_code = compile(ast.Module(body=[roster_block], type_ignores=[]), '<actual FreshSeries roster block>', 'exec')
wrapper = s4.module('admission_fresh_wrapper', s4.HERE.parent / 's5/run_s5_native.py')
def series_roster(actual, planned):
obj = SimpleNamespace(name='s4', read=lambda _: json.dumps(actual).encode(), inputs={'parse_paths': planned})
exec(roster_code, dict(wrapper=wrapper, HERE=s4.HERE.parent, selected=selected,
self=obj, loads=series.loads, require=series.require))
check('FreshSeries-exact-roster', lambda: series_roster(paths, paths))
check('FreshSeries-actual-substitution', lambda: series_roster(mutations['same-count-substitution'], paths), True)
check('FreshSeries-plan-substitution', lambda: series_roster(paths, mutations['same-count-substitution']), True)
main_tree = ast.parse(Path(s4.__file__).read_text())
cache_guard = next(n for n in ast.walk(main_tree) if isinstance(n, ast.Expr)
and 'cache selection is exact product plus checker' in ast.unparse(n))
cache_code = compile(ast.Module(body=[cache_guard], type_ignores=[]), '<actual cache set guard>', 'exec')
def cache_keys(old):
exec(cache_code, dict(r=r, old_map=old, current_map=product, CHECKER=s4.CHECKER))
s4.admission(old, product, paths)
cache = dict(product, **{s4.CHECKER: {}})
check('cache-exact-product-plus-checker', lambda: cache_keys(cache))
check('cache-missing-checker', lambda: cache_keys(product), True)
check('cache-same-count-substitution', lambda: cache_keys(dict(list(cache.items())[1:], unexpected={})), True)
historical = s4.HERE / 'historical/run_s4_native_r28.py.reference'
series.verify_s4_wrapper(hashlib.sha256(historical.read_bytes()).hexdigest(), historical=True)
tree = ast.parse(historical.read_text())
guard = next(n for n in ast.walk(tree) if isinstance(n, ast.Expr)
and 'complete 316-script parse roster' in ast.unparse(n))
code = compile(ast.Module(body=[guard], type_ignores=[]), str(historical), 'exec')
# A finite synthetic historical receipt; not a replacement for native raw.
check('historical-316-receipt', lambda: exec(code, {'r': r, 'paths': paths[:316]}))
check('historical-rejects-current-roster', lambda: exec(code, {'r': r, 'paths': paths}), True)
uid_plan, snapshot, _ = s4.module('admission_uid', s4.HERE.parent / 's5/run_s5_native.py').load_uid_helper(root, r, selected)
for label in ('original-no-uid', 'partial-import', 'unknown-uid', 'changed-source'):
stage = scratch / label
(stage / 'tests/special').mkdir(parents=True)
(stage / 'tests/special/a.gd').write_text('extends Node\n')
(stage / 'tests/special/b.gd').write_text('extends Node\n')
source = r.inventory(stage, stage=True)
refs = {n: dict(p, path=str(stage / n)) for n, p in source.items()}
plan = uid_plan(refs)
assert len(plan) >= 2
if label == 'partial-import':
(stage / next(iter(plan))).write_text('uid://partial\n')
elif label == 'unknown-uid':
(stage / 'unknown.uid').write_text('uid://unknown\n')
elif label == 'changed-source':
(stage / 'tests/special/a.gd').write_text('changed\n')
state = dict(phase=None, accepted=None)
check('pre-import-' + label, lambda: s4.cleanup_uid_guard(r, snapshot, stage, source, plan, state), label != 'original-no-uid')
if label != 'original-no-uid':
state['phase'] = 'import'
check('attempted-import-' + label, lambda: s4.cleanup_uid_guard(r, snapshot, stage, source, plan, state), True)
result = dict(passed=True, controls=results, source_files=len(selected),
product_files=len(product), scripts=len(paths), engine_launches=0,
formal_acceptance=False)
(scratch / 'results.json').write_text(json.dumps(result, indent=2) + '\n')
print(json.dumps(result, indent=2))
if __name__ == '__main__':
main()
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Root-only synthetic receipt lifecycle fixtures. No engine/process is started.
"""Historical-pinned r28 synthetic lifecycle fixtures. No engine/process starts.
Uses the real main, exclusive JSON writer, source_refs and inventories. The
identity section is compiled directly from S4Series.__init__, not reimplemented.
@@ -10,6 +10,7 @@ from collections import Counter
from contextlib import ExitStack, redirect_stdout
import copy
import importlib.util
from importlib.machinery import SourceFileLoader
import io
import json
from pathlib import Path
@@ -21,17 +22,18 @@ from unittest.mock import patch
sys.dont_write_bytecode = True
HERE = Path(__file__).resolve().parent
ROOT = next(p for p in HERE.parents if (p / 'tools/run_all_tests.py').is_file())
WORKTREE = next(p for p in HERE.parents if (p / 'tools/run_all_tests.py').is_file())
ROOT = WORKTREE / 'build/owner-handoff/p04-after-p26-closure-r01/verified-m-engine-free-r45'
def load(name, path):
spec = importlib.util.spec_from_file_location(name, path)
spec = importlib.util.spec_from_file_location(name, path, loader=SourceFileLoader(name, str(path)))
value = importlib.util.module_from_spec(spec)
spec.loader.exec_module(value)
return value
s4 = load('receipt_s4', HERE / 'run_s4_native.py')
s4 = load('receipt_s4_historical', HERE / 'historical/run_s4_native_r28.py.reference')
r = load('receipt_standard', ROOT / 'tools/run_all_tests.py')
series = load('receipt_series', HERE.parent / 'p04_native_series.py')
@@ -78,7 +80,7 @@ def case(base, failure=None):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
# Real 1338-entry source_refs and 316-entry script roster, synthetic bytes.
# Historical-pinned r28 fixture only. Current-derived cases: admission_controls.py.
scripts = [s4.TEST, 'tests/transport/test_runtime_transport.gd',
'tests/staff/test_rescue_elevator.gd']
scripts += ['scripts/fixture_%03d.gd' % i for i in range(313)]
@@ -92,7 +94,7 @@ def case(base, failure=None):
assert len(r.source_refs(product)) == 1338
qa = staged / 'p04_s4_power_probe.gd'
put(qa, b'extends Node\n# synthetic override\n')
put(staged / 'run_s4_native.py', (HERE / 'run_s4_native.py').read_bytes())
put(staged / 'run_s4_native.py', (HERE / 'historical/run_s4_native_r28.py.reference').read_bytes())
put(staged / 'S4_ROOT_RUN.md', b'synthetic fixture material\n')
contracts = base / 'contracts.json'
r.write(contracts, {})
@@ -97,6 +97,28 @@ def engine_pin(path, expected=None):
raise ValueError("engine identity changed")
return value
CHECKER = "tools/check_p04_order_evidence.py"
def admission(selected, product, paths):
"""Engine-free exact selection and parse admission; no historical counts."""
if CHECKER in product:
raise ValueError("checker is not a product input")
if set(selected) - set(product) not in (set(), {CHECKER}):
raise ValueError("only checker may be a selected extra")
if {n: p for n, p in selected.items() if n != CHECKER} != product:
raise ValueError("exact selected/product source map")
expected = ["res://" + n for n in sorted(product) if n.endswith(".gd")]
if paths != expected:
raise ValueError("exact source-derived parse roster")
return paths
def cleanup_uid_guard(r, snapshot, stage, source, plan, state):
if state["phase"] is None:
r.require(state["accepted"] is None and r.inventory(stage, stage=True) == source,
"exact original source and no generated UID before import")
else:
r.require(snapshot(stage, source, plan) == state["accepted"], "UID seal after cleanup")
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--project-root", type=Path, default=PRODUCT)
@@ -147,6 +169,9 @@ def main():
refs = dict(source_before)
refs[TEST] = pin(QA)
template_guard(root, historical=not fresh)
product_refs = {n: p for n, p in refs.items() if n != CHECKER}
paths = ["res://" + n for n in sorted(product_refs) if n.endswith(".gd")]
admission(refs, product_refs, paths)
expected = {n: dict(bytes=p["bytes"], sha256=p["sha256"]) for n, p in refs.items()}
fixed = [pin(p) for p in ([] if fresh else MATERIALS) + [
QA, Path(__file__), HERE / "S4_ROOT_RUN.md",
@@ -171,7 +196,7 @@ def main():
break
plan = dict(protocol="p04-s4-preparation-v1", project_root=str(root), output=str(out), engine_path=str(args.godot.resolve()),
source_before=source_before, selected_source_map=selected["map"],
QA_overrides=[TEST], fixed_input_pins=fixed,
QA_overrides=[TEST], parse_paths=paths, fixed_input_pins=fixed,
cache_run=str(args.cache_run) if args.cache_run is not None else None,
import_mode="cache" if args.cache_run is not None else "fresh",
engine_sha256=ENGINE_SHA256, engine=engine_descriptor,
@@ -237,7 +262,9 @@ def main():
r.require(cache_input["engine_sha256"] == ENGINE_SHA256, "cache engine identity")
old_map = cache_input["selected_source_map"]
current_map = {n: p["sha256"] for n, p in source_before.items()}
r.require({k: v for k, v in old_map.items() if k != "tools/check_p04_order_evidence.py"} == current_map and len(old_map) == 1339, "immutable full cache: all 1338 product inputs equal; checker is not parsed/executed by product suites")
r.require(set(old_map) == (set(current_map) - {CHECKER}) | {CHECKER},
"cache selection is exact product plus checker")
admission(old_map, {k: v for k, v in current_map.items() if k != CHECKER}, paths)
protected = r.default_player(root)
outer = module("s4_owned_outer", r.SUPPORT / "owned_outer.py")
r.require(not outer.native.helper["live_engines"](), "another Godot owns CPU")
@@ -297,8 +324,7 @@ def main():
r.write(out / "cache-reuse.json", dict(ended_run=str(old), receipts=cache_receipts,
whole_cache=cache_before, independent_inode=True,
ordinary_import_required=True, ordinary_parse_required=True))
paths = ["res://" + n for n in sorted(refs) if n.endswith(".gd")]
r.require(len(paths) == 316, "complete 316-script parse roster")
admission(refs, product_refs, paths)
r.write(out / "parse-paths.json", paths)
markers = module("s4_markers", root / "tools/strict_test_markers.py")
contracts = markers.load_contracts(root / shared.C_REL if fresh else CONTRACTS)
@@ -377,7 +403,7 @@ def main():
# Fixed guard always follows ownership cleanup, even if setup or an engine failed.
try:
fixed_guard()
r.require(uid_snapshot(stage, uid_source, uid_plan_map) == uid_state["accepted"], "UID seal after cleanup")
cleanup_uid_guard(r, uid_snapshot, stage, uid_source, uid_plan_map, uid_state)
r.write(out / "fresh-engine-checks.json", engine_checks)
after = r.source_refs(root)
r.write(out / "source_after.json", after)