Advance keyboard epochs after validated input enqueue

JSON_RPC_API.md

@@ -539,13 +539,14 @@ queues the corresponding break code (`scan_code | 0x80`). Events are queued in
 the listed order. Result:
 
 ```json
-{"accepted":2,"state_revision":12345}
+{"accepted":2,"state_revision":12346}
 ```
 
 Keyboard requests are allowed while running or paused. The complete batch
-is validated before any event is enqueued. Enqueuing does not advance
-`state_revision` or the CPU; this revision alone cannot prove the input queue
-is unchanged. `keyboard.scancode` defaults omitted `pressed` to true.
+is validated before any event is enqueued. Each successfully validated batch
+increments `state_revision` once without advancing the CPU. Invalid batches
+leave the queue, pressed state and revision unchanged. `keyboard.scancode`
+defaults omitted `pressed` to true.
 
 ## `input.state`
 

docs/HANDOFF.md

@@ -1,3 +1,21 @@
+# Handoff - keyboard input epochs, 2026-10-06
+
+FINISHED local: each validated input.keyboard/keyboard.scancode batch advances
+state_revision once after enqueue. Running input remains supported; enqueue does
+not execute the CPU or complete an active operation. Invalid batches preserve
+complete machine/control state. Stale RPC coherent-observation and snapshot tokens
+now refuse; fresh snapshot metadata includes the changed keyboard queue.
+FINISHED gates: 39 affected tests pass on CPython and PyPy; omitting the enqueue
+epoch bump fails all three new named controls, then exact restored source passes.
+The real JSON-lines server services input while its production CPU loop runs.
+Review: RPC batch handling is correct; separate confirmed VNC/Telnet producer
+epoch debt remains (host input can bypass revision). No frontend fix is implied.
+API documentation rebase onto merged 48867c9 is complete, including the updated
+enqueue paragraph/sample. WIP: final-head public CI and integration; parent
+stays pinned to the previous runtime until integration.
+
+Earlier handoff notes below describe their historical slices.
+
 # Handoff - exclusive text address selectors, 2026-10-06
 
 FINISHED local: video.text and state.observe video_text reject simultaneous

main.py

@@ -858,6 +858,8 @@ def handle_debug(request):
                 scancodes.append(code if pressed else code | 0x80)
             for scancode in scancodes:
                 kb.PushKeyboardScancode(scancode)
+            # Queued input is snapshot state even before the CPU consumes it.
+            control['revision'] += 1
             return {'accepted': len(events), 'state_revision': control['revision']}
 
         if method == 'input.state':

tests/test_execution_rpc.py

@@ -20,6 +20,7 @@
 import debughardware
 import debugserver
 import i8088
+import keyboard
 
 
 class IdleTransport:
@@ -83,6 +84,23 @@ def interrupt_bp(self):
 
 
 class ExecutionRPCTests(unittest.TestCase):
+    def test_running_keyboard_enqueue_advances_epoch_without_completing_operation(self):
+        machine = HeadlessMachine()
+        machine.namespace['kb'] = keyboard.Keyboard()
+        machine.load(b'\x90')
+        operation = machine.rpc('execution.continue')
+        before = machine.rpc('state.get_registers')
+        accepted = machine.rpc('input.keyboard', events=[{'scan_code': 77},
+                                                         {'scan_code': 77, 'pressed': False}])
+        self.assertEqual(accepted['state_revision'], before['state_revision'] + 1)
+        self.assertFalse(machine.control['paused'])
+        self.assertEqual(machine.rpc('execution.wait', operation_id=operation['operation_id']),
+                         {'running': True})
+        after = machine.rpc('state.get_registers')
+        self.assertEqual(after['clock'], before['clock'])
+        self.assertEqual(after['ip'], before['ip'])
+        self.assertEqual(list(machine.namespace['kb']._keyboard_buffer.queue), [77, 205])
+
     def test_plain_continue_batches_sixty_four_unobserved_instructions(self):
         machine = HeadlessMachine()
         machine.load(bytes.fromhex('eb fe'))
@@ -248,6 +266,23 @@ def test_parse_error_does_not_close_connection(self):
         self.assertEqual(response['error']['code'], -32700)
         self.assertEqual(self.request('state.get_registers')['ip'], 0x100)
 
