diff --git a/CHANGES b/CHANGES index 7a691b0cc9..6a060ada84 100644 --- a/CHANGES +++ b/CHANGES @@ -45,6 +45,34 @@ $ uvx --from 'libtmux' --prerelease allow python _Notes on the upcoming release will go here._ +### What's new + +#### Read only what a pane wrote since you last looked (#740) + +{meth}`Pane.capture_since() ` returns the rows a +pane has written since a {class}`~libtmux.capture.CaptureCursor`, along with a +fresh cursor to resume from. Watching a pane over time — tailing a build, +following a long-running command — no longer means re-reading the whole screen +each tick and diffing it yourself. + +Cursors are immutable and the call never advances the one it was given, so +replaying a cursor always returns the same rows. They serialize through `str()` +and {meth}`~libtmux.capture.CaptureCursor.from_str` for callers that carry them +across a process or wire boundary. + +Naive screen-diffing goes wrong in ways tmux makes easy to hit, and a cursor +accounts for each: output that scrolled past the visible region between reads, +`clear-history` and `history-limit` trims renumbering the grid under a stored +offset, and a respawned pane reusing its `pane_id` while running a different +program. Where the anchored rows are genuinely gone, the result reports +`lines_missed` and falls back to the visible screen rather than returning a +delta that quietly omits them. Where continuing would mean reading another +process's output, it raises {exc}`~libtmux.exc.PaneLifecycleChanged` or +{exc}`~libtmux.exc.InvalidCaptureCursor` — both under a shared +{exc}`~libtmux.exc.CaptureCursorError` base. + +See {ref}`capture-since` for a walkthrough. + ### Documentation #### Cleaner `from_env` examples (#719) diff --git a/docs/api/index.md b/docs/api/index.md index 23cd9043b1..9236e7a2e7 100644 --- a/docs/api/index.md +++ b/docs/api/index.md @@ -31,7 +31,8 @@ Use {meth}`pane.send_keys() ` and :::{grid-item-card} Capture output from a pane? :link: libtmux.pane :link-type: doc -Use {meth}`pane.capture_pane() `. +Use {meth}`pane.capture_pane() ` for a snapshot, or +{meth}`pane.capture_since() ` for only what is new. ::: :::{grid-item-card} Write tests against tmux? @@ -108,6 +109,12 @@ tmux option get/set. tmux hook management. ::: +:::{grid-item-card} Capture +:link: libtmux.capture +:link-type: doc +Cursors for incremental pane reads. +::: + :::{grid-item-card} Constants :link: libtmux.constants :link-type: doc @@ -176,6 +183,7 @@ Common Neo Options Hooks +Capture Constants Exceptions ``` diff --git a/docs/api/libtmux.capture.md b/docs/api/libtmux.capture.md new file mode 100644 index 0000000000..4c438919c2 --- /dev/null +++ b/docs/api/libtmux.capture.md @@ -0,0 +1,23 @@ +(capture)= + +# Capture + +Incremental pane reading for {meth}`~libtmux.Pane.capture_since`. + +Where {meth}`~libtmux.Pane.capture_pane` returns a snapshot of a pane, +{meth}`~libtmux.Pane.capture_since` returns a *delta* — the rows written since a +{class}`~libtmux.capture.CaptureCursor` — plus a fresh cursor to resume from. + +Cursors are immutable, so a call never advances the cursor it was handed, and +serialize through `str()` / {meth}`~libtmux.capture.CaptureCursor.from_str` for +callers that carry them across a process or wire boundary. + +See {ref}`capture-since` for a worked walkthrough. + +```{eval-rst} +.. automodule:: libtmux.capture + :members: + :private-members: + :show-inheritance: + :member-order: bysource +``` diff --git a/docs/topics/pane_interaction.md b/docs/topics/pane_interaction.md index 6bde824ec6..4505d5645d 100644 --- a/docs/topics/pane_interaction.md +++ b/docs/topics/pane_interaction.md @@ -266,6 +266,89 @@ True flags (`start`, `end`, `escape_sequences`, etc.) — tmux ignores them when `-P` is set. +(capture-since)= + +### Capturing only what is new + +{meth}`~libtmux.Pane.capture_pane` answers "what is on screen right now?". When +you watch a pane over time — tailing a build, following a long-running command — +the question becomes "what changed since I last looked?", and re-capturing the +whole screen every tick makes you answer it yourself. + +{meth}`~libtmux.Pane.capture_since` answers it directly. It returns the rows +written since a {class}`~libtmux.capture.CaptureCursor`, plus a fresh cursor to +resume from: + +```python +>>> from libtmux.test.retry import retry_until + +>>> first = pane.capture_since() +>>> pane.send_keys('echo watching') + +>>> retry_until( +... lambda: any( +... 'watching' in line +... for line in pane.capture_since(first.cursor).lines +... ), +... 2, +... ) +True +``` + +The cursor is immutable and the call never advances it, so replaying one is +always safe. Assign the returned cursor to move forward: + +```python +>>> latest = pane.capture_since() +>>> pane.capture_since(latest.cursor).lines +[] +``` + +#### When output is genuinely gone + +tmux keeps a bounded scrollback. If `clear-history` runs, or output floods past +`history-limit`, the rows a cursor pointed at stop existing. Rather than return +a delta that quietly omits them, `capture_since` falls back to the current +visible screen and sets `lines_missed`: + +```python +>>> from libtmux.test.retry import retry_until + +>>> pane.send_keys('printf "scroll %s\\n" $(seq 1 60)') +>>> retry_until(lambda: pane.capture_since().cursor.history_size > 0, 3) +True + +>>> before_clear = pane.capture_since() +>>> pane.cmd('clear-history') + + +>>> pane.send_keys('echo after') +>>> retry_until( +... lambda: pane.capture_since(before_clear.cursor).lines_missed is True, 3 +... ) +True +``` + +Treat `lines_missed=True` as "some output was lost" — the returned rows are +still real, they are just not the complete delta. + +Two conditions raise instead of degrading, because continuing would mean reading +a different program's output through a cursor that looks valid: +{exc}`~libtmux.exc.PaneLifecycleChanged` when the pane died or was respawned, +and {exc}`~libtmux.exc.InvalidCaptureCursor` when a cursor is replayed against +another pane. Both derive from {exc}`~libtmux.exc.CaptureCursorError`, so one +`except` clause covers every way a cursor stops being usable. + +Cursors serialize for callers that hand them across a process or wire boundary: + +```python +>>> from libtmux.capture import CaptureCursor + +>>> cursor = pane.capture_since().cursor +>>> CaptureCursor.from_str(str(cursor)) == cursor +True +``` + ## Waiting for output tmux runs commands asynchronously: {meth}`~libtmux.Pane.send_keys` returns the diff --git a/src/libtmux/capture.py b/src/libtmux/capture.py new file mode 100644 index 0000000000..6efed18fe3 --- /dev/null +++ b/src/libtmux/capture.py @@ -0,0 +1,1033 @@ +"""Incremental pane capture for :meth:`libtmux.pane.Pane.capture_since`. + +:meth:`~libtmux.pane.Pane.capture_pane` returns a snapshot. Watching a pane +over time needs a *delta*: the rows written since the last look. Computing one +by diffing successive snapshots is wrong in three ways tmux makes easy to hit +-- output can scroll past the visible region between reads, ``history-limit`` +trimming and ``clear-history`` renumber the grid under a stored row offset, and +a respawned pane keeps its ``pane_id`` while running a different process. + +A :class:`CaptureCursor` anchors a position against all three. When the anchor +provably survives, the delta is exact; when it provably does not, the read +degrades to the current visible screen and says so through +:attr:`CaptureSince.lines_missed`. It never returns a silently incomplete +delta. + +This module is deliberately split, at the ``TMUX I/O BOUNDARY`` comment +partway down. Above it everything is pure: it decides what a read *means* +given values, and runs without a tmux server. Below it everything performs +tmux round-trips. Only the second half is execution-model-specific, so an +alternate driver can reuse the first half rather than reimplement the anchor +arithmetic. +""" + +from __future__ import annotations + +import base64 +import binascii +import dataclasses +import hashlib +import json +import logging +import typing as t + +from libtmux import exc +from libtmux.common import raise_if_stderr + +if t.TYPE_CHECKING: + from libtmux.pane import Pane + +logger = logging.getLogger(__name__) + + +#: Serialized-cursor prefix. Versioned so the wire format can change without +#: a decoder silently misreading an older payload as a newer one. +CURSOR_PREFIX = "capture-since-v1:" + +_CURSOR_VERSION = 1 + +#: How many times a read re-samples pane state before giving up on getting an +#: untorn snapshot and reporting :attr:`CaptureSince.lines_missed`. +_STABLE_READ_ATTEMPTS = 3 + + +class CaptureSince(t.NamedTuple): + """Rows written since a cursor, plus the cursor that follows them. + + Attributes + ---------- + lines : list[str] + Captured rows, oldest first. Empty when nothing was written since + the cursor. + cursor : CaptureCursor + A fresh cursor anchored after ``lines``. Pass it to the next + :meth:`~libtmux.pane.Pane.capture_since` call. + lines_missed : bool + ``True`` when the previous anchor could not be proven to survive, + so ``lines`` is the current visible screen rather than a complete + delta. Rows written between the old anchor and the visible region + are gone. + """ + + lines: list[str] + cursor: CaptureCursor + lines_missed: bool + + +@dataclasses.dataclass(frozen=True) +class CaptureCursor: + """An immutable anchor into one pane's grid. + + Carries the pane it belongs to, so replaying it against a different + pane is caught rather than silently reading the wrong process. + + Frozen because :meth:`~libtmux.pane.Pane.capture_since` returns a new + cursor rather than advancing the one it was given -- the same cursor + can be replayed and yields the same delta. + + Attributes + ---------- + pane_id : str + Pane the cursor was taken from (e.g. ``'%1'``). + pane_pid : str + PID of the pane's process when the cursor was taken. A change + means the pane was respawned and the anchor describes another + process's output. + history_size : int + Rows in the pane's scrollback when the cursor was taken. + pane_height : int + Visible rows when the cursor was taken. Distinguishes a resize + from a real history trim. + anchor_abs : int + Absolute grid row of the anchor, as ``history_size + cursor_y``. + anchor_hash : str | None + Content hash of the anchor row, or ``None`` when the cursor sat + below the visible region. + below_hashes : tuple[str, ...] + Content hashes of the rows beneath the anchor, used to re-locate + the anchor when tmux may have renumbered the grid. + + Examples + -------- + >>> cursor = pane.capture_since().cursor + >>> cursor.pane_id == pane.pane_id + True + >>> CaptureCursor.from_str(str(cursor)) == cursor + True + """ + + pane_id: str + pane_pid: str + history_size: int + pane_height: int + anchor_abs: int + anchor_hash: str | None + below_hashes: tuple[str, ...] + + def __str__(self) -> str: + """Serialize to an opaque, round-trippable string. + + Examples + -------- + >>> str(pane.capture_since().cursor).startswith('capture-since-v1:') + True + """ + payload: dict[str, t.Any] = { + "version": _CURSOR_VERSION, + "pane_id": self.pane_id, + "pane_pid": self.pane_pid, + "history_size": self.history_size, + "pane_height": self.pane_height, + "anchor_abs": self.anchor_abs, + "anchor_hash": self.anchor_hash, + "below_hashes": list(self.below_hashes), + } + raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode() + encoded = base64.urlsafe_b64encode(raw).decode().rstrip("=") + return f"{CURSOR_PREFIX}{encoded}" + + @classmethod + def from_str(cls, value: str) -> CaptureCursor: + """Decode a cursor serialized by :meth:`__str__`. + + Parameters + ---------- + value : str + A string produced by ``str(cursor)``. + + Returns + ------- + CaptureCursor + + Raises + ------ + libtmux.exc.InvalidCaptureCursor + If the string is not a cursor, is a version this build cannot + read, or carries a malformed payload. + + Examples + -------- + >>> CaptureCursor.from_str(str(pane.capture_since().cursor)) + CaptureCursor(pane_id='%...', ...) + + >>> CaptureCursor.from_str('nope') + Traceback (most recent call last): + libtmux.exc.InvalidCaptureCursor: invalid capture_since cursor: \ +unsupported cursor format + """ + if not value.startswith(CURSOR_PREFIX): + _raise_invalid_cursor("unsupported cursor format") + encoded = value.removeprefix(CURSOR_PREFIX) + padding = "=" * (-len(encoded) % 4) + try: + raw = base64.urlsafe_b64decode(f"{encoded}{padding}") + payload: t.Any = json.loads(raw) + except (binascii.Error, json.JSONDecodeError, UnicodeDecodeError) as err: + msg = "invalid capture_since cursor: could not decode payload" + raise exc.InvalidCaptureCursor(msg) from err + + if not isinstance(payload, dict): + _raise_invalid_cursor("payload is not an object") + if payload.get("version") != _CURSOR_VERSION: + _raise_invalid_cursor("unsupported cursor version") + + anchor_hash = payload.get("anchor_hash") + if anchor_hash is not None and not isinstance(anchor_hash, str): + _raise_invalid_cursor("missing or invalid anchor_hash") + below_hashes = payload.get("below_hashes") + if not isinstance(below_hashes, list) or not all( + isinstance(item, str) for item in below_hashes + ): + _raise_invalid_cursor("missing or invalid below_hashes") + + return cls( + pane_id=_cursor_str(payload, "pane_id"), + pane_pid=_cursor_str(payload, "pane_pid"), + history_size=_cursor_int(payload, "history_size"), + pane_height=_cursor_int(payload, "pane_height"), + anchor_abs=_cursor_int(payload, "anchor_abs"), + anchor_hash=anchor_hash, + below_hashes=tuple(below_hashes), + ) + + +class _PaneState(t.NamedTuple): + """Per-read snapshot of a pane's grid and lifecycle. + + Read in one ``display-message`` round-trip so a caller does not pay a + subprocess per format field. ``history_size + cursor_y`` is the + absolute grid row of the cursor. + + Attributes + ---------- + history_size : int + Rows currently in scrollback. + cursor_y : int + Cursor row within the visible region. + pane_height : int + Visible rows. + pane_pid : str + PID of the pane's process. + pane_dead : bool + Whether the pane's process has exited (``remain-on-exit``). + alternate_on : bool + Whether the pane is on the alternate screen. Reported for + completeness because it rides along free in the same round-trip; + never acted on. A pane on the alternate screen has handed the + whole grid to a full-screen program that repaints it, so "rows + below the anchor" stops carrying delta meaning -- but + ``capture-pane -S`` still returns real main-screen scrollback, so + the anchor itself stays arithmetically valid. + """ + + history_size: int + cursor_y: int + pane_height: int + pane_pid: str + pane_dead: bool + alternate_on: bool = False + + +#: tmux format read by :func:`_read_pane_state`. A fixed literal -- no +#: caller-supplied text is ever interpolated into a tmux format string, +#: because tmux's parser treats ``#`` and ``}`` structurally and either one +#: silently corrupts the surrounding fields. +PANE_STATE_FORMAT = ( + "#{history_size}|#{cursor_y}|#{pane_height}|#{pane_pid}|#{pane_dead}" + "|#{alternate_on}" +) + +#: ``history-limit`` read, split out because it never changes between reads. +HISTORY_LIMIT_FORMAT = "#{history_limit}" + + +def _raise_invalid_cursor(reason: str) -> t.NoReturn: + """Raise :exc:`~libtmux.exc.InvalidCaptureCursor` with a uniform message. + + Examples + -------- + >>> _raise_invalid_cursor('unsupported cursor version') + Traceback (most recent call last): + libtmux.exc.InvalidCaptureCursor: invalid capture_since cursor: \ +unsupported cursor version + """ + msg = f"invalid capture_since cursor: {reason}" + raise exc.InvalidCaptureCursor(msg) + + +def _cursor_str(payload: t.Mapping[str, t.Any], key: str) -> str: + """Read a required non-empty string from a cursor payload. + + Examples + -------- + >>> _cursor_str({'pane_id': '%1'}, 'pane_id') + '%1' + + >>> _cursor_str({'pane_id': ''}, 'pane_id') + Traceback (most recent call last): + libtmux.exc.InvalidCaptureCursor: invalid capture_since cursor: \ +missing or invalid pane_id + """ + value = payload.get(key) + if not isinstance(value, str) or not value: + _raise_invalid_cursor(f"missing or invalid {key}") + return value + + +def _cursor_int(payload: t.Mapping[str, t.Any], key: str) -> int: + """Read a required non-negative integer from a cursor payload. + + Rejects :class:`bool`, which is an :class:`int` subclass and would + otherwise decode as ``0`` or ``1``. + + Examples + -------- + >>> _cursor_int({'anchor_abs': 12}, 'anchor_abs') + 12 + + >>> _cursor_int({'anchor_abs': True}, 'anchor_abs') + Traceback (most recent call last): + libtmux.exc.InvalidCaptureCursor: invalid capture_since cursor: \ +missing or invalid anchor_abs + """ + value = payload.get(key) + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + _raise_invalid_cursor(f"missing or invalid {key}") + return value + + +def _line_hash(line: str) -> str: + """Return a stable content hash for a tmux row. + + ``surrogateescape`` because tmux can hand back bytes that are not + valid UTF-8, and a capture must never fail on undecodable output. + + Examples + -------- + >>> _line_hash('') == _line_hash('') + True + >>> _line_hash('a') == _line_hash('b') + False + """ + return hashlib.sha256(line.encode("utf-8", "surrogateescape")).hexdigest() + + +def _parse_pane_state(raw: str) -> _PaneState: + """Parse one :data:`PANE_STATE_FORMAT` line into a :class:`_PaneState`. + + ``maxsplit`` is one below the field count so a ``pane_pid`` or future + field containing ``|`` cannot shift the parse. tmux builds that do not + know ``alternate_on`` emit the literal format text instead of a value, + so anything but ``"1"`` is treated as off -- this read sits on a poll + path and must degrade rather than raise. + + Examples + -------- + >>> _parse_pane_state('100|5|24|4242|0|0') + _PaneState(history_size=100, cursor_y=5, pane_height=24, pane_pid='4242', \ +pane_dead=False, alternate_on=False) + + A build without ``alternate_on`` still parses: + + >>> _parse_pane_state('0|0|24|4242|1').pane_dead + True + >>> _parse_pane_state('0|0|24|4242|1|#{alternate_on}').alternate_on + False + """ + parts = raw.split("|", 5) + history_size, cursor_y, pane_height, pane_pid, pane_dead = parts[:5] + alternate = parts[5] if len(parts) > 5 else "0" + return _PaneState( + history_size=int(history_size), + cursor_y=int(cursor_y), + pane_height=int(pane_height), + pane_pid=pane_pid, + pane_dead=pane_dead == "1", + alternate_on=alternate == "1", + ) + + +def _cursor_anchor_lost(cursor: CaptureCursor, state: _PaneState) -> bool: + """Whether sampled state proves tmux destroyed the cursor's anchor. + + ``anchor_abs`` below ``history_size`` is *not* loss -- the anchor + scrolled into retained scrollback, where ``capture-pane -S`` still + addresses it with a negative start offset. + + The ``pane_height`` comparison separates a resize-grow, which pulls + rows out of history back into the visible region without freeing + anything, from a real trim, where row data is gone. + + That comparison deliberately ignores *how much* history shrank, which + is safe for two reasons rather than one. Arithmetically, ``anchor_abs`` + is never below ``cursor.history_size``, so any shrink larger than + ``pane_height - 1`` pushes the anchor past the grid bottom and trips + the first check regardless of a resize. Within that bounded window, + losing rows a resize cannot account for means tmux trimmed, which only + happens at ``history-limit`` — and :func:`_history_limit_trim_risk` + routes that to a content re-anchor instead of trusting offsets. + + Parameters + ---------- + cursor : CaptureCursor + The anchor being checked. + state : _PaneState + A current snapshot of the same pane. + + Returns + ------- + bool + + Examples + -------- + An unchanged pane keeps its anchor: + + >>> steady = CaptureCursor('%1', '1', 100, 24, 110, None, ()) + >>> _cursor_anchor_lost(steady, _PaneState(100, 10, 24, '1', False)) + False + + Growing the pane explains a shrunken history, so it is not loss -- + the rows moved back into the visible region rather than being freed: + + >>> _cursor_anchor_lost(steady, _PaneState(90, 10, 34, '1', False)) + False + + Each branch below is shown against state where only it fires, since a + real ``clear-history`` trips all three at once and would not show + which one is load-bearing. + + An anchor past the bottom of the grid cannot exist, even with history + intact and unshrunken: + + >>> off_grid = CaptureCursor('%1', '1', 10, 24, 100, None, ()) + >>> _cursor_anchor_lost(off_grid, _PaneState(10, 5, 24, '1', False)) + True + + A history wiped to zero destroys the anchor even when a simultaneous + pane grow leaves it inside the new grid and explains the shrink: + + >>> wiped = CaptureCursor('%1', '1', 5, 24, 10, None, ()) + >>> _cursor_anchor_lost(wiped, _PaneState(0, 0, 30, '1', False)) + True + + A partial trim at constant height destroys rows while the anchor + still addresses a row that exists: + + >>> trimmed = CaptureCursor('%1', '1', 100, 24, 50, None, ()) + >>> _cursor_anchor_lost(trimmed, _PaneState(60, 5, 24, '1', False)) + True + """ + bottom_abs = state.history_size + state.pane_height - 1 + if cursor.anchor_abs > bottom_abs: + return True + # A complete history wipe (``clear-history``) always destroys the anchor + # regardless of pane height -- the grid is reset to zero. + if state.history_size == 0 and cursor.history_size > 0: + return True + return state.history_size < cursor.history_size and ( + state.pane_height <= cursor.pane_height + ) + + +def _history_limit_trim_risk( + cursor: CaptureCursor, + state: _PaneState, + history_limit: int, +) -> bool: + """Whether tmux may have rebased retained-history rows. + + tmux trims scrollback in batches rather than one row at a time, so + ``history_size`` alone cannot prove rows were not renumbered. Anywhere + within one batch of the limit, positional arithmetic is untrustworthy + and the anchor must be re-located by content instead. + + Parameters + ---------- + cursor : CaptureCursor + The anchor being checked. + state : _PaneState + A current snapshot of the same pane. + history_limit : int + The pane's ``history-limit``. + + Returns + ------- + bool + + Examples + -------- + >>> cursor = CaptureCursor('%1', '1', 10, 24, 20, None, ()) + >>> state = _PaneState(10, 5, 24, '1', False) + + Far from the limit, offsets can be trusted: + + >>> _history_limit_trim_risk(cursor, state, 2000) + False + + Close to it, they cannot: + + >>> _history_limit_trim_risk(cursor, state, 10) + True + + A pane with no scrollback at all is always at risk: + + >>> _history_limit_trim_risk(cursor, state, 0) + True + """ + if history_limit <= 0: + return True + trim_batch = max(history_limit // 10, 1) + risk_floor = history_limit - trim_batch + return cursor.history_size >= risk_floor or state.history_size >= risk_floor + + +def _find_unique_cursor_match(rows: list[str], cursor: CaptureCursor) -> int | None: + """Locate the cursor's anchor in ``rows`` by content fingerprint. + + Used when :func:`_history_limit_trim_risk` says offsets cannot be + trusted. Matches the anchor row *and* the rows recorded beneath it, so + a repeated single line (a bare shell prompt, say) does not anchor to + the wrong place. + + Parameters + ---------- + rows : list[str] + Captured rows to search, oldest first. + cursor : CaptureCursor + The anchor to locate. + + Returns + ------- + int | None + Index of the anchor row, or ``None`` when the fingerprint is + absent or appears more than once -- either way the anchor cannot + be proven, and the caller must report a missed read. + + Examples + -------- + >>> rows = ['alpha', 'beta', 'gamma'] + >>> cursor = CaptureCursor( + ... '%1', '1', 0, 24, 0, _line_hash('beta'), (_line_hash('gamma'),) + ... ) + >>> _find_unique_cursor_match(rows, cursor) + 1 + + An absent fingerprint does not match: + + >>> _find_unique_cursor_match(['alpha'], cursor) is None + True + + An ambiguous fingerprint is refused rather than guessed: + + >>> ambiguous = CaptureCursor('%1', '1', 0, 24, 0, _line_hash('alpha'), ()) + >>> _find_unique_cursor_match(['alpha', 'beta', 'alpha'], ambiguous) is None + True + + A cursor with no anchor row has nothing to match on: + + >>> _find_unique_cursor_match(rows, CaptureCursor( + ... '%1', '1', 0, 24, 0, None, () + ... )) is None + True + """ + if cursor.anchor_hash is None: + return None + + fingerprint = (cursor.anchor_hash, *cursor.below_hashes) + if len(rows) < len(fingerprint): + return None + + # Hash each row once. Windows overlap, so hashing per-window would + # re-hash a row once for every window it appears in -- up to the + # fingerprint's length -- on the path that is already the expensive + # fallback. + hashes = [_line_hash(line) for line in rows] + + match_index: int | None = None + for index in range(len(hashes) - len(fingerprint) + 1): + if tuple(hashes[index : index + len(fingerprint)]) != fingerprint: + continue + if match_index is not None: + return None + match_index = index + return match_index + + +def _drop_previously_seen_rows( + rows: list[str], + cursor: CaptureCursor, +) -> list[str]: + """Drop rows the cursor already reported. + + Compares hashes rather than counting, so a row rewritten in place -- + a progress bar redrawing over itself with a carriage return -- is + reported as new content instead of skipped as already-seen. + + Parameters + ---------- + rows : list[str] + Captured rows starting at the anchor, oldest first. + cursor : CaptureCursor + The anchor these rows were captured from. + + Returns + ------- + list[str] + + Examples + -------- + >>> cursor = CaptureCursor( + ... '%1', '1', 0, 24, 0, _line_hash('prompt'), (_line_hash('below'),) + ... ) + + Unchanged rows are dropped, new ones kept: + + >>> _drop_previously_seen_rows(['prompt', 'below', 'fresh'], cursor) + ['fresh'] + + A rewritten anchor row is new content, while the unchanged row below + it stays dropped: + + >>> _drop_previously_seen_rows(['prompt$ ls', 'below'], cursor) + ['prompt$ ls'] + + Matching stops at the first difference, so nothing after a changed + row is dropped on position alone: + + >>> _drop_previously_seen_rows(['prompt', 'rewritten', 'below'], cursor) + ['rewritten', 'below'] + + >>> _drop_previously_seen_rows([], cursor) + [] + """ + if not rows: + return [] + + output: list[str] = [] + if cursor.anchor_hash is None or _line_hash(rows[0]) != cursor.anchor_hash: + output.append(rows[0]) + tail = rows[1:] + + drop = 0 + for expected_hash, line in zip(cursor.below_hashes, tail, strict=False): + if _line_hash(line) != expected_hash: + break + drop += 1 + output.extend(tail[drop:]) + return output + + +def _build_cursor( + pane_id: str, + state: _PaneState, + cursor_rows: list[str], +) -> CaptureCursor: + """Build the cursor describing where a completed read stopped. + + Parameters + ---------- + pane_id : str + Pane the read came from. + state : _PaneState + The snapshot the read settled on. + cursor_rows : list[str] + Rows from the cursor row through the visible bottom. + + Returns + ------- + CaptureCursor + + Examples + -------- + >>> _build_cursor('%1', _PaneState(100, 5, 24, '42', False), ['a', 'b']) + CaptureCursor(pane_id='%1', pane_pid='42', history_size=100, \ +pane_height=24, anchor_abs=105, anchor_hash='...', below_hashes=('...',)) + + A cursor below the visible region has no rows to fingerprint: + + >>> _build_cursor('%1', _PaneState(0, 0, 24, '42', False), []).anchor_hash \ +is None + True + """ + return CaptureCursor( + pane_id=pane_id, + pane_pid=state.pane_pid, + history_size=state.history_size, + pane_height=state.pane_height, + anchor_abs=state.history_size + state.cursor_y, + anchor_hash=_line_hash(cursor_rows[0]) if cursor_rows else None, + below_hashes=tuple(_line_hash(line) for line in cursor_rows[1:]), + ) + + +def _raise_if_lifecycle_changed( + pane_id: str | None, + state: _PaneState, + baseline_pid: str, +) -> None: + """Raise when a cursor's process identity no longer holds. + + Examples + -------- + >>> _raise_if_lifecycle_changed('%1', _PaneState(0, 0, 24, '42', False), '42') + + >>> _raise_if_lifecycle_changed('%1', _PaneState(0, 0, 24, '99', False), '42') + Traceback (most recent call last): + libtmux.exc.PaneLifecycleChanged: pane %1 was respawned (pid 42 -> 99); \ +cursor anchor is no longer valid + + >>> _raise_if_lifecycle_changed('%1', _PaneState(0, 0, 24, '42', True), '42') + Traceback (most recent call last): + libtmux.exc.PaneLifecycleChanged: pane %1 died; cursor anchor is no \ +longer valid + """ + if state.pane_dead: + msg = f"pane {pane_id} died; cursor anchor is no longer valid" + raise exc.PaneLifecycleChanged(msg) + if state.pane_pid != baseline_pid: + msg = ( + f"pane {pane_id} was respawned " + f"(pid {baseline_pid} -> {state.pane_pid}); " + "cursor anchor is no longer valid" + ) + raise exc.PaneLifecycleChanged(msg) + + +# -------------------------------------------------------------------------- +# TMUX I/O BOUNDARY +# +# Everything above decides what a read *means* given values, and runs +# without a tmux server. Everything below performs tmux round-trips. Keep +# new logic above the line unless it genuinely needs to talk to tmux. +# -------------------------------------------------------------------------- + + +class _PaneRead(t.NamedTuple): + """One completed tmux read, before it becomes a :class:`CaptureSince`. + + Attributes + ---------- + state : _PaneState + The snapshot the read settled on. + cursor_rows : list[str] + Rows from the cursor row through the visible bottom, used to + fingerprint the next cursor. + lines : list[str] + Rows to return to the caller. + lines_missed : bool + Whether ``lines`` is a fallback visible capture rather than a + complete delta. + """ + + state: _PaneState + cursor_rows: list[str] + lines: list[str] + lines_missed: bool + + +def _read_pane_state(pane: Pane) -> _PaneState: + """Snapshot ``pane``'s grid and lifecycle in one round-trip. + + Examples + -------- + >>> state = _read_pane_state(pane) + >>> state.pane_height > 0 + True + >>> state.pane_dead + False + """ + stdout = pane.display_message(PANE_STATE_FORMAT, get_text=True) + return _parse_pane_state(stdout[0] if stdout else "0|0|0||0") + + +def _read_history_limit(pane: Pane) -> int: + """Read ``pane``'s ``history-limit`` once. + + Fixed at pane creation -- a retroactive ``set-option history-limit`` + only takes effect from tmux 3.7 (commit ``e7b1575``), and older + versions need a new pane. Safe to cache for one capture, and kept out + of :func:`_read_pane_state` so per-tick reads do not pay for a value + that cannot change between ticks. + + Examples + -------- + >>> _read_history_limit(pane) > 0 + True + """ + stdout = pane.display_message(HISTORY_LIMIT_FORMAT, get_text=True) + return int(stdout[0] if stdout else "0") + + +def _capture_rows( + pane: Pane, + *, + start: t.Literal["-"] | int | None = None, + end: t.Literal["-"] | int | None = None, +) -> list[str]: + """Capture pane rows, refusing to mistake a failed read for silence. + + Issues ``capture-pane`` directly rather than through + :meth:`~libtmux.pane.Pane.capture_pane`, which returns tmux's stdout + without inspecting stderr. A blank pane and a failed capture both + yield no rows there, and this module cannot tell a caller "nothing + was written" unless it knows the read succeeded. + + Examples + -------- + >>> isinstance(_capture_rows(pane), list) + True + """ + args = ["capture-pane", "-p"] + if start is not None: + args.extend(["-S", str(start)]) + if end is not None: + args.extend(["-E", str(end)]) + proc = pane.cmd(*args) + raise_if_stderr(proc, "capture-pane") + return list(proc.stdout) + + +def _capture_cursor_rows(pane: Pane, state: _PaneState) -> list[str]: + """Capture from the cursor row through the visible bottom. + + Examples + -------- + >>> isinstance(_capture_cursor_rows(pane, _read_pane_state(pane)), list) + True + + A cursor below the visible region has no rows: + + >>> _capture_cursor_rows(pane, _PaneState(0, 99, 24, '1', False)) + [] + """ + if state.cursor_y >= state.pane_height: + return [] + return _capture_rows(pane, start=state.cursor_y, end=None) + + +def _read_stable_visible( + pane: Pane, + *, + baseline_pid: str | None = None, +) -> _PaneRead: + """Capture the visible pane, re-sampling until the grid holds still. + + Samples state, captures, then re-samples. A difference means the pane + moved mid-capture and the rows may be torn across two grid states, so + the read is retried. After :data:`_STABLE_READ_ATTEMPTS` the rows are + returned with ``lines_missed`` set rather than presented as exact. + + Exhausting the attempts returns the *last attempt's* reads rather than + taking fresh ones. On a pane writing continuously enough to defeat + three brackets, another round of unbracketed samples would pair an + anchor row number with a fingerprint taken at a different instant, and + the resulting cursor would claim to anchor content it never saw. The + last attempt's ``before`` snapshot and the rows captured against it at + least describe one moment. + + Parameters + ---------- + pane : Pane + Pane to read. + baseline_pid : str, optional + PID a cursor expects. When omitted this is a first read, so any + live PID is accepted and only pane death is an error. + + Returns + ------- + _PaneRead + + Examples + -------- + >>> read = _read_stable_visible(pane) + >>> read.lines_missed + False + >>> isinstance(read.lines, list) + True + """ + before = _read_pane_state(pane) + lines: list[str] = [] + cursor_rows: list[str] = [] + for _attempt in range(_STABLE_READ_ATTEMPTS): + before = _read_pane_state(pane) + if baseline_pid is None: + _raise_if_dead_without_baseline(pane, before) + expected_pid = before.pane_pid + else: + expected_pid = baseline_pid + _raise_if_lifecycle_changed(pane.pane_id, before, expected_pid) + + lines = _capture_rows(pane) + cursor_rows = _capture_cursor_rows(pane, before) + after = _read_pane_state(pane) + _raise_if_lifecycle_changed(pane.pane_id, after, expected_pid) + if before == after: + return _PaneRead( + state=after, + cursor_rows=cursor_rows, + lines=lines, + lines_missed=False, + ) + + logger.debug( + "pane never settled across %s reads; reporting a missed capture", + _STABLE_READ_ATTEMPTS, + extra={"tmux_pane": pane.pane_id, "tmux_stdout_len": len(lines)}, + ) + return _PaneRead( + state=before, + cursor_rows=cursor_rows, + lines=lines, + lines_missed=True, + ) + + +def _raise_if_dead_without_baseline(pane: Pane, state: _PaneState) -> None: + """Raise when a first read finds the pane already dead. + + Examples + -------- + >>> _raise_if_dead_without_baseline(pane, _read_pane_state(pane)) + """ + if state.pane_dead: + msg = f"pane {pane.pane_id} died during pane read" + raise exc.PaneLifecycleChanged(msg) + + +def _read_delta(pane: Pane, cursor: CaptureCursor) -> _PaneRead: + """Capture rows written since ``cursor``, or fall back on anchor loss. + + Parameters + ---------- + pane : Pane + Pane to read. + cursor : CaptureCursor + Anchor to read from. + + Returns + ------- + _PaneRead + + Examples + -------- + >>> read = _read_delta(pane, pane.capture_since().cursor) + >>> read.lines_missed + False + """ + history_limit = _read_history_limit(pane) + for _attempt in range(_STABLE_READ_ATTEMPTS): + before = _read_pane_state(pane) + _raise_if_lifecycle_changed(pane.pane_id, before, cursor.pane_pid) + if _cursor_anchor_lost(cursor, before): + return _missed_read(pane, cursor) + + trim_risk = _history_limit_trim_risk(cursor, before, history_limit) + start = cursor.anchor_abs - before.history_size + if trim_risk: + rows = _capture_rows(pane, start="-", end=None) + else: + # ``_cursor_anchor_lost`` returning False above already proved + # ``anchor_abs`` sits at or below the grid bottom, so ``start`` + # is always below ``pane_height``. It may still be negative, + # which is how ``capture-pane -S`` addresses retained history. + rows = _capture_rows(pane, start=start, end=None) + cursor_rows = _capture_cursor_rows(pane, before) + + after = _read_pane_state(pane) + _raise_if_lifecycle_changed(pane.pane_id, after, cursor.pane_pid) + if before != after: + continue + + if trim_risk: + match_index = _find_unique_cursor_match(rows, cursor) + if match_index is None: + return _missed_read(pane, cursor) + rows = rows[match_index:] + return _PaneRead( + state=after, + cursor_rows=cursor_rows, + lines=_drop_previously_seen_rows(rows, cursor), + lines_missed=False, + ) + + return _missed_read(pane, cursor) + + +def _missed_read(pane: Pane, cursor: CaptureCursor) -> _PaneRead: + """Fall back to the visible screen and mark the delta incomplete. + + Examples + -------- + >>> _missed_read(pane, pane.capture_since().cursor).lines_missed + True + """ + missed = _read_stable_visible(pane, baseline_pid=cursor.pane_pid) + return missed._replace(lines_missed=True) + + +def _capture_since(pane: Pane, cursor: CaptureCursor | None = None) -> CaptureSince: + """Capture rows written to ``pane`` since ``cursor``. + + Implements :meth:`libtmux.pane.Pane.capture_since`; call that instead. + + Parameters + ---------- + pane : Pane + Pane to read. + cursor : CaptureCursor, optional + Anchor from a previous call. When omitted the current visible + screen is captured and a first cursor is opened. + + Returns + ------- + CaptureSince + + Raises + ------ + libtmux.exc.InvalidCaptureCursor + If ``cursor`` belongs to a different pane. + libtmux.exc.PaneLifecycleChanged + If the pane died or was respawned since ``cursor`` was taken, or, + when no ``cursor`` is given, if the pane is already dead. + + Examples + -------- + >>> first = _capture_since(pane) + >>> _capture_since(pane, first.cursor).lines + [] + """ + if pane.pane_id is None: + raise exc.PaneNotFound + if cursor is not None and cursor.pane_id != pane.pane_id: + msg = ( + f"invalid capture_since cursor: cursor pane {cursor.pane_id} " + f"does not match requested pane {pane.pane_id}" + ) + raise exc.InvalidCaptureCursor(msg) + + read = _read_stable_visible(pane) if cursor is None else _read_delta(pane, cursor) + return CaptureSince( + lines=read.lines, + cursor=_build_cursor(pane.pane_id, read.state, read.cursor_rows), + lines_missed=read.lines_missed, + ) diff --git a/src/libtmux/exc.py b/src/libtmux/exc.py index 57bb06102f..83c4722654 100644 --- a/src/libtmux/exc.py +++ b/src/libtmux/exc.py @@ -371,6 +371,27 @@ def __init__(self, pane_id: str | None = None, *args: object) -> None: super().__init__("Pane not found") +class CaptureCursorError(LibTmuxException): + """Any reason a :class:`~libtmux.capture.CaptureCursor` stopped being usable. + + Catch this to handle every cursor invalidation in one place, rather + than enumerating malformed payloads, cross-pane replays, and pane + lifecycle changes separately. + """ + + +class InvalidCaptureCursor(CaptureCursorError, ValueError): + """Capture cursor is malformed, unreadable, or from a different pane.""" + + +class PaneLifecycleChanged(CaptureCursorError, PaneError): + """Pane died or was respawned, so a capture cursor no longer applies. + + The ``pane_id`` outlives the process it pointed at, so continuing + would read a different program's output as if it were the original's. + """ + + class WindowError(LibTmuxException): """Any type of window related error.""" diff --git a/src/libtmux/pane.py b/src/libtmux/pane.py index e0c2f59619..0b61d74ee6 100644 --- a/src/libtmux/pane.py +++ b/src/libtmux/pane.py @@ -15,6 +15,7 @@ from libtmux import exc from libtmux._internal.env import pane_id_from_env +from libtmux.capture import CaptureCursor, CaptureSince, _capture_since from libtmux.common import get_version_str, has_gte_version, raise_if_stderr, tmux_cmd from libtmux.constants import ( PANE_DIRECTION_FLAG_MAP, @@ -691,6 +692,82 @@ def capture_pane( return None return proc.stdout + def capture_since(self, cursor: CaptureCursor | None = None) -> CaptureSince: + """Capture only the rows written since ``cursor``. + + Where :meth:`capture_pane` returns a snapshot, this returns a + *delta* plus a fresh cursor to resume from, for watching a pane + over time without re-reading and re-diffing the whole screen. + + The cursor is never modified: replaying the same one returns the + same rows. Assign the returned cursor to advance. + + When tmux destroyed the anchor -- ``clear-history``, or a + ``history-limit`` trim that discarded it -- + :attr:`~libtmux.capture.CaptureSince.lines_missed` is ``True`` and + the rows are the current visible screen instead of a complete + delta. Rows are never silently dropped without that flag. + + Parameters + ---------- + cursor : CaptureCursor, optional + Cursor from a previous call. When omitted, captures the + current visible screen and opens a first cursor. + + Returns + ------- + CaptureSince + ``(lines, cursor, lines_missed)``. + + Raises + ------ + libtmux.exc.InvalidCaptureCursor + If ``cursor`` belongs to a different pane. + libtmux.exc.PaneLifecycleChanged + If the pane died or was respawned since ``cursor`` was taken, + or, when no ``cursor`` is given, if the pane is already dead. + + See Also + -------- + libtmux.pane.Pane.capture_pane : Snapshot of the whole pane. + + Examples + -------- + A first call opens a cursor: + + >>> first = pane.capture_since() + >>> first.lines_missed + False + + Nothing new has been written, so replaying it returns nothing: + + >>> pane.capture_since(first.cursor).lines + [] + + New output comes back on its own: + + >>> pane.send_keys('echo capture_since_demo', enter=True) + >>> from libtmux.test.retry import retry_until + >>> retry_until( + ... lambda: any( + ... 'capture_since_demo' in line + ... for line in pane.capture_since(first.cursor).lines + ... ), + ... 2, + ... ) + True + + The cursor it was handed is untouched, so the next delta is taken + from the cursor the call returned: + + >>> second = pane.capture_since(first.cursor) + >>> second.cursor == first.cursor + False + + .. versionadded:: 0.63 + """ + return _capture_since(self, cursor) + def send_keys( self, cmd: str | None = None, diff --git a/tests/test_capture_since.py b/tests/test_capture_since.py new file mode 100644 index 0000000000..044655d9ef --- /dev/null +++ b/tests/test_capture_since.py @@ -0,0 +1,371 @@ +"""Tests for :meth:`libtmux.pane.Pane.capture_since` and its cursor. + +Covers the delta contract (only new rows come back), the anchor-loss +contract (``lines_missed`` is set rather than returning a silently +incomplete delta), and the lifecycle contract (a cursor never reads a +different process's output). +""" + +from __future__ import annotations + +import itertools +import typing as t +import uuid + +import pytest + +from libtmux import capture, exc +from libtmux.capture import CaptureCursor +from libtmux.test.retry import retry_until + +if t.TYPE_CHECKING: + from libtmux.pane import Pane + from libtmux.session import Session + + +def run_and_wait(pane: Pane, payload: str) -> None: + """Run ``payload`` in ``pane`` and block until the shell finishes it. + + Polls rather than signalling through ``tmux wait-for``: a missed + signal on a channel blocks forever, while a poll degrades to a + :exc:`~libtmux.exc.WaitTimeout`. + + The sentinel is assembled by ``printf`` from two arguments, so the + string being polled for never appears in the command line the shell + echoes back. Polling for text that is present in the typed command + matches that echo and returns before the payload has run at all. + """ + token = uuid.uuid4().hex[:8].upper() + sentinel = f"SETTLED{token}" + pane.send_keys(f"{payload}; printf '%s%s\\n' SETTLED {token}", enter=True) + retry_until( + lambda: any(sentinel in line for line in pane.capture_pane()), + 5, + raises=True, + ) + + +def test_first_call_returns_visible_screen_and_cursor(session: Session) -> None: + """An initial call captures visible content and opens a cursor.""" + pane = session.new_window(window_name="capture_since_first").active_pane + assert pane is not None + marker = "CAPTURE_SINCE_INITIAL" + run_and_wait(pane, f"echo {marker}") + + result = pane.capture_since() + + assert result.cursor.pane_id == pane.pane_id + assert result.lines_missed is False + assert any(marker in line for line in result.lines) + + +def test_followup_returns_only_new_output(session: Session) -> None: + """Follow-up calls return content written after the previous cursor.""" + pane = session.new_window(window_name="capture_since_delta").active_pane + assert pane is not None + old_marker = "CAPTURE_SINCE_OLD" + new_marker = "CAPTURE_SINCE_NEW" + run_and_wait(pane, f"echo {old_marker}") + first = pane.capture_since() + + run_and_wait(pane, f"echo {new_marker}") + second = pane.capture_since(first.cursor) + third = pane.capture_since(second.cursor) + + assert any(new_marker in line for line in second.lines) + assert not any(old_marker in line for line in second.lines) + assert third.lines == [] + + +def test_capture_since_does_not_mutate_the_cursor(session: Session) -> None: + """Replaying one cursor twice yields the same delta (#740).""" + pane = session.new_window(window_name="capture_since_pure").active_pane + assert pane is not None + marker = "CAPTURE_SINCE_REPLAY" + first = pane.capture_since() + snapshot = first.cursor + + run_and_wait(pane, f"echo {marker}") + once = pane.capture_since(first.cursor) + twice = pane.capture_since(first.cursor) + + assert first.cursor == snapshot + assert once.lines == twice.lines + + +def test_follows_anchor_into_retained_history(session: Session) -> None: + """A cursor stays exact after its anchor scrolls into history.""" + pane = session.new_window(window_name="capture_since_scroll").active_pane + assert pane is not None + first = pane.capture_since() + pane_height = int(pane.display_message("#{pane_height}", get_text=True)[0]) + markers = [f"CAPTURE_SINCE_SCROLL_{index:02d}" for index in range(pane_height + 8)] + + run_and_wait(pane, "printf '%s\\n' " + " ".join(markers)) + second = pane.capture_since(first.cursor) + + assert second.lines_missed is False + assert any(markers[-1] in line for line in second.lines) + + +def test_reports_same_row_rewrite(session: Session) -> None: + """Carriage-return rewrites on the cursor row count as new content.""" + pane = session.new_window(window_name="capture_since_rewrite").active_pane + assert pane is not None + script = ( + "printf OLD_REWRITE_CAPTURE_SINCE; " + "IFS= read -r line; " + "printf '\\r%s' \"$line\"; " + "sleep 60" + ) + + def on_screen(marker: str) -> bool: + return any(marker in line for line in pane.capture_pane()) + + pane.respawn(kill=True, shell=f"sh -c '{script}'") + retry_until(lambda: on_screen("OLD_REWRITE_CAPTURE_SINCE"), 5, raises=True) + first = pane.capture_since() + + pane.send_keys("NEW_REWRITE_CAPTURE_SINCE", enter=True) + retry_until(lambda: on_screen("NEW_REWRITE_CAPTURE_SINCE"), 5, raises=True) + second = pane.capture_since(first.cursor) + + assert any("NEW_REWRITE_CAPTURE_SINCE" in line for line in second.lines) + + +def test_marks_lines_missed_after_history_clear(session: Session) -> None: + """Lost history returns current visible content with ``lines_missed``.""" + pane = session.new_window(window_name="capture_since_clear").active_pane + assert pane is not None + fill = "; ".join(f"echo CAPTURE_SINCE_HISTORY_{i}" for i in range(40)) + run_and_wait(pane, fill) + first = pane.capture_since() + + pane.cmd("clear-history") + run_and_wait(pane, "echo CAPTURE_SINCE_AFTER_CLEAR") + second = pane.capture_since(first.cursor) + + assert second.lines_missed is True + assert any("CAPTURE_SINCE_AFTER_CLEAR" in line for line in second.lines) + + +def test_marks_lines_missed_after_clear_history_with_resize(session: Session) -> None: + """``clear-history`` plus a pane grow still detects anchor loss. + + Regression: an early ``pane_height`` guard returned False when the + pane grew after ``clear-history``, masking the complete history wipe. + """ + window = session.new_window(window_name="capture_since_resize") + pane = window.split() + fill = "; ".join(f"echo RESIZE_CLEAR_{i}" for i in range(40)) + run_and_wait(pane, fill) + first = pane.capture_since() + + pane.cmd("clear-history") + assert pane.pane_height is not None + pane.set_height(int(pane.pane_height) + 3) + run_and_wait(pane, "echo AFTER_RESIZE_CLEAR") + second = pane.capture_since(first.cursor) + + assert second.lines_missed is True + assert any("AFTER_RESIZE_CLEAR" in line for line in second.lines) + + +def test_marks_lines_missed_after_history_limit_trim(session: Session) -> None: + """History-limit trims return visible content with ``lines_missed``. + + Floods past ``history-limit`` then clears history to guarantee the + anchor is destroyed. The flood alone is not deterministic -- tmux + retains enough of the original prompt that the fingerprint search + can legitimately re-anchor on a surviving hash. + """ + session.cmd("set-option", "-g", "history-limit", "20") + window = session.new_window(window_name="capture_since_trim") + pane = window.split() + + def hlimit_locked() -> bool: + raw = pane.display_message("#{history_limit}", get_text=True) + return bool(raw) and int(raw[0]) == 20 + + retry_until(hlimit_locked, 5, raises=True) + run_and_wait( + pane, + "for i in $(seq 1 25); do printf 'PREFILL_%03d\\n' \"$i\"; done", + ) + first = pane.capture_since() + + run_and_wait( + pane, + "for i in $(seq 1 120); do printf 'TRIM_%03d\\n' \"$i\"; done", + ) + pane.cmd("clear-history") + run_and_wait(pane, "echo TRIM_DONE") + second = pane.capture_since(first.cursor) + + assert second.lines_missed is True + assert any("TRIM" in line for line in second.lines) + + +def test_rejects_malformed_cursor() -> None: + """Malformed cursor strings fail loudly instead of guessing.""" + with pytest.raises(exc.InvalidCaptureCursor, match="unsupported cursor format"): + CaptureCursor.from_str("not-a-valid-cursor") + + +def test_rejects_cursor_for_a_different_pane(session: Session) -> None: + """A cursor cannot be replayed against a different pane.""" + window = session.new_window(window_name="capture_since_other") + pane = window.active_pane + assert pane is not None + first = pane.capture_since() + other_pane = window.split() + + with pytest.raises(exc.InvalidCaptureCursor, match="cursor pane"): + other_pane.capture_since(first.cursor) + + +def test_rejects_respawned_pane_cursor(session: Session) -> None: + """Pane respawn invalidates the cursor's process identity.""" + pane = session.new_window(window_name="capture_since_respawn").active_pane + assert pane is not None + first = pane.capture_since() + + pane.respawn(kill=True, shell="sleep 60") + + with pytest.raises(exc.PaneLifecycleChanged, match="respawned"): + pane.capture_since(first.cursor) + + +def test_rejects_dead_pane_cursor(session: Session) -> None: + """Pane death invalidates the cursor instead of returning stale rows.""" + window = session.new_window(window_name="capture_since_dead") + pane = window.active_pane + assert pane is not None + first = pane.capture_since() + window.cmd("set-option", "-w", "remain-on-exit", "on") + pane.respawn(kill=True, shell="true") + + def is_dead() -> bool: + out = pane.cmd("display-message", "-p", "#{pane_dead}").stdout + return bool(out) and out[0].strip() == "1" + + retry_until(is_dead, 5, raises=True) + + with pytest.raises(exc.PaneLifecycleChanged, match="died"): + pane.capture_since(first.cursor) + + +def test_rejects_a_dead_pane_without_a_cursor(session: Session) -> None: + """A first call on an already-dead pane refuses rather than reading it.""" + window = session.new_window(window_name="capture_since_dead_first") + pane = window.active_pane + assert pane is not None + window.cmd("set-option", "-w", "remain-on-exit", "on") + pane.respawn(kill=True, shell="true") + + def is_dead() -> bool: + out = pane.cmd("display-message", "-p", "#{pane_dead}").stdout + return bool(out) and out[0].strip() == "1" + + retry_until(is_dead, 5, raises=True) + + with pytest.raises(exc.PaneLifecycleChanged, match="died"): + pane.capture_since() + + +def test_a_failed_capture_raises_instead_of_reading_as_empty( + session: Session, monkeypatch: pytest.MonkeyPatch +) -> None: + """A tmux capture failure is never reported as "nothing was written". + + ``Pane.capture_pane`` returns tmux's stdout without inspecting stderr, + so a failed capture and a blank pane are indistinguishable there. Uses + ``monkeypatch`` because provoking a real ``capture-pane`` failure + against a live, healthy pane is not otherwise reachable. + """ + pane = session.new_window(window_name="capture_since_failed_read").active_pane + assert pane is not None + real_cmd = type(pane).cmd + + def failing_capture(self: Pane, *args: str) -> t.Any: + proc = real_cmd(self, *args) + if args and args[0] == "capture-pane": + proc.stderr = ["no such pane"] + proc.stdout = [] + return proc + + monkeypatch.setattr(type(pane), "cmd", failing_capture) + + with pytest.raises(exc.LibTmuxException, match="capture-pane"): + pane.capture_since() + + +def test_cursor_round_trips_through_a_string(session: Session) -> None: + """A serialized cursor decodes back to an equal cursor.""" + pane = session.new_window(window_name="capture_since_codec").active_pane + assert pane is not None + first = pane.capture_since() + + encoded = str(first.cursor) + decoded = CaptureCursor.from_str(encoded) + + assert encoded.startswith("capture-since-v1:") + assert decoded == first.cursor + + +def test_deserialized_cursor_still_captures_a_delta(session: Session) -> None: + """A cursor that survived serialization behaves like the original.""" + pane = session.new_window(window_name="capture_since_codec_delta").active_pane + assert pane is not None + marker = "CAPTURE_SINCE_CODEC" + first = pane.capture_since() + + run_and_wait(pane, f"echo {marker}") + second = pane.capture_since(CaptureCursor.from_str(str(first.cursor))) + + assert any(marker in line for line in second.lines) + + +def never_settles(pane: Pane, monkeypatch: pytest.MonkeyPatch) -> None: + """Make every pane-state sample differ from the one before it. + + Simulates a pane written to continuously, so no read ever brackets a + capture with two matching snapshots. ``monkeypatch`` rather than a + real busy pane because the retry loop guards a race: a genuine pane + cannot be made to move between *every* pair of samples on demand, so + the condition has to be injected to be asserted on at all. + """ + steady = capture._read_pane_state(pane) + samples = itertools.count() + monkeypatch.setattr( + capture, + "_read_pane_state", + lambda _pane: steady._replace(cursor_y=next(samples) % 5), + ) + + +def test_unstable_pane_reports_a_missed_first_read( + session: Session, monkeypatch: pytest.MonkeyPatch +) -> None: + """A pane that never holds still reports a missed read, not a torn one.""" + pane = session.new_window(window_name="capture_since_unstable").active_pane + assert pane is not None + never_settles(pane, monkeypatch) + + assert pane.capture_since().lines_missed is True + + +def test_unstable_pane_reports_a_missed_delta( + session: Session, monkeypatch: pytest.MonkeyPatch +) -> None: + """The delta path degrades the same way when the grid keeps moving. + + The injected states leave the anchor valid, so a missed read can only + come from the stability retry giving up -- not from anchor loss. + """ + pane = session.new_window(window_name="capture_since_unstable_delta").active_pane + assert pane is not None + first = pane.capture_since() + never_settles(pane, monkeypatch) + + assert pane.capture_since(first.cursor).lines_missed is True