Merge commit 'a7e3069b167511b3194242b3403fbcc35244a526' into HEAD

autoload/simulation_runtime.gd

@@ -20,6 +20,7 @@ const OPERATING_EXPENSE_ACCOUNT := &"OPERATING_EXPENSE"
 const AUTHORITY_FINE_ACCOUNT := &"AUTHORITY_FINE_EXPENSE"
 const WAGE_EXPENSE_ACCOUNT := &"WAGE_EXPENSE"
 const RUNTIME_SNAPSHOT_VERSION := 14
+const PeepingRooms := preload("res://scripts/special/peeping_room_service.gd")
 const AUTO_PURCHASE_RUNTIME_SNAPSHOT_VERSION := 14
 const EVENT_THREAT_RUNTIME_SNAPSHOT_VERSION := 13
 const MIMI_PURCHASE_RUNTIME_SNAPSHOT_VERSION := 15
@@ -97,6 +98,7 @@ const DEFENSE_COMMAND_TYPES: Array[StringName] = [
 var definition_registry := DefinitionRegistry.new()
 var command_queue := CommandQueue.new()
 var id_generator := StableIdGenerator.new()
+var peeping_room_service := PeepingRooms.new()
 var domain_events := DomainEventStream.new()
 var world_service := WorldService.new()
 var path_service := PathService.new()
@@ -233,6 +235,7 @@ func initialize(clock: SimulationKernel, state: Variant, seed := DEFAULT_WORLD_S
 	_hired_staff_candidate_ids.clear()
 	command_queue = CommandQueue.new()
 	id_generator = StableIdGenerator.new()
+	peeping_room_service = PeepingRooms.new()
 	domain_events = DomainEventStream.new()
 	world_service = WorldService.new()
 	path_service = PathService.new()
@@ -253,6 +256,7 @@ func initialize(clock: SimulationKernel, state: Variant, seed := DEFAULT_WORLD_S
 	_initialize_s5_defense()
 	construction_service = ConstructionService.new(world_service, construction_resource_state)
 	construction_service.current_minute = _clock.absolute_minute
+	peeping_room_service.configure(world_service, facility_service, construction_service, path_service, id_generator)
 	_connect_construction_signals()
 	var initial_infrastructure_ok := _initialize_infrastructure_services()
 	_next_command_id = 1
@@ -298,6 +302,8 @@ func initialize(clock: SimulationKernel, state: Variant, seed := DEFAULT_WORLD_S
 	register_command_executor(&"build_facility", Callable(self, "_execute_build_facility"))
 	register_command_executor(&"cancel_construction", Callable(self, "_execute_cancel_construction"))
 	register_command_executor(&"demolish_facility", Callable(self, "_execute_demolish_facility"))
+	for command_type: StringName in [&"peeping_room_validate", &"peeping_room_place", &"peeping_room_relink", &"peeping_room_record_evidence"]:
+		register_command_executor(command_type, Callable(self, "_execute_peeping_room"))
 	register_command_executor(&"set_elevator_operation", Callable(self, "_execute_set_elevator_operation"))
 	register_command_executor(&"start_excavation", Callable(self, "_execute_start_excavation"))
 	register_command_executor(&"pause_excavation", Callable(self, "_execute_pause_excavation"))
@@ -565,6 +571,8 @@ func create_snapshot() -> Dictionary:
 	}
 	if PurchaseQuoteV2.has_history(owned_systems[&"runtime"]):
 		owned_systems[&"runtime"][&"runtime_schema_version"] = MIMI_PURCHASE_RUNTIME_SNAPSHOT_VERSION
+	if not peeping_room_service.is_empty():
+		owned_systems[&"runtime"][&"special_peeping_rooms"] = peeping_room_service.create_snapshot()
 	# Every component above returns detached containers. Transfer this fresh
 	# tree into the envelope instead of deep-copying all component snapshots a
 	# second time. Keep generic capture's caller-owned-extra contract unchanged.
@@ -666,6 +674,8 @@ func _restore_snapshot_internal(snapshot: Dictionary, recovery_mode: bool) -> bo
 			runtime_current_keys.append(&"facility_cleanliness")
 		if runtime_schema_version >= AUTO_PURCHASE_RUNTIME_SNAPSHOT_VERSION:
 			runtime_current_keys.append(&"auto_purchase_plans")
+		if runtime_data.has(&"special_peeping_rooms"):
+			runtime_current_keys.append(&"special_peeping_rooms")
 		if not _dictionary_has_exact_canonical_keys(runtime_data, runtime_current_keys) \
 				or typeof(runtime_data.get(&"runtime_schema_version", null)) != TYPE_INT \
 				or typeof(runtime_data.get(&"world_seed", null)) != TYPE_INT \
@@ -1290,6 +1300,16 @@ func _restore_snapshot_internal(snapshot: Dictionary, recovery_mode: bool) -> bo
 	var restored_next_id := int(runtime_data.get(&"next_command_id", 0))
 	if restored_next_id <= 0:
 		return false
+	var restored_peeping_rooms := PeepingRooms.new()
+	restored_peeping_rooms.configure(restored_world, restored_facilities, restored_construction, restored_path, restored_id_generator)
+	if runtime_data.has(&"special_peeping_rooms"):
+		if not runtime_data[&"special_peeping_rooms"] is Dictionary:
+			return false
+		for raw_key: Variant in runtime_data:
+			if raw_key == &"special_peeping_rooms" and typeof(raw_key) != TYPE_STRING_NAME:
+				return false
+	if not restored_peeping_rooms.restore_section(runtime_data.get(&"special_peeping_rooms", null), restored_minute):
+		return false
 	if not SimulationSnapshot.restore(snapshot, _clock, _state):
 		return false
 	world_seed = restored_world_seed
@@ -1297,6 +1317,7 @@ func _restore_snapshot_internal(snapshot: Dictionary, recovery_mode: bool) -> bo
 	_staffing_projection_enabled = restored_staffing_projection
 	command_queue = restored_queue
 	id_generator = restored_id_generator
+	peeping_room_service = restored_peeping_rooms
 	domain_events = restored_domain_events
 	_rng_streams = restored_streams
 	world_service = restored_world
@@ -5546,6 +5567,7 @@ func _connect_facility_signals() -> void:
 		return
 	facility_service.instance_registered.connect(Callable(self, "_on_cleanliness_facility_registered"))
 	facility_service.instance_unregistered.connect(Callable(self, "_on_cleanliness_facility_unregistered"))
+	facility_service.instance_unregistered.connect(Callable(self, "_on_peeping_target_removed"))
 	facility_service.facility_state_changed.connect(Callable(self, "_on_facility_state_changed"))
 	facility_service.service_started.connect(Callable(self, "_on_service_started"))
 	facility_service.service_interrupted.connect(Callable(self, "_on_service_interrupted"))
@@ -10817,6 +10839,7 @@ func _on_excavation_completed(report: Dictionary) -> void:
 	))
 
 func _on_facility_demolished(instance_id: StringName, refund: Dictionary) -> void:
+	peeping_room_service.target_removed(instance_id)
 	if not mana_network_service.get_node(instance_id).is_empty():
 		mana_network_service.unregister_node(instance_id)
 	if facility_service.has_instance(instance_id):
@@ -10831,6 +10854,31 @@ func _on_facility_demolished(instance_id: StringName, refund: Dictionary) -> voi
 		0
 	))
 
+func _on_peeping_target_removed(instance_id: StringName) -> void:
+	peeping_room_service.target_removed(instance_id)
+
+func _execute_peeping_room(command: SimulationCommand) -> CommandResult:
+	var payload := command.payload
+	var report: Dictionary
+	if command.command_type == &"peeping_room_place":
+		report = PeepingRooms.result(&"BUILD_DEFINITION_PENDING_D04")
+	elif command.command_type == &"peeping_room_validate":
+		if typeof(payload.get(&"origin")) != TYPE_VECTOR3I or typeof(payload.get(&"footprint", Vector2i(2, 1))) != TYPE_VECTOR2I or typeof(payload.get(&"target_instance_id", &"")) not in [TYPE_STRING, TYPE_STRING_NAME]:
+			report = PeepingRooms.result(&"INVALID_PAYLOAD")
+		else:
+			report = peeping_room_service.validate_placement(payload[&"origin"], payload.get(&"footprint", Vector2i(2, 1)), StringName(payload.get(&"target_instance_id", &"")))
+	else:
+		var key: StringName = &"target_instance_id" if command.command_type == &"peeping_room_relink" else &"source_event_key"
+		if typeof(payload.get(&"room_id")) not in [TYPE_STRING, TYPE_STRING_NAME] or typeof(payload.get(key, &"")) not in [TYPE_STRING, TYPE_STRING_NAME]:
+			report = PeepingRooms.result(&"INVALID_PAYLOAD")
+		elif command.command_type == &"peeping_room_relink":
+			report = peeping_room_service.relink(StringName(payload[&"room_id"]), StringName(payload.get(key, &"")))
+		else:
+			report = peeping_room_service.record_evidence(StringName(payload[&"room_id"]), StringName(payload.get(key, &"")), _runtime_event_minute())
+	if report[&"ok"]:
+		return CommandResult.succeeded(command.command_id, report)
+	return CommandResult.failed(CommandResult.Code.VALIDATION_FAILED, command.command_id, String(report[&"reason"]), report)
+
 func _on_ledger_transaction_posted(transaction_id: StringName, _entries: Array) -> void:
 	if _operations_flushing_economy_notifications.has(transaction_id):
 		return

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

