main.py

#! /usr/bin/python3

from typing import List
import argparse
import base64
import hashlib
import bus
import cga
import debugserver
import debugbreakpoints
import debughardware
import debugtrace
import dosmailbox
import gameport
import i8088
import i8253
import i8255
import keyboard
import mda
import rom
import telnet
import time
import vncserver
import vncspeed
import vga
import videohistory
import xtide
import uart8250
import virtualfat16
from biosservice import VGAInterruptService
from machinesnapshots import LockedDisplay, MachineSnapshots


def ParseArguments():
    parser = argparse.ArgumentParser(description='Run the PyPC 8088 emulator')
    parser.add_argument(
        '--host-dir', metavar='PATH',
        help='expose a host directory as a write-through FAT16 D: drive')
    parser.add_argument(
        '--video', choices=('cga', 'vga'), default='cga',
        help='select the emulated text/video adapter (default: cga)')
    parser.add_argument('--vnc-speed-overlay', action='store_true',
        help='show emulated/wall speed at top right over VNC pixels')
    parser.add_argument('--dos-mailbox', action='store_true',
        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})')
    result = parser.parse_args()
    ports = (result.rpc_port, result.telnet_port, result.vnc_port)
    if any(port < 1 or port > 65535 for port in ports):
        parser.error('listener ports must be in 1..65535')
    if len(set(ports)) != len(ports):
        parser.error('listener ports must be distinct')
    return result

def GetRegisters(state) -> str:
    return f'{state.GetFlagsAsString()} AX:{state.GetAX():04x} BX:{state.GetBX():04x} CX:{state.GetCX():04x} DX:{state.GetDX():04x} SP:{state.GetSP():04x} BP:{state.GetBP():04x} SI:{state.GetSI():04x} DI:{state.GetDI():04x} flags:{state.GetFlags():04x} ES:{state.GetES():04x} CS:{state.GetCS():04x} SS:{state.GetSS():04x} DS:{state.GetDS():04x} IP:{state.GetIP():04x}'


def ParseNumber(value):
    if isinstance(value, int):
        return value
    if isinstance(value, str):
        return int(value, 0)
    raise ValueError('number expected')


def ReadTextScreen(scr, display_address=None):
    columns = scr.GetTextColumns()
    address_mask = scr.GetTextAddressMask()
    rows = []
    if display_address is None:
        display_address = scr._display_address & address_mask
    else:
        display_address &= address_mask
    for y in range(25):
        row = bytearray()
        for x in range(columns):
            offset = (display_address + (y * columns + x) * 2) & address_mask
            character = scr.ReadTextByte(offset)
            row.append(character if character >= 32 else 32)
        rows.append(row.decode('cp437', errors='replace').rstrip())
    return rows


def ReadTextCells(scr, display_address=None):
    columns = scr.GetTextColumns()
    address_mask = scr.GetTextAddressMask()
    cells = []
    if display_address is None:
        display_address = scr._display_address & address_mask
    else:
        display_address &= address_mask
    for y in range(25):
        row = []
        for x in range(columns):
            offset = (display_address + (y * columns + x) * 2) & address_mask
            character = scr.ReadTextByte(offset)
            attribute = scr.ReadTextByte((offset + 1) & address_mask)
            foreground, background, blink = scr.DecodeTextAttribute(attribute)
            row.append({
                'code': character,
                'char': bytes((character,)).decode('cp437', errors='replace'),
                'attribute': attribute,
                'foreground': foreground,
                'background': background,
                'blink': blink,
            })
        cells.append(row)
    return cells

# GIL Status
import sysconfig
status = sysconfig.get_config_var("Py_GIL_DISABLED")
if status is None:
    print("GIL cannot be disabled")
if status == 0:
    print("GIL is active")
if status == 1:
    print("GIL is disabled")

