Invalidate stale snapshot guards after UART batches and port reads

JSON_RPC_API.md

@@ -186,8 +186,11 @@ Result:
 
 Register values and `clock` are unsigned integers. `state_revision` advances after
 each executed instruction and after a successful paused-state mutation through
-`state.set_registers`, `memory.write`, or `io.write`. It identifies the snapshot
-boundary at which the read was made. Unlike the full DOSBox-X service, PyPC permits
+`state.set_registers`, `memory.write`, or `io.write`. Successfully accepted
+keyboard/joystick input, `serial.write` batches, and `io.read` bus operations also
+advance it once. Port reads conservatively invalidate guards even if a particular
+read leaves no observable device change. It identifies the snapshot boundary at
+which the read was made. Unlike the full DOSBox-X service, PyPC permits
 this read while the CPU is running because the request is executed synchronously at
 an instruction boundary.
 
@@ -502,12 +505,15 @@ Parameters:
 `port` is a required 16-bit integer. Result:
 
 ```json
-{"port":984,"value":9,"state_revision":12345}
+{"port":984,"value":9,"state_revision":12346}
 ```
 
 The read uses the same emulated I/O dispatch as an 8088 `IN` byte operation. It may
-have device-specific read side effects. Use `io.write` for explicit paused-state
-mutation.
+have device-specific read side effects. Every successful read advances
+`state_revision` once, including ports whose particular read leaves no observable
+change; this operation does not infer device purity or execute a CPU instruction.
+Invalid port parameters leave the machine and revision unchanged. Reads remain
+allowed while running or paused. Use `io.write` for explicit paused-state mutation.
 
 ## `io.write`
 
@@ -976,8 +982,12 @@ INT 14h and DOS COM1 programs; it is not a serial timing validation instrument.
 ## `serial.write`
 
 Params: `{"data_base64":"Ug0="}`. Queues the bytes `R` and CR on COM1 and returns
-`{"accepted":2,"rx_pending":2}`. Accepts 1..4096 bytes, validates the entire input
-before mutation, and refuses a full 65536-byte input queue or UART loopback mode.
+`{"accepted":2,"rx_pending":2,"state_revision":12346}`. Accepts 1..4096 bytes,
+validates the entire input before mutation, and refuses a full 65536-byte input queue or UART loopback mode.
+Each successfully accepted batch advances `state_revision` once without executing
+the CPU; rejected input preserves the UART, machine and revision. Calls remain
+supported while running or paused. These serial and port-read methods are PyPC
+capabilities; MartyPC currently refuses them as unsupported.
 A network timeout after submission is uncertain: do not blindly resend.
 
 ## `serial.read`

docs/HANDOFF.md

@@ -1,3 +1,22 @@
+# Handoff - device operation epochs, 2026-10-06
+
+FINISHED local: successful serial.write batches and io.read bus operations advance
+state_revision once. Port reads conservatively invalidate guards even when that
+particular device read leaves no observable change; no per-device purity guesses.
+Running calls remain supported. Invalid input is atomic, and stale coherent
+observation/snapshot export/import tokens refuse after UART enqueue/consumption.
+PR75 frontend policy integrated at 86c0f17 after exact-head CI and no-defect review.
+FINISHED gates: 59 affected unittest checks plus four historical UART probes pass
+on CPython/PyPy. Each omitted serial/read epoch mutant makes the named stale-token
+control accept the old revision; exact restored source passes. Invalid parameters,
+full queue and loopback refusals preserve complete Machine/control state.
+Initial scoped review found no runtime defect; its running-loop proof and central
+revision-summary gaps are fixed with an actual JSON-lines/CPU-loop UART test.
+WIP: targeted final review and final-head public CI/integration.
+Parent source pin remains fixed until the combined repair/compiler witness.
+
+Earlier handoff notes below describe their historical slices.
+
 # Handoff - controlled frontend input, 2026-10-06
 
 FINISHED local: default JSON-RPC runs give Telnet/VNC no keyboard target, so

main.py

@@ -625,7 +625,9 @@ def handle_debug(request):
             data = params.get('data_base64')
             if not isinstance(data, str) or len(data) > 5464:
                 raise ValueError('bounded base64 serial input required')
-            return serial.host_write(base64.b64decode(data, validate=True))
+            accepted = serial.host_write(base64.b64decode(data, validate=True))
+            control['revision'] += 1
+            return {**accepted, 'state_revision': control['revision']}
         if method == 'serial.read':
             return serial.host_read(params.get('offset'), params.get('max_bytes', 4096))
 