@@ -3734,6 +3734,61 @@
         "path": "docs/spec/acceptance/P23-03.json",
         "scope": "component and read-only observation only; native unexecuted; whole unit blocked"
       }
+    },
+    {
+      "test": "tests/special/test_peeping_room_geometry.gd",
+      "template": "Peeping room geometry: <checks> checks, <failures> failures",
+      "required": true,
+      "fixed_check_count_used": false,
+      "source_sha256_at_schema_audit": "103d85f83e1e97a8637892504264275eef23f172eeec56c94e5c022d81bf6945",
+      "authority": {
+        "path": "docs/plans/original-completion-2026-10-02/execution/p13-02/CODE1_TASK_SPEC.json",
+        "scope": "new backend suite marker; native execution pending"
+      }
+    },
+    {
+      "test": "tests/special/test_peeping_room_reference.gd",
+      "template": "Peeping room reference: <checks> checks, <failures> failures",
+      "required": true,
+      "fixed_check_count_used": false,
+      "source_sha256_at_schema_audit": "188b00a60f7aefda69dc4973f1de9c14447a0d4bcdd1257a2eabe7c4782e4caf",
+      "authority": {
+        "path": "docs/plans/original-completion-2026-10-02/execution/p13-02/CODE1_TASK_SPEC.json",
+        "scope": "new backend suite marker; native execution pending"
+      }
+    },
+    {
+      "test": "tests/special/test_peeping_room_evidence.gd",
+      "template": "Peeping room evidence: <checks> checks, <failures> failures",
+      "required": true,
+      "fixed_check_count_used": false,
+      "source_sha256_at_schema_audit": "390e0032d14de9a8875ced163a922341ce655744200d78b4ffa776d42c4f8c2a",
+      "authority": {
+        "path": "docs/plans/original-completion-2026-10-02/execution/p13-02/CODE1_TASK_SPEC.json",
+        "scope": "new backend suite marker; native execution pending"
+      }
+    },
+    {
+      "test": "tests/special/test_peeping_room_save.gd",
+      "template": "Peeping room save: <checks> checks, <failures> failures",
+      "required": true,
+      "fixed_check_count_used": false,
+      "source_sha256_at_schema_audit": "ba0e7644277477c1821135cae6d6fd533651799c6f9bd8c43ccf45e63b96c63c",
+      "authority": {
+        "path": "docs/plans/original-completion-2026-10-02/execution/p13-02/CODE1_TASK_SPEC.json",
+        "scope": "new backend suite marker; native execution pending"
+      }
+    },
+    {
+      "test": "tests/special/test_peeping_room_runtime.gd",
+      "template": "Peeping room runtime: <checks> checks, <failures> failures",
+      "required": true,
+      "fixed_check_count_used": false,
+      "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"
+      }
     }
   ],
   "unknown_roster_suite_policy": "refuse, explicitly review its source/schema before accepting",

docs/plans/original-completion-2026-10-02/execution/p13-02/CODE1_IMPLEMENTATION.md

@@ -0,0 +1,148 @@
+# P13-02-CODE1 implementation handoff
+
+This is non-graphic state/reference/occupancy/save machinery only. It does not
+complete P13-02, grant partial credit, or change the original 30/88 denominator
+or formal 14/88 count. Native execution and acceptance are pending.
+
+## Declared scope and verification
+
+Before editing, the worker declared the new service, bounded Runtime hooks,
+special tests, narrow runner, marker additions and this documentation. SaveService
+was conditional on an exact-key dependency; inspection found
+`SupplyAutoPurchaseQuotePlans.valid_runtime_format` has that dependency.
+Only allowed paths were edited. No asset, UI, content, model, policy, clock phase,
+or P04 change is involved.
+
+The full ignored `build/owner-handoff/ACTUAL_OPUS_RUN_DECISION.json` was read and
+left untouched. Base HEAD is `1e8d4ea88876865a72547ff0edf43ed5b0e74c7e`.
+The immutable spec was not edited. Its SHA-256 without the terminal newline is
+the registered `8ea65f2eefedc772c08170186f5cd03ac017432f4b069ade4f9210d4215c2d43`;
+the on-disk file including that newline is
+`988d6e18a49372549863cd433432c2852c962fa64b138e490361c023e80249dc`.
+
+Worker verification: Python AST parsing/compilation only, without executing the
+runner or importing its dependencies; source/JSON/hash review. GDScript parsing,
+all tests, `--prepare-only`, native collection and regression were **not run**.
+No check totals or native success receipts are claimed. No git writes, commits,
+branches, other agents, integration or Runner operations were performed.
+
+## Service and commands
+
+`PeepingRoomService` is a separate RefCounted service with no content or UI.
+All command executors use the existing Runtime queue; queue/history receipts
+remain normal Runtime behavior even for rejected/read-only domain commands.
+
+| Command | Preconditions and result |
+| --- | --- |
+| `peeping_room_validate` | `origin: Vector3i`, optional `footprint: Vector2i` (2x1), optional `target_instance_id: String/StringName`. Pure domain validation. |
+| `peeping_room_place` | Always `BUILD_DEFINITION_PENDING_D04`; no occupancy, resource, room or ID mutation. |
+| `peeping_room_relink` | Existing `room_id`, explicit nonempty `target_instance_id`; full geometry and ownership validation excluding only the room's exact own occupancy. Leaves `LINKED`, stopped, `RELINKED_PENDING_RESUME`. |
+| `peeping_room_record_evidence` | Existing `room_id`, nonblank named `source_event_key`. Uses Runtime minute, never caller-supplied time. Checks stopped, target power, target closure/opening hours, completed construction authority, and lost target in that order. Otherwise rejects with `RECORDER_ELIGIBILITY_PENDING_D04_D08_P16`. Never writes the ledger. |
+
+Malformed command field types return `INVALID_PAYLOAD`. Missing/blank source keys
+return `SOURCE_EVENT_KEY_REQUIRED` before operational checks. Loss latches both
+`TARGET_LOST` and stopped, so collection after normal demolition returns `STOPPED`
+while the room's link/stop reason identifies `TARGET_LOST`. No resume command or
+automatic resume is defined. Power/closure guards read actual target conditions;
+they do not define special-room power demand, staffing or normal operation.
+
+Placement is exactly 2x1 BACK, fully behind a single completed FRONT
+`public_bath` or `adult_private_room`. It checks bounds/excavation, both real
+FacilityService and ConstructionService ownership, FRONT cells, World occupancy,
+and both World walkability and PathService membership. Exact rejection families:
+`WRONG_LAYER`, `WRONG_FOOTPRINT`, `OUT_OF_BOUNDS`, `NOT_EXCAVATED`,
+`TARGET_INCOMPLETE`, `TARGET_MISSING`, `PARTIAL_TARGET`, `MULTIPLE_TARGETS`,
+`TARGET_MISMATCH`, `TARGET_UNKNOWN`, `TARGET_NOT_ALLOWED`, `TARGET_WRONG_LAYER`,
+`CORRIDOR_CONFLICT`, `BACK_OCCUPIED`. The named `vip_suite` request returns
+`PENDING_P10_01`; `vip_private_salon` is not an allowed substitute.
+
+Any existing corridor cell conflicts. A proposed cut through a connected corridor
+is rejected before mutation, preserving the route; no reroute or bypass is
+invented. Accepted fixture cells were already nonwalkable and become occupied
+under `special_peeping_room`. No special-room construction weight/cost is assumed.
+
+Runtime demolition and FacilityService unregistration latch loss. Collection also
+detects missing FRONT occupancy/identity. Rebuilding the same cells produces a
+different construction ID and never silently reattaches rooms. Relink is explicit.
+
+## CONTROLLED_FIXTURE boundary
+
+Only service methods named `controlled_fixture_place`,
+`controlled_fixture_set_stopped`, and `controlled_fixture_record_evidence` create
+test rooms, enable fixture collection, or write evidence. They are labeled
+CONTROLLED_FIXTURE in code and tests. They have no command/palette registration,
+and no Runtime executor calls them. Test helpers disclose direct underground
+geometry setup and MIMI registration; those do not assert natural unlock/join or
+production eligibility. Target construction uses existing definitions/costs/times.
+
+Room IDs come from the existing StableIdGenerator's `peeping_room` counter.
+Evidence IDs encode the UTF-8 lengths and hex bytes of the full
+`(room_id, target_instance_id, source_event_key)` tuple. This is injective rather
+than delimiter concatenation or a hash with possible collisions. Replays return
+the existing ID with `duplicated=true` and retain the first minute. Evidence is
+only IDs, source key and minute. Room `target_history` retains prior bindings so
+relink does not orphan existing opaque evidence.
+
+## Save invariants
+
+Runtime versions remain 14 and 15, with version 15 only for existing v2 purchase
+history. Optional canonical StringName key `special_peeping_rooms` has section
+version 1 and exact keys `version`, `rooms`, `evidence`. Empty service state emits
+no key, preserving original snapshots. Absent sections restore empty only if no
+world occupancy uses the special kind or `peeping_room.` prefix. Thus removing
+the section from a current room-bearing save is rejected without a version bump.
+
+Restore stages the entire section against staged world, path, facilities,
+construction and ID generator before committing any Runtime state. It checks
+canonical keys/types, version, duplicate room/evidence identities, deterministic
+evidence tuple encoding, bounds/geometry, room/occupancy correspondence in both
+directions, generator high-water mark, known historical facility IDs, present
+target definition/geometry, loss/stopped coherence, and evidence room/history/time
+references. Null/present-empty sections are invalid; absence is the only empty
+encoding. Existing pending command/reference and schema migration checks remain.
+
+SaveService validates a present section through the same staged service validator,
+then supplies a shallow copy without this one key to the original purchase-format
+allowlist. No other keys are removed. The original format 1 (Runtime 1–14),
+format 2 (Runtime 15), manifest duplicate 15 and MIMI checks remain unchanged.
+
+## Verification handoff
+
+Five strict suites cover geometry/corridor, removal/relink, evidence/guards,
+typed atomic save corruption and physical format 1, and Runtime commands plus
+physical MIMI format 2. Tests use `var_to_bytes` for whole Runtime atomicity and
+roundtrip checks. New marker rows carry final test source hashes; prior rows and
+policy are unchanged. A helper under `tests/special/` is not a discovered suite.
+
+`tools/run_p13_reference_tests.py` defaults to read-only verification of
+`$BUILD_OUTPUT_DIR/p13-reference-native`, failing closed without an absolute
+environment path. Explicit `--output --collect` is for Root only. The strict flag
+parser rejects abbreviations, repeated flags, positional inputs and `=` aliases.
+`--prepare-only` is non-native but was not executed by this worker.
+
+The runner imports the existing P18 core by path. Since its three entry functions
+hardcode their label and seven/nine counts, a bounded AST adapter changes only
+protocol/runner identities, those explicit roster cardinalities, and diagnostic
+labels. It checks the exact adaptation-site counts and fails closed on drift.
+All confinement, inventory, JSON, sealing, source/engine/marker/policy binding,
+process, diagnostic, profile and cleanup gates are reused. It neither copies the
+whole runner nor modifies P18/helpers. The whole selected source map binds both
+runner sources. Phase count derives from `len(ROSTER) + 2`.
+
+Roster includes all five new suites and existing construction/world/path/runtime
+construction, supply migration, transport save, Runtime weather, snapshot
+serialization, save lifecycle and MIMI purchase suites. Receipts retain
+`formal_acceptance=false` and `original_unit_complete=false`.
+
+Root must perform serialized native collection with the allocation/private
+profile, then registered `p13-reference` validation, a separate actual Opus
+review, and the final exact-version required regression. Only those steps can
+establish execution results.
+
+## Still pending, without invented values
+
+D04 build/upkeep/build minutes and other product numbers; recorder staffing and
+natural triggers/rates D04/D08/P16; VIP suite P10-01; D05 transaction decisions
+(no tick-phase changes here); P19 evidence use and P24 camera UI; normal palette
+registration/construction/operation and natural production evidence; whole P13
+acceptance, final review/integration, and applicable native Windows conditions.

