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

docs/plans/original-completion-2026-10-02/execution/p04-01/s2b/S2B_ROOT_RUN.md

@@ -0,0 +1,98 @@
+# P04-01 S2b code preparation
+
+変更は本ディレクトリの `p04_s2b_shift_probe.gd`、`run_s2b_native.py`、本書の3ファイルのみ。
+S1 r02、S2a native r03（ROOT_S2A_NATIVE_R03_RESULT.json と build/p04-s2a-native-r03/）は保持し、再実行しない。
+原15条件＋EV、7領域、3安全結果、原30/88、部分点なしを維持する。本提出による完成加点は0。
+原P04全体の達成、D05、独立レビュー、統合は確定しない。
+
+## 構成と固定入力
+
+QA overrideは `tests/s3/test_runtime_staff_assignment.gd` の1件だけ。
+単一GDがWriter内で2回fresh Runtime/Clock/GameStateを生成する。Readerも保存点ごと、replay開始時にfresh生成する。
+型推論によるVariant受取を避け、GDの変数宣言は明示型。state snapshotはString定義と比較する。
+
+- night_absent: supplied `approved-shift-current-r01.gd` のseed420410、2 DAY servers、実通勤、元のcafeteria/kitchen fixtureを使用。
+  queue_atだけ `boundary + 2 - base_service_minutes - 1` にし、予定完了を1082へ移す。
+  service duration/priceは定義値、Guest planned_exitは元の `boundary + base_service_minutes` のまま。
+  kitchenの物理配置済FLEX cookは既存診断fixtureの前提であり、自然雇用の証拠と混同しない。
+- overtime: `tests/supply/test_supply_logistics_bridge.gd:40-79` のseed653322、warehouse/kitchen/staff_door施工、
+  catalogue Gina公開hire/DAY assignment、実通勤、購入8、公開request、実pickupから_loadedを得る。
+  synthetic dispatch rejection、expiry snapshot変更、fault injectionは採用しない。
+  実taskのinterrupt_level==0を確認し、元のtarget cellだけ一時blockして通常clockを進める。
+
+ROOT_S2B_EXISTING_SOURCE_AUDIT.jsonの8ファイルは毎回bytes/SHAを選択rootと照合する。
+供給18:00 GD、監査JSON、S1/S2a wrapperと関連文書/GD、本3ファイルをplanでpinする。
+inherit先hardening/kitchen fixtureを含む選択root全sourceもpinする。
+旧source checkout1300のcheck-onlyは製品1338の代替証拠にしない。
+
+## 境界判定
+
+night_absentでは1079を基準に1080/1081で未完了、wallet/player cash/ledger無変更、同じ非空予約IDとACTIVE予約内容、
+元requestの所有を検査する。DAYの退出とNIGHT登録0を確認し、1081でWAITING_FOR_STAFFを必須にする。
+退出予定式と元5分graceも検査する。完了を早めたり、元のduration、price、shift値を変更しない。
+
+overtimeでは1078→1079にworked+1/overtime+0、1079→1080と1080→1081はworked+0/overtime+1を検査する。
+各tickの総勤務差分は1。正常勤務分と残業分を両方+1と誤読しない。
+同じstaff/task/shift/request/cargo所有、8個のcargo、crosslinks、is_overtimeを実オブジェクト／snapshotで検査する。
+各1079/1080/1081のtick後に公開 `StaffService.advance_one_minute(current_minute)` を1回呼び、
+実戻り値MINUTE_ALREADY_APPLIEDと全Runtime bytes不変で二重勤務記録なしを確認する。
+private `_update_shift_states` と `record_duty_minute` は呼ばない。
+解除後は各tickを保存し、safe unloading→元task/request COMPLETED→OFF_DUTY、8個移送と所有解放を確認する。
+
+## Formal save / Reader
+
+両ケースの1079/1080/1081を公開SaveServiceで保存・inspectする（合計6slot）。
+Writer全Runtime snapshotとinspect snapshotのvar_to_bytes完全一致、保存時のlive Runtime不変を必須とする。
+Readerは別PID/profileで各slotのjournal.log/manifest.json/thumbnail.png/world.binを全コピーし、
+原本／コピーの全SHAと、load前後／replay後の不変を検査する。余分なmember/directoryは失敗。
+公開load/restore後はWriterの全Runtimeをvar_to_bytesで比較する。項目の削除、JSON正規化による一致化はしない。
+
+Readerはさらに各1079 slotをfresh loadして通常clockの2tickを再現し、1080/1081全Runtimeと勤務差分、
+同分guardを検査する。FACILITY_COMPLETION前後とAI_DECISIONS後のphase順と全snapshotもWriterに一致させる。
+overtimeはunblock後のCOMPLETED/OFF_DUTYまで全Runtime一致を追加確認する。
+raw Dictionary getterと保存用canonical Operationsは別々に保存する。getter joinだけStringName/Stringとkey順差を明示し、
+型・値・欠落・配列順を検査する。全Runtime typed条件は緩和しない。
+
+## Rootの正式実行
+
+Rootだけが製品1338でfresh plan/nativeを実行する。workerはGodot起動（parse/version含む）、Runner操作、全回帰、git書込みをしない。
+以下は本提出checkoutからRootが実行する手順。既存output/planを上書きせず、失敗時は最初のphaseで停止し自動再試行しない。
+
+```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/s2b/run_s2b_native.py --check-only --plan "$BUILD_OUTPUT_DIR/p04-s2b-root-plan-r01.json" --output "$BUILD_OUTPUT_DIR/p04-s2b-native-r01"
+python3 docs/plans/original-completion-2026-10-02/execution/p04-01/s2b/run_s2b_native.py --plan "$BUILD_OUTPUT_DIR/p04-s2b-root-plan-r01.json" --output "$BUILD_OUTPUT_DIR/p04-s2b-native-r01"
+```
+
+製品root既定値は `/Users/korizatou114/Projects/とりにく工房ゲーム制作部`。
+native admissionはこのrootかつsource1338を必須にする。worker checkout planをnativeへ転用しない。
+check-onlyはsource/material/cache receiptのみ読取り、engine0、default/private profile読取り0。
+cache receiptの読取拒否はplanに実エラーを残し、nativeは拒否する。Rootは自身の環境で確認する。
+
+S1/S2aと同じsource_refs→QA1→private source stage→終了済cacheの別inode clone→import→parse→Writer→Reader→
+owned finally cleanup→全guardを維持する。source/default player/private source/original cache不変、
+固定pin、実PIDとreceipt、4phase/4owned PIDの終了、正式slot不変は失敗時にも検査する。
+通常policy、watchdog、strict marker/diagnostic/ownership/audio gateを変更しない。
+
+## 必須出力と残件
+
+rawにはseed/input、実command結果（継承fixture分もcommand_completedで採取）、minute、実staff/task/shift/request/service ID、
+reservation/cargo、全ledger、全Runtime、phase観測、duty差分、同分guard結果、save/load結果をJSONと型保持.varの対で残す。
+Writerのloaded、1078-1079-duty、各1079/1080/1081とsave、phase、branch、unload、Readerの全restored/replayとSHA表を保持する。
+正式6slotはWriter/Reader private profileに保持する。
+source_before/after、selected-source-map、runner-input、cache-reuse、parse-paths、contracts、各phaseログ／receipt、
+combined-raw.log、PID-set、outer-owned-finally、execution、runner-ended、raw-inventoryも保持する。
+obsolete 500000-byte capに合わせた省略・圧縮はしない。
+
+worker検証範囲はpy_compile、wrapper --check-only、GDのVariant推論宣言grepのみ。
+GDScript parse、製品1338 nativeの条件成立、formal replay一致は未実行であり合格を主張しない。
+Root native実測と正式独立レビューが残る。これらは予定された後続作業であり、未実装を外部blockedとは報告しない。
+
+worker実施記録: Python py_compile成功。GD grepは `:=` と禁止された2直接呼出しが0件（rg exit1）。
+既定製品rootでのcheck-onlyは `tools/run_all_tests.py` 読取りがPermissionErrorとなりexit2。
+明示worktree rootでのcheck-onlyはexit0、source1300、QA1、固定pin13、engine0。
+終了済cacheのrunner-ended.jsonもPermissionErrorであり、planのcache_accessへ実path/errorを記録した。
+private/default profileの読取りは0。製品1338のcheck-only成功・native成功を主張しない。
+最初のworker planは `build/p04-s2b-worker-check-r01.json`。その後の本書／GD調整を含む最終pin確認には
+fresh `build/p04-s2b-worker-check-r02.json` を使用し、r01を上書きしない。

