From 3916e40d7c0f6621b837b036a2509c20e54fb3d2 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Mon, 17 Aug 2026 06:45:30 -0500 Subject: [PATCH 1/8] Engines(fix[control]): Preserve transport truth why: Direct argv parsing, pinned server scope, control-mode correlation, and async subprocess cleanup could misroute commands, misattribute results, leak process resources, or leave pane output encoded. what: - Protect literal semicolons and reject incompatible engine connections - Correlate control replies without treating guard-shaped output as protocol - Drain and reap async children through cancellation and reader failures - Expose decoded pane-output bytes to control-mode consumers --- docs/api/libtmux.engines.md | 2 +- docs/experimental/engines.md | 14 +- .../engines/async-control-mode.md | 43 +- docs/experimental/engines/control-mode.md | 11 +- docs/experimental/engines/imsg.md | 2 +- docs/experimental/tutorials/control-mode.md | 4 +- docs/experimental/tutorials/imsg-parity.md | 2 +- docs/topics/engines.md | 69 ++- src/libtmux/_internal/tmux_argv.py | 234 ++++++++ src/libtmux/engines/__init__.py | 4 +- src/libtmux/engines/base.py | 54 +- src/libtmux/engines/connection.py | 259 ++++++++- src/libtmux/engines/subprocess.py | 6 +- src/libtmux/exc.py | 55 +- src/libtmux/experimental/engines/__init__.py | 6 + .../engines/async_control_mode.py | 282 +++++++-- src/libtmux/experimental/engines/asyncio.py | 199 ++++++- src/libtmux/experimental/engines/base.py | 159 ++--- .../experimental/engines/connection.py | 2 +- .../experimental/engines/control_mode.py | 171 +++--- src/libtmux/experimental/engines/imsg/base.py | 39 +- .../experimental/engines/subprocess.py | 28 + src/libtmux/experimental/mcp/_settle.py | 61 +- src/libtmux/experimental/mcp/events.py | 2 +- src/libtmux/neo.py | 9 +- src/libtmux/pane.py | 3 +- src/libtmux/server.py | 152 +++-- src/libtmux/window.py | 9 +- tests/_internal/test_tmux_argv.py | 191 ++++++ .../contract/test_async_control_engine.py | 99 +++- ..._async_control_engine_workspace_builder.py | 26 +- .../contract/test_async_engine.py | 236 ++++++++ .../test_async_control_mode_lifecycle.py | 20 + .../test_async_control_mode_supervisor.py | 347 ++++++++++- tests/experimental/engines/test_base.py | 14 + .../engines/test_control_group_failures.py | 204 +++++++ .../engines/test_control_mode_correlation.py | 78 ++- tests/experimental/engines/test_imsg.py | 31 + tests/experimental/mcp/test_settle.py | 10 +- tests/test_engine_connections.py | 544 ++++++++++++++++++ tests/test_engines.py | 40 +- tests/test_pane.py | 44 ++ tests/test_server.py | 10 +- 43 files changed, 3230 insertions(+), 545 deletions(-) create mode 100644 src/libtmux/_internal/tmux_argv.py create mode 100644 tests/_internal/test_tmux_argv.py create mode 100644 tests/experimental/engines/test_control_group_failures.py create mode 100644 tests/test_engine_connections.py diff --git a/docs/api/libtmux.engines.md b/docs/api/libtmux.engines.md index 862ca28c0f..4ff8799c25 100644 --- a/docs/api/libtmux.engines.md +++ b/docs/api/libtmux.engines.md @@ -41,7 +41,7 @@ also implement. A {class}`~libtmux.engines.connection.ServerConnection` is the pair every engine needs before it can dispatch anything: which tmux *binary* to run, and the -connection flags (`-L`/`-S`/`-f`/`-2`/`-8`) naming one tmux server. It is the +connection flags (`-L`/`-S`/`-f`/`-2`) naming one tmux server. It is the single place either is computed. ```{eval-rst} diff --git a/docs/experimental/engines.md b/docs/experimental/engines.md index 652ce63962..de2dbb0b5f 100644 --- a/docs/experimental/engines.md +++ b/docs/experimental/engines.md @@ -39,10 +39,10 @@ event loop. ## Engine boundary A synchronous engine satisfies -{class}`~libtmux.experimental.engines.base.TmuxEngine`; an async engine +{class}`~libtmux.engines.base.TmuxEngine`; an async engine satisfies {class}`~libtmux.experimental.engines.base.AsyncTmuxEngine`. Both -accept a {class}`~libtmux.experimental.engines.base.CommandRequest` and produce -a raw {class}`~libtmux.experimental.engines.base.CommandResult`. +accept a {class}`~libtmux.engines.base.CommandRequest` and produce +a raw {class}`~libtmux.engines.base.CommandResult`. {func}`~libtmux.experimental.ops.run` and {func}`~libtmux.experimental.ops.arun` own the next boundary: they render an operation and convert the raw command outcome to its declared typed result. @@ -68,16 +68,10 @@ engine-owned workflow. ## Shared API ```{eval-rst} -.. autoclass:: libtmux.experimental.engines.base.TmuxEngine - :members: - .. autoclass:: libtmux.experimental.engines.base.AsyncTmuxEngine :members: -.. autoclass:: libtmux.experimental.engines.base.CommandRequest - :members: - -.. autoclass:: libtmux.experimental.engines.base.CommandResult +.. autoclass:: libtmux.experimental.engines.base.SupportsAsyncTmuxVersion :members: .. autofunction:: libtmux.experimental.engines.registry.available_engines diff --git a/docs/experimental/engines/async-control-mode.md b/docs/experimental/engines/async-control-mode.md index b17a88d8de..0f1ef376a7 100644 --- a/docs/experimental/engines/async-control-mode.md +++ b/docs/experimental/engines/async-control-mode.md @@ -62,8 +62,20 @@ and Per-subscriber queues are bounded; {attr}`~libtmux.experimental.engines.async_control_mode.AsyncControlModeEngine.dropped_notifications` reports overflow. Connection failures and timeouts raise at the engine -boundary, while tmux command errors remain result data. Sequence anomalies are -logged. +boundary, while tmux command errors remain result data. The engine drains the +control process's stderr concurrently and includes its bounded recent tail in +connection and protocol failures. Cleanup joins that reader and reaps the +process before returning, including when the caller waiting on cleanup is +cancelled. A backwards command number or a solicited block with no pending +request raises a protocol error and restarts the connection rather than risking +result misattribution. + +Tmux does not escape command-output lines inside these blocks. Output that +resembles a nonmatching guard remains data, but output byte-for-byte identical +to its own closing guard is indistinguishable from protocol framing. This is a +tmux control-protocol limitation; use +{class}`~libtmux.experimental.engines.asyncio.AsyncSubprocessEngine` when +arbitrary output must round-trip without that ambiguity. The persistent process has normal tmux client semantics: `list-clients` shows it, `session_attached` includes it, and client attach and detach hooks can @@ -83,15 +95,40 @@ after a safe session exists; direct startup raises available. The API has no subscriber-readiness signal, so code must not assume a notification emitted before the first iteration will be delivered. +## Pane-output bytes + +{class}`~libtmux.experimental.engines.async_control_mode.ControlNotification` +keeps the encoded, human-readable control line in `raw` and its exact bytes in +`raw_bytes` for wire diagnostics. For `%output` and `%extended-output`, it also +exposes the pane ID and decoded bytes in `pane_id` and `payload`. Decode text +only at the application boundary: tmux passes pane bytes through without +validating UTF-8. + +```python +>>> from libtmux.experimental.engines import ControlNotification +>>> event = ControlNotification.parse(b"%output %7 hello\\012world\\134") +>>> event.pane_id, event.payload +('%7', b'hello\nworld\\') +>>> event.raw +'%output %7 hello\\012world\\134' +``` + +Tmux uses a backslash followed by exactly three octal digits for an escaped +byte. Shorter octal-looking text remains literal. Consumers normally read +`payload`; `raw` remains useful when diagnosing the protocol stream. + ## API ```{eval-rst} .. autoclass:: libtmux.experimental.engines.async_control_mode.AsyncControlModeEngine :members: :special-members: __aenter__, __aexit__ + +.. autoclass:: libtmux.experimental.engines.async_control_mode.ControlNotification + :members: ``` ## Related tutorial See {doc}`../tutorials/async-control-plans` to compose forward-referenced -operations and fold them into control-mode dispatches. +operations and pipeline their ordered requests over control mode. diff --git a/docs/experimental/engines/control-mode.md b/docs/experimental/engines/control-mode.md index 2d421bd576..16834a5e7e 100644 --- a/docs/experimental/engines/control-mode.md +++ b/docs/experimental/engines/control-mode.md @@ -53,7 +53,16 @@ writes the batch before collecting correlated control-mode result blocks. The engine drains unsolicited notifications so they are not mistaken for command replies. Timeouts, connection death, and write failures raise {exc}`~libtmux.experimental.engines.control_mode.ControlModeError`; tmux -`%error` blocks remain command-result data. Sequence anomalies are logged. +`%error` blocks remain command-result data. A backwards command number or a +solicited block with no pending request raises a protocol error and closes the +connection rather than risking result misattribution. + +Tmux does not escape command-output lines inside these blocks. Output that +resembles a nonmatching guard remains data, but output byte-for-byte identical +to its own closing guard is indistinguishable from protocol framing. This is a +tmux control-protocol limitation; use +{class}`~libtmux.experimental.engines.subprocess.SubprocessEngine` when +arbitrary output must round-trip without that ambiguity. The persistent process has normal tmux client semantics: `list-clients` shows it, `session_attached` includes it, and client attach and detach hooks can diff --git a/docs/experimental/engines/imsg.md b/docs/experimental/engines/imsg.md index f8d3619248..76760a2d7f 100644 --- a/docs/experimental/engines/imsg.md +++ b/docs/experimental/engines/imsg.md @@ -22,7 +22,7 @@ method. Local queries and commands that must start a missing server use the tmux binary instead. Unlike the subprocess and control-mode engines, it has no `for_server()` helper. Put a private server's raw `-L` or `-S` global argument in every -{class}`~libtmux.experimental.engines.base.CommandRequest`. +{class}`~libtmux.engines.base.CommandRequest`. ```python >>> from libtmux.experimental.engines import CommandRequest, ImsgEngine diff --git a/docs/experimental/tutorials/control-mode.md b/docs/experimental/tutorials/control-mode.md index bcd5e5f5da..0d08fca839 100644 --- a/docs/experimental/tutorials/control-mode.md +++ b/docs/experimental/tutorials/control-mode.md @@ -3,7 +3,7 @@ {class}`~libtmux.experimental.engines.control_mode.ControlModeEngine` keeps one `tmux -C` client alive once connected and correlates tmux's framed replies with submitted -{class}`~libtmux.experimental.engines.base.CommandRequest` values. +{class}`~libtmux.engines.base.CommandRequest` values. {meth}`~libtmux.experimental.engines.control_mode.ControlModeEngine.run_batch` pipelines an ordered batch instead of starting one process per request. @@ -60,7 +60,7 @@ True ``` The values are live raw -{class}`~libtmux.experimental.engines.base.CommandResult` instances because +{class}`~libtmux.engines.base.CommandResult` instances because `run_batch` is the engine boundary. Use {func}`~libtmux.experimental.ops.run` or a plan when the caller needs operation-specific result subtypes. diff --git a/docs/experimental/tutorials/imsg-parity.md b/docs/experimental/tutorials/imsg-parity.md index 601b3fe627..af91a68b3c 100644 --- a/docs/experimental/tutorials/imsg-parity.md +++ b/docs/experimental/tutorials/imsg-parity.md @@ -10,7 +10,7 @@ standard output. `ImsgEngine` has no `for_server()` helper. Include the server's `-L` socket name or `-S` socket path in every -{class}`~libtmux.experimental.engines.base.CommandRequest`. The subprocess +{class}`~libtmux.engines.base.CommandRequest`. The subprocess engine accepts the same global argument in the request, which keeps the comparison exact. diff --git a/docs/topics/engines.md b/docs/topics/engines.md index 3100072a25..674049f8a0 100644 --- a/docs/topics/engines.md +++ b/docs/topics/engines.md @@ -172,14 +172,28 @@ server would silently dispatch to whichever server a flagless `tmux` reaches: ('-Lengines_doc_c',) ``` -An engine that *does* name a server is left exactly as you built it: +An engine already on the requested server is left exactly as you built it: ```python >>> from libtmux.engines import SubprocessEngine >>> from libtmux.server import Server ->>> pinned = SubprocessEngine.of(server_args=("-Lengines_doc_pinned",)) +>>> pinned = SubprocessEngine.of(server_args=("-Lengines_doc_c",)) >>> Server(socket_name="engines_doc_c", engine=pinned).engine.server_args -('-Lengines_doc_pinned',) +('-Lengines_doc_c',) +``` + +Conflicting explicit scopes fail before a command can reach the wrong server: + +```python +>>> from libtmux import exc +>>> from libtmux.engines import SubprocessEngine +>>> from libtmux.server import Server +>>> pinned = SubprocessEngine.of(server_args=("-Lengines_doc_pinned",)) +>>> try: +... Server(socket_name="engines_doc_c", engine=pinned).engine +... except exc.EngineConfigurationMismatch: +... print("connection mismatch") +connection mismatch ``` An in-memory engine has no connection at all, so neither rule applies and it is @@ -192,19 +206,26 @@ An engine may implement extra protocols. Each is optional; libtmux checks with {class}`~libtmux.engines.base.SupportsCommandLine` renders the argv an engine *would* run, which is how the full command line reaches the debug log before -dispatch. {class}`~libtmux.engines.base.SupportsConnection` marks an engine that -dispatches over a named server and can be rebound — the protocol behind the -adoption rule above. +dispatch. {class}`~libtmux.engines.base.HasConnection` exposes a transport's +server scope for validation. {class}`~libtmux.engines.base.SupportsConnection` +adds safe rebinding; stateless subprocess engines implement both, while a live +control connection implements only the read-only capability and must already +match the Server. ```python >>> from libtmux.engines import ( ... SubprocessEngine, +... HasConnection, ... SupportsCommandLine, ... SupportsConnection, ... ) >>> engine = SubprocessEngine() ->>> isinstance(engine, SupportsCommandLine), isinstance(engine, SupportsConnection) -(True, True) +>>> ( +... isinstance(engine, SupportsCommandLine), +... isinstance(engine, HasConnection), +... isinstance(engine, SupportsConnection), +... ) +(True, True, True) ``` An engine that implements neither simply is not matched: @@ -227,10 +248,11 @@ in-memory fake — omits it, and the caller assumes the newest tmux. ## Explicit command separators -tmux treats a bare `;` argument as a boundary between two commands, but only -when it arrives unquoted. A `;` that is *data* — a pane title, a shell fragment -bound for `send-keys` — must not be mistaken for one. Guessing from the string -alone cannot tell them apart, so the intent rides in the type: +Tmux's direct argv parser treats an unescaped `;` at the end of any token as a +boundary between two commands. That includes both a standalone `;` and a value +such as `title;`; an interior semicolon remains data. Guessing intent from the +string alone cannot distinguish a literal suffix from command structure, so +the intent rides in the type: {class}`~libtmux.engines.base.CommandSeparator` marks a real boundary, and {func}`~libtmux.engines.base.is_command_separator` finds it. @@ -243,8 +265,7 @@ alone cannot tell them apart, so the intent rides in the type: [False, False, True, False, False] ``` -A plain `";"` is data and stays data, so nothing an existing caller passes can -become a boundary by accident: +A plain `";"` and any other ordinary trailing semicolon are encoded as data: ```python >>> from libtmux.engines import is_command_separator @@ -252,15 +273,21 @@ become a boundary by accident: False ``` -The marker survives normalization, so an engine that chains commands into one -dispatch can find the boundaries while every other engine ignores them. The -default {class}`~libtmux.engines.subprocess.SubprocessEngine` sends one command -per dispatch and has no use for them. +An existing `\;` suffix remains tmux escape syntax and is not escaped again; +tmux consumes that backslash while producing a literal semicolon. Use the +typed marker for structure and an ordinary unescaped suffix for new literal +data. + +The marker survives request normalization. Direct subprocess engines render it +as a bare structural token while escaping ordinary suffix semicolons; control +engines render ordinary values as quoted data. Callers that intentionally used +a plain `";"` to group commands must replace it with `CommandSeparator(";")`. ## What an engine does not change An engine chooses *how* a command runs, not what libtmux does with the answer. -Arguments reach tmux exactly as they always have, results read exactly as they -always have, and {meth}`Server.cmd() ` still returns a +Unescaped suffix semicolons remain argument data while each engine applies its +transport encoding, results read as before, and +{meth}`Server.cmd() ` still returns a {class}`~libtmux.common.tmux_cmd`. Under the default engine there is nothing new -to learn and nothing to migrate. +to configure. diff --git a/src/libtmux/_internal/tmux_argv.py b/src/libtmux/_internal/tmux_argv.py new file mode 100644 index 0000000000..ab10aa6618 --- /dev/null +++ b/src/libtmux/_internal/tmux_argv.py @@ -0,0 +1,234 @@ +"""Parse and normalize arguments passed directly to the tmux executable.""" + +from __future__ import annotations + +import typing as t + +from libtmux.engines.base import CommandSeparator, is_command_separator + + +class DirectArgv(t.NamedTuple): + """Tmux arguments split at the client-option boundary. + + Attributes + ---------- + global_args : tuple[str, ...] + Client-global options and their values. + command_argv : tuple[str | CommandSeparator, ...] + Command names, flags, data, and explicit separators. + """ + + global_args: tuple[str, ...] + command_argv: tuple[str | CommandSeparator, ...] + + +class ClientOption(t.NamedTuple): + """One parsed tmux client-global option. + + Attributes + ---------- + name : str + Single-character option name without ``-``. + value : str or None + Attached or separate value for value-taking options. + """ + + name: str + value: str | None + + +_CLIENT_OPTION_VALUE_CHARACTERS = frozenset("cfLST") +_CLIENT_FLAG_CHARACTERS = frozenset("2CDdhlNquUvV") + + +def _scan_client_prefix( + raw_args: tuple[object, ...], +) -> tuple[int, tuple[ClientOption, ...]]: + """Return the client-prefix boundary and its normalized options.""" + if any("\0" in str(arg) for arg in raw_args): + msg = "tmux command arguments cannot contain NUL" + raise ValueError(msg) + options: list[ClientOption] = [] + index = 0 + while index < len(raw_args): + arg = raw_args[index] + if type(arg) is CommandSeparator: + break + token = str(arg) + if "\0" in token: + msg = "tmux command arguments cannot contain NUL" + raise ValueError(msg) + if token == "--": + index += 1 + break + if len(token) <= 1 or not token.startswith("-"): + break + + option_start = len(options) + is_client_option = True + position = 0 + cluster = token[1:] + while position < len(cluster): + option = cluster[position] + if option in _CLIENT_FLAG_CHARACTERS: + options.append(ClientOption(option, None)) + position += 1 + continue + if option in _CLIENT_OPTION_VALUE_CHARACTERS: + attached = cluster[position + 1 :] + if attached: + value: str | None = attached + elif index + 1 < len(raw_args): + index += 1 + value = str(raw_args[index]) + else: + value = None + options.append(ClientOption(option, value)) + break + is_client_option = False + del options[option_start:] + break + if not is_client_option: + break + index += 1 + return index, tuple(options) + + +def split_direct_argv(args: t.Iterable[object]) -> DirectArgv: + """Separate tmux client-global options from command arguments. + + Parameters + ---------- + args : iterable[object] + Arguments passed after the tmux executable. + + Returns + ------- + DirectArgv + Client-global prefix and command region. + + Examples + -------- + >>> parsed = split_direct_argv(("-L", "socket", "display-message", ";")) + >>> parsed.global_args + ('-L', 'socket') + >>> parsed.command_argv + ('display-message', ';') + """ + raw_args = tuple(args) + command_start, _options = _scan_client_prefix(raw_args) + + global_args = tuple(str(arg) for arg in raw_args[:command_start]) + command_argv = tuple( + arg if type(arg) is CommandSeparator else str(arg) + for arg in raw_args[command_start:] + ) + if any("\0" in str(arg) for arg in command_argv): + msg = "tmux command arguments cannot contain NUL" + raise ValueError(msg) + if any( + type(arg) is CommandSeparator and not is_command_separator(arg) + for arg in command_argv + ): + msg = "a command separator must be exactly ';'" + raise ValueError(msg) + return DirectArgv(global_args=global_args, command_argv=command_argv) + + +def parse_client_options(args: t.Iterable[object]) -> tuple[ClientOption, ...]: + """Parse normalized options from the leading tmux client argv region. + + Parameters + ---------- + args : iterable[object] + Arguments passed after the tmux executable. + + Returns + ------- + tuple[ClientOption, ...] + Options in command-line order, with clusters and values expanded. + + Examples + -------- + >>> options = parse_client_options(("-uqLwork", "list-sessions")) + >>> [(option.name, option.value) for option in options] + [('u', None), ('q', None), ('L', 'work')] + >>> parse_client_options(("-f", "/tmp/tmux.conf", "list-sessions"))[-1] + ClientOption(name='f', value='/tmp/tmux.conf') + """ + _boundary, options = _scan_client_prefix(tuple(args)) + return options + + +def encode_command_argv(args: t.Iterable[object]) -> tuple[str, ...]: + r"""Encode argv already known to be in tmux's command parser region. + + Parameters + ---------- + args : iterable[object] + Command names, flags, values, and typed separators. + + Returns + ------- + tuple[str, ...] + Command argv with literal suffix semicolons protected. + + Examples + -------- + >>> encode_command_argv(("-L", "value;", CommandSeparator(";"))) + ('-L', 'value\\;', ';') + """ + command_argv = tuple( + arg if type(arg) is CommandSeparator else str(arg) for arg in args + ) + if any("\0" in str(arg) for arg in command_argv): + msg = "tmux command arguments cannot contain NUL" + raise ValueError(msg) + if any( + type(arg) is CommandSeparator and not is_command_separator(arg) + for arg in command_argv + ): + msg = "a command separator must be exactly ';'" + raise ValueError(msg) + + encoded: list[str] = [] + for arg in command_argv: + if is_command_separator(arg): + encoded.append(str.__str__(arg)) + elif arg.endswith(";") and not arg.endswith(r"\;"): + encoded.append(f"{arg[:-1]}\\;") + else: + encoded.append(arg) + return tuple(encoded) + + +def encode_direct_argv(args: t.Iterable[object]) -> tuple[str, ...]: + r"""Render direct tmux argv with untyped trailing semicolons as data. + + An ordinary string ending in a bare semicolon receives the tmux-level + escape needed to keep it literal. Existing ``\;`` suffixes are preserved, + while a :class:`~libtmux.engines.base.CommandSeparator` renders as command + structure. Client-global option values remain unchanged because tmux + removes them before parsing command separators. + + Parameters + ---------- + args : iterable[object] + Arguments passed after the tmux executable. + + Returns + ------- + tuple[str, ...] + Normalized subprocess arguments. + + Examples + -------- + >>> encode_direct_argv(("-Lsocket;", "display-message", "value;")) + ('-Lsocket;', 'display-message', 'value\\;') + >>> encode_direct_argv(("display-message", CommandSeparator(";"))) + ('display-message', ';') + >>> encode_direct_argv(("display-message", r"already\;")) + ('display-message', 'already\\;') + """ + direct = split_direct_argv(args) + return (*direct.global_args, *encode_command_argv(direct.command_argv)) diff --git a/src/libtmux/engines/__init__.py b/src/libtmux/engines/__init__.py index bba2ae94c4..3dc952d7d4 100644 --- a/src/libtmux/engines/__init__.py +++ b/src/libtmux/engines/__init__.py @@ -30,7 +30,7 @@ >>> engine.seen [('list-sessions',)] -The connection flags (``-L``/``-S``/``-f``/``-2``/``-8``) are *not* part of a +The connection flags (``-L``/``-S``/``-f``/``-2``) are *not* part of a request: they belong to the engine's :class:`~libtmux.engines.connection.ServerConnection`, so every engine sees the same request regardless of which tmux server it targets. @@ -42,6 +42,7 @@ CommandRequest, CommandResult, CommandSeparator, + HasConnection, SupportsCommandLine, SupportsConnection, SupportsTmuxVersion, @@ -63,6 +64,7 @@ "CommandResult", "CommandSeparator", "CountingSink", + "HasConnection", "InstrumentedEngine", "ServerConnection", "Sink", diff --git a/src/libtmux/engines/base.py b/src/libtmux/engines/base.py index 224b56859b..921381a387 100644 --- a/src/libtmux/engines/base.py +++ b/src/libtmux/engines/base.py @@ -19,16 +19,17 @@ from typing_extensions import Self + from libtmux.engines.connection import ServerConnection + class CommandSeparator(str): """A caller-authored command boundary, distinct from a literal ``";"``. - tmux treats a bare ``;`` argument as a command separator only when it - arrives unquoted, so a ``";"`` that is *data* -- a pane title, a shell - fragment passed to ``send-keys`` -- must not be mistaken for one. Marking - the boundary with its own type keeps the distinction in the value rather - than in a parsing convention, so an engine that chains commands can find - the real boundaries and every other engine can ignore them. + Tmux's direct argv parser treats any unescaped trailing ``;`` as command + structure, including a suffix on a larger value. A literal suffix -- a pane + title or shell fragment passed to ``send-keys`` -- must not become one. + Marking a real boundary with its own type lets direct engines escape + ordinary values while rendering intentional command groups structurally. Examples -------- @@ -127,7 +128,7 @@ class CommandRequest: """A tmux command, ready for an engine to execute. Carries the subcommand and its arguments only. Connection flags - (``-L``/``-S``/``-f``/``-2``/``-8``) belong to the engine's + (``-L``/``-S``/``-f``/``-2``) belong to the engine's :class:`~libtmux.engines.connection.ServerConnection`, so every engine sees the same request no matter which tmux server it targets. @@ -338,13 +339,33 @@ def command_line(self, request: CommandRequest) -> tuple[str, ...]: @t.runtime_checkable -class SupportsConnection(t.Protocol): - """An engine that dispatches over a named tmux server and can be rebound. +class HasConnection(t.Protocol): + """An engine whose tmux connection can be inspected without changing it. + + Persistent engines implement this read-only capability even though their + live transport cannot be cloned safely. In-memory engines with no tmux + connection omit it. + + Examples + -------- + >>> from libtmux.engines import HasConnection, SubprocessEngine + >>> isinstance(SubprocessEngine(), HasConnection) + True + """ + + @property + def connection(self) -> ServerConnection: + """Return the tmux binary and global flags this engine uses.""" + ... + + +@t.runtime_checkable +class SupportsConnection(HasConnection, t.Protocol): + """An inspectable engine that can safely return a rebound equivalent. - Optional capability. :attr:`Server.engine ` reads it - so an injected engine that names no server of its own adopts the server's - connection instead of silently reaching the ambient tmux server. In-memory - engines have no connection and simply do not implement it. + Optional capability. Stateless subprocess engines implement it. Persistent + transports expose :class:`HasConnection` only, because cloning one could + duplicate or abandon live connection state. Examples -------- @@ -363,12 +384,7 @@ class SupportsConnection(t.Protocol): False """ - @property - def connection(self) -> t.Any: - """Return the tmux binary and flags this engine dispatches over.""" - ... - - def with_connection(self, connection: t.Any) -> TmuxEngine: + def with_connection(self, connection: ServerConnection) -> Self: """Return an equivalent engine bound to *connection*.""" ... diff --git a/src/libtmux/engines/connection.py b/src/libtmux/engines/connection.py index 318c4619e1..6405753005 100644 --- a/src/libtmux/engines/connection.py +++ b/src/libtmux/engines/connection.py @@ -1,7 +1,7 @@ """The connection an engine talks to: which tmux binary, which tmux server. Every engine needs the same two things before it can dispatch anything: a tmux -*binary* to exec, and the *connection flags* (``-L``/``-S``/``-f``/``-2``/``-8``) +*binary* to exec, and the *connection flags* (``-L``/``-S``/``-f``/``-2``) that point at one particular tmux server. :class:`ServerConnection` is that pair as one frozen value, and it is the only place in libtmux where either is computed -- :meth:`libtmux.Server.cmd`, :meth:`libtmux.Server.raise_if_dead` and @@ -26,6 +26,238 @@ from collections.abc import Sequence +_GLOBAL_OPTIONS_WITH_VALUE = frozenset({"c", "f", "L", "S", "T"}) +_GLOBAL_OPTIONS_WITHOUT_VALUE = frozenset( + {"2", "C", "D", "d", "h", "l", "N", "q", "u", "U", "v", "V"}, +) + + +@dataclass(frozen=True) +class _ConnectionSettings: + """Effective tmux globals relevant to Server connection constraints. + + Attributes + ---------- + socket_name : str or None + Final ``-L`` socket name, before ``-S`` precedence is applied. + socket_path : str or None + Final ``-S`` socket path. Any value overrides ``socket_name``. + config_files : tuple[str, ...] + Every cumulative ``-f`` value in command-line order. + colors : int or None + Requested client color mode, if declared. + """ + + socket_name: str | None = None + socket_path: str | None = None + config_files: tuple[str, ...] = () + colors: int | None = None + + @property + def socket(self) -> tuple[str, str] | None: + """Return tmux's effective socket selector. + + ``-S`` wins regardless of where ``-L`` appears, matching tmux's global + option semantics. + + Examples + -------- + >>> _ConnectionSettings(socket_name="ignored", socket_path="/tmp/s").socket + ('path', '/tmp/s') + >>> _ConnectionSettings(socket_name="work").socket + ('name', 'work') + """ + if self.socket_path is not None: + return ("path", self.socket_path) + if self.socket_name is not None: + return ("name", self.socket_name) + return None + + +@dataclass(frozen=True) +class _ConnectionResolution: + """A constraint overlay plus the fields that could not be reconciled. + + Attributes + ---------- + connection : ServerConnection + Rebound target after adding non-conflicting missing requirements. + conflicts : tuple[str, ...] + Explicit settings whose engine and Server values disagree. + missing : tuple[str, ...] + Server settings the engine must add through safe rebinding. + """ + + connection: ServerConnection + conflicts: tuple[str, ...] = () + missing: tuple[str, ...] = () + + +def _connection_settings(args: Sequence[str]) -> _ConnectionSettings: + """Parse effective connection globals from tmux's short-option argv. + + Attached and separate values normalize to the same settings, repeated + values use the last occurrence, and ``-S`` makes every ``-L`` ineffective. + + Parameters + ---------- + args : Sequence[str] + Tmux client-global arguments. + + Returns + ------- + _ConnectionSettings + Effective settings used for constraint comparison. + + Examples + -------- + >>> _connection_settings(("-2q", "-f", "/tmp/c", "-Lwork")) + _ConnectionSettings(socket_name='work', socket_path=None, + config_files=('/tmp/c',), colors=256) + >>> _connection_settings(("-S/tmp/s", "-Lignored")).socket + ('path', '/tmp/s') + """ + socket_name: str | None = None + socket_path: str | None = None + config_files: list[str] = [] + colors: int | None = None + index = 0 + + while index < len(args): + token = args[index] + if token == "--": + break + if not token.startswith("-") or token == "-": + break + + cluster = token[1:] + option_index = 0 + while option_index < len(cluster): + option = cluster[option_index] + # tmux removed ``-8`` before libtmux's minimum supported version. + # Recognize it only so an injected engine cannot hide a conflicting + # legacy color constraint; ServerConnection never emits it. + if option == "8": + colors = 88 + option_index += 1 + continue + if option in _GLOBAL_OPTIONS_WITH_VALUE: + attached = cluster[option_index + 1 :] + value: str | None + if attached: + value = attached + elif index + 1 < len(args): + index += 1 + value = args[index] + else: + value = None + + if value is not None: + if option == "f": + config_files.append(value) + elif option == "L": + socket_name = value + elif option == "S": + socket_path = value + break + + if option not in _GLOBAL_OPTIONS_WITHOUT_VALUE: + break + if option == "2": + colors = 256 + elif option == "8": + colors = 88 + option_index += 1 + index += 1 + + return _ConnectionSettings( + socket_name=socket_name, + socket_path=socket_path, + config_files=tuple(config_files), + colors=colors, + ) + + +def _merge_connection_constraints( + connection: ServerConnection, + constraints: ServerConnection, +) -> _ConnectionResolution: + """Overlay explicit Server constraints onto an engine connection. + + Existing engine settings survive when the Server is silent. Missing + settings are appended in canonical tmux order. Contradictory explicit + settings are reported instead of choosing an authority silently. + + Parameters + ---------- + connection : ServerConnection + The injected engine's current connection. + constraints : ServerConnection + Explicit values declared by the Server. + + Returns + ------- + _ConnectionResolution + Target connection, conflicting fields, and fields requiring a rebind. + + Examples + -------- + >>> current = ServerConnection.of(args=("-q", "-Lwork")) + >>> required = ServerConnection.of(args=("-2", "-f/tmp/c")) + >>> _merge_connection_constraints(current, required).connection.args + ('-q', '-Lwork', '-2', '-f/tmp/c') + >>> _merge_connection_constraints( + ... ServerConnection.of(args=("-Lone",)), + ... ServerConnection.of(args=("-Ltwo",)), + ... ).conflicts + ('socket',) + """ + current = _connection_settings(connection.args) + required = _connection_settings(constraints.args) + conflicts: list[str] = [] + missing: list[str] = [] + additions: list[str] = [] + + tmux_bin = connection.tmux_bin + if constraints.tmux_bin is not None: + if tmux_bin is None: + tmux_bin = constraints.tmux_bin + missing.append("binary") + elif tmux_bin != constraints.tmux_bin: + conflicts.append("binary") + + if required.colors is not None: + if current.colors is None: + additions.append("-2" if required.colors == 256 else "-8") + missing.append("color") + elif current.colors != required.colors: + conflicts.append("color") + + if required.config_files: + if not current.config_files: + additions.extend(f"-f{path}" for path in required.config_files) + missing.append("config") + elif current.config_files != required.config_files: + conflicts.append("config") + + if required.socket is not None: + if current.socket is None: + kind, value = required.socket + additions.append(f"-{'L' if kind == 'name' else 'S'}{value}") + missing.append("socket") + elif current.socket != required.socket: + conflicts.append("socket") + + return _ConnectionResolution( + connection=ServerConnection.of( + tmux_bin=tmux_bin, + args=(*connection.args, *additions), + ), + conflicts=tuple(conflicts), + missing=tuple(missing), + ) + + class _BinaryResolver: """Memoized tmux-binary resolution and ``tmux -V`` probe. @@ -198,7 +430,7 @@ def from_server(cls, server: t.Any) -> ServerConnection: Raises ------ :exc:`~libtmux.exc.UnknownColorOption` - ``colors`` is truthy but is neither ``256`` nor ``88``. + ``colors`` is truthy but is not ``256``. Examples -------- @@ -213,7 +445,7 @@ def from_server(cls, server: t.Any) -> ServerConnection: ... ServerConnection.from_server(types.SimpleNamespace(colors=16)) ... except exc.UnknownColorOption as e: ... print(e) - Server.colors must equal 88 or 256 + Server.colors must equal 256 """ args: list[str] = [] @@ -221,8 +453,6 @@ def from_server(cls, server: t.Any) -> ServerConnection: if colors: if colors == 256: args.append("-2") - elif colors == 88: - args.append("-8") else: raise exc.UnknownColorOption @@ -261,18 +491,11 @@ def is_unconfigured(self) -> bool: @property def names_server(self) -> bool: - """Whether this connection carries connection flags of its own. + """Whether this connection effectively selects a tmux server. - :attr:`Server.engine ` reads this on the - *engine's* side of adoption: an engine that already carries flags knows - which tmux server it talks to and is left alone, while one that carries - none is bound to the server's flags so it cannot silently dispatch to - the ambient server. - - :attr:`tmux_bin` deliberately does not count. It selects which tmux - *program* to exec, which says nothing about which server that program - connects to -- a custom binary with no ``-L``/``-S`` reaches the same - ambient server as the stock one. + Only effective ``-L`` and ``-S`` values count. ``-S`` overrides ``-L`` + exactly as tmux documents. The binary, color, configuration and quiet + globals affect execution but do not select a socket. Returns ------- @@ -289,8 +512,10 @@ def names_server(self) -> bool: >>> ServerConnection.of(tmux_bin="/usr/bin/tmux").names_server False + >>> ServerConnection.of(args=("-2", "-f/dev/null", "-q")).names_server + False """ - return bool(self.args) + return _connection_settings(self.args).socket is not None def resolve_bin(self) -> str: """Return the tmux binary path (memoized). diff --git a/src/libtmux/engines/subprocess.py b/src/libtmux/engines/subprocess.py index 886dded420..374e5f4095 100644 --- a/src/libtmux/engines/subprocess.py +++ b/src/libtmux/engines/subprocess.py @@ -14,6 +14,7 @@ import typing as t from libtmux import exc +from libtmux._internal.tmux_argv import encode_direct_argv from libtmux.engines.base import CommandResult from libtmux.engines.connection import ServerConnection @@ -211,7 +212,10 @@ def command_line(self, request: CommandRequest) -> tuple[str, ...]: ... ) ('tmux', '-Lwork', 'send-keys', 'echo hi') """ - return self._conn.argv(*request.args, tmux_bin=request.tmux_bin) + return self._conn.argv( + *encode_direct_argv(request.args), + tmux_bin=request.tmux_bin, + ) def run(self, request: CommandRequest) -> CommandResult: """Execute one tmux command via :mod:`subprocess` and return its result. diff --git a/src/libtmux/exc.py b/src/libtmux/exc.py index 0dd2326cd0..76bf46a65c 100644 --- a/src/libtmux/exc.py +++ b/src/libtmux/exc.py @@ -10,6 +10,7 @@ import typing as t if t.TYPE_CHECKING: + from libtmux.engines.connection import ServerConnection from libtmux.neo import ListExtraArgs @@ -160,6 +161,58 @@ def __init__(self, engine: object, method: str, *args: object) -> None: super().__init__(msg, *args) +class EngineConfigurationMismatch(ValueError): + """An inspectable engine contradicts explicit Server configuration. + + Raised when dispatch cannot honor both the injected engine's declared + connection and the values explicitly set on :class:`libtmux.Server`. It is + also raised when satisfying missing Server constraints would require + cloning a pinned transport, or when a claimed safe rebind reports the wrong + connection. + + Parameters + ---------- + engine : object + Injected engine whose connection cannot satisfy the Server. + reason : str + Concrete conflicting or unsafe configuration detail. + server_connection : ServerConnection, optional + Connection constraints derived from the Server at dispatch time. + engine_connection : object, optional + Connection reported by the injected engine. + *args : object + Forwarded to :class:`ValueError`. + + Examples + -------- + >>> from libtmux import exc + >>> print(exc.EngineConfigurationMismatch(object(), "socket conflict")) + object configuration does not satisfy Server: socket conflict + + It is a configuration error, not a tmux command failure: + + >>> issubclass(exc.EngineConfigurationMismatch, exc.LibTmuxException) + False + """ + + def __init__( + self, + engine: object, + reason: str, + *args: object, + server_connection: ServerConnection | None = None, + engine_connection: object | None = None, + ) -> None: + self.engine = engine + self.reason = reason + self.server_connection = server_connection + self.engine_connection = engine_connection + super().__init__( + f"{type(engine).__name__} configuration does not satisfy Server: {reason}", + *args, + ) + + class NotInsideTmux(LibTmuxException): """Raised when the process is not running inside a tmux pane. @@ -396,7 +449,7 @@ class UnknownColorOption(UnknownOption): """Unknown color option.""" def __init__(self, *args: object) -> None: - super().__init__("Server.colors must equal 88 or 256") + super().__init__("Server.colors must equal 256") class InvalidOption(OptionError): diff --git a/src/libtmux/experimental/engines/__init__.py b/src/libtmux/experimental/engines/__init__.py index 6b9d6b7c9e..10228af22b 100644 --- a/src/libtmux/experimental/engines/__init__.py +++ b/src/libtmux/experimental/engines/__init__.py @@ -21,6 +21,9 @@ CommandResult, EngineKind, EngineSpec, + HasConnection, + SupportsAsyncTmuxVersion, + SupportsConnection, SupportsTmuxVersion, TmuxEngine, ) @@ -52,10 +55,13 @@ "ControlNotification", "EngineKind", "EngineSpec", + "HasConnection", "ImsgEngine", "MockEngine", "ServerConnection", "SubprocessEngine", + "SupportsAsyncTmuxVersion", + "SupportsConnection", "SupportsTmuxVersion", "TmuxEngine", "available_engines", diff --git a/src/libtmux/experimental/engines/async_control_mode.py b/src/libtmux/experimental/engines/async_control_mode.py index e09cb47bfc..855b18ffda 100644 --- a/src/libtmux/experimental/engines/async_control_mode.py +++ b/src/libtmux/experimental/engines/async_control_mode.py @@ -12,10 +12,10 @@ the I/O layer differs from the sync engine (``await stdout.read`` instead of ``selectors``). - Command correlation is a FIFO of futures resolved in block-arrival order. A - block that arrives with *no* pending command is **unsolicited** (a hook- - triggered command, or the startup ACK) and is skipped, so correlation never - desyncs. The startup ACK is consumed synchronously in :meth:`_spawn` before - the reader runs, closing the startup race. + block with unsolicited flags (a hook-triggered command or startup ACK) is + skipped. A solicited block with no pending command is a protocol error rather + than a result to guess at. The startup ACK is consumed synchronously in + :meth:`_spawn` before the reader runs, closing the startup race. - A supervisor owns the process lifecycle. :meth:`start` launches it once; it attaches to an exact existing session without updating its environment, replays the desired subscriptions, and runs the reader inline (one reader at @@ -32,7 +32,7 @@ - A reader failure or EOF marks the engine *dead* and fails every pending command, rather than hanging; the supervisor then reconnects. - Notifications go to a bounded queue; on overflow the oldest is dropped and - counted (backpressure), mirroring control mode's own ``%pause`` philosophy. + counted. Producers are not slowed. """ from __future__ import annotations @@ -46,7 +46,11 @@ from libtmux import exc from libtmux.experimental.engines.asyncio import AsyncSubprocessEngine -from libtmux.experimental.engines.base import CommandRequest, render_control_line +from libtmux.experimental.engines.base import ( + CommandRequest, + render_control_line, + unescape_control_output, +) from libtmux.experimental.engines.connection import ServerConnection from libtmux.experimental.engines.control_mode import ( BlockSequenceMonitor, @@ -71,6 +75,9 @@ _DEFAULT_TIMEOUT = 30.0 _STARTUP_TIMEOUT = 5.0 _STOP_TIMEOUT = 2.0 +_STDERR_TAIL_LINES = 20 +_STDERR_TAIL_BYTES = 16 * 1024 +_STDERR_LOG_BYTES = 4 * 1024 # A connection must survive at least this long to count as healthy and reset the # reconnect backoff; a shorter-lived one is treated as a failed attempt so a # persistently flapping proc escalates instead of fork-storming. @@ -90,12 +97,20 @@ class ControlNotification: args : tuple[str, ...] Whitespace-separated notification arguments. raw : str - Decoded control-mode line before tokenization. + Control-mode line before tokenization and pane-output unescaping. + raw_bytes : bytes + Exact control-mode wire line for diagnostics that require byte fidelity. + pane_id : str or None + Pane identifier for an output notification, otherwise ``None``. + payload : bytes or None + Octal-unescaped pane bytes for ``%output`` and ``%extended-output``; + ``None`` for every other notification. Examples -------- - >>> ControlNotification.parse(b"%window-add @3") - ControlNotification(kind='window-add', args=('@3',), raw='%window-add @3') + >>> notification = ControlNotification.parse(b"%window-add @3") + >>> notification.kind, notification.args, notification.payload + ('window-add', ('@3',), None) >>> ControlNotification.parse(b"%output %1 hello world").kind 'output' """ @@ -103,15 +118,43 @@ class ControlNotification: kind: str args: tuple[str, ...] raw: str + raw_bytes: bytes = b"" + pane_id: str | None = None + payload: bytes | None = None @classmethod def parse(cls, line: bytes) -> ControlNotification: - """Parse a raw ``%``-notification line.""" + r"""Parse one notification and decode pane data at the wire boundary. + + Tmux encodes every non-printable pane byte and backslash as exactly + three octal digits. The human-readable :attr:`raw` line remains encoded + for diagnostics while :attr:`payload` carries the original pane bytes. + """ text = line.decode(errors="replace") body = text.removeprefix("%") parts = body.split(" ") kind = parts[0] if parts else "" - return cls(kind=kind, args=tuple(parts[1:]), raw=text) + pane_id: str | None = None + payload: bytes | None = None + if line.startswith(b"%output "): + output_parts = line.split(b" ", 2) + if len(output_parts) == 3: + pane_id = output_parts[1].decode(errors="replace") + payload = unescape_control_output(output_parts[2]) + elif line.startswith(b"%extended-output "): + metadata, separator, payload_wire = line.partition(b" : ") + fields = metadata.split(b" ") + if separator and len(fields) >= 3: + pane_id = fields[1].decode(errors="replace") + payload = unescape_control_output(payload_wire) + return cls( + kind=kind, + args=tuple(parts[1:]), + raw=text, + raw_bytes=line, + pane_id=pane_id, + payload=payload, + ) @dataclass(slots=True) @@ -161,7 +204,7 @@ def _force_put(queue: asyncio.Queue[t.Any], item: t.Any) -> None: Like :func:`_offer` but drop-count-free: used to land the stream-end sentinel even on a queue already at ``maxsize``, so a slow consumer that hit - backpressure still gets closed instead of hanging on ``queue.get()``. Pulled + overflow still gets closed instead of hanging on ``queue.get()``. Pulled out of the broadcast loop so the ``try``/``except`` stays out of it. """ try: @@ -198,7 +241,7 @@ class AsyncControlModeEngine: timeout : float Seconds to await a command's result before failing it. event_queue_size : int - Bounded size of the notification queue (backpressure). + Bounded notification count; overflow drops the oldest item. Notes ----- @@ -226,6 +269,9 @@ def __init__( self._subscribers: set[asyncio.Queue[t.Any]] = set() self._dropped_notifications = 0 self._proc: asyncio.subprocess.Process | None = None + self._stderr_tail = bytearray() + self._stderr_proc: asyncio.subprocess.Process | None = None + self._stderr_task: asyncio.Task[None] | None = None self._start_lock = asyncio.Lock() self._write_lock = asyncio.Lock() self._started = False @@ -265,12 +311,22 @@ def tmux_version(self) -> str | None: """Report the connected server's tmux version (``tmux -V``), memoized. Implements - :class:`~libtmux.experimental.engines.base.SupportsTmuxVersion` so + :class:`~libtmux.engines.base.SupportsTmuxVersion` so version-gated operations render correctly over control mode; in-memory engines omit it and resolution assumes latest. """ return self._conn.tmux_version() + async def atmux_version(self) -> str | None: + """Probe the connected tmux version without blocking the event loop. + + Examples + -------- + >>> asyncio.run(AsyncControlModeEngine().atmux_version()) is not None + True + """ + return await self._bootstrap.atmux_version() + def add_subscription(self, spec: str) -> None: """Record a desired ``refresh-client -B`` subscription (idempotent). @@ -380,7 +436,7 @@ async def _spawn(self) -> None: # exited, so this is a no-op there. old = self._proc if old is not None: - await self._stop_process(old) + await self._stop_connection(old) target = self._next_attach_target self._next_attach_target = None if target is None: @@ -398,13 +454,18 @@ async def _spawn(self) -> None: "-t", target, ) - try: - proc = await asyncio.create_subprocess_exec( + creation = asyncio.create_task( + asyncio.create_subprocess_exec( *cmd, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, - ) + ), + name="libtmux-async-control-process-start", + ) + cancelled = await self._wait_for_owned_task(creation) + try: + proc = creation.result() except FileNotFoundError: raise exc.TmuxCommandNotFound from None self._proc = proc @@ -413,11 +474,19 @@ async def _spawn(self) -> None: # run_batch must hit the dead guard instead of writing a reply that # _consume_startup would drain and discard. try: - await self._consume_startup() - except BaseException: - await self._stop_process(proc) - self._proc = None + self._start_stderr_reader(proc) + if not cancelled: + await self._consume_startup() + except BaseException as error: + await self._stop_connection(proc) + if isinstance(error, asyncio.CancelledError): + raise + if isinstance(error, ControlModeError): + raise self._died(str(error)) from error raise + if cancelled: + await self._stop_connection(proc) + raise asyncio.CancelledError self._dead = None async def _consume_startup(self) -> None: @@ -542,7 +611,9 @@ async def run_batch( # so the reader can drain any replies, but make their futures # terminal so close/reconnect cannot publish unobserved errors. for queued in appended: - if not queued.future.done(): + if queued.future.done(): + _swallow_future(queued.future) + else: queued.future.cancel() raise except (BrokenPipeError, OSError) as error: @@ -552,7 +623,9 @@ async def run_batch( for queued in appended: with contextlib.suppress(ValueError): self._pending.remove(queued) - if not queued.future.done(): + if queued.future.done(): + _swallow_future(queued.future) + else: queued.future.cancel() raise cm_error from error @@ -632,20 +705,20 @@ async def aclose(self) -> None: self._aclose_impl(), name="libtmux-async-control-close", ) - cancelled = await self._wait_for_close_cleanup(cleanup) + cancelled = await self._wait_for_owned_task(cleanup) cleanup.result() if cancelled: raise asyncio.CancelledError @staticmethod - async def _wait_for_close_cleanup(cleanup: asyncio.Task[None]) -> bool: + async def _wait_for_owned_task(task: asyncio.Task[t.Any]) -> bool: """Wait through every caller cancellation; report whether one occurred.""" try: - await asyncio.wait((cleanup,)) + await asyncio.wait((task,)) except asyncio.CancelledError: # Recursion gives each accepted ``cancel()`` its own handler without - # propagating cancellation into the independently owned cleanup task. - await AsyncControlModeEngine._wait_for_close_cleanup(cleanup) + # propagating cancellation into the independently owned task. + await AsyncControlModeEngine._wait_for_owned_task(task) return True return False @@ -682,9 +755,7 @@ async def _close_locked(self) -> None: self._fail_pending(ControlModeError("control-mode engine closed")) proc = self._proc if proc is not None: - await self._stop_process(proc) - if self._proc is proc: - self._proc = None + await self._stop_connection(proc) async def __aenter__(self) -> Self: """Start when a safe session exists; otherwise remain lazy to bootstrap.""" @@ -785,17 +856,18 @@ async def _supervisor( proc = self._proc if proc is not None: try: - await self._stop_process(proc) + await self._stop_connection(proc) except Exception as cleanup_error: failure = ControlModeError( f"control-mode supervisor failed: {error}; " f"process cleanup failed: {cleanup_error}" ) - else: - if self._proc is proc: - self._proc = None await self._finish_failed_start(first_attempt, failure) finally: + proc = self._proc + if proc is not None: + with contextlib.suppress(Exception): + await self._stop_connection(proc) if first_attempt is not None and not first_attempt.done(): first_attempt.set_exception( ControlModeError("control-mode engine closed before connecting"), @@ -857,6 +929,107 @@ async def _stop_process(proc: asyncio.subprocess.Process) -> None: msg = "tmux control process did not exit after kill" raise ControlModeError(msg) from None + def _start_stderr_reader(self, proc: asyncio.subprocess.Process) -> None: + """Start the one stderr drain task owned by *proc*'s generation.""" + if self._stderr_task is not None: + msg = "control-mode stderr reader already exists" + raise ControlModeError(msg) + self._stderr_tail.clear() + if getattr(proc, "stderr", None) is None: + return + self._stderr_proc = proc + self._stderr_task = asyncio.create_task( + self._drain_stderr(proc), + name="libtmux-async-control-stderr", + ) + + async def _drain_stderr(self, proc: asyncio.subprocess.Process) -> None: + """Continuously drain *proc* stderr into a bounded diagnostic tail.""" + stderr = proc.stderr + if stderr is None: + return + try: + while True: + chunk = await stderr.read(_READ_CHUNK) + if not chunk: + return + self._append_stderr(chunk) + preview = chunk[-_STDERR_LOG_BYTES:].decode(errors="replace") + logger.debug( + "tmux control-mode stderr", + extra={"tmux_stderr": [preview]}, + ) + except asyncio.CancelledError: + raise + except Exception as error: + self._append_stderr(f"stderr read failed: {error}\n".encode()) + + def _append_stderr(self, chunk: bytes) -> None: + """Append stderr while retaining bounded bytes and logical lines.""" + self._stderr_tail.extend(chunk) + overflow = len(self._stderr_tail) - _STDERR_TAIL_BYTES + if overflow > 0: + del self._stderr_tail[:overflow] + lines = self._stderr_tail.splitlines(keepends=True) + if len(lines) > _STDERR_TAIL_LINES: + self._stderr_tail[:] = b"".join(lines[-_STDERR_TAIL_LINES:]) + + def _stderr_lines(self) -> tuple[str, ...]: + """Return the retained non-empty stderr lines for diagnostics.""" + return tuple( + line.decode(errors="replace") + for line in self._stderr_tail.splitlines() + if line.strip() + ) + + async def _finish_stderr_reader( + self, + proc: asyncio.subprocess.Process, + ) -> None: + """Join and forget *proc*'s stderr task without leaking on cancellation.""" + if self._stderr_proc is not proc: + return + task = self._stderr_task + try: + if task is not None: + waiter = asyncio.gather(task, return_exceptions=True) + try: + await asyncio.wait_for(waiter, timeout=_STOP_TIMEOUT) + except asyncio.TimeoutError: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + except asyncio.CancelledError: + if task is not None and not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + raise + finally: + if self._stderr_proc is proc: + self._stderr_proc = None + self._stderr_task = None + + async def _stop_connection(self, proc: asyncio.subprocess.Process) -> None: + """Finish connection cleanup even when this caller is cancelled.""" + cleanup = asyncio.create_task( + self._stop_connection_impl(proc), + name="libtmux-async-control-process-close", + ) + cancelled = await self._wait_for_owned_task(cleanup) + cleanup.result() + if cancelled: + raise asyncio.CancelledError + + async def _stop_connection_impl(self, proc: asyncio.subprocess.Process) -> None: + """Terminate and reap *proc*, then join its stderr drain task.""" + try: + await self._stop_process(proc) + finally: + try: + await self._finish_stderr_reader(proc) + finally: + if self._proc is proc: + self._proc = None + async def _replay_subscriptions(self) -> None: """Re-issue every desired subscription to the freshly connected proc. @@ -961,7 +1134,8 @@ async def _reader(self) -> None: while True: chunk = await stdout.read(_READ_CHUNK) if not chunk: - self._mark_dead(ControlModeError("tmux -C closed stdout")) + await self._stop_connection(proc) + self._mark_dead(self._died("tmux -C closed stdout")) return self._parser.feed(chunk) for block in self._parser.blocks(): @@ -971,35 +1145,28 @@ async def _reader(self) -> None: except asyncio.CancelledError: raise except Exception as error: - self._mark_dead(ControlModeError(f"control-mode reader failed: {error}")) + with contextlib.suppress(Exception): + await self._stop_connection(proc) + self._mark_dead(self._died(f"control-mode reader failed: {error}")) def _dispatch_block(self, block: ControlModeBlock) -> None: """Accumulate a solicited block; resolve the command once it has them all. - A ``;``-folded command emits one block per sub-command; unsolicited blocks + A semicolon command group emits one block per member; unsolicited blocks (hook-triggered commands, the startup ACK) carry flags 0 and are skipped, so FIFO correlation never desyncs. """ if block.flags != 1: return # unsolicited (hook-triggered command or startup ACK): skip if not self._pending: - # A solicited reply with no command waiting: its command's future was - # already resolved, cancelled, or failed. FIFO is now one block off. - logger.warning( - "control-mode dropped solicited block #%s with no pending command", - block.number, - extra={ - "tmux_stdout": [ - line.decode(errors="replace") for line in block.body - ], - "tmux_stdout_len": len(block.body), - }, - ) - return + msg = "control-mode received a solicited result with no pending request" + raise ControlModeError(msg) pending = self._pending[0] - self._sequence.check(block, pending.argv) + if not self._sequence.check(block, pending.argv): + msg = "control-mode command block sequence moved backwards" + raise ControlModeError(msg) pending.blocks.append(block) - if len(pending.blocks) < pending.expected: + if not block.is_error and len(pending.blocks) < pending.expected: return self._pending.popleft() if not pending.future.done(): @@ -1019,7 +1186,7 @@ def _broadcast_stream_end(self) -> None: """Push the stream-end sentinel to every subscriber, then clear them. Uses :func:`_force_put` so the sentinel lands even on a queue already at - ``maxsize`` (a slow consumer that hit backpressure); otherwise the + ``maxsize`` (a slow consumer that hit overflow); otherwise the sentinel would be lost and the consumer would hang forever on ``queue.get()`` -- the exact bug this guards against. """ @@ -1034,6 +1201,11 @@ def _mark_dead(self, error: BaseException) -> None: self._fail_pending(error) self._broadcast_stream_end() + def _died(self, message: str) -> ControlModeError: + """Build a connection error carrying tmux's bounded stderr tail.""" + tail = "; ".join(self._stderr_lines()) + return ControlModeError(f"{message}: {tail}" if tail else message) + def _fail_pending(self, error: BaseException) -> None: """Fail every queued command future with *error*.""" while self._pending: diff --git a/src/libtmux/experimental/engines/asyncio.py b/src/libtmux/experimental/engines/asyncio.py index 00944008e5..f08a783016 100644 --- a/src/libtmux/experimental/engines/asyncio.py +++ b/src/libtmux/experimental/engines/asyncio.py @@ -12,17 +12,137 @@ import asyncio import contextlib +import re +import sys import typing as t from libtmux import exc -from libtmux.experimental.engines.base import CommandResult, encode_direct_argv +from libtmux.experimental.engines.base import ( + CommandRequest, + CommandResult, + encode_direct_argv, +) from libtmux.experimental.engines.connection import ServerConnection if t.TYPE_CHECKING: import pathlib from collections.abc import Sequence - from libtmux.experimental.engines.base import CommandRequest + +_TERMINATE_TIMEOUT = 1.0 +_KILL_TIMEOUT = 1.0 + + +def _version_from_result(result: CommandResult) -> str | None: + """Normalize native async ``tmux -V`` output like the sync probe. + + Examples + -------- + >>> _version_from_result( + ... CommandResult(cmd=("tmux", "-V"), stdout=("tmux 3.4a",)) + ... ) + '3.4' + >>> _version_from_result(CommandResult(cmd=("tmux", "-V"), stderr=("bad",))) + """ + from libtmux.common import TMUX_MAX_VERSION + + if result.stderr: + if ( + sys.platform.startswith("openbsd") + and result.stderr[0] == "tmux: unknown option -- V" + ): + return f"{TMUX_MAX_VERSION}-openbsd" + return None + if not result.stdout: + return None + _prefix, separator, version = result.stdout[0].partition("tmux ") + if not separator or not version: + return None + if version == "master": + return f"{TMUX_MAX_VERSION}-master" + return re.sub(r"[a-z-]", "", version) or None + + +class _AsyncProcessOwner: + """Own one subprocess and every task required to communicate or reap it.""" + + __slots__ = ("_argv",) + + def __init__(self, argv: Sequence[str]) -> None: + self._argv = tuple(argv) + + async def communicate(self) -> tuple[bytes, bytes, int]: + """Spawn, communicate, and own exceptional cleanup for one process. + + Examples + -------- + >>> asyncio.run(_AsyncProcessOwner(("tmux", "-V")).communicate())[2] + 0 + """ + process = await asyncio.create_subprocess_exec( + *self._argv, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + communication = asyncio.create_task( + process.communicate(), + name="libtmux-async-subprocess-communicate", + ) + try: + stdout, stderr = await asyncio.shield(communication) + except BaseException: + + async def finish_communication(timeout: float) -> bool: + """Await pipe draining without cancelling it on timeout.""" + if communication.done(): + await asyncio.gather(communication, return_exceptions=True) + return True + done, _pending = await asyncio.wait( + (communication,), + timeout=timeout, + ) + if not done: + return False + await asyncio.gather(communication, return_exceptions=True) + return True + + async def cleanup_process() -> None: + """Drain pipes while performing bounded TERM-to-KILL cleanup.""" + if process.returncode is None: + with contextlib.suppress(Exception): + process.terminate() + drained = await finish_communication(_TERMINATE_TIMEOUT) + if drained and process.returncode is not None: + return + if process.returncode is None: + with contextlib.suppress(Exception): + process.kill() + drained = await finish_communication(_KILL_TIMEOUT) + if drained: + return + # A killed direct child closes both pipes. This fallback bounds + # pathological inherited-pipe cases where a descendant keeps an + # fd open after the child has exited. + communication.cancel() + await asyncio.gather(communication, return_exceptions=True) + + cleanup_task = asyncio.create_task( + cleanup_process(), + name="libtmux-async-subprocess-cleanup", + ) + while not cleanup_task.done(): + try: + await asyncio.shield(cleanup_task) + except asyncio.CancelledError: # noqa: PERF203 + continue + with contextlib.suppress(asyncio.CancelledError, Exception): + cleanup_task.result() + raise + return ( + stdout, + stderr, + process.returncode if process.returncode is not None else -1, + ) class AsyncSubprocessEngine: @@ -52,6 +172,9 @@ def __init__( server_args: Sequence[str] = (), ) -> None: self._conn = ServerConnection.of(tmux_bin, server_args) + self._async_version: str | None = None + self._async_version_probed = False + self._async_version_lock = asyncio.Lock() @property def connection(self) -> ServerConnection: @@ -76,30 +199,42 @@ def tmux_version(self) -> str | None: """ return self._conn.tmux_version() + async def atmux_version(self) -> str | None: + """Probe ``tmux -V`` natively asynchronously and memoize the result. + + A cancelled probe remains retryable. + + Examples + -------- + >>> asyncio.run(AsyncSubprocessEngine().atmux_version()) is not None + True + """ + if self._async_version_probed: + return self._async_version + async with self._async_version_lock: + if not self._async_version_probed: + try: + result = await self.run(CommandRequest.from_args("-V")) + except exc.LibTmuxException: + version = None + else: + version = _version_from_result(result) + self._async_version = version + self._async_version_probed = True + return self._async_version + async def run(self, request: CommandRequest) -> CommandResult: """Execute one tmux command asynchronously and return its result.""" argv = encode_direct_argv(request.args) cmd = self._conn.argv(*argv, tmux_bin=request.tmux_bin) try: - process = await asyncio.create_subprocess_exec( - *cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) + stdout_bytes, stderr_bytes, returncode = await _AsyncProcessOwner( + cmd, + ).communicate() except FileNotFoundError: raise exc.TmuxCommandNotFound from None - try: - stdout_bytes, stderr_bytes = await process.communicate() - except asyncio.CancelledError: - # The child may have already exited (terminate races the reap); - # suppress so the cancellation propagates, not ProcessLookupError. - with contextlib.suppress(ProcessLookupError): - process.terminate() - await process.wait() - raise - stdout = stdout_bytes.decode(errors="backslashreplace") stderr = stderr_bytes.decode(errors="backslashreplace") @@ -112,7 +247,7 @@ async def run(self, request: CommandRequest) -> CommandResult: cmd=tuple(cmd), stdout=tuple(stdout_lines), stderr=tuple(stderr_lines), - returncode=process.returncode if process.returncode is not None else -1, + returncode=returncode, ) async def run_batch( @@ -122,6 +257,34 @@ async def run_batch( """Execute requests sequentially (preserving tmux command ordering).""" return [await self.run(req) for req in requests] + def with_connection(self, connection: ServerConnection) -> AsyncSubprocessEngine: + """Return an equivalent stateless async engine bound to *connection*. + + Parameters + ---------- + connection : ServerConnection + The connection the returned engine dispatches over. + + Returns + ------- + AsyncSubprocessEngine + A new engine; this engine is unchanged. + + Examples + -------- + >>> from libtmux.experimental.engines import ServerConnection + >>> engine = AsyncSubprocessEngine() + >>> rebound = engine.with_connection( + ... ServerConnection.of(args=("-Lwork",)) + ... ) + >>> rebound.server_args, engine.server_args + (('-Lwork',), ()) + """ + return type(self)( + tmux_bin=connection.tmux_bin, + server_args=connection.args, + ) + @classmethod def for_server(cls, server: t.Any) -> AsyncSubprocessEngine: """Build an async engine bound to a live :class:`libtmux.Server`'s socket.""" diff --git a/src/libtmux/experimental/engines/base.py b/src/libtmux/experimental/engines/base.py index 006d5d62f3..2d6870865a 100644 --- a/src/libtmux/experimental/engines/base.py +++ b/src/libtmux/experimental/engines/base.py @@ -22,10 +22,17 @@ import typing as t from dataclasses import dataclass +from libtmux._internal.tmux_argv import ( + DirectArgv, + encode_direct_argv, + split_direct_argv, +) from libtmux.engines.base import ( CommandRequest, CommandResult, CommandSeparator, + HasConnection, + SupportsConnection, SupportsTmuxVersion, TmuxEngine, is_command_separator, @@ -38,14 +45,6 @@ #: tmux escapes a byte in ``%output`` as a backslash plus three octal digits. _CONTROL_OCTAL = re.compile(rb"\\([0-7]{3})") -# tmux parses these options with getopt before its command argv reaches -# cmd_parse_from_arguments. Only values in the latter have structural -# trailing-semicolon semantics. -_GLOBAL_OPTIONS_WITH_VALUE = frozenset({"c", "f", "L", "S", "T"}) -_GLOBAL_OPTIONS_WITHOUT_VALUE = frozenset( - {"2", "8", "C", "D", "d", "h", "l", "N", "q", "u", "U", "v", "V"}, -) - __all__ = ( "AsyncTmuxEngine", "CommandRequest", @@ -54,6 +53,9 @@ "DirectArgv", "EngineKind", "EngineSpec", + "HasConnection", + "SupportsAsyncTmuxVersion", + "SupportsConnection", "SupportsTmuxVersion", "TmuxEngine", "encode_direct_argv", @@ -64,116 +66,6 @@ ) -class DirectArgv(t.NamedTuple): - """The client-global and command portions of direct tmux argv. - - Attributes - ---------- - global_args : tuple[str, ...] - Leading options consumed by tmux's client-level ``getopt`` parser. - command_argv : tuple[str, ...] - The subcommand and arguments passed to ``cmd_parse_from_arguments``. - """ - - global_args: tuple[str, ...] - command_argv: tuple[str, ...] - - -def _global_option_consumes_next(token: str) -> bool | None: - """Return a global option's separate-value arity, or ``None`` if unknown. - - Examples - -------- - >>> _global_option_consumes_next("-L") - True - >>> _global_option_consumes_next("-Lwork") - False - >>> _global_option_consumes_next("list-sessions") is None - True - """ - if not token.startswith("-") or token in {"-", "--"}: - return None - cluster = token[1:] - if not cluster: - return None - for index, option in enumerate(cluster): - if option in _GLOBAL_OPTIONS_WITH_VALUE: - return index == len(cluster) - 1 - if option not in _GLOBAL_OPTIONS_WITHOUT_VALUE: - return None - return False - - -def split_direct_argv(argv: Sequence[str]) -> DirectArgv: - """Split raw tmux argv at the client-global/command parser boundary. - - The split follows tmux's leading short-option ``getopt`` grammar, including - attached values and ``--``. Global values remain byte-for-byte data because - tmux removes them before parsing command separators. - - Examples - -------- - >>> split_direct_argv(("-L", "socket;", "display-message", "text;")) - DirectArgv(global_args=('-L', 'socket;'), command_argv=('display-message', 'text;')) - >>> split_direct_argv(("-Lsocket;", "--", "display-message")) - DirectArgv(global_args=('-Lsocket;', '--'), command_argv=('display-message',)) - """ - args = tuple(argv) - if any("\0" in token for token in args): - msg = "tmux command arguments cannot contain NUL" - raise ValueError(msg) - - index = 0 - while index < len(args): - token = args[index] - if token == "--": - index += 1 - break - consumes_next = _global_option_consumes_next(token) - if consumes_next is None: - break - index += 2 if consumes_next and index + 1 < len(args) else 1 - return DirectArgv(global_args=args[:index], command_argv=args[index:]) - - -def _encode_command_argv(argv: Sequence[str]) -> tuple[str, ...]: - r"""Escape literal separators in argv already known to be command-scoped. - - Examples - -------- - >>> _encode_command_argv(("display-message", "literal;")) - ('display-message', 'literal\\;') - """ - encoded: list[str] = [] - for token in argv: - if not is_command_separator(token) and token.endswith(";"): - token = f"{token[:-1]}\\;" - encoded.append(str(token)) - return tuple(encoded) - - -def encode_direct_argv(argv: Sequence[str]) -> tuple[str, ...]: - r"""Encode literal arguments for tmux's direct argv parser. - - Tmux first removes client-global options, then routes only the remaining - command argv through ``cmd_parse_from_arguments``, where a final ``;`` is - structural. Prefixing that final byte with one backslash preserves it as - data. Global option values remain unchanged, and a - :class:`CommandSeparator` remains structural. - - Examples - -------- - >>> encode_direct_argv(("send-keys", "text;")) - ('send-keys', 'text\\;') - >>> encode_direct_argv(("-L", "socket;", "send-keys", "text;")) - ('-L', 'socket;', 'send-keys', 'text\\;') - >>> encode_direct_argv(("a", CommandSeparator(";"), "b")) - ('a', ';', 'b') - """ - direct = split_direct_argv(argv) - return (*direct.global_args, *_encode_command_argv(direct.command_argv)) - - def _quote_control_token(token: str) -> str: r"""Quote one literal token for tmux's line-oriented control parser.""" if "\0" in token: @@ -208,7 +100,7 @@ def render_control_line(argv: Sequence[str]) -> str: ) -def unescape_control_output(payload: str) -> bytes: +def unescape_control_output(payload: str | bytes) -> bytes: r"""Decode a control-mode ``%output`` payload back to the bytes the pane wrote. tmux does not forward pane output verbatim: in a ``%output`` notification it @@ -217,8 +109,9 @@ def unescape_control_output(payload: str) -> bytes: this first, or it can never match: an ``ESC`` (``0x1b``) arrives on the wire as the four *characters* ``\``, ``0``, ``3``, ``3``. - Bytes tmux left alone pass through untouched, so feeding this an already-raw - payload is harmless. + Bytes outside an exact three-digit escape pass through untouched. Apply the + decoder once to wire data; it is not idempotent when decoded pane bytes + themselves contain a backslash followed by three octal digits. Examples -------- @@ -237,7 +130,11 @@ def unescape_control_output(payload: str) -> bytes: >>> unescape_control_output(r"caf\303\251").decode() 'café' """ - raw = payload.encode("utf-8", "surrogateescape") + raw = ( + payload.encode("utf-8", "surrogateescape") + if isinstance(payload, str) + else payload + ) return _CONTROL_OCTAL.sub(lambda m: bytes((int(m.group(1), 8),)), raw) @@ -305,6 +202,24 @@ def imsg(cls, *, protocol_version: int | None = None) -> EngineSpec: return cls(kind=EngineKind.IMSG, protocol_version=protocol_version) +@t.runtime_checkable +class SupportsAsyncTmuxVersion(t.Protocol): + """An async engine that can report its tmux version without blocking. + + Examples + -------- + >>> class Versioned: + ... async def atmux_version(self): + ... return "3.4" + >>> isinstance(Versioned(), SupportsAsyncTmuxVersion) + True + """ + + async def atmux_version(self) -> str | None: + """Return the engine's tmux version string, or ``None`` if unknown.""" + ... + + @t.runtime_checkable class AsyncTmuxEngine(t.Protocol): """An asynchronous executor of tmux commands.""" diff --git a/src/libtmux/experimental/engines/connection.py b/src/libtmux/experimental/engines/connection.py index 59d4244676..97e5d49807 100644 --- a/src/libtmux/experimental/engines/connection.py +++ b/src/libtmux/experimental/engines/connection.py @@ -3,7 +3,7 @@ :class:`~libtmux.engines.connection.ServerConnection` graduated to Core as part of the command execution seam, so the experimental engines dispatch over the same connection value the object API does: one tmux binary resolution, one set -of ``-L``/``-S``/``-f``/``-2``/``-8`` flags, one ``tmux -V`` probe. +of ``-L``/``-S``/``-f``/``-2`` flags, one ``tmux -V`` probe. This module is the import path the experimental engines have always used; it carries no definition of its own. diff --git a/src/libtmux/experimental/engines/control_mode.py b/src/libtmux/experimental/engines/control_mode.py index ad740ad807..d4657815bf 100644 --- a/src/libtmux/experimental/engines/control_mode.py +++ b/src/libtmux/experimental/engines/control_mode.py @@ -78,12 +78,15 @@ class ControlModeBlock: Whether tmux closed the block with ``%error``. body : tuple[bytes, ...] Raw output lines between the opening and closing guards. + timestamp : int + Timestamp echoed by both guards for exact block correlation. """ number: int flags: int is_error: bool body: tuple[bytes, ...] + timestamp: int = 0 @dataclasses.dataclass(slots=True) @@ -98,11 +101,14 @@ class _PendingBlock: Flags from the opening control-mode guard. body : list[bytes] Raw output lines collected before the closing guard. + timestamp : int + Timestamp from the opening guard. """ number: int flags: int body: list[bytes] + timestamp: int class ControlModeParser: @@ -159,30 +165,31 @@ def notifications(self) -> list[bytes]: def _handle_line(self, line: bytes) -> None: if self._pending is not None: - if _matches_pending_close(line, self._pending.number): + if _matches_pending_close(line, self._pending): self._close_block(line) return self._pending.body.append(line) return if line.startswith(_BEGIN_PREFIX): self._open_block(line) + elif line.startswith((_END_PREFIX, _ERROR_PREFIX)): + msg = "control-mode received a closing guard without %begin" + raise ControlModeError(msg) elif line.startswith(b"%"): self._notifications.append(line) def _open_block(self, line: bytes) -> None: - number, flags = _parse_guard(line, _BEGIN_PREFIX) - if number is None: - # A %begin whose guard will not parse is dropped: its body lines then - # fall through as notifications and a LATER command's block fills the - # count, silently mis-attributing output. Never observed in practice - # (tmux always writes "%begin