diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7269b1988..ac41d9c1f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -60,6 +60,14 @@ jobs: - uses: actions/setup-python@v5 with: python-version: '3.14' + - name: Install Neovim + run: | + curl --fail --location --silent --show-error \ + https://github.com/neovim/neovim/releases/download/v0.12.4/nvim-linux-x86_64.tar.gz \ + --output "$RUNNER_TEMP/neovim.tar.gz" + tar --extract --gzip --file "$RUNNER_TEMP/neovim.tar.gz" --directory "$RUNNER_TEMP" + echo "$RUNNER_TEMP/nvim-linux-x86_64/bin" >> "$GITHUB_PATH" + "$RUNNER_TEMP/nvim-linux-x86_64/bin/nvim" --version - uses: ./.github/actions/core-from-main with: path: .artifacts/core @@ -83,4 +91,4 @@ jobs: # xdist reports a collection error as a worker crash; collect alone first so it reads as one. - run: .venv/bin/pytest --collect-only -q # A test stuck past 120 s dumps every thread, so the red names where it hung. - - run: .venv/bin/pytest -n auto --basetemp .artifacts/pilots -o faulthandler_timeout=120 -rfE + - run: .venv/bin/pytest -n auto --basetemp .artifacts/pilots -o faulthandler_timeout=120 -rfE -vv --tb=short diff --git a/pyproject.toml b/pyproject.toml index f709fc75e..0c7948e4b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,7 @@ dependencies = [ "ui-theme", "formbridge==0.2.0", "objectstate==5.0.0", + "msgpack==1.2.2", ] [tool.uv.workspace] diff --git a/src/toad/acp/agent_session.py b/src/toad/acp/agent_session.py index c8bfed59d..c8cc30efe 100644 --- a/src/toad/acp/agent_session.py +++ b/src/toad/acp/agent_session.py @@ -110,7 +110,9 @@ class FailedSessionDisposition(UnavailableSessionDisposition): @property def status(self) -> SessionDisconnected: - return SessionDisconnected(self.failure.message) + return SessionDisconnected( + self.failure.message, details=tuple(self.failure.details.splitlines()) + ) async def present_failure(self, failure: AgentFail, view) -> None: if failure is self.failure: diff --git a/src/toad/key_modes.py b/src/toad/key_modes.py index 031f56f2a..7020e8b79 100644 --- a/src/toad/key_modes.py +++ b/src/toad/key_modes.py @@ -1,35 +1,32 @@ -"""Keyboard profiles compose the native editor modes; no second editor state.""" +"""Saved profiles select an editing owner, not a handwritten command subset.""" from __future__ import annotations from agent_comms.declared_choice import DeclaredChoice -from textual.document._editing import ( - EditingMode, - VimInsert as InsertEditing, - VimMode as ModalEditing, - VimNormal as NormalEditing, - VimVisual as VisualEditing, -) +from textual.document._editing import EditingMode +from toad.neovim_editing import NeovimEditing + class KeyMode(EditingMode, DeclaredChoice, affix=""): - """A preference selects the initial native mode, whose own transitions keep editing.""" + """The saved preference selects native editing or a real Neovim initial state.""" + class Standard(KeyMode): - """Ordinary text entry and navigation.""" + """Ordinary native text entry and navigation.""" label = "STANDARD" -class VimMode(KeyMode, ModalEditing): - """The shared modal-key capability; the native key handler keeps this abstract.""" +class VimInsert(NeovimEditing, KeyMode): + """Start Neovim in Insert mode.""" + initial_keys = "i" -class VimInsert(VimMode, InsertEditing): - """Start modal editing in Insert.""" +class VimNormal(NeovimEditing, KeyMode): + """Start Neovim in Normal mode.""" -class VimNormal(VimMode, NormalEditing): - """Start modal editing in Normal.""" +class VimVisual(NeovimEditing, KeyMode): + """Start Neovim in Visual mode.""" -class VimVisual(VimNormal, VisualEditing): - """Start modal editing in Visual.""" + initial_keys = "v" diff --git a/src/toad/keymap.py b/src/toad/keymap.py index c1ddddd1a..81bae44fb 100644 --- a/src/toad/keymap.py +++ b/src/toad/keymap.py @@ -113,8 +113,7 @@ def capture(cls, modes: tuple[type[KeyMode], ...]) -> KeymapCatalog: for binding in declaration.exclusive_bindings(native_mode)) owned = [application] for owner in DOMNode.contexts(): - defaults = (*(binding for _, binding in owner.binding_defaults(native_mode)), - *owner.editor_command_bindings(native_mode)) + defaults = tuple(binding for _, binding in owner.binding_defaults(native_mode)) if not issubclass(owner, ModalScreen): defaults = (*defaults, *application) owned.append(defaults) diff --git a/src/toad/neovim_editing.lua b/src/toad/neovim_editing.lua new file mode 100644 index 000000000..7cc7bebc6 --- /dev/null +++ b/src/toad/neovim_editing.lua @@ -0,0 +1,186 @@ +-- Application integration, not a Vim interpreter. Neovim interprets every key. +local mappings, channel = ... +local api = vim.api +local namespace = api.nvim_create_namespace('toad-editor') +vim.opt.swapfile = false +vim.opt.undofile = false +vim.opt.writebackup = false +vim.opt.backup = false +vim.opt.modeline = false +vim.opt.hidden = true +vim.opt.clipboard = '' +vim.opt.shortmess:append('I') +vim.bo.buftype = 'nofile' +vim.bo.bufhidden = 'hide' + +-- Admission is not completion. Only SafeState, which excludes typeahead, +-- mappings and unfinished commands, may certify a completed generation. +local generation, completed = 0, -1 +local scheduled = false +local function location(position) + local row = position[2] - 1 + local line = api.nvim_buf_get_lines(0, row, row + 1, true)[1] + local byte = math.max(0, math.min(position[3] - 1, #line)) + return {row, vim.str_utfindex(line, 'utf-32', byte, false)} +end +local function state() + local cursor, anchor = vim.fn.getpos('.'), vim.fn.getpos('v') + local code = vim.fn.mode(1) + local kind = ({v = 'v', V = 'V', ['\022'] = '\022', + s = 'v', S = 'V', ['\019'] = '\022'})[code] + local regions, selected = {}, '' + if kind then + for _, region in ipairs(vim.fn.getregionpos(anchor, cursor, {type = kind, eol = true})) do + local first, last = location(region[1]), location(region[2]) + last[2] = last[2] + 1 + regions[#regions + 1] = {start = first, ['end'] = last} + end + selected = table.concat(vim.fn.getregion(anchor, cursor, {type = kind}), '\n') + if kind == 'V' then selected = selected .. '\n' end + end + return {mode = api.nvim_get_mode(), cursor = location(cursor), + command_line = vim.fn.getcmdtype() .. vim.fn.getcmdline(), + regions = regions, selected = selected, completed = completed} +end +local function publish() + vim.rpcnotify(channel, 'toad_editor_state', state()) +end +local function schedule_publish() + if scheduled then return end + scheduled = true + vim.schedule(function() scheduled = false; publish() end) +end +-- Buffer subscriptions belong to this engine, including :edit/:buffer changes. +-- Only its current buffer publishes rendering deltas; returning to a buffer is +-- an ownership transfer, not another callback registration. +local subscriptions = {} +local function entered(buffer) + vim.rpcnotify(channel, 'toad_editor_lines', {first = 0, last = -1, + lines = api.nvim_buf_get_lines(buffer, 0, -1, true)}) + if subscriptions[buffer] then return end + subscriptions[buffer] = true + api.nvim_buf_attach(buffer, false, { + on_lines = function(_, buf, tick, first, last, new_last) + if api.nvim_get_current_buf() == buf then + completed = -1 + vim.rpcnotify(channel, 'toad_editor_lines', {first = first, last = last, + lines = api.nvim_buf_get_lines(buf, first, new_last, true)}) + schedule_publish() + end + end, + on_detach = function(_, buf) subscriptions[buf] = nil end, + }) +end +api.nvim_create_autocmd('BufEnter', {callback = function(event) + completed = -1 + entered(event.buf) + schedule_publish() +end}) +entered(api.nvim_get_current_buf()) +api.nvim_create_autocmd({'ModeChanged', 'CursorMoved', 'CursorMovedI', + 'TextChanged', 'TextChangedI', 'CmdlineChanged'}, + {callback = schedule_publish}) +api.nvim_create_autocmd('SafeState', {callback = function() + if completed ~= generation then + completed = generation + -- Do not queue work from the idle notification: running that work returns + -- to SafeState and would schedule the next idle notification forever. + publish() + end +end}) +vim.on_key(function() + -- lua.txt: on_key runs BEFORE further processing. This invalidates a proof; + -- it cannot create one, even if the host has already received the input reply. + completed = -1 + schedule_publish() +end, namespace) + +-- Install the actual native arrow declarations, including saved aliases, inside +-- Neovim. Its mapping engine knows that the k in fk is an operand, not Up. +for _, mapping in ipairs(mappings) do + local lhs, rhs, boundary = unpack(mapping) + for _, mode in ipairs({'', 'i'}) do + vim.keymap.set(mode, lhs, function() + if vim.v.count == 0 and not vim.fn.mode(1):match('^no') then + if vim.rpcrequest(channel, 'toad_cursor_boundary', boundary, state()) then return '' end + end + return api.nvim_replace_termcodes(rhs, true, false, true) + end, {expr = true, noremap = true, silent = true}) + end +end + +ToadEditor = {} +function ToadEditor.input(serial, keys) + generation = math.max(generation, serial) + completed = -1 + return api.nvim_input(keys) +end +function ToadEditor.call(serial, operation, arguments) + generation = math.max(generation, serial) + completed = -1 + local result = operation(unpack(arguments)) + schedule_publish() + return result +end +function ToadEditor.load(text) + api.nvim_buf_set_lines(0, 0, -1, true, vim.split(text, '\n', {plain = true})) + api.nvim_win_set_cursor(0, {1, 0}) +end +function ToadEditor.cursor_end() + local count = api.nvim_buf_line_count(0) + local line = api.nvim_buf_get_lines(0, count - 1, count, true)[1] + api.nvim_win_set_cursor(0, {count, #line}) +end +local function byte_position(location) + -- Convert application codepoint coordinates against the actual buffer now, + -- not against a potentially older Python rendering when the edit was queued. + local row = math.max(0, math.min(location[1], api.nvim_buf_line_count(0) - 1)) + local line = api.nvim_buf_get_lines(0, row, row + 1, true)[1] + local column = math.max(0, math.min(location[2], vim.str_utfindex(line, 'utf-32'))) + return {row + 1, vim.str_byteindex(line, 'utf-32', column, false)} +end +function ToadEditor.cursor(location) + api.nvim_win_set_cursor(0, byte_position(location)) +end +function ToadEditor.select(first, last) + -- TextArea API ranges are end-exclusive; Neovim owns the actual Visual state. + first, last = byte_position(first), byte_position(last) + local row, byte = last[1], last[2] + if byte == 0 and row > 1 then + row = row - 1 + byte = #api.nvim_buf_get_lines(0, row - 1, row, true)[1] + end + local line = api.nvim_buf_get_lines(0, row - 1, row, true)[1] + local characters = vim.str_utfindex(line, 'utf-32', byte, false) + byte = vim.str_byteindex(line, 'utf-32', math.max(0, characters - 1), false) + api.nvim_win_set_cursor(0, first) + vim.cmd.normal({args = {'v'}, bang = true}) + api.nvim_win_set_cursor(0, {row, byte}) +end +function ToadEditor.insert(lines, position, maintain_offset) + -- MessagePack nil in an argument array enters Lua as vim.NIL. + if position == vim.NIL then + local cursor = api.nvim_win_get_cursor(0) + local line = api.nvim_get_current_line() + position = {cursor[1] - 1, vim.str_utfindex(line, 'utf-32', cursor[2], false)} + end + return ToadEditor.replace(position, position, lines, maintain_offset) +end +function ToadEditor.replace(first, last, lines, maintain_offset) + local top, bottom = byte_position(first), byte_position(last) + local top_row, top_byte = top[1] - 1, top[2] + local bottom_row, bottom_byte = bottom[1] - 1, bottom[2] + local cursor = api.nvim_win_get_cursor(0) + local mark = api.nvim_buf_set_extmark(0, namespace, cursor[1] - 1, cursor[2], + {right_gravity = true}) + api.nvim_buf_set_text(0, top_row, top_byte, bottom_row, bottom_byte, lines) + local retained = api.nvim_buf_get_extmark_by_id(0, namespace, mark, {}) + api.nvim_buf_del_extmark(0, namespace, mark) + if maintain_offset then + api.nvim_win_set_cursor(0, {retained[1] + 1, retained[2]}) + else + local row = top_row + #lines - 1 + local column = #lines[#lines] + (#lines == 1 and top_byte or 0) + api.nvim_win_set_cursor(0, {row + 1, column}) + end +end diff --git a/src/toad/neovim_editing.py b/src/toad/neovim_editing.py new file mode 100644 index 000000000..0b5072520 --- /dev/null +++ b/src/toad/neovim_editing.py @@ -0,0 +1,1221 @@ +"""Real modal editing. Neovim owns the buffer, registers, undo and input grammar.""" +from __future__ import annotations + +import asyncio +from abc import ABC, abstractmethod +from collections.abc import Callable, Mapping, Sequence +from contextlib import AsyncExitStack +from dataclasses import dataclass, field +from functools import cached_property +from importlib.resources import files +from itertools import count +from typing import ClassVar, Self + +import msgpack +import toad +from agent_comms.child_process import AttachedChild, join_retirement +from agent_comms.declared_family import DeclaredFamily, UnknownFamilyMember +from agent_comms.field_codec import FieldCodec, MalformedRecord, NullMember, SingleFieldValue, SubclassUnion, boundary_decoder +from metaclass_registry.mro_dispatch import MroDispatch, handles +from textual import log +from textual.binding import Binding +from textual.cursor_shape import BlinkingBar, CursorShape, SteadyBlock +from textual.document._document import DocumentBase, Location, RangedEditResult, Selection +from textual.document._editing import EditingFailure, EditingMode, VimMode +from textual.document._edit import Edit, InsertionPoint +from textual.events import Key +from textual.keys import key_to_character +from textual.message import Message +from textual.widgets import TextArea + + +class NeovimKey(DeclaredFamily, affix="Key"): + """Only vocabulary differences are declared; native Neovim key names pass through.""" + + notation: ClassVar[str] + + @classmethod + def spelling(cls, name: str) -> str: + try: + return cls.member_named(name).notation + except UnknownFamilyMember: + return name + + +class EnterKey(NeovimKey): + notation = "CR" + + +class EscapeKey(NeovimKey): + notation = "Esc" + + +class BackspaceKey(NeovimKey): + notation = "BS" + + +class DeleteKey(NeovimKey): + notation = "Del" + + +class CtrlKey(NeovimKey): + notation = "C" + + +class ShiftKey(NeovimKey): + notation = "S" + + +class AltKey(NeovimKey): + notation = "A" + + +class MetaKey(NeovimKey): + notation = "M" + + +class SuperKey(NeovimKey): + notation = "D" + + +class NeovimWireFailure(EditingFailure): + """A child, transport or external Neovim protocol failure, not a Vim prefix.""" + + +@dataclass(frozen=True) +class NeovimApiType: + id: int + prefix: str + + +@dataclass(frozen=True) +class NeovimApiMetadata: + types: dict[str, NeovimApiType] + + @classmethod + @boundary_decoder + def from_response(cls, metadata: object) -> Self: + if not isinstance(metadata, dict): + raise NeovimWireFailure("Invalid Neovim API metadata") + # Decode the declared field through its existing binding. Unused native + # metadata stays opaque; no second record decoder or copied field key. + binding, = FieldCodec.argument_fields(cls) + return FieldCodec.construct(cls, binding.decoded_bound(metadata, {})) + + +@dataclass(frozen=True) +class NeovimApiInfo: + channel: int + metadata: NeovimApiMetadata + + @classmethod + @boundary_decoder + def from_response(cls, response: object) -> Self: + if not isinstance(response, list) or len(response) != 2: + raise NeovimWireFailure("Invalid Neovim API information response") + channel, metadata = response + return cls(FieldCodec.decode(int, channel), NeovimApiMetadata.from_response(metadata)) + + +@dataclass(frozen=True) +class NeovimHandle(DeclaredFamily, affix="Handle"): + """Native extension identity; its code belongs to this child's API metadata.""" + + value: int + + @classmethod + def api_name(cls) -> str: + return cls.declared_name.capitalize() + + def extension(self, types: Mapping[str, NeovimApiType]) -> msgpack.ExtType: + return msgpack.ExtType(types[self.api_name()].id, msgpack.packb(self.value)) + + +class BufferHandle(NeovimHandle): + pass + + +class WindowHandle(NeovimHandle): + pass + + +class TabpageHandle(NeovimHandle): + pass + + +@dataclass(frozen=True) +class NeovimMode: + """The actual nvim_get_mode answer, decoded once inside a display update.""" + + mode: str + blocking: bool + + +@dataclass(frozen=True) +class NeovimRegion: + start: Location + end: Location + + def selection(self) -> Selection: + return Selection(self.start, self.end) + + +@dataclass(frozen=True) +class NeovimDisplay: + mode: NeovimMode + cursor: Location + command_line: str + regions: tuple[NeovimRegion, ...] + selected: str + completed: int + + def state(self) -> NeovimState: + member = next((member for member in NeovimState.members_with(NeovimState) + if member.recognizes(self.mode.mode)), NeovimState) + return member(self.mode.blocking, self.cursor, self.command_line, + tuple(region.selection() for region in self.regions), self.selected) + + +@dataclass(frozen=True) +class NeovimState(DeclaredFamily, affix="State"): + """Behaviour derives from a mode declaration, never from the last input key.""" + + blocking: bool = True + cursor: Location = (0, 0) + command_line: str = "" + regions: tuple[Selection, ...] = () + region_text: str = "" + mode_codes: ClassVar[tuple[str, ...]] = () + mode_prefixes: ClassVar[tuple[str, ...]] = () + mode_name: ClassVar[str] = "NEOVIM" + cursor_shape: ClassVar[CursorShape] = SteadyBlock() + + @classmethod + def recognizes(cls, code: str) -> bool: + return code in cls.mode_codes or any(code.startswith(prefix) for prefix in cls.mode_prefixes) + + @property + def label(self) -> str: + return self.mode_name + (" (PENDING)" if self.blocking else "") + + @property + def accepts_text(self) -> bool: + return False + + @property + def allows_application_commands(self) -> bool: + return False + + @property + def allows_history_navigation(self) -> bool: + return False + + @property + def transferable(self) -> bool: + return False + + def update(self, display: NeovimDisplay) -> NeovimState: + return display.state() + + def completion(self, completion: NeovimCompletion) -> NeovimCompletion: + return completion + + def fail(self, error: EditingFailure) -> NeovimState: + return FailedState(cursor=self.cursor, command_line=self.command_line, + regions=self.regions, region_text=self.region_text, error=error) + + @property + def binding_answers(self) -> tuple[bool, bool, bool]: + return self.accepts_text, self.allows_application_commands, self.allows_history_navigation + + def raise_failure(self) -> None: + pass + + def close_error(self) -> EditingFailure: + return EditingFailure("Neovim editor closed; input was not replayed") + + def selection(self) -> Selection: + return Selection.cursor(self.cursor) + + def selection_for_line(self, row: int) -> Selection: + return self.selection() + + def selected_text(self) -> str: + return "" + + +class TransferableState(NeovimState, declared_name=None): + @property + def transferable(self) -> bool: + return not self.blocking + + +class TextEntryState(TransferableState, declared_name=None): + cursor_shape = BlinkingBar() + + @property + def accepts_text(self) -> bool: + return not self.blocking + + +class HistoryState(NeovimState, declared_name=None): + @property + def allows_history_navigation(self) -> bool: + return not self.blocking + + +class ApplicationCommandState(TransferableState, declared_name=None): + @property + def allows_application_commands(self) -> bool: + return not self.blocking + + +class StartingState(NeovimState): + mode_name = "NEOVIM STARTING" + + def snapshot(self) -> NeovimSnapshot: + return NeovimSnapshot(self, NotCompleted()) + + @property + def label(self) -> str: + return self.mode_name + + +@dataclass(frozen=True) +class FailedState(NeovimState): + error: EditingFailure = field(kw_only=True) + + def update(self, display: NeovimDisplay) -> NeovimState: + return self + + def completion(self, completion: NeovimCompletion) -> NeovimCompletion: + return NotCompleted() + + def fail(self, error: EditingFailure) -> NeovimState: + return self + + @property + def label(self) -> str: + return f"NEOVIM FAILED — {self.error} — draft preserved" + + def raise_failure(self) -> None: + raise self.error + + def close_error(self) -> EditingFailure: + return self.error + + +class InsertState(TextEntryState, HistoryState): + mode_prefixes = ("i",) + mode_name = "INSERT" + + +class NormalState(ApplicationCommandState, HistoryState): + mode_codes = ("n",) + mode_name = "NORMAL" + + +class ReplaceState(TextEntryState): + mode_prefixes = ("R",) + mode_name = "REPLACE" + + +class OperatorState(NeovimState): + mode_prefixes = ("no",) + mode_name = "OPERATOR" + + +class CommandLineState(NeovimState): + mode_codes = ("c",) + mode_name = "COMMAND" + + +class VisualState(TransferableState): + mode_codes = ("v", "s") + mode_name = "VISUAL" + + def selection(self) -> Selection: + if not self.regions: + return super().selection() + return Selection(self.regions[0].start, self.regions[-1].end) + + def selection_for_line(self, row: int) -> Selection: + return next((region for region in self.regions if region.start[0] == row), + Selection.cursor(self.cursor)) + + def selected_text(self) -> str: + return self.region_text + + +class LineVisualState(VisualState): + mode_codes = ("V", "S") + mode_name = "VISUAL LINE" + + +class BlockVisualState(VisualState): + mode_codes = ("\x16", "\x13") + mode_name = "VISUAL BLOCK" + + +class DetachedReceiver: + """An engine may outlive its widget; detachment is not child failure.""" + + def attach(self, editor: TextArea) -> AttachedReceiver: + self.prepare_editor(editor) + return AttachedReceiver(editor) + + def prepare_editor(self, editor: TextArea) -> None: + pass + + def require(self) -> TextArea: + raise EditingFailure("Neovim requires an attached editor") + + def owns(self, editor: TextArea) -> bool: + return False + + def visit(self, action: Callable[[TextArea], object]) -> None: + pass + + def boundary(self, binding: Binding) -> bool: + return False + + def project(self, changes: Sequence[NeovimChange]) -> None: + pass + + def finish_projection(self) -> None: + pass + + +class RetainedReceiver(DetachedReceiver): + """A captured document remains the projection without retaining its widget.""" + + def __init__(self, document: DocumentBase) -> None: + self._document = document + + @property + def document(self) -> DocumentBase: + return self._document + + def prepare_editor(self, editor: TextArea) -> None: + if editor.document is not self.document: + editor.set_document(self.document) + + def project(self, changes: Sequence[NeovimChange]) -> None: + for change in changes: + self.project_edit(change.apply(self.document)) + + def project_edit(self, result: RangedEditResult) -> None: + pass + + +class AttachedReceiver(RetainedReceiver): + def __init__(self, editor: TextArea) -> None: + self._editor = editor + + @property + def document(self) -> DocumentBase: + return self._editor.document + + def project_edit(self, result: RangedEditResult) -> None: + self._editor.project_engine_edit(result) + + def finish_projection(self) -> None: + self._editor.finish_engine_changes() + + def require(self) -> TextArea: + return self._editor + + def owns(self, editor: TextArea) -> bool: + return self._editor is editor + + def visit(self, action: Callable[[TextArea], object]) -> None: + if self._editor.is_attached: + action(self._editor) + + def boundary(self, binding: Binding) -> bool: + return self._editor.is_attached and binding.engine_boundary(self._editor) + + +class NeovimRequest(ABC): + """One queued request; the writer never waits for its eventual reply.""" + + _identities = count(1) + + def __init__(self) -> None: + self.identity = next(self._identities) + self.reply: asyncio.Future[object] = asyncio.get_running_loop().create_future() + self.dispatched: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + @property + @abstractmethod + def method(self) -> str: ... + + @property + @abstractmethod + def arguments(self) -> Sequence[object]: ... + + async def send(self, owner: NeovimEditing) -> None: + self.dispatched.set_result(None) + await owner.write_frame([0, self.identity, self.method, list(self.arguments)]) + + def resolve(self, value: object) -> None: + self.reply.set_result(value) + + async def result(self) -> object: + await asyncio.shield(self.dispatched) + return await self.wait_reply() + + async def wait_reply(self) -> object: + return await asyncio.shield(self.reply) + + +class NeovimFastRequest(NeovimRequest): + async def wait_reply(self) -> object: + try: + async with asyncio.timeout(5): + return await super().wait_reply() + except TimeoutError as error: + raise NeovimWireFailure(f"Neovim fast request {self.method} did not reply") from error + + +class NeovimCall(NeovimRequest): + def __init__(self, method: str, arguments: Sequence[object]) -> None: + super().__init__() + self._method, self._arguments = method, arguments + + @property + def method(self) -> str: + return self._method + + @property + def arguments(self) -> Sequence[object]: + return self._arguments + + +class NeovimOperation(NeovimCall): + """Programmatic mutation executes on the same engine and invalidates its proof.""" + + operation: ClassVar[str] = "assert(vim.api[operation], operation)" + + def __init__(self, generation: int, method: str, arguments: Sequence[object]) -> None: + super().__init__("nvim_exec_lua", [ + "local serial, operation, arguments = ...; " + f"return ToadEditor.call(serial, {self.operation}, arguments)", + [generation, method, list(arguments)], + ]) + + +class NeovimLuaOperation(NeovimOperation): + """Lua chunks are not API names: nvim_exec_lua itself is RPC-only.""" + + operation = "assert(loadstring(operation))" + + +class NeovimInput(NeovimFastRequest): + method = "nvim__exec_lua_fast" + + def __init__(self, generation: int, keys: str) -> None: + super().__init__() + self.generation, self.keys = generation, keys + + @property + def arguments(self) -> Sequence[object]: + return ("return ToadEditor.input(...)", [self.generation, self.keys]) + + def resolve(self, value: object) -> None: + accepted = FieldCodec.decode(int, value) + if accepted != len(self.keys.encode("utf-8")): + raise NeovimWireFailure("Neovim did not accept the complete key; it was not replayed") + super().resolve(accepted) + + +class NeovimChange(ABC): + @abstractmethod + def apply(self, document: DocumentBase) -> RangedEditResult: ... + + +@dataclass(frozen=True) +class NeovimLineChange(NeovimChange): + first: int + last: int + lines: tuple[str, ...] + + def apply(self, document: DocumentBase) -> RangedEditResult: + return document.replace_lines(self.first, self.last, self.lines) + + +@dataclass(frozen=True) +class NeovimBufferTransfer(NeovimChange): + lines: tuple[str, ...] + + def apply(self, document: DocumentBase) -> RangedEditResult: + return document.replace_lines(0, document.line_count, self.lines) + + +class NeovimCompletion(ABC): + @classmethod + @boundary_decoder + def from_native(cls, generation: int) -> NeovimCompletion: + if generation == -1: + return NotCompleted() + if generation < 0: + raise NeovimWireFailure("Invalid Neovim completed generation") + return CompletedGeneration(generation) + + @abstractmethod + def matches(self, generation: int) -> bool: ... + + +@dataclass(frozen=True) +class NotCompleted(NeovimCompletion): + def matches(self, generation: int) -> bool: + return False + + +@dataclass(frozen=True) +class CompletedGeneration(NeovimCompletion): + generation: int + + def matches(self, generation: int) -> bool: + return self.generation == generation + + +@dataclass(frozen=True) +class NeovimSnapshot: + state: NeovimState + completed: NeovimCompletion + + def with_lines(self) -> NeovimSnapshot: + return NeovimSnapshot(self.state, self.completed) + + def with_display(self, display: NeovimDisplay) -> NeovimSnapshot: + state = self.state.update(display) + return NeovimSnapshot(state, state.completion(NeovimCompletion.from_native(display.completed))) + + def with_boundary(self, display: NeovimDisplay) -> NeovimSnapshot: + # A mapping's in-command display is not a SafeState certificate. + return NeovimSnapshot(self.state.update(display), NotCompleted()) + + def with_failure(self, error: EditingFailure) -> NeovimSnapshot: + return NeovimSnapshot(self.state.fail(error), NotCompleted()) + + def unpainted(self) -> UnpaintedSnapshot: + return UnpaintedSnapshot(self.state, self.completed) + + def matches[Answer](self, current: NeovimState, + answer: Callable[[NeovimState], Answer]) -> bool: + return answer(self.state) == answer(current) + + +class UnpaintedSnapshot(NeovimSnapshot): + """A new view has presented none of the retained owner's answers yet.""" + + def matches[Answer](self, current: NeovimState, + answer: Callable[[NeovimState], Answer]) -> bool: + return False + + +class NeovimUpdated(Message): + def __init__(self, owner: NeovimEditing) -> None: + super().__init__() + self.owner = owner + + +class NeovimMethod: + method: ClassVar[str] + + @classmethod + def selection_key(cls) -> str: + return cls.method + + +class NeovimNotification(NeovimMethod, DeclaredFamily, affix="Notification"): + @classmethod + @abstractmethod + def receive(cls, owner: NeovimEditing, arguments: object) -> None: ... + + +class DisplayNotification(NeovimNotification): + method = "toad_editor_state" + + @classmethod + def receive(cls, owner: NeovimEditing, arguments: object) -> None: + display, = FieldCodec.decode(tuple[NeovimDisplay], arguments) + owner.accept_display(display) + + +class LinesNotification(NeovimNotification): + method = "toad_editor_lines" + + @classmethod + def receive(cls, owner: NeovimEditing, arguments: object) -> None: + change, = FieldCodec.decode(tuple[NeovimLineChange], arguments) + owner.accept_lines(change) + + +class RedrawNotification(NeovimNotification): + """The attached UI drives Neovim, but TextArea paints buffer projections.""" + + method = "redraw" + + @classmethod + def receive(cls, owner: NeovimEditing, arguments: object) -> None: + pass + + +class NeovimHostRequest(NeovimMethod, DeclaredFamily, affix="Request"): + @classmethod + @abstractmethod + def receive(cls, owner: NeovimEditing, arguments: object) -> object: ... + + +class CursorBoundaryRequest(NeovimHostRequest): + method = "toad_cursor_boundary" + + @classmethod + def receive(cls, owner: NeovimEditing, arguments: object) -> bool: + boundary, display = FieldCodec.decode(tuple[int, NeovimDisplay], arguments) + return owner.engine_boundary(boundary, display) + + +class NeovimReply(SubclassUnion, affix="Reply"): + @abstractmethod + def resolve(self, request: NeovimRequest, value: object) -> None: ... + + +class SuccessfulReply(NullMember, NeovimReply): + def resolve(self, request: NeovimRequest, value: object) -> None: + request.resolve(value) + + +@dataclass(frozen=True) +class ErrorReply(SingleFieldValue, NeovimReply): + details: tuple[int, str] + + def resolve(self, request: NeovimRequest, value: object) -> None: + code, message = self.details + request.reply.set_exception(NeovimWireFailure(f"Neovim {request.method} ({code}): {message}")) + + +class NeovimFrame(DeclaredFamily, affix="Frame"): + kind: ClassVar[int] + + @classmethod + def selection_key(cls) -> int: + return cls.kind + + @abstractmethod + def __init__(self, *payload: object) -> None: + if payload: + raise NeovimWireFailure("Unexpected Neovim RPC header fields") + + @classmethod + @boundary_decoder + def from_payload(cls, payload: Sequence[object]) -> Self: + try: + return cls(*payload) + except TypeError as error: + raise NeovimWireFailure(f"Invalid {cls.__name__} header or arity") from error + + @abstractmethod + async def receive(self, owner: NeovimEditing) -> None: ... + + +class IdentifiedFrame(NeovimFrame, declared_name=None): + def __init__(self, identity: object, *payload: object) -> None: + self.identity = FieldCodec.decode(int, identity) + super().__init__(*payload) + + +class MethodFrame(NeovimFrame, declared_name=None): + def __init__(self, method: object, arguments: object) -> None: + self.method = FieldCodec.decode(str, method) + self.arguments = arguments + super().__init__() + + +class ResponseFrame(IdentifiedFrame): + kind = 1 + + def __init__(self, identity: object, reply: object, value: object) -> None: + super().__init__(identity) + self.reply = FieldCodec.decode(NeovimReply, reply) + self.value = value + + async def receive(self, owner: NeovimEditing) -> None: + owner.resolve_reply(self.identity, self.reply, self.value) + + +class NotificationFrame(MethodFrame): + kind = 2 + + async def receive(self, owner: NeovimEditing) -> None: + NeovimNotification.member_selected(self.method).receive(owner, self.arguments) + + +class RequestFrame(IdentifiedFrame, MethodFrame): + kind = 0 + + async def receive(self, owner: NeovimEditing) -> None: + reply = NeovimHostRequest.member_selected(self.method).receive(owner, self.arguments) + await owner.answer_request(self.identity, reply) + + +class NeovimEditing(VimMode, MroDispatch): + """One engine, one input queue, completed snapshots only from SafeState.""" + + initial_keys = "" + + def __init__(self) -> None: + self._receiver: DetachedReceiver = DetachedReceiver() + self._pending: dict[int, NeovimRequest] = {} + self._outgoing: asyncio.Queue[NeovimRequest] = asyncio.Queue() + self._generation = 0 + self.snapshot = StartingState().snapshot() + self._changes: list[NeovimChange] = [] + self._rendered: NeovimSnapshot = self.snapshot.unpainted() + self._update_queued = False + self._boundaries: list[Binding] = [] + self._api_types: dict[str, NeovimApiType] = {} + + @property + def label(self) -> str: + state = self.snapshot.state + return f"{state.label} {state.command_line}".rstrip() + + @property + def completed(self) -> bool: + return (self.snapshot.completed.matches(self._generation) + and not self._pending and not self._changes) + + @property + def accepts_text(self) -> bool: + return self.completed and self.snapshot.state.accepts_text + + @property + def allows_application_commands(self) -> bool: + return self.completed and self.snapshot.state.allows_application_commands + + @property + def allows_history_navigation(self) -> bool: + # The engine's expression mapping calls this boundary while an arrow is + # executing. That input cannot already have a SafeState completion. + return self.snapshot.state.allows_history_navigation + + @property + def cursor_shape(self) -> CursorShape: + return self.snapshot.state.cursor_shape + + @property + def binding_mode(self) -> type[EditingMode]: + return type(self) + + @cached_property + def _started(self) -> asyncio.Task[None]: + editor = self._receiver.require() + self._initial_lines = tuple(editor.document.lines) + self._initial_selection = editor.selection + self._child_ready: asyncio.Future[AttachedChild] = asyncio.get_running_loop().create_future() + self._lifetime = editor.app.run_worker(self._run(), name="Neovim editor", exit_on_error=True) + task = asyncio.create_task(self._start()) + task.add_done_callback(self._observe_failure) + return task + + def attach(self, editor: TextArea) -> None: + changed_receiver = not self._receiver.owns(editor) + self._receiver = self._receiver.attach(editor) + if changed_receiver: + self._update_queued = False + editor.history.clear() + # Even the same receiver may have rebuilt its width-dependent view. + self._rendered = self._rendered.unpainted() + started = self._started + if started.done() and not started.cancelled(): + self.paint(editor) + + def resume(self, editor: TextArea) -> None: + pass + + def suspend(self, editor: TextArea) -> None: + pass + + def capture(self, editor: TextArea) -> NeovimEditing: + self._receiver = RetainedReceiver(editor.document) + return self + + async def unmount(self, editor: TextArea) -> None: + if self._receiver.owns(editor): + await self.close() + + async def close(self) -> None: + started = self._started + self._receiver = DetachedReceiver() + started.cancel() + self._lifetime.cancel() + await join_retirement(asyncio.create_task(self._join())) + + async def _join(self) -> None: + await asyncio.gather(self._started, return_exceptions=True) + try: + await self._lifetime.wait() + except asyncio.CancelledError: + pass + + async def _run(self) -> None: + try: + async with AsyncExitStack() as cleanup: + async def acquire_child() -> AttachedChild: + child = await AttachedChild.start(("nvim", "--embed", "--clean", "-n")) + cleanup.push_async_callback(child.stop) + return child + + spawn = asyncio.create_task(acquire_child()) + try: + child = await asyncio.shield(spawn) + except asyncio.CancelledError: + await join_retirement(spawn) + raise + self._child = child + if self._child_ready.cancelled(): + return + self._child_ready.set_result(child) + async with asyncio.TaskGroup() as tasks: + tasks.create_task(self._write_queued()) + await self._read(child) + except* (EditingFailure, OSError, msgpack.UnpackException, + UnicodeDecodeError, MalformedRecord) as errors: + self._fail(NeovimWireFailure("; ".join(str(error) for error in errors.exceptions))) + finally: + error = self.snapshot.state.close_error() + if not self._child_ready.done(): + self._child_ready.set_exception(error) + for request in self._pending.values(): + if not request.dispatched.done(): + request.dispatched.set_exception(error) + if not request.reply.done(): + request.reply.set_exception(error) + self._pending.clear() + + @cached_property + def _unpacker(self) -> msgpack.Unpacker: + return msgpack.Unpacker(raw=False, strict_map_key=False, + max_buffer_size=16 * 1024 * 1024, + ext_hook=self._decode_handle) + + @boundary_decoder + async def _read(self, child: AttachedChild) -> None: + stream = child.stdout + assert stream is not None + while chunk := await stream.read(65536): + self._unpacker.feed(chunk) + for message in self._unpacker: + await self._receive(message) + raise NeovimWireFailure(f"Neovim exited: {await child.stderr_tail()}") + + def accept_api_info(self, response: object) -> NeovimApiInfo: + api = NeovimApiInfo.from_response(response) + self._api_types = api.metadata.types + return api + + @boundary_decoder + def _decode_handle(self, code: int, data: bytes) -> NeovimHandle: + try: + name = next(name for name, definition in self._api_types.items() + if definition.id == code) + member = NeovimHandle.member_named(name.casefold()) + except (StopIteration, UnknownFamilyMember) as error: + raise NeovimWireFailure(f"Unknown Neovim extension code {code}") from error + return member(FieldCodec.decode(int, msgpack.unpackb(data, raw=False))) + + @boundary_decoder + def _encode_handle(self, value: object) -> msgpack.ExtType: + try: + encoder: Callable[[NeovimEditing, object], msgpack.ExtType] = self.handler_for(type(value)) + except StopIteration as error: + raise NeovimWireFailure("Unsupported Neovim MessagePack value") from error + return encoder(self, value) + + @handles(NeovimHandle) + def _native_handle(self, value: NeovimHandle) -> msgpack.ExtType: + try: + return value.extension(self._api_types) + except KeyError as error: + raise NeovimWireFailure("Unknown Neovim extension type") from error + + async def write_frame(self, frame: Sequence[object]) -> None: + try: + await self._child.write(msgpack.packb( + frame, use_bin_type=True, default=self._encode_handle)) + except OSError as error: + raise NeovimWireFailure(str(error)) from error + + async def _write_queued(self) -> None: + await self._started + while True: + request = await self._outgoing.get() + await request.send(self) + + def _register(self, request: NeovimRequest) -> None: + request.reply.add_done_callback(self._observe_failure) + request.dispatched.add_done_callback(self._observe_failure) + self.snapshot.state.raise_failure() + self._pending[request.identity] = request + + async def _bootstrap(self, request: NeovimRequest) -> object: + self._register(request) + await request.send(self) + return await request.result() + + def _enqueue(self, request: NeovimRequest) -> asyncio.Task[object]: + try: + self._register(request) + except EditingFailure as error: + request.dispatched.set_exception(error) + request.reply.set_exception(error) + else: + self._outgoing.put_nowait(request) + result = asyncio.create_task(request.result()) + result.add_done_callback(self._observe_failure) + return result + + def _next_generation(self) -> int: + self._generation += 1 + return self._generation + + def _call(self, method: str, arguments: Sequence[object]) -> asyncio.Future[object]: + return self._enqueue(NeovimOperation(self._next_generation(), method, arguments)) + + def _lua(self, code: str, arguments: Sequence[object]) -> asyncio.Future[object]: + return self._enqueue(NeovimLuaOperation(self._next_generation(), code, arguments)) + + def _input(self, keys: str) -> asyncio.Future[object]: + return self._enqueue(NeovimInput(self._next_generation(), keys)) + + async def _start(self) -> None: + await self._child_ready + editor = self._receiver.require() + width, height = max(1, editor.wrap_width), max(3, editor.content_size.height + 2) + api = self.accept_api_info(await self._bootstrap(NeovimCall("nvim_get_api_info", []))) + await self._bootstrap(NeovimCall("nvim_ui_attach", [width, height, + {"ext_linegrid": True, "ext_cmdline": True, "ext_messages": True}])) + initial_lines = self._initial_lines + undo_levels = await self._bootstrap(NeovimCall("nvim_get_option_value", ["undolevels", {"buf": 0}])) + await self._bootstrap(NeovimCall("nvim_set_option_value", ["undolevels", -1, {"buf": 0}])) + await self._bootstrap(NeovimCall("nvim_buf_set_lines", [0, 0, -1, True, list(initial_lines)])) + await self._bootstrap(NeovimCall("nvim_set_option_value", ["undolevels", undo_levels, {"buf": 0}])) + initial_cursor = self._initial_selection.end + row, column = initial_cursor + await self._bootstrap(NeovimCall("nvim_win_set_cursor", [0, + [row + 1, len(initial_lines[row][:column].encode("utf-8"))]])) + script = files(toad).joinpath("neovim_editing.lua").read_text() + mappings: list[list[object]] = [] + for key, binding in editor.binding_map(type(self)): + for spelling, native in binding.declared_key.input_bindings(key): + index = len(self._boundaries) + self._boundaries.append(binding) + mappings.append([self.input_code(spelling), self.input_code(native), index]) + await self._bootstrap(NeovimCall("nvim_exec_lua", [script, [mappings, api.channel]])) + await self._bootstrap(NeovimCall("nvim_set_option_value", ["tabstop", editor.indent_width, {}])) + if self.initial_keys: + await self._bootstrap(NeovimInput(self._next_generation(), self.initial_keys)) + await self._bootstrap(NeovimLuaOperation(self._next_generation(), + "return ToadEditor.cursor(...)", [initial_cursor])) + + @boundary_decoder + async def _receive(self, message: object) -> None: + """Decode the envelope tag here; its member alone decodes the raw payload.""" + if not isinstance(message, list) or not message: + raise NeovimWireFailure("Invalid Neovim RPC message") + kind = FieldCodec.decode(int, message[0]) + try: + frame = NeovimFrame.member_selected(kind).from_payload(message[1:]) + await frame.receive(self) + except UnknownFamilyMember as error: + raise NeovimWireFailure(str(error)) from error + + def resolve_reply(self, identity: int, reply: NeovimReply, value: object) -> None: + try: + request = self._pending[identity] + except KeyError as error: + raise NeovimWireFailure("Unexpected Neovim RPC response") from error + reply.resolve(request, value) + del self._pending[identity] + # SafeState can arrive before the final RPC response. Its + # publication and that response must invalidate the same view owner. + if self.completed: + self.request_paint() + + def engine_boundary(self, boundary: int, display: NeovimDisplay) -> bool: + if not 0 <= boundary < len(self._boundaries): + raise NeovimWireFailure("Unknown Neovim cursor boundary") + self.publish(self.snapshot.with_boundary(display)) + self._receiver.visit(self.render) + return self._receiver.boundary(self._boundaries[boundary]) + + async def answer_request(self, identity: int, reply: object) -> None: + await self.write_frame([1, identity, None, reply]) + + def accept_lines(self, change: NeovimLineChange) -> None: + if change.last == -1: + self._changes.append(NeovimBufferTransfer(change.lines)) + else: + self._changes.append(change) + self.snapshot = self.snapshot.with_lines() + + def accept_display(self, display: NeovimDisplay) -> None: + self.publish(self.snapshot.with_display(display)) + + def publish(self, snapshot: NeovimSnapshot) -> None: + """Project preceding lines before exposing the display that describes them.""" + changes, self._changes = self._changes, [] + if changes: + self._receiver.project(changes) + self.snapshot = snapshot + if changes: + self._receiver.finish_projection() + self.request_paint() + + def request_paint(self) -> None: + """One coalesced paint slot for display and RPC completion changes.""" + if not self._update_queued: + def post(editor: TextArea) -> None: + self._update_queued = editor.post_message(NeovimUpdated(self)) + self._receiver.visit(post) + + def paint(self, editor: TextArea) -> None: + if not self._receiver.owns(editor): + return + self._update_queued = False + # A later line batch may precede its display while a message waits. + if not self._changes: + self.render(editor) + + def render(self, editor: TextArea) -> None: + if editor.editing_mode is not self or not self._receiver.owns(editor): + return + if self._changes: + raise RuntimeError("Cannot paint an unpublished Neovim line batch") + if self.snapshot is self._rendered: + return + previous, current = self._rendered, self.snapshot.state + if not previous.matches(current, lambda state: state.cursor): + editor.paint_selection(Selection.cursor(current.cursor)) + editor.record_cursor_width() + if not previous.matches(current, lambda state: state.regions): + editor.invalidate_engine_regions((*previous.state.regions, *current.regions)) + if not previous.matches(current, lambda state: (state.label, state.command_line)): + editor.show_editing_mode(self) + if not previous.matches(current, lambda state: state.binding_answers): + editor.refresh_bindings() + self._rendered = self.snapshot + + def _observe_failure[Result](self, result: asyncio.Future[Result]) -> None: + try: + result.result() + except asyncio.CancelledError: + pass + except EditingFailure as error: + self._fail(error) + + def _fail(self, error: EditingFailure) -> None: + failed = self.snapshot.with_failure(error) + if failed.state is self.snapshot.state: + return + log.error("Neovim editing failed:", error) + self.publish(failed) + def notify(editor: TextArea) -> None: + editor.notify(f"{error}. The displayed draft is preserved. Input was not replayed.", + title="Neovim editing failed", severity="error", timeout=0) + self._receiver.visit(notify) + + async def finish(self, editor: TextArea) -> bool: + return await self.flush(editor) + + async def flush(self, editor: TextArea) -> bool: + """Never wait for deferred RPC in App dispatch. Missing completion refuses.""" + self.snapshot.state.raise_failure() + if not self.completed: + return False + self.render(editor) + return self.snapshot.state.transferable + + async def prepare_input(self, editor: TextArea) -> bool: + # Only fence the widget queue, never the Neovim event loop. Escape must + # remain admissible while a deferred operation is waiting for an operand. + finished: asyncio.Future[bool] = asyncio.get_running_loop().create_future() + def ready() -> None: + finished.set_result(True) + if not editor.call_later(ready): + return False + return await finished + + async def handle_key(self, editor: TextArea, event: Key) -> bool: + try: + await self._input(self.input_code(event.key, event.character or "")) + except EditingFailure: + pass # The child/wire failure was reported; never replay input. + return True + + @staticmethod + def input_code(key: str, character: str = "") -> str: + """Textual's foreign key vocabulary enters Neovim at this boundary.""" + *modifiers, name = key.split("+") + literal = character or key_to_character(name) or "" + is_character = bool(literal) and literal.isprintable() + if not modifiers and is_character: + return literal.replace("<", "") + base = literal if is_character else NeovimKey.spelling(name) + base = base.replace("<", "LT") + return "<" + "-".join([*(NeovimKey.member_named(part).notation for part in modifiers), base]) + ">" + + def resize(self, editor: TextArea) -> None: + # This UI API is RPC-only, like nvim_ui_attach during startup. A + # viewport resize is not a buffer edit and must not advance its generation. + self._enqueue(NeovimCall("nvim_ui_try_resize", + [max(1, editor.wrap_width), max(3, editor.content_size.height + 2)])) + + def enter_text(self, editor: TextArea) -> None: + self._input("i") + + def load_text(self, editor: TextArea, text: str) -> None: + self._lua("return ToadEditor.load(...)", [text]) + + def edit(self, editor: TextArea, edit: Edit) -> None: + self._lua("return ToadEditor.replace(...)", + [edit.top, edit.bottom, edit.text.split("\n"), edit.maintain_selection_offset]) + + def insert(self, editor: TextArea, text: str, point: InsertionPoint, maintain_selection_offset: bool) -> None: + self._lua("return ToadEditor.insert(...)", + [text.split("\n"), point.engine_location(), maintain_selection_offset]) + + def clear(self, editor: TextArea) -> None: + self.load_text(editor, "") + + def move_to_end(self, editor: TextArea) -> None: + self._lua("return ToadEditor.cursor_end()", []) + + def paste(self, editor: TextArea, text: str) -> None: + self._call("nvim_paste", [text, True, -1]) + + def undo(self, editor: TextArea) -> None: + self._call("nvim_command", ["undo"]) + + def redo(self, editor: TextArea) -> None: + self._call("nvim_command", ["redo"]) + + def request_selection(self, editor: TextArea, selection: Selection) -> Selection: + if not selection.is_empty: + self._input("") + self._lua("return ToadEditor.select(...)", sorted(selection)) + elif selection.end != self.snapshot.state.cursor: + self._lua("return ToadEditor.cursor(...)", [selection.end]) + return editor.selection + + def selection_range(self, editor: TextArea) -> Selection: + return self.snapshot.state.selection() + + def selection_for_line(self, editor: TextArea, row: int) -> Selection: + return self.snapshot.state.selection_for_line(row) + + def selected_text(self, editor: TextArea) -> str: + return self.snapshot.state.selected_text() diff --git a/src/toad/prompt_cursor.py b/src/toad/prompt_cursor.py index 45eb0a564..13cd68719 100644 --- a/src/toad/prompt_cursor.py +++ b/src/toad/prompt_cursor.py @@ -3,15 +3,19 @@ from abc import abstractmethod from dataclasses import dataclass -from typing import TYPE_CHECKING, Callable +from types import FunctionType +from typing import TYPE_CHECKING, ClassVar from weakref import ref from agent_comms.declared_family import DeclaredFamily from textual.content import Content +from textual.document._editing import EditingMode +from textual.widgets import TextArea from textual.widgets.text_area import Selection +from toad.core.input_events import HistoryMove + if TYPE_CHECKING: - from textual.document._editing import EditingMode, LineMotion, VimNormal from toad.slash_command import SlashCommand from toad.widgets.prompt import PromptTextArea @@ -81,6 +85,7 @@ def retreat(self, editor, movement): class HistoryCursor(DeclaredFamily, affix="Cursor"): direction: int + native_action: ClassVar[FunctionType] @classmethod @abstractmethod @@ -88,8 +93,6 @@ def at_edge(cls, editor: PromptTextArea) -> bool: ... @classmethod def try_move(cls, editor: PromptTextArea, select: bool, mode: EditingMode) -> bool: - from toad.core.input_events import HistoryMove - if (mode.allows_history_navigation and editor.selection.is_empty and not select): if cls.at_edge(editor): @@ -98,21 +101,19 @@ def try_move(cls, editor: PromptTextArea, select: bool, mode: EditingMode) -> bo return False @classmethod - def move(cls, editor: PromptTextArea, select: bool, ordinary: Callable[[bool], object]) -> None: + def move(cls, editor: PromptTextArea, select: bool) -> None: if not cls.try_move(editor, select, editor.editing_mode): - ordinary(select) + cls.native_action(editor, select) @classmethod - def move_motion(cls, editor: PromptTextArea, mode: VimNormal, motion: type[LineMotion]) -> bool: - # Direction is the native vertical displacement, not a mode/key name. - # New counted vertical motions borrow the same two history boundaries. - cursor = next(cursor for cursor in cls.members_with(cls) - if cursor.direction == motion.direction) - return cursor.try_move(editor, False, mode) + def try_input(cls, editor: PromptTextArea, action: FunctionType) -> bool: + return any(cursor.try_move(editor, False, editor.editing_mode) + for cursor in cls.members_with(cls) if cursor.native_action is action) class PreviousHistoryCursor(HistoryCursor): direction = -1 + native_action = TextArea.action_cursor_up @classmethod def at_edge(cls, editor): @@ -121,6 +122,7 @@ def at_edge(cls, editor): class NextHistoryCursor(HistoryCursor): direction = 1 + native_action = TextArea.action_cursor_down @classmethod def at_edge(cls, editor): diff --git a/src/toad/session_presentation.py b/src/toad/session_presentation.py index c2d86cc23..340a19ffe 100644 --- a/src/toad/session_presentation.py +++ b/src/toad/session_presentation.py @@ -11,6 +11,7 @@ from agent_comms.acp_extension import CoordinationChangedUpdate from agent_comms.session_load import EnsuringSessionLoadPlan from agent_comms.ui_model.message_category import MessageCategory +from textual.document._editing import EditingMode from textual.widget import Widget from textual.widgets.text_area import TextAreaState @@ -35,8 +36,30 @@ class SessionSurfaceSlot(Widget): DEFAULT_CSS = "SessionSurfaceSlot { display: none; }" +class NativeEditorSelection: + @property + def initial_editing_mode(self) -> EditingMode: + return EditingMode() + + +class SessionViewState(NativeEditorSelection): + """An uncaptured surface has no retained resources to restore or close.""" + + def prepare_conversation(self, conversation: Conversation) -> None: + pass + + def prepare_transcript(self, conversation: Conversation) -> None: + pass + + async def restore(self, conversation: Conversation) -> None: + pass + + async def close(self) -> None: + pass + + @dataclass(frozen=True) -class SessionViewState: +class RetainedSessionViewState(SessionViewState): """The actual editor document/history and reader intent, not a transcript copy.""" editor: TextAreaState @@ -47,7 +70,7 @@ class SessionViewState: initial_prompt: str | None @classmethod - def capture(cls, conversation: Conversation) -> SessionViewState: + def capture(cls, conversation: Conversation) -> RetainedSessionViewState: editor = conversation.prompt.prompt_text_area return cls( editor.capture_editor_state(), @@ -58,18 +81,28 @@ def capture(cls, conversation: Conversation) -> SessionViewState: conversation.take_initial_prompt(), ) + @property + def initial_editing_mode(self) -> EditingMode: + return self.editor.editing_mode + + def prepare_conversation(self, conversation: Conversation) -> None: + conversation._initial_prompt = self.initial_prompt + + async def close(self) -> None: + await self.editor.editing_mode.close() + def prepare_transcript(self, conversation: Conversation) -> None: conversation.visible_categories = self.visible_categories conversation.window.reader_position = self.reader_position - def restore(self, conversation: Conversation) -> None: - conversation.input_histories = self.input_histories + async def restore(self, conversation: Conversation) -> None: editor = conversation.prompt.prompt_text_area - editor.restore_editor_state(self.editor) + await editor.restore_editor_state(self.editor) + conversation.input_histories = self.input_histories editor.shell_mode = self.shell_mode -class SessionSurfaceLifetime(ABC): +class SessionSurfaceLifetime(NativeEditorSelection, ABC): """A session declares its own UI placement and activation lifetime.""" @property @@ -93,7 +126,11 @@ class EditorSessionSurfaceLifetime(SessionSurfaceLifetime): """Own retained editor state and active shell close semantics.""" def __init__(self) -> None: - self.state: SessionViewState | None = None + self.state: SessionViewState = SessionViewState() + + @property + def initial_editing_mode(self) -> EditingMode: + return self.state.initial_editing_mode @abstractmethod async def release_binding( @@ -349,17 +386,14 @@ async def acquire(self, screen: MainScreen): self.widget = conversation screen._initial_prompt = None if not returning: - if self.state is not None: - conversation._initial_prompt = self.state.initial_prompt + self.state.prepare_conversation(conversation) await content.mount(conversation, before=slot) screen.thread_identity.bind_history(conversation.input_histories) yield conversation - if self.state is not None: - self.state.prepare_transcript(conversation) + self.state.prepare_transcript(conversation) await self.transcript.bind(conversation).restore_documents(self.transcript) - if self.state is not None: - self.state.restore(conversation) - self.state = None + await self.state.restore(conversation) + self.state = SessionViewState() await self.attach_binding(conversation) await self.transcript.bind(conversation).restore(self.transcript) if returning: @@ -367,7 +401,8 @@ async def acquire(self, screen: MainScreen): conversation.prompt.focus() async def close(self, screen: MainScreen) -> None: - self.state = None + await self.state.close() + self.state = SessionViewState() async with AsyncResourceRelease() as release: release.push_async_callback(self.sources.close) release.push_async_callback(self.transcript.close) @@ -377,7 +412,10 @@ async def evict(self, screen: MainScreen) -> None: """Release the owned native tree while retaining its actual editor resource.""" if (conversation := self.widget) is None: return - self.state = SessionViewState.capture(conversation) + editor = conversation.prompt.prompt_text_area + if not await editor.editing_mode.flush(editor): + return # Retain the actual view/engine while its modal input is pending. + self.state = RetainedSessionViewState.capture(conversation) await self.remove_native_tree(screen) def external_transcript_changed(self) -> None: diff --git a/src/toad/widgets/conversation.py b/src/toad/widgets/conversation.py index e091d0c12..eff132af1 100644 --- a/src/toad/widgets/conversation.py +++ b/src/toad/widgets/conversation.py @@ -974,6 +974,11 @@ def compose(self) -> ComposeResult: model_history_scope=Conversation.model_history_scope, status=Conversation.status, ) + # A retained editor is installed before Prompt.on_mount can apply a + # profile. Returning to a tab must not spawn a throwaway Neovim. + prompt.initial_editing_mode = self.query_ancestor( + SessionScope + ).presentation.initial_editing_mode self.goal_bar: GoalBar = GoalBar(self, prompt=prompt) yield self.goal_bar yield prompt diff --git a/src/toad/widgets/prompt.py b/src/toad/widgets/prompt.py index 94639b4e1..6b51387b3 100644 --- a/src/toad/widgets/prompt.py +++ b/src/toad/widgets/prompt.py @@ -4,6 +4,7 @@ import shlex from dataclasses import replace from pathlib import Path +from types import FunctionType from agent_comms.repo import Checkout from typing import TYPE_CHECKING, Literal, Self @@ -18,7 +19,8 @@ from textual.app import ComposeResult from textual.binding import ModeBinding from textual.widgets.text_area import EditingMode, VimMode -from textual.document._editing import LineMotion, VimNormal +from toad.neovim_editing import NeovimUpdated +from textual.document._editing import EditingFailure from textual.content import Content, Span from textual.css.scalar import Scalar from textual.geometry import Spacing @@ -228,6 +230,7 @@ def __init__(self, *, prompt: Prompt) -> None: super().__init__() # The composer that built this editor owns its turns, popups and modes. self.prompt = prompt + self.set_reactive(TextArea.editing_mode, prompt.initial_editing_mode) self.set_reactive(PromptTextArea.project_path, prompt.project_path) self._submit_pending = False self._submit_immediate = False @@ -271,6 +274,7 @@ def highlight_shell(self, text: str) -> Content: def on_mount(self) -> None: self.highlight_cursor_line = False self.hide_suggestion_on_blur = False + self.editing_mode.attach(self) self.apply_keyboard_profile() self.call_after_refresh(lambda: self.show_editing_mode(self.editing_mode)) @@ -283,7 +287,16 @@ def watch_has_focus(self, focus: bool) -> None: self.show_editing_mode(self.editing_mode) def apply_keyboard_profile(self) -> None: - self.editing_mode = self.screen.key_mode() + self.call_later(self._apply_keyboard_profile) + + async def _apply_keyboard_profile(self) -> None: + mode = self.screen.key_mode + if self.key_mode is not mode: + await self.switch_editing_mode(mode()) + + def on_neovim_updated(self, event: NeovimUpdated) -> None: + event.stop() + event.owner.paint(self) @property def submission_allowed(self) -> bool: @@ -304,9 +317,7 @@ def clear_suggestions(self) -> None: def _paste_clipboard_text(self, text: str) -> None: # A fresh Paste event posted directly here would bubble to App and be # forwarded back to the focused TextArea, inserting it a second time. - self.editing_mode = self.editing_mode.pasted() - if result := self._replace_via_keyboard(text, *self.selection): - self.move_cursor(result.end_location) + self.editing_mode.paste(self, text) @work(group='clipboard-paste', exclusive=True) async def action_paste_clipboard(self) -> None: @@ -378,9 +389,17 @@ async def action_tab_complete(self) -> None: priority=True), ) - def _submit(self) -> None: + async def _submit(self) -> None: """Silently cancel a submission superseded by focus, mode or popup changes.""" self._submit_pending = False + mode = self.editing_mode + try: + if not await mode.flush(self): + return + except EditingFailure: + return # The engine already reported its failure; never submit stale text. + if mode is not self.editing_mode: + return immediate, self._submit_immediate = self._submit_immediate, False if not self.has_focus or not self.submission_allowed: return @@ -410,15 +429,14 @@ def _submit(self) -> None: if shell: self.prompt.shell_mode = False - def move_line_motion(self, motion: type[LineMotion], mode: VimNormal, count: int) -> None: - if not HistoryCursor.move_motion(self, mode, motion): - super().move_line_motion(motion, mode, count) + def before_engine_input(self, action: FunctionType) -> bool: + return HistoryCursor.try_input(self, action) def action_cursor_up(self, select: bool = False): - PreviousHistoryCursor.move(self, select, super().action_cursor_up) + PreviousHistoryCursor.move(self, select) def action_cursor_down(self, select: bool = False): - NextHistoryCursor.move(self, select, super().action_cursor_down) + NextHistoryCursor.move(self, select) def action_delete_left(self) -> None: if not self.editing_mode.accepts_text: @@ -512,6 +530,7 @@ def __init__( super().__init__(name=name, id=id, classes=classes, disabled=disabled) self.set_reactive(Prompt.project_path, project) self.ask_queue: list[Ask] = [] + self.initial_editing_mode: EditingMode = EditingMode() self.turns = turns self.submissions = submissions @@ -539,10 +558,9 @@ def text(self) -> str: @text.setter def text(self, text: str) -> None: - self.prompt_text_area.text = text - self.prompt_text_area.selection = Selection.cursor( - self.prompt_text_area.get_cursor_line_end_location() - ) + editor = self.prompt_text_area + editor.text = text + editor.editing_mode.move_to_end(editor) @property def supports_completion(self) -> bool: @@ -772,10 +790,9 @@ def enter_insert(self) -> None: """ if not self.is_mounted: return - from toad.key_modes import VimInsert from toad.widgets.conversation import Conversation - self.prompt_text_area.editing_mode = VimInsert() + self.prompt_text_area.editing_mode.enter_text(self.prompt_text_area) self.query_ancestor(Conversation).focus_prompt() MODE_BINDINGS = ( diff --git a/tests/app/test_agent_startup_failure.py b/tests/app/test_agent_startup_failure.py index a6bcca840..14a2d5487 100644 --- a/tests/app/test_agent_startup_failure.py +++ b/tests/app/test_agent_startup_failure.py @@ -17,6 +17,8 @@ async def test_a_process_that_exits_during_startup_is_reported(workspace): async def test_a_refused_handshake_is_reported_as_not_sent(workspace): # The refusing process stays attached: a later owner may accept the same connection. async with workspace.app(size=(139, 25), binding=AgentAttached(peer("refuses_initialize"))) as ui: - await ui.until_shows("AGENT_REFUSED_TO_START", seconds=20) + await ui.until_shows("Thread disconnected: Request failed", seconds=20) + await ui.click("Thread disconnected: Request failed") + await ui.until_shows("AGENT_REFUSED_TO_START") assert ui.shows("Not sent") assert not ui.agent_ready() diff --git a/tests/textual/test_acp_log_link.py b/tests/textual/test_acp_log_link.py index c16d9ec20..10623d299 100644 --- a/tests/textual/test_acp_log_link.py +++ b/tests/textual/test_acp_log_link.py @@ -26,7 +26,7 @@ async def shown_log(ui): async def test_the_log_link_highlights_under_the_pointer_and_opens_the_agents_log_in_one_tab(workspace): async with workspace.app(size=(139, 30), binding=AgentAttached(peer("refuses_initialize"))) as ui: app, pilot = ui.textual - await ui.until_shows("AGENT_REFUSED_TO_START", seconds=20) + await ui.until_shows(LINK, seconds=20) await ui.until(lambda: ui.shows(LINK) and ui.link_at(LINK) is not None, why="the log link drawn as a link") owner = ui.selected_tab cell = ui._cell_of(LINK, 0) diff --git a/tests/textual/test_active_mode_keymap.py b/tests/textual/test_active_mode_keymap.py index 49e87116e..19b8362b3 100644 --- a/tests/textual/test_active_mode_keymap.py +++ b/tests/textual/test_active_mode_keymap.py @@ -1,40 +1,21 @@ -"""Application remaps follow the live editor mode and shared sidebar focus.""" +"""Application remaps work from shared sidebar focus and the Close button.""" -from textual.document._editing import VimInsert, VimNormal, VimVisual from textual.widgets import Input from toad.application_actions import CommanderAction -from toad.key_modes import VimNormal as NormalProfile +from toad.key_modes import VimNormal from toad.preferences import SETTINGS from toad.screens.commander import CommanderPage from toad.widgets.side_bar import SideBar from toad.widget_actions import DismissAction -async def test_commander_remaps_follow_editor_transitions_and_sidebar_focus(workspace): +async def test_commander_remaps_from_sidebar_and_close_button(workspace): async with workspace.app(size=(120, 40)) as ui: app, _pilot = ui.textual - app.saved_settings.set(SETTINGS.ui.key_mode, NormalProfile) - editor = ui.conversation.prompt.prompt_text_area - remaps = { - CommanderAction.binding_id(VimNormal): "f3", - CommanderAction.binding_id(VimInsert): "f4", - CommanderAction.binding_id(VimVisual): "f5", - } + app.saved_settings.set(SETTINGS.ui.key_mode, VimNormal) + remaps = {CommanderAction.binding_id(VimNormal): "f3"} app.set_keymap(remaps) - for mode, key, other in ((VimNormal(), "f3", "f4"), - (VimInsert(), "f4", "f3"), - (VimVisual(), "f5", "f3")): - editor.editing_mode = mode - editor.focus() - await ui.press(other) - assert app.screen is app.workspace_screen - await ui.press(key) - await ui.until(lambda: isinstance(app.screen, CommanderPage), - why="the commander key for the editor's active mode") - await ui.press("escape") - await ui.until(lambda: app.screen is app.workspace_screen, - why="return from the commander page") sidebar = app.workspace_chrome.channels sidebar.reveal() assert isinstance(sidebar, SideBar) diff --git a/tests/textual/test_keymap_screen.py b/tests/textual/test_keymap_screen.py index f96983409..5b9572131 100644 --- a/tests/textual/test_keymap_screen.py +++ b/tests/textual/test_keymap_screen.py @@ -5,9 +5,8 @@ from textual.binding import ModeBinding from toad.application_actions import CommanderAction, KeymapAction -from toad.key_modes import KeyMode, Standard, VimInsert +from toad.key_modes import KeyMode, Standard from toad.preferences import SETTINGS, SavedSettings -from textual.document._editing import Escape, VimInsert as InsertEditing, VimNormal from toad.screens.keymap import KeymapScreen from toad.keymap import KeymapRow from toad.widget_actions import ProfileKeys @@ -86,28 +85,6 @@ async def test_f1_menu_remaps_itself_saves_and_refuses_a_conflict(workspace, mod why="the saved menu remap after restarting the app") -async def test_the_menu_remaps_the_native_insert_escape_without_changing_normal(workspace): - async with workspace.app(size=(140, 45)) as ui: - app = ui.app - app.saved_settings.set(SETTINGS.ui.key_mode, VimInsert) - editor = ui.conversation.prompt.prompt_text_area - editor.focus() - await ui.press("f1") - await ui.until(lambda: isinstance(app.screen, KeymapScreen), why="the native command menu") - await choose(ui, Escape(), VimInsert) - app.screen.query_one("#keymap-keys", Input).value = "f6" - await ui.pilot.click("#keymap-apply") - await ui.until(lambda: Escape.binding_id(InsertEditing) in app._keymap, - why="the native Insert escape override") - await ui.press("escape") - await ui.until(lambda: app.screen is app.workspace_screen, why="the menu dismissed") - editor.focus() - await ui.press("escape") - assert isinstance(editor.editing_mode, InsertEditing) - await ui.press("f6") - assert type(editor.editing_mode) is VimNormal - - async def test_the_menu_discovers_and_remaps_a_native_callable_with_both_aliases(workspace): async with workspace.app(size=(140, 45)) as ui: counter = CallCounter("0") diff --git a/tests/textual/test_neovim_wire.py b/tests/textual/test_neovim_wire.py new file mode 100644 index 000000000..bb1cbe5f0 --- /dev/null +++ b/tests/textual/test_neovim_wire.py @@ -0,0 +1,49 @@ +"""The retained native win_viewport handle must survive the actual RPC decoder.""" + +import msgpack +import pytest + +from toad.neovim_editing import ( + BufferHandle, + NeovimCall, + NeovimEditing, + NeovimWireFailure, + TabpageHandle, + WindowHandle, +) + + +async def test_native_redraw_and_response_keep_nominal_window() -> None: + owner = NeovimEditing() + owner.accept_api_info([1, {"types": { + "Buffer": {"id": 0, "prefix": "nvim_buf_"}, + "Window": {"id": 1, "prefix": "nvim_win_"}, + "Tabpage": {"id": 2, "prefix": "nvim_tabpage_"}, + }}]) + # Root's real nvim 0.12.4 first redraw carried this extension in win_viewport. + native_window = msgpack.ExtType(1, b"\xcd\x03\xe8") + owner._unpacker.feed(msgpack.packb([ + 2, "redraw", [["win_viewport", [1, native_window, 0, 1, 0, 0, 1, 0]]], + ], use_bin_type=True)) + redraw, = owner._unpacker + window = redraw[2][0][1][1] + assert window == WindowHandle(1000) + assert window != BufferHandle(1000) + assert window != TabpageHandle(1000) + await owner._receive(redraw) + + request = NeovimCall("nvim_get_current_win", []) + owner._register(request) + owner._unpacker.feed(msgpack.packb([1, request.identity, None, native_window])) + response, = owner._unpacker + await owner._receive(response) + assert request.reply.result() == window + assert owner._encode_handle(window) == native_window + + with pytest.raises(NeovimWireFailure, match="arity"): + await owner._receive([2, "redraw"]) + owner._unpacker.feed(msgpack.packb([ + 2, "redraw", [msgpack.ExtType(127, msgpack.packb(1000))], + ])) + with pytest.raises(NeovimWireFailure, match="Unknown Neovim extension code"): + tuple(owner._unpacker) diff --git a/tests/textual/test_prompt_enter_insert.py b/tests/textual/test_prompt_enter_insert.py index f4258fe5d..d73c4be36 100644 --- a/tests/textual/test_prompt_enter_insert.py +++ b/tests/textual/test_prompt_enter_insert.py @@ -1,22 +1,8 @@ """Textual-facing: vim `i` asks the prompt for insert mode; the prompt owns whether its editor exists.""" from agent_comms.repo import Checkout -from textual.widgets.text_area import VimInsert - -from support.agents import peer -from toad.session_scope import AgentAttached -from toad.widgets.prompt import MessagePrompt, Prompt +from toad.widgets.prompt import MessagePrompt def test_enter_insert_before_compose_has_no_editor_to_touch(tmp_path): MessagePrompt(project=Checkout(tmp_path)).enter_insert() - - -async def test_enter_insert_sets_insert_mode_and_focuses_the_editor(workspace): - async with workspace.app(binding=AgentAttached(peer("echo"))) as ui: - await ui.until_agent_ready() - app, _pilot = ui.textual - prompt = app.selected_session.query_one(Prompt) - prompt.enter_insert() - await ui.until(lambda: app.focused is prompt.prompt_text_area, why="the editor to take focus") - assert isinstance(prompt.prompt_text_area.editing_mode, VimInsert) diff --git a/uv.lock b/uv.lock index eb50833df..eb16acae4 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 3 +revision = 5 requires-python = ">=3.14.8, <3.15" [[package]] @@ -234,6 +234,7 @@ dependencies = [ { name = "httpx" }, { name = "jeepney", marker = "sys_platform == 'linux'" }, { name = "metaclass-registry" }, + { name = "msgpack" }, { name = "notify-py" }, { name = "objectstate" }, { name = "packaging" }, @@ -279,6 +280,7 @@ requires-dist = [ { name = "httpx", specifier = "==0.28.1" }, { name = "jeepney", marker = "sys_platform == 'linux'", specifier = "==0.9.0" }, { name = "metaclass-registry", git = "https://github.com/OpenHCSDev/metaclass-registry?branch=main" }, + { name = "msgpack", specifier = "==1.2.2" }, { name = "notify-py", specifier = "==0.3.43" }, { name = "objectstate", git = "https://github.com/OpenHCSDev/objectstate.git?rev=b5dd9f78b9293df145d7629f21deb89b042b9400" }, { name = "packaging", specifier = "==26.2" }, @@ -1326,7 +1328,7 @@ wheels = [ [[package]] name = "textual" version = "8.2.8" -source = { git = "https://github.com/OpenHCSDev/textual.git?branch=main#e7a48c59e22d3a4ea9ebcb64c5a2dcb40673f780" } +source = { git = "https://github.com/OpenHCSDev/textual.git?branch=main#bea0d377b9bc414dc377f1ebcbb51405fc7b0e03" } dependencies = [ { name = "markdown-it-py", extra = ["linkify"] }, { name = "mdit-py-plugins" },