docs/plans/original-completion-2026-10-02/execution/p04-01/s2b/p04_s2b_shift_probe.gd

@@ -0,0 +1,522 @@
+extends "res://tests/s3/test_runtime_actor_hardening.gd"
+
+const REQUEST: StringName = &"supply.bridge_test"
+const PORTER: StringName = &"staff_gina"
+var _evidence_dir: String = ""
+var _mode: String = ""
+var _case: String = ""
+var _seed: int = 0
+var _serial: int = 0
+var _slots: Dictionary = {}
+var _branches: Dictionary = {}
+var _commands: Array = []
+var _command_results: Array = []
+var _phases: Array = []
+var _facility_id: StringName = &""
+var _service_id: StringName = &""
+var _request_id: StringName = &"p04.order.shift.request"
+var _guest_id: StringName = &"p04.order.shift.guest"
+var _staff_ids: Array[StringName] = [&"p04.order.server.a", &"p04.order.server.b"]
+var _source: StringName = &""
+var _destination: StringName = &""
+var _loaded: Dictionary = {}
+var _task_id: StringName = &""
+var _shift_id: StringName = &""
+var _target: Vector3i = Vector3i.ZERO
+
+func _run() -> void:
+	_evidence_dir = OS.get_environment("THE_TREE_S2B_EVIDENCE")
+	_mode = OS.get_environment("THE_TREE_S2B_MODE")
+	if _evidence_dir.is_empty() or _mode not in ["writer", "reader"]:
+		quit(2)
+		return
+	_expect(DirAccess.make_dir_recursive_absolute(_evidence_dir) == OK, "evidence directory")
+	if _mode == "writer":
+		for branch: String in ["night_absent", "overtime"]: _writer_branch(branch)
+	else:
+		_reader()
+	_write("branches", _branches)
+	var receipt: Dictionary = {&"pid": OS.get_process_id(), &"profile": OS.get_user_data_dir(),
+		&"slots": _slots, &"checks": _checks, &"failures": _failures, &"branches": _branches.keys(),
+		&"whole_goal_accepted": false, &"unit_accepted": false}
+	_write(_mode, receipt, true)
+	_cleanup()
+	print("S2b shift: %d checks, %d failures" % [_checks, _failures])
+	quit(1 if _failures else 0)
+
+func _fresh(seed: int) -> void:
+	_seed = seed
+	_serial = 0
+	_commands.clear()
+	_command_results.clear()
+	_phases.clear()
+	_facility_id = &""
+	_service_id = &""
+	_source = &""
+	_destination = &""
+	_task_id = &""
+	_shift_id = &""
+	_target = Vector3i.ZERO
+	_loaded = {}
+	_reset_runtime(seed)
+	_runtime.configure_save_policy({&"enabled": false, &"daily_checkpoint_enabled": false})
+	_runtime.command_completed.connect(func(result: CommandResult) -> void:
+		_command_results.append({&"minute": _clock.absolute_minute, &"result": result.to_dictionary()})
+	)
+
+func _cmd(kind: StringName, payload: Dictionary) -> CommandResult:
+	_serial += 1
+	var result: CommandResult = _runtime.submit_command(kind, payload, &"player", &"", 10, -1, StringName("p04.order.shift.command.%d" % _serial))
+	_commands.append({&"kind": kind, &"input": payload.duplicate(true), &"minute": _clock.absolute_minute, &"result": result.to_dictionary()})
+	return result
+
+func _command(kind: StringName, payload: Dictionary) -> bool:
+	# Supply fixture retains its original public submit defaults.
+	var result: CommandResult = _runtime.submit_command(kind, payload)
+	_commands.append({&"kind": kind, &"input": payload.duplicate(true), &"minute": _clock.absolute_minute, &"result": result.to_dictionary()})
+	return result.is_success()
+
+func _prepare_night() -> void:
+	_fresh(420410)
+	var facility: FacilityInstance = _build_cafeteria(true)
+	if facility == null: return
+	_facility_id = facility.instance_id
+	var definition: FacilityDefinition = _runtime.definition_registry.get_facility(facility.definition_id)
+	for id: StringName in _staff_ids:
+		_expect(_cmd(&"register_staff", {&"staff_id": id, &"role_id": &"server", &"options": {&"state": StaffRuntime.State.OFF_DUTY, &"current_cell": Vector3i.ZERO}}).is_success(), "register real Staff actor " + String(id))
+		_expect(_cmd(&"set_staff_work_assignment", {&"staff_id": id, &"template_id": &"DAY", &"facility_id": _facility_id}).is_success(), "original DAY 06:00-18:00 shift " + String(id))
+	# Publicly registered Staff actors walk normally. The inherited fixture's
+	# kitchen cook is explicitly placed at the kitchen and uses its existing FLEX
+	# shift; this is not natural hiring/unlock or a proof of the whole labor flow.
+	var boundary: int = ShiftInstance.DAY_END
+	var queue_at: int = boundary + 2 - definition.base_service_minutes - 1
+	_expect(queue_at > _clock.absolute_minute, "real ticks can reach exact admission minute")
+	_clock.advance_minutes(queue_at - _clock.absolute_minute)
+	_expect(facility.active_staff_ids == _staff_ids, "both daytime actors are physically projected before service")
+	for id: StringName in _staff_ids:
+		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, "actual workplace arrival and attendance plan " + String(id))
+	_expect(_stock(facility.local_inventory_id, &"ingredient/basic_meal", 1, &"p04.order.shift.stock"), "one actual meal to reserve")
+	_expect(_cmd(&"spawn_guest", {&"guest_id": _guest_id, &"guest_type_id": &"traveler", &"gold": definition.base_price * 2, &"options": {&"state": GuestRuntime.State.DECIDING, &"current_cell": Vector3i.ZERO, &"planned_exit_at": boundary + definition.base_service_minutes}}).is_success(), "real public Guest actor and wallet")
+	_expect(_cmd(&"create_service_request", {&"facility_id": _facility_id, &"request_id": _request_id, &"guest_ids": [_guest_id]}).is_success(), "original price and duration service request")
+	_clock.advance_one_minute()
+	var active: Array = _runtime.facility_service.get_active_services(_facility_id)
+	_expect(active.size() == 1, "exactly one actual service starts")
+	if active.size() != 1: return
+	var service: ServiceInstance = active[0]; _service_id = service.service_id
+	_expect(service.duration_minutes == definition.base_service_minutes and service.price_snapshot == definition.base_price, "no price or duration override")
+	_expect(service.started_at + service.duration_minutes == boundary + 2, "unchanged duration with admission shifted two minutes")
+	_clock.advance_minutes(boundary - 1 - _clock.absolute_minute)
+	_expect(service.elapsed_minutes == service.duration_minutes - 3, "three real service minutes remain at 17:59")
+
+func _prepare_overtime() -> void:
+	_fresh(653322)
+	for row: Array in [[&"warehouse", Vector3i(-8, 0, 1)], [&"kitchen", Vector3i(2, 0, 1)], [&"staff_door", Vector3i(0, 0, 1)]]:
+		_expect(_command(&"build_facility", {&"definition_id": row[0], &"origin": row[1]}), "build %s" % row[0])
+	_clock.advance_minutes(180)
+	for facility: FacilityInstance in _runtime.facility_service.get_instances():
+		if facility.definition_id == &"warehouse": _source = facility.instance_id
+		if facility.definition_id == &"kitchen": _destination = facility.instance_id
+	if _source.is_empty() or _destination.is_empty():
+		_expect(false, "normal construction completes")
+		return
+	_expect(_command(&"hire_staff_candidate", {&"candidate_id": PORTER}), "hire the catalogue porter")
+	_expect(_command(&"set_staff_work_assignment", {&"staff_id": PORTER, &"facility_id": _source, &"template_id": &"DAY"}), "assign porter to a repeating real shift")
+	_expect(_worker().state == StaffRuntime.State.COMMUTING, "porter assignment begins a real warehouse commute")
+	for minute in range(40):
+		if _worker().state == StaffRuntime.State.AVAILABLE: break
+		_clock.advance_minutes(1)
+	_expect(_worker().state == StaffRuntime.State.AVAILABLE and StaffCommuteService.is_at_facility(_worker(), _runtime.facility_service.get_instance(_source)), "porter checks in physically before taking a supply job")
+	_expect(_command(&"purchase_supplies", {&"warehouse_id": _source, &"item_id": SupplyCatalog.RAW_FOOD, &"quantity": 8}), "purchase the source stock through the ledger")
+	_expect(_command(&"create_logistics_request", {&"request_id": REQUEST, &"source_inventory_id": _inventory(_source), &"destination_inventory_id": _inventory(_destination), &"item_requirements": {SupplyCatalog.RAW_FOOD: 8}}), "create the reserved supply request")
+	_expect(_quantity(_inventory(_destination)) == 0, "request creation does not deliver anything")
+	var check_in_cell: Vector3i = _worker().current_cell
+	_clock.advance_minutes(1)
+	_expect(_request().assigned_carrier_id == PORTER and _worker().current_cell.distance_to(check_in_cell) <= 1.0, "dispatcher selects the checked-in employee without teleporting")
+	_wait_for(LogisticsRequest.State.DELIVERING, 40)
+	_expect(_request().state == LogisticsRequest.State.DELIVERING and _worker().current_cell.distance_to(check_in_cell) <= 1.0, "physical pickup starts delivery from the checked-in warehouse access without teleporting")
+	if _request().state != LogisticsRequest.State.DELIVERING:
+		return
+	_expect(_quantity(_inventory(_source)) == 0 and _quantity(_cargo()) == 8 and _quantity(_inventory(_destination)) == 0, "pickup moves all eight units into owned cargo only")
+	_expect(_crosslinks(), "loaded Request, Task, reservations and cargo agree")
+	_loaded = _runtime.create_snapshot()
+
+	_write("loaded", {&"seed": _seed, &"snapshot": _loaded, &"commands": _commands, &"command_results": _command_results})
+	_expect(_task().interrupt_level == 0, "actual loaded task is non-interruptible")
+	_task_id = _task().task_id
+	_shift_id = _worker().shift_id
+	_target = _task().target_cell
+	_runtime.path_service.set_cell_temporary_block(_target, true)
+	_clock.advance_minutes(1078 - _clock.absolute_minute)
+	var before: Dictionary = _capture()
+	_clock.advance_one_minute()
+	var after: Dictionary = _capture()
+	_write("1078-1079-duty", _duty_delta(before, after))
+	_expect(_clock.absolute_minute == 1079, "real loaded branch reaches pre-boundary")
+
+func _wait_for(state: int, minutes: int) -> void:
+	for index: int in minutes:
+		if _request().state == state: return
+		_clock.advance_minutes(1)
+
+
+func _inventory(id: StringName) -> StringName:
+	return _runtime.facility_service.get_instance(id).local_inventory_id
+
+
+func _request() -> LogisticsRequest:
+	return _runtime.logistics_requests[REQUEST]
+
+
+func _worker() -> StaffRuntime:
+	return _runtime.staff_service.get_staff(PORTER)
+
+
+func _task() -> StaffTask:
+	return _runtime.staff_service.get_task(_worker().current_task_id)
+
+
+func _cargo() -> StringName:
+	return StringName("cargo:%s" % PORTER)
+
+
+func _quantity(inventory_id: StringName) -> int:
+	return _runtime.inventory_service.total_quantity(inventory_id, SupplyCatalog.RAW_FOOD)
+
+
+func _crosslinks() -> bool:
+	return SupplyLogisticsBridge.validate_crosslinks(_runtime.logistics_requests, _runtime.staff_service, _runtime.inventory_service, _runtime.facility_service)
+
+func _writer_branch(branch: String) -> void:
+	_case = branch
+	if branch == "night_absent": _prepare_night()
+	else: _prepare_overtime()
+	if _clock.absolute_minute != 1079:
+		_expect(false, "setup must reach exact pre-boundary minute")
+		return
+	_connect_phases()
+	var before: Dictionary = _capture()
+	if branch == "overtime": _guard_same_minute()
+	_save_point("1079", before)
+	var previous: Dictionary = before
+	var points: Dictionary = {"1079": before}
+	for minute: int in [1080, 1081]:
+		_clock.advance_one_minute()
+		var actual: Dictionary = _capture()
+		_expect(actual[&"minute"] == minute, "normal clock advances one boundary tick")
+		if branch == "overtime":
+			_write(str(minute) + "-duty", _duty_delta(previous, actual))
+			_guard_same_minute()
+		else:
+			_assert_night(before, actual)
+		_save_point(str(minute), actual)
+		points[str(minute)] = actual
+		previous = actual
+	if branch == "overtime": points["unload"] = _unload()
+	points["phase_observations"] = _phases.duplicate(true)
+	points["commands"] = _commands.duplicate(true)
+	points["command_results_including_inherited_fixture"] = _command_results.duplicate(true)
+	_branches[branch] = points
+	_write("branch", points)
+
+func _capture() -> Dictionary:
+	var snapshot: Dictionary = _runtime.create_snapshot()
+	var raw_day: Dictionary = _runtime.operations_coordinator.get_day_summary(1)
+	var canonical_operations: Dictionary = _runtime.operations_coordinator.create_snapshot()
+	var canonical_day: Dictionary = {}
+	var day_count: int = 0
+	for day: Dictionary in canonical_operations["day_summaries"]:
+		if int(day["day"]) == 1:
+			canonical_day = day
+			day_count += 1
+	var getter_differences: Array = []
+	_expect(_getter_join(raw_day, canonical_day, "$", getter_differences) and day_count == (0 if raw_day.is_empty() else 1), "raw getter joins canonical save contract without relaxing Runtime equality")
+	var ledger: Array = []
+	for entry: LedgerEntry in _runtime.economy_service.get_entries(): ledger.append(entry.to_dictionary())
+	var row: Dictionary = {&"seed": _seed, &"minute": _clock.absolute_minute, &"snapshot": snapshot,
+		&"inputs": {&"case": _case, &"seed": _seed, &"facility_id": _facility_id, &"service_id": _service_id,
+			&"source": _source, &"destination": _destination, &"task_id": _task_id, &"shift_id": _shift_id,
+			&"target": _target, &"request_id": _request_id, &"guest_id": _guest_id,
+			&"boundary": ShiftInstance.DAY_END},
+		&"ledger": ledger, &"raw_day_getter": raw_day,
+		&"canonical_operations": canonical_operations, &"canonical_day1": canonical_day,
+		&"getter_differences": getter_differences, &"getter_bytes_equal": var_to_bytes(raw_day) == var_to_bytes(canonical_day),
+		&"commands": _commands.duplicate(true), &"command_results": _command_results.duplicate(true)}
+	# Raw getters retain their actual StringName keys and insertion order. They
+	# are diagnostics, never substituted for the entire canonical Runtime save.
+	if _case == "night_absent":
+		row[&"inputs"].merge({&"planned_completion": 1082, &"admission_offset": 2,
+			&"kitchen_premise": "original physically placed FLEX cook; not natural hire"})
+		var facility: FacilityInstance = _runtime.facility_service.get_instance(_facility_id)
+		var service: ServiceInstance = _runtime.facility_service.get_service(_service_id)
+		var guest: GuestRuntime = _runtime.guest_service.get_guest(_guest_id)
+		var workers: Array = []
+		var night_count: int = 0
+		var staff_snapshot: Dictionary = _runtime.staff_service.create_snapshot()
+		for shift: Dictionary in staff_snapshot[&"shifts"]:
+			if shift[&"template_id"] == "NIGHT" and not shift[&"cancelled"]: night_count += 1
+		for id: StringName in _staff_ids:
+			var worker: StaffRuntime = _runtime.staff_service.get_staff(id)
+			workers.append(worker.create_snapshot())
+		var reservations: Array = []
+		for id: StringName in service.inventory_reservations:
+			var reservation: InventoryReservation = _runtime.inventory_service.get_reservation(id)
+			_expect(reservation != null, "original service reservation exists")
+			if reservation != null: reservations.append(reservation.create_snapshot())
+		var request: ServiceRequest = _runtime.facility_service.get_request(_request_id)
+		row.merge({&"facility": facility.create_snapshot(), &"service": service.create_snapshot(),
+			&"guest": guest.create_snapshot(), &"workers": workers, &"night_staff_count": night_count,
+			&"reservation_ids": service.inventory_reservations.duplicate(), &"reservations": reservations,
+			&"request": request.create_snapshot() if request != null else {},
+			&"wallet": _runtime.economy_service.get_balance(StringName("GUEST_WALLET:" + String(_guest_id))),
+			&"player_cash": _runtime.economy_service.get_balance(_runtime.PLAYER_CASH_ACCOUNT)})
+	else:
+		row[&"inputs"].merge({&"supply_request_id": REQUEST, &"porter_id": PORTER, &"quantity": 8,
+			&"staff_premise": "public catalogue Gina hire and real DAY commute"})
+		var worker: StaffRuntime = _worker()
+		var task: StaffTask = _runtime.staff_service.get_task(_task_id)
+		var shift: ShiftInstance = _runtime.staff_service.get_shift(_shift_id)
+		row.merge({&"worker": worker.create_snapshot(), &"task": task.create_snapshot() if task != null else {},
+			&"shift": shift.create_snapshot() if shift != null else {}, &"request": _request().create_snapshot(),
+			&"cargo_id": _cargo(), &"cargo_quantity": _quantity(_cargo()),
+			&"source_quantity": _quantity(_inventory(_source)), &"destination_quantity": _quantity(_inventory(_destination)),
+			&"inventory": _runtime.inventory_service.create_snapshot(), &"crosslinks": _crosslinks(),
+			&"reserved_load": _runtime.logistics_dispatcher.get_reserved_load(PORTER)})
+	return row
+
+func _assert_night(before: Dictionary, actual: Dictionary) -> void:
+	var service: Dictionary = actual[&"service"]
+	_expect(service[&"state"] != "COMPLETED" and int(service[&"elapsed_minutes"]) < int(service[&"duration_minutes"]), "original service remains unfinished")
+	_expect(actual[&"night_staff_count"] == 0, "no NIGHT shift registered")
+	_expect(actual[&"wallet"] == before[&"wallet"] and actual[&"player_cash"] == before[&"player_cash"], "unfinished service charges neither wallet nor player")
+	_expect(var_to_bytes(actual[&"ledger"]) == var_to_bytes(before[&"ledger"]), "no ledger posting while unfinished at boundary")
+	_expect(var_to_bytes(actual[&"reservation_ids"]) == var_to_bytes(before[&"reservation_ids"]) and not actual[&"reservation_ids"].is_empty(), "same nonempty original reservation IDs")
+	_expect(var_to_bytes(actual[&"reservations"]) == var_to_bytes(before[&"reservations"]), "original active reservations unchanged")
+	_expect(not actual[&"request"].is_empty(), "original request retained")
+	for reservation: Dictionary in actual[&"reservations"]:
+		_expect(reservation[&"state"] == "ACTIVE" and reservation[&"requester_id"] == String(_request_id) and reservation[&"purpose"] == "facility_service", "reservation state and original request ownership")
+		_expect(actual[&"request"][&"reserved_inventory_ids"].has(reservation[&"reservation_id"]), "request still owns service reservation")
+	_expect(actual[&"facility"][&"active_staff_ids"].is_empty(), "DAY staff have left cafeteria")
+	for worker: Dictionary in actual[&"workers"]:
+		_expect(worker[&"state"] == "OFF_DUTY", "DAY worker is off duty")
+	_expect(int(actual[&"guest"][&"planned_exit_at"]) == 1080 + int(service[&"duration_minutes"]) and int(actual[&"guest"][&"planned_exit_at"]) > 1081, "original planned exit formula; not early exit")
+	if actual[&"minute"] == 1081:
+		_expect(service[&"state"] == "WAITING_FOR_STAFF", "night_staff_absent at 1081")
+		_expect(int(service[&"staff_grace_deadline"]) == int(service[&"interrupted_at"]) + FacilityService.STAFF_REPLACEMENT_GRACE_MINUTES, "original five-minute replacement grace")
+
+func _duty_delta(before: Dictionary, after: Dictionary) -> Dictionary:
+	var worked: int = int(after[&"shift"][&"worked_minutes"]) - int(before[&"shift"][&"worked_minutes"])
+	var overtime: int = int(after[&"shift"][&"overtime_minutes"]) - int(before[&"shift"][&"overtime_minutes"])
+	var minute: int = after[&"minute"]
+	var expected_overtime: int = 1 if minute >= ShiftInstance.DAY_END else 0
+	_expect(worked == 1 - expected_overtime and overtime == expected_overtime, "exactly one duty minute in original normal/overtime bucket")
+	_expect(after[&"worker"][&"is_overtime"] == (minute >= ShiftInstance.DAY_END), "actual overtime flag at each minute")
+	for point: Dictionary in [before, after]:
+		_expect(point[&"task"][&"interrupt_level"] == 0, "actual task interrupt_level zero")
+		_expect(point[&"worker"][&"current_task_id"] == String(_task_id) and point[&"task"][&"assigned_staff_id"] == String(PORTER), "same Staff/Task pair")
+		_expect(point[&"worker"][&"shift_id"] == String(_shift_id) and point[&"shift"][&"staff_id"] == String(PORTER), "same real ShiftInstance")
+		_expect(point[&"worker"][&"carried_inventory_id"] == String(_cargo()) and point[&"cargo_quantity"] == 8 and point[&"destination_quantity"] == 0 and point[&"source_quantity"] == 0, "same eight-unit cargo owner while held")
+		_expect(point[&"request"][&"assigned_carrier_id"] == String(PORTER) and point[&"request"][&"state"] == "DELIVERING" and point[&"crosslinks"], "loaded request crosslinks remain valid")
+	return {&"before": before, &"after": after, &"worked_delta": worked, &"overtime_delta": overtime,
+		&"duty_recorded_once": worked == 1 - expected_overtime and overtime == expected_overtime}
+
+func _guard_same_minute() -> void:
+	var before: Dictionary = _runtime.create_snapshot()
+	var minute: int = _clock.absolute_minute
+	var result: Dictionary = _runtime.staff_service.advance_one_minute(minute)
+	var after: Dictionary = _runtime.create_snapshot()
+	_expect(not bool(result.get(&"ok", true)) and result.get(&"reason") == &"MINUTE_ALREADY_APPLIED", "public same-minute duplicate guard")
+	_expect(var_to_bytes(before) == var_to_bytes(after), "same-minute guard records no second duty minute; all Runtime fields unchanged")
+	_write(str(minute) + "-guard", {&"minute": minute, &"result": result, &"before": before, &"after": after})
+
+func _unload() -> Dictionary:
+	_runtime.path_service.set_cell_temporary_block(_target, false)
+	var observations: Array = []
+	for index: int in range(40):
+		if _request().state == LogisticsRequest.State.COMPLETED: break
+		_clock.advance_one_minute()
+		observations.append(_capture())
+	var completed: Dictionary = _capture()
+	_expect(_request().state == LogisticsRequest.State.COMPLETED and completed[&"task"][&"state"] == "COMPLETED", "safe unloading completes original request and task")
+	_expect(_quantity(_inventory(_destination)) == 8 and _quantity(_cargo()) == 0 and _crosslinks(), "safe unloading conserves all eight units")
+	_expect(_worker().current_task_id.is_empty() and _worker().carried_inventory_id.is_empty() and _runtime.logistics_dispatcher.get_reserved_load(PORTER) == 0, "completion releases cargo/task/dispatcher ownership")
+	_clock.advance_one_minute()
+	var off: Dictionary = _capture()
+	_expect(_worker().state == StaffRuntime.State.OFF_DUTY and not _worker().is_overtime, "OFF_DUTY after safe completion")
+	var row: Dictionary = {&"observations": observations, &"completed": completed, &"off_duty": off}
+	_write("unload", row)
+	return row
+
+func _connect_phases() -> void:
+	_clock.register_tick_handler(_clock.TickPhase.FACILITY_COMPLETION, &"aa.p04.s2b.observe", func(minute: int) -> void:
+		if minute in [1080, 1081]: _phase("before-facility")
+	)
+	_clock.register_tick_handler(_clock.TickPhase.FACILITY_COMPLETION, &"zz.p04.s2b.observe", func(minute: int) -> void:
+		if minute in [1080, 1081]: _phase("after-facility")
+	)
+	_clock.register_tick_handler(_clock.TickPhase.AI_DECISIONS, &"zz.p04.s2b.observe", func(minute: int) -> void:
+		if minute in [1080, 1081]: _phase("after-staff")
+	)
+
+func _phase(label: String) -> void:
+	var row: Dictionary = {&"label": label, &"minute": _clock.absolute_minute, &"snapshot": _runtime.create_snapshot()}
+	_phases.append(row)
+	_write(str(_clock.absolute_minute) + "-" + label, row)
+
+func _save_point(label: String, point: Dictionary) -> void:
+	_write(label, point)
+	var path: String = "user://saves/p04_s2b/" + _case + "-" + label
+	var result: Dictionary = SaveService.save_snapshot(path, point[&"snapshot"])
+	var inspected: Dictionary = SaveService.inspect_snapshot(path)
+	_expect(bool(result.get(&"ok", false)) and bool(inspected.get(&"ok", false)) and bool(inspected.get(&"read_only", false)), "public formal save and inspect")
+	_expect(var_to_bytes(inspected.get(&"snapshot", {})) == var_to_bytes(point[&"snapshot"]), "formal save retains every typed Runtime field")
+	_expect(var_to_bytes(_runtime.create_snapshot()) == var_to_bytes(point[&"snapshot"]), "saving does not mutate live Runtime")
+	_slots[_case + "-" + label] = ProjectSettings.globalize_path(path)
+	_write(label + "-save", {&"save_result": result, &"inspect_result": inspected})
+
+func _write(label: String, value: Variant, global_name: bool = false) -> void:
+	var name: String = label if global_name else _mode + "-" + _case + "-" + label
+	var file: FileAccess = FileAccess.open(_evidence_dir.path_join(name + ".var"), FileAccess.WRITE)
+	_expect(file != null, "typed artifact " + name)
+	if file != null:
+		file.store_buffer(var_to_bytes(value)); file.flush(); file.close()
+	file = FileAccess.open(_evidence_dir.path_join(name + ".json"), FileAccess.WRITE)
+	_expect(file != null, "JSON artifact " + name)
+	if file != null:
+		file.store_string(JSON.stringify(value, "\t")); file.flush(); file.close()
+
+func _read_point(label: String) -> Dictionary:
+	var file: FileAccess = FileAccess.open(_evidence_dir.path_join("writer-" + _case + "-" + label + ".var"), FileAccess.READ)
+	_expect(file != null, "Writer typed artifact " + label)
+	if file == null: return {}
+	var value: Variant = bytes_to_var(file.get_buffer(file.get_length()))
+	file.close()
+	_expect(value is Dictionary, "typed dictionary artifact")
+	return value if value is Dictionary else {}
+
+func _restore_inputs(point: Dictionary) -> void:
+	var inputs: Dictionary = point[&"inputs"]
+	_seed = int(inputs[&"seed"])
+	_facility_id = StringName(inputs[&"facility_id"])
+	_service_id = StringName(inputs[&"service_id"])
+	_source = StringName(inputs[&"source"])
+	_destination = StringName(inputs[&"destination"])
+	_task_id = StringName(inputs[&"task_id"])
+	_shift_id = StringName(inputs[&"shift_id"])
+	_target = inputs[&"target"]
+
+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 String(writer["profile"]) != OS.get_user_data_dir(), "independent Reader PID/profile")
+	for branch: String in ["night_absent", "overtime"]:
+		_case = branch
+		var points: Dictionary = {}
+		var hashes: Dictionary = {}
+		for label: String in ["1079", "1080", "1081"]:
+			var key: String = branch + "-" + label
+			var original: String = writer["slots"][key]
+			var target: String = ProjectSettings.globalize_path("user://saves/p04_s2b/" + key)
+			_expect(DirAccess.make_dir_recursive_absolute(target) == OK, "fresh Reader slot")
+			var directory: DirAccess = DirAccess.open(original)
+			_expect(directory != null, "Writer slot accessible")
+			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"]), "exact four-member formal package")
+			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 formal member")
+				var copied_sha: String = FileAccess.get_sha256(target.path_join(member))
+				_expect(sha == copied_sha and not sha.is_empty(), "original/copy 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_point(label)
+			if expected.is_empty(): return
+			points[label] = expected
+			_fresh(int(expected[&"seed"]))
+			_restore_inputs(expected)
+			var loaded: Dictionary = SaveService.load_snapshot(target)
+			_expect(bool(loaded.get(&"ok", false)), "public formal load")
+			if not bool(loaded.get(&"ok", false)): return
+			_expect(var_to_bytes(loaded[&"snapshot"]) == var_to_bytes(expected[&"snapshot"]), "all typed saved fields match Writer")
+			_expect(_runtime.restore_snapshot(loaded[&"snapshot"]), "public Runtime restore")
+			var actual: Dictionary = _capture()
+			_expect(var_to_bytes(actual[&"snapshot"]) == var_to_bytes(expected[&"snapshot"]), "all typed Runtime fields match Writer after restore")
+			_write(label + "-restored", {&"expected": expected, &"actual": actual, &"load_result": loaded})
+			if branch == "night_absent" and label != "1079": _assert_night(points["1079"], actual)
+		# A separate fresh load of the pre-boundary package drives both real ticks.
+		_fresh(int(points["1079"][&"seed"]))
+		_restore_inputs(points["1079"])
+		var pre: Dictionary = SaveService.load_snapshot(_slots[branch + "-1079"])
+		_expect(bool(pre.get(&"ok", false)), "replay public pre-boundary load")
+		if not bool(pre.get(&"ok", false)): return
+		_expect(_runtime.restore_snapshot(pre[&"snapshot"]), "replay public restore")
+		_expect(var_to_bytes(_runtime.create_snapshot()) == var_to_bytes(points["1079"][&"snapshot"]), "replay starts at exact all-field pre-boundary state")
+		_connect_phases()
+		var previous: Dictionary = _capture()
+		if branch == "overtime": _guard_same_minute()
+		for label: String in ["1080", "1081"]:
+			_clock.advance_one_minute()
+			var actual: Dictionary = _capture()
+			_expect(var_to_bytes(actual[&"snapshot"]) == var_to_bytes(points[label][&"snapshot"]), "two-tick replay entire typed equality " + label)
+			if branch == "overtime":
+				_write(label + "-duty", _duty_delta(previous, actual))
+				_guard_same_minute()
+			else: _assert_night(points["1079"], actual)
+			_write(label + "-replay", actual)
+			previous = actual
+		var writer_branch: Dictionary = _read_point("branch")
+		_expect(var_to_bytes(_phases) == var_to_bytes(writer_branch["phase_observations"]), "same boundary phase order and full typed states")
+		if branch == "overtime":
+			var unload: Dictionary = _unload()
+			for label: String in ["completed", "off_duty"]:
+				_expect(var_to_bytes(unload[label][&"snapshot"]) == var_to_bytes(writer_branch["unload"][label][&"snapshot"]), "safe unloading replay all typed fields " + label)
+		for key: String in hashes:
+			var row: Dictionary = hashes[key]
+			for member: String in row[&"members"]:
+				var member_row: Dictionary = row[&"members"][member]
+				member_row[&"original_after"] = FileAccess.get_sha256(String(row[&"original"]).path_join(member))
+				member_row[&"copy_after"] = FileAccess.get_sha256(String(row[&"copy"]).path_join(member))
+				_expect(member_row[&"original_before"] == member_row[&"original_after"] and member_row[&"copy_before"] == member_row[&"copy_after"], "original and copied members unchanged after loads/replay")
+		_branches[branch] = {&"member_hashes": hashes, &"phase_observations": _phases.duplicate(true)}
+		_write("branch", _branches[branch])
+
+func _getter_join(raw: Variant, saved: Variant, path: String, differences: Array) -> bool:
+	# Only diagnostic getter-to-save joining permits StringName/String and key
+	# ordering differences. All formal/replay Runtime checks above remain bytes.
+	if typeof(raw) != typeof(saved):
+		var text_pair: bool = typeof(raw) in [TYPE_STRING, TYPE_STRING_NAME] and typeof(saved) in [TYPE_STRING, TYPE_STRING_NAME]
+		differences.append({"path": path, "kind": "StringName_String" if text_pair else "type_mismatch", "raw_type": typeof(raw), "saved_type": typeof(saved)})
+		return text_pair and String(raw) == String(saved)
+	if raw is Dictionary:
+		var matched: Dictionary = {}
+		var ok: bool = raw.size() == saved.size()
+		var raw_order: Array = []
+		var saved_order: Array = []
+		for key: Variant in saved: saved_order.append(String(key))
+		for key: Variant in raw:
+			var name: String = String(key)
+			raw_order.append(name)
+			if matched.has(name) or not saved.has(name):
+				ok = false
+				continue
+			matched[name] = true
+			if typeof(key) != TYPE_STRING:
+				differences.append({"path": path + "/" + name, "kind": "key_StringName_String" if typeof(key) == TYPE_STRING_NAME else "invalid_key_type", "raw_type": typeof(key), "saved_type": TYPE_STRING})
+				ok = typeof(key) == TYPE_STRING_NAME and ok
+			ok = _getter_join(raw[key], saved[name], path + "/" + name, differences) and ok
+		if raw_order != saved_order: differences.append({"path": path, "kind": "key_order", "raw": raw_order, "saved": saved_order})
+		return ok
+	if raw is Array:
+		if raw.size() != saved.size(): return false
+		var ok: bool = true
+		for index: int in range(raw.size()): ok = _getter_join(raw[index], saved[index], path + "/" + str(index), differences) and ok
+		return ok
+	return var_to_bytes(raw) == var_to_bytes(saved)
+

docs/plans/original-completion-2026-10-02/execution/p04-01/s2b/run_s2b_native.py

@@ -0,0 +1,274 @@
+#!/usr/bin/env python3
+"""S2b-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_s2b_shift_probe.gd"
+ENGINE = Path("/Applications/Godot.app/Contents/MacOS/Godot")
+PATTERNS = BUILD / "audit-support/p04-whole-closure-r01/current-patterns"
+AUDIT = BUILD / "audit-support/p04-whole-closure-r01/ROOT_S2B_EXISTING_SOURCE_AUDIT.json"
+MATERIALS = [PATTERNS / "approved-shift-current-r01.gd", AUDIT] + [
+    HERE.parent / folder / name for folder, name in [
+        ("s1", "run_s1_native.py"), ("s1", "S1_ROOT_RUN.md"),
+        ("s2", "run_s2a_native.py"), ("s2", "p04_s2a_midnight_probe.gd"),
+        ("s2", "S2_ROOT_RUN.md"),
+    ]
+]
+
+def template_guard(root):
+    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 must use explicit types; inferred declaration found")
+    for forbidden in ["._update_shift_states(", ".record_duty_minute("]:
+        if forbidden in probe:
+            raise ValueError("unsupported duty mutation: " + 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-s2b-native-r01")
+    parser.add_argument("--plan", type=Path, default=BUILD / "p04-s2b-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("s2b_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 + [
+        QA, Path(__file__), HERE / "S2B_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-s2b-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("s2b_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_S2B_EVIDENCE=str(out / "raw"), THE_TREE_S2B_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("s2b_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="S2b shift: <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("s2b_diagnostics", r.SUPPORT / "diagnostic_gate.py")
+        owners = module("s2b_owners", r.SUPPORT / "ordinary_proof_gate.py")
+        audio = module("s2b_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(len(formal_before) == 6, "three formal points for each of two fresh branches")
+            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,
+                          "S2b actual outcome receipt")
+                r.require(sorted(receipt["branches"]) == ["night_absent", "overtime"],
+                          "exact independent Writer cases and formal Reader case")
+                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)