Reference flat disk files without losing host FAT snapshot state

JSON_RPC_API.md

@@ -871,13 +871,30 @@ Both methods require a paused machine and `expected_state_revision` equal to
 `session.status.state_revision`. Paths name files on the emulator host.
 
 - `machine.snapshot.export`: `path` must be a new file. Optional `disk_mode`
-  is `auto` (default), `embed`, or `reference`. Returns archive `sha256` and
-  current `state_revision`. The ZIP contains JSON plus hashed binary payloads.
-- `machine.snapshot.import`: `path`, new `disk_root` directory, optional archive
-  `sha256`, and optional `references` mapping string disk indices `0`/`1` to
+  is `auto` (default), `embed`, `reference`, or `reference-files`. Returns
+  `path`, archive `bytes`/`sha256` and current `state_revision`. The ZIP contains JSON plus hashed binary payloads.
+- `machine.snapshot.import`: `path`, new `disk_root` directory, required archive
+  legacy `sha256` or canonical `expected_sha256`, and optional `references` mapping string disk indices `0`/`1` to
   source image paths. Referenced images must match saved size and SHA-256.
   Returns the new revision and `machine_snapshot_restored` stop reason.
 
+`reference-files` stores flat-file disks as size/hash references while embedding
+host-mapped FAT disks completely, including their live image, host file contents,
+sync hashes and pending synchronization state. It does not reference arbitrary
+host directories. Legacy policies are unchanged: `reference` still refuses host
+mounts. Import copies each verified file reference into a new writable disk;
+subsequent guest writes cannot change the immutable reference.
+
+Every import requires one independently retained archive digest: canonical
+`expected_sha256` or legacy `sha256`,64 hexadecimal characters. New shared clients
+use the canonical name. Supplying both aliases is rejected. The former unguarded
+RPC import is intentionally no longer accepted; internal codec APIs are unchanged.
+Optional `preserve_breakpoints` is boolean, default `true`: retain permanent host
+breakpoint definitions and hit counters, clear transient predicates/operations
+and journals. `false` clears all breakpoints after successful installation.
+All new flags/hash fields are validated before reading/restoring the archive;
+refused requests retain the machine and existing debugger configuration.
+
 Import reconstructs CPU/RAM/ROM/device state, host RNG, disk bytes and host-FAT
 synchronization state. It remains paused; execution resumes only by a later
 execution request. Restored disks use new paths under `disk_root`. No existing

diskcodec.py

@@ -70,10 +70,12 @@ def _key(key):
 
 
 def dump_disk_state(disk, mode='auto', embed_limit=64*1024*1024):
-    if (mode not in ('auto', 'embed', 'reference')
+    if (mode not in ('auto', 'embed', 'reference', 'reference-files')
             or type(embed_limit) is not int or embed_limit < 0):
         raise ValueError('invalid disk storage policy')
     if type(disk) is str:
+        if mode == 'reference-files':
+            mode = 'reference'
         data = Path(disk).read_bytes()
         if mode == 'auto':
             if len(data) > embed_limit:
@@ -84,6 +86,8 @@ def dump_disk_state(disk, mode='auto', embed_limit=64*1024*1024):
                 {'image': data} if mode == 'embed' else {})
     if type(disk) is not HostDirectoryFAT16:
         raise ValueError('unsupported disk type')
+    if mode == 'reference-files':
+        mode = 'embed'
     if mode == 'reference':
         raise ValueError(
             'host mounts require embedding their live image and sync state')

docs/HANDOFF.md