+    def test_keyboard_request_is_serviced_by_running_loop_over_json_lines(self):
+        self.machine.namespace['kb'] = keyboard.Keyboard()
+        self.machine.load(bytes.fromhex('eb fe'))
+        operation = self.request('execution.continue')
+        accepted = self.request('input.keyboard', events=[{'scan_code': 77},
+                                                         {'scan_code': 77, 'pressed': False}])
+        self.assertEqual(accepted['accepted'], 2)
+        self.assertGreater(accepted['state_revision'], operation['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['kb']._keyboard_buffer.queue), [77, 205])
+        self.assertEqual(self.machine.namespace['kb'].GetPressedScancodes(), [])
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_keyboard_rpc.py

@@ -40,6 +40,7 @@ def test_invalid_second_event_has_no_side_effects(self):
                 with self.assertRaises(ValueError):
                     self.call({'events': [{'scan_code': 42}, bad_event]})
                 self.assertEqual(self.snapshot(), before)
+                self.assertEqual(self.handler.__globals__['control']['revision'], 123)
 
     def test_rejected_batch_does_not_release_an_existing_key(self):
         self.kb.PushKeyboardScancode(42)
@@ -48,17 +49,18 @@ def test_rejected_batch_does_not_release_an_existing_key(self):
             self.call({'events': [{'scan_code': 42, 'pressed': False},
                                   {'scan_code': 128}]})
         self.assertEqual(self.snapshot(), before)
+        self.assertEqual(self.handler.__globals__['control']['revision'], 123)
 
     def test_valid_batch_preserves_make_break_order(self):
         result = self.call({'events': [{'scan_code': '0x2a'}, {'scan_code': 30},
                                       {'scan_code': 30, 'pressed': False},
                                       {'scan_code': 42, 'pressed': False}]})
-        self.assertEqual(result, {'accepted': 4, 'state_revision': 123})
+        self.assertEqual(result, {'accepted': 4, 'state_revision': 124})
         self.assertEqual(self.snapshot(), ([], [42, 30, 158, 170]))
 
     def test_single_event_alias_and_default_pressed(self):
-        self.call({'scan_code': 30}, method='keyboard.scancode')
-        self.call({'scan_code': 30, 'pressed': False})
+        self.assertEqual(self.call({'scan_code': 30}, method='keyboard.scancode')['state_revision'], 124)
+        self.assertEqual(self.call({'scan_code': 30, 'pressed': False})['state_revision'], 125)
         self.assertEqual(self.snapshot(), ([], [30, 158]))
 
     def test_batch_limits(self):
@@ -68,7 +70,9 @@ def test_batch_limits(self):
                 with self.assertRaises(ValueError):
                     self.call({'events': events})
                 self.assertEqual(self.snapshot(), before)
-        self.assertEqual(self.call({'events': [{'scan_code': 30}] * 32})['accepted'], 32)
+                self.assertEqual(self.handler.__globals__['control']['revision'], 123)
+        self.assertEqual(self.call({'events': [{'scan_code': 30}] * 32}),
+                         {'accepted': 32, 'state_revision': 124})
         self.assertEqual(self.snapshot(), ([30], [30] * 32))
 
 

tests/test_observation_rpc.py

@@ -12,6 +12,7 @@
 import cga
 import rom
 import vga
+import keyboard
 from tests.test_execution_rpc import HeadlessMachine
 from tests import test_execution_rpc as execution_tests
 
@@ -30,6 +31,22 @@ def machine(video='cga'):
 
 
 class ObservationTests(unittest.TestCase):
+    def test_keyboard_enqueue_invalidates_old_observation_epoch(self):
+        m, _ = machine()
+        m.namespace['kb'] = keyboard.Keyboard()
+        for method, params in (('input.keyboard', {'events': [{'scan_code': 77},
+                                                             {'scan_code': 77, 'pressed': False}]}),
+                               ('keyboard.scancode', {'scan_code': 30})):
+            before = m.rpc('state.get_registers')
+            accepted = m.rpc(method, **params)
+            self.assertEqual(accepted['state_revision'], before['state_revision'] + 1)
+            with self.assertRaisesRegex(ValueError, 'expected state revision'):
+                m.rpc('state.observe', expected_state_revision=before['state_revision'])
+            actual = m.rpc('state.observe', expected_state_revision=accepted['state_revision'])
+            self.assertEqual(actual['registers']['clock'], before['clock'])
+            self.assertEqual(actual['registers']['ip'], before['ip'])
+            self.assertEqual(actual['state_revision'], accepted['state_revision'])
+
     def test_equal_separate_reads_preserve_captured_state(self):
         for adapter in ('cga', 'vga'):
             with self.subTest(adapter=adapter):

tests/test_snapshot_contract.py

@@ -17,6 +17,39 @@
 
 
 class SnapshotContractTests(unittest.TestCase):
+    def test_keyboard_epochs_guard_complete_snapshot_state_and_rejected_batches(self):
+        self.h.namespace['kb'] = self.cpu._devices[1]
+        old = self.export()
+        for index, (method, params) in enumerate((
+                ('input.keyboard', {'events': [{'scan_code': 42}, {'scan_code': 42, 'pressed': False}]}),
+                ('keyboard.scancode', {'scan_code': 77}))):
+            before = capture_machine(self.cpu)
+            controls = copy.deepcopy(self.h.control)
+            with self.assertRaises(ValueError):
+                self.h.rpc('input.keyboard', events=[{'scan_code': 42}, {'scan_code': 128}])
+            self.assertEqual(capture_machine(self.cpu), before)
+            self.assertEqual(self.h.control, controls)
+            accepted = self.h.rpc(method, **params)
+            changed = capture_machine(self.cpu)
+            self.assertNotEqual(changed[0]['keyboard'], before[0]['keyboard'])
+            self.assertEqual(accepted['state_revision'], controls['revision'] + 1)
+            refused = self.root / ('stale-' + str(index))
+            current_controls = copy.deepcopy(self.h.control)
+            for operation in ('export', 'import'):
+                with self.assertRaisesRegex(ValueError, 'expected state revision'):
+                    self.h.rpc('machine.snapshot.' + operation,
+                               path=old['path'] if operation == 'import' else str(self.root / 'stale.pypc'),
+                               expected_state_revision=controls['revision'], disk_root=str(refused),
+                               expected_sha256=old['sha256'], references={'0': str(self.reference)})
+                self.assertFalse(refused.exists())
+                self.assertFalse((self.root / 'stale.pypc').exists())
+                self.assertEqual(capture_machine(self.cpu), changed)
+                self.assertEqual(self.h.control, current_controls)
+            current = self.h.rpc('machine.snapshot.export', path=str(self.root / ('queued-' + str(index) + '.pypc')),
+                                 expected_state_revision=accepted['state_revision'])
+            manifest, _ = read_bundle(current['path'], current['sha256'])
+            self.assertEqual(manifest['keyboard'], changed[0]['keyboard'])
+
     def setUp(self):
         scratch = Path(__file__).resolve().parents[1] / '.test-tmp'
         scratch.mkdir(exist_ok=True)