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

This commit is contained in:
Agent Control Runner
2026-10-16 03:11:58 +09:00
parent 5475391647
commit 64617018ed
6 changed files with 910 additions and 15 deletions
@@ -1,5 +1,28 @@
# 初期監査の検証器
## repair7 定義(以前の版識別説明に優先)
registered は deb1bcb、execution_origin は 9767b0a、checker は実際に
実行したcheckerのcommitとbytes、product_target は
`{main_anchor, submitted, composed_tree}` とする。checkerのSHAを製品版に
代用しない。提出後にしか確定しない自己blob照合は未実施としてRootが検証する。
`duty_payments` は決算日のpayroll settlement取引のうち、
ShiftInstance.record_duty_minute の勤務recordとstaff_id/day単位で
結合できる取引の件数。期待値は1のまま。独立した通貨項目ではない。
collectorはpayroll行と勤務recordの実rawを結合し、結合不能なら失敗する。
融資の公開返済API `repay_loan_principal` とledger分類
`LOAN_PRINCIPAL_REPAID` / `financing_principal_outflows` は存在する。
自動返済や全UIの完成は主張しない。「融資機構がない」という説明は撤回する。
midnight_replayは同minuteの決算API再呼出しによる実測までpending。
1442完了やservice request再送を1440売上・決算API再呼出しの証明にしない。
shift_completionはapproved-shift-current-r01.gdの1080観測を使う。
D05のEV停止帰属、停電と休業の厳密な順序、費用時点の未確定部分はpending。
全case収集、登録validate、独立whole review、統合準備の順序を維持する。
本修理での試験・engine実行はworkerでは行わない。
## repair6 現在値(2026-10-16)
旧値: C全体は `native ordering bytes` によりblocked、実applicabilityは拒否、
@@ -0,0 +1,98 @@
# repair7 作業状態
状態: blocked(実装未完了)。whole合格、登録validate成功、実測成功を意味しない。
提出547のGit objectと既存C実装は保持。今回の変更は未コミット。
engine・全回帰・checker試験・登録validate・Runner操作は未実行。
原15条件+EV、7領域、3安全性結果、30/88、部分点0を維持する。
## 実装した部分と未検証部分
checkerに名前付きtargetの照合を追加した。main_anchorは
9694f50ee63750192376ca69bc7dda3f884784c5、submittedは提出後に確定する実commit。
composed_treeはRoot例の904bbbを固定せず、毎回merge-treeから再計算する。
実行時の一時object directoryとcanonical objectsのread-only alternateを使い、
canonical refs/HEAD/indexへの書込み命令を使わない。この実装はまだ実行していない。
全tree entryと選択sourceを区別し、1338件のpath/bytes/SHA256をGit blobと
runner-input.source_beforeで両方向比較する。source選択規則はmain_anchorの
tools/run_all_tests.pyのSHAに固定。追加選択pathはcheckerのみ。
3139件の変更されていない既存metadataを明示分類し、追加・変更・削除pathは
登録scope内のみ許す。submitted差分もregisteredから独立にscope検査する。
ORDER_FUNCTIONS、REACH_REFS、applicability_deltaはcomposed_treeのbytesを読む。
candidate_treeは実候補のtree OIDとし、composed_treeの全entryとの一致を要求する。
checker metadataだけのcommit SHA差はengine再実行の理由にしない。
checker source bytesはsubmitted/composed/checker commitと別々に比較する。
自己blob一致は提出前なので**未実施**。Rootが提出後に検証する。
## 新しい入力形式(未試験)
既存indexの条件・authority・domain・rawを省略する形式ではない。
- `candidate_commit`: 実際にcheckerを実行するcheckoutのHEAD。
- `candidate_tree`: 実際に検証する候補tree。登録時にも指定し、比較を省略しない。
- `product_binding.product_target`: main_anchor/submitted/composed_treeの完全OID。
- `product_binding.tested_product`: 実測製品commit。製品manifestをGitから再計算する。
- `product_binding.runner_inputs`: 実runner-inputファイルのdescriptor配列。
descriptorはpath/bytes/sha256に加えてmaterialsとwrapperを持つ。
materialsはnative固定input/QAの原pathから供給済み実ファイルdescriptorへのmap。
原pathは由来のみで、そこからファイルを開かない。wrapperも実ファイルdescriptor。
Cの3 runのrunner-inputを全て含める。各selected_source_mapの差分は宣言済みQA
override 1件だけに限定する。
- `versions`: registered、execution_origin、checker、testedの4役割。
checkerはcommit/sha256。testedはproduct_target/manifest_sha256。
manifest digestはpath→{bytes,sha256}をsort_keys=True、separators=(',', ':')の
JSON UTF-8で符号化したSHA256。checkerのSHAは製品digestに代用しない。
- `native_bundle`: 従来のmanifestキーからpath/bytes/sha256 descriptorへ変更。
OWNERSHIP_RAW_PROVENANCE.jsonの実ファイルを指定する。
descriptorの相対pathはp04-evidence配下。絶対pathはbuild_dir/audit-support配下の
p04-whole-closure-r01、native-first-current、p04-current-native-series-r01の3系統のみ。
ファイルを追加コピーせずその場で読む。symlink・欠落・SHA/bytes不一致は失敗する。
## 残作業(Rootのnative実行へ進める状態ではない)
1. S1/S2/S3のprobeと、import/parse各1回→実測→正式Writer→別PID Readerを
結合するwrapperは未実装。既存templateを読んだだけで完成とは扱わない。
2. 全条件のactor/service/予約/帳簿/同minute/正式保存rawを結合するcollectorは
未実装。既存verify_nativeの有限3条件を全条件collectorに代用できない。
verified_conditionsの全条件への動的導出も未実装。
3. 新しいtarget側のconfig/input/environment実体照合は未完了。
現在は固定input/QA/wrapperと既存environment pin検査があり、これだけで
engine環境を全runnerの実起動に結合したとは主張しない。
4. 新入力形式へのfixture更新・Runner正式試験・提出後自己blob検査は未実施。
既存repair6の試験結果を今回の変更の合格として転記しない。
実装残作業を仕様不足として偽装しない。原本の未確定D05部分は従来どおりpending。
新しい金額、費用帰属、EV停止規則を追加して残作業を埋めない。
## B/Cの確定事項と実体確認
main969のautoload/simulation_runtime.gdのrepay_loan_principal本体と
tests/mvp/test_original_midnight_settlement.gdの返済テストをGit objectから読んだ。
公開返済APIとledger分類は存在する。自動返済・全UI完成は主張しない。
duty_paymentsはstaff_id/dayで勤務recordに結合できるpayroll settlement取引数で、
期待値1のまま。実コードではWAGE行のsource_entity_idがpayroll settlement_id、
settlement.rowsがstaff_id/payment、ShiftInstance snapshotがstaff_id/start_dayと
worked_minutes/overtime_minutesを保持する。給与行の数や勤務分数をそのまま
duty_paymentsに代入してはならない。day基準と勤務の実測差分も照合対象。
具体的に読んだ追加rawは
`p04-current-native-series-r01/p04-paid-shift-midnight-native-r01/order-evidence/midnight-midnight.json`
のstaff/payroll/economy部分と
`p04-current-native-series-r01/p04-paid-shift-midnight-formal-native-r01/runner-input.json`
のsource_beforeサンプル・fixed_input_pins。これを39raw読了・採用とはしない。
加えてnative-first-currentのactual-native、dependency-midnight-native、
dependency-formal-native各runner-input.jsonのキーとQA/QA_overridesを読んだ。
前2件にはfixed_input_pinsがなく、formalのみ保存入力のpinを持つ。
midnight_replayは実際の同minute決算API再呼出しまでpending。
paid資料の1442完了は1440売上境界を証明せず、request再送は決算API再呼出しを
証明しない。shift_completionはROOT_APPROVED_SHIFT_CURRENT_RESULTと
approved-shift-current-r01.gdの1080系列を結合する必要がある。
順序はworker実装完了→Root資源確認→S1→S2→S3→collector→登録validate→
独立whole review→prepare-integration/final。全case収集前のwhole合格はない。
S1準備に移行、A未検証。repair8はs1/の3ファイルによるコード準備のみ。未実測・原P04 whole未達。
@@ -0,0 +1,90 @@
# S1 Root run(コード準備のみ)
S1コード準備完了・未実測・原P04 whole未達。D05 pending、A未検証。
engine起動・全回帰・コミットはworkerでは実施しない。
assertは各caseの停止中の課金行0、予約の孤立0、到着1回だけ。
費用帰属、金額、EV停止規則を決めない。remaining_minutes保持は生観測として保存する。
## Rootが実行するコマンド
以下はSource checkout外のtask buildにplan/cache/rawを保存する。
Rootは先に空き容量・他のengine・終了済cacheへのアクセスを確認する。
既存出力は削除しない。再試行する場合はplan/outputの両方に新しいr番号を指定する。
```sh
export PYTHONPYCACHEPREFIX='/Users/korizatou114/Projects/TheTree-managed-worktrees/p04-01/.runtime/P04-01/build/s1-python-cache'
python3 '/Users/korizatou114/Projects/TheTree-managed-worktrees/p04-01/P04-01-implement/docs/plans/original-completion-2026-10-02/execution/p04-01/s1/run_s1_native.py' --check-only --plan '/Users/korizatou114/Projects/TheTree-managed-worktrees/p04-01/.runtime/P04-01/build/p04-s1-root-plan-r01.json' --output '/Users/korizatou114/Projects/TheTree-managed-worktrees/p04-01/.runtime/P04-01/build/p04-s1-native-r01'
python3 '/Users/korizatou114/Projects/TheTree-managed-worktrees/p04-01/P04-01-implement/docs/plans/original-completion-2026-10-02/execution/p04-01/s1/run_s1_native.py' --plan '/Users/korizatou114/Projects/TheTree-managed-worktrees/p04-01/.runtime/P04-01/build/p04-s1-root-plan-r01.json' --output '/Users/korizatou114/Projects/TheTree-managed-worktrees/p04-01/.runtime/P04-01/build/p04-s1-native-r01'
```
既定の製品sourceは
`/Users/korizatou114/Projects/とりにく工房ゲーム制作部`。
workerのprobeだけを `tests/s3/test_runtime_staff_assignment.gd` にQA overrideする。
製品sourceの全選択ファイルをpinし、終了済cacheのselected mapとの完全一致を実runで要求する。
sourceの役割変更や不一致を自動承認しない。
既定cache:
`/Users/korizatou114/Projects/TheTree-managed-worktrees/p18-01-model/.runtime/P18-01-CODE/build/p18-relationship-evidence`。
workerからreceiptの読取りはOSに拒否された。check-onlyはその具体的path/errorを
cache_accessに記録し、engineやdefault playerには触らない。
Rootのcheck-onlyでは読める必要がある。実runではreceiptがpassedであること、
cacheの存在・全hash・cp -cの別inodeを必須とする。アクセス権やRoot境界は変更しない。
## 実行順序とexit
source_refs → QA override 1件 → private stage → bootstrap → cache cp -c/別inode →
import 1回 → parse 1回 → writer → 別PID reader →
finally owned_outer.finish_captured → fixed_guard。
normal watchdogは既存1200秒、importは既存180秒で変更しない。
実runのdefault-player保護は既存runnerを使う。
check-onlyの期待exitは0。engine起動0で入力SHAとplanを出力する。
cache_accessが非nullならRoot確認が未完でありnative開始条件を満たさない。
実runの期待exitは0(4phase成功、Writer/Reader別PID/profile、6安全性checksずつ、
14保存pointの正式復元、Writer保存物不変、owned cleanupとsource/default-player guard成功)。
strict phase失敗またはguard失敗は1、native開始前の入力・資源不備は2。
最初の失敗で停止し、自動再試行しない。exit0でも原P04 whole合格ではない。
## 出力とraw一覧
出力root:
`/Users/korizatou114/Projects/TheTree-managed-worktrees/p04-01/.runtime/P04-01/build/p04-s1-native-r01/`
- source_before.json、source_after.json、selected-source-map.json、runner-input.json
- runner-started.json、runner-ended.json、execution.json、PID-set.json、outer-owned-finally.json、raw-inventory.json
- stage/(別inodeの製品source+QA、.godot clone)、starters/、bootstrap/
- cache-reuse.json(全cache hashと終了済receipt pin)、parse-paths.json、contracts.json
- import/、parse/、writer/、reader/のstdout.log、stderr.log、engine-native.log、
strict runner生成のprocess/profile/start/end/capture証拠。全フォルダを保存する
- 各phaseの*-receipt.json、*-actual-phase.json、writer-combined-raw.log、reader-combined-raw.log
- raw/writer.json、raw/reader.json(PID/profile、正式slot path、checks/errors)
- raw/{writer,reader}-observations.{json,var}
- raw/{writer,reader}-{landing,no-safe-landing}-observations.{json,var}
- raw/{writer,reader}-{landing,no-safe-landing}-{before,boarded,depleted,stopped,held,landing,recovered}.{json,var}
- phase receiptが指すuser://saves/p04_s1/の全正式package。
WriterとReader各14slot、各slotのjournal.log、manifest.json、thumbnail.png、world.binを保持する。
private startup/runtime profileのlog/lock等も既存runner方式で保持し削除しない
JSONは閲覧用でVariant型を落とすため、必ずtyped .varと対で読む。
case観測はseed、実task.required_items、予約配列、ledger全行、trip/elevator report、
domain event、facility/service ID、全Runtime snapshot、コマンド結果を含む。
割当時のreserved_before_travelはイベント内の生観測で、正式保存pointとは区別する。
Readerは正式4-member packageを自身のprivate profileへ全コピーしてpublic load/restore後、
tickせず観測しWriter typed snapshotとの完全一致を確認する。
到着イベント履歴はWriter採取rawであることをReader側にも明示する。
no-safe-landingの残り時間保持とclosureによるrelease時点はrawで確認する。
給電復帰後も対象clinicは直接休業のままにして追加の搬送を防ぐ。
S2/S3、collector、A仕上げ、登録validate、独立whole review、統合は後続であり本提出の範囲外。
## worker静的確認の適用範囲
製品rootのtools/run_all_tests.pyもworkerから読取り拒否されたため、workerのcheck-onlyは
--project-rootにSource checkoutを明示した。選択sourceは1300件であり、Rootのcurrent製品1338件の検証を代行しない。
Rootは上記既定product_rootで独自のfresh planを作る。worker planを実runに転用しない。
停電と着床を別規則に分離しない。reserveをpublic commandで0にし、同じ更新境界の前に
対象clinicへ直接休業commandを送り、通常tickによる停電・予約解放・安全到着の実順序を記録する。
着床不能caseの予約は乗車中保持され得るため、releaseの時点を予測しない。
strict runnerの同一test名logはphaseごとにcombined-rawとして退避し、actual-phase/executionの
logを退避pathへ結合する。元strict receiptは変更せずstrict_runner_log_pathも残す。
@@ -0,0 +1,257 @@
extends SceneTree
# S1 only; setup follows test_rescue_elevator and transport:94-149.
const SEED := 712
const TASK := &"s1_rescue_with_tool"
var runtime: Variant
var clock: Variant
var state: Variant
var folder := ""
var mode := ""
var clinic_id: StringName
var checks := 0
var failures := 0
var errors: Array = []
var slots: Dictionary = {}
var observations: Dictionary = {}
var commands: Array = []
var arrivals: Array = []
var pretravel: Dictionary = {}
func _initialize() -> void: call_deferred("run")
# Exactly three safety assertions per case; operational failures are separate.
func outcome(ok: bool, label: String) -> void:
checks += 1
if not ok:
failures += 1
push_error(label)
func io(ok: bool, label: String) -> bool:
if not ok:
errors.append(label)
push_error(label)
return ok
func raw(label: String, value: Variant) -> void:
var file := FileAccess.open(folder.path_join(mode + "-" + label + ".var"), FileAccess.WRITE)
if io(file != null, "typed output " + label):
file.store_buffer(var_to_bytes(value)); file.flush(); file.close()
file = FileAccess.open(folder.path_join(mode + "-" + label + ".json"), FileAccess.WRITE)
if io(file != null, "JSON output " + label):
file.store_string(JSON.stringify(value, "\t")); file.flush(); file.close()
func read_json(path: String) -> Dictionary:
var file := FileAccess.open(path, FileAccess.READ)
if not io(file != null, "read " + path): return {}
var value: Variant = JSON.parse_string(file.get_as_text()); file.close()
if not io(value is Dictionary, "dictionary " + path): return {}
return value
func typed(path: String) -> Dictionary:
var file := FileAccess.open(path, FileAccess.READ)
if not io(file != null, "read typed " + path): return {}
var value: Variant = bytes_to_var(file.get_buffer(file.get_length())); file.close()
if not io(value is Dictionary, "typed dictionary " + path): return {}
return value
func cleanup() -> void:
if runtime != null: runtime.free()
if clock != null: clock.free()
if state != null: state.free()
runtime = null; clock = null; state = null
func reset() -> bool:
cleanup(); commands = []; arrivals = []; pretravel = {}; clinic_id = &""
clock = load("res://autoload/game_clock.gd").new()
state = load("res://autoload/game_state.gd").new()
runtime = load("res://autoload/simulation_runtime.gd").new()
clock.reset(1, 6, 0); state.reset()
var initialized: Dictionary = runtime.initialize(clock, state, SEED)
clock.set_paused(true)
runtime.configure_save_policy({&"enabled": false, &"daily_checkpoint_enabled": false})
runtime.domain_events.subscribe(&"ELEVATOR_ARRIVED", &"s1.raw", on_arrival)
runtime.domain_events.subscribe(&"STAFF_TASK_ASSIGNED", &"s1.reservation", on_assignment)
return io(bool(initialized.get(&"valid", false)), "runtime initialization")
func on_assignment(event: SimulationDomainEvent) -> void:
if TASK in event.entity_ids and pretravel.is_empty():
pretravel = capture()
pretravel[&"staff_cell"] = runtime.staff_service.get_staff(&"medic").current_cell
func on_arrival(event: SimulationDomainEvent) -> void:
arrivals.append(event.to_dictionary())
func cmd(kind: StringName, payload: Dictionary) -> bool:
var result: CommandResult = runtime.submit_command(kind, payload, &"player")
commands.append(result.to_dictionary())
return io(result.is_success(), "public command " + String(kind))
func capture() -> Dictionary:
var reservations: Array = []
for reservation: InventoryReservation in runtime.inventory_service.get_reservations():
reservations.append(reservation.create_snapshot())
var ledger: Array = []
for entry: LedgerEntry in runtime.economy_service.get_entries(): ledger.append(entry.to_dictionary())
var events: Array = []
for event: SimulationDomainEvent in runtime.domain_events.recent_events(): events.append(event.to_dictionary())
var services: Array = []
for facility: FacilityInstance in runtime.facility_service.get_instances():
services.append({&"facility_id": facility.instance_id, &"active_service_ids": facility.active_services.keys()})
var task: StaffTask = runtime.staff_service.get_task(TASK)
return {&"seed": SEED, &"minute": clock.absolute_minute, &"snapshot": runtime.create_snapshot(),
&"ledger": ledger, &"reservations": reservations, &"task": task.create_snapshot() if task != null else {},
&"trip": runtime.elevators.get_actor_trip(&"staff:medic"),
&"elevator_report": runtime.elevators.get_report(), &"domain_events": events,
&"arrivals": arrivals.duplicate(true), &"commands": commands.duplicate(true),
&"service_ids": services, &"staff_invariants": runtime.staff_service.validate_invariants()}
func construct() -> bool:
if not reset(): return false
for spec: Array in [[&"infirmary", Vector3i(-6,0,1)], [&"service_elevator", Vector3i(0,0,1)]]:
if not cmd(&"build_facility", {&"definition_id": spec[0], &"origin": spec[1]}): return false
clock.advance_minutes(400)
for facility: FacilityInstance in runtime.facility_service.get_instances():
if facility.definition_id == &"infirmary": clinic_id = facility.instance_id
if not io(not clinic_id.is_empty(), "constructed clinic"): return false
for id: StringName in [&"patient", &"medic"]:
if not cmd(&"register_staff", {&"staff_id": id, &"role_id": &"medic", &"options": {
&"current_cell": Vector3i(0,1,1) if id == &"patient" else Vector3i(5,0,1),
&"health": 0 if id == &"patient" else 100, &"state": StaffRuntime.State.AVAILABLE,
&"role_tags": [&"medic"], &"permitted_task_tags": [&"rescue"]}}): return false
# Actual task.required_items uses the ordinary Runtime reservation callback.
if not cmd(&"add_inventory_stack", {&"owner_inventory_id": &"staff.supplies", &"item_id": &"tool/cleaning_kit", &"quantity": 1}): return false
return cmd(&"create_staff_task", {&"task_id": TASK, &"task_type": &"rescue_pickup", &"priority": 95, &"options": {
&"target_entity_id": &"patient", &"target_cell": Vector3i(0,1,1),
&"required_role_tags": [&"medic"], &"task_tags": [&"rescue"],
&"required_items": {&"tool/cleaning_kit": 1}, &"estimated_minutes": 30,
&"interrupt_level": 0, &"emergency": true}})
func save_point(label: String) -> void:
# The raw snapshot is observed immediately before the official save.
var point := capture()
raw(label, point)
var path := "user://saves/p04_s1/" + label
var saved := SaveService.save_snapshot(path, point[&"snapshot"])
if not io(bool(saved.get(&"ok", false)), "formal save " + label): return
var inspected := SaveService.inspect_snapshot(path)
io(bool(inspected.get(&"ok", false)) and bool(inspected.get(&"read_only", false)) and var_to_bytes(inspected.get(&"snapshot", {})) == var_to_bytes(point[&"snapshot"]), "formal inspect " + label)
slots[label] = ProjectSettings.globalize_path(path)
func safety(row: Dictionary) -> void:
var charge_rows: int = row[&"held_after"][&"ledger"].size() - row[&"reserve_exhausted_before_tick"][&"ledger"].size()
outcome(charge_rows == 0, "停止中の課金行0")
var orphans := 0
# Check each observed boundary, both directions, using actual task ownership.
for point_name: StringName in [&"boarded", &"stopped", &"held_after", &"final"]:
var point: Dictionary = row[point_name]
var tasks: Array = point[&"snapshot"][&"systems"][&"runtime"][&"staff"][&"tasks"]
var active: Dictionary = {}
for reservation: Dictionary in point[&"reservations"]:
if int(reservation[&"state"]) != InventoryReservation.State.ACTIVE: continue
if String(reservation[&"purpose"]) != "staff_task": continue
active[String(reservation[&"reservation_id"])] = reservation
var linked := false
for task: Dictionary in tasks:
if String(task[&"task_id"]) == String(reservation[&"requester_id"]) and String(reservation[&"reservation_id"]) in task[&"reservation_ids"]: linked = true
if not linked: orphans += 1
for task: Dictionary in tasks:
for id: Variant in task[&"reservation_ids"]:
if not active.has(String(id)): orphans += 1
outcome(orphans == 0, "予約の孤立0")
var counts := {&"staff:medic": 0, &"staff:patient": 0}
var stops: Array = row[&"boarded"][&"elevator_report"][&"shafts"][0][&"stops"]
var requests: Dictionary = {}
for trip: Dictionary in row[&"boarded"][&"elevator_report"][&"trips"]: requests[trip[&"actor_id"]] = trip[&"request_id"]
for event: Dictionary in row[&"final"][&"arrivals"]:
var payload: Dictionary = event["payload"]
var actor := StringName(payload.get(&"actor_id", &""))
if counts.has(actor) and payload.get(&"request_id") == requests.get(actor) and payload.get(&"cell") in stops: counts[actor] += 1
outcome(counts[&"staff:medic"] == 1 and counts[&"staff:patient"] == 1, "到着1回")
row[&"safety_raw"] = {&"stopped_ledger_delta": charge_rows, &"orphans": orphans, &"arrivals_by_actor": counts}
func writer_case(label: String, blocked: bool) -> void:
if not construct(): return
var row := {&"seed": SEED, &"case": label, &"before": capture()}
save_point(label + "-before")
var reserved := false
for tick: int in 130:
clock.advance_one_minute()
var task: StaffTask = runtime.staff_service.get_task(TASK)
if not reserved and not task.reservation_ids.is_empty():
row[&"reserved_before_travel"] = pretravel.duplicate(true); reserved = true
if task.task_type == &"rescue_transport" and runtime.elevators.get_actor_trip(&"staff:patient").get(&"state") == PathService.ELEVATOR_REQUEST_BOARDED: break
if not io(reserved and not pretravel.is_empty() and pretravel.get(&"staff_cell") == Vector3i(5,0,1) and not pretravel[&"task"][&"reservation_ids"].is_empty() and runtime.elevators.get_actor_trip(&"staff:patient").get(&"state") == PathService.ELEVATOR_REQUEST_BOARDED, "real reserved task reaches boarded pair"): return
row[&"boarded"] = capture(); save_point(label + "-boarded")
var stops: Array = runtime.elevators.get_report()[&"shafts"][0][&"stops"]
# Public fixture API reused literally from transport test, not a private mutation.
if blocked:
for cell: Vector3i in stops: runtime.path_service.configure_cell_traffic(cell, 4, 0, true)
var reserve: int = state.get_resource(&"mana")
if not cmd(&"change_resources", {&"deltas": {&"mana": -reserve}}): return
row[&"reserve_exhausted_before_tick"] = capture()
if not cmd(&"set_facility_conditions", {&"facility_id": clinic_id, &"conditions": {&"operator_open": false}}): return
clock.advance_one_minute(); row[&"depleted"] = capture(); save_point(label + "-depleted")
clock.advance_one_minute(); row[&"stopped"] = capture(); save_point(label + "-stopped")
clock.advance_minutes(20); row[&"held_after"] = capture(); save_point(label + "-held")
row[&"remaining_minutes_before"] = row[&"stopped"][&"elevator_report"][&"shafts"][0][&"remaining_minutes"]
row[&"remaining_minutes_after"] = row[&"held_after"][&"elevator_report"][&"shafts"][0][&"remaining_minutes"]
row[&"held_remaining_unchanged"] = row[&"remaining_minutes_before"] == row[&"remaining_minutes_after"]
if blocked:
for cell: Vector3i in stops: runtime.path_service.configure_cell_traffic(cell)
clock.advance_one_minute()
row[&"landing"] = capture(); save_point(label + "-landing")
if not cmd(&"change_resources", {&"deltas": {&"mana": reserve}}): return
clock.advance_minutes(45 if blocked else 1)
row[&"final"] = capture(); save_point(label + "-recovered")
row[&"reservation_release_raw"] = runtime.staff_service.get_task(TASK).create_snapshot()
safety(row); observations[label] = row; raw(label + "-observations", row)
func reader() -> void:
var receipt := read_json(folder.path_join("writer.json"))
if not io(int(receipt.get("pid", 0)) != OS.get_process_id() and String(receipt.get("profile", "")) != OS.get_user_data_dir(), "distinct Reader PID/profile"): return
for label: String in receipt["slots"]:
if not reset(): return
var original := String(receipt["slots"][label])
var target := ProjectSettings.globalize_path("user://saves/p04_s1/" + label)
if not io(DirAccess.make_dir_recursive_absolute(target) == OK, "reader package directory"): return
var directory := DirAccess.open(original)
if not io(directory != null, "writer package exists"): return
var members := directory.get_files(); members.sort()
if not io(directory.get_directories().is_empty() and members == PackedStringArray(["journal.log", "manifest.json", "thumbnail.png", "world.bin"]), "complete formal package"): return
for member: String in members:
if not io(DirAccess.copy_absolute(original.path_join(member), target.path_join(member)) == OK and FileAccess.get_sha256(original.path_join(member)) == FileAccess.get_sha256(target.path_join(member)), "exact formal member"): return
var loaded := SaveService.load_snapshot(target)
if not io(bool(loaded.get(&"ok", false)), "formal Reader load"): return
if not io(runtime.restore_snapshot(loaded[&"snapshot"]), "public Reader restore"): return
var expected := typed(folder.path_join("writer-" + label + ".var"))
if not io(expected.has(&"snapshot"), "writer point exists"): return
io(var_to_bytes(runtime.create_snapshot()) == var_to_bytes(expected[&"snapshot"]), "Reader exact typed snapshot")
var point := capture() # No tick after formal restore.
point[&"writer_observation"] = expected
observations[label] = point; slots[label] = target; raw(label, point)
for label: String in ["landing", "no-safe-landing"]:
var row := typed(folder.path_join("writer-" + label + "-observations.var"))
if not io(not row.is_empty(), "writer case exists"): return
for pair: Array in [[&"boarded", "boarded"], [&"stopped", "stopped"], [&"held_after", "held"], [&"final", "recovered"]]:
var point: Dictionary = observations[label + "-" + pair[1]]
for key: StringName in [&"snapshot", &"ledger", &"reservations"]: row[pair[0]][key] = point[key]
# Arrival history is Writer raw; Reader proves restored state separately.
safety(row); raw(label + "-observations", row)
func run() -> void:
folder = OS.get_environment("THE_TREE_S1_EVIDENCE"); mode = OS.get_environment("THE_TREE_S1_MODE")
if folder.is_empty() or mode not in ["writer", "reader"]: quit(2); return
if not io(DirAccess.make_dir_recursive_absolute(folder) == OK, "evidence directory"): quit(2); return
if mode == "writer":
writer_case("landing", false); writer_case("no-safe-landing", true)
else: reader()
raw("observations", observations)
var receipt := {&"pid": OS.get_process_id(), &"profile": OS.get_user_data_dir(), &"slots": slots,
&"checks": checks, &"failures": failures, &"operational_errors": errors,
&"whole_goal_accepted": false, &"unit_accepted": false}
var file := FileAccess.open(folder.path_join(mode + ".json"), FileAccess.WRITE)
if io(file != null, "phase receipt"):
file.store_string(JSON.stringify(receipt, "\t")); file.flush(); file.close()
cleanup()
print("S1 EV power: %d checks, %d failures" % [checks, failures])
quit(1 if failures or not errors.is_empty() or checks != 6 else 0)
@@ -0,0 +1,253 @@
#!/usr/bin/env python3
"""S1-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_s1_ev_power_probe.gd"
ENGINE = Path("/Applications/Godot.app/Contents/MacOS/Godot")
MATERIALS = [
BUILD / "audit-support/p04-whole-closure-r01/current-patterns/p04-idempotency-paid-formal-native-r01.py",
BUILD / "audit-support/p04-whole-closure-r01/current-patterns/idempotency-paid-formal-r01.gd",
BUILD / "audit-support/p04-whole-closure-r01/SOURCE_MANIFEST.json",
BUILD / "audit-support/p04-native-existing-sources/p04-01-power-formal-recovery-r04.gd",
BUILD / "audit-support/p04-native-existing-sources/p04-01-power-formal-recovery-native-r04.py",
BUILD / "audit-support/p04-native-existing-sources/SOURCE_MANIFEST.json",
BUILD / "audit-support/native-first-current/SOURCE_MANIFEST.json",
BUILD / "audit-support/native-first-current/actual-native/runner-input.json",
BUILD / "audit-support/p04-current-native-series-r01/SOURCE_MANIFEST.json",
BUILD / "audit-support/p04-current-native-series-r01/p04-idempotency-paid-formal-native-r01/runner-input.json",
]
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-s1-native-r01")
parser.add_argument("--plan", type=Path, default=BUILD / "p04-s1-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()
r = module("s1_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)
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 / "S1_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-s1-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(args.plan.is_file() and r.read(args.plan) == plan, "run exact check-only plan first")
def fixed_guard():
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("s1_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_S1_EVIDENCE=str(out / "raw"), THE_TREE_S1_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("s1_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="S1 EV power: <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("s1_diagnostics", r.SUPPORT / "diagnostic_gate.py")
owners = module("s1_owners", r.SUPPORT / "ordinary_proof_gate.py")
audio = module("s1_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) == 14, "all fourteen formal writer points")
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"] == 6 and receipt["failures"] == 0 and not receipt["operational_errors"],
"exact S1 outcome/operational receipt")
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")
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)
+189 -15
View File
@@ -5,6 +5,7 @@ Required conditions and authority bindings are fixed here, never supplied by an
See execution/p04-01/CHECKER_PROTOCOL.md for the supported input subset.
"""
import argparse
from contextlib import contextmanager
import hashlib
import json
import os
@@ -12,6 +13,7 @@ from pathlib import Path
import re
import subprocess
import sys
import tempfile
ROOT = Path(__file__).resolve().parents[1]
SPEC = ROOT / 'docs/spec/acceptance/P04-01.json'
@@ -20,6 +22,11 @@ DOMAINS = {'閉店', '停電', '破壊', '退出', '課金', 'シフト交代',
OWNER_HASH = '4583ee9aad45c4db183d72df4420df48aa8e0897e3b5efda13016f0b5cdbd68e'
REGISTERED_COMMIT = 'deb1bcb2374eb378b840ca345db952bd8572cb6e'
EXECUTION_ORIGIN = '9767b0a28ca8d252520307dbbac60a1d8cfb87ab'
CHECKER_PATH = 'tools/check_p04_order_evidence.py'
DOC_SCOPE = 'docs/plans/original-completion-2026-10-02/execution/p04-01/'
MAIN_ANCHOR = '9694f50ee63750192376ca69bc7dda3f884784c5'
SELECTION_SHA = '320ca668732b4ff72fc4449998af2327d4e5c527b06896740806aaf0487282b4'
_GIT_ENV = None
# Fixed membership, never supplied by the evidence index.
ORDER_FUNCTIONS = {
'scripts/core/simulation_kernel.gd': ('advance_one_minute', '_run_phase', '_handle_time_boundaries'),
@@ -128,7 +135,173 @@ def literal(a, b):
def git_bytes(*args):
return subprocess.check_output(['git', *args], cwd=ROOT)
env = _GIT_ENV if _GIT_ENV is not None else dict(os.environ, GIT_OPTIONAL_LOCKS='0')
return subprocess.check_output(['git', '--no-replace-objects', *args], cwd=ROOT, env=env)
@contextmanager
def isolated_objects():
"""merge-tree writes only disposable objects; canonical refs/index stay untouched.
Keep this context alive while reading the composed tree. No stage switch can
bypass it. The alternate is only an object lookup path, never a write target.
"""
global _GIT_ENV
require(_GIT_ENV is None, 'nested object store')
objects = git_bytes('rev-parse', '--path-format=absolute', '--git-path', 'objects').decode().strip()
with tempfile.TemporaryDirectory(prefix='p04-object-audit-') as temporary:
env = dict(os.environ)
for key in ('GIT_OBJECT_DIRECTORY', 'GIT_ALTERNATE_OBJECT_DIRECTORIES',
'GIT_INDEX_FILE', 'GIT_QUARANTINE_PATH'):
env.pop(key, None)
env.update(GIT_OBJECT_DIRECTORY=temporary, GIT_ALTERNATE_OBJECT_DIRECTORIES=objects,
GIT_OPTIONAL_LOCKS='0', GIT_NO_REPLACE_OBJECTS='1')
_GIT_ENV = env
try:
yield
finally:
_GIT_ENV = None
def tree_entries(tree):
entries = {}
for row in git_bytes('ls-tree', '-r', '-z', tree).split(b'\0'):
if not row:
continue
metadata, encoded = row.split(b'\t', 1)
mode, kind, oid = metadata.decode().split()
path = encoded.decode('utf-8')
require(kind == 'blob' and mode in {'100644', '100755'}, 'nonordinary tree entry: ' + path)
require(path not in entries, 'duplicate tree path')
entries[path] = (mode, oid)
return entries
def selected_path(path):
parts = path.split('/')
return (parts[0] in {'autoload', 'scripts', 'scenes', 'data', 'tests', 'assets', 'tools'}
and len(parts) > 1 and parts[-1] != '.DS_Store' and '__pycache__' not in parts
or path in {'project.godot', 'export_presets.cfg', 'Makefile', 'AGENTS.md'})
def allowed_path(path):
return path == CHECKER_PATH or path.startswith(DOC_SCOPE)
def product_manifest(entries):
result = {}
for path, (_, oid) in sorted(entries.items()):
if selected_path(path) and path != CHECKER_PATH:
data = git_bytes('cat-file', 'blob', oid)
result[path] = {'bytes': len(data), 'sha256': hashlib.sha256(data).hexdigest()}
return result
def manifest_digest(manifest):
return hashlib.sha256(json.dumps(manifest, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
def supplied_file(directory, descriptor):
"""Read supplied material in place, never copy it or follow provenance paths."""
name = descriptor['path']
path = Path(name)
if path.is_absolute():
support = directory / 'audit-support'
roots = [support / n for n in ('p04-whole-closure-r01', 'native-first-current',
'p04-current-native-series-r01')]
matches = [root for root in roots if path.is_relative_to(root)]
require(len(matches) == 1, 'unsupplied external material: ' + name)
path = safe(matches[0], path.relative_to(matches[0]).as_posix())
else:
path = safe(directory / 'p04-evidence', name)
require(type(descriptor.get('bytes')) is int and path.stat().st_size == descriptor['bytes']
and sha(path) == descriptor.get('sha256'), 'supplied material pin: ' + name)
return path
def verify_product_target(index, directory, checker_commit):
"""Four actual roles, complete source selection, and candidate equality.
This does not establish any native boundary condition. Runner input files
are checked against Git blobs, never against a Root-reported digest alone.
"""
binding = index['product_binding']
target = binding['product_target']
require(set(target) == {'main_anchor', 'submitted', 'composed_tree'}, 'product target shape')
require(target['main_anchor'] == MAIN_ANCHOR, 'main anchor changed')
for key in target:
require(re.fullmatch('[0-9a-f]{40}', target[key]) is not None, 'full product target OID')
for commit in (REGISTERED_COMMIT, EXECUTION_ORIGIN, target['main_anchor'], target['submitted'], checker_commit):
require(git_bytes('cat-file', '-t', commit).strip() == b'commit', 'role is not a commit')
require(_GIT_ENV is not None, 'merge-tree needs isolated object store')
composed = git_bytes('-c', 'rerere.enabled=false', 'merge-tree', '--write-tree',
target['main_anchor'], target['submitted']).decode().strip()
require(composed == target['composed_tree'], 'recomputed composed tree mismatch')
anchor, submission, tree = [tree_entries(t) for t in (MAIN_ANCHOR, target['submitted'], composed)]
selection = git_bytes('show', MAIN_ANCHOR + ':tools/run_all_tests.py')
require(hashlib.sha256(selection).hexdigest() == SELECTION_SHA, 'source selection authority')
changed = git_bytes('diff', '--no-ext-diff', '--no-renames', '--name-only', '-z',
REGISTERED_COMMIT, target['submitted']).split(b'\0')
require(all(allowed_path(p.decode()) for p in changed if p), 'submission outside registered scope')
for path in anchor.keys() | tree.keys():
if anchor.get(path) != tree.get(path):
require(allowed_path(path), 'new/changed path outside scope: ' + path)
baseline = product_manifest(anchor)
product = product_manifest(tree)
require(len(baseline) == 1338 and product == baseline, 'complete selected product mismatch')
metadata = sorted(p for p in anchor if not selected_path(p) and anchor[p] == tree.get(p))
require(len(metadata) == 3139, 'unchanged existing metadata membership changed')
require(set(p for p in tree if selected_path(p)) == set(product) | {CHECKER_PATH},
'unknown selected source cannot be metadata')
own = Path(__file__).read_bytes()
for role, entries in (('submitted', submission), ('composed', tree),
('checker', tree_entries(checker_commit))):
require(CHECKER_PATH in entries and git_bytes('cat-file', 'blob', entries[CHECKER_PATH][1]) == own,
'checker self blob not verified: ' + role)
tested = binding['tested_product']
require(git_bytes('cat-file', '-t', tested).strip() == b'commit'
and product_manifest(tree_entries(tested)) == product, 'tested product manifest mismatch')
candidate = tree_entries(index['candidate_tree'])
require(candidate == tree, 'candidate product/QA/collector/tree bytes differ from target')
runners = binding['runner_inputs']
require(isinstance(runners, list) and runners, 'actual runner inputs missing')
runner_paths = set()
for descriptor in runners:
runner_path = supplied_file(directory, descriptor)
require(runner_path not in runner_paths, 'duplicate runner input')
runner_paths.add(runner_path)
native = load(runner_path)
before = native['source_before']
require(set(before) == set(product) and all(
{'bytes': v['bytes'], 'sha256': v['sha256']} == product[p] for p, v in before.items()),
'runner source_before differs from complete product')
require(native['runner_sha256'] == product['tools/run_all_tests.py']['sha256'], 'runner source pin')
overrides = native['QA_overrides']
require(isinstance(overrides, list) and len(overrides) == 1
and overrides[0].startswith('tests/') and overrides[0] in product, 'one declared QA override')
selected = {p: v['sha256'] for p, v in product.items()}
selected[overrides[0]] = native['QA']['sha256']
require(native['selected_source_map'] == selected, 'selected source/QA override mismatch')
# A relocation is explicit and hash-bound; never open the original path.
materials = descriptor['materials']
fixed = native.get('fixed_input_pins', [])
require(isinstance(fixed, list), 'native fixed input pin list')
needed = fixed + [native['QA']]
require(set(materials) == {p['path'] for p in needed}, 'runner input/config material membership')
for pin in needed:
real = supplied_file(directory, materials[pin['path']])
require(real.stat().st_size == pin['bytes'] and sha(real) == pin['sha256'], 'native input/QA bytes')
wrapper = supplied_file(directory, descriptor['wrapper'])
require(sha(wrapper) == native['wrapper_sha256'], 'native wrapper bytes')
digest = manifest_digest(product)
require(index['versions'] == {
'registered': REGISTERED_COMMIT, 'execution_origin': EXECUTION_ORIGIN,
'checker': {'commit': checker_commit, 'sha256': hashlib.sha256(own).hexdigest()},
'tested': {'product_target': target, 'manifest_sha256': digest}}, 'four version roles')
require(index['source_map'] == {p: v['sha256'] for p, v in product.items()}, 'product source map')
return composed, {'manifest_sha256': digest, 'existing_metadata': metadata,
'runner_paths': runner_paths,
'tested_product': tested, 'product_target': target}
def function_bytes(data):
@@ -198,21 +371,19 @@ def applicability(index, spec, source, head):
for path, digest in registered.items():
require(hashlib.sha256(git_bytes('show', REGISTERED_COMMIT + ':' + path)).hexdigest() == digest,
'registered Git pin: ' + path)
require(index.get('versions') == {'execution_origin': EXECUTION_ORIGIN,
'registered': REGISTERED_COMMIT, 'tested': head, 'integration': None}, 'version roles')
# Refuse uncommitted game bytes; otherwise Git diff would not describe tested.
# head is the independently recomputed product tree, not the checker HEAD.
game_paths = git_bytes('ls-tree', '-r', '--name-only', '-z', head, '--', 'autoload/', 'scripts/').split(b'\0')
for encoded in game_paths:
if not encoded.endswith(b'.gd'):
continue
path = encoded.decode('utf-8')
require(path in source and sha(safe(ROOT, path)) == hashlib.sha256(git_bytes('show', head + ':' + path)).hexdigest(),
require(path in source and source[path] == hashlib.sha256(git_bytes('show', head + ':' + path)).hexdigest(),
'tested game differs from Git: ' + path)
delta = source_delta(REGISTERED_COMMIT, head)
require(index.get('applicability_delta') == delta, 'git diff/applicability_delta mismatch')
for path, names in ORDER_FUNCTIONS.items():
old = function_bytes(git_bytes('show', REGISTERED_COMMIT + ':' + path))
new = function_bytes(safe(ROOT, path).read_bytes())
new = function_bytes(git_bytes('show', head + ':' + path))
for name in names:
require(name in old and name in new and old[name] == new[name], 'stale ordering function: ' + path + ':' + name)
bindings = index.get('delta_cases')
@@ -227,7 +398,7 @@ def applicability(index, spec, source, head):
require(case_id == declaration.get('case_id'), 'boundary delta engine case: ' + key)
def coverage(index, raw, source):
def coverage(index, raw, source, product_tree):
authority = index['authority']
for ref in {r for _, refs, _ in REQUIRED.values() for r in refs}:
entry = authority.get(ref, {})
@@ -294,7 +465,7 @@ def coverage(index, raw, source):
require(ref.get('path') in source and ref.get('sha256') == source[ref['path']], 'reachability source pin')
identity = (ref['path'], ref.get('function'), ref.get('anchor_sha256'), ref.get('line_count'))
require(identity in REACH_REFS[name], 'unknown reachability anchor')
resolve_anchor(safe(ROOT, ref['path']).read_bytes(), *identity[1:])
resolve_anchor(git_bytes('show', product_tree + ':' + ref['path']), *identity[1:])
require(len(refs) == len(REACH_REFS[name]) and set(REACH_REFS[name]) == {
(r['path'], r['function'], r['anchor_sha256'], r['line_count']) for r in refs},
'missing static reachability chain: ' + name)
@@ -646,10 +817,10 @@ def verify(directory):
index = load(safe(evidence, 'index.json'))
require(type(index.get('schema_version')) is int and index['schema_version'] == 1, 'schema_version')
require(index.get('task_id') == 'P04-01' and index.get('formal_acceptance') is False, 'task/acceptance')
head = subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=ROOT, text=True).strip()
require(index.get('candidate_commit') == head, 'candidate_commit mismatch')
checker_commit = git_bytes('rev-parse', 'HEAD').decode().strip()
require(index.get('candidate_commit') == checker_commit, 'checker invocation commit mismatch')
head, target_result = verify_product_target(index, directory, checker_commit)
source = index.get('source_map')
pins(source, ROOT)
applicability(index, spec, source, head)
oldmap = load(ROOT / 'docs/plans/original-completion-2026-10-02/execution/p04-01-input-r01/historical-source-map.json')
required = {p for p in oldmap if p.endswith('.gd') and p.startswith(('autoload/', 'scripts/'))}
@@ -719,15 +890,17 @@ def verify(directory):
raise ValueError('unsupported predicate: ' + str(op))
require(result, name + ': predicate failed')
require(index.get('review_status') in {'pending_review', 'approved'}, 'review status')
coverage(index, raw, source)
coverage(index, raw, source, head)
# Scalar conditions remain necessary declarations, never sufficient proof.
binding = index.get('native_bundle')
require(isinstance(binding, dict), 'native bundle required; self-reported scalar evidence is insufficient')
manifest = safe(evidence, binding['manifest'])
manifest = supplied_file(directory, binding)
require(manifest.name == 'OWNERSHIP_RAW_PROVENANCE.json' and sha(manifest) == binding['sha256'],
'native bundle provenance pin')
native = verify_native(manifest.parent)
require(native['production_commit'] == head, 'native production/tested mismatch; A/B reuse coverage required')
require({safe(manifest.parent, run + '/runner-input.json') for run in NATIVE_RUNS}
<= target_result['runner_paths'], 'C runner inputs absent from complete product binding')
require(native['production_commit'] == target_result['tested_product'], 'native production role mismatch')
# This adapter covers only the supplied dependency/formal series. Never
# promote it to the missing original scenarios through scalar booleans.
missing = (set(REQUIRED) | {'EV'}) - set(native['verified_conditions'])
@@ -741,7 +914,8 @@ def main():
parser.parse_args()
try:
require(bool(os.environ.get('BUILD_OUTPUT_DIR')), 'BUILD_OUTPUT_DIR required')
verify(Path(os.environ['BUILD_OUTPUT_DIR']))
with isolated_objects():
verify(Path(os.environ['BUILD_OUTPUT_DIR']).resolve())
except (ValueError, OSError, KeyError, TypeError, IndexError, AttributeError, subprocess.SubprocessError) as exc:
print(json.dumps({'task_id': 'P04-01', 'machine_verified': False, 'formal_acceptance': False, 'error': str(exc)}, ensure_ascii=False))
return 1