Give RPC-controlled runs exclusive keyboard ownership

JSON_RPC_API.md

@@ -22,6 +22,13 @@ see `guest/README.md`.
 Ports must be distinct integers in 1..65535. `agent.capabilities.endpoint`
 reports the selected RPC port. Each instance should use its own writable disks.
 
+With JSON-RPC enabled (the default), Telnet and VNC are view-only: their keyboard
+messages do not enter the guest queue. Use `input.keyboard` or `keyboard.scancode`
+for controlled input. This prevents a frontend thread from changing input during
+a guarded observation or snapshot, matching MartyPC's controlled-input policy.
+For ordinary manual play, `--no-rpc` disables the RPC listener and enables normal
+Telnet/VNC keyboard input. There is no concurrent frontend-input/RPC-control mode.
+
 Each request and response is one UTF-8 JSON object terminated by `LF` (`\n`). A
 client may keep the connection open and send multiple requests. Requests are
 executed by the CPU thread at an instruction boundary. The returned state is

README.md

@@ -14,6 +14,11 @@
 > `127.0.0.1:2301` using JSON-lines framing; despite the repository name, it is
 > not yet a Model Context Protocol server.
 >
+> RPC control is enabled by default. Telnet/VNC then display the machine but
+> cannot inject keys; use the RPC keyboard methods. Pass `--no-rpc` for ordinary
+> Telnet/VNC keyboard control with no RPC listener. This keeps controlled input
+> and guarded observations/snapshots under the CPU thread's ownership.
+>
 > **Host-directory FAT16 mount:** From the repository root, pass a host
 > directory with `--host-dir`; it appears inside DOS as the guest `D:` drive.
 > The existing `C:` boot image is unchanged.

docs/HANDOFF.md

@@ -1,3 +1,19 @@
+# Handoff - controlled frontend input, 2026-10-06
+
+FINISHED local: default JSON-RPC runs give Telnet/VNC no keyboard target, so
+their callbacks cannot mutate queued input during guarded captures. RPC keyboard
+input still works. --no-rpc disables the RPC listener and retains normal frontend
+keyboard input; the CPU loop supports an absent debug server. This closes the
+separate frontend epoch debt recorded below, without a callback revision race.
+FINISHED gates: 50 affected tests pass on CPython/PyPy, including actual startup
+listeners with and without RPC and every VNC mapped key under controlled policy.
+Restoring the old frontend keyboard binding fails the complete-state control;
+exact restored source passes. PR73 and PR74 rebase-integrated at d8b3c0ef after
+exact final public CI; this branch is rebased on that master. WIP: scoped review,
+final public CI and frontend-policy integration. Parent pin remains unchanged.
+
+Earlier handoff notes below describe their historical slices.
+
 # Handoff - idempotent stopped pause, 2026-10-06
 
 FINISHED local: execution.pause on a stopped machine preserves the complete

main.py

@@ -45,6 +45,8 @@ def ParseArguments():
         help='reserve D8000h-D9FFFh for the DOS control worker')
     parser.add_argument('--game-port', action='store_true',
         help='attach two analog two-button joysticks at port201h')
+    parser.add_argument('--no-rpc', action='store_true',
+        help='disable JSON-RPC control and permit Telnet/VNC keyboard input')
     for name, default in (('rpc', 2301), ('telnet', 2300), ('vnc', 5902)):
         parser.add_argument(f'--{name}-port', type=int, default=default,
                             help=f'{name} listener port (default: {default})')
@@ -163,12 +165,14 @@ def ReadTextCells(scr, display_address=None):
         p.SetInterruptServiceHook(VGAInterruptService(scr))
 
     frontend_display = LockedDisplay(scr)
-    t = telnet.Telnet(arguments.telnet_port, kb, frontend_display)
+    # A controlled machine has one input owner; frontend threads remain viewers.
+    frontend_keyboard = kb if arguments.no_rpc else None
+    t = telnet.Telnet(arguments.telnet_port, frontend_keyboard, frontend_display)
     vnc_display = (vncspeed.SpeedDisplay(frontend_display, state.GetClock)
                    if arguments.vnc_speed_overlay else frontend_display)
-    v = vncserver.VNCServer(vnc_display, kb, arguments.vnc_port, False)
+    v = vncserver.VNCServer(vnc_display, frontend_keyboard, arguments.vnc_port, False)
     machine_snapshots = MachineSnapshots(p, frontend_display, v)