scripts/core/save_service.gd

@@ -1048,6 +1048,13 @@ static func _snapshot_package_format(snapshot: Dictionary) -> int:
 	if not runtime_data is Dictionary: return FORMAT_VERSION # no owned Runtime authority
 	var version: Variant = runtime_data.get(&"runtime_schema_version", 1)
 	if typeof(version) != TYPE_INT: return -1
+	if runtime_data.has(&"special_peeping_rooms"):
+		var section_clock: Variant = snapshot.get(&"clock")
+		if not section_clock is Dictionary or typeof(section_clock.get(&"absolute_minute")) != TYPE_INT: return -1
+		if not preload("res://scripts/special/peeping_room_service.gd").valid_runtime_section(runtime_data, section_clock[&"absolute_minute"]): return -1
+		# Keep PurchaseQuotePlansV2's exact original allowlist and all format rules.
+		runtime_data = runtime_data.duplicate()
+		runtime_data.erase(&"special_peeping_rooms")
 	if version >= 1 and version <= 14:
 		return -1 if PurchaseQuoteV2.has_history(runtime_data) else FORMAT_VERSION
 	if version != 15: return -1

scripts/special/peeping_room_service.gd

@@ -0,0 +1,271 @@
+class_name PeepingRoomService
+extends RefCounted
+
+## Non-graphic state/reference machinery; production definitions remain pending.
+const SECTION_VERSION := 1
+const ALLOWED_TARGETS := [&"public_bath", &"adult_private_room"]
+const OCCUPANCY_KIND := &"special_peeping_room"
+const ID_PREFIX := &"peeping_room"
+const ROOM_KEYS := [&"room_id", &"origin", &"footprint", &"cells", &"target_instance_id", &"target_definition_id", &"target_history", &"link_state", &"stopped", &"stop_reason"]
+const EVIDENCE_KEYS := [&"evidence_id", &"room_id", &"target_instance_id", &"source_event_key", &"minute"]
+
+var _rooms: Dictionary = {}
+var _evidence: Dictionary = {}
+var _world: WorldService
+var _facilities: FacilityService
+var _construction: ConstructionService
+var _path: PathService
+var _ids: StableIdGenerator
+
+func configure(world: WorldService, facilities: FacilityService, construction: ConstructionService, path: PathService, ids: StableIdGenerator) -> void:
+	_world = world
+	_facilities = facilities
+	_construction = construction
+	_path = path
+	_ids = ids
+
+static func result(reason: StringName = &"OK") -> Dictionary:
+	return {&"ok": reason == &"OK", &"reason": reason}
+
+func is_empty() -> bool:
+	return _rooms.is_empty() and _evidence.is_empty()
+
+func get_room(room_id: StringName) -> Dictionary:
+	return (_rooms.get(room_id, {}) as Dictionary).duplicate(true)
+
+## Pure validation. own_room is supplied internally only for explicit relink/restore.
+func validate_placement(origin: Vector3i, footprint: Vector2i = Vector2i(2, 1), target_id: StringName = &"", own_room: StringName = &"") -> Dictionary:
+	if origin.z != WorldService.LAYER_BACK: return result(&"WRONG_LAYER")
+	if footprint != Vector2i(2, 1): return result(&"WRONG_FOOTPRINT")
+	if target_id == &"vip_suite": return result(&"PENDING_P10_01")
+	var cells := _world.cells_for_footprint(origin, footprint)
+	var targets: Array[StringName] = []
+	var missing := false
+	for cell: Vector3i in cells:
+		if not _world.is_in_bounds(cell): return result(&"OUT_OF_BOUNDS")
+		if cell.y < 0 and not _world.is_excavated(cell): return result(&"NOT_EXCAVATED")
+		var front := Vector3i(cell.x, cell.y, WorldService.LAYER_FRONT)
+		if _world.get_occupant_kind(front) == &"construction": return result(&"TARGET_INCOMPLETE")
+		var occupant := _world.get_occupant(front)
+		if occupant.is_empty(): missing = true
+		elif occupant not in targets: targets.append(occupant)
+	if targets.size() > 1: return result(&"MULTIPLE_TARGETS")
+	if missing: return result(&"TARGET_MISSING" if targets.is_empty() else &"PARTIAL_TARGET")
+	var resolved_id := targets[0]
+	if not target_id.is_empty() and target_id != resolved_id: return result(&"TARGET_MISMATCH")
+	var target := _facilities.get_instance(resolved_id)
+	if target == null: return result(&"TARGET_UNKNOWN")
+	if target.definition_id not in ALLOWED_TARGETS: return result(&"TARGET_NOT_ALLOWED")
+	if target.origin.z != WorldService.LAYER_FRONT: return result(&"TARGET_WRONG_LAYER")
+	var built := _construction.get_facility(resolved_id)
+	if built.is_empty(): return result(&"TARGET_INCOMPLETE")
+	if built.get(&"definition_id") != target.definition_id: return result(&"TARGET_MISMATCH")
+	for cell: Vector3i in cells:
+		var front := Vector3i(cell.x, cell.y, WorldService.LAYER_FRONT)
+		if front not in target.occupied_cells or front not in built[&"occupied_cells"]: return result(&"PARTIAL_TARGET")
+		# Reject cutting an existing corridor; never construct an implicit bypass.
+		if _world.is_walkable(cell) or _path.has_cell(cell): return result(&"CORRIDOR_CONFLICT")
+	var occupancy := _world.validate_occupancy(cells)
+	if not occupancy[&"ok"]:
+		if own_room.is_empty() or _world.get_entity_cells(own_room) != cells or _world.get_entity_kind(own_room) != OCCUPANCY_KIND:
+			return result(&"BACK_OCCUPIED")
+	var report := result()
+	report.merge({&"cells": cells, &"target_instance_id": resolved_id, &"target_definition_id": target.definition_id})
+	return report
+
+## CONTROLLED_FIXTURE: tests only. No Runtime executor or palette calls this.
+func controlled_fixture_place(origin: Vector3i, target_id: StringName = &"") -> Dictionary:
+	var report := validate_placement(origin, Vector2i(2, 1), target_id)
+	if not report[&"ok"]: return report
+	var room_id := _ids.next_id(ID_PREFIX)
+	if room_id.is_empty(): return result(&"ID_EXHAUSTED")
+	if not _world.occupy(room_id, report[&"cells"], OCCUPANCY_KIND): return result(&"BACK_OCCUPIED")
+	_rooms[room_id] = {&"room_id": room_id, &"origin": origin, &"footprint": Vector2i(2, 1),
+		&"cells": report[&"cells"].duplicate(), &"target_instance_id": report[&"target_instance_id"],
+		&"target_definition_id": report[&"target_definition_id"],
+		&"target_history": [report[&"target_instance_id"]], &"link_state": &"LINKED",
+		&"stopped": true, &"stop_reason": &"CONTROLLED_FIXTURE"}
+	report[&"room_id"] = room_id
+	return report
+
+func target_removed(target_id: StringName) -> void:
+	for room: Dictionary in _rooms.values():
+		if room[&"target_instance_id"] == target_id:
+			room[&"link_state"] = &"TARGET_LOST"
+			room[&"stopped"] = true
+			room[&"stop_reason"] = &"TARGET_LOST"
+
+func detect_target_loss() -> void:
+	for room: Dictionary in _rooms.values():
+		if room[&"link_state"] == &"TARGET_LOST": continue
+		if not _target_still_present(room): target_removed(room[&"target_instance_id"])
+
+func _target_still_present(room: Dictionary) -> bool:
+	var target := _facilities.get_instance(room[&"target_instance_id"])
+	if target == null or target.definition_id != room[&"target_definition_id"]: return false
+	for cell: Vector3i in room[&"cells"]:
+		if _world.get_occupant(Vector3i(cell.x, cell.y, WorldService.LAYER_FRONT)) != target.instance_id: return false
+	return true
+
+func relink(room_id: StringName, target_id: StringName) -> Dictionary:
+	if not _rooms.has(room_id): return result(&"ROOM_MISSING")
+	if target_id.is_empty(): return result(&"TARGET_MISSING")
+	var room: Dictionary = _rooms[room_id]
+	var report := validate_placement(room[&"origin"], room[&"footprint"], target_id, room_id)
+	if not report[&"ok"]: return report
+	room[&"target_instance_id"] = target_id
+	room[&"target_definition_id"] = report[&"target_definition_id"]
+	if target_id not in room[&"target_history"]: room[&"target_history"].append(target_id)
+	room[&"link_state"] = &"LINKED"
+	room[&"stopped"] = true
+	room[&"stop_reason"] = &"RELINKED_PENDING_RESUME"
+	return result()
+
+func _collection_precondition(room_id: StringName, source_event_key: StringName, minute: int) -> Dictionary:
+	if String(source_event_key).strip_edges().is_empty(): return result(&"SOURCE_EVENT_KEY_REQUIRED")
+	if minute < 0: return result(&"INVALID_MINUTE")
+	if not _rooms.has(room_id): return result(&"ROOM_MISSING")
+	detect_target_loss()
+	var room: Dictionary = _rooms[room_id]
+	# Required precedence: stopped, power, closed, construction, lost target.
+	if room[&"stopped"]: return result(&"STOPPED")
+	var target := _facilities.get_instance(room[&"target_instance_id"])
+	if target != null:
+		if not target.power_available: return result(&"NO_POWER")
+		if not target.operator_open or not target.is_within_opening_hours(minute): return result(&"CLOSED")
+		if _construction.get_facility(target.instance_id).is_empty(): return result(&"UNDER_CONSTRUCTION")
+	if room[&"link_state"] == &"TARGET_LOST" or target == null: return result(&"TARGET_LOST")
+	return result()
+
+func record_evidence(room_id: StringName, source_event_key: StringName, minute: int) -> Dictionary:
+	var report := _collection_precondition(room_id, source_event_key, minute)
+	return report if not report[&"ok"] else result(&"RECORDER_ELIGIBILITY_PENDING_D04_D08_P16")
+
+## CONTROLLED_FIXTURE: explicit test state, never a production resume command.
+func controlled_fixture_set_stopped(room_id: StringName, stopped: bool) -> bool:
+	if not _rooms.has(room_id): return false
+	detect_target_loss()
+	var room: Dictionary = _rooms[room_id]
+	if not stopped and room[&"link_state"] == &"TARGET_LOST": return false
+	room[&"stopped"] = stopped
+	room[&"stop_reason"] = &"CONTROLLED_FIXTURE" if stopped else &""
+	return true
+
+## CONTROLLED_FIXTURE: sole ledger write entry point, opaque event keys only.
+func controlled_fixture_record_evidence(room_id: StringName, source_event_key: StringName, minute: int) -> Dictionary:
+	var report := _collection_precondition(room_id, source_event_key, minute)
+	if not report[&"ok"]: return report
+	var target_id: StringName = _rooms[room_id][&"target_instance_id"]
+	var evidence_id := _evidence_id(room_id, target_id, source_event_key)
+	var duplicated := _evidence.has(evidence_id)
+	if not duplicated:
+		_evidence[evidence_id] = {&"evidence_id": evidence_id, &"room_id": room_id,
+			&"target_instance_id": target_id, &"source_event_key": source_event_key, &"minute": minute}
+	return {&"ok": true, &"reason": &"OK", &"evidence_id": evidence_id, &"duplicated": duplicated}
+
+static func _evidence_id(room_id: StringName, target_id: StringName, event_key: StringName) -> StringName:
+	# Length-prefixed UTF-8 hex is injective; delimiters in input cannot alias.
+	var pieces: Array[String] = []
+	for value: StringName in [room_id, target_id, event_key]:
+		var bytes := String(value).to_utf8_buffer()
+		pieces.append("%d:%s" % [bytes.size(), bytes.hex_encode()])
+	return StringName("peeping_evidence:" + ":".join(pieces))
+
+func create_snapshot() -> Dictionary:
+	var rooms: Array[Dictionary] = []
+	var evidence: Array[Dictionary] = []
+	var keys := _rooms.keys()
+	keys.sort()
+	for key: StringName in keys: rooms.append(_rooms[key].duplicate(true))
+	keys = _evidence.keys()
+	keys.sort()
+	for key: StringName in keys: evidence.append(_evidence[key].duplicate(true))
+	return {&"version": SECTION_VERSION, &"rooms": rooms, &"evidence": evidence}
+
+static func _exact(value: Variant, keys: Array) -> bool:
+	if not value is Dictionary or value.size() != keys.size(): return false
+	for key: Variant in value:
+		if typeof(key) != TYPE_STRING_NAME or key not in keys: return false
+	return true
+
+## Save format 15 has a fixed purchase-key allowlist outside this task's scope.
+## Validate this independent section before passing only the original keys to it.
+static func valid_runtime_section(data: Dictionary, minute: int) -> bool:
+	if not data.get(&"special_peeping_rooms") is Dictionary: return false
+	for key: Variant in data:
+		if key == &"special_peeping_rooms" and typeof(key) != TYPE_STRING_NAME: return false
+	for key: StringName in [&"world", &"path", &"facilities", &"construction", &"id_generator"]:
+		if not data.get(key) is Dictionary: return false
+	var world := WorldService.new()
+	var path := PathService.new()
+	var facilities := FacilityService.new()
+	var ids := StableIdGenerator.new()
+	if not world.restore_snapshot(data[&"world"]) or not path.restore_snapshot(data[&"path"]) or not facilities.restore_snapshot(data[&"facilities"]) or not ids.restore_snapshot(data[&"id_generator"]): return false
+	var construction := ConstructionService.new(world)
+	if not construction.restore_snapshot(data[&"construction"]): return false
+	var service := PeepingRoomService.new()
+	service.configure(world, facilities, construction, path, ids)
+	return service.restore_section(data[&"special_peeping_rooms"], minute)
+
+## Staged restore validates both directions of world ownership before committing.
+func restore_section(section: Variant, minute: int) -> bool:
+	var rooms: Dictionary = {}
+	var evidence: Dictionary = {}
+	if section != null:
+		if not _exact(section, [&"version", &"rooms", &"evidence"]): return false
+		if typeof(section[&"version"]) != TYPE_INT or section[&"version"] != SECTION_VERSION: return false
+		if not section[&"rooms"] is Array or not section[&"evidence"] is Array: return false
+		if section[&"rooms"].is_empty(): return false # absence represents empty state
+		var known: Array = _facilities.create_snapshot()[&"known_instance_ids"]
+		for raw: Variant in section[&"rooms"]:
+			if not _exact(raw, ROOM_KEYS): return false
+			for key: StringName in [&"room_id", &"target_instance_id", &"target_definition_id", &"link_state", &"stop_reason"]:
+				if typeof(raw[key]) != TYPE_STRING_NAME: return false
+			var room_id: StringName = raw[&"room_id"]
+			var suffix := String(room_id).trim_prefix("peeping_room.")
+			if suffix.length() != 12 or not suffix.is_valid_int() or int(suffix) <= 0 or String(room_id) != "peeping_room.%012d" % int(suffix): return false
+			if int(suffix) > _ids.get_counter(ID_PREFIX) or rooms.has(room_id): return false
+			if typeof(raw[&"origin"]) != TYPE_VECTOR3I or typeof(raw[&"footprint"]) != TYPE_VECTOR2I: return false
+			if raw[&"origin"].z != WorldService.LAYER_BACK or raw[&"footprint"] != Vector2i(2, 1): return false
+			var cells := _world.cells_for_footprint(raw[&"origin"], raw[&"footprint"])
+			if not raw[&"cells"] is Array or raw[&"cells"] != cells: return false
+			for cell: Variant in raw[&"cells"]:
+				if typeof(cell) != TYPE_VECTOR3I: return false
+			if _world.get_entity_cells(room_id) != cells or _world.get_entity_kind(room_id) != OCCUPANCY_KIND: return false
+			for cell: Vector3i in cells:
+				if _world.is_walkable(cell) or _path.has_cell(cell): return false
+			if raw[&"target_definition_id"] not in ALLOWED_TARGETS: return false
+			if typeof(raw[&"stopped"]) != TYPE_BOOL or not raw[&"target_history"] is Array: return false
+			var history := {}
+			for target: Variant in raw[&"target_history"]:
+				if typeof(target) != TYPE_STRING_NAME or String(target) not in known or history.has(target): return false
+				history[target] = true
+			if not history.has(raw[&"target_instance_id"]): return false
+			var current_target := _facilities.get_instance(raw[&"target_instance_id"])
+			if current_target != null and current_target.definition_id != raw[&"target_definition_id"]: return false
+			if raw[&"link_state"] == &"TARGET_LOST":
+				if not raw[&"stopped"] or raw[&"stop_reason"] != &"TARGET_LOST" or _target_still_present(raw): return false
+			elif raw[&"link_state"] == &"LINKED":
+				if not validate_placement(raw[&"origin"], raw[&"footprint"], raw[&"target_instance_id"], room_id)[&"ok"]: return false
+				if _facilities.get_instance(raw[&"target_instance_id"]).definition_id != raw[&"target_definition_id"]: return false
+				if raw[&"stopped"]:
+					if raw[&"stop_reason"] not in [&"CONTROLLED_FIXTURE", &"RELINKED_PENDING_RESUME"]: return false
+				elif raw[&"stop_reason"] != &"": return false
+			else: return false
+			rooms[room_id] = raw.duplicate(true)
+		for raw: Variant in section[&"evidence"]:
+			if not _exact(raw, EVIDENCE_KEYS): return false
+			for key: StringName in [&"evidence_id", &"room_id", &"target_instance_id", &"source_event_key"]:
+				if typeof(raw[key]) != TYPE_STRING_NAME or raw[key].is_empty(): return false
+			if String(raw[&"source_event_key"]).strip_edges().is_empty(): return false
+			if typeof(raw[&"minute"]) != TYPE_INT or raw[&"minute"] < 0 or raw[&"minute"] > minute: return false
+			if not rooms.has(raw[&"room_id"]) or raw[&"target_instance_id"] not in rooms[raw[&"room_id"]][&"target_history"]: return false
+			var expected := _evidence_id(raw[&"room_id"], raw[&"target_instance_id"], raw[&"source_event_key"])
+			if expected != raw[&"evidence_id"] or evidence.has(expected): return false
+			evidence[expected] = raw.duplicate(true)
+	for occupied: Dictionary in _world.create_snapshot()[&"occupancies"]:
+		if occupied[&"kind"] == OCCUPANCY_KIND or String(occupied[&"entity_id"]).begins_with("peeping_room."):
+			if not rooms.has(occupied[&"entity_id"]): return false
+	_rooms = rooms
+	_evidence = evidence
+	return true

