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

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

@@ -17,7 +17,7 @@ MANIFEST_SHA256 = 'a15f29fa12537f6d8aaf7cabea390922a9f16e3eb9c10981e5742aafc8c4c
 MANIFEST_BYTES = 1244135
 PRODUCT = '5f6bc11d6bb6e92d22d787989b219fef3a435bc1'
 RUNS = {'s1': 'p04-s1-native-r02', 's2a': 'p04-s2a-native-r03',
-        's2b': 'p04-s2b-native-r02', 's3': 'p04-s3-native-r01', 's4': 'p04-s4-native-r24'}
+        's2b': 'p04-s2b-native-r02', 's3': 'p04-s3-native-r01', 's4': 'p04-s4-native-r26'}
 # Native descriptors pinned from literal files, never ROOT result passed fields.
 NATIVE_PINS = {
     's1': ('1e26d9585396da9415747d87d77b70f23425a3ee0714e4fca114748f9479fd4e', '01372beb0ffec181c892d3b4e88bce32001e08fb2007938d7bf19135511a7acc', '0b6d11eb34d8af64b22cb806caeb06eed4ec521fa9ee24d1204ea185345e655c'),
@@ -1023,11 +1023,29 @@ class S4Series(Series):
             require(path.is_relative_to(self.build), 'S4 material outside own B')
             self.read_build(path.relative_to(self.build).as_posix(), pin)
         verify_s4_engine(inputs['engine'], execution['engine_identity'])
-        require(inputs['engine_identity'] == execution['engine_identity'], 'S4 input/execution engine identity')
+        initial = inputs['engine_identity_initial']
+        require(inputs['engine_identity_before'] == inputs['engine'] == initial['before'] ==
+                execution['engine_identity']['before'], 'S4 immutable initial engine descriptor')
+        require(set(initial) == {'before', 'during', 'after'} and initial['after'] is None and
+                isinstance(initial['during'], list) and
+                initial['during'] == execution['engine_identity']['during'][:len(initial['during'])],
+                'S4 initial observations are an immutable prefix')
         observations = execution['engine_observations']
-        require(any(row['stage'] == 'after-cleanup' for row in observations) and
+        stages = [row['stage'] for row in observations]
-                all(row.get('descriptor') == inputs['engine'] for row in observations),
+        required = ['before-import', 'before-parse', 'before-writer', 'before-reader',
-                'S4 runtime engine observations')
+                    'after-collection', 'after-cleanup', 'after-output-collection']
+        require(all(stages.count(stage) == 1 for stage in required) and
+                [stages.index(stage) for stage in required] == sorted(stages.index(stage) for stage in required) and
+                all(row.get('descriptor') == inputs['engine'] and 'error' not in row for row in observations),
+                'S4 ordered runtime engine observations')
+        for phase in ('import', 'parse', 'writer', 'reader'):
+            index = stages.index('before-' + phase)
+            require(loads(self.read('engine-identity-before-' + phase + '.json')) ==
+                    dict(observations[index], observation_index=index), 'S4 immutable per-phase engine receipt')
+        require(loads(self.read('engine-identity.json')) ==
+                dict(observations=observations, identity=execution['engine_identity']), 'S4 final engine receipt')
+        require(observations[stages.index('after-cleanup')]['descriptor'] == inputs['engine'],
+                'S4 exact after-cleanup engine descriptor')
         require(execution['error'] is None and execution['guard_errors'] == [] and execution['cleanup_errors'] == [] and
                 [p['phase'] for p in execution['phases']] == ['import', 'parse', 'writer', 'reader'], 'S4 native phase/guard failure')
         for phase in execution['phases']:

docs/plans/original-completion-2026-10-02/execution/p04-01/s4/S4_ROOT_RUN.md

@@ -14,20 +14,24 @@ provenance only. Never open their original production locations.
 export B=/Users/korizatou114/Projects/TheTree-managed-worktrees/p04-01/.runtime/P04-01/build
 export E=docs/plans/original-completion-2026-10-02/execution/p04-01
 export PYTHONDONTWRITEBYTECODE=1
-mkdir "$B/audit-support/p04-s4-source-r24"
+python3 "$E/s4/engine_identity_negative_fixtures.py"
-cp "$E/s4/run_s4_native.py" "$E/s4/p04_s4_power_probe.gd" "$E/s4/S4_ROOT_RUN.md" "$B/audit-support/p04-s4-source-r24/"
+python3 "$E/s4/receipt_lifecycle_fixtures.py"
-python3 "$B/audit-support/p04-s4-source-r24/run_s4_native.py" \
+mkdir "$B/audit-support/p04-s4-source-r26"
+cp "$E/s4/run_s4_native.py" "$E/s4/p04_s4_power_probe.gd" "$E/s4/S4_ROOT_RUN.md" "$B/audit-support/p04-s4-source-r26/"
+python3 "$B/audit-support/p04-s4-source-r26/run_s4_native.py" \
   --project-root "$B/audit-support/p04-r17-full-inputs-r01/product" \
-  --output "$B/p04-s4-native-r24" --plan "$B/p04-s4-plan-r24.json" \
+  --output "$B/p04-s4-native-r26" --plan "$B/p04-s4-plan-r26.json" \
   --godot /Applications/Godot.app/Contents/MacOS/Godot --check-only
-python3 "$B/audit-support/p04-s4-source-r24/run_s4_native.py" \
+python3 "$B/audit-support/p04-s4-source-r26/run_s4_native.py" \
   --project-root "$B/audit-support/p04-r17-full-inputs-r01/product" \
-  --output "$B/p04-s4-native-r24" --plan "$B/p04-s4-plan-r24.json" \
+  --output "$B/p04-s4-native-r26" --plan "$B/p04-s4-plan-r26.json" \
   --godot /Applications/Godot.app/Contents/MacOS/Godot
 ```
 
-The first invocation only prepares the pinned plan: zero engine launches. The
+Root must stop if either fixture or check-only fails; the commands are not an
-second is one collection, exactly four sequential owned processes: import,
+unconditional batch. Both fixtures start zero engines. The check-only invocation
+only prepares the pinned plan: zero engine launches. The final invocation
+is one collection, exactly four sequential owned processes: import,
 parse, Writer, Reader. Each retains the framework's native diagnostics,
 ownership, exit/end checks, full stream SHA, default-player protection, private
 profiles, source-before/after guards and cleanup. Failure stops the collection;
@@ -37,7 +41,7 @@ inputs. Reader uses only the existing IO1 formal package mapping.
 
 Exact input footprints:
 
-- Current wrapper SHA256 is recorded in `../REPAIR24_STATUS.md`. The plan pins
+- Historical wrapper hashes do not pin this repair. The fresh plan pins
   its exact bytes and Stage R checks them against the current source checkout.
 - `B/audit-support/p04-r17-full-inputs-r01/MANIFEST.json`, SHA256
   `a15f29fa12537f6d8aaf7cabea390922a9f16e3eb9c10981e5742aafc8c4c8a8`.
@@ -48,7 +52,8 @@ Exact input footprints:
 - The three staged S4 files above, pinned by the plan, and the installed executable
   `/Applications/Godot.app/Contents/MacOS/Godot`, pinned directly against the
   constant SHA before launch, during collection, and explicitly after cleanup.
-  Before/during/after descriptors are retained in runner-input and execution. The only stage override
+  Runner-input retains an immutable initial descriptor and observation prefix;
+  execution and engine-identity retain the final before/during/after identity. The only stage override
   is `tests/s3/test_runtime_staff_assignment.gd` with the S4 probe. Harness and
   kitchen fixture are the unchanged product files; inherited sources are
   included in the full1338 map. The parse roster is all316 `.gd` inputs.
@@ -64,8 +69,8 @@ Exact input footprints:
   proof and remains ineligible. Its diagnostics are at
   `audit-support/p04-root-d393-s4-r01/ENGINE_STAGING_DIAGNOSTICS.json`.
 
-Exact output footprints are `B/p04-s4-plan-r24.json` and the fresh
+Exact output footprints are `B/p04-s4-plan-r26.json` and the fresh
-`B/p04-s4-native-r24/**`: private selected source stage, complete phase receipts
+`B/p04-s4-native-r26/**`: private selected source stage, complete phase receipts
 and logs, raw JSON/Variant points and all phase observations, full source maps,
 PID and cleanup receipts. `formal/{writer,reader}/S4-{boundary,held,recovered}/`
 mirrors all four formal members (`journal.log`, `manifest.json`, `thumbnail.png`,
@@ -107,8 +112,115 @@ restore and is not part of Runtime snapshot equality. The derived
 `related_facility_operating` flag is measured separately and checked after restore.
 
 This source-only dispatch authorizes no worker execution. Root stages the three
-files, creates the r24 plan with --check-only, then collects once to the fresh
+files, creates the r26 plan with --check-only, then collects once to the fresh
-r24 output. Do not rerun S1-S3, C or full178. No engine copy, signing, xattr,
+r26 output. Do not read or write r24 outputs, cleanup, stages or plans.
+Do not rerun S1-S3, C or full178. No engine copy, signing, xattr,
 codesign call or security-setting change is part of this procedure.
 Native evidence, registered validation, independent review, support53cb rebind
 and formal acceptance remain pending. D05 remains pending.
+
+## REPAIR26 receipt lifecycle audit (static; NOT RUN)
+
+All names below are relative to the fresh output unless marked otherwise.
+`r.write` is the unchanged exclusive `open('x')` writer in
+`tools/run_all_tests.py`. No retry, overwrite, replace or atomic rename is used
+for a JSON proof. A write attempt may fail or leave a partial file; it is never
+retried. “At most once” describes wrapper control flow, not a durability claim.
+The initial input is serialized once with `engine=engine_descriptor`,
+`engine_identity_before=engine_descriptor` and a deep-copied initial identity.
+`repin()` itself is unchanged (it returns None); each per-phase receipt uses
+the descriptor it just appended and the index of that exact observation.
+
+| Writer / destination | Maximum attempts per run / uniqueness | Success and failure reachability |
+| --- | --- | --- |
+| `r.write(args.plan)` outside output | 1, check-only branch returns immediately | Native branch only reads the exact plan; existing plan rejected in check-only |
+| `runner-started.json` | 1 | First write inside protected setup; exception enters cleanup |
+| `source_before.json` | 1 | Setup before source copy; no fallback write |
+| `selected-source-map.json` | 1 | Setup before source copy; no fallback write |
+| `r.copy_source` to `stage/<source name>` | 1 per unique source-map key | Fresh stage; loop stops on setup failure; source files are not proof receipts |
+| `write_bytes` to `bootstrap/<name>` | 1 per pinned bootstrap key | The reviewed map names `profile_guard.gd` and `parse_full_dependencies.gd`; setup only |
+| `/bin/cp -c` to `stage/.godot/<name>` | 1 per cache inventory key | Optional eligible-cache branch only; fresh mode never accesses cache |
+| `cache-reuse.json` | 0 or 1 | Only after all optional cache clones and guards |
+| `parse-paths.json` | 1 | Setup after 316-script guard |
+| `contracts.json` | 1 | Setup after unchanged suite-policy guard |
+| `runner-input.json` | 1 | Setup; no final read/rewrite; absent if earlier setup fails |
+| `engine-identity-before-<phase>.json` | 1 each for import, parse, writer, reader | Immediately after that phase's repin; failure stops loop, never resumes it |
+| `write_bytes` to `writer-combined-raw.log`, `reader-combined-raw.log` | 1 each | Binary copy immediately after returned phase row, even a failed row if it has a log; later phases never reuse names |
+| `<phase>-actual-phase.json` | 1 each for four unique phases | Before `require(row['passed'])`; mid-loop failure retains its row and stops later phases |
+| `outer-owned-finally.json` | 1 | Single finally after setup/loop success or failure; cleanup called once; cleanup exception recorded; receipt write exception becomes guard error |
+| `source_after.json` | 0 or 1 | Single post-cleanup guarded block, after fixed_guard/source_refs succeed; later guard failure does not re-enter or retry the write |
+| `shutil.copyfile` to `formal/<role>/<slot>/<member>` | 1 per tuple | Only preliminary pass; two unique roles, unique slot map keys, four unique member keys, fresh target directory; exception stops collection |
+| `formal-relocation.json` | 0 or 1 | Only after successful complete formal mirror |
+| `raw-inventory.json` | 0 or 1 | Once after mirror branch, including setup/phase/cleanup failure; earlier output-collection exception can skip it |
+| `PID-set.json` | 0 or 1 | Once after raw inventory; earlier output-collection exception can skip it |
+| `engine-identity.json` | 1 | One final attempt after cleanup and both after-cleanup/after-output-collection repin attempts, including caught setup/phase/cleanup/collection failures; own write failure records guard error and makes both passed fields false |
+| `execution.json` | 1 | After identity attempt, with final observations, identity and guard errors |
+| `runner-ended.json` | 1 | After execution; no retry if final receipt storage itself fails |
+
+The wrapper has no `write_text` call. The unchanged `execute_phase` dependency
+also writes the following. Its shared text log is an explicit exception to
+write-once naming, and is not presented as two independent phase proofs.
+
+| Dependency destination | Writer and lifecycle |
+| --- | --- |
+| `<phase>-startup-profile-before.json` | Exclusive `write`, once per distinct phase before engine start |
+| `<phase>-profile.json` | Guard report path for parse/writer/reader, one per phase; actual Guard checks file absence before writing |
+| `starters/<phase>.gd` | `write_text`, once each for writer/reader |
+| `<phase>/combined-raw.log` | `write_text`, once each for import/parse |
+| `tests__s3__test_runtime_staff_assignment.gd.log` (general `out/<test>.log`) | Both writer and reader call `write_text` on this same per-test name; reader replaces the shared text. Wrapper preserves separate binary copies as `writer-combined-raw.log` and `reader-combined-raw.log`, retaining `strict_runner_log_path` and the original log hash |
+| `<phase>-receipt.json` | Exclusive `write`, once per phase after its finally and guards |
+| `stage/project.godot` | `write_bytes` for private startup settings and restoration in each phase finally; mutable private source, not immutable proof |
+| `<phase>/**`, private profiles, raw QA outputs | Delegated owned-native capture/engine/QA outputs; not wrapper `r.write`/`write_bytes`/`write_text` calls; their existing ownership/diagnostic/formal guards remain unchanged |
+
+Path coverage: success traverses each distinct phase once. A failed parse row is
+recorded, then `require(row['passed'])` jumps to the single finally; no writer or
+reader names are used. A protected setup failure also enters that finally, then
+attempts post-cleanup guards and final identity/execution/end receipts. Cleanup
+exceptions are caught separately, set `cleanup_errors`, and do not prevent final
+receipts. A post-cleanup fixed/source guard failure is caught without retrying
+`source_after`; formal/output-collection exceptions now become guard errors and
+still reach the final repin and identity receipt. A final identity write failure
+is caught without touching existing bytes and forces failure in execution/end.
+
+Admission failures before the protected setup (invalid plan/product/engine,
+live engine, insufficient disk, existing output, directory creation failure)
+remain fail-closed and do not promise final receipts. They have no repeated
+wrapper proof write. Persistent storage failure writing execution/end can still
+prevent a complete terminal receipt pair; no code can guarantee receipt storage
+when that storage fails. Such runs cannot provide accepted evidence.
+
+## Fixture scope and remaining gaps
+
+`receipt_lifecycle_fixtures.py` imports the real wrapper and standard writer.
+It builds 1338 synthetic source entries including 316 scripts in resolved fresh
+temporary paths; engine identity, template_guard, disk/default-player, helpers
+and phase execution are mocked. Four distinct fake PIDs and complete synthetic
+formal slot members exercise wrapper guards and copying without starting an
+engine. The actual identity section of `S4Series.__init__` is compiled from its
+AST, and actual `verify_s4_engine` checks a non-executable temporary data file
+under fixture-only patched constants. Other verifier sections are not claimed
+as covered by this synthetic fixture.
+
+Cases cover success after check-only, parse failure, cleanup exception, setup
+copy failure, post-cleanup collection failure and final identity write failure.
+Every exclusive write attempt is counted. Input bytes are compared with their
+creation bytes before negative mutation. Fresh-output refusal, each missing or
+tampered phase receipt and mismatched initial-before are negative checks. The
+fixture's deliberate proof mutations are confined to its disposable temporary
+files. No registered plan/output/stage is read or changed.
+
+Reasoning only about the 7af6 wrapper: import writes `engine-identity.json` via
+exclusive create. Before parse, the same name raises FileExistsError, so the
+four-phase success assertion cannot pass. After cleanup the old final write
+tries that same name again and escapes main, preventing final execution/end.
+Even if that collision alone were removed, the old runner-input final rewrite
+would hit the exclusive writer, and the required initial/per-phase receipts
+are absent. No 7af6 execution was performed to establish this reasoning.
+
+Worker performed source inspection and file edits only: no fixtures, check-only,
+tests, Git, engines, Runner, agents or commits. Root still must run both fixtures
+and check-only before any native r26 collection, then exact-version verification
+and independent review. The registered validation gate remains unchanged at
+7af6, exit 2, fresh MANIFEST missing; this is the supplied gate record, not a new
+worker run. Native/product proof, full formal checks, D05 and original P04-01
+acceptance remain pending. This repair is submitted preparation, not completion.

docs/plans/original-completion-2026-10-02/execution/p04-01/s4/engine_identity_negative_fixtures.py

@@ -33,7 +33,7 @@ def run():
 
     # Same bytes/SHA cannot admit an own-B bare copy or another outside-B path.
     for label, path in [
-        ("copied bare engine in B with same SHA", s4.BUILD / "audit-support/p04-s4-source-r24/Godot"),
+        ("copied bare engine in B with same SHA", s4.BUILD / "audit-support/p04-s4-source-r26/Godot"),
         ("other outside-B path", Path("/Applications/OtherGodot.app/Contents/MacOS/Godot")),
     ]:
         other = dict(pin, path=str(path))

docs/plans/original-completion-2026-10-02/execution/p04-01/s4/receipt_lifecycle_fixtures.py

@@ -0,0 +1,255 @@
+#!/usr/bin/env python3
+"""Root-only synthetic receipt lifecycle fixtures. No engine/process is started.
+
+Uses the real main, exclusive JSON writer, source_refs and inventories. The
+identity section is compiled directly from S4Series.__init__, not reimplemented.
+This is not native/product/ownership/formal acceptance evidence.
+"""
+import ast
+from collections import Counter
+from contextlib import ExitStack, redirect_stdout
+import copy
+import importlib.util
+import io
+import json
+from pathlib import Path
+import shutil
+import sys
+import tempfile
+from types import SimpleNamespace
+from unittest.mock import patch
+
+sys.dont_write_bytecode = True
+HERE = Path(__file__).resolve().parent
+ROOT = next(p for p in HERE.parents if (p / 'tools/run_all_tests.py').is_file())
+
+
+def load(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
+
+
+s4 = load('receipt_s4', HERE / 'run_s4_native.py')
+r = load('receipt_standard', ROOT / 'tools/run_all_tests.py')
+series = load('receipt_series', HERE.parent / 'p04_native_series.py')
+
+
+def identity_code():
+    tree = ast.parse((HERE.parent / 'p04_native_series.py').read_text())
+    cls = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == 'S4Series')
+    init = next(n for n in cls.body if isinstance(n, ast.FunctionDef) and n.name == '__init__')
+    start = next(i for i, n in enumerate(init.body) if isinstance(n, ast.Expr) and
+                 isinstance(n.value, ast.Call) and isinstance(n.value.func, ast.Name) and
+                 n.value.func.id == 'verify_s4_engine')
+    end = next(i for i in range(start + 1, len(init.body))
+               if "execution['error']" in ast.unparse(init.body[i]))
+    return compile(ast.Module(body=init.body[start:end], type_ignores=[]), '<real S4 identity section>', 'exec')
+
+
+def verify_identity(out, descriptor):
+    # The real engine verifier reads only the synthetic ordinary file here.
+    # Production constants/functions on disk are never changed.
+    with patch.object(series, 'S4_ENGINE_PATH', descriptor['path']), \
+            patch.object(series, 'S4_ENGINE_SHA256', descriptor['sha256']):
+        namespace = dict(vars(series), inputs=r.read(out / 'runner-input.json'),
+                         execution=r.read(out / 'execution.json'),
+                         self=SimpleNamespace(read=lambda name: (out / name).read_bytes()))
+        exec(identity_code(), namespace)
+
+
+def rejected(action):
+    try:
+        action()
+    except (ValueError, AssertionError, OSError, KeyError, TypeError):
+        return
+    raise AssertionError('negative fixture was accepted')
+
+
+def case(base, failure=None):
+    product = base / 'product'
+    support = base / 'support'
+    staged = base / 'source'
+    for path in (product, support, staged):
+        path.mkdir(parents=True)
+
+    def put(path, data):
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(data)
+
+    # Real 1338-entry source_refs and 316-entry script roster, synthetic bytes.
+    scripts = [s4.TEST, 'tests/transport/test_runtime_transport.gd',
+               'tests/staff/test_rescue_elevator.gd']
+    scripts += ['scripts/fixture_%03d.gd' % i for i in range(313)]
+    for name in scripts:
+        put(product / name, b'extends Node\n')
+    for name in ('project.godot', 'export_presets.cfg', 'Makefile', 'AGENTS.md'):
+        put(product / name, b'fixture\n')
+    put(product / 'tools/run_all_tests.py', (ROOT / 'tools/run_all_tests.py').read_bytes())
+    for i in range(1017):
+        put(product / ('data/fixture_%04d.txt' % i), b'fixture\n')
+    assert len(r.source_refs(product)) == 1338
+    qa = staged / 'p04_s4_power_probe.gd'
+    put(qa, b'extends Node\n# synthetic override\n')
+    put(staged / 'run_s4_native.py', (HERE / 'run_s4_native.py').read_bytes())
+    put(staged / 'S4_ROOT_RUN.md', b'synthetic fixture material\n')
+    contracts = base / 'contracts.json'
+    r.write(contracts, {})
+    material = base / 'material.json'
+    r.write(material, {})
+    r.write(support / 'suite-policy.json', {'suite_policies': {}})
+    r.write(support / 'bootstrap-scripts.json', {
+        'parse_full_dependencies.gd': {'text': 'fixture\n', 'sha256': r.sha(product / 'Makefile')}})
+    engine = base / 'synthetic-engine-data'
+    put(engine, b'not an executable; never launched\n')
+    st = engine.stat()
+    descriptor = dict(path=str(engine), resolved_path=str(engine), bytes=st.st_size,
+                      sha256=r.sha(engine), mtime_ns=st.st_mtime_ns, mode=st.st_mode,
+                      device=st.st_dev, inode=st.st_ino, ctime_ns=st.st_ctime_ns)
+    out = base / 'p04-s4-native-r26'
+    plan = base / 'p04-s4-plan-r26.json'
+    attempts, created, phases = Counter(), {}, []
+    cleanup_calls = []
+    real_write, real_inventory = r.write, r.inventory
+    protected = dict(path=str(base / 'default-player'), existed=False, files={}, logs_and_locks_included=True)
+
+    def write(path, value):
+        path = Path(path)
+        attempts[path] += 1
+        if failure == 'identity-write' and path == out / 'engine-identity.json':
+            raise OSError('injected final identity write error')
+        real_write(path, value)  # unchanged open('x'), including collisions
+        created[path] = path.read_bytes()
+
+    def finish(owned):
+        cleanup_calls.append(list(owned))
+        if failure == 'cleanup':
+            raise RuntimeError('injected cleanup error')
+        return [dict(pid=p.pid, engine_ended=True) for p in owned], []
+
+    outer = SimpleNamespace(native=SimpleNamespace(helper={'live_engines': lambda: []}),
+                            OWNED=[SimpleNamespace(pid=91001+i) for i in range(4)],
+                            finish_captured=finish, run_phase=lambda *a, **k: None)
+
+    def execute(engine_arg, stage, output, phase, test, selected, expected, previous,
+                policy, contracts_arg, paths, public, player, root, *helpers):
+        phases.append(phase)
+        assert engine_arg == str(engine) and output == out and player == protected
+        assert r.source_refs(root) == selected['root_before']
+        assert r.inventory(stage, stage=True) == dict(expected, **previous)
+        pid = outer.OWNED[len(phases)-1].pid
+        row = dict(phase=phase, passed=not (failure == 'parse' and phase == 'parse'), actual_pid=pid)
+        if test:
+            slots = {}
+            for slot in s4.SLOTS:
+                directory = base / 'profiles' / phase / slot
+                for member in ('journal.log', 'manifest.json', 'thumbnail.png', 'world.bin'):
+                    put(directory / member, (slot + '/' + member).encode())
+                slots[slot] = str(directory)
+            (out / 'raw').mkdir(exist_ok=True)
+            write(out / ('raw/' + phase + '.json'), dict(checks=1, failures=0, branches=['S4'],
+                  slots=slots, pid=pid, profile=str(base / 'profiles' / phase)))
+            # Reproduce the runner's shared text log and let main copy its bytes.
+            log = test.replace('/', '__') + '.log'
+            (out / log).write_text(phase + '\n')
+            row['log'] = log
+        return row
+
+    def inventory(path, **kwargs):
+        if failure == 'output' and Path(path) == out / 'raw':
+            raise OSError('injected post-cleanup inventory error')
+        return real_inventory(path, **kwargs)
+
+    def copy_source(source, target):
+        if failure == 'setup':
+            raise OSError('injected setup copy error')
+        shutil.copyfile(source, target)
+
+    def modules(name, path):
+        if name == 's4_standard_runner':
+            return r
+        if name == 's4_owned_outer':
+            return outer
+        if name == 's4_markers':
+            return SimpleNamespace(load_contracts=lambda path: {})
+        if name in ('s4_diagnostics', 's4_owners', 's4_audio'):
+            return SimpleNamespace()
+        raise AssertionError('unexpected helper import: ' + name)
+
+    with ExitStack() as stack:
+        for obj, values in [(s4, dict(BUILD=base, PRODUCT=product, HERE=staged, QA=qa,
+                ENGINE=engine, CONTRACTS=contracts, MATERIALS=[material],
+                __file__=str(staged / 'run_s4_native.py'), module=modules)),
+                (r, dict(SUPPORT=support, write=write, execute_phase=execute,
+                         copy_source=copy_source, inventory=inventory))]:
+            for key, value in values.items():
+                stack.enter_context(patch.object(obj, key, value))
+        pin_mock = stack.enter_context(patch.object(s4, 'engine_pin', return_value=descriptor))
+        guard_mock = stack.enter_context(patch.object(s4, 'template_guard', return_value=None))
+        player_mock = stack.enter_context(patch.object(r, 'default_player', return_value=protected))
+        disk_mock = stack.enter_context(patch.object(s4.shutil, 'disk_usage',
+                          return_value=SimpleNamespace(free=10 * 1024**3)))
+        stack.enter_context(redirect_stdout(io.StringIO()))
+        argv = ['run_s4_native.py', '--project-root', str(product), '--output', str(out),
+                '--plan', str(plan), '--godot', str(engine)]
+        with patch.object(sys, 'argv', argv + ['--check-only']):
+            assert s4.main() == 0
+        assert not phases and not cleanup_calls and not out.exists()
+        with patch.object(sys, 'argv', argv):
+            result = s4.main()
+            rejected(s4.main)  # same output must be rejected before any new writes
+        assert pin_mock.called and guard_mock.called and player_mock.called and disk_mock.called
+
+    assert result == (0 if failure is None else 1)
+    assert len(cleanup_calls) == 1
+    assert all(count == 1 for count in attempts.values()), attempts
+    for leaf in ('outer-owned-finally.json', 'engine-identity.json', 'execution.json', 'runner-ended.json'):
+        assert attempts[out / leaf] == 1
+        if failure != 'identity-write' or leaf != 'engine-identity.json':
+            assert (out / leaf).is_file()
+    if failure != 'setup':
+        assert (out / 'runner-input.json').read_bytes() == created[out / 'runner-input.json']
+        assert r.read(out / 'runner-input.json')['engine_identity_initial']['after'] is None
+    execution = r.read(out / 'execution.json')
+    assert execution['passed'] is (failure is None)
+    assert 'FileExistsError' not in json.dumps(execution)
+    assert phases == ([] if failure == 'setup' else ['import', 'parse'] if failure == 'parse'
+                      else ['import', 'parse', 'writer', 'reader'])
+    if failure == 'cleanup':
+        assert execution['cleanup_errors'] == ['injected cleanup error']
+    if failure in ('output', 'identity-write'):
+        assert execution['guard_errors']
+    if failure is None:
+        assert len(list(out.glob('engine-identity*.json'))) == 5
+        verify_identity(out, descriptor)
+        assert (out / 'writer-combined-raw.log').read_bytes() == b'writer\n'
+        assert (out / 'reader-combined-raw.log').read_bytes() == b'reader\n'
+        # Mutations below affect only this disposable fixture's synthetic proofs.
+        for phase in ('import', 'parse', 'writer', 'reader'):
+            path = out / ('engine-identity-before-' + phase + '.json')
+            original = path.read_bytes()
+            path.write_text('{}')
+            rejected(lambda: verify_identity(out, descriptor))
+            path.unlink()
+            rejected(lambda: verify_identity(out, descriptor))
+            path.write_bytes(original)
+        path = out / 'runner-input.json'
+        initial = r.read(path)
+        changed = copy.deepcopy(initial)
+        changed['engine_identity_initial']['before']['inode'] += 1
+        path.write_text(json.dumps(changed))
+        rejected(lambda: verify_identity(out, descriptor))
+        path.write_bytes(created[path])
+        verify_identity(out, descriptor)
+    return dict(case=failure or 'success', passed=True, native_engine_launches=0)
+
+
+def run():
+    with tempfile.TemporaryDirectory(prefix='s4-receipt-fixtures-') as temporary:
+        return [case(Path(temporary).resolve() / (failure or 'success'), failure)
+                for failure in (None, 'parse', 'cleanup', 'setup', 'output', 'identity-write')]
+
+
+if __name__ == '__main__':
+    print(json.dumps(dict(results=run(), formal_acceptance=False), indent=2))

docs/plans/original-completion-2026-10-02/execution/p04-01/s4/run_s4_native.py

@@ -2,6 +2,7 @@
 """S4-only native entry. --check-only never loads ownership helpers or starts Godot."""
 from pathlib import Path
 import argparse
+import copy
 import hashlib
 import importlib.util
 import json
@@ -98,8 +99,8 @@ def engine_pin(path, expected=None):
 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-s4-native-r24")
+    parser.add_argument("--output", type=Path, default=BUILD / "p04-s4-native-r26")
-    parser.add_argument("--plan", type=Path, default=BUILD / "p04-s4-plan-r24.json")
+    parser.add_argument("--plan", type=Path, default=BUILD / "p04-s4-plan-r26.json")
     parser.add_argument("--cache-run", type=Path, default=None,
                         help="optional passed registered cache; default is fresh, without cache access")
     parser.add_argument("--godot", type=Path, default=ENGINE)
@@ -222,9 +223,9 @@ def main():
             env.update(THE_TREE_S4_EVIDENCE=str(out / "raw"), THE_TREE_S4_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 / "runner-started.json", dict(started=r.now(), owner_runner_pid=os.getpid(),
+                plan=pin(args.plan), engines_launched=0))
         r.write(out / "source_before.json", source_before)
         r.write(out / "selected-source-map.json", refs)
         # Private source stage first, then literal reviewed bootstrap.
@@ -262,7 +263,9 @@ def main():
         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), engine=engine_descriptor, engine_identity=engine_identity, runner_sha256=r.sha(root / "tools/run_all_tests.py"),
+                wrapper_sha256=r.sha(__file__), QA=pin(QA), engine=engine_descriptor,
+                engine_identity_before=engine_descriptor, engine_identity_initial=copy.deepcopy(engine_identity),
+                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("s4_diagnostics", r.SUPPORT / "diagnostic_gate.py")
@@ -278,12 +281,15 @@ def main():
                 r.require(not (stage / ".godot").exists() and not (stage / ".godot").is_symlink(),
                           "fresh private .godot absent at actual import")
             repin("before-" + phase)
-            r.write(out / "engine-identity.json", engine_observations)
+            r.write(out / ("engine-identity-before-" + phase + ".json"),
+                    dict(engine_observations[-1], observation_index=len(engine_observations)-1))
             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.
+            # execute_phase writer/reader share out/<test>.log via write_text.
+            # Keep binary per-phase copies under these existing unique names.
+            # The shared runner log is not write-once; see S4_ROOT_RUN.md.
             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"]
@@ -349,36 +355,43 @@ def main():
     # IO1 has already copied all four members into a separate Reader profile.
     # Mirror those exact complete directories to own B for read-only R verification.
     # Original profile paths are provenance after this step, never verifier inputs.
-    if passed:
+    try:
-        relocated = {}
+        if passed:
-        for role in ("writer", "reader"):
+            relocated = {}
-            receipt = r.read(out / ("raw/" + role + ".json"))
+            for role in ("writer", "reader"):
-            relocated[role] = {}
+                receipt = r.read(out / ("raw/" + role + ".json"))
-            for slot, original in receipt["slots"].items():
+                relocated[role] = {}
-                target = out / "formal" / role / slot
+                for slot, original in receipt["slots"].items():
-                target.mkdir(parents=True)
+                    target = out / "formal" / role / slot
-                members = r.inventory(Path(original))
+                    target.mkdir(parents=True)
-                r.require(set(members) == {"journal.log", "manifest.json", "thumbnail.png", "world.bin"}, "all formal members")
+                    members = r.inventory(Path(original))
-                for member in members:
+                    r.require(set(members) == {"journal.log", "manifest.json", "thumbnail.png", "world.bin"}, "all formal members")
-                    shutil.copyfile(Path(original) / member, target / member)
+                    for member in members:
-                r.require(r.inventory(target) == members == formal_before[slot], "whole formal mirror")
+                        shutil.copyfile(Path(original) / member, target / member)
-                relocated[role][slot] = dict(provenance=original, relative=str(target.relative_to(out)), members=members)
+                    r.require(r.inventory(target) == members == formal_before[slot], "whole formal mirror")
-        r.write(out / "formal-relocation.json", relocated)
+                    relocated[role][slot] = dict(provenance=original, relative=str(target.relative_to(out)), members=members)
-    r.write(out / "raw-inventory.json", r.inventory(out / "raw"))
+            r.write(out / "formal-relocation.json", relocated)
-    r.write(out / "PID-set.json", dict(owner_runner_pid=os.getpid(), engine_PIDs=pids))
+        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))
+    except BaseException as caught:
+        guards.append("output collection: " + type(caught).__name__ + ": " + str(caught))
+        passed = False
+        execution["passed"] = False
     try:
         repin("after-output-collection")
     except BaseException as caught:
         guards.append(str(caught))
         passed = False
         execution["passed"] = False
-    r.write(out / "engine-identity.json", engine_observations)
+    try:
+        r.write(out / "engine-identity.json",
+                dict(observations=engine_observations, identity=engine_identity))
+    except BaseException as caught:
+        guards.append("engine identity receipt: " + type(caught).__name__ + ": " + str(caught))
+        passed = False
+        execution["passed"] = False
     execution["engine_observations"] = engine_observations
     execution["engine_identity"] = engine_identity
-    if (out / "runner-input.json").is_file():
-        final_inputs = r.read(out / "runner-input.json")
-        final_inputs["engine_identity"] = engine_identity
-        r.write(out / "runner-input.json", final_inputs)
     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))))