P13-02-CODE1: P13-02 背面対象参照・占有・保存の非描写機構（全原成果の一部、加算なし）

docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json

@@ -3784,7 +3784,7 @@
       "template": "Peeping room runtime: <checks> checks, <failures> failures",
       "required": true,
       "fixed_check_count_used": false,
-      "source_sha256_at_schema_audit": "30ed2fe11cc6b0353e9a3a6d995cf8690d322858d2c3dbb38ef32fd941072d2d",
+      "source_sha256_at_schema_audit": "924efd34414c12a325a1b700f799aef0396de9f15441fd88a3608292ff221657",
       "authority": {
         "path": "docs/plans/original-completion-2026-10-02/execution/p13-02/CODE1_TASK_SPEC.json",
         "scope": "new backend suite marker; native execution pending"

tests/special/test_p13_uid_policy.py

@@ -0,0 +1,96 @@
+"""Static UID policy cases; no engine, native worker or regression invocation."""
+import importlib.util
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+
+sys.dont_write_bytecode = True
+TOOLS = Path(__file__).resolve().parents[2] / 'tools'
+sys.path.insert(0, str(TOOLS))
+spec = importlib.util.spec_from_file_location('p13_uid_policy', TOOLS / 'run_p13_reference_tests.py')
+runner = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(runner)
+
+
+class UIDPolicyTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.stage = Path(self.tmp.name)
+        self.owner = 'tests/special/example.gd'
+        self.other = 'scripts/special/peeping_room_service.gd'
+        for name in (self.owner, self.other):
+            self.write(name, b'extends RefCounted\n')
+        self.write('project.godot', b'[application]\n')
+        self.write('scripts/existing.gd.uid', b'uid://existing\n')
+        self.source = runner.shared.inventory(self.stage, stage=True)
+        self.plan = runner.uid_plan(self.source)
+        self.write(self.owner + '.uid', b'uid://first\n')
+        self.write(self.other + '.uid', b'uid://second\n')
+
+    def write(self, name, raw):
+        path = self.stage / name
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(raw)
+
+    def snapshot(self):
+        return runner.uid_snapshot(self.stage, self.source, self.plan)
+
+    def test_accept_and_owner_seal(self):
+        accepted = self.snapshot()
+        self.assertEqual(set(accepted), {self.owner + '.uid', self.other + '.uid'})
+        pin = accepted[self.owner + '.uid']
+        self.assertEqual(pin['raw_bytes'], list(b'uid://first\n'))
+        self.assertEqual(pin['text'], 'uid://first\n')
+        self.assertEqual(pin['owner_sha256'], self.source[self.owner]['sha256'])
+        self.assertEqual(pin['sha256'], runner.shared.sha(self.stage / (self.owner + '.uid')))
+
+    def test_unknown_and_extra(self):
+        for name in ('scripts/unknown.gd.uid', 'extra.txt'):
+            with self.subTest(name=name):
+                self.write(name, b'uid://extra\n')
+                with self.assertRaises(AssertionError):
+                    self.snapshot()
+                (self.stage / name).unlink()
+
+    def test_missing(self):
+        (self.stage / (self.owner + '.uid')).unlink()
+        with self.assertRaises(AssertionError):
+            self.snapshot()
+
+    def test_malformed(self):
+        for raw in (b'uid://UPPER\n', b'uid://first', b'uid://first\nextra',
+                    b'uid://12345678901234\n', b'uid://\n', b'uid://first\r\n'):
+            with self.subTest(raw=raw):
+                self.write(self.owner + '.uid', raw)
+                with self.assertRaises(AssertionError):
+                    self.snapshot()
+
+    def test_duplicate_and_existing_collision(self):
+        for raw in (b'uid://second\n', b'uid://existing\n'):
+            with self.subTest(raw=raw):
+                self.write(self.owner + '.uid', raw)
+                with self.assertRaises(AssertionError):
+                    self.snapshot()
+
+    def test_modified_original(self):
+        for name in (self.owner, 'project.godot', 'scripts/existing.gd.uid'):
+            with self.subTest(name=name):
+                path = self.stage / name
+                original = path.read_bytes()
+                path.write_bytes(original + b'tamper')
+                with self.assertRaises(AssertionError):
+                    self.snapshot()
+                path.write_bytes(original)
+
+    def test_symlink(self):
+        path = self.stage / (self.owner + '.uid')
+        path.unlink()
+        path.symlink_to(self.stage / (self.other + '.uid'))
+        with self.assertRaises(AssertionError):
+            self.snapshot()
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/special/test_peeping_room_runtime.gd

@@ -12,10 +12,43 @@ func _run() -> void:
 	expect(before == var_to_bytes(runtime.create_snapshot()), "validator changes no whole Runtime bytes")
 	var public_validation: CommandResult = runtime.submit_command(&"peeping_room_validate", validation.payload)
 	expect(public_validation.is_success(), "registered validation command runs through real queue")
-	var domain_before := [runtime.world_service.create_snapshot(), runtime.economy_service.create_snapshot(), runtime.id_generator.create_snapshot(), runtime.peeping_room_service.create_snapshot()]
+	var placement := SimulationCommand.new()
+	placement.command_type = &"peeping_room_place"
+	placement.payload = validation.payload.duplicate(true)
+	var placement_before := var_to_bytes(runtime.create_snapshot())
+	var direct_unavailable: CommandResult = runtime._execute_peeping_room(placement)
+	expect(not direct_unavailable.is_success() and direct_unavailable.details[&"reason"] == &"BUILD_DEFINITION_PENDING_D04", "direct production placement unavailable with precise reason")
+	expect(placement_before == var_to_bytes(runtime.create_snapshot()), "direct placement adds no room, occupancy, cost or ID in whole typed Runtime")
+	var domain_before := var_to_bytes([runtime.world_service.create_snapshot(), runtime.economy_service.create_snapshot(), runtime.peeping_room_service.create_snapshot()])
+	var ids_before: Dictionary = runtime.id_generator.create_snapshot()
+	var counters_before: Dictionary = ids_before["counters"]
+	var event_counter_before: int = counters_before.get("domain_event", 0)
+	var event_ids_before := {}
+	for event: SimulationDomainEvent in runtime.domain_events.recent_events():
+		event_ids_before[event.event_id] = true
 	var unavailable: CommandResult = runtime.submit_command(&"peeping_room_place", validation.payload)
 	expect(not unavailable.is_success() and unavailable.details[&"reason"] == &"BUILD_DEFINITION_PENDING_D04", "production placement unavailable with precise reason")
-	expect(var_to_bytes(domain_before) == var_to_bytes([runtime.world_service.create_snapshot(), runtime.economy_service.create_snapshot(), runtime.id_generator.create_snapshot(), runtime.peeping_room_service.create_snapshot()]), "placement adds no room, occupancy, cost or ID (normal queue receipt only)")
+	expect(domain_before == var_to_bytes([runtime.world_service.create_snapshot(), runtime.economy_service.create_snapshot(), runtime.peeping_room_service.create_snapshot()]), "queued placement adds no room, occupancy or cost")
+	var ids_after: Dictionary = runtime.id_generator.create_snapshot()
+	var counters_after: Dictionary = ids_after["counters"]
+	expect(ids_before["version"] == ids_after["version"], "queued placement preserves ID generator version")
+	for prefix: Variant in counters_before:
+		if prefix != "domain_event":
+			expect(counters_after.has(prefix) and counters_after.get(prefix) == counters_before[prefix], "queued placement consumes no non-event ID or removes prefix: %s" % prefix)
+	for prefix: Variant in counters_after:
+		if prefix != "domain_event":
+			expect(counters_before.has(prefix), "queued placement adds no non-event ID prefix: %s" % prefix)
+	expect(counters_after.has("domain_event") and counters_after["domain_event"] == event_counter_before + 1, "queued placement allocates exactly one normal receipt event ID")
+	var new_events: Array[SimulationDomainEvent] = []
+	for event: SimulationDomainEvent in runtime.domain_events.recent_events():
+		if not event_ids_before.has(event.event_id):
+			new_events.append(event)
+	expect(new_events.size() == 1, "queued placement publishes exactly one new receipt event")
+	if new_events.size() == 1:
+		var receipt: SimulationDomainEvent = new_events[0]
+		expect(receipt.event_id == StringName("domain_event.%012d" % (event_counter_before + 1)), "placement receipt uses the next domain event ID")
+		expect(receipt.event_type == &"COMMAND_FAILED", "unavailable placement receipt is COMMAND_FAILED")
+		expect(receipt.causation_command_id == unavailable.command_id, "placement receipt causation matches rejected command")
 	var room := place(f)
 	enable_fixture(f, room)
 	var blocked: CommandResult = runtime.submit_command(&"peeping_room_record_evidence", {&"room_id": room, &"source_event_key": &"fixture.command"})

tools/run_p13_reference_tests.py

@@ -5,6 +5,9 @@ import ast
 import inspect
 import json
 import os
+import re
+import stat
+import types
 from pathlib import Path
 import shutil
 import sys
@@ -13,6 +16,7 @@ sys.dont_write_bytecode = True
 import run_all_tests as shared
 
 PROTOCOL = 'guarded-p13-reference-v1'
+CANONICAL_EVIDENCE = 'p13-reference-native-uid-corrected'
 NEW_TESTS = [
     'tests/special/test_peeping_room_geometry.gd',
     'tests/special/test_peeping_room_reference.gd',
@@ -34,12 +38,112 @@ ROSTER = NEW_TESTS + [
 ]
 
 
+def uid_plan(refs):
+    return {name + '.uid': {'owner': name, 'owner_sha256': pin['sha256'],
+                             'file_type': 'regular'}
+            for name, pin in sorted(refs.items())
+            if name.endswith('.gd') and name + '.uid' not in refs
+            and (name.startswith('tests/special/')
+                 or name == 'scripts/special/peeping_room_service.gd')}
+
+
+def uid_snapshot(stage, source, plan):
+    """Accept exactly the source inventory and finite native UID plan."""
+    actual = shared.inventory(stage, stage=True)
+    shared.require(set(actual) == set(source) | set(plan), 'exact source and planned UID set')
+    shared.require({n: actual[n] for n in source} == source, 'original source/settings unchanged')
+    seen = set()
+    for name in source:
+        if name.endswith('.uid'):
+            raw = (stage / name).read_bytes().rstrip(b'\r\n')
+            shared.require(raw not in seen, 'existing project UID collision')
+            seen.add(raw)
+    result = {}
+    for name, binding in plan.items():
+        path = stage / name
+        shared.require(stat.S_ISREG(path.lstat().st_mode) and not path.is_symlink()
+                       and path.stat().st_nlink == 1, 'ordinary generated UID')
+        raw = path.read_bytes()
+        shared.require(re.fullmatch(rb'uid://[a-z0-9]{1,13}\n', raw) is not None,
+                       'complete canonical UID bytes')
+        shared.require(raw.rstrip(b'\n') not in seen, 'generated UID collision')
+        seen.add(raw.rstrip(b'\n'))
+        shared.require(stat.S_ISREG((stage / binding['owner']).lstat().st_mode)
+                       and shared.sha(stage / binding['owner']) == binding['owner_sha256'],
+                       'UID owner content binding')
+        result[name] = dict(binding, bytes=len(raw), raw_bytes=list(raw),
+                            sha256=shared.sha(path), text=raw.decode('ascii'))
+    return result
+
+
+def install_uid_policy(namespace):
+    """Specialize private globals; shared and P18 functions remain untouched."""
+    original_collect, original_seal = namespace['collect'], namespace['seal']
+    state = {}
+
+    def reports(stage, expected, previous, requested):
+        shared.require(not requested and not previous, 'P13 has no generated reports')
+        snapshot = uid_snapshot(stage, state['source'], state['plan'])
+        if state['accepted'] is None:
+            shared.require(state['phase'] == 'import', 'UID generation only during import')
+            state['accepted'] = snapshot
+            expected.update({n: {'bytes': p['bytes'], 'sha256': p['sha256']}
+                             for n, p in snapshot.items()})
+        shared.require(snapshot == state['accepted'], 'sealed UID bytes unchanged')
+        return shared.declared_reports(stage, expected, previous, requested)
+
+    phase_globals = dict(vars(shared), declared_reports=reports)
+    execute = types.FunctionType(shared.execute_phase.__code__, phase_globals)
+
+    def phase(*args):
+        stage, out, label = args[1:4]
+        state['phase'] = label
+        if label == 'import':
+            shared.require(shared.inventory(stage, stage=True) == state['source'],
+                           'original stage before import')
+            before = {}
+            shared.write(out / 'uid-plan.json', state['plan'])
+        else:
+            before = uid_snapshot(stage, state['source'], state['plan'])
+            shared.require(before == state['accepted'], 'UID set before phase unchanged')
+        row = execute(*args)
+        after = uid_snapshot(stage, state['source'], state['plan'])
+        shared.require(after == state['accepted'], 'UID set after phase unchanged')
+        shared.write(out / (label + '-uids.json'), {'before': before, 'after': after})
+        return row
+
+    namespace['shared'] = types.SimpleNamespace(**dict(vars(shared), execute_phase=phase))
+
+    def collect(args, root, out, refs, *rest):
+        state.update(source=namespace['content_map'](refs), plan=uid_plan(refs), accepted=None)
+        return original_collect(args, root, out, refs, *rest)
+
+    def seal(out, refs, engine, contract_sha):
+        shared.write(out / 'accepted-uids.json', state['accepted'])
+        return original_seal(out, refs, engine, contract_sha)
+
+    def retained(stage, source, out):
+        plan = namespace['read_json'](out / 'uid-plan.json')
+        shared.require(plan == uid_plan(source), 'independently derived exact UID plan')
+        accepted = uid_snapshot(stage, source, plan)
+        shared.require(namespace['read_json'](out / 'accepted-uids.json') == accepted,
+                       'retained raw UID bytes and owner bindings')
+        for label in ['import', 'parse'] + [str(i).zfill(3) for i in range(len(ROSTER))]:
+            shared.require(namespace['read_json'](out / (label + '-uids.json')) ==
+                           {'before': {} if label == 'import' else accepted, 'after': accepted},
+                           'all phase UID before/after seals')
+        return True
+
+    namespace.update(collect=collect, seal=seal, uid_retained=retained)
+
+
 def protocol_core():
     """Reuse P18's complete guarded core without editing or copying its runner.
 
-    P18 predates configurable protocol/roster constants. Specialize ONLY its
-    three entry functions' identity literals and explicit roster cardinalities.
-    All proof gates and pure helpers are the imported original functions.
+    P18 predates configurable protocol/roster constants. Specialize its three
+    entry functions' identity literals and explicit roster cardinalities, plus
+    one retained-stage comparison for the finite UID policy. The local phase
+    adapter retains every original shared gate and seals its accepted sidecars.
     Fail closed if an upstream function changes any of these adapter sites.
     Both runner sources remain bound by the whole selected source map.
     """
@@ -92,7 +196,22 @@ def protocol_core():
         tree = adapter.visit(ast.parse(inspect.getsource(getattr(core, name))))
         shared.require((adapter.identities, adapter.counts, adapter.summary) == counts,
                        'unchanged upstream adaptation sites: ' + name)
+        if name == 'verify':
+            old = "shared.inventory(confined(out, 'stage'), stage=True) == source"
+            matches = [node for node in ast.walk(tree)
+                       if isinstance(node, ast.Compare) and ast.unparse(node) == old]
+            shared.require(len(matches) == 1, 'unchanged retained-stage UID adapter site')
+
+            class Retained(ast.NodeTransformer):
+                def visit_Compare(self, node):
+                    if ast.unparse(node) == old:
+                        return ast.copy_location(ast.parse(
+                            "uid_retained(confined(out, 'stage'), source, out)", mode='eval').body, node)
+                    return self.generic_visit(node)
+
+            tree = Retained().visit(tree)
         exec(compile(ast.fix_missing_locations(tree), str(shared.HERE / 'run_p18_relationship_tests.py'), 'exec'), namespace)
+    install_uid_policy(namespace)
     return namespace
 
 
@@ -118,7 +237,7 @@ def main(argv=None):
     if args.output is None:
         base = os.environ.get('BUILD_OUTPUT_DIR')
         shared.require(bool(base) and Path(base).is_absolute(), 'absolute BUILD_OUTPUT_DIR required')
-        args.output = str(Path(base) / 'p13-reference-native')
+        args.output = str(Path(base) / CANONICAL_EVIDENCE)
     args.timeout = 1200  # unchanged guarded core timeout; not a product value
     core = protocol_core()
     root = shared.HERE.parent