-    debug = debugserver.DebugServer(arguments.rpc_port)
+    debug = None if arguments.no_rpc else debugserver.DebugServer(arguments.rpc_port)
     control = {
         'paused': False, 'step': False, 'revision': 0,
         'snapshot_number': 0, 'snapshots': {},
@@ -1034,13 +1038,15 @@ def handle_debug(request):
 
         raise LookupError(f'unknown method: {method}')
 
-    print(f'Use: "telnet localhost {arguments.telnet_port}" to interact with the emulated system')
-    print(f'and/or connect using a VNC client to localhost:{arguments.vnc_port} (preferred)')
+    if arguments.no_rpc:
+        print(f'Use Telnet localhost:{arguments.telnet_port} or VNC localhost:{arguments.vnc_port} for keyboard input')
+    else:
+        print(f'Telnet localhost:{arguments.telnet_port} and VNC localhost:{arguments.vnc_port} are view-only; use JSON-RPC keyboard input')
 
     p_time = time.time()
     p_cycles = 0
     while True:
-        if debug.has_pending():
+        if debug is not None and debug.has_pending():
             debug.process_pending(handle_debug)
         if control['paused']:
             time.sleep(0.001)

telnet.py

@@ -32,6 +32,8 @@ def __init__(self, port: int, kb: keyboard.Keyboard, scr: mda.MDA):
         t.start()
 
     def push(self, c):
+        if self._kb is None:
+            return
         key_map = dict()
         key_map[8] = ( 0x0e, )
         key_map[9] = ( 0x0f, )

tests/test_frontend_input_rpc.py

@@ -0,0 +1,100 @@
+"""Actual frontend input callbacks under the production launch ownership policy."""
+import ast
+import copy
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+import telnet
+import vncserver
+from machinecodec import capture_machine
+from tests.machine_fixture import rpc_machine
+from tests.test_execution_rpc import HeadlessMachine
+
+
+def frontends(kb, no_rpc, display):
+    path = Path(__file__).resolve().parents[1] / 'main.py'
+    tree = ast.parse(path.read_text())
+    body = next(n.body for n in tree.body if isinstance(n, ast.Try))
+    names = ('frontend_keyboard', 't', 'v', 'debug')
+    assignments = [n for n in body if isinstance(n, ast.Assign) and
+                   any(isinstance(t, ast.Name) and t.id in names for t in n.targets)]
+    assert len(assignments) == len(names)
+    namespace = dict(kb=kb, arguments=SimpleNamespace(no_rpc=no_rpc, telnet_port=2300,
+                                                    vnc_port=5902, rpc_port=2301),
+                     frontend_display=display, vnc_display=display,
+                     telnet=telnet, vncserver=vncserver,
+                     debugserver=SimpleNamespace(DebugServer=lambda port: ('rpc', port)))
+    # Keep real constructors/key maps, but do not launch listeners for this owner test.
+    with patch('threading.Thread.start'):
+        exec(compile(ast.Module(body=assignments, type_ignores=[]), str(path), 'exec'), namespace)
+    return namespace['t'], namespace['v'], namespace['debug']
+
+
+class FrontendInputTests(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, vnc = rpc_machine(self.root)
+        self.kb = self.cpu._devices[1]
+        self.h.namespace['kb'] = self.kb
+        self.display = vnc._display
+
+    def test_controlled_frontends_cannot_mutate_machine_or_stale_guards(self):
+        t, v, debug = frontends(self.kb, False, self.display)
+        self.assertEqual(debug, ('rpc', 2301))
+        before = capture_machine(self.cpu)
+        controls = copy.deepcopy(self.h.control)
+        def all_vnc_keys():
+            for key in v._key_map:
+                v.PushChar(key, True)
+                v.PushChar(key, False)
+        for producer in (lambda: t.push(ord('a')),
+                         all_vnc_keys):
+            producer()
+            self.assertEqual(capture_machine(self.cpu), before)
+            self.assertEqual(self.h.control, controls)
+        self.assertEqual(self.h.rpc('state.observe', expected_state_revision=0)['state_revision'], 0)
+        self.h.rpc('machine.snapshot.export', path=str(self.root / 'unchanged.pypc'),
+                   expected_state_revision=0)
+        self.assertEqual(capture_machine(self.cpu), before)
+
+    def test_controlled_rpc_keyboard_still_changes_epoch_and_guest_queue(self):
+        frontends(self.kb, False, self.display)
+        accepted = self.h.rpc('input.keyboard', events=[{'scan_code': 77},
+                                                       {'scan_code': 77, 'pressed': False}])
+        self.assertEqual(accepted, {'accepted': 2, 'state_revision': 1})
+        self.assertEqual(list(self.kb._keyboard_buffer.queue)[-2:], [77, 205])
+        with self.assertRaisesRegex(ValueError, 'expected state revision'):
+            self.h.rpc('state.observe', expected_state_revision=0)
+
+    def test_uncontrolled_frontends_retain_normal_keyboard_input(self):
+        t, v, debug = frontends(self.kb, True, self.display)
+        self.assertIsNone(debug)
+        before = capture_machine(self.cpu)
+        t.push(ord('a'))
+        v.PushChar(ord('a'), True)
+        v.PushChar(ord('a'), False)
+        self.assertEqual(list(self.kb._keyboard_buffer.queue)[-4:], [30, 158, 30, 158])
+        v.PushChar(0xffe1, True)
+        v.PushChar(0xffe1, False)
+        self.assertEqual(list(self.kb._keyboard_buffer.queue)[-2:], [42, 170])
+        self.assertNotEqual(capture_machine(self.cpu)[0]['keyboard'], before[0]['keyboard'])
+
+    def test_cpu_loop_runs_without_rpc_listener(self):
+        h = HeadlessMachine()
+        h.load(bytes.fromhex('eb fe'))
+        h.namespace['debug'] = None
+        h.control['paused'] = False
+        h.pump()
+        self.assertEqual(h.control['revision'], 64)
+        self.assertGreater(h.state.GetClock(), 0)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_listener_ports.py

@@ -28,6 +28,8 @@ def test_defaults_and_isolated_ports(self):
         result = self.parse()
         self.assertEqual((result.rpc_port, result.telnet_port, result.vnc_port),
                          (2301, 2300, 5902))
+        self.assertFalse(result.no_rpc)
+        self.assertTrue(self.parse('--no-rpc').no_rpc)
         result = self.parse('--rpc-port', '12301', '--telnet-port', '12300',
                             '--vnc-port', '15902')
         self.assertEqual((result.rpc_port, result.telnet_port, result.vnc_port),
@@ -51,7 +53,10 @@ def test_process_binds_selected_ports_and_advertises_rpc_endpoint(self):
                 if attempt == 2:
                     raise
 
-    def exercise_process(self):
+    def test_process_without_rpc_binds_only_frontends(self):
+        self.exercise_process(no_rpc=True)
+
+    def exercise_process(self, no_rpc=False):
         root = Path(__file__).resolve().parents[1]
         # Reserve distinct free ports together; release immediately before launch.
         reservations = [socket.socket() for _ in range(3)]
@@ -76,11 +81,11 @@ def exercise_process(self):
                 process = subprocess.Popen(
                     [sys.executable, '-u', '-c', code, str(root),
                      '--rpc-port', str(ports[0]), '--telnet-port', str(ports[1]),
-                     '--vnc-port', str(ports[2])], cwd=scratch,
+                     '--vnc-port', str(ports[2])] + (['--no-rpc'] if no_rpc else []), cwd=scratch,
                     stdout=log, stderr=log)
                 try:
                     deadline = time.monotonic() + 15
-                    for port in ports:
+                    for port in ports[1:] if no_rpc else ports:
                         while True:
                             try:
                                 connection = socket.create_connection(('127.0.0.1', port), timeout=1)
@@ -91,11 +96,16 @@ def exercise_process(self):
                                     log.seek(0)
                                     raise ConnectionError(log.read().decode(errors='replace'))
                                 time.sleep(0.02)
-                    with socket.create_connection(('127.0.0.1', ports[0]), timeout=3) as client:
-                        client.sendall(b'{"jsonrpc":"2.0","id":1,"method":"agent.capabilities"}\n')
-                        with client.makefile('rb') as stream:
-                            reply = json.loads(stream.readline())
-                    self.assertEqual(reply['result']['endpoint'], f'127.0.0.1:{ports[0]}')
+                    if no_rpc:
+                        with socket.socket() as client:
+                            client.settimeout(1)
+                            self.assertNotEqual(client.connect_ex(('127.0.0.1', ports[0])), 0)
+                    else:
+                        with socket.create_connection(('127.0.0.1', ports[0]), timeout=3) as client:
+                            client.sendall(b'{"jsonrpc":"2.0","id":1,"method":"agent.capabilities"}\n')
+                            with client.makefile('rb') as stream:
+                                reply = json.loads(stream.readline())
+                        self.assertEqual(reply['result']['endpoint'], f'127.0.0.1:{ports[0]}')
                     # Current server greetings identify which listener owns a port;
                     # this does not assert complete protocol conformance.
                     for port, banner in ((ports[1], b'\xff\xf4\x25'),