Retain immutable pane capture observations

CHANGES

@@ -45,6 +45,14 @@ $ uvx --from 'libtmux' --prerelease allow python
 _Notes on the upcoming release will go here._
 <!-- END PLACEHOLDER - ADD NEW CHANGELOG ENTRIES BELOW THIS LINE -->
 
+### What's new
+
+#### Retained pane observations
+
+Retain immutable pane text and capture details with
+`PaneObservation.capture()`, even after the pane changes or disappears.
+These sampled values remain readable without another tmux query. (#810)
+
 ### Documentation
 
 #### Cleaner `from_env` examples (#719)

docs/api/index.md

@@ -173,6 +173,7 @@ Window <libtmux.window>
 Pane <libtmux.pane>
 Client <libtmux.client>
 Common <libtmux.common>
+Retained observations <libtmux.observation>
 Neo <libtmux.neo>
 Options <libtmux.options>
 Hooks <libtmux.hooks>

docs/api/libtmux.observation.md

@@ -0,0 +1,48 @@
+# Retained pane observations
+
+`PaneObservation.capture(pane)` keeps decoded pane text and its capture command
+in a frozen value. Read it after redraw or pane removal without querying tmux
+again. The pane and server remain borrowed.
+
+```python
+>>> from libtmux import PaneObservation
+>>> observed = PaneObservation.capture(pane)
+>>> observed.pane_id == pane.pane_id
+True
+>>> isinstance(observed.lines, tuple)
+True
+>>> observed.text == "\n".join(observed.lines)
+True
+>>> observed.target_lifetime_verified is None
+True
+```
+
+Use `start`, `end`, and `join_wrapped` to select text with tmux's ordinary
+capture semantics. The result retains the selection, actual command argv, and
+the controller's monotonic capture interval in nanoseconds. Text uses the
+existing command decoder and trailing-empty-line policy. It contains no raw
+byte or terminal-column positions.
+
+`max_text_codepoints` bounds admitted decoded text, including LF between lines.
+It counts Python Unicode code points. Oversized text raises `ValueError`;
+tmux failures raise instead of becoming successful empty observations. The
+limit applies after the existing transport returns, so it does not bound
+transport memory. The call keeps the selected backend's blocking behavior.
+
+The legacy pane ID cannot prove which daemon or pane lifetime supplied a
+capture after replacement; `target_lifetime_verified` remains unknown. The
+interval describes one sampled command, without an atomic metadata/content
+checkpoint. Retained text remains readable after observer or server loss while
+you retain the value. Process exit loses the in-memory value. No serializer or
+persistent record is supplied.
+
+A later capture can recover then-visible text, not omitted output, erased
+history, or hidden terminal state. Visible text and echoed commands do not
+prove application completion; use an explicit producer signal when your
+application cooperates.
+
+```{eval-rst}
+.. autoclass:: libtmux.PaneObservation
+    :members:
+    :member-order: bysource
+```

src/libtmux/__init__.py

@@ -15,6 +15,7 @@
     __version__,
 )
 from .client import Client
+from .observation import PaneObservation
 from .pane import Pane
 from .server import Server
 from .session import Session