@@ -1,3 +1,27 @@
+# Handoff - mandatory common snapshot guard, 2026-10-06
+
+Shared RPC imports now require exactly one valid retained expected_sha256 or
+legacy sha256. Missing, malformed, mismatched and ambiguous guards refuse before
+restore; old unguarded RPC callers intentionally fail. Internal codec APIs and
+historical evidence unchanged. Current lock/input test uses its export receipt.
+Five stdlib storage cases cover both aliases, invalid flags with a valid digest,
+full fresh-process continuation and refusal without machine/debugger/disk writes.
+Final-head local/public checks and storage review must pass before PR70 merge.
+
+# Handoff - mixed disk references and guarded import, 2026-10-06
+
+FINISHED local: reference-files references flat images while fully embedding
+host FAT image/files/sync state. Legacy policies unchanged; immutable references
+are verified then copied before guest writes. Canonical expected_sha256 import
+guard rejects malformed/mismatched hashes and both aliases; legacy sha256 stays.
+Explicit preserve_breakpoints defaults true with definitions/hits preserved;
+false clears all after successful restore. Exports add path/bytes metadata.
+Five stdlib tests pass on Windows CPython and bare PyPy, including complete
+fresh-process continuation, pending XTIDE write, host sync, immutable reference
+isolation, mismatch refusal before disk/state mutation and breakpoint policy.
+PR69 is rebase-integrated at c98c4af after all five final-head public CI checks.
+WIP: independent storage review/final-head CI, then parent pin/live-game replay.
+
 # Handoff - bare-PyPy full observation proof, 2026-10-06
 
 FINISHED: complete motherboard factory is shared in tests/machine_fixture.py,

main.py

@@ -631,20 +631,35 @@ def handle_debug(request):
             if not isinstance(path, str) or not path:
                 raise ValueError('path must be a nonempty host path')
             if method == 'machine.snapshot.export':
+                from pathlib import Path
                 digest = machine_snapshots.export(path, params.get('disk_mode', 'auto'))
-                return {'sha256': digest, 'state_revision': control['revision']}
+                return {'path': path, 'bytes': Path(path).stat().st_size,
+                        'sha256': digest, 'state_revision': control['revision']}
+            if 'expected_sha256' in params and 'sha256' in params:
+                raise ValueError('supply expected_sha256 or legacy sha256, not both')
+            expected_sha256 = params.get('expected_sha256', params.get('sha256'))
+            if (not isinstance(expected_sha256, str) or len(expected_sha256) != 64
+                    or any(c not in '0123456789abcdefABCDEF' for c in expected_sha256)):
+                raise ValueError('expected_sha256 or sha256 must be a 64-character hexadecimal digest')
+            expected_sha256 = expected_sha256.lower()
+            preserve = params.get('preserve_breakpoints', True)
+            if not isinstance(preserve, bool):
+                raise ValueError('preserve_breakpoints must be boolean')
             disk_root = params.get('disk_root')
             if not isinstance(disk_root, str) or not disk_root:
                 raise ValueError('disk_root must be a new host directory path')
             references = params.get('references', {})
             if (not isinstance(references, dict)
                     or any(k not in ('0', '1') or not isinstance(v, str) or not v for k,v in references.items())):
                 raise ValueError('references must map disk indices 0/1 to host paths')
-            machine_snapshots.restore(path, disk_root, params.get('sha256'),
+            machine_snapshots.restore(path, disk_root, expected_sha256,
                                       {int(k):v for k,v in references.items()})
             video_history.reset()
             p._io.SetVideoWriteHook(None, None)
             rpc_clear_run_until()
+            if not preserve:
+                for breakpoint in breakpoints.list():
+                    breakpoints.delete(breakpoint['breakpoint_id'])
             control['operations'].clear()
             control['active_operation'] = None
             control['snapshots'].clear()

tests/test_machine_snapshot_rpc.py

@@ -60,7 +60,7 @@ def test_rpc_refusal_preserves_machine(tmp_path, failure):
 def test_input_during_prepare_is_deferred_until_after_install(tmp_path, monkeypatch):
     h, cpu, vnc = rpc_machine(tmp_path)
     path = str(tmp_path/'saved.pypc')
-    h.rpc('machine.snapshot.export', path=path, expected_state_revision=0)
+    exported = h.rpc('machine.snapshot.export', path=path, expected_state_revision=0)
     old_queue = capture_machine(cpu)[0]['keyboard']['fields']['queue']
     attempted, delivered = threading.Event(), threading.Event()
     def input_thread():
@@ -75,7 +75,8 @@ def prepare(*args, **kwargs):
         assert not delivered.wait(0.05)
         return original(*args, **kwargs)
     monkeypatch.setattr(machinesnapshots, 'prepare_machine', prepare)
-    h.rpc('machine.snapshot.import', path=path, disk_root=str(tmp_path/'restore'), expected_state_revision=0)
+    h.rpc('machine.snapshot.import', path=path, disk_root=str(tmp_path/'restore'), expected_state_revision=0,
+          expected_sha256=exported['sha256'])
     producer.join(1)
     assert delivered.is_set()
     assert capture_machine(cpu)[0]['keyboard']['fields']['queue'] == old_queue + [0x30]

tests/test_snapshot_contract.py

@@ -0,0 +1,135 @@
+"""Mixed disk references and guarded restore on a complete machine, no pytest."""
+import copy
+import hashlib
+import json
+from pathlib import Path
+import random
+import subprocess
+import sys
+from tempfile import TemporaryDirectory
+import unittest
+
+from checkpointbundle import read_bundle, write_bundle
+from diskcodec import dump_disk_state
+from machinecodec import capture_machine, prepare_machine
+from tests.machine_fixture import rpc_machine, run
+from virtualfat16 import HostDirectoryFAT16
+
+
+class SnapshotContractTests(unittest.TestCase):
+    def setUp(self):
+        scratch = Path(__file__).resolve().parents[1] / '.test-tmp'
+        scratch.mkdir(exist_ok=True)
+        self.directory = TemporaryDirectory(dir=scratch)
+        self.addCleanup(self.directory.cleanup)
+        self.root = Path(self.directory.name)
+        self.h, self.cpu, _ = rpc_machine(self.root)
+        host = self.root / 'host'; host.mkdir()
+        (host / 'PYRO.DAT').write_bytes(b'original data')
+        controller = self.cpu._devices[4]
+        host_disk = HostDirectoryFAT16(host)
+        controller._disks.append(host_disk)
+        controller._serial_numbers.append(str(hash(str(host_disk))))
+        # Extend the fixture inventory while retaining its pending disk0 write.
+        self.reference = self.root / 'immutable.img'
+        self.reference.write_bytes(Path(self.cpu._devices[4]._disks[0]).read_bytes())
+
+    def export(self, name='saved.pypc'):
+        return self.h.rpc('machine.snapshot.export', path=str(self.root/name),
+                          expected_state_revision=0, disk_mode='reference-files')
+
+    def test_mixed_policy_preserves_complete_host_state_and_legacy_refusal(self):
+        saved = capture_machine(self.cpu, disk_mode='reference-files')
+        self.assertEqual([d['mode'] for d in saved[0]['disks']], ['reference','embed'])
+        self.assertNotIn('disk0/image', saved[1])
+        self.assertIn('disk1/image', saved[1])
+        self.assertEqual(dump_disk_state(self.cpu._devices[4]._disks[1], 'reference-files'),
+                         dump_disk_state(self.cpu._devices[4]._disks[1], 'embed'))
+        with self.assertRaisesRegex(ValueError, 'require embedding'):
+            capture_machine(self.cpu, disk_mode='reference')
+        expected = capture_machine(self.cpu)
+        restored, rng = prepare_machine(*saved, self.root/'restore', {0:str(self.reference)})
+        random.setstate(rng.getstate())
+        self.assertEqual(capture_machine(restored), expected)
+        self.assertNotEqual(restored._devices[4]._disks[0], str(self.reference))
+
+    def test_reference_mismatch_refuses_before_any_machine_or_disk_mutation(self):
+        exported = self.export()
+        before = capture_machine(self.cpu)
+        self.reference.write_bytes(b'x' * self.reference.stat().st_size)
+        with self.assertRaisesRegex(ValueError, 'size/hash'):
+            self.h.rpc('machine.snapshot.import', path=exported['path'],
+                       disk_root=str(self.root/'refused'), expected_state_revision=0,
+                       expected_sha256=exported['sha256'], references={'0':str(self.reference)})
+        self.assertEqual(capture_machine(self.cpu), before)
+        self.assertFalse((self.root/'refused').exists())
+        self.assertEqual(self.h.control['revision'], 0)
+
+    def test_hash_alias_and_breakpoint_flag_preflight(self):
+        exported = self.export()
+        before = capture_machine(self.cpu)
+        controls = copy.deepcopy(self.h.control)
+        invalid = [dict(), dict(sha256=None), dict(sha256=True),
+                   dict(sha256='wrong'), dict(expected_sha256=None), dict(expected_sha256=True),
+                   dict(expected_sha256='f'*63), dict(expected_sha256=' '*60+'abcd'),
+                   dict(expected_sha256='g'*64), dict(expected_sha256='0'*64),
+                   dict(expected_sha256=exported['sha256'], sha256=exported['sha256']),
+                   dict(preserve_breakpoints=1, expected_sha256=exported['sha256']),
+                   dict(preserve_breakpoints=None, expected_sha256=exported['sha256'])]
+        for params in invalid:
+            with self.subTest(params=params), self.assertRaises(ValueError):
+                self.h.rpc('machine.snapshot.import', path=exported['path'],
+                           disk_root=str(self.root/'refused'), expected_state_revision=0,
+                           references={'0':str(self.reference)}, **params)
+            self.assertEqual(capture_machine(self.cpu), before)
+            self.assertEqual(self.h.control, controls)
+            self.assertFalse((self.root/'refused').exists())
+
+    def test_preserved_definitions_and_hitcounts_or_explicit_clear(self):
+        exported = self.export()
+        self.assertEqual(exported['bytes'], Path(exported['path']).stat().st_size)
+        manager = self.h.namespace['breakpoints']
+        bp = self.h.rpc('breakpoints.create', address={'space':'segmented','segment':0x1000,'offset':0})
+        manager.check(0x1000, 0, {})
+        definitions = self.h.rpc('breakpoints.list')
+        self.assertEqual(definitions['breakpoints'][0]['hit_count'], 1)
+        self.h.rpc('machine.snapshot.import', path=exported['path'],
+                   disk_root=str(self.root/'first'), expected_state_revision=0,
+                   expected_sha256=exported['sha256'].upper(), references={'0':str(self.reference)})
+        self.assertEqual(self.h.rpc('breakpoints.list'), definitions)
+        self.h.rpc('machine.snapshot.import', path=exported['path'],
+                   disk_root=str(self.root/'second'), expected_state_revision=1,
+                   sha256=exported['sha256'], references={'0':str(self.reference)},
+                   preserve_breakpoints=False)
+        self.assertEqual(self.h.rpc('breakpoints.list'), {'breakpoints':[]})
+        self.assertFalse(self.h.control['breakpoints_active'])
+
+    def test_fresh_process_continuation_and_reference_write_isolation(self):
+        saved = capture_machine(self.cpu, disk_mode='reference-files')
+        checkpoint = self.root/'saved.pypc'; digest = write_bundle(checkpoint, *saved)
+        expected_trace = run(self.cpu)
+        expected = capture_machine(self.cpu)
+        reference_hash = hashlib.sha256(self.reference.read_bytes()).hexdigest()
+        child = '''import sys,random,json
+from checkpointbundle import read_bundle,write_bundle
+from machinecodec import prepare_machine,capture_machine
+from tests.machine_fixture import run
+cpu,rng=prepare_machine(*read_bundle(sys.argv[1],sys.argv[2]),sys.argv[3],{0:sys.argv[4]})
+random.setstate(rng.getstate())
+trace=run(cpu)
+write_bundle(sys.argv[5],*capture_machine(cpu))
+open(sys.argv[6],'w').write(json.dumps(trace))
+'''
+        out, trace = self.root/'after.pypc', self.root/'trace.json'
+        result = subprocess.run([sys.executable,'-c',child,str(checkpoint),digest,
+                                 str(self.root/'fresh'),str(self.reference),str(out),str(trace)],
+                                cwd=Path(__file__).resolve().parents[1], stdin=subprocess.DEVNULL,
+                                stdout=subprocess.PIPE,stderr=subprocess.STDOUT, timeout=30)
+        self.assertEqual(result.returncode, 0, result.stdout.decode(errors='replace'))
+        self.assertEqual(read_bundle(out), expected)
+        self.assertEqual(json.loads(trace.read_text()), json.loads(json.dumps(expected_trace)))
+        self.assertEqual(hashlib.sha256(self.reference.read_bytes()).hexdigest(), reference_hash)
+
+
+if __name__ == '__main__':
+    unittest.main()