tests/special/peeping_room_fixture.gd

@@ -0,0 +1,59 @@
+extends SceneTree
+
+## CONTROLLED_FIXTURE. No production placement, recorder or eligibility claim.
+const Rooms := preload("res://scripts/special/peeping_room_service.gd")
+var checks := 0
+var failures := 0
+var fixtures: Array[Dictionary] = []
+
+func _initialize() -> void:
+	call_deferred("_run")
+
+func expect(ok: bool, label: String) -> void:
+	checks += 1
+	if not ok:
+		failures += 1
+		push_error(label)
+
+func fixture(build_bath := true) -> Dictionary:
+	var clock: Variant = load("res://autoload/game_clock.gd").new()
+	var state: Variant = load("res://autoload/game_state.gd").new()
+	var runtime: Variant = load("res://autoload/simulation_runtime.gd").new()
+	clock.reset(1, 6, 0)
+	state.reset()
+	expect(runtime.initialize(clock, state, 1302)[&"valid"], "CONTROLLED_FIXTURE real Runtime initializes")
+	clock.set_paused(true)
+	runtime.configure_save_policy({&"enabled": false, &"daily_checkpoint_enabled": false})
+	var f := {&"clock": clock, &"state": state, &"runtime": runtime, &"target": &"", &"origin": Vector3i(-6, 1, 1)}
+	fixtures.append(f)
+	if build_bath:
+		var built: CommandResult = runtime.submit_command(&"build_facility", {&"definition_id": &"public_bath", &"origin": Vector3i(-6, 0, 0)})
+		expect(built.is_success(), "real paid public bath construction")
+		clock.advance_minutes(runtime.definition_registry.get_facility(&"public_bath").build_minutes)
+		for target: FacilityInstance in runtime.facility_service.get_instances():
+			if target.definition_id == &"public_bath": f[&"target"] = target.instance_id
+		expect(not f[&"target"].is_empty(), "completed FRONT target has real stable identity")
+	return f
+
+func place(f: Dictionary, origin: Variant = null) -> StringName:
+	var report: Dictionary = f[&"runtime"].peeping_room_service.controlled_fixture_place(f[&"origin"] if origin == null else origin)
+	expect(report[&"ok"], "CONTROLLED_FIXTURE only room placement: " + String(report[&"reason"]))
+	return report.get(&"room_id", &"")
+
+func enable_fixture(f: Dictionary, room: StringName) -> void:
+	var runtime: Variant = f[&"runtime"]
+	expect(runtime.peeping_room_service.controlled_fixture_set_stopped(room, false), "CONTROLLED_FIXTURE explicitly enables ledger exercise")
+	expect(runtime.facility_service.set_operational_status(f[&"target"], {&"power_available": true, &"operator_open": true}, f[&"clock"].absolute_minute).is_success(), "CONTROLLED_FIXTURE target conditions")
+
+func rejects_atomically(runtime: Variant, snapshot: Dictionary) -> bool:
+	var before := var_to_bytes(runtime.create_snapshot())
+	return not runtime.restore_snapshot(snapshot) and before == var_to_bytes(runtime.create_snapshot())
+
+func finish(label: String) -> void:
+	for f: Dictionary in fixtures:
+		f[&"runtime"].free()
+		f[&"clock"].free()
+		f[&"state"].free()
+	fixtures.clear()
+	print("%s: %d checks, %d failures" % [label, checks, failures])
+	quit(1 if failures else 0)

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_evidence.gd

