emrg: Windows TUI Unicode input via ReadConsoleInputW

Agent.md

@@ -93,7 +93,7 @@ Community needs voiced in HN agent-UI discussions map directly to EMRG's design:
 pkill -f "emrg.server"; rm -f ~/.emrg/emrgd.port; python -m emrg
 ```
 
-Python: `uv run pytest tests/ -v` (548) — import check: `uv run python -c "from emrg.client.app import run_client"`
+Python: `uv run pytest tests/ -v` (572) — import check: `uv run python -c "from emrg.client.app import run_client"`
 GUI: `cd emrg/gui && npm test` (91: 22 daemon_client + 22 app-commands + 22 renderer smoke + 15 i18n + 7 integration + 3 commands) — syntax: `node --check main.js preload.js daemon_client.js renderer/js/*.js`
 CI: `uv run pytest` + GUI tests + **actionlint workflow lint** (`rhysd/actionlint@v1.7.12` gate, #444 — workflow 解析错误在 PR CI 即失败，如 `if:` secrets 上下文)
 Re-trigger: `scripts/re-trigger-ci.sh [branch]` (workflow_dispatch, #527 — 替代空 commit 重触发：Actions outage 会整段丢弃 push 事件，dispatch 走 API 路径不受影响)

README.md

@@ -273,7 +273,7 @@ EMRG doesn't just keep up — it catches up on its own.
 git clone https://github.com/argszero/emrg.git
 cd emrg
 uv sync # install deps
-uv run pytest tests/ -v # run tests (currently 548 items)
+uv run pytest tests/ -v # run tests (currently 572 items)
 uv run python -m emrg # launch TUI
 # CI includes actionlint workflow gate (#444): workflow parse errors fail PR CI
 

emrg/client/app.py

@@ -896,12 +896,24 @@ def _stdin_reader() -> None:
 
  if sys.platform == "win32":
  def _win_stdin_loop() -> None:
+ # ReadConsoleInputW 主路径（rant 2026-08-07T21:35:47）：
+ # os.read 字节流按 OEM 代码页（中文系统 GBK）交付 IME 字符，
+ # UTF-8 假设的输入链必然乱码；宽字符 API 直接给 UTF-16。
+ from emrg.client.python_tui.win32 import (
+ flush_console_input,
+ read_console_unicode,
+ )
+ try:
+ flush_console_input(stdin_fd) # 丢弃切换前的字节流残余
+ except (OSError, ValueError):
+ pass
  while not _win_stdin_stop.is_set():
  try:
- data = os.read(stdin_fd, 4096)
- if not data:
- break
- loop.call_soon_threadsafe(stdin_queue.put_nowait, data)
+ data = read_console_unicode(stdin_fd)
+ if data:
+ loop.call_soon_threadsafe(stdin_queue.put_nowait, data)
+ else:
+ _win_stdin_stop.wait(0.005) # 防忙轮询
  except (OSError, ValueError):
  break
 

emrg/client/python_tui/win32.py

@@ -11,24 +11,38 @@
  - set the fd to binary mode so CRLF translation is off (key sequences like
  the arrow prefix ESC [ A arrive byte-identical to POSIX)
 
-ENABLE_VIRTUAL_TERMINAL_INPUT (rant 2026-08-07T10:38:21) is required:
-without it conhost delivers keystrokes in the OEM code page (GBK on Chinese
-systems → UTF-8-assuming input chain garbles CJK IME text) and arrow keys
-arrive as legacy 0xE0 scan codes instead of ANSI ESC [ A/B/C/D. With VT
-input enabled the console delivers UTF-8 + standard ANSI sequences, fixing
-both problems at once. Pre-Win10-1607 consoles reject the flag — fall back
-to window-input-only raw mode (InputParser defensively translates legacy
-scan codes).
+ENABLE_VIRTUAL_TERMINAL_INPUT (rant 2026-08-07T10:38:21) makes conhost
+deliver function/arrow keys as ANSI ESC [ A/B/C/D sequences (fixing the
+legacy 0xE0 scan-code problem) and is kept as the raw-mode input flag.
+However it does NOT fix CJK IME input: character keystrokes still arrive
+through the os.read byte stream encoded in the console input code page
+(GBK/CP936 on Chinese systems), which the UTF-8-assuming input chain
+garbles (rant 2026-08-07T21:35:47, host-verified on v0.2.11).
+
+The reliable path for Unicode characters is the wide-char API
+ReadConsoleInputW — it returns KEY_EVENT_RECORDs whose UnicodeChar holds
+the IME-confirmed UTF-16 character. read_console_unicode() is the primary
+stdin reader on Windows; the VT-input byte path remains only as a fallback
+for older conhost versions.
+
+The module imports cleanly on POSIX (guarded msvcrt / windll access) so
+its pure KEY_EVENT → bytes translation logic is unit-testable everywhere.
 """
 
 from __future__ import annotations
 
 import ctypes
-import msvcrt
 import os
 from ctypes import wintypes
 from typing import Any
 
+try:
+ import msvcrt
+except ImportError: # POSIX — module imported only for its pure helpers
+ msvcrt = None # type: ignore[assignment]
+
+from emrg.client.python_tui.events import _LEGACY_SCAN_TO_ANSI
+
 # ── Console input/output mode flags (wincon.h) ───────────────────────────
 ENABLE_PROCESSED_INPUT = 0x0001
 ENABLE_LINE_INPUT = 0x0002
@@ -52,13 +66,18 @@ class Win32Console:
  """Raw-mode / VT support for a console handle (ctypes, no pywin32)."""
 
  def __init__(self) -> None:
- self._kernel32 = ctypes.windll.kernel32
+ try:
+ self._kernel32 = ctypes.windll.kernel32
+ except AttributeError: # POSIX — imported only for pure helpers
+ self._kernel32 = None
  self._saved_modes: dict[int, tuple[int, int]] = {} # fd -> (in, out)
 
  # -- internal helpers ------------------------------------------------
  @staticmethod
  def _fd_to_handle(fd: int) -> int | None:
  """Return the Win32 HANDLE for a std fd (via _get_osfhandle)."""
+ if msvcrt is None: # POSIX
+ return None
  try:
  return msvcrt.get_osfhandle(fd)
  except OSError:
@@ -122,3 +141,115 @@ def active(self) -> bool:
 # _enter_raw_mode / _exit_raw_mode calls (a fresh instance per call would
 # lose _saved_modes and fail to restore the terminal).
 win32_console = Win32Console()
+
+
+# ── Wide-char input (rant 2026-08-07T21:35:47) ─────────────────────────
+# CJK IME-confirmed text must be read via ReadConsoleInputW (UTF-16
+# KEY_EVENT_RECORDs), not the os.read byte stream (OEM code page — GBK on
+# Chinese systems — garbles the UTF-8-assuming input chain).
+
+KEY_EVENT = 0x0001
+
+
+class _KEY_EVENT_RECORD(ctypes.Structure):
+ # ABI-critical: use FIXED-WIDTH ctypes (c_int/c_uint/c_ushort), NOT
+ # wintypes.BOOL/DWORD — those are c_long/c_ulong, 4B on Windows but 8B
+ # on LP64 POSIX, inflating the struct and breaking Win32 ABI parity
+ # (review ❌ finding on #553: sizes were 32/40 instead of 16/20).
+ _fields_ = [
+ ("bKeyDown", ctypes.c_int), # Win32 BOOL
+ ("wRepeatCount", ctypes.c_ushort),
+ ("wVirtualKeyCode", ctypes.c_ushort),
+ ("wVirtualScanCode", ctypes.c_ushort),
+ # uChar is the union { WCHAR UnicodeChar; CHAR AsciiChar; } — we
+ # read the UnicodeChar view (c_ushort, 2B, matches WCHAR everywhere).
+ ("uChar", ctypes.c_ushort),
+ ("dwControlKeyState", ctypes.c_uint), # Win32 DWORD
+ ]
+
+
+class _INPUT_RECORD(ctypes.Structure):
+ _fields_ = [
+ ("EventType", wintypes.WORD),
+ ("Event", _KEY_EVENT_RECORD),
+ ]
+
+
+try:
+ _k32 = ctypes.windll.kernel32
+except AttributeError: # POSIX — no kernel32
+ _k32 = None
+
+_ReadConsoleInputW = None
+_FlushConsoleInputBuffer = None
+if _k32 is not None:
+ _ReadConsoleInputW = _k32.ReadConsoleInputW
+ _ReadConsoleInputW.argtypes = [
+ wintypes.HANDLE,
+ ctypes.POINTER(_INPUT_RECORD),
+ wintypes.DWORD,
+ ctypes.POINTER(wintypes.DWORD),
+ ]
+ _ReadConsoleInputW.restype = wintypes.BOOL
+ _FlushConsoleInputBuffer = _k32.FlushConsoleInputBuffer
+ _FlushConsoleInputBuffer.argtypes = [wintypes.HANDLE]
+ _FlushConsoleInputBuffer.restype = wintypes.BOOL
+
+
+def _key_event_to_bytes(b_key_down: bool, unicode_char: str, scan_code: int) -> bytes:
+ """Translate one KEY_EVENT_RECORD to input bytes (pure, unit-testable).
+
+ - key-up events are dropped (avoids duplicate characters)
+ - UnicodeChar != 0 → UTF-8 encode (covers ASCII, Ctrl chars, and
+ IME-confirmed CJK — the whole point of ReadConsoleInputW)
+ - UnicodeChar == 0 (function/arrow keys) → legacy scan-code table
+ (same _LEGACY_SCAN_TO_ANSI map the VT-input path uses)
+ """
+ if not b_key_down:
+ return b""
+ if unicode_char and unicode_char != "\x00":
+ return unicode_char.encode("utf-8", errors="replace")
+ return _LEGACY_SCAN_TO_ANSI.get(scan_code, b"")
+
+
+def flush_console_input(fd: int) -> None:
+ """Flush the console input buffer (drop stale byte-stream residue)."""
+ if _FlushConsoleInputBuffer is None:
+ return
+ handle = Win32Console._fd_to_handle(fd)
+ if handle is not None:
+ _FlushConsoleInputBuffer(handle)
+
+
+def read_console_unicode(fd: int, max_events: int = 32) -> bytes:
+ """Read console input via ReadConsoleInputW, returning UTF-8 bytes.
+
+ Character keys (incl. IME-confirmed CJK) are UTF-8 encoded; function
+ keys (UnicodeChar == 0) are translated to ANSI CSI via the scan-code
+ table. Returns b"" when no key events are pending — callers should
+ sleep briefly to avoid busy-polling.
+ """
+ if _ReadConsoleInputW is None:
+ return b""
+ handle = Win32Console._fd_to_handle(fd)
+ if handle is None:
+ return b""
+ records = (_INPUT_RECORD * max_events)()
+ n_read = wintypes.DWORD(0)
+ if not _ReadConsoleInputW(handle, records, max_events, ctypes.byref(n_read)):
+ return b""
+ out = bytearray()
+ for i in range(n_read.value):
+ rec = records[i]
+ if rec.EventType != KEY_EVENT:
+ continue
+ key = rec.Event
+ uchar = int(key.uChar) # c_ushort → int (0 = no char / function key)
+ out.extend(
+ _key_event_to_bytes(
+ bool(key.bKeyDown),
+ chr(uchar) if uchar else "\x00",
+ int(key.wVirtualScanCode),
+ )
+ )
+ return bytes(out)

tests/test_win32_input.py

@@ -0,0 +1,164 @@
+"""Tests for win32 wide-char input (rant 2026-08-07T21:35:47).
+
+ReadConsoleInputW returns KEY_EVENT_RECORDs whose UnicodeChar holds the
+IME-confirmed UTF-16 character (the os.read byte path delivers OEM code
+page bytes — GBK on Chinese systems — garbling the UTF-8 input chain).
+The ctypes kernel32 call itself is Windows-only, so the tests exercise the
+pure _key_event_to_bytes translation: CJK chars, ASCII/Ctrl chars, and
+function keys via the legacy scan-code table.
+"""
+
+from emrg.client.python_tui.win32 import _key_event_to_bytes
+
+
+class TestKeyEventToBytes:
+ """KEY_EVENT_RECORD → input bytes translation (positive + negative)."""
+
+ def test_ime_cjk_char(self):
+ """IME 确认后的中文 → UTF-8 字节（不再按 GBK 交付）。"""
+ assert _key_event_to_bytes(True, "中", 0) == "中".encode("utf-8")
+
+ def test_ime_cjk_multi_char(self):
+ """一次组合确认多个汉字（如"你好"）逐字符编码。"""
+ assert _key_event_to_bytes(True, "你", 0) == "你".encode("utf-8")
+ assert _key_event_to_bytes(True, "好", 0) == "好".encode("utf-8")
+
+ def test_ascii_char(self):
+ assert _key_event_to_bytes(True, "a", 0) == b"a"
+
+ def test_digit_char(self):
+ assert _key_event_to_bytes(True, "7", 0) == b"7"
+
+ def test_ctrl_char(self):
+ """Ctrl+C 等控制字符照常交付（0x03 < 0x20 由上层解释）。"""
+ assert _key_event_to_bytes(True, "\x03", 0) == b"\x03"
+
+ def test_return_key(self):
+ assert _key_event_to_bytes(True, "\r", 0) == b"\r"
+
+ def test_backspace_key(self):
+ assert _key_event_to_bytes(True, "\x08", 0) == b"\x08"
+
+ def test_arrow_up_scan_code(self):
+ """UnicodeChar==0（功能键）→ 走 scan-code 翻译表（与 #546 同表）。"""
+ assert _key_event_to_bytes(True, "\x00", 0x48) == b"\x1b[A"
+
+ def test_arrow_down_scan_code(self):
+ assert _key_event_to_bytes(True, "\x00", 0x50) == b"\x1b[B"
+
+ def test_arrow_left_right_scan_code(self):
+ assert _key_event_to_bytes(True, "\x00", 0x4B) == b"\x1b[D"
+ assert _key_event_to_bytes(True, "\x00", 0x4D) == b"\x1b[C"
+
+ def test_home_end_scan_code(self):
+ assert _key_event_to_bytes(True, "\x00", 0x47) == b"\x1b[H"
+ assert _key_event_to_bytes(True, "\x00", 0x4F) == b"\x1b[F"
+
+ def test_pageup_pagedown_scan_code(self):
+ assert _key_event_to_bytes(True, "\x00", 0x49) == b"\x1b[5~"
+ assert _key_event_to_bytes(True, "\x00", 0x51) == b"\x1b[6~"
+
+ def test_unknown_scan_code(self):
+ """未识别扫描码 → 空（不产生垃圾字节）。"""
+ assert _key_event_to_bytes(True, "\x00", 0x99) == b""
+
+ def test_key_up_drops_cjk(self):
+ """key-up 事件丢弃，避免重复字符。"""
+ assert _key_event_to_bytes(False, "中", 0x48) == b""
+
+ def test_key_up_drops_scan_code(self):
+ assert _key_event_to_bytes(False, "\x00", 0x48) == b""
+
+ def test_key_up_drops_ascii(self):
+ assert _key_event_to_bytes(False, "a", 0) == b""
+
+ def test_zero_unicode_with_unknown_scan(self):
+ """修饰键（UnicodeChar==0 且非功能键）→ 空。"""
+ assert _key_event_to_bytes(True, "\x00", 0) == b""
+
+
+import ctypes # noqa: E402
+
+import emrg.client.python_tui.win32 as win32 # noqa: E402
+
+
+class TestInputRecordLayout:
+ """ctypes struct must match the Win32 ABI on every platform (review ❌
+ finding: wintypes.WCHAR/c_wchar is 4B on POSIX, inflating the layout;
+ uChar is now c_ushort so sizes are platform-stable and simulatable)."""
+
+ def test_key_event_record_size(self):
+ assert ctypes.sizeof(win32._KEY_EVENT_RECORD) == 16
+
+ def test_input_record_size(self):
+ assert ctypes.sizeof(win32._INPUT_RECORD) == 20
+
+ def test_event_offset(self):
+ # EventType (WORD) at 0; union is 4-aligned (DWORD members) → 2B pad,
+ # Event at offset 4, INPUT_RECORD = 20B total (Win32 ABI)
+ assert win32._INPUT_RECORD.Event.offset == 4
+
+
+class TestReadConsoleUnicodeLoop:
+ """Simulated INPUT_RECORD buffers through the real read loop — the part
+ the pure-function tests cannot reach. _ReadConsoleInputW is faked to
+ write a record into the buffer the same way kernel32 would."""
+
+ @staticmethod
+ def _monkeypatch(monkeypatch, records):
+ monkeypatch.setattr(
+ win32.Win32Console, "_fd_to_handle", staticmethod(lambda fd: 12345)
+ )
+
+ def _set_path(obj, path, value):
+ # ctypes setattr with a dotted name silently no-ops — traverse
+ parts = path.split(".")
+ for p in parts[:-1]:
+ obj = getattr(obj, p)
+ setattr(obj, parts[-1], value)
+
+ def fake_read(handle, buf, max_events, n_read):
+ rec = ctypes.cast(buf, ctypes.POINTER(win32._INPUT_RECORD)).contents
+ for field, value in records:
+ _set_path(rec, field, value)
+ # n_read arrives as the byref() CArgObject wrapping the DWORD
+ ctypes.cast(n_read, ctypes.POINTER(ctypes.c_uint32)).contents.value = 1
+ return True
+
+ monkeypatch.setattr(win32, "_ReadConsoleInputW", fake_read)
+
+ def test_cjk_char_through_loop(self, monkeypatch):
+ self._monkeypatch(monkeypatch, [
+ ("EventType", win32.KEY_EVENT),
+ ("Event.bKeyDown", True),
+ ("Event.wVirtualScanCode", 0),
+ ("Event.uChar", ord("中")),
+ ])
+ assert win32.read_console_unicode(0) == "中".encode("utf-8")
+
+ def test_arrow_scan_code_through_loop(self, monkeypatch):
+ self._monkeypatch(monkeypatch, [
+ ("EventType", win32.KEY_EVENT),
+ ("Event.bKeyDown", True),
+ ("Event.wVirtualScanCode", 0x48),
+ ("Event.uChar", 0),
+ ])
+ assert win32.read_console_unicode(0) == b"\x1b[A"
+
+ def test_key_up_dropped_through_loop(self, monkeypatch):
+ self._monkeypatch(monkeypatch, [
+ ("EventType", win32.KEY_EVENT),
+ ("Event.bKeyDown", False),
+ ("Event.wVirtualScanCode", 0x48),
+ ("Event.uChar", ord("中")),
+ ])
+ assert win32.read_console_unicode(0) == b""
+
+ def test_mouse_event_skipped_through_loop(self, monkeypatch):
+ self._monkeypatch(monkeypatch, [
+ ("EventType", 0x0002), # MOUSE_EVENT — not a key, must be skipped
+ ("Event.bKeyDown", True),
+ ("Event.wVirtualScanCode", 0),
+ ("Event.uChar", ord("a")),
+ ])
+ assert win32.read_console_unicode(0) == b""