@@ -25,6 +26,7 @@
 __all__ = (
     "Client",
     "Pane",
+    "PaneObservation",
     "Server",
     "Session",
     "Window",

src/libtmux/observation.py

@@ -0,0 +1,132 @@
+"""Immutable sampled pane text and capture provenance."""
+
+from __future__ import annotations
+
+import time
+import typing as t
+from dataclasses import dataclass, field
+
+from libtmux import exc
+from libtmux.common import raise_if_stderr
+
+if t.TYPE_CHECKING:
+    from libtmux.pane import Pane
+
+
+@dataclass(frozen=True, slots=True)
+class PaneObservation:
+    """Retain decoded capture text without a live pane or server reference.
+
+    ``lines`` and ``command`` are owned tuples. The interval counts controller
+    monotonic nanoseconds around one command; it is not a server timestamp.
+    ``pane_id`` records the requested target. Its daemon/object lifetime remains
+    unknown with the legacy pane handle, so ``target_lifetime_verified`` is None.
+    """
+
+    pane_id: str
+    lines: tuple[str, ...] = field(repr=False)
+    command: tuple[str, ...] = field(repr=False)
+    start: t.Literal["-"] | int | None
+    end: t.Literal["-"] | int | None
+    join_wrapped: bool
+    started_ns: int
+    finished_ns: int
+    target_lifetime_verified: bool | None = field(default=None, init=False)
+
+    @property
+    def text(self) -> str:
+        """Return retained lines joined with LF, without terminal-column semantics."""
+        return "\n".join(self.lines)
+
+    @classmethod
+    def capture(
+        cls,
+        pane: Pane,
+        *,
+        start: t.Literal["-"] | int | None = None,
+        end: t.Literal["-"] | int | None = None,
+        join_wrapped: bool = False,
+        max_text_codepoints: int = 1_048_576,
+    ) -> PaneObservation:
+        """Capture once through the supplied pane's command route.
+
+        The call retains plain decoded text using the existing command's
+        trailing-empty-line policy. The text limit includes LF between lines
+        and counts Python Unicode code points. It applies after transport
+        decoding; it is not a transport memory cap or command deadline.
+
+        Parameters
+        ----------
+        pane : Pane
+            Borrowed pane providing the backend and target. No cleanup is acquired.
+        start, end : int | "-" | None
+            Requested tmux line range, using ordinary capture-pane semantics.
+        join_wrapped : bool
+            Join wrapped physical lines using tmux's capture policy.
+        max_text_codepoints : int
+            Positive admission limit for retained decoded text, including LF.
+
+        Raises
+        ------
+        ValueError
+            Invalid selection, absent pane ID, or text exceeding the limit.
+        TypeError
+            ``join_wrapped`` is not a bool.
+        libtmux.exc.LibTmuxException
+            The capture command reports stderr or a nonzero/unknown exit status.
+
+        Notes
+        -----
+        Backend exceptions propagate unchanged. Blocking defaults remain those
+        of ``pane.cmd``; this method adds no cancellation operation. A retained
+        value survives pane removal while referenced, but supplies no continuity,
+        durable storage, hidden terminal state, or application-completion proof.
+        """
+        if (
+            isinstance(max_text_codepoints, bool)
+            or not isinstance(max_text_codepoints, int)
+            or max_text_codepoints <= 0
+        ):
+            msg = "max_text_codepoints must be a positive integer"
+            raise ValueError(msg)
+        for name, value in (("start", start), ("end", end)):
+            if value is not None and (
+                isinstance(value, bool) or not (isinstance(value, int) or value == "-")
+            ):
+                msg = f"{name} must be an integer, '-' or None"
+                raise ValueError(msg)
+        if not isinstance(join_wrapped, bool):
+            msg = "join_wrapped must be a bool"
+            raise TypeError(msg)
+        pane_id = pane.pane_id
+        if not pane_id:
+            msg = "Pane must have a pane_id to capture"
+            raise ValueError(msg)
+        args = ["-p"]
+        if start is not None:
+            args.extend(("-S", str(start)))
+        if end is not None:
+            args.extend(("-E", str(end)))
+        if join_wrapped:
+            args.append("-J")
+        started_ns = time.monotonic_ns()
+        result = pane.cmd("capture-pane", *args, target=pane_id)
+        finished_ns = time.monotonic_ns()
+        raise_if_stderr(result, "capture-pane")
+        if result.returncode != 0:
+            msg = f"command exited with status {result.returncode}"
+            raise exc.LibTmuxException(msg, subcommand="capture-pane")
+        text_size = sum(map(len, result.stdout)) + max(0, len(result.stdout) - 1)
+        if text_size > max_text_codepoints:
+            msg = "captured text exceeds max_text_codepoints"
+            raise ValueError(msg)
+        return cls(
+            pane_id=pane_id,
+            lines=tuple(result.stdout),
+            command=tuple(result.cmd),
+            start=start,
+            end=end,
+            join_wrapped=join_wrapped,
+            started_ns=started_ns,
+            finished_ns=finished_ns,
+        )

tests/test_observation.py

@@ -0,0 +1,142 @@
+"""Retained capture values preserve text, routing and failure distinctions."""
+
+from __future__ import annotations
+
+import dataclasses
+import os
+import pathlib
+import shlex
+import subprocess
+import sys
+import typing as t
+
+import pytest
+
+from libtmux import Pane, PaneObservation, exc
+from libtmux.common import tmux_cmd
+
+if t.TYPE_CHECKING:
+    from libtmux.server import Server
+
+
+def test_observation_owns_text_and_actual_command(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A controlled backend permits alias mutation after the capture returns."""
+    result = object.__new__(tmux_cmd)
+    result.stdout = ["界 e\u0301 😀", "second"]
+    result.stderr = []
+    result.returncode = 0
+    result.cmd = ["backend", "capture-pane", "-p", "-S", "0", "-E", "-", "-J"]
+    pane = object.__new__(Pane)
+    pane.pane_id = "%23"
+    calls = []
+
+    def capture(cmd: str, *args: object, target: str | int | None = None) -> tmux_cmd:
+        calls.append((cmd, args, target))
+        return result
+
+    monkeypatch.setattr(pane, "cmd", capture)
+    value = PaneObservation.capture(pane, start=0, end="-", join_wrapped=True)
+    result.stdout[0] = "changed"
+    result.cmd[0] = "changed"
+    assert value.lines == ("界 e\u0301 😀", "second")
+    assert value.command[0] == "backend"
+    assert value.text == "界 e\u0301 😀\nsecond"
+    assert (value.start, value.end, value.join_wrapped) == (0, "-", True)
+    assert value.pane_id == "%23"
+    assert calls == [("capture-pane", ("-p", "-S", "0", "-E", "-", "-J"), "%23")]
+    assert 0 < value.started_ns <= value.finished_ns
+    assert value.target_lifetime_verified is None
+    with pytest.raises(dataclasses.FrozenInstanceError):
+        setattr(value, "pane_id", "%24")  # noqa: B010 - test runtime freezing
+    assert not isinstance(value, Pane)
+    assert not hasattr(value, "server") and not hasattr(value, "cmd")
+
+
+def test_observation_bounds_and_backend_failures(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A controlled backend distinguishes empty text, refusal, and native errors."""
+    pane = object.__new__(Pane)
+    pane.pane_id = "%23"
+    calls = []
+    result = object.__new__(tmux_cmd)
+    result.stdout, result.stderr, result.cmd, result.returncode = [], [], [], 0
+    failure = RuntimeError("remote backend failed")
+
+    def capture(*args: object, **kwargs: object) -> tmux_cmd:
+        calls.append(args)
+        return result
+
+    monkeypatch.setattr(pane, "cmd", capture)
+    for limit in (0, -1, True, 1.5):
+        with pytest.raises(ValueError, match="positive integer"):
+            PaneObservation.capture(pane, max_text_codepoints=t.cast("int", limit))
+    with pytest.raises(ValueError, match="start"):
+        PaneObservation.capture(pane, start=t.cast("int", "invalid"))
+    assert not calls
+    assert PaneObservation.capture(pane).lines == ()
+    result.stdout = ["😀", "e\u0301"]
+    assert PaneObservation.capture(pane, max_text_codepoints=4).text == "😀\ne\u0301"
+    with pytest.raises(ValueError, match="exceeds"):
+        PaneObservation.capture(pane, max_text_codepoints=3)
+    result.stderr = ["no such pane"]
+    with pytest.raises(exc.LibTmuxException, match="no such pane"):
+        PaneObservation.capture(pane)
+    result.stderr, result.returncode = [], 1
+    with pytest.raises(exc.LibTmuxException, match="status 1"):
+        PaneObservation.capture(pane)
+
+    def failed(*args: object, **kwargs: object) -> tmux_cmd:
+        raise failure
+
+    monkeypatch.setattr(pane, "cmd", failed)
+    with pytest.raises(RuntimeError) as raised:
+        PaneObservation.capture(pane)
+    assert raised.value is failure
+    pane.pane_id = None
+    with pytest.raises(ValueError, match="pane_id"):
+        PaneObservation.capture(pane)
+
+
+def test_observation_retains_cooperative_producer_after_removal(
+    server: Server,
+    tmp_path: pathlib.Path,
+) -> None:
+    """A signalled producer supplies text without requiring shell startup control."""
+    server.config_file = os.devnull
+    server.new_session(session_name="keeper", window_command="cat")
+    producer = tmp_path / "observation.py"
+    producer.write_text(
+        "import subprocess, sys\n"
+        "sys.stdout.write('界 e\\u0301 😀\\nretained\\n')\n"
+        "sys.stdout.flush()\n"
+        "subprocess.run(['tmux', '-L' + sys.argv[1], 'wait-for', '-S', "
+        "'observation-ready'], check=True, timeout=0.5)\n"
+        "sys.stdin.readline()\n",
+    )
+    observed = server.new_session(
+        session_name="observed",
+        window_command=shlex.join(
+            [sys.executable, str(producer), str(server.socket_name)]
+        ),
+    )
+    pane = observed.active_pane
+    assert pane is not None
+    subprocess.run(
+        ["tmux", f"-L{server.socket_name}", "wait-for", "observation-ready"],
+        check=True,
+        capture_output=True,
+        timeout=0.5,
+    )
+    value = PaneObservation.capture(pane, start=-20)
+    assert "界 e\u0301 😀" in value.lines
+    assert "retained" in value.lines
+    before = value.lines
+    assert f"-L{server.socket_name}" in value.command
+    pane.kill()
+    assert value.lines == before and "retained" in value.text
+    with pytest.raises(exc.LibTmuxException):
+        PaneObservation.capture(pane)
+    assert server.is_alive()