@@ -823,7 +825,10 @@ def handle_debug(request):
             port = rpc_number(params.get('port'), 'port')
             if port < 0 or port > 0xffff:
                 raise ValueError('port must be a 16-bit number')
-            return {'port': port, 'value': p._io.In(port, False),
+            value = p._io.In(port, False)
+            # A device read can consume input or change its register-read phase.
+            control['revision'] += 1
+            return {'port': port, 'value': value,
                     'state_revision': control['revision']}
 
         if method == 'io.write':

tests/machine_fixture.py

@@ -2,7 +2,7 @@
 import ast
 import threading
 from pathlib import Path
-import bus, i8088, i8253, i8255, keyboard, vga, xtide
+import bus, i8088, i8253, i8255, keyboard, vga, xtide, uart8250
 import debughardware
 import debugtrace
 import videohistory
@@ -11,10 +11,12 @@
 from tests.test_execution_rpc import HeadlessMachine
 
 
-def machine(tmp_path):
+def machine(tmp_path, com1=False):
     disk = tmp_path/'disk.img'; disk.write_bytes(bytes(4096))
     kb = keyboard.Keyboard()
     devices = [i8253.i8253(), kb, i8255.i8255(kb), vga.VGA(False), xtide.XTIDE([str(disk)])]
+    if com1:
+        devices.append(uart8250.UART8250())
     b = bus.Bus(1048576, devices, [])
     cpu = i8088.i8088(b, devices, True)
     state = cpu.GetState(); state.SetCS(0x1000); state.SetIP(0); state.SetDS(0x1000)
@@ -39,9 +41,9 @@ def run(cpu, count=200):
     return events
 
 
-def rpc_machine(tmp_path):
+def rpc_machine(tmp_path, com1=False):
     harness = HeadlessMachine()
-    cpu = machine(tmp_path)
+    cpu = machine(tmp_path, com1=com1)
     display = LockedDisplay(cpu._devices[3])
     vnc = vncserver.VNCServer.__new__(vncserver.VNCServer)
     vnc._display = display
@@ -51,6 +53,8 @@ def rpc_machine(tmp_path):
                              hardware_trace=debughardware.HardwareTraceRecorder(),
                              machine_snapshots=MachineSnapshots(cpu, display, vnc),
                              video_history=videohistory.VideoHistory(cpu.GetState().GetClock))
+    if com1:
+        harness.namespace['serial'] = cpu._devices[5]
     # Bind production register methods to this real motherboard before testing.
     tree = ast.parse((Path(__file__).resolve().parents[1]/'main.py').read_text())
     body = next(n.body for n in tree.body if isinstance(n, ast.Try))

tests/test_device_io_epochs.py

