diff --git a/JSON_RPC_API.md b/JSON_RPC_API.md index 3c8c391..1d94b70 100644 --- a/JSON_RPC_API.md +++ b/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` diff --git a/docs/HANDOFF.md b/docs/HANDOFF.md index 8a2c6e4..cf6766f 100644 --- a/docs/HANDOFF.md +++ b/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 diff --git a/main.py b/main.py index 7ddcf4c..828e3e5 100755 --- a/main.py +++ b/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': diff --git a/tests/machine_fixture.py b/tests/machine_fixture.py index f08b842..72fed44 100644 --- a/tests/machine_fixture.py +++ b/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)) diff --git a/tests/test_device_io_epochs.py b/tests/test_device_io_epochs.py new file mode 100644 index 0000000..02af197 --- /dev/null +++ b/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() diff --git a/tests/test_execution_rpc.py b/tests/test_execution_rpc.py index 4dda13b..f726e77 100644 --- a/tests/test_execution_rpc.py +++ b/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() diff --git a/tests/test_uart8250.py b/tests/test_uart8250.py index 2f0dff8..d1a416f 100644 --- a/tests/test_uart8250.py +++ b/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