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

docs/plans/original-completion-2026-10-02/execution/p04-01/s3/S3_ROOT_RUN.md

@@ -0,0 +1,95 @@
+# P04-01 S3 code-ready preparation
+
+変更は本ディレクトリのGD、Python wrapper、本書の3本のみ。製品修正0。
+本提出は原P04-01の残る5条件の検証コード準備であり、原成果の加点は0。
+原7領域、3安全結果、15条件＋EV、30/88、部分点0を維持する。
+S1r02、S2ar03、S2b native r02は保持し再実行しない。
+
+## 固定入力と6実ケース
+
+最初に供給SOURCE_MANIFEST.jsonの7ファイルについて全byte SHA256とbyte数の一致を確認した。
+manifest自体のSHAは `d4993398f7619ca37bcd64c32b79bdee892d44e31f332cc0dd5f31fb7bc28314`。
+wrapperは同manifest、7供給ファイル、Rootの現行source audit、S2bの3ファイル、本3ファイルをpinする。
+選択rootの全sourceもsource_refsでpinし、その後にQAを1件だけ明示overrideする。
+override先はS2bと同じ `tests/s3/test_runtime_staff_assignment.gd`。
+これはprivate stageだけのoverrideで、継承先ActorHardeningを置き換えない。
+
+| case | seed | 境界と検証対象 |
+| --- | --- | --- |
+| B1 | 420410 | 959に開始済Aと実queue B。960に有限mana枯渇と対象への直接休業。960/961はAの進捗・同じ予約を保持し課金0、Bはrelease・在庫返却・忍耐不消費。元reserveへ公開復旧し、休業のまま962にAだけ1回課金。963で重複なし。 |
+| B2a | 420430 | 原定義のduration/priceで960完了・課金後に通常退出、961 Despawnで二重課金なし。 |
+| B2b | 420431 | 959の実queueを960の公開leaveでadmission前にrelease。961通常退出、962 Despawn。課金0。 |
+| B3 | 420435 | 元traveler.base_patience、実サーバー不足queue。template同様にfixture時刻＋base_patienceを境界とし、通常tickで枯渇。ABANDONING→DECIDING→LEAVING→Despawn、返却1回・課金0。planned_exitは元templateの境界値。 |
+| B4a | 420420 | 元diagnostic automatic DEFENSE Event、foreground=false、durability=0。960の完了・1回課金が後段Event破壊に先行。961で重複なし。 |
+| B4b | 420421 | 元durationの入場offset1。960は進捗後にEvent破壊、961でFAILED・release・返却1回、課金0。 |
+
+7件目は作らない。各caseはfresh Runtime/Clock/GameState、厨房は現行ActorHardeningと
+required_kitchen_setupの物理配置FLEX fixture。自然hire/unlockの証拠ではない。
+サービス価格・時間は定義取得、capacity/staff_min変更なし、忍耐・elapsedの直接変更なし。
+旧1299 wrapper、未承認18:00 predicate、本番Operations enableは使用しない。
+
+D-034（供給DESIGN_DECISIONS.md 252–262行、特に258行）の直接休業では開始済を継続する。
+関連厨房不稼働によるpauseとは区別する。interruption_reasonは実snapshotのまま残し、
+RELATED_FACILITY_INACTIVEをNO_POWERへ置換しない。
+enum snapshotは各create_snapshotのString定義に従い、stateをintへ変換しない。
+
+## Rawと正式保存
+
+Writerの各before/after/next、B1 recovered/duplicate_guard、B2b/B3 despawnを正式保存する（計22slot）。
+各点の全Runtime typed snapshot、入力・seed・serial・実ID・Guest/wallet/planned_exit、
+service/request/reservation、在庫、ledger、Event instanceを無加工の.varとJSONで一度だけ保存する。
+各tickの全11phase開始点も同様に保存し、phase順と境界内部の状態を残す。
+case/indexはrawのpath/SHAを参照し、巨大snapshotのaggregate多重コピーを作らない。
+全体rawをobsolete 500000-byte capに合わせて省略・圧縮しない。
+public command入力と結果、継承fixture分を含むcommand_completed結果をcase別rawに保存する。
+fixture自身の公開command入力・IDは全Runtime command queue履歴にも残る。
+Operationsは正式snapshotのdisabled/coordinator={}と実coordinator不在を別々に記録し、無関係なgetterを呼ばない。
+
+公開SaveService.save_snapshot/inspect_snapshotの成功・全typed一致・live Runtime不変を確認する。
+Readerは別PID/profileで22slotの全member（journal.log/manifest.json/thumbnail.png/world.bin）をコピーし、
+原/copyの全SHA、load/replay後の両者不変を確認する。余分なmember/directoryは失敗する。
+全点を公開load/restoreしてWriterの全Runtime typed bytesと一致させる。
+さらに各beforeをfresh loadし、通常clockで同じ全境界を再生する。
+B1のfuture recoveryについてserial、idempotency key、command ID、issue minute、元reserveをbeforeに保存する。
+Readerはserial0から始め、保存済probe入力を復元して未来コマンドだけを実行する。setup keyを再利用しない。
+全pointとphaseのtyped比較からフィールドを削除せず、JSON正規化や要約への置換も行わない。
+再生中のphaseはWriterと同じ順・全typed状態、pointは入力・receiptまで含む全typed一致を必須とする。
+
+## Root手順
+
+Rootだけが現行製品1338でfresh plan→nativeを1回実行する。
+worker checkout planを製品nativeへ転用しない。既存plan/outputは上書きしない。
+
+```sh
+export BUILD_OUTPUT_DIR='/Users/korizatou114/Projects/TheTree-managed-worktrees/p04-01/.runtime/P04-01/build'
+export PYTHONDONTWRITEBYTECODE=1
+python3 docs/plans/original-completion-2026-10-02/execution/p04-01/s3/run_s3_native.py --check-only --plan "$BUILD_OUTPUT_DIR/p04-s3-root-plan-r01.json" --output "$BUILD_OUTPUT_DIR/p04-s3-native-r01"
+python3 docs/plans/original-completion-2026-10-02/execution/p04-01/s3/run_s3_native.py --plan "$BUILD_OUTPUT_DIR/p04-s3-root-plan-r01.json" --output "$BUILD_OUTPUT_DIR/p04-s3-native-r01"
+```
+
+製品root既定値: `/Users/korizatou114/Projects/とりにく工房ゲーム制作部`。
+出力先: `/Users/korizatou114/Projects/TheTree-managed-worktrees/p04-01/.runtime/P04-01/build/p04-s3-native-r01/`。
+一つのprivate stage/cacheでimport1回、parse1回、Writer1回、別PID Reader1回。
+S2bのstrict guard、policy/watchdog、marker/diagnostic/ownership/audio gateを踏襲する。
+source/cacheは別inode、owned finally後に全source/default profile/private source/pin/original cacheを検査する。
+Writer formal slot不変も失敗時を含めて検査し、4phase・4実PIDの終了とreceipt PIDを必須とする。
+check-onlyはengine0、private/default profileアクセス0、所有helper未読込。cache receiptの実読取エラーはplanに残しnativeを拒否する。
+
+## 残gapとworker検証
+
+workerはpy_compile、check-only、型/参照の静的確認だけを行う。
+Godot parse・native・全回帰・Runner操作・git書込み・他agent起動は0。
+GDの正式parseと6ケースの実成立、全正式Save/Load/replay一致はRoot nativeまで未実証。
+S3 raw取得後のcollector/A仕上げ、原15＋EV全証拠の適用性照合、D05の原未確定規約、
+必要なOS別条件、正式独立レビュー・統合は本提出で確定しない。
+未知の原規約を新たに作らずpendingとし、P04全体完成を主張しない。
+
+worker実施記録: py_compile成功（出力はtask buildのp04-s3-worker-pycompile.pyc）。
+GDの明示型変数・loop・関数戻り値、禁止されたpatience/elapsed直接代入なし、継承先literal参照、
+15のservice method参照、Runtime method/signal参照、括弧整合を静的確認した。Godot parseの代替ではない。
+製品rootでのcheck-onlyはtools/run_all_tests.py読取りがPermissionErrorでexit2。
+明示worktree rootではcheck-only成功（source1300、QA1、fixed pins18、engine0）。
+終了済cacheのrunner-ended.jsonはPermissionErrorであり、planのcache_accessに実path/errorを記録した。
+製品1338のcheck-only成功やcache再利用の成立は主張しない。Rootが自身の環境で確認する。
+初回worktree planはbuild/p04-s3-worker-check-r01.json。最終3ファイルのpin確認にはfreshな
+build/p04-s3-worker-check-r02.jsonを使用し、r01は保持する。

docs/plans/original-completion-2026-10-02/execution/p04-01/s3/p04_s3_closure_probe.gd

@@ -0,0 +1,480 @@
+extends "res://tests/s3/test_runtime_actor_hardening.gd"
+
+const CASES: Array[String] = ["B1", "B2a", "B2b", "B3", "B4a", "B4b"]
+const SEEDS: Dictionary = {"B1": 420410, "B2a": 420430, "B2b": 420431, "B3": 420435, "B4a": 420420, "B4b": 420421}
+const GUEST: StringName = &"p04.s3.guest.a"
+const REQUEST: StringName = &"p04.s3.request.a"
+const QUEUED_GUEST: StringName = &"p04.s3.guest.b"
+const QUEUED_REQUEST: StringName = &"p04.s3.request.b"
+const SERVERS: Array[StringName] = [&"p04.order.server.a", &"p04.order.server.b"]
+var _evidence_dir: String = ""
+var _mode: String = ""
+var _case: String = ""
+var _serial: int = 0
+var _facility_id: StringName = &""
+var _service_id: StringName = &""
+var _boundary: int = 960
+var _price: int = 0
+var _duration: int = 0
+var _patience: int = 0
+var _reserve: int = 0
+var _recovery: Dictionary = {}
+var _commands: Array = []
+var _results: Array = []
+var _receipts: Array = []
+var _phases: Array = []
+var _points: Dictionary = {}
+var _branches: Dictionary = {}
+var _slots: Dictionary = {}
+
+func _run() -> void:
+	_evidence_dir = OS.get_environment("THE_TREE_S3_EVIDENCE")
+	_mode = OS.get_environment("THE_TREE_S3_MODE")
+	if _evidence_dir.is_empty() or _mode not in ["writer", "reader"]:
+		quit(2)
+		return
+	_expect(DirAccess.make_dir_recursive_absolute(_evidence_dir) == OK, "external evidence directory")
+	if _mode == "writer":
+		for branch: String in CASES:
+			_case = branch
+			_writer_case()
+	else:
+		_reader()
+	_write(_mode, {&"pid": OS.get_process_id(), &"profile": OS.get_user_data_dir(), &"slots": _slots,
+		&"branches": _branches, &"checks": _checks, &"failures": _failures,
+		&"whole_goal_accepted": false, &"unit_accepted": false}, true)
+	_cleanup()
+	print("S3 closure: %d checks, %d failures" % [_checks, _failures])
+	quit(1 if _failures else 0)
+
+func _fresh() -> void:
+	_serial = 0
+	_facility_id = &""
+	_service_id = &""
+	_boundary = 960
+	_price = 0
+	_duration = 0
+	_patience = 0
+	_reserve = 0
+	_recovery = {}
+	_commands = []
+	_results = []
+	_receipts = []
+	_phases = []
+	_points = {}
+	_reset_runtime(SEEDS[_case])
+	_runtime.configure_save_policy({&"enabled": false, &"daily_checkpoint_enabled": false})
+	_runtime.command_completed.connect(func(result: CommandResult) -> void:
+		_results.append({&"minute": _clock.absolute_minute, &"result": result.to_dictionary()})
+		_expect(result.is_success(), "actual public command completion " + str(result.command_id))
+	)
+
+func _cmd(kind: StringName, payload: Dictionary, at: int = -1) -> CommandResult:
+	_serial += 1
+	var key: StringName = StringName("p04.s3.%s.command.%d" % [_case, _serial])
+	var result: CommandResult = _runtime.submit_command(kind, payload, &"player", &"", 10, at, key)
+	_commands.append({&"serial": _serial, &"kind": kind, &"payload": payload.duplicate(true),
+		&"at": at, &"issued_at": _clock.absolute_minute, &"idempotency_key": key, &"result": result.to_dictionary()})
+	_expect(result.is_success(), "public command " + String(kind))
+	return result
+
+func _queue(guest_id: StringName, request_id: StringName, exit_at: int) -> void:
+	_cmd(&"spawn_guest", {&"guest_id": guest_id, &"guest_type_id": &"traveler", &"gold": _price * 2,
+		&"options": {&"state": GuestRuntime.State.DECIDING, &"current_cell": Vector3i.ZERO, &"planned_exit_at": exit_at}})
+	_cmd(&"create_service_request", {&"facility_id": _facility_id, &"request_id": request_id, &"guest_ids": [guest_id]})
+	var guest: GuestRuntime = _runtime.guest_service.get_guest(guest_id)
+	var request: ServiceRequest = _runtime.facility_service.get_request(request_id)
+	_expect(guest != null and request != null, "actual Guest and request exist")
+	if guest == null or request == null: return
+	_expect(guest.state == GuestRuntime.State.QUEUING and guest.queue_request_id == request_id, "actual queued ownership before advancing clock")
+	_expect(request.quoted_price == _price and request.service_minutes == _duration and request.reserved_inventory_ids.size() == 1, "definition price/duration and one meal reservation")
+	for id: StringName in request.reserved_inventory_ids:
+		var reservation: InventoryReservation = _runtime.inventory_service.get_reservation(id)
+		_expect(reservation != null and reservation.is_active() and reservation.requester_id == request_id, "public queue actually reserves meal")
+
+func _prepare() -> bool:
+	_fresh()
+	var facility: FacilityInstance = _build_cafeteria(true)
+	if facility == null: return false
+	_facility_id = facility.instance_id
+	var definition: FacilityDefinition = _runtime.definition_registry.get_facility(facility.definition_id)
+	var traveler: GuestDefinition = _runtime.guest_service.get_definition(&"traveler")
+	_price = definition.base_price
+	_duration = definition.base_service_minutes
+	_patience = traveler.base_patience
+	if _case == "B3":
+		_expect(_patience > 1 and _patience < 240, "original patience precedes reservation expiry")
+		if _patience <= 1 or _patience >= 240: return false
+		_boundary = _clock.absolute_minute + _patience
+		_expect(facility.operator_open and facility.is_within_opening_hours(_clock.absolute_minute) and facility.related_facility_operating and facility.active_staff_ids.size() < facility.staff_min, "real no-server queue with operating kitchen")
+	else:
+		for id: StringName in SERVERS:
+			_cmd(&"register_staff", {&"staff_id": id, &"role_id": &"server", &"options": {&"state": StaffRuntime.State.OFF_DUTY, &"current_cell": Vector3i.ZERO}})
+			_cmd(&"set_staff_work_assignment", {&"staff_id": id, &"template_id": &"DAY", &"facility_id": _facility_id})
+		var queue_at: int = 959 if _case == "B2b" else 960 - _duration - 1 + (1 if _case == "B4b" else 0)
+		_expect(queue_at > _clock.absolute_minute, "normal ticks reach queue time")
+		_clock.advance_minutes(queue_at - _clock.absolute_minute)
+		_expect(facility.active_staff_ids == SERVERS, "unchanged real server attendance")
+		for id: StringName in SERVERS:
+			var worker: StaffRuntime = _runtime.staff_service.get_staff(id)
+			_expect(_runtime.staff_service.requires_attendance(id) and StaffCommuteService.is_at_facility(worker, facility) and worker.current_cell != Vector3i.ZERO, "ordinary public server commute")
+	if _case in ["B4a", "B4b"]:
+		var action: EventAction = EventAction.new().configure(&"p04.destroy.action", EventAction.Type.SET_FACILITY_STATE, {&"facility_id": _facility_id, &"conditions": {&"durability": 0, &"suspension_reason": &"DESTROYED"}})
+		_cmd(&"register_event_definition", {&"definition": {&"event_id": &"p04.destroy.event", &"category": &"DEFENSE", &"foreground": false, &"transaction_policy": &"ROLLBACK_ALL", &"automatic_actions": [action.create_snapshot()]}})
+	_expect(_stock(facility.local_inventory_id, &"ingredient/basic_meal", 1, &"p04.order.shift.stock"), "original public stock helper")
+	var exit_at: int = _boundary if _case in ["B2a", "B2b", "B3"] else _boundary + _duration
+	_queue(GUEST, REQUEST, exit_at)
+	if _case == "B3":
+		_clock.advance_minutes(_patience - 1)
+		var guest: GuestRuntime = _runtime.guest_service.get_guest(GUEST)
+		_expect(guest.state == GuestRuntime.State.QUEUING and guest.patience_remaining == 1 and _runtime.facility_service.get_active_services(_facility_id).is_empty(), "normal ticks exhaust original patience, no manual mutation")
+	elif _case == "B2b":
+		_cmd(&"leave_facility_queue", {&"facility_id": _facility_id, &"request_id": REQUEST}, 960)
+	else:
+		_clock.advance_one_minute()
+		var active: Array = _runtime.facility_service.get_active_services(_facility_id)
+		_expect(active.size() == 1, "one public service starts")
+		if active.size() != 1: return false
+		var service: ServiceInstance = active[0]
+		_service_id = service.service_id
+		var offset: int = 1 if _case == "B4b" else 0
+		_expect(service.duration_minutes == _duration and service.price_snapshot == _price and service.started_at + _duration == 960 + offset, "original service completion alignment")
+		_clock.advance_minutes(959 - _clock.absolute_minute)
+		_expect(service.elapsed_minutes == _duration - 1 - offset, "measured boundary-before progress")
+	if _case == "B1":
+		_expect(_stock(facility.local_inventory_id, &"ingredient/basic_meal", 1, &"p04.s3.stock.b"), "second actual meal at 959")
+		_queue(QUEUED_GUEST, QUEUED_REQUEST, 960 + _duration)
+		_reserve = _state.get_resource(&"mana")
+		_expect(_reserve > 0, "original positive finite mana reserve")
+		_cmd(&"change_resources", {&"deltas": {&"mana": -_reserve}}, 960)
+		_cmd(&"set_facility_policy", {&"facility_id": _facility_id, &"patch": {&"business_open": false}}, 960)
+		var snapshot: Dictionary = _runtime.create_snapshot()
+		_recovery = {&"serial": _serial + 1, &"kind": &"change_resources", &"payload": {&"deltas": {&"mana": _reserve}},
+			&"idempotency_key": StringName("p04.s3.%s.command.%d" % [_case, _serial + 1]),
+			&"command_id": snapshot[&"systems"][&"runtime"][&"next_command_id"], &"at": -1, &"issue_minute": 961}
+	if _case in ["B4a", "B4b"]:
+		_cmd(&"enqueue_event", {&"event_id": &"p04.destroy.event", &"actor_ids": [], &"options": {&"event_instance_id": &"p04.destroy.instance"}})
+	_expect(_clock.absolute_minute == _boundary - 1, "exact boundary-before time")
+	return _clock.absolute_minute == _boundary - 1
+
+func _connect_observations() -> void:
+	_runtime.economy_service.transaction_posted.connect(func(id: StringName, entries: Array) -> void:
+		var rows: Array = []
+		for entry: LedgerEntry in entries: rows.append(entry.to_dictionary())
+		_receipts.append({&"id": id, &"minute": _clock.absolute_minute, &"entries": rows})
+	)
+	_clock.tick_phase_started.connect(func(phase: int, minute: int) -> void:
+		var label: String = "phase-%d-%s" % [minute, SimulationKernel.TickPhase.keys()[phase]]
+		var row: Dictionary = _capture()
+		row[&"phase"] = SimulationKernel.TickPhase.keys()[phase]
+		_assert_phase(row)
+		_phases.append(_write(label, row))
+	)
+
+func _assert_phase(point: Dictionary) -> void:
+	if point[&"minute"] != _boundary: return
+	var phase: String = point[&"phase"]
+	if _case in ["B4a", "B4b"]:
+		if phase == "INCIDENTS_AND_EVENTS":
+			_expect(point[&"facility"][&"durability"] > 0 and point[&"event"][&"state"] == "QUEUED", "facility phase precedes diagnostic destruction")
+			_expect(point[&"service"][&"state"] == ("COMPLETED" if _case == "B4a" else "ACTIVE"), "actual service state before Event phase")
+			_expect(point[&"receipts"].size() == (1 if _case == "B4a" else 0), "completion payment occurs before destruction, not while stopped")
+		if phase == "FACILITY_QUEUES":
+			_expect(point[&"facility"][&"durability"] == 0 and point[&"event"][&"state"] == "COMPLETED", "Event committed before queue phase")
+	if _case == "B3" and phase == "AI_DECISIONS":
+		var guest: Dictionary = point[&"guests"][String(GUEST)]
+		_expect(guest[&"state"] == "ABANDONING" and guest[&"patience_remaining"] == 0 and guest[&"queue_request_id"].is_empty(), "actual auto-abandon before AI redecision")
+	if _case == "B2b" and phase == "FACILITY_COMPLETION":
+		_expect(point[&"requests"][String(REQUEST)].is_empty() and point[&"active_services"] == 0 and point[&"available_meals"] == 1, "scheduled leave released before admission")
+	if _case == "B1" and phase == "AI_DECISIONS":
+		var queued: Dictionary = point[&"guests"][String(QUEUED_GUEST)]
+		_expect(not queued.is_empty() and queued[&"queue_request_id"].is_empty() and queued[&"patience_remaining"] == _patience, "direct closure releases B before AI with no patience penalty")
+
+func _capture() -> Dictionary:
+	var snapshot: Dictionary = _runtime.create_snapshot()
+	var operations: Dictionary = snapshot[&"systems"][&"runtime"][&"operations"]
+	var facility: FacilityInstance = _runtime.facility_service.get_instance(_facility_id)
+	var service: ServiceInstance = _runtime.facility_service.get_service(_service_id)
+	var event: EventInstance = _runtime.event_service.get_instance(&"p04.destroy.instance")
+	var actors: Dictionary = {}
+	var requests: Dictionary = {}
+	var wallets: Dictionary = {}
+	for id: StringName in [GUEST, QUEUED_GUEST]:
+		var guest: GuestRuntime = _runtime.guest_service.get_guest(id)
+		actors[String(id)] = guest.create_snapshot() if guest != null else {}
+		wallets[String(id)] = _runtime.economy_service.get_balance(StringName("GUEST_WALLET:" + String(id)))
+	for id: StringName in [REQUEST, QUEUED_REQUEST]:
+		var request: ServiceRequest = _runtime.facility_service.get_request(id)
+		requests[String(id)] = request.create_snapshot() if request != null else {}
+	var reservations: Array = []
+	for reservation: InventoryReservation in _runtime.inventory_service.get_reservations(false): reservations.append(reservation.create_snapshot())
+	var ledger: Array = []
+	for entry: LedgerEntry in _runtime.economy_service.get_entries(): ledger.append(entry.to_dictionary())
+	return {&"snapshot": snapshot, &"minute": _clock.absolute_minute,
+		&"inputs": {&"case": _case, &"seed": SEEDS[_case], &"serial": _serial, &"facility_id": _facility_id, &"service_id": _service_id,
+			&"boundary": _boundary, &"price": _price, &"duration": _duration, &"patience": _patience, &"reserve": _reserve, &"recovery": _recovery.duplicate(true),
+			&"kitchen_premise": "ActorHardening physical FLEX fixture; not natural hire/unlock"},
+		&"facility": facility.create_snapshot(), &"service": service.create_snapshot() if service != null else {},
+		&"event": event.create_snapshot() if event != null else {}, &"guests": actors, &"requests": requests, &"wallets": wallets,
+		&"reservations": reservations, &"ledger": ledger, &"receipts": _receipts.duplicate(true),
+		&"player_cash": _runtime.economy_service.get_balance(_runtime.PLAYER_CASH_ACCOUNT),
+		&"available_meals": _runtime.inventory_service.available_quantity(facility.local_inventory_id, &"ingredient/basic_meal"),
+		&"active_services": _runtime.facility_service.get_active_services(_facility_id).size(),
+		&"operations_enabled": operations[&"enabled"], &"saved_coordinator": operations[&"coordinator"],
+		&"actual_coordinator_absent": _runtime.operations_coordinator == null, &"unrelated_getters_called": false}
+
+func _labels() -> Array[String]:
+	if _case == "B1": return ["before", "after", "next", "recovered", "duplicate_guard"]
+	if _case in ["B2b", "B3"]: return ["before", "after", "next", "despawn"]
+	return ["before", "after", "next"]
+
+func _advance(label: String) -> void:
+	if label == "recovered":
+		_expect(_serial + 1 == _recovery[&"serial"] and _clock.absolute_minute == _recovery[&"issue_minute"], "saved future recovery serial and issue minute")
+		var result: CommandResult = _cmd(_recovery[&"kind"], _recovery[&"payload"], _recovery[&"at"])
+		var command: Dictionary = _commands.back()
+		_expect(result.command_id == _recovery[&"command_id"] and command[&"idempotency_key"] == _recovery[&"idempotency_key"], "same future public recovery command ID/key")
+	_clock.advance_one_minute()
+
+func _writer_case() -> void:
+	if not _prepare(): return
+	_connect_observations()
+	var before: Dictionary = _capture()
+	_assert_before(before)
+	_save_point("before", before)
+	for label: String in _labels():
+		if label == "before": continue
+		_advance(label)
+		var actual: Dictionary = _capture()
+		_assert_outcome(before, actual)
+		_save_point(label, actual)
+	var branch: Dictionary = {&"points": _points.duplicate(true), &"phases": _phases.duplicate(true),
+		&"commands": _write("commands", {&"submitted": _commands, &"completed_including_fixture": _results})}
+	_branches[_case] = _write("case", branch)
+
+func _assert_before(point: Dictionary) -> void:
+	_expect(not point[&"operations_enabled"] and point[&"actual_coordinator_absent"] and point[&"saved_coordinator"] == {}, "Operations disabled and actual absence measured")
+	var guest: Dictionary = point[&"guests"][String(GUEST)]
+	_expect(guest[&"planned_exit_at"] >= _boundary, "planned exit does not short-circuit queue/service")
+	var expected_count: int = 2 if _case == "B1" else 1
+	_expect(point[&"reservations"].size() == expected_count, "only actual requested meals reserved")
+	for reservation: Dictionary in point[&"reservations"]:
+		_expect(reservation[&"state"] == "ACTIVE" and reservation[&"purpose"] == "facility_service", "actual ACTIVE reservation before boundary")
+	_expect(point[&"available_meals"] == 0, "all stocked meals are reserved")
+	if _case == "B1":
+		var queued: Dictionary = point[&"guests"][String(QUEUED_GUEST)]
+		_expect(queued[&"state"] == "QUEUING" and queued[&"patience_remaining"] == _patience, "B exists at 959 with unchanged original patience")
+
+func _assert_reservation(before: Dictionary, point: Dictionary, request_id: StringName, state: String, terminal: int) -> void:
+	var prior: Array = []
+	var current: Array = []
+	for row: Dictionary in before[&"reservations"]:
+		if row[&"requester_id"] == String(request_id): prior.append(row)
+	for row: Dictionary in point[&"reservations"]:
+		if row[&"requester_id"] == String(request_id): current.append(row)
+	_expect(prior.size() == 1 and current.size() == 1, "one original reservation per request")
+	if prior.size() != 1 or current.size() != 1: return
+	_expect(prior[0][&"reservation_id"] == current[0][&"reservation_id"] and current[0][&"state"] == state and current[0][&"terminal_at"] == terminal, "same reservation ID/state/terminal time; no second release")
+	if state == "ACTIVE":
+		_expect(var_to_bytes(prior[0]) == var_to_bytes(current[0]), "entire held reservation unchanged")
+		var request: Dictionary = point[&"requests"][String(request_id)]
+		_expect(not request.is_empty() and request[&"reserved_inventory_ids"].has(current[0][&"reservation_id"]), "active reservation has live request owner")
+	else:
+		_expect(point[&"requests"][String(request_id)].is_empty(), "terminal reservation has no orphan request")
+	if state == "RELEASED":
+		var release_count: int = 0
+		var inventory: Dictionary = point[&"snapshot"][&"systems"][&"runtime"][&"inventory"]
+		for operation: Dictionary in inventory[&"idempotency_results"]:
+			var result: Dictionary = operation[&"result"]
+			if operation[&"action"] == "release" and String(result.get(&"reservation_id", "")) == current[0][&"reservation_id"]:
+				release_count += 1
+				_expect(bool(result.get(&"ok", false)) and result.get(&"quantity", 0) == current[0][&"quantity"], "release returned actual reserved quantity, not an orphan repair/retry")
+		_expect(release_count == 1, "one raw inventory release operation for original reservation")
+
+func _assert_outcome(before: Dictionary, point: Dictionary) -> void:
+	var minute: int = point[&"minute"]
+	var service: Dictionary = point[&"service"]
+	var guest: Dictionary = point[&"guests"][String(GUEST)]
+	var paid: bool = _case in ["B2a", "B4a"] or (_case == "B1" and minute >= 962)
+	_expect(point[&"wallets"][String(GUEST)] == _price * (1 if paid else 2), "actual wallet delta follows boundary")
+	_expect(point[&"receipts"].size() == (1 if paid else 0), "actual payment notification exactly once or zero")
+	_expect(point[&"player_cash"] == before[&"player_cash"] + (_price if paid else 0), "actual player cash delta exactly matches single price")
+	if not paid:
+		_expect(var_to_bytes(point[&"ledger"]) == var_to_bytes(before[&"ledger"]), "zero ledger charge while stopped or abandoned")
+	if paid:
+		var expected_ledger: Array = before[&"ledger"].duplicate(true)
+		for receipt: Dictionary in point[&"receipts"]: expected_ledger.append_array(receipt[&"entries"])
+		_expect(var_to_bytes(point[&"ledger"]) == var_to_bytes(expected_ledger), "whole ledger delta is exactly actual payment receipt, no extra posting")
+		var completed_at: int = 962 if _case == "B1" else 960
+		_expect(service[&"state"] == "COMPLETED" and service[&"completed_at"] == completed_at, "one actual completion")
+		_assert_reservation(before, point, REQUEST, "CONSUMED", completed_at)
+		_expect(point[&"active_services"] == 0, "completed service releases active slot")
+	if _case == "B1":
+		_expect(not point[&"facility"][&"operator_open"], "direct closure remains in force including recovery")
+		_assert_reservation(before, point, QUEUED_REQUEST, "RELEASED", 960)
+		var queued: Dictionary = point[&"guests"][String(QUEUED_GUEST)]
+		if minute == 960: _expect(not queued.is_empty(), "released B measured on closure tick")
+		if not queued.is_empty():
+			_expect(queued[&"queue_request_id"].is_empty() and queued[&"patience_remaining"] == before[&"guests"][String(QUEUED_GUEST)][&"patience_remaining"], "closure releases B without patience consumption; later natural Despawn is allowed")
+		_expect(point[&"wallets"][String(QUEUED_GUEST)] == _price * 2 and point[&"available_meals"] == 1, "B stock returned once and wallet unchanged")
+		if not paid:
+			_expect(not point[&"facility"][&"power_available"] and service[&"state"] == "PAUSED" and service[&"elapsed_minutes"] == before[&"service"][&"elapsed_minutes"], "outage holds original progress; actual interruption_reason retained in raw")
+			_assert_reservation(before, point, REQUEST, "ACTIVE", -1)
+	elif _case in ["B2b", "B3"]:
+		_assert_reservation(before, point, REQUEST, "RELEASED", _boundary)
+		_expect(point[&"active_services"] == 0 and point[&"available_meals"] == 1, "no service/orphan and exact stock return")
+		if minute == _boundary:
+			_expect(not guest.is_empty() and guest[&"state"] == "DECIDING" and guest[&"queue_request_id"].is_empty(), "automatic redecision after release")
+			if _case == "B3": _expect(guest[&"patience_remaining"] == 0, "normal queue tick exhausted original patience")
+		elif minute == _boundary + 1:
+			_expect(not guest.is_empty() and guest[&"state"] == "LEAVING", "normal next-minute exit")
+		else: _expect(guest.is_empty(), "normal Despawn boundary")
+	elif _case == "B2a":
+		_expect(point[&"available_meals"] == 0, "completed meal consumed")
+		if minute == 960: _expect(not guest.is_empty() and guest[&"state"] == "LEAVING" and guest[&"state_entered_at"] == 960, "completion precedes normal planned exit")
+		else: _expect(guest.is_empty(), "next-tick despawn without second payment")
+	else:
+		_expect(point[&"facility"][&"durability"] == 0 and point[&"facility"][&"state"] == "BROKEN" and point[&"event"][&"state"] == "COMPLETED", "actual automatic destruction Event committed")
+		if _case == "B4a": _expect(point[&"available_meals"] == 0, "completion before destruction consumes meal")
+		elif minute == 960:
+			_expect(service[&"state"] == "ACTIVE" and service[&"elapsed_minutes"] == _duration - 1, "late Event follows facility completion phase")
+			_assert_reservation(before, point, REQUEST, "ACTIVE", -1)
+		else:
+			_expect(service[&"state"] == "FAILED" and service[&"elapsed_minutes"] == _duration - 1 and point[&"available_meals"] == 1 and point[&"active_services"] == 0, "next facility tick fails without progress or payment")
+			_assert_reservation(before, point, REQUEST, "RELEASED", 961)
+	# Any remaining active reservation must still belong to the same live request
+	# and started service. Terminal history is retained, never filtered away.
+	for reservation: Dictionary in point[&"reservations"]:
+		if reservation[&"state"] != "ACTIVE": continue
+		var request: Dictionary = point[&"requests"].get(reservation[&"requester_id"], {})
+		_expect(not request.is_empty() and request[&"reserved_inventory_ids"].has(reservation[&"reservation_id"]) and not service.is_empty() and service[&"inventory_reservations"].has(reservation[&"reservation_id"]), "no active reservation orphan")
+
+func _save_point(label: String, point: Dictionary) -> void:
+	_points[label] = _write(label, point)
+	var path: String = "user://saves/p04_s3/" + _case + "-" + label
+	var saved: Dictionary = SaveService.save_snapshot(path, point[&"snapshot"])
+	var inspected: Dictionary = SaveService.inspect_snapshot(path)
+	_expect(bool(saved.get(&"ok", false)) and bool(inspected.get(&"ok", false)) and bool(inspected.get(&"read_only", false)), "public formal save/inspect")
+	_expect(var_to_bytes(inspected.get(&"snapshot", {})) == var_to_bytes(point[&"snapshot"]), "all typed fields preserved by formal package")
+	_expect(var_to_bytes(_runtime.create_snapshot()) == var_to_bytes(point[&"snapshot"]), "formal saving leaves Runtime unchanged")
+	_slots[_case + "-" + label] = ProjectSettings.globalize_path(path)
+	# Snapshot bytes exist once in point raw and once in the required formal
+	# package; indexes do not embed inspection/load copies of those bytes.
+
+func _write(label: String, value: Variant, global_name: bool = false) -> Dictionary:
+	var name: String = label if global_name else _mode + "-" + _case + "-" + label
+	var reference: Dictionary = {}
+	for extension: String in ["var", "json"]:
+		var path: String = _evidence_dir.path_join(name + "." + extension)
+		_expect(not FileAccess.file_exists(path), "raw point written exactly once " + name)
+		var file: FileAccess = FileAccess.open(path, FileAccess.WRITE)
+		_expect(file != null, "raw file " + name)
+		if file == null: continue
+		if extension == "var": file.store_buffer(var_to_bytes(value))
+		else: file.store_string(JSON.stringify(value, "\t"))
+		file.flush()
+		file.close()
+		reference[extension] = {&"path": path, &"sha256": FileAccess.get_sha256(path)}
+	return reference
+
+func _read(reference: Dictionary) -> Dictionary:
+	var item: Dictionary = reference["var"]
+	var path: String = item["path"]
+	_expect(FileAccess.get_sha256(path) == item["sha256"], "raw typed SHA")
+	var json_ref: Dictionary = reference["json"]
+	_expect(FileAccess.get_sha256(json_ref["path"]) == json_ref["sha256"], "raw JSON SHA")
+	var file: FileAccess = FileAccess.open(path, FileAccess.READ)
+	_expect(file != null, "typed input readable")
+	if file == null: return {}
+	var value: Variant = bytes_to_var(file.get_buffer(file.get_length()))
+	file.close()
+	_expect(value is Dictionary, "typed input dictionary")
+	return value if value is Dictionary else {}
+
+func _restore_inputs(point: Dictionary) -> void:
+	var inputs: Dictionary = point[&"inputs"]
+	_expect(_serial == 0, "Reader starts with serial zero before importing external inputs")
+	_serial = inputs[&"serial"]
+	_facility_id = inputs[&"facility_id"]
+	_service_id = inputs[&"service_id"]
+	_boundary = inputs[&"boundary"]
+	_price = inputs[&"price"]
+	_duration = inputs[&"duration"]
+	_patience = inputs[&"patience"]
+	_reserve = inputs[&"reserve"]
+	_recovery = inputs[&"recovery"].duplicate(true)
+
+func _load_point(path: String, expected: Dictionary) -> bool:
+	_fresh()
+	_restore_inputs(expected)
+	var loaded: Dictionary = SaveService.load_snapshot(path)
+	_expect(bool(loaded.get(&"ok", false)), "public formal load")
+	if not bool(loaded.get(&"ok", false)): return false
+	_expect(var_to_bytes(loaded[&"snapshot"]) == var_to_bytes(expected[&"snapshot"]), "all saved typed fields equal Writer")
+	var restored: bool = _runtime.restore_snapshot(loaded[&"snapshot"])
+	_expect(restored, "public Runtime restore")
+	_expect(var_to_bytes(_runtime.create_snapshot()) == var_to_bytes(expected[&"snapshot"]), "all restored typed fields equal Writer")
+	return restored
+
+func _reader() -> void:
+	var file: FileAccess = FileAccess.open(_evidence_dir.path_join("writer.json"), FileAccess.READ)
+	_expect(file != null, "Writer receipt")
+	if file == null: return
+	var writer: Dictionary = JSON.parse_string(file.get_as_text())
+	file.close()
+	_expect(int(writer["pid"]) != OS.get_process_id() and writer["profile"] != OS.get_user_data_dir(), "independent Reader PID/profile")
+	for branch: String in CASES:
+		_case = branch
+		var writer_case: Dictionary = _read(writer["branches"][branch])
+		var hashes: Dictionary = {}
+		var restored_refs: Dictionary = {}
+		for label: String in _labels():
+			var key: String = branch + "-" + label
+			var original: String = writer["slots"][key]
+			var target: String = ProjectSettings.globalize_path("user://saves/p04_s3/" + key)
+			_expect(DirAccess.make_dir_recursive_absolute(target) == OK, "private Reader formal slot")
+			var directory: DirAccess = DirAccess.open(original)
+			_expect(directory != null, "Writer formal slot readable")
+			if directory == null: return
+			var members: PackedStringArray = directory.get_files()
+			members.sort()
+			_expect(directory.get_directories().is_empty() and members == PackedStringArray(["journal.log", "manifest.json", "thumbnail.png", "world.bin"]), "all formal members enumerated")
+			var member_hashes: Dictionary = {}
+			for member: String in members:
+				var sha: String = FileAccess.get_sha256(original.path_join(member))
+				_expect(DirAccess.copy_absolute(original.path_join(member), target.path_join(member)) == OK, "copy every formal member")
+				var copied_sha: String = FileAccess.get_sha256(target.path_join(member))
+				_expect(not sha.is_empty() and sha == copied_sha, "original/copy member SHA equality")
+				member_hashes[member] = {&"original_before": sha, &"copy_before": copied_sha}
+			hashes[key] = {&"original": original, &"copy": target, &"members": member_hashes}
+			_slots[key] = target
+			var expected: Dictionary = _read(writer_case[&"points"][label])
+			if not _load_point(target, expected): return
+			restored_refs[label] = _write(label + "-restored", _capture())
+		var before: Dictionary = _read(writer_case[&"points"]["before"])
+		if not _load_point(_slots[branch + "-before"], before): return
+		_assert_before(_capture())
+		_connect_observations()
+		var replay_refs: Dictionary = {}
+		for label: String in _labels():
+			if label == "before": continue
+			_advance(label)
+			var actual: Dictionary = _capture()
+			var expected: Dictionary = _read(writer_case[&"points"][label])
+			_expect(var_to_bytes(actual) == var_to_bytes(expected), "replay complete typed point including Runtime, inputs, receipts " + label)
+			_assert_outcome(before, actual)
+			replay_refs[label] = _write(label + "-replay", actual)
+		var expected_phases: Array = writer_case[&"phases"]
+		_expect(_phases.size() == expected_phases.size(), "complete phase count")
+		if _phases.size() == expected_phases.size():
+			for index: int in range(_phases.size()):
+				_expect(var_to_bytes(_read(_phases[index])) == var_to_bytes(_read(expected_phases[index])), "same phase order and all typed phase state")
+		for key: String in hashes:
+			var row: Dictionary = hashes[key]
+			for member: String in row[&"members"]:
+				var item: Dictionary = row[&"members"][member]
+				item[&"original_after"] = FileAccess.get_sha256(String(row[&"original"]).path_join(member))
+				item[&"copy_after"] = FileAccess.get_sha256(String(row[&"copy"]).path_join(member))
+				_expect(item[&"original_before"] == item[&"original_after"] and item[&"copy_before"] == item[&"copy_after"], "original/copy unchanged after all loads and replay")
+		_branches[branch] = _write("case", {&"restored": restored_refs, &"replay": replay_refs, &"phases": _phases,
+			&"member_hashes": hashes, &"recovery_commands": _commands, &"replay_completed_commands": _results})

docs/plans/original-completion-2026-10-02/execution/p04-01/s3/run_s3_native.py

@@ -0,0 +1,293 @@
+#!/usr/bin/env python3
+"""S3-only native entry. --check-only never loads ownership helpers or starts Godot."""
+from pathlib import Path
+import argparse
+import hashlib
+import importlib.util
+import json
+import os
+import shutil
+import subprocess
+import sys
+sys.dont_write_bytecode = True
+
+HERE = Path(__file__).resolve().parent
+BUILD = Path("/Users/korizatou114/Projects/TheTree-managed-worktrees/p04-01/.runtime/P04-01/build")
+PRODUCT = Path("/Users/korizatou114/Projects/とりにく工房ゲーム制作部")
+CACHE_RUN = Path("/Users/korizatou114/Projects/TheTree-managed-worktrees/p18-01-model/.runtime/P18-01-CODE/build/p18-relationship-evidence")
+TEST = "tests/s3/test_runtime_staff_assignment.gd"
+QA = HERE / "p04_s3_closure_probe.gd"
+ENGINE = Path("/Applications/Godot.app/Contents/MacOS/Godot")
+MANIFEST = BUILD / "audit-support/p04-s3-existing-source-r01/SOURCE_MANIFEST.json"
+MANIFEST_SHA = "d4993398f7619ca37bcd64c32b79bdee892d44e31f332cc0dd5f31fb7bc28314"
+AUDIT = BUILD / "audit-support/p04-whole-closure-r01/ROOT_S2B_EXISTING_SOURCE_AUDIT.json"
+CASES = ["B1", "B2a", "B2b", "B3", "B4a", "B4b"]
+LABELS = {"B1": ["before", "after", "next", "recovered", "duplicate_guard"],
+          "B2a": ["before", "after", "next"], "B2b": ["before", "after", "next", "despawn"],
+          "B3": ["before", "after", "next", "despawn"],
+          "B4a": ["before", "after", "next"], "B4b": ["before", "after", "next"]}
+SLOTS = sorted(case + "-" + label for case in CASES for label in LABELS[case])
+MATERIALS = [MANIFEST, AUDIT, HERE.parent / "s2b/run_s2b_native.py",
+             HERE.parent / "s2b/p04_s2b_shift_probe.gd", HERE.parent / "s2b/S2B_ROOT_RUN.md"]
+
+def template_guard(root):
+    if pin(MANIFEST)["sha256"] != MANIFEST_SHA:
+        raise ValueError("supplied seven-source manifest changed")
+    manifest = json.loads(MANIFEST.read_text())
+    if len(manifest["files"]) != 7:
+        raise ValueError("exact seven source files required")
+    for row in manifest["files"]:
+        actual = pin(row["supplied"])
+        if (actual["bytes"], actual["sha256"]) != (row["bytes"], row["sha256"]):
+            raise ValueError("supplied template SHA mismatch: " + row["supplied"])
+    audit = json.loads(AUDIT.read_text())
+    for row in audit["source_pins"]:
+        actual = pin(root / row["path"])
+        if (actual["bytes"], actual["sha256"]) != (row["production_bytes"], row["production_sha256"]):
+            raise ValueError("Root source audit mismatch: " + row["path"])
+    probe = QA.read_text()
+    if ":=" in probe:
+        raise ValueError("probe requires explicit typed declarations")
+    import re
+    if re.search(r"\bvar\s+\w+\s*=", probe) or re.search(r"\bfor\s+\w+\s+in\b", probe):
+        raise ValueError("untyped declaration/loop")
+    if re.search(r"\.(?:patience_remaining|elapsed_minutes)\s*=(?!=)", probe):
+        raise ValueError("manual patience/progress mutation")
+    for forbidden in ["._update_shift_states(", ".record_duty_minute(", "get_day_summary(",
+                      "configure_operations("]:
+        if forbidden in probe:
+            raise ValueError("out-of-scope mutation/getter: " + forbidden)
+
+
+def module(name, path):
+    spec = importlib.util.spec_from_file_location(name, path)
+    value = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(value)
+    return value
+
+def pin(path):
+    path = Path(path)
+    if path.is_symlink() or not path.is_file():
+        raise ValueError("missing exact input: " + str(path))
+    data = path.read_bytes()
+    return dict(path=str(path), bytes=len(data), sha256=hashlib.sha256(data).hexdigest())
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--project-root", type=Path, default=PRODUCT)
+    parser.add_argument("--output", type=Path, default=BUILD / "p04-s3-native-r01")
+    parser.add_argument("--plan", type=Path, default=BUILD / "p04-s3-plan-r01.json")
+    parser.add_argument("--cache-run", type=Path, default=CACHE_RUN)
+    parser.add_argument("--godot", type=Path, default=ENGINE)
+    parser.add_argument("--check-only", action="store_true")
+    args = parser.parse_args()
+    root = args.project_root.resolve()
+    out = args.output.resolve()
+    if not out.is_relative_to(BUILD.resolve()) or not args.plan.resolve().is_relative_to(BUILD.resolve()):
+        raise ValueError("plan/output must stay in task build")
+    r = module("s3_standard_runner", root / "tools/run_all_tests.py")
+    # Preserve the reviewed source_refs -> exactly one QA override order.
+    source_before = r.source_refs(root)
+    refs = dict(source_before)
+    refs[TEST] = pin(QA)
+    template_guard(root)
+    expected = {n: dict(bytes=p["bytes"], sha256=p["sha256"]) for n, p in refs.items()}
+    fixed = [pin(p) for p in MATERIALS + [Path(row["supplied"]) for row in json.loads(MANIFEST.read_text())["files"]] + [
+        QA, Path(__file__), HERE / "S3_ROOT_RUN.md",
+        root / "tests/transport/test_runtime_transport.gd",
+        root / "tests/staff/test_rescue_elevator.gd",
+        root / "docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json",
+    ]]
+    selected = dict(root_before=source_before, map={n: p["sha256"] for n, p in refs.items()}, timeout=1200)
+    cache_note = None
+    cache_receipts = []
+    for leaf in ["runner-ended.json", "runner-input.json"]:
+        path = args.cache_run / leaf
+        try:
+            cache_receipts.append(pin(path))
+        except OSError as caught:
+            cache_note = dict(path=str(path), error=type(caught).__name__ + ": " + str(caught),
+                              required_before_native=True)
+            break
+    plan = dict(protocol="p04-s3-preparation-v1", project_root=str(root), output=str(out),
+                source_before=source_before, selected_source_map=selected["map"],
+                QA_overrides=[TEST], fixed_input_pins=fixed, cache_run=str(args.cache_run),
+                cache_receipts=cache_receipts, cache_access=cache_note,
+                phases=["import", "parse", "writer", "reader"],
+                native_engine_launches=0, whole_goal_accepted=False, unit_accepted=False)
+    # Check-only only reads source/materials and writes the plan in task build.
+    # It deliberately does not read default-player data, inspect processes or stat Godot.
+    if args.check_only:
+        args.plan.parent.mkdir(parents=True, exist_ok=True)
+        if args.plan.exists():
+            r.require(r.read(args.plan) == plan, "existing plan differs; use a fresh plan path")
+        else:
+            r.write(args.plan, plan)
+        print(json.dumps(dict(check_only=True, engines=0, source_files=len(source_before),
+                              QA_overrides=[TEST], fixed_pins=len(fixed), plan=str(args.plan),
+                              plan_sha256=r.sha(args.plan), cache_access=cache_note), ensure_ascii=False))
+        return 0
+    r.require(root == PRODUCT.resolve() and len(source_before) == 1338, "Root native requires product 1338; worker plan is not native evidence")
+    r.require(args.plan.is_file() and r.read(args.plan) == plan, "run exact check-only plan first")
+    def fixed_guard():
+        template_guard(root)
+        for value in fixed + cache_receipts:
+            r.require(pin(value["path"]) == value, "fixed input changed: " + value["path"])
+    fixed_guard()
+    # Native admission is Root-only and fails closed on unreadable cache.
+    r.require(cache_note is None, "Root must read exact cache receipts: " + str(cache_note))
+    old = args.cache_run
+    r.require(r.read(old / "runner-ended.json")["passed"], "ended cache run must have passed")
+    old_map = r.read(old / "runner-input.json")["selected_source_map"]
+    current_map = {n: p["sha256"] for n, p in source_before.items()}
+    r.require(old_map == current_map, "ended cache selected source must equal current product")
+    protected = r.default_player(root)
+    outer = module("s3_owned_outer", r.SUPPORT / "owned_outer.py")
+    r.require(not outer.native.helper["live_engines"](), "another Godot owns CPU")
+    engine = args.godot.resolve()
+    r.require(engine.is_file(), "Root native Godot path")
+    r.require(shutil.disk_usage(out.parent).free >= 1024**3 + engine.stat().st_size + 136*1024**2,
+              "reviewed native disk admission")
+    r.require(not out.exists(), "fresh output required; preserve previous attempts")
+    out.mkdir()
+    stage = out / "stage"
+    stage.mkdir()
+    (out / "starters").mkdir()
+    (out / "bootstrap").mkdir()
+    phases = []
+    previous = {}
+    error = None
+    guards = []
+    cleanup = []
+    cleanup_errors = []
+    cache_before = None
+    formal_before = None
+    formal_after = None
+    original_phase = outer.run_phase
+    def run_phase(argv, folder, timeout, env, **kwargs):
+        env = dict(env)
+        if folder.name in ["writer", "reader"]:
+            env.update(THE_TREE_S3_EVIDENCE=str(out / "raw"), THE_TREE_S3_MODE=folder.name)
+        return original_phase(argv, folder, timeout, env, **kwargs)
+    outer.run_phase = run_phase
+    r.write(out / "runner-started.json", dict(started=r.now(), owner_runner_pid=os.getpid(),
+            plan=pin(args.plan), engines_launched=0))
+    try:
+        r.write(out / "source_before.json", source_before)
+        r.write(out / "selected-source-map.json", refs)
+        # Private source stage first, then literal reviewed bootstrap.
+        for name, value in refs.items():
+            target = stage / name
+            target.parent.mkdir(parents=True, exist_ok=True)
+            r.copy_source(Path(value["path"]), target)
+        r.require(r.inventory(stage, stage=True) == expected, "whole private source")
+        for name, value in r.read(r.SUPPORT / "bootstrap-scripts.json").items():
+            data = value["text"].encode()
+            r.require(hashlib.sha256(data).hexdigest() == value["sha256"], "bootstrap pin")
+            (out / "bootstrap" / name).write_bytes(data)
+        cache = old / "stage/.godot"
+        cache_before = r.inventory(cache)
+        r.require(bool(cache_before), "nonempty ended cache")
+        for name in cache_before:
+            src = cache / name
+            dst = stage / ".godot" / name
+            dst.parent.mkdir(parents=True, exist_ok=True)
+            subprocess.run(["/bin/cp", "-c", str(src), str(dst)], check=True)
+            r.require((src.stat().st_dev, src.stat().st_ino) != (dst.stat().st_dev, dst.stat().st_ino),
+                      "cache clone is a separate inode")
+        r.require(r.inventory(stage / ".godot") == cache_before, "whole cache clone hashes")
+        r.write(out / "cache-reuse.json", dict(ended_run=str(old), receipts=cache_receipts,
+                whole_cache=cache_before, independent_inode=True,
+                ordinary_import_required=True, ordinary_parse_required=True))
+        paths = ["res://" + n for n in sorted(refs) if n.endswith(".gd")]
+        r.write(out / "parse-paths.json", paths)
+        markers = module("s3_markers", root / "tools/strict_test_markers.py")
+        contracts = markers.load_contracts(root / "docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json")
+        contracts[TEST] = dict(test=TEST, template="S3 closure: <checks> checks, <failures> failures", required=True)
+        policy = r.read(r.SUPPORT / "suite-policy.json")
+        r.require(TEST not in policy["suite_policies"], "no normal suite watchdog/policy override")
+        r.write(out / "contracts.json", contracts)
+        r.write(out / "runner-input.json", dict(plan, owner_runner_pid=os.getpid(),
+                wrapper_sha256=r.sha(__file__), QA=pin(QA), runner_sha256=r.sha(root / "tools/run_all_tests.py"),
+                default_player_before=protected, fixed_input_pins=fixed,
+                functional_watchdog_seconds=1200, source_adoption=False))
+        diagnostics = module("s3_diagnostics", r.SUPPORT / "diagnostic_gate.py")
+        owners = module("s3_owners", r.SUPPORT / "ordinary_proof_gate.py")
+        audio = module("s3_audio", r.SUPPORT / "audio_evidence_gate.py")
+        public = (stage / "project.godot").read_bytes()
+        for phase, test in [("import", None), ("parse", None), ("writer", TEST), ("reader", TEST)]:
+            if phase == "reader":
+                writer = r.read(out / "raw/writer.json")
+                formal_before = {n: r.inventory(Path(p)) for n, p in writer["slots"].items()}
+                r.require(sorted(formal_before) == SLOTS, "all 22 formal points for six fresh cases")
+            row = r.execute_phase(str(engine), stage, out, phase, test, selected, expected,
+                                  previous, policy, contracts, paths, public, protected, root,
+                                  outer, markers, diagnostics, owners, audio)
+            phases.append(row)
+            # execute_phase uses the same test log name; retain each phase's exact log too.
+            if test and row.get("log"):
+                (out / (phase + "-combined-raw.log")).write_bytes((out / row["log"]).read_bytes())
+                row["strict_runner_log_path"] = row["log"]
+                row["log"] = phase + "-combined-raw.log"
+            r.write(out / (phase + "-actual-phase.json"), row)
+            print(json.dumps(dict(phase=phase, passed=row["passed"], error=row.get("validation_error"))), flush=True)
+            r.require(row["passed"], "first strict phase failure: " + phase)
+            if phase in ["writer", "reader"]:
+                receipt = r.read(out / ("raw/" + phase + ".json"))
+                r.require(receipt["checks"] > 0 and receipt["failures"] == 0,
+                          "S3 actual outcome receipt")
+                r.require(sorted(receipt["branches"]) == CASES,
+                          "exact six independent Writer/Reader cases")
+                r.require(sorted(receipt["slots"]) == SLOTS, "all formal point slots")
+                r.require(receipt["pid"] == row["actual_pid"], "actual native PID/receipt")
+        writer = r.read(out / "raw/writer.json")
+        reader = r.read(out / "raw/reader.json")
+        r.require(writer["pid"] != reader["pid"] and writer["profile"] != reader["profile"], "distinct formal processes")
+        formal_after = {n: r.inventory(Path(p)) for n, p in writer["slots"].items()}
+        r.require(formal_before == formal_after, "Reader leaves Writer packages unchanged")
+    except BaseException as caught:
+        error = type(caught).__name__ + ": " + str(caught)
+    finally:
+        cleanup, cleanup_errors = outer.finish_captured(outer.OWNED)
+        try:
+            r.write(out / "outer-owned-finally.json", dict(cleanup=cleanup, errors=cleanup_errors))
+        except BaseException as caught:
+            guards.append("finally receipt: " + type(caught).__name__ + ": " + str(caught))
+    # Fixed guard always follows ownership cleanup, even if setup or an engine failed.
+    try:
+        fixed_guard()
+        after = r.source_refs(root)
+        r.write(out / "source_after.json", after)
+        r.require(after == source_before, "whole product source unchanged")
+        r.require(r.default_player(root) == protected, "default player unchanged")
+        if (stage / "project.godot").exists():
+            r.require(r.inventory(stage, stage=True) == dict(expected, **previous), "whole private source unchanged")
+        if cache_before is not None:
+            r.require(r.inventory(old / "stage/.godot") == cache_before, "original cache unchanged")
+        if formal_before is not None:
+            writer = r.read(out / "raw/writer.json")
+            formal_after = {n: r.inventory(Path(p)) for n, p in writer["slots"].items()}
+            r.require(formal_before == formal_after, "Writer formal packages unchanged after Reader, including failure")
+    except BaseException as caught:
+        guards.append(type(caught).__name__ + ": " + str(caught))
+    pids = [p.pid for p in outer.OWNED]
+    passed = (error is None and not guards and not cleanup_errors and len(phases) == 4
+              and all(p["passed"] for p in phases) and len(pids) == len(set(pids)) == 4
+              and all(p["engine_ended"] for p in cleanup))
+    execution = dict(passed=passed, error=error, guard_errors=guards, phases=phases,
+                     actual_captured_PIDs=pids, cleanup=cleanup, cleanup_errors=cleanup_errors,
+                     formal_before=formal_before, formal_after=formal_after,
+                     whole_goal_accepted=False, unit_accepted=False)
+    r.write(out / "raw-inventory.json", r.inventory(out / "raw"))
+    r.write(out / "PID-set.json", dict(owner_runner_pid=os.getpid(), engine_PIDs=pids))
+    r.write(out / "execution.json", execution)
+    r.write(out / "runner-ended.json", dict(execution, ended=r.now()))
+    print(json.dumps(dict(passed=passed, error=error, guard_errors=guards, output=str(out))))
+    return 0 if passed else 1
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except Exception as caught:
+        print(type(caught).__name__ + ": " + str(caught), file=sys.stderr)
+        raise SystemExit(2)