@@ -0,0 +1,141 @@
+"""Side-effecting RPC operations must invalidate guarded complete-machine captures."""
+import base64
+import copy
+from pathlib import Path
+from tempfile import TemporaryDirectory
+import unittest
+
+from checkpointbundle import read_bundle
+from machinecodec import capture_machine
+from tests.machine_fixture import rpc_machine
+from tests.test_execution_rpc import HeadlessMachine
+from tests import test_execution_rpc as execution_tests
+from uart8250 import UART8250
+
+
+class DeviceIOEpochTests(unittest.TestCase):
+    def setUp(self):
+        scratch = Path(__file__).resolve().parents[1] / '.test-tmp'
+        scratch.mkdir(exist_ok=True)
+        temporary = TemporaryDirectory(dir=scratch)
+        self.addCleanup(temporary.cleanup)
+        self.root = Path(temporary.name)
+        self.h, self.cpu, _ = rpc_machine(self.root, com1=True)
+        self.uart = self.cpu._devices[5]
+
+    def assert_stale_refusals_preserve_complete_state(self, old_revision, archive):
+        before = capture_machine(self.cpu)
+        controls = copy.deepcopy(self.h.control)
+        with self.assertRaisesRegex(ValueError, 'expected state revision'):
+            self.h.rpc('state.observe', expected_state_revision=old_revision)
+        for method in ('export', 'import'):
+            with self.assertRaisesRegex(ValueError, 'expected state revision'):
+                self.h.rpc('machine.snapshot.' + method,
+                           path=archive['path'] if method == 'import' else str(self.root / 'stale.pypc'),
+                           expected_sha256=archive['sha256'], disk_root=str(self.root / 'refused'),
+                           expected_state_revision=old_revision)
+            self.assertEqual(capture_machine(self.cpu), before)
+            self.assertEqual(self.h.control, controls)
+            self.assertFalse((self.root / 'stale.pypc').exists())
+            self.assertFalse((self.root / 'refused').exists())
+
+    def test_serial_enqueue_and_consuming_port_read_invalidate_snapshot_tokens(self):
+        archive = self.h.rpc('machine.snapshot.export', path=str(self.root / 'original.pypc'),
+                             expected_state_revision=0)
+        before = self.h.rpc('state.get_registers')
+        accepted = self.h.rpc('serial.write', data_base64=base64.b64encode(b'R\r').decode())
+        self.assert_stale_refusals_preserve_complete_state(0, archive)
+        self.assertEqual(accepted, {'accepted': 2, 'rx_pending': 2, 'state_revision': 1})
+        self.assertEqual(list(self.uart.rx), [ord('R'), 13])
+        read = self.h.rpc('io.read', port='0x3f8')
+        self.assertEqual(list(self.uart.rx), [13])
+        self.assert_stale_refusals_preserve_complete_state(1, archive)
+        self.assertEqual(read, {'port': 0x3f8, 'value': ord('R'), 'state_revision': 2})
+        current = self.h.rpc('state.observe', expected_state_revision=2)
+        self.assertEqual(current['registers']['clock'], before['clock'])
+        self.assertEqual(current['registers']['ip'], before['ip'])
+        fresh = self.h.rpc('machine.snapshot.export', path=str(self.root / 'fresh.pypc'),
+                           expected_state_revision=2)
+        manifest, _ = read_bundle(fresh['path'], fresh['sha256'])
+        self.assertEqual(manifest['serial'], capture_machine(self.cpu)[0]['serial'])
+
+    def test_rejected_serial_and_port_parameters_preserve_complete_state(self):
+        before = capture_machine(self.cpu)
+        controls = copy.deepcopy(self.h.control)
+        invalid = [('serial.write', {'data_base64': data})
+                   for data in (None, '', '!', 'A' * 5465)]
+        invalid.extend(('io.read', {'port': port}) for port in (None, True, -1, 65536))
+        for method, params in invalid:
+            with self.subTest(method=method, params=params), self.assertRaises(ValueError):
+                self.h.rpc(method, **params)
+            self.assertEqual(capture_machine(self.cpu), before)
+            self.assertEqual(self.h.control, controls)
+        for _ in range(16):
+            self.uart.host_write(bytes(4096))
+        full = capture_machine(self.cpu)
+        with self.assertRaisesRegex(ValueError, 'queue full'):
+            self.h.rpc('serial.write', data_base64='QQ==')
+        self.assertEqual(capture_machine(self.cpu), full)
+        self.assertEqual(self.h.control, controls)
+        self.uart.IO_Write(0x3fc, 16)
+        loopback = capture_machine(self.cpu)
+        with self.assertRaisesRegex(ValueError, 'loopback mode'):
+            self.h.rpc('serial.write', data_base64='QQ==')
+        self.assertEqual(capture_machine(self.cpu), loopback)
+        self.assertEqual(self.h.control, controls)
+
+    def test_running_device_calls_preserve_execution_operation_and_cpu_boundary(self):
+        operation = self.h.rpc('execution.continue')
+        before = self.h.rpc('state.get_registers')
+        accepted = self.h.rpc('serial.write', data_base64='QQ==')
+        self.assertEqual(accepted['state_revision'], before['state_revision'] + 1)
+        read = self.h.rpc('io.read', port=0x3f8)
+        self.assertEqual(read['state_revision'], before['state_revision'] + 2)
+        self.assertEqual(read['value'], ord('A'))
+        self.assertFalse(self.h.control['paused'])
+        self.assertEqual(self.h.rpc('execution.wait', operation_id=operation['operation_id']),
+                         {'running': True})
+        after = self.h.rpc('state.get_registers')
+        self.assertEqual(after['clock'], before['clock'])
+        self.assertEqual(after['ip'], before['ip'])
+
+    def test_successful_port_read_advances_conservatively_without_inferred_purity(self):
+        before = capture_machine(self.cpu)
+        read = self.h.rpc('io.read', port=0x3fd)
+        self.assertEqual(read, {'port': 0x3fd, 'value': 0x60, 'state_revision': 1})
+        self.assertEqual(capture_machine(self.cpu), before)
+        with self.assertRaisesRegex(ValueError, 'expected state revision'):
+            self.h.rpc('state.observe', expected_state_revision=0)
+
+
+class DeviceIORunningSocketTests(unittest.TestCase):
+    # Use the same JSON-lines socket and production-loop worker as execution tests.
+    tearDown = execution_tests.SocketRPCTests.tearDown
+    request = execution_tests.SocketRPCTests.request
+
+    def setUp(self):
+        uart = UART8250()
+        machine = HeadlessMachine(devices=[uart], run_IO=True)
+        machine.namespace['serial'] = uart
+        execution_tests.SocketRPCTests.setUp(self, machine=machine)
+
+    def test_device_requests_are_serviced_while_cpu_actually_runs(self):
+        self.machine.load(bytes.fromhex('eb fe'))
+        operation = self.request('execution.continue')
+        accepted = self.request('serial.write', data_base64='Ug0=')
+        self.assertEqual(accepted['accepted'], 2)
+        self.assertGreater(accepted['state_revision'], operation['state_revision'])
+        read = self.request('io.read', port=0x3f8)
+        self.assertEqual(read['value'], ord('R'))
+        self.assertGreater(read['state_revision'], accepted['state_revision'])
+        self.assertEqual(self.request('execution.wait', operation_id=operation['operation_id']),
+                         {'running': True})
+        registers = self.request('state.get_registers')
+        self.assertFalse(self.machine.control['paused'])
+        self.assertGreater(registers['clock'], operation['clock'])
+        self.request('execution.pause')
+        self.assertEqual(list(self.machine.namespace['serial'].rx), [13])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_execution_rpc.py