try:
    arguments = ParseArguments()
    devices: List[object] = []
    devices.append(i8253.i8253())
    kb = keyboard.Keyboard()
    devices.append(kb)
    devices.append(i8255.i8255(kb, arguments.video))
    #scr = mda.MDA()
    scr = vga.VGA(False) if arguments.video == 'vga' else cga.CGA(False)
    devices.append(scr)
    disks = ['harddisk.img']
    if arguments.host_dir:
        host_disk = virtualfat16.HostDirectoryFAT16(arguments.host_dir)
        disks.append(host_disk)
        print(f'Exposing {host_disk.directory} as guest drive D: (FAT16, write-through)')
    devices.append(xtide.XTIDE(disks));
    if arguments.dos_mailbox:
        devices.append(dosmailbox.DOSMailbox())

    serial = uart8250.UART8250()
    devices.append(serial)
    joystick = gameport.GamePort() if arguments.game_port else None
    if joystick is not None:
        devices.append(joystick)

    roms = []
    roms.append(rom.Rom('roms/GLABIOS.ROM', 0xf000 * 16 + 0xe000))
    if arguments.video == 'vga':
        roms.append(rom.Rom('roms/PYPCVGA.ROM', 0xc000 * 16))
    roms.append(rom.Rom('roms/ide_xt.bin', 0xd000 * 16 + 0x0000))

    b = bus.Bus(1024 * 1024, devices, roms)
    p = i8088.i8088(b, devices, True)
    state = p.GetState()
    video_history = videohistory.VideoHistory(state.GetClock)
    state.SetCS(0xf000)
    state.SetIP(0xfff0)
    if arguments.video == 'vga':
        p.SetInterruptServiceHook(VGAInterruptService(scr))

    frontend_display = LockedDisplay(scr)
    # 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, frontend_keyboard, arguments.vnc_port, False)
    machine_snapshots = MachineSnapshots(p, frontend_display, v)
    debug = None if arguments.no_rpc else debugserver.DebugServer(arguments.rpc_port)
    control = {
        'paused': False, 'step': False, 'revision': 0,
        'snapshot_number': 0, 'snapshots': {},
        'last_stop': None, 'skip_breakpoint_id': None,
        'breakpoints_active': False,
        'memory_watchpoints_active': False,
        'interrupt_breakpoints_active': False,
        'trace_active': False, 'instruction_hooks_active': False,
        'hardware_trace_active': False,
        'next_operation': 1, 'operations': {}, 'active_operation': None,
        'run_until_id': None, 'run_until_start_clock': None,
        'run_until_deadline_clock': None, 'run_until_requested_ns': None,
    }
    breakpoints = debugbreakpoints.BreakpointManager()
    trace = debugtrace.TraceRecorder()
    hardware_trace = debughardware.HardwareTraceRecorder()

    def rpc_params(request):
        params = request.get('params', {})
        if not isinstance(params, dict):
            raise ValueError('params must be an object')
        return params

    def rpc_number(value, name):
        if isinstance(value, bool):
            raise ValueError(f'{name} must be a number')
        if isinstance(value, int):
            return value
        if isinstance(value, str):
            return int(value, 0)
        raise ValueError(f'{name} must be a number')

    def rpc_address(params):
        address = params.get('address', params.get('offset'))
        if isinstance(address, dict):
            space = address.get('space', 'physical')
            offset = rpc_number(address.get('offset'), 'address.offset')
            if space == 'segmented':
                segment = rpc_number(address.get('segment'), 'address.segment')
                if not 0 <= segment <= 0xffff or not 0 <= offset <= 0xffff:
                    raise ValueError('segmented address segment and offset must be 16-bit values')
                offset = (segment * 16 + offset) & 0xfffff
            elif space not in ('physical', 'linear'):
                raise ValueError('address.space must be physical, linear, or segmented')
        else:
            offset = rpc_number(address, 'address')
        if offset < 0 or offset >= 1024 * 1024:
            raise ValueError('address is outside the 1 MiB address space')
        return offset

    def rpc_registers():
        return {
            'general': {
                'ax': state.GetAX(), 'bx': state.GetBX(), 'cx': state.GetCX(),
                'dx': state.GetDX(), 'sp': state.GetSP(), 'bp': state.GetBP(),
                'si': state.GetSI(), 'di': state.GetDI(),
            },
            'segments': {
                'cs': state.GetCS(), 'ds': state.GetDS(),
                'es': state.GetES(), 'ss': state.GetSS(),
            },
            'ip': state.GetIP(),
            'flags': state.GetFlags(),
            'flags_text': state.GetFlagsAsString(),
            'clock': state.GetClock(),
            'in_hlt': state.GetInHlt(),
            'state_revision': control['revision'],
        }

    def rpc_flat_registers():
        registers = rpc_registers()
        return {
            **registers['general'], **registers['segments'],
            'ip': registers['ip'], 'flags': registers['flags'],
        }

    def rpc_last_stop(kind, **details):
        # A breakpoint or deadline completes a pending step too.
        control['step'] = False
        stop = {'kind': kind, **details, 'registers': rpc_registers()}
        control['last_stop'] = stop
        operation = control['active_operation']
        if operation is not None:
            operation['done'] = True
            operation['stop_reason'] = stop
            control['active_operation'] = None
        return stop

    def rpc_resume_skip_id():
        # Execution and INT breakpoints stop before execution; watchpoints
        # stop after it. Never suppress the next instruction's memory access.
        stop = control['last_stop']
        if stop and stop.get('kind') == 'breakpoint' and 'access' not in stop:
            return stop.get('breakpoint_id')
        return None

    def rpc_start_operation(kind):
        if control['active_operation'] is not None:
            raise ValueError('an execution operation is already running')
        operation_id = f'op-{control["next_operation"]}'
        control['next_operation'] += 1
        operation = {'operation_id': operation_id, 'kind': kind, 'done': False}
        control['operations'][operation_id] = operation
        while len(control['operations']) > 256:
            oldest_id, oldest = next(iter(control['operations'].items()))
            if not oldest['done']:
                break
            del control['operations'][oldest_id]
        control['active_operation'] = operation
        return operation

    def rpc_clear_run_until():
        predicate_id = control['run_until_id']
        if predicate_id is not None and breakpoints.contains(predicate_id):
            breakpoints.delete(predicate_id)
        control['run_until_id'] = None
        control['run_until_start_clock'] = None
        control['run_until_deadline_clock'] = None
        control['run_until_requested_ns'] = None
        rpc_refresh_instruction_hooks()

    def rpc_clock_to_ns(clock):
        return (clock * 1_000_000_000) // 4_770_000

    def rpc_refresh_instruction_hooks():
        control['breakpoints_active'] = breakpoints.has_execution()
        control['memory_watchpoints_active'] = breakpoints.has_memory_access()
        control['interrupt_breakpoints_active'] = breakpoints.has_interrupt()
        control['instruction_hooks_active'] = (
            control['breakpoints_active'] or control['trace_active'])
        p.SetMemoryAccessHook(
            rpc_memory_access if control['memory_watchpoints_active'] else None)
        p.SetInterruptHook(
            rpc_interrupt if control['interrupt_breakpoints_active'] else None)
        p.SetMemoryTraceHook(
            rpc_trace_memory if control['trace_active'] else None)
        p.SetIOTraceHook(
            rpc_trace_io if control['trace_active'] else None)
        p.SetHardwareTraceHook(
            rpc_hardware_event if control['hardware_trace_active'] else None)

    def rpc_trace_event():
        registers = rpc_registers()
        physical = ((registers['segments']['cs'] << 4) + registers['ip']) & 0xfffff
        segment_base = registers['segments']['cs'] << 4
        opcode = bytes(b.ReadByte(
            (segment_base + ((registers['ip'] + i) & 0xffff)) & 0xfffff)[0]
            for i in range(8))
        return {
            'kind': 'hlt' if registers['in_hlt'] else 'instruction',
            'address': {
                'space': 'segmented', 'segment': registers['segments']['cs'],
                'offset': registers['ip'],
            },
            'physical': physical,
            'opcode_hex': opcode.hex(),
            'clock_before': registers['clock'],
            'registers_before': registers,
            'effects': [],
        }

    def rpc_trace_memory(kind, physical, old, new):
        event = control.get('trace_event')
        if event is None:
            return
        effect = {
            'kind': kind,
            'address': {'space': 'linear', 'offset': physical},
            'byte_count': 1,
        }
        if kind == 'memory_read':
            effect['data_base64'] = base64.b64encode(bytes((new,))).decode('ascii')
        else:
            effect['before_base64'] = (
                None if old is None else base64.b64encode(bytes((old,))).decode('ascii'))
            effect['after_base64'] = base64.b64encode(bytes((new,))).decode('ascii')
        event['effects'].append(effect)

    def rpc_trace_io(kind, port, value, byte_count, handled):
        event = control.get('trace_event')
        if event is None:
            return
        event['effects'].append({
            'kind': kind, 'port': port, 'byte_count': byte_count,
            'value': value, 'handled': bool(handled),
        })

    def rpc_hardware_event(*args):
        if len(args) == 1 and isinstance(args[0], dict):
            hardware_trace.capture(args[0])
            return
        kind, port, value, byte_count, handled, address, clock = args
        hardware_trace.capture({
            'kind': kind,
            'emulated_time': clock,
            'address': address,
            'port': port,
            'byte_count': byte_count,
            'value': value,
        })

    def rpc_memory_access(kind, physical, old, new, instruction_address):
        if instruction_address is None:
            return None
        access = {
            'kind': kind,
            'address': {'space': 'linear', 'offset': physical},
            'byte_count': 1,
            'instruction_address': {
                'space': 'segmented',
                'segment': instruction_address['segment'],
                'offset': instruction_address['offset'],
            },
            'new_value': new,
        }
        if old is not None:
            access['old_value'] = old
        return breakpoints.check_memory_access(
            kind, physical, access, control['skip_breakpoint_id'])

    def rpc_interrupt(number, ah, al, interrupt_state, instruction_address):
        if instruction_address is None:
            return None
        return breakpoints.check_interrupt(
            number, ah, al, rpc_flat_registers, control['skip_breakpoint_id'])

    register_access = {
        'ax': (state.GetAX, state.SetAX), 'bx': (state.GetBX, state.SetBX),
        'cx': (state.GetCX, state.SetCX), 'dx': (state.GetDX, state.SetDX),
        'sp': (state.GetSP, state.SetSP), 'bp': (state.GetBP, state.SetBP),
        'si': (state.GetSI, state.SetSI), 'di': (state.GetDI, state.SetDI),
        'cs': (state.GetCS, state.SetCS), 'ds': (state.GetDS, state.SetDS),
        'es': (state.GetES, state.SetES), 'ss': (state.GetSS, state.SetSS),
        'ip': (state.GetIP, state.SetIP), 'flags': (state.GetFlags, state.SetFlags),
    }

    def rpc_require_paused():
        if not control['paused']:
            raise ValueError('target must be paused for this operation')

    def rpc_hash(data):
        return hashlib.sha256(data).hexdigest()

    def rpc_snapshot_component(data):
        return {
            'byte_count': len(data), 'sha256': rpc_hash(data),
        }

    def rpc_video_snapshot():
        vram = bytes(scr._ram)
        text = '\n'.join(ReadTextScreen(scr)).encode('utf-8')
        snapshot_data = {'vram': vram, 'text': text}
        graphics_vram = (scr.GetGraphicsMemorySnapshot()
                         if hasattr(scr, 'GetGraphicsMemorySnapshot') else None)
        if graphics_vram is not None:
            snapshot_data['graphics_vram'] = graphics_vram
        if hasattr(scr, 'GetFontMemory'):
            snapshot_data['font'] = scr.GetFontMemory()
        control['snapshot_number'] += 1
        snapshot_id = f'snap-{control["snapshot_number"]}'
        control['snapshots'][snapshot_id] = snapshot_data
        while len(control['snapshots']) > 8:
            del control['snapshots'][next(iter(control['snapshots']))]
        result = {
            'snapshot_id': snapshot_id,
            'state_revision': control['revision'],
            'captured_ticks': state.GetClock(),
            'adapter': scr.GetName(),
            'video_mode': scr._graphics_mode,
            'columns': scr.GetTextColumns(), 'rows': 25,
            'display_address': scr._display_address & scr.GetTextAddressMask(),
            'active_page': ((scr._display_address & scr.GetTextAddressMask()) //
                            (scr.GetTextColumns() * 25 * 2)),
            'text': rpc_snapshot_component(text),
            'vram': rpc_snapshot_component(vram),
        }
        if graphics_vram is not None:
            result['graphics_vram'] = rpc_snapshot_component(graphics_vram)
            result['graphics_vram_layout'] = (
                'chain4_guest_order' if scr._graphics_mode == 0x13
                else 'planes_0_to_3_concatenated')
        if 'font' in snapshot_data:
            result['font'] = rpc_snapshot_component(snapshot_data['font'])
        if hasattr(scr, 'GetCursorInfo'):
            result['cursor'] = scr.GetCursorInfo()
        return result

    def rpc_text_address(params):
        if 'page' in params and 'display_address' in params:
            raise ValueError('page and display_address are mutually exclusive')
        columns = scr.GetTextColumns()
        page_size = columns * 25 * 2
        page_count = len(scr._ram) // page_size
        active_address = scr._display_address & scr.GetTextAddressMask()
        if 'page' in params:
            page = rpc_number(params['page'], 'page')
            if page < 0 or page >= page_count:
                raise ValueError(f'page must be 0..{page_count - 1}')
            display_address = page * page_size
        elif 'display_address' in params:
            display_address = rpc_number(params['display_address'], 'display_address')
            if display_address < 0 or display_address >= len(scr._ram):
                raise ValueError('display_address must be within video memory')
            display_address &= scr.GetTextAddressMask()
        else:
            display_address = active_address
            display_address &= scr.GetTextAddressMask()
        return columns, page_size, page_count, active_address, display_address

    def rpc_text(params):
        columns, page_size, page_count, active_address, display_address = rpc_text_address(params)
        result = {
            'adapter': scr.GetName(),
            'columns': columns, 'rows': 25,
            'page_size_bytes': page_size, 'page_count': page_count,
            'page': display_address // page_size,
            'active_page': active_address // page_size,
            'is_active_page': display_address == active_address,
            'display_address': display_address,
            'mode': scr._cga_mode.name,
            'graphics_mode': scr._graphics_mode,
            'text': ReadTextScreen(scr, display_address),
            'cells': ReadTextCells(scr, display_address),
            'state_revision': control['revision'],
        }
        if hasattr(scr, 'GetCursorInfo'):
            result['cursor'] = scr.GetCursorInfo()
        return result


    def rpc_observe_plan(address, length):
        # Resolve every span against bus priority without invoking device reads.
        # VGA planar reads latch data; mailbox/device reads may consume state.
        plan = []
        end = address + length
        while address < end:
            entry = next((item for item in b._cache
                          if item.start_addr <= address < item.end_addr), None)
            if entry is None:
                raise ValueError('observation memory is not backed by safe storage')
            stop = min(end, entry.end_addr)
            for item in b._cache:
                if address < item.start_addr < stop:
                    stop = item.start_addr
            owner = entry.device
            if owner is b._m:
                storage, offset = owner._m, address
            elif type(owner) is rom.Rom:
                storage, offset = owner._contents, address - owner._offset
            elif owner is scr and address >= scr._ram_offset:
                storage = scr._ram
                offset = (address - scr._ram_offset) % len(storage)
                # CGA mirrors its 16 KiB backing throughout the32 KiB aperture.
                stop = min(stop, address + len(storage) - offset)
            else:
                raise ValueError('observation memory maps a device without a safe peek')
            plan.append((storage, offset, stop - address))
            address = stop
        return plan

    def rpc_observe_bytes(data, **fields):
        return {**fields, 'byte_count': len(data),
                'data_base64': base64.b64encode(data).decode('ascii'),
                'data_hex': data.hex(), 'sha256': rpc_hash(data),
                'state_revision': control['revision']}

    def rpc_observe(params):
        rpc_require_paused()
        expected = rpc_number(params.get('expected_state_revision'), 'expected_state_revision')
        if expected != control['revision']:
            raise ValueError('expected state revision does not match the live state')
        if set(params) - {'expected_state_revision', 'memory', 'video_text', 'video_memory'}:
            raise ValueError('unsupported observation parameter')
        windows = params.get('memory', [])
        if not isinstance(windows, list) or len(windows) > 16:
            raise ValueError('memory must contain at most 16 windows')
        plans, total = [], 0
        for window in windows:
            if not isinstance(window, dict) or set(window) - {'address', 'offset', 'length'}:
                raise ValueError('memory window must contain only address and length')
            address = rpc_address(window)
            length = rpc_number(window.get('length', 1), 'length')
            if length < 1 or length > 65536 or address + length > 1024 * 1024:
                raise ValueError('length must be 1..65536 and stay within memory')
            total += length
            if total > 65536:
                raise ValueError('observation memory exceeds 65536 bytes')
            plans.append((address, rpc_observe_plan(address, length)))
        text_params = params.get('video_text')
        if 'video_text' in params:
            if not isinstance(text_params, dict) or set(text_params) - {'page', 'display_address'}:
                raise ValueError('video_text must be an object with page or display_address')
            rpc_text_address(text_params)
        video_memory = params.get('video_memory', False)
        if not isinstance(video_memory, bool):
            raise ValueError('video_memory must be boolean')
        # Only now copy bytes: malformed later windows/options cannot read devices.
        result = {'state_revision': control['revision'], 'registers': rpc_registers(),
                  'memory': [rpc_observe_bytes(b''.join(bytes(source[offset:offset + size])
                                      for source, offset, size in plan), address=address)
                             for address, plan in plans]}
        if text_params is not None:
            result['video_text'] = rpc_text(text_params)
        if video_memory:
            result['video_memory'] = rpc_observe_bytes(bytes(scr._ram),
                                                      address=scr._ram_offset,
                                                      adapter=scr.GetName())
        return result

    def handle_debug(request):
        global trace, hardware_trace
        method = request['method']
        params = rpc_params(request)

        if method in ('agent.capabilities', 'emulator.info'):
            return {
                'protocol': 'JSON-RPC 2.0 over localhost JSON-lines',
                'endpoint': f'127.0.0.1:{arguments.rpc_port}',
                'cpu': '8088', 'memory_bytes': 1024 * 1024,
                'video_adapters': ['CGA', 'VGA'],
                'address_spaces': ['physical', 'linear', 'segmented'],
                'limits': {'max_memory_bytes': 65536, 'max_keyboard_events': 32,
                           'max_trace_events': 65536,
                           'max_observation_windows': 16,
                           'max_observation_memory_bytes': 65536,
                           'retained_video_snapshots': 8},
                'methods': [
                    'agent.capabilities', 'emulator.info', 'state.get_registers',
                    'serial.status', 'serial.write', 'serial.read',
                    'state.get', 'state.observe', 'state.set_registers', 'session.status',
                    'memory.read', 'memory.write', 'video.text', 'video.snapshot',
                    'video.snapshot.read', 'video.history.start',
                    'video.history.read', 'video.history.stop',
                    'io.read', 'io.write', 'input.keyboard',
                    'keyboard.scancode', 'input.state',
                    'input.joystick', 'input.joystick.state', 'execution.pause',
                    'execution.continue', 'execution.go', 'execution.run_until',
                    'execution.wait', 'execution.step',
                    'breakpoints.create', 'breakpoints.list', 'breakpoints.delete',
                    'trace.start', 'trace.read', 'trace.stop',
                    'hardware.trace.start', 'hardware.trace.read',
                    'hardware.trace.stop',
                    'machine.snapshot.export', 'machine.snapshot.import',
                ],
            }

        if method == 'serial.status':
            return serial.status()
        if method == 'serial.write':
            data = params.get('data_base64')
            if not isinstance(data, str) or len(data) > 5464:
                raise ValueError('bounded base64 serial input required')
            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))

        if method in ('machine.snapshot.export', 'machine.snapshot.import'):
            rpc_require_paused()
            if rpc_number(params.get('expected_state_revision'), 'expected_state_revision') != control['revision']:
                raise ValueError('expected state revision does not match the live state')
            path = params.get('path')
            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 {'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, 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()
            control['skip_breakpoint_id'] = None
            control['trace_event'] = None
            trace = debugtrace.TraceRecorder()
            hardware_trace = debughardware.HardwareTraceRecorder()
            control['trace_active'] = False
            control['hardware_trace_active'] = False
            control['revision'] += 1
            rpc_refresh_instruction_hooks()
            return {'state_revision': control['revision'],
                    'stop_reason': rpc_last_stop('machine_snapshot_restored')}

        if method == 'session.status':
            session_id = params.get('session_id')
            if session_id is not None and session_id != 'pypc':
                raise ValueError('session_id must be pypc')
            return {
                'session_id': 'pypc',
                'state': 'stopped' if control['paused'] else 'running',
                'state_revision': control['revision'],
                'clock': state.GetClock(),
                'target': {'cpu': '8088', 'memory_bytes': 1024 * 1024,
                           'video': scr.GetName()},
                'last_stop': control['last_stop'],
            }

        if method in ('state.get_registers', 'state.get'):
            return rpc_registers()

        if method == 'state.observe':
            return rpc_observe(params)

        if method == 'state.set_registers':
            rpc_require_paused()
            expected_revision = rpc_number(
                params.get('expected_state_revision'), 'expected_state_revision')
            expected = params.get('expected')
            values = params.get('set')
            if not isinstance(expected, dict) or not isinstance(values, dict) or not values:
                raise ValueError('expected and set must be non-empty objects')
            if expected_revision != control['revision']:
                raise ValueError('expected state revision does not match the live state')
            for name, value in values.items():
                if name not in register_access:
                    raise ValueError(f'unsupported register: {name}')
                if name not in expected:
                    raise ValueError(f'register {name} is not guarded by expected')
                if rpc_number(expected[name], f'expected.{name}') != register_access[name][0]():
                    raise ValueError(f'register precondition mismatch: {name}')
                value = rpc_number(value, f'set.{name}')
                if value < 0 or value > 0xffff:
                    raise ValueError(f'set.{name} must be a 16-bit value')
            before = rpc_registers()
            for name, value in values.items():
                register_access[name][1](rpc_number(value, f'set.{name}'))
            control['revision'] += 1
            return {'before': before, 'after': rpc_registers()}

        if method == 'memory.read':
            address = rpc_address(params)
            length = rpc_number(params.get('length', 1), 'length')
            if length < 1 or length > 65536 or address + length > 1024 * 1024:
                raise ValueError('length must be 1..65536 and stay within memory')
            data = bytes(b.ReadByte(address + i)[0] for i in range(length))
            return {
                'address': address, 'byte_count': length,
                'data_base64': base64.b64encode(data).decode('ascii'),
                'data_hex': data.hex(),
                'sha256': hashlib.sha256(data).hexdigest(),
                'state_revision': control['revision'],
            }

        if method == 'memory.write':
            rpc_require_paused()
            address = rpc_address(params)
            encoded = params.get('data_base64')
            if not isinstance(encoded, str):
                raise ValueError('data_base64 is required')
            try:
                data = base64.b64decode(encoded, validate=True)
            except Exception as error:
                raise ValueError('data_base64 is invalid') from error
            if not data or len(data) > 65536 or address + len(data) > 1024 * 1024:
                raise ValueError('data must contain 1..65536 bytes and stay within memory')
            before = bytes(b.ReadByte(address + i)[0] for i in range(len(data)))
            expected_sha256 = params.get('expected_sha256')
            if expected_sha256 is not None:
                if not isinstance(expected_sha256, str) or len(expected_sha256) != 64:
                    raise ValueError('expected_sha256 must be a 64-character hexadecimal SHA-256 value')
                try:
                    int(expected_sha256, 16)
                except ValueError as error:
                    raise ValueError('expected_sha256 must be hexadecimal') from error
                if rpc_hash(before) != expected_sha256.lower():
                    raise ValueError('expected_sha256 does not match current memory')
            for i, value in enumerate(data):
                b.WriteByte(address + i, value)
            after = bytes(b.ReadByte(address + i)[0] for i in range(len(data)))
            control['revision'] += 1
            return {
                'address': address, 'byte_count': len(data),
                'before_sha256': rpc_hash(before), 'after_sha256': rpc_hash(after),
                'state_revision': control['revision'],
            }

        if method == 'video.snapshot':
            return rpc_video_snapshot()

        if method == 'video.history.start':
            capacity = rpc_number(params.get('capacity', 500000), 'capacity')
            result = video_history.start(scr, capacity)
            p._io.SetVideoWriteHook(scr, video_history.port_write)
            return result

        if method == 'video.history.read':
            cursor = params.get('cursor')
            if cursor is not None:
                cursor = rpc_number(cursor, 'cursor')
            limit = rpc_number(params.get('limit', 1024), 'limit')
            return video_history.read(params.get('history_id'), cursor, limit)

        if method == 'video.history.stop':
            result = video_history.stop()
            p._io.SetVideoWriteHook(None, None)
            return result

        if method == 'video.snapshot.read':
            snapshot_id = params.get('snapshot_id')
            component = params.get('component')
            if snapshot_id not in control['snapshots']:
                raise ValueError('snapshot_id was not found or has expired')
            if component not in control['snapshots'][snapshot_id]:
                raise ValueError('component is not available in this video snapshot')
            offset = rpc_number(params.get('offset', 0), 'offset')
            length = rpc_number(params.get('length'), 'length')
            data = control['snapshots'][snapshot_id][component]
            if offset < 0 or length < 1 or length > 65536 or offset + length > len(data):
                raise ValueError('snapshot range is invalid')
            chunk = data[offset:offset + length]
            return {
                'snapshot_id': snapshot_id, 'component': component,
                'offset': offset, 'byte_count': len(chunk),
                'component_byte_count': len(data),
                'component_sha256': rpc_hash(data),
                'data_base64': base64.b64encode(chunk).decode('ascii'),
                'sha256': rpc_hash(chunk),
            }

        if method == 'video.text':
            return rpc_text(params)

        if method == 'io.read':
            port = rpc_number(params.get('port'), 'port')
            if port < 0 or port > 0xffff:
                raise ValueError('port must be a 16-bit number')
            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':
            rpc_require_paused()
            port = rpc_number(params.get('port'), 'port')
            value = rpc_number(params.get('value'), 'value')
            if port < 0 or port > 0xffff:
                raise ValueError('port must be a 16-bit number')
            if value < 0 or value > 0xff:
                raise ValueError('value must be an 8-bit number')
            handled = p._io.Out(port, value, False)
            control['revision'] += 1
            return {
                'port': port, 'value': value, 'handled': bool(handled),
                'state_revision': control['revision'],
            }

        if method in ('input.keyboard', 'keyboard.scancode'):
            events = params.get('events')
            if events is None:
                events = [params]
            if not isinstance(events, list) or not events or len(events) > 32:
                raise ValueError('events must contain 1..32 keyboard events')
            # Validate the complete batch before queuing any input. A rejected
            # request must not leave a modifier pressed or release an existing key.
            scancodes = []
            for event in events:
                if not isinstance(event, dict):
                    raise ValueError('keyboard event must be an object')
                code = rpc_number(event.get('scan_code'), 'scan_code')
                if code < 0 or code > 0x7f:
                    raise ValueError('scan_code must be an XT make code (0..127)')
                pressed = event.get('pressed', True)
                if not isinstance(pressed, bool):
                    raise ValueError('pressed must be boolean')
                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':
            return {
                'keyboard': {'pressed_scancodes': kb.GetPressedScancodes()},
                'joysticks': joystick.input_state() if joystick else [],
                'state_revision': control['revision'],
            }

        if method in ('input.joystick', 'input.joystick.state'):
            if method == 'input.joystick':
                rpc_require_paused()
                if joystick is None:
                    raise ValueError('no game port configured')
                joystick.set_input(params)
                control['revision'] += 1
            return {'joysticks': joystick.input_state() if joystick else [],
                    'state_revision': control['revision']}

        if method == 'breakpoints.create':
            result = breakpoints.create(params)
            rpc_refresh_instruction_hooks()
            return result

        if method == 'breakpoints.list':
            return {'breakpoints': breakpoints.list()}

        if method == 'breakpoints.delete':
            breakpoint_id = params.get('breakpoint_id')
            if not isinstance(breakpoint_id, str):
                raise ValueError('breakpoint_id is required')
            breakpoints.delete(breakpoint_id)
            rpc_refresh_instruction_hooks()
            return {'breakpoint_id': breakpoint_id, 'deleted': True}

        if method == 'trace.start':
            rpc_require_paused()
            detail = params.get('detail', 'normal')
            instruction_count = rpc_number(
                params.get('instruction_count', 256), 'instruction_count')
            result = trace.start(detail, instruction_count)
            control['trace_active'] = True
            rpc_refresh_instruction_hooks()
            return result

        if method == 'trace.read':
            limit = rpc_number(params.get('limit', 128), 'limit')
            return trace.read(params.get('cursor'), limit)

        if method == 'trace.stop':
            result = trace.stop()
            control['trace_active'] = False
            rpc_refresh_instruction_hooks()
            return result

        if method == 'hardware.trace.start':
            result = hardware_trace.start(
                params.get('capacity', 4096),
                params.get('include_io', True),
                params.get('include_irq', True),
                params.get('ports'), params.get('irqs'))
            control['hardware_trace_active'] = True
            rpc_refresh_instruction_hooks()
            return result

        if method == 'hardware.trace.read':
            limit = rpc_number(params.get('limit', 128), 'limit')
            return hardware_trace.read(params.get('cursor'), limit)

        if method == 'hardware.trace.stop':
            result = hardware_trace.stop()
            control['hardware_trace_active'] = False
            rpc_refresh_instruction_hooks()
            return result

        if method == 'execution.wait':
            operation_id = params.get('operation_id')
            if not isinstance(operation_id, str):
                raise ValueError('operation_id is required')
            timeout_ms = rpc_number(params.get('timeout_ms', 0), 'timeout_ms')
            if timeout_ms < 0 or timeout_ms > 60000:
                raise ValueError('timeout_ms must be 0..60000')
            operation = control['operations'].get(operation_id)
            if operation is None:
                raise ValueError('operation_id was not found')
            if not operation['done']:
                return {'running': True}
            return {
                'state': 'stopped',
                'stop_reason': operation['stop_reason'],
            }

        if method == 'execution.pause':
            if control['paused']:
                return {'paused': True, **rpc_registers()}
            rpc_clear_run_until()
            control['paused'] = True
            control['step'] = False
            control['skip_breakpoint_id'] = None
            rpc_last_stop('pause')
            return {'paused': True, **rpc_registers()}

        if method in ('execution.continue', 'execution.go'):
            operation = rpc_start_operation('continue')
            control['paused'] = False
            control['step'] = False
            control['skip_breakpoint_id'] = rpc_resume_skip_id()
            control['last_stop'] = None
            return {
                'operation_id': operation['operation_id'], 'state': 'running',
                'paused': False, **rpc_registers(),
            }

        if method == 'execution.run_until':
            rpc_require_paused()
            predicate = params.get('predicate')
            if not isinstance(predicate, dict):
                raise ValueError('predicate must be an object')
            if predicate.get('once', False):
                raise ValueError('run_until predicates are always one-shot')
            requested_ns = None
            if 'max_emulated_ns' in params:
                requested_ns = rpc_number(
                    params.get('max_emulated_ns'), 'max_emulated_ns')
                if requested_ns < 1 or requested_ns > 0xffffffffffffffff:
                    raise ValueError('max_emulated_ns must be 1..2^64-1')
            predicate_params = dict(predicate)
            predicate_params['once'] = True
            predicate_result = breakpoints.create(
                predicate_params, id_prefix='until', private=True)
            try:
                operation = rpc_start_operation('run_until')
            except Exception:
                breakpoints.delete(predicate_result['breakpoint_id'])
                raise
            control['run_until_id'] = predicate_result['breakpoint_id']
            deadline_clock = None
            if requested_ns is not None:
                cycles = max(1, (requested_ns * 4_770_000 +
                                 1_000_000_000 - 1) // 1_000_000_000)
                deadline_clock = state.GetClock() + cycles
            control['run_until_start_clock'] = state.GetClock()
            control['run_until_deadline_clock'] = deadline_clock
            control['run_until_requested_ns'] = requested_ns
            rpc_refresh_instruction_hooks()
            control['paused'] = False
            control['step'] = False
            control['skip_breakpoint_id'] = rpc_resume_skip_id()
            control['last_stop'] = None
            return {
                'operation_id': operation['operation_id'],
                'predicate_id': predicate_result['breakpoint_id'],
                'state': 'running',
                **({'max_emulated_ns': requested_ns}
                   if requested_ns is not None else {}),
            }

        if method == 'execution.step':
            mode = params.get('mode', 'into')
            if mode != 'into':
                raise ValueError('execution.step supports mode=into only')
            if not control['paused']:
                raise ValueError('execution.step requires a paused emulator')
            operation = rpc_start_operation('step')
            rpc_clear_run_until()
            control['step'] = True
            control['paused'] = False
            control['skip_breakpoint_id'] = rpc_resume_skip_id()
            control['last_stop'] = None
            return {'stepping': True, 'operation_id': operation['operation_id'],
                    **rpc_registers()}

        raise LookupError(f'unknown method: {method}')

    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 is not None and debug.has_pending():
            debug.process_pending(handle_debug)
        if control['paused']:
            time.sleep(0.001)
            continue
        # With no instruction-boundary observer active, keep a small batch in
        # one Python loop. Devices still tick after every instruction inside
        # CPU.Tick. A newly arrived RPC waits for at most 64 instructions;
        # enabling any exact debugger feature immediately selects the path
        # below on the next outer iteration.
        if (not control['breakpoints_active']
                and not control['instruction_hooks_active']
                and not control['memory_watchpoints_active']
                and not control['interrupt_breakpoints_active']
                and not control['step']
                and control['run_until_deadline_clock'] is None
                and control['skip_breakpoint_id'] is None):
            executed = 0
            rc = 0
            tick = p.Tick
            for _normal_tick in range(64):
                rc = tick()
                if rc == -1:
                    break
                executed += 1
            control['revision'] += executed
            if rc == -1:
                break
            cur_cycles = state.GetClock()
            c_diff = cur_cycles - p_cycles
            if c_diff >= 4700000:
                p_cycles = cur_cycles
                now = time.time()
                print(f'{100 * c_diff / (now - p_time) / 4700000:.2f}%')
                p_time = now
            continue
        if control['breakpoints_active']:
            breakpoint = breakpoints.check(
                state.GetCS(), state.GetIP(), rpc_flat_registers,
                control['skip_breakpoint_id'])
            # Keep the resume exemption through Tick's interrupt hook.
            if breakpoint is not None:
                control['paused'] = True
                control['step'] = False
                is_run_until = breakpoint['breakpoint_id'] == control['run_until_id']
                if control['run_until_id'] is not None:
                    rpc_clear_run_until()
                details = {
                    'breakpoint_id': breakpoint['breakpoint_id'],
                    'address': breakpoint['address'],
                    'hit_count': breakpoint['hit_count'],
                }
                if is_run_until:
                    details['predicate_id'] = breakpoint['breakpoint_id']
                rpc_last_stop('run_until' if is_run_until else 'breakpoint', **details)
                continue
        if control['instruction_hooks_active'] and control['trace_active']:
            trace_before = rpc_trace_event()
        else:
            trace_before = None
        control['trace_event'] = trace_before
        # print(f'{state.GetCS():04x}:{state.GetIP():04x} {GetRegisters(state)}')
        rc = p.Tick()
        if rc == -1:
            control['trace_event'] = None
            break
        control['revision'] += 1
        memory_stop = (p.ConsumeMemoryWriteStop()
                       if control['memory_watchpoints_active'] else None)
        if memory_stop is not None:
            control['paused'] = True
            is_run_until = memory_stop['breakpoint_id'] == control['run_until_id']
            if control['run_until_id'] is not None:
                rpc_clear_run_until()
            details = {
                'breakpoint_id': memory_stop['breakpoint_id'],
                'address': memory_stop['address'],
                'length': memory_stop['length'],
                'hit_count': memory_stop['hit_count'],
                'access': memory_stop['access'],
            }
            if is_run_until:
                details['predicate_id'] = memory_stop['breakpoint_id']
            rpc_last_stop('run_until' if is_run_until else 'breakpoint', **details)
            rpc_refresh_instruction_hooks()
        interrupt_stop = (p.ConsumeInterruptStop()
                          if control['interrupt_breakpoints_active'] else None)
        if memory_stop is None and interrupt_stop is not None:
            control['paused'] = True
            is_run_until = interrupt_stop['breakpoint_id'] == control['run_until_id']
            if control['run_until_id'] is not None:
                rpc_clear_run_until()
            details = {
                'breakpoint_id': interrupt_stop['breakpoint_id'],
                'event': interrupt_stop['event'],
                'hit_count': interrupt_stop['hit_count'],
            }
            if is_run_until:
                details['predicate_id'] = interrupt_stop['breakpoint_id']
            rpc_last_stop('run_until' if is_run_until else 'breakpoint', **details)
            rpc_refresh_instruction_hooks()
        deadline_clock = control['run_until_deadline_clock']
        if (memory_stop is None and interrupt_stop is None and
                deadline_clock is not None and state.GetClock() >= deadline_clock):
            start_clock = control['run_until_start_clock']
            requested_ns = control['run_until_requested_ns']
            actual_clock = state.GetClock()
            rpc_clear_run_until()
            control['paused'] = True
            actual_ns = rpc_clock_to_ns(actual_clock)
            deadline_ns = rpc_clock_to_ns(deadline_clock)
            details = {
                'emulated_time_ns': actual_ns,
                'emulated_time_limit': {
                    'requested_duration_ns': requested_ns,
                    'start_emulated_time_ns': rpc_clock_to_ns(start_clock),
                    'deadline_emulated_time_ns': deadline_ns,
                    'actual_stop_emulated_time_ns': actual_ns,
                    'reached': True,
                    'overshoot_ns': max(0, actual_ns - deadline_ns),
                },
            }
            rpc_last_stop('emulated_time_limit', **details)
        control['skip_breakpoint_id'] = None
        if trace_before is not None:
            interrupt = p.GetLastInterruptDispatch()
            if interrupt is not None:
                trace_before['kind'] = 'interrupt_dispatch'
                trace_before['interrupt'] = interrupt
            trace_before['clock_after'] = state.GetClock()
            trace_before['clock_delta'] = (
                trace_before['clock_after'] - trace_before['clock_before'])
            if trace.detail in ('normal', 'long'):
                trace_before['registers_after'] = rpc_registers()
            elif trace.detail == 'csip':
                trace_before.pop('registers_before', None)
            trace.capture(trace_before)
            control['trace_event'] = None
            control['trace_active'] = trace.active
            rpc_refresh_instruction_hooks()
        if control['step']:
            control['step'] = False
            control['paused'] = True
            rpc_last_stop('step')
        cur_cycles = state.GetClock()
        c_diff = cur_cycles - p_cycles
        if c_diff >= 4700000:
            p_cycles = cur_cycles
            now = time.time()
            ##print(f'\033[1;82H{100 * c_diff / (now - p_time) / 4700000:.2f}%', end='')
            print(f'{100 * c_diff / (now - p_time) / 4700000:.2f}%')
            p_time = now

except KeyboardInterrupt as ki:
    print('exit')