@@ -0,0 +1,40 @@
+extends "res://tests/special/peeping_room_fixture.gd"
+
+func _run() -> void:
+	var f := fixture()
+	var runtime: Variant = f[&"runtime"]
+	var room := place(f)
+	var other := place(f, Vector3i(-4, 1, 1))
+	var service: Variant = runtime.peeping_room_service
+	var minute: int = f[&"clock"].absolute_minute
+	expect(service.record_evidence(room, &"", minute)[&"reason"] == &"SOURCE_EVENT_KEY_REQUIRED", "named source key required first")
+	expect(service.controlled_fixture_record_evidence(room, &"fixture.event", minute)[&"reason"] == &"STOPPED", "initial stopped state rejects even fixture ledger")
+	enable_fixture(f, room)
+	enable_fixture(f, other)
+	var target: FacilityInstance = runtime.facility_service.get_instance(f[&"target"])
+	target.power_available = false
+	target.operator_open = false
+	expect(service.record_evidence(room, &"fixture.event", minute)[&"reason"] == &"NO_POWER", "power rejection precedes closed")
+	target.power_available = true
+	expect(service.record_evidence(room, &"fixture.event", minute)[&"reason"] == &"CLOSED", "closed target rejects")
+	target.operator_open = true
+	# CONTROLLED_FIXTURE temporarily omit construction authority, leaving live target.
+	var empty_construction := ConstructionService.new(runtime.world_service)
+	service.configure(runtime.world_service, runtime.facility_service, empty_construction, runtime.path_service, runtime.id_generator)
+	expect(service.record_evidence(room, &"fixture.event", minute)[&"reason"] == &"UNDER_CONSTRUCTION", "uncompleted construction authority rejects collection")
+	service.configure(runtime.world_service, runtime.facility_service, runtime.construction_service, runtime.path_service, runtime.id_generator)
+	var before := var_to_bytes(service.create_snapshot())
+	expect(service.record_evidence(room, &"fixture.event", minute)[&"reason"] == &"RECORDER_ELIGIBILITY_PENDING_D04_D08_P16", "production collection never invents eligibility")
+	expect(before == var_to_bytes(service.create_snapshot()), "production collection leaves ledger unchanged")
+	var first: Dictionary = service.controlled_fixture_record_evidence(room, &"fixture.event", minute)
+	var replay: Dictionary = service.controlled_fixture_record_evidence(room, &"fixture.event", minute + 1)
+	expect(first[&"ok"] and not first[&"duplicated"] and replay[&"duplicated"] and replay[&"evidence_id"] == first[&"evidence_id"], "tuple replay returns exactly one stable ID")
+	var second: Dictionary = service.controlled_fixture_record_evidence(other, &"fixture.event", minute)
+	var third: Dictionary = service.controlled_fixture_record_evidence(room, &"fixture:event:日本語", minute)
+	expect(second[&"evidence_id"] != first[&"evidence_id"] and third[&"evidence_id"] != first[&"evidence_id"], "different room/key distinct, delimiter and UTF-8 input supported")
+	expect(service.create_snapshot()[&"evidence"].size() == 3, "replay adds no ledger record")
+	var snapshot: Dictionary = runtime.create_snapshot()
+	expect(runtime.restore_snapshot(snapshot), "active fixture evidence restores")
+	var restored: Dictionary = runtime.peeping_room_service.controlled_fixture_record_evidence(room, &"fixture.event", minute)
+	expect(restored[&"duplicated"] and restored[&"evidence_id"] == first[&"evidence_id"], "evidence stable across restore")
+	finish("Peeping room evidence")

tests/special/test_peeping_room_geometry.gd

@@ -0,0 +1,62 @@
+extends "res://tests/special/peeping_room_fixture.gd"
+
+func _run() -> void:
+	var f := fixture()
+	var runtime: Variant = f[&"runtime"]
+	var service: Variant = runtime.peeping_room_service
+	var origin: Vector3i = f[&"origin"]
+	var before := var_to_bytes(runtime.create_snapshot())
+	expect(service.validate_placement(origin)[&"ok"], "public bath entirely behind one completed FRONT target")
+	expect(service.validate_placement(Vector3i(-6, 1, 0))[&"reason"] == &"WRONG_LAYER", "FRONT room rejected")
+	expect(service.validate_placement(origin, Vector2i(1, 2))[&"reason"] == &"WRONG_FOOTPRINT", "rotated footprint rejected")
+	expect(service.validate_placement(origin, Vector2i(2, 1), &"vip_suite")[&"reason"] == &"PENDING_P10_01", "VIP unavailable, never substituted")
+	expect(service.validate_placement(Vector3i(-1, 1, 1))[&"reason"] == &"PARTIAL_TARGET", "partial target rejected")
+	expect(service.validate_placement(Vector3i(8, 1, 1))[&"reason"] == &"TARGET_MISSING", "missing target rejected")
+	expect(service.validate_placement(origin, Vector2i(2, 1), &"missing")[&"reason"] == &"TARGET_MISMATCH", "named reference must match FRONT occupant")
+	expect(before == var_to_bytes(runtime.create_snapshot()), "geometry validation is pure across whole Runtime")
+	# Complete real second bath; the 2x1 footprint at the boundary spans targets.
+	expect(runtime.submit_command(&"build_facility", {&"definition_id": &"public_bath", &"origin": Vector3i(0, 0, 0)}).is_success(), "second paid target starts")
+	expect(service.validate_placement(Vector3i(0, 1, 1))[&"reason"] == &"TARGET_INCOMPLETE", "construction reservation is not a completed target")
+	f[&"clock"].advance_minutes(runtime.definition_registry.get_facility(&"public_bath").build_minutes)
+	expect(service.validate_placement(Vector3i(-1, 1, 1))[&"reason"] == &"MULTIPLE_TARGETS", "multiple FRONT target footprints rejected")
+	# A real path joining both sides through the proposed footprint must survive.
+	for x in range(-7, -3): runtime.world_service.set_walkable(Vector3i(x, 1, 1), true)
+	expect(runtime.rebuild_path_graph_from_world(), "CONTROLLED_FIXTURE corridor graph rebuild")
+	var path_before: Array = runtime.path_service.find_path(Vector3i(-7, 1, 1), Vector3i(-4, 1, 1))
+	expect(not path_before.is_empty(), "proposed footprint lies across a connected corridor")
+	expect(service.validate_placement(origin)[&"reason"] == &"CORRIDOR_CONFLICT", "would disconnect BACK corridor: reject with no bypass")
+	expect(runtime.path_service.find_path(Vector3i(-7, 1, 1), Vector3i(-4, 1, 1)) == path_before, "rejected cut preserves route")
+	for x in range(-7, -3): runtime.world_service.set_walkable(Vector3i(x, 1, 1), false)
+	expect(service.validate_placement(origin)[&"reason"] == &"CORRIDOR_CONFLICT", "stale path authority also rejects conflict")
+	expect(runtime.rebuild_path_graph_from_world(), "corridor fixture removed explicitly")
+	var room := place(f)
+	expect(service.validate_placement(origin)[&"reason"] == &"BACK_OCCUPIED", "dedicated room occupancy blocks overlap")
+	expect(runtime.world_service.get_entity_kind(room) == Rooms.OCCUPANCY_KIND, "dedicated cross-reference occupancy kind")
+	_private_geometry()
+	finish("Peeping room geometry")
+
+func _private_geometry() -> void:
+	var f := fixture(false)
+	var runtime: Variant = f[&"runtime"]
+	# CONTROLLED_FIXTURE underground geometry only: no unlock/normal palette claim.
+	var bounds: Dictionary = runtime.world_service.get_bounds()
+	bounds[&"min_y"] = -1
+	expect(runtime.world_service.expand_bounds(bounds), "CONTROLLED_FIXTURE underground bounds")
+	var definition: FacilityDefinition = runtime.definition_registry.get_facility(&"adult_private_room")
+	for x in range(2):
+		var cell := Vector3i(x, -1, 0)
+		runtime.world_service.set_excavated(cell, true)
+		runtime.world_service.set_support(cell, definition.weight_per_cell, true)
+	var started: CommandResult = runtime.construction_service.start_construction(definition, Vector3i(0, -1, 0))
+	expect(started.is_success(), "CONTROLLED_FIXTURE paid real adult_private_room definition")
+	f[&"clock"].advance_minutes(definition.build_minutes)
+	var service: Variant = runtime.peeping_room_service
+	expect(service.validate_placement(Vector3i(0, -1, 1))[&"ok"], "adult_private_room valid completed geometry")
+	var target_id: StringName = runtime.world_service.get_occupant(Vector3i(0, -1, 0))
+	var target: FacilityInstance = runtime.facility_service.get_instance(target_id)
+	if target != null:
+		target.definition_id = &"vip_private_salon"
+		expect(service.validate_placement(Vector3i(0, -1, 1))[&"reason"] == &"TARGET_NOT_ALLOWED", "salon is not VIP suite substitute")
+		target.definition_id = &"adult_private_room"
+		runtime.facility_service.unregister_instance(target_id)
+		expect(service.validate_placement(Vector3i(0, -1, 1))[&"reason"] == &"TARGET_UNKNOWN", "world occupant without actual FacilityService target rejected")

tests/special/test_peeping_room_reference.gd

@@ -0,0 +1,37 @@
+extends "res://tests/special/peeping_room_fixture.gd"
+
+func _run() -> void:
+	var f := fixture()
+	var runtime: Variant = f[&"runtime"]
+	var room := place(f)
+	enable_fixture(f, room)
+	var old_target: StringName = f[&"target"]
+	var old_evidence: Dictionary = runtime.peeping_room_service.controlled_fixture_record_evidence(room, &"fixture.event", f[&"clock"].absolute_minute)
+	expect(old_evidence[&"ok"], "fixture evidence before removal")
+	expect(runtime.submit_command(&"demolish_facility", {&"instance_id": old_target}).is_success(), "real Runtime demolition")
+	var lost: Dictionary = runtime.peeping_room_service.get_room(room)
+	expect(lost[&"link_state"] == &"TARGET_LOST" and lost[&"stopped"], "target removal immediately marks lost and stopped")
+	expect(runtime.peeping_room_service.record_evidence(room, &"fixture.after_loss", f[&"clock"].absolute_minute)[&"reason"] == &"STOPPED", "collection rejected after loss with prescribed stopped precedence")
+	expect(not runtime.peeping_room_service.controlled_fixture_set_stopped(room, false), "lost target cannot resume even through fixture")
+	var lost_snapshot: Dictionary = runtime.create_snapshot()
+	expect(runtime.restore_snapshot(lost_snapshot), "lost reference roundtrip retains historical identity")
+	expect(var_to_bytes(lost_snapshot) == var_to_bytes(runtime.create_snapshot()), "lost snapshot exact typed roundtrip")
+	expect(runtime.submit_command(&"build_facility", {&"definition_id": &"public_bath", &"origin": Vector3i(-6, 0, 0)}).is_success(), "rebuild at same geometry starts")
+	f[&"clock"].advance_minutes(runtime.definition_registry.get_facility(&"public_bath").build_minutes)
+	var new_target: StringName = runtime.world_service.get_occupant(Vector3i(-6, 0, 0))
+	expect(new_target != old_target, "rebuild receives different target identity")
+	expect(runtime.peeping_room_service.get_room(room)[&"target_instance_id"] == old_target, "no implicit relink on geometry reuse")
+	var rejected: CommandResult = runtime.submit_command(&"peeping_room_relink", {&"room_id": room, &"target_instance_id": old_target})
+	expect(not rejected.is_success(), "explicit old target relink rejected")
+	expect(runtime.submit_command(&"peeping_room_relink", {&"room_id": room, &"target_instance_id": new_target}).is_success(), "explicit new target full validation")
+	var linked: Dictionary = runtime.peeping_room_service.get_room(room)
+	expect(linked[&"stopped"] and linked[&"stop_reason"] == &"RELINKED_PENDING_RESUME", "explicit relink never auto resumes")
+	f[&"target"] = new_target
+	enable_fixture(f, room)
+	var new_evidence: Dictionary = runtime.peeping_room_service.controlled_fixture_record_evidence(room, &"fixture.event", f[&"clock"].absolute_minute)
+	expect(new_evidence[&"ok"] and new_evidence[&"evidence_id"] != old_evidence.get(&"evidence_id"), "same room/event on different target never aliases")
+	var linked_snapshot: Dictionary = runtime.create_snapshot()
+	expect(runtime.restore_snapshot(linked_snapshot) and var_to_bytes(linked_snapshot) == var_to_bytes(runtime.create_snapshot()), "relinked ledger keeps both historical target bindings")
+	expect(runtime.facility_service.unregister_instance(new_target).is_success(), "equivalent FacilityService removal path")
+	expect(runtime.peeping_room_service.get_room(room)[&"link_state"] == &"TARGET_LOST", "unregister signal also invalidates target")
+	finish("Peeping room reference")

tests/special/test_peeping_room_runtime.gd

@@ -0,0 +1,91 @@
+extends "res://tests/special/peeping_room_fixture.gd"
+
+func _run() -> void:
+	var f := fixture()
+	var runtime: Variant = f[&"runtime"]
+	# Direct executor invocation distinguishes read-only domain state from queue receipts.
+	var validation := SimulationCommand.new()
+	validation.command_type = &"peeping_room_validate"
+	validation.payload = {&"origin": f[&"origin"]}
+	var before := var_to_bytes(runtime.create_snapshot())
+	expect(runtime._execute_peeping_room(validation).is_success(), "read-only validator executor")
+	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 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(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"})
+	expect(not blocked.is_success() and blocked.details[&"reason"] == &"RECORDER_ELIGIBILITY_PENDING_D04_D08_P16", "registered evidence command cannot reach ledger")
+	var missing: CommandResult = runtime.submit_command(&"peeping_room_record_evidence", {&"room_id": room})
+	expect(not missing.is_success() and missing.details[&"reason"] == &"SOURCE_EVENT_KEY_REQUIRED", "public command requires named key")
+	var malformed: CommandResult = runtime.submit_command(&"peeping_room_validate", {&"origin": [0, 1, 1]})
+	expect(not malformed.is_success() and malformed.details[&"reason"] == &"INVALID_PAYLOAD", "wrong payload type safely rejected")
+	expect(runtime.peeping_room_service.create_snapshot()[&"evidence"].is_empty(), "commands cannot inject arbitrary content events")
+	var fixture_command: CommandResult = runtime.submit_command(&"controlled_fixture_place", validation.payload)
+	expect(not fixture_command.is_success(), "no fixture command registration")
+	_mimi_format(f)
+	finish("Peeping room runtime")
+
+func _mimi_format(f: Dictionary) -> void:
+	var runtime: Variant = f[&"runtime"]
+	const Quote := preload("res://scripts/supply/supply_purchase_quote_v2.gd")
+	expect(runtime.submit_command(&"build_facility", {&"definition_id": &"warehouse", &"origin": Vector3i(2, 0, 1)}).is_success(), "paid warehouse for independent existing MIMI format path")
+	f[&"clock"].advance_minutes(runtime.definition_registry.get_facility(&"warehouse").build_minutes)
+	var warehouse: StringName = &""
+	for facility: FacilityInstance in runtime.facility_service.get_instances():
+		if facility.definition_id == &"warehouse": warehouse = facility.instance_id
+	expect(not warehouse.is_empty(), "completed real warehouse")
+	var mimi := StaffRuntime.new().configure(Quote.MIMI, &"porter", {&"employment_kind": StaffRuntime.EmploymentKind.UNIQUE})
+	expect(runtime.staff_service.register_staff(mimi).is_success(), "CONTROLLED_FIXTURE existing MIMI unique registration, no natural join claim")
+	expect(runtime.submit_command(&"set_staff_work_assignment", {&"staff_id": Quote.MIMI, &"facility_id": warehouse, &"template_id": &"NIGHT"}).is_success(), "existing persistent MIMI assignment")
+	expect(runtime.submit_command(&"purchase_supplies", {&"warehouse_id": warehouse, &"item_id": SupplyCatalog.RAW_FOOD, &"quantity": 1}).is_success(), "existing real v2 paid purchase")
+	var snapshot: Dictionary = runtime.create_snapshot()
+	expect(snapshot[&"systems"][&"runtime"][&"runtime_schema_version"] == 15 and snapshot[&"systems"][&"runtime"].has(&"special_peeping_rooms"), "v2 purchase history and special section coexist")
+	expect(runtime.restore_snapshot(snapshot) and var_to_bytes(snapshot) == var_to_bytes(runtime.create_snapshot()), "v15 optional section exact typed restore")
+	var directory := "user://tests/p13_reference_mimi_%d" % OS.get_process_id()
+	expect(SaveService.save_snapshot(directory, snapshot)[&"ok"], "physical v15 save accepts validated extension without loosening purchase allowlist")
+	var loaded := SaveService.load_snapshot(directory)
+	expect(loaded[&"ok"], "physical MIMI section save loads")
+	if loaded[&"ok"]:
+		expect(loaded[&"manifest"]["format_version"] == 2 and loaded[&"manifest"]["runtime_schema_version"] == 15, "format2 manifest still exactly15")
+		expect(runtime.restore_snapshot(loaded[&"snapshot"]), "MIMI package restore reaches same staged section validator")
+	var bad := snapshot.duplicate(true)
+	bad[&"systems"][&"runtime"][&"special_peeping_rooms"][&"version"] = 2
+	expect(not SaveService.save_snapshot(directory + "_bad", bad)[&"ok"], "save-format adapter rejects malformed optional section")

tests/special/test_peeping_room_save.gd

@@ -0,0 +1,66 @@
+extends "res://tests/special/peeping_room_fixture.gd"
+
+func _run() -> void:
+	var f := fixture()
+	var runtime: Variant = f[&"runtime"]
+	var empty: Dictionary = runtime.create_snapshot()
+	expect(not empty[&"systems"][&"runtime"].has(&"special_peeping_rooms"), "empty service does not alter existing snapshot keys")
+	expect(runtime.restore_snapshot(empty) and var_to_bytes(empty) == var_to_bytes(runtime.create_snapshot()), "pre-extension v14 absence exact roundtrip")
+	var legacy := empty.duplicate(true)
+	legacy[&"systems"][&"runtime"][&"runtime_schema_version"] = 13
+	legacy[&"systems"][&"runtime"].erase(&"auto_purchase_plans")
+	expect(runtime.restore_snapshot(legacy) and runtime.peeping_room_service.is_empty(), "genuine v13 absent section migrates empty")
+	expect(runtime.restore_snapshot(empty), "restore canonical baseline before controlled room")
+	var room := place(f)
+	enable_fixture(f, room)
+	expect(runtime.peeping_room_service.controlled_fixture_record_evidence(room, &"fixture.save", f[&"clock"].absolute_minute)[&"ok"], "controlled evidence before mid-state save")
+	expect(runtime.submit_command(&"peeping_room_relink", {&"room_id": room, &"target_instance_id": f[&"target"]}, &"player", room, 0, f[&"clock"].absolute_minute + 1).is_success(), "future explicit relink remains pending at checkpoint")
+	var snapshot: Dictionary = runtime.create_snapshot()
+	expect(snapshot[&"systems"][&"runtime"][&"runtime_schema_version"] == 14, "extension never bumps Runtime version")
+	expect(runtime.restore_snapshot(snapshot) and var_to_bytes(snapshot) == var_to_bytes(runtime.create_snapshot()), "exact typed room/evidence whole Runtime roundtrip")
+	var directory := "user://tests/p13_reference_save_%d" % OS.get_process_id()
+	var saved := SaveService.save_snapshot(directory, snapshot)
+	expect(saved[&"ok"], "actual SaveService writes section in format 1")
+	var loaded := SaveService.load_snapshot(directory)
+	expect(loaded[&"ok"], "actual SaveService reads section")
+	if loaded[&"ok"]:
+		expect(loaded[&"manifest"]["format_version"] == 1, "existing format mapping unchanged")
+		expect(runtime.restore_snapshot(loaded[&"snapshot"]), "loaded physical package reaches Runtime validation")
+		expect(var_to_bytes(snapshot) == var_to_bytes(runtime.create_snapshot()), "physical save retains all typed fields")
+	for mutation in range(18):
+		var bad := snapshot.duplicate(true)
+		var data: Dictionary = bad[&"systems"][&"runtime"]
+		var section: Dictionary = data[&"special_peeping_rooms"]
+		match mutation:
+			0: data.erase(&"special_peeping_rooms")
+			1: data[&"special_peeping_rooms"] = null
+			2: section[&"version"] = 2
+			3: section[&"version"] = 1.0
+			4: section[&"unknown"] = true
+			5: section.erase(&"evidence")
+			6: section[&"rooms"].append(section[&"rooms"][0].duplicate(true))
+			7: section[&"evidence"].append(section[&"evidence"][0].duplicate(true))
+			8: section[&"rooms"][0][&"target_instance_id"] = &"missing.target"
+			9: section[&"rooms"][0][&"target_definition_id"] = &"adult_private_room"
+			10: section[&"rooms"][0][&"origin"] = Vector3i(8, 1, 1)
+			11: section[&"rooms"][0][&"stopped"] = 1
+			12: section[&"evidence"][0][&"room_id"] = &"peeping_room.999999999999"
+			13: section[&"evidence"][0][&"source_event_key"] = &"aliased.event"
+			14: section[&"evidence"][0][&"minute"] = f[&"clock"].absolute_minute + 1
+			15: data[&"id_generator"]["counters"]["peeping_room"] = 0
+			16:
+				section.erase(&"version")
+				section["version"] = 1
+			17:
+				section[&"rooms"][0][&"link_state"] = &"TARGET_LOST"
+				section[&"rooms"][0][&"stopped"] = true
+				section[&"rooms"][0][&"stop_reason"] = &"TARGET_LOST"
+		expect(rejects_atomically(runtime, bad), "malformed/current missing/reference/duplicate rejection atomic case %d" % mutation)
+	var orphan := empty.duplicate(true)
+	orphan[&"systems"][&"runtime"][&"world"] = snapshot[&"systems"][&"runtime"][&"world"].duplicate(true)
+	expect(rejects_atomically(runtime, orphan), "world-owned room without any section rejects atomically")
+	var removed_world := snapshot.duplicate(true)
+	removed_world[&"systems"][&"runtime"][&"world"] = empty[&"systems"][&"runtime"][&"world"].duplicate(true)
+	expect(rejects_atomically(runtime, removed_world), "section-owned room without world occupancy rejects atomically")
+	expect(runtime.restore_snapshot(empty) and runtime.peeping_room_service.is_empty(), "absent legacy section clears previous nonempty service on valid load")
+	finish("Peeping room save")

tools/run_p13_reference_tests.py

@@ -0,0 +1,283 @@
+#!/usr/bin/env python3
+"""P13 reference evidence: read-only verification by default; collection is Root-only."""
+import argparse
+import ast
+import inspect
+import json
+import os
+import re
+import stat
+import types
+from pathlib import Path
+import shutil
+import sys
+
+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',
+    'tests/special/test_peeping_room_evidence.gd',
+    'tests/special/test_peeping_room_save.gd',
+    'tests/special/test_peeping_room_runtime.gd',
+]
+ROSTER = NEW_TESTS + [
+    'tests/s1/test_construction_service.gd',
+    'tests/s1/test_world_service.gd',
+    'tests/s1/test_path_service.gd',
+    'tests/s1/test_runtime_world_construction.gd',
+    'tests/supply/test_supply_save_migration.gd',
+    'tests/transport/test_transport_save.gd',
+    'tests/environment/test_runtime_weather.gd',
+    'tests/s6/test_runtime_snapshot_serialization.gd',
+    'tests/s6/test_runtime_save_lifecycle.gd',
+    'tests/supply/test_mimi_purchase_quote.gd',
+]
+
+
+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 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.
+    """
+    core = shared.module('p13_p18_core', shared.HERE / 'run_p18_relationship_tests.py')
+    namespace = dict(vars(core), ROSTER=ROSTER, __file__=__file__)
+    replacements = {
+        'guarded-p18-v2': PROTOCOL,
+        'tools/run_p18_relationship_tests.py': 'tools/run_p13_reference_tests.py',
+        'exact seven unique tests': 'exact unique P13 roster',
+        'exact nine phases': 'exact P13 phase count',
+        'all nine captured engines': 'all P13 captured engines',
+    }
+    cardinalities = {
+        'len(set(ROSTER)) == 7': 'len(set(ROSTER)) == len(ROSTER)',
+        'len(phases) == 9': 'len(phases) == len(ROSTER) + 2',
+        "len(tests) == 7": 'len(tests) == len(ROSTER)',
+    }
+
+    class Specialize(ast.NodeTransformer):
+        def __init__(self):
+            self.identities = 0
+            self.counts = 0
+            self.summary = 0
+
+        def visit_Constant(self, node):
+            if isinstance(node.value, str) and node.value in replacements:
+                self.identities += 1
+                return ast.copy_location(ast.Constant(replacements[node.value]), node)
+            return node
+
+        def visit_Compare(self, node):
+            expression = ast.unparse(node)
+            if expression in cardinalities:
+                self.counts += 1
+                return ast.copy_location(ast.parse(cardinalities[expression], mode='eval').body, node)
+            return self.generic_visit(node)
+
+        def visit_Dict(self, node):
+            for index, key in enumerate(node.keys):
+                if isinstance(key, ast.Constant) and key.value == 'actual_suites':
+                    shared.require(isinstance(node.values[index], ast.Constant)
+                                   and node.values[index].value == 7, 'upstream suite summary unchanged')
+                    node.values[index] = ast.parse('len(ROSTER)', mode='eval').body
+                    self.summary += 1
+            return self.generic_visit(node)
+
+    expected = {'collect': (2, 0, 0), 'seal': (1, 0, 0), 'verify': (6, 3, 1)}
+    for name, counts in expected.items():
+        adapter = Specialize()
+        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
+
+
+def main(argv=None):
+    argv = list(sys.argv[1:] if argv is None else argv)
+    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
+    parser.add_argument('--output')
+    modes = parser.add_mutually_exclusive_group()
+    modes.add_argument('--collect', action='store_true', help='Root only; fresh native collection')
+    modes.add_argument('--prepare-only', action='store_true')
+    seen, index = set(), 0
+    while index < len(argv):
+        flag = argv[index]
+        shared.require(flag in ('--output', '--collect', '--prepare-only', '--help')
+                       and flag not in seen, 'unknown, duplicate or malformed flag')
+        seen.add(flag)
+        if flag == '--output':
+            index += 1
+            shared.require(index < len(argv) and argv[index] and not argv[index].startswith('--'), 'output value required')
+        index += 1
+    args = parser.parse_args(argv)
+    shared.require(not args.collect or args.output is not None, 'Root collection requires explicit --output')
+    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) / CANONICAL_EVIDENCE)
+    args.timeout = 1200  # unchanged guarded core timeout; not a product value
+    core = protocol_core()
+    root = shared.HERE.parent
+    out = core['plain_path'](args.output)
+    refs = shared.source_refs(root)
+    shared.require(len(set(ROSTER)) == len(ROSTER) and set(ROSTER).issubset(shared.roster(root)), 'complete unique discovered roster')
+    markers = shared.module('p13_exact_markers', shared.HERE / 'strict_test_markers.py')
+    contract_path = root / 'docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json'
+    contract_sha = shared.sha(contract_path)
+    contracts = markers.load_contracts(contract_path)
+    shared.require(all(name in contracts for name in ROSTER), 'complete marker roster')
+    policy = shared.read(shared.SUPPORT / 'suite-policy.json')
+    shared.policy_coverage(ROSTER, refs, policy)
+    for test in ROSTER:
+        shared.phase_policy(test, Path(refs[test]['path']), policy['suite_policies'])
+    for test in NEW_TESTS:
+        shared.require(contracts[test]['source_sha256_at_schema_audit'] == refs[test]['sha256'], 'new strict source-hashed marker')
+    parse_paths = ['res://' + name for name in sorted(refs) if name.endswith('.gd')]
+    if args.prepare_only:
+        ancestor = out.parent
+        while not ancestor.exists():
+            ancestor = ancestor.parent
+        free = shutil.disk_usage(ancestor).free
+        size = sum(pin['bytes'] for pin in refs.values())
+        print(json.dumps(dict(protocol=PROTOCOL, roster=ROSTER, source_files=len(refs),
+            source_bytes=size, free_bytes=free, conservative_required_bytes=size + shared.MIN_SPACE,
+            capacity_available=free >= size + shared.MIN_SPACE, parse_paths=len(parse_paths),
+            future_native_phases=len(ROSTER) + 2, actual_checks=None, engines=0, stages=0,
+            formal_acceptance=False, original_unit_complete=False)))
+        return 0
+    if args.collect:
+        shared.require(not out.exists() and out.parent.is_dir(), 'fresh output with existing parent required')
+        shared.require(not out.is_relative_to(root), 'private collection outside source tree')
+        return core['collect'](args, root, out, refs, markers, contracts, policy, contract_path, contract_sha, parse_paths)
+    return core['verify'](out, refs, markers, contracts, policy, contract_sha, parse_paths)
+
+
+if __name__ == '__main__':
+    try:
+        raise SystemExit(main())
+    except (AssertionError, OSError, ValueError, KeyError, TypeError) as error:
+        print(type(error).__name__ + ': ' + str(error), file=sys.stderr)
+        raise SystemExit(1)