@@ -30,12 +30,12 @@ def has_pending(self):
 
 
 class HeadlessMachine:
-    def __init__(self, main_path=None, devices=None):
+    def __init__(self, main_path=None, devices=None, run_IO=False):
         main_path = main_path or Path(__file__).resolve().parents[1] / 'main.py'
         tree = ast.parse(main_path.read_text(), filename=str(main_path))
         body = next(node.body for node in tree.body if isinstance(node, ast.Try))
         self.memory = bus.Bus(1 << 20, [], [])
-        self.cpu = i8088.i8088(self.memory, [] if devices is None else devices, False)
+        self.cpu = i8088.i8088(self.memory, [] if devices is None else devices, run_IO)
         self.state = self.cpu.GetState()
         for name, value in dict(CS=0x1000, DS=0x2000, ES=0x3000,
                                 SS=0x2000, IP=0x100, SP=0x9000, Flags=2).items():
@@ -242,8 +242,8 @@ def test_trace_fetch_wrap_matches_cpu(self):
 
 
 class SocketRPCTests(unittest.TestCase):
-    def setUp(self):
-        self.machine = HeadlessMachine()
+    def setUp(self, machine=None):
+        self.machine = machine if machine is not None else HeadlessMachine()
         self.server = debugserver.DebugServer(0)
         self.machine.namespace['debug'] = self.server
         self.stop = threading.Event()

tests/test_uart8250.py

@@ -55,16 +55,18 @@ def test_real_rpc_handler_validation():
     from pathlib import Path
     tree=ast.parse(Path('main.py').read_text(encoding='utf-8'))
     functions=[n for n in ast.walk(tree) if isinstance(n,ast.FunctionDef) and n.name in ('rpc_params','rpc_number','handle_debug')]
-    uart=UART8250();ns={'serial':uart,'base64':base64}
+    uart=UART8250();ns={'serial':uart,'base64':base64,'control':{'revision':123}}
     exec(compile(ast.Module(body=functions,type_ignores=[]),'main.py','exec'),ns)
     call=lambda m,p:ns['handle_debug']({'method':m,'params':p})
     assert call('serial.write',{'data_base64':'Ug0='})['accepted']==2
+    assert ns['control']['revision']==124
     before=uart.dump()
     for bad in ('!',None,'A'*5465,''):
         try:call('serial.write',{'data_base64':bad})
         except ValueError:pass
         else:raise AssertionError('bad serial input accepted')
         assert uart.dump()==before
+        assert ns['control']['revision']==124
     assert call('serial.status',{})['rx_pending']==2