Saving inputs and outputs of each pane

[ch1c0t]

Hi @tony, thank you for libtmux.

I want to save inputs and outputs of interactive zsh sessions. Similarly to what Atuin does, but it should also store outputs resulted from inputs, as described in this issue.

Can it be implemented with libtmux? Probably, yes. Should it be implemented with libtmux?

I am not sure how to answer these questions. So far, I have tried tmux pipe-pane, but it only saves outputs to text files, whereas I want something alike to a Python list representing session inputs, which also provide access to outputs associated with them.

How would you approach answering the above questions?

[tony]

Saving inputs and outputs of each pane

Short answer: yes. Try this as a starting point (you can feed into an LLM).

You can build exactly what you described - "a list of inputs that also give access to the outputs associated with them" - on top of libtmux today. The one piece neither tmux nor libtmux can synthesize for you is where one command ends and the next begins; that signal has to come from your shell, via OSC 133 semantic prompts. Once your shell emits those, libtmux already ships the two primitives you need:

APIs available in libtmux as of v0.60.0

• We have Pane.capture_pane() (source) - a one-shot snapshot of the pane.
• We have Pane.pipe() (source) - wraps tmux pipe-pane, streaming the pane's raw output to a command (libtmux ≥ 0.56.0).

Why tmux pipe-pane only gave you text

pipe-pane (and capture_pane, and tmux control-mode %output) all hand you the raw pane byte stream: program output and the terminal's echo of your keystrokes, interleaved, with no command boundaries and no exit codes. That is why your pipe-pane experiment produced a flat text file and not paired input/output records. tmux can't tell "this is what I typed" from "this is what the program printed" - that knowledge lives in the shell.

This is the same wall Atuin hit. Its classic capture is shell preexec/precmd hooks that record only the command (no output - that's atuinsh/atuin#2179); its newer atuin-pty-proxy captures output by parsing OSC 133 markers out of the PTY stream. We can do the same thing with Pane.pipe() as the tap.

The missing piece: OSC 133

OSC 133 marks four points in every shell interaction:

Marker	Meaning
A	prompt is about to print
B	prompt ended - your input begins
C	command submitted - output begins
D;n	command finished with exit code n

Most shell integrations already emit these (Atuin, Starship, and the iTerm2 / Ghostty / WezTerm shell-integration snippets all do - reuse whatever you have). A minimal zsh emitter, used by both examples below:

# rc.zsh - make zsh emit OSC 133 semantic-prompt markers
unsetopt PROMPT_SP PROMPT_CR          # no '%' partial-line marker polluting output
autoload -Uz add-zsh-hook
_op() { print -rn -- $'\e]133;C\a' }                              # output begins
_or() { local e=$?; print -rn -- $'\e]133;D;'"$e"$'\a\e]133;A\a' } # prev exit; prompt start
add-zsh-hook preexec _op
add-zsh-hook precmd  _or
PS1=$'READY$ %{\e]133;B\a%}'           # B = your input begins

Those escapes pass through pipe-pane verbatim (tmux tees the raw PTY bytes before parsing them), so a small state machine can rebuild structured records from the stream.

Example 1 - sync (batch): pipe() → file → records

Drive a pane, tee its raw stream to a file with Pane.pipe(), then segment it. Tested with libtmux 0.60.0 on tmux 3.7.

"""Sync OSC 133 recorder on shipped libtmux. Run: python sync_demo.py"""
from __future__ import annotations

import json
import pathlib
import re
import tempfile
import time

import libtmux

_OSC133 = re.compile(rb"\x1b\]133;([^\x07\x1b]*)(?:\x07|\x1b\\)")
_ESC = re.compile(
    rb"\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)|\x1b[\[\]][0-9;?]*[ -/]*[@-~]|\x1b[@-Z\\-_]"
)


def _clean(b: bytes) -> str:
    """Strip escapes, then replay backspace like a terminal would."""
    out = bytearray()
    for ch in _ESC.sub(b"", b):
        if ch == 0x08:  # backspace
            if out and out[-1] != 0x0A:
                out.pop()
        elif ch == 0x0D or (ch < 0x20 and ch not in (0x09, 0x0A)):
            pass
        else:
            out.append(ch)
    return out.decode("utf-8", "replace")


def segment(raw: bytes) -> list[dict[str, object]]:
    r"""Reconstruct ``{input, output, exit_code}`` records from an OSC 133 stream.

    >>> segment(b"\x1b]133;B\x07ls\x1b]133;C\x07a.txt\n\x1b]133;D;0\x07")
    [{'input': 'ls', 'output': 'a.txt\n', 'exit_code': 0}]
    >>> segment(b"\x1b]133;B\x07false\x1b]133;C\x07\x1b]133;D;1\x07")
    [{'input': 'false', 'output': '', 'exit_code': 1}]
    """
    records: list[dict[str, object]] = []
    state, cur_in, cur_out, last = "idle", b"", b"", 0
    for m in _OSC133.finditer(raw):
        seg = raw[last : m.start()]
        if state == "input":
            cur_in += seg
        elif state == "output":
            cur_out += seg
        last = m.end()
        params = m.group(1)
        kind = params[:1]
        if kind == b"A":
            state = "prompt"
        elif kind == b"B":
            state, cur_in = "input", b""
        elif kind == b"C":
            state, cur_out = "output", b""
        elif kind == b"D":
            code = int(params.split(b";", 1)[1]) if b";" in params else None
            cmd = _clean(cur_in).strip()
            if cmd:
                records.append({"input": cmd, "output": _clean(cur_out), "exit_code": code})
            state, cur_in, cur_out = "idle", b"", b""
    return records


RC_ZSH = r"""
unsetopt PROMPT_SP PROMPT_CR
autoload -Uz add-zsh-hook
_op() { print -rn -- $'\e]133;C\a' }
_or() { local e=$?; print -rn -- $'\e]133;D;'"$e"$'\a\e]133;A\a' }
add-zsh-hook preexec _op
add-zsh-hook precmd  _or
PS1=$'READY$ %{\e]133;B\a%}'
"""


def main() -> None:
    tmp = pathlib.Path(tempfile.mkdtemp(prefix="osc133s_"))
    (tmp / "rc.zsh").write_text(RC_ZSH)
    log = tmp / "pane.raw"
    log.write_bytes(b"")

    server = libtmux.Server(socket_name="osc133_sync")
    session = server.new_session("rec", kill_session=True, window_command="zsh -f")
    pane = session.active_window.active_pane
    commands = ["echo hello world", "false", r"printf 'a\nb\n'", "expr 6 \\* 7"]
    try:
        time.sleep(0.6)
        pane.send_keys(f"source {tmp / 'rc.zsh'}", enter=True)
        time.sleep(0.6)
        pane.pipe(f"cat >> {log}")  # <- shipped Pane.pipe() == tmux pipe-pane
        time.sleep(0.3)
        pane.send_keys("", enter=True)  # prime one clean prompt cycle
        time.sleep(0.3)
        for c in commands:
            pane.send_keys(c, enter=True)
            time.sleep(0.5)
        deadline = time.time() + 8
        while time.time() < deadline:
            if log.read_bytes().count(b"\x1b]133;D") >= len(commands) + 1:
                break
            time.sleep(0.2)
        pane.pipe()  # stop piping
        raw = log.read_bytes()
    finally:
        server.kill()

    print(json.dumps(segment(raw), indent=2))


if __name__ == "__main__":
    main()

Output:

[
  {"input": "echo hello world", "output": "hello world\n", "exit_code": 0},
  {"input": "false",            "output": "",              "exit_code": 1},
  {"input": "printf 'a\\nb\\n'", "output": "a\nb\n",        "exit_code": 0},
  {"input": "expr 6 \\* 7",      "output": "42\n",          "exit_code": 0}
]

Example 2 - asyncio (live streaming): records as they complete

Same idea, but tee into a FIFO and consume it with asyncio so each record is emitted the instant its command finishes - handy for a live recorder/daemon. libtmux drives tmux synchronously (the calls are fast); only the output stream is async. Tested with libtmux 0.60.0 on tmux 3.7.

"""Async live-streaming OSC 133 recorder on shipped libtmux. Run: python async_demo.py"""
from __future__ import annotations

import asyncio
import os
import pathlib
import re
import tempfile

import libtmux

_OSC133 = re.compile(rb"\x1b\]133;([^\x07\x1b]*)(?:\x07|\x1b\\)")
_ESC = re.compile(
    rb"\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)|\x1b[\[\]][0-9;?]*[ -/]*[@-~]|\x1b[@-Z\\-_]"
)


def _clean(b: bytes) -> str:
    out = bytearray()
    for ch in _ESC.sub(b"", b):
        if ch == 0x08:
            if out and out[-1] != 0x0A:
                out.pop()
        elif ch == 0x0D or (ch < 0x20 and ch not in (0x09, 0x0A)):
            pass
        else:
            out.append(ch)
    return out.decode("utf-8", "replace")


class Segmenter:
    r"""Incremental OSC 133 segmenter: ``feed()`` bytes, get completed records.

    >>> s = Segmenter()
    >>> s.feed(b"\x1b]133;B\x07false\x1b]133;C\x07")
    []
    >>> s.feed(b"\x1b]133;D;1\x07")
    [{'input': 'false', 'output': '', 'exit_code': 1}]
    """

    def __init__(self) -> None:
        self.state = "idle"
        self.cur_in = bytearray()
        self.cur_out = bytearray()
        self._tail = bytearray()

    def _absorb(self, seg: bytes) -> None:
        if self.state == "input":
            self.cur_in += seg
        elif self.state == "output":
            self.cur_out += seg

    def feed(self, chunk: bytes) -> list[dict[str, object]]:
        data = bytes(self._tail) + chunk
        records: list[dict[str, object]] = []
        pos = 0
        for m in _OSC133.finditer(data):
            self._absorb(data[pos : m.start()])
            pos = m.end()
            params = m.group(1)
            kind = params[:1]
            if kind == b"A":
                self.state = "prompt"
            elif kind == b"B":
                self.state, self.cur_in = "input", bytearray()
            elif kind == b"C":
                self.state, self.cur_out = "output", bytearray()
            elif kind == b"D":
                code = int(params.split(b";", 1)[1]) if b";" in params else None
                cmd = _clean(bytes(self.cur_in)).strip()
                self.state = "idle"
                if cmd:
                    records.append(
                        {"input": cmd, "output": _clean(bytes(self.cur_out)), "exit_code": code}
                    )
        rest = data[pos:]
        cut = rest.rfind(b"\x1b")  # keep a possibly-incomplete trailing marker
        if cut == -1:
            self._absorb(rest)
            self._tail = bytearray()
        else:
            self._absorb(rest[:cut])
            self._tail = bytearray(rest[cut:])
        return records


RC_ZSH = r"""
unsetopt PROMPT_SP PROMPT_CR
autoload -Uz add-zsh-hook
_op() { print -rn -- $'\e]133;C\a' }
_or() { local e=$?; print -rn -- $'\e]133;D;'"$e"$'\a\e]133;A\a' }
add-zsh-hook preexec _op
add-zsh-hook precmd  _or
PS1=$'READY$ %{\e]133;B\a%}'
"""


async def record(pane, fifo: pathlib.Path, commands: list[str]) -> list[dict[str, object]]:
    loop = asyncio.get_running_loop()
    fd = os.open(fifo, os.O_RDONLY | os.O_NONBLOCK)  # open reader first (FIFO ordering)
    reader = asyncio.StreamReader()
    await loop.connect_read_pipe(
        lambda: asyncio.StreamReaderProtocol(reader), os.fdopen(fd, "rb", buffering=0)
    )
    pane.pipe(f"cat >> {fifo}")  # writer attaches now that a reader exists

    seg = Segmenter()
    records: list[dict[str, object]] = []

    async def drive() -> None:
        await asyncio.sleep(0.3)
        pane.send_keys("", enter=True)  # prime
        for c in commands:
            await asyncio.sleep(0.4)
            pane.send_keys(c, enter=True)

    driver = asyncio.create_task(drive())
    try:
        while len(records) < len(commands):
            chunk = await asyncio.wait_for(reader.read(65536), timeout=10)
            if not chunk:
                break
            for rec in seg.feed(chunk):
                print(f"live -> {rec}")
                records.append(rec)
    finally:
        driver.cancel()
        pane.pipe()  # stop piping
    return records


async def main() -> None:
    tmp = pathlib.Path(tempfile.mkdtemp(prefix="osc133a_"))
    (tmp / "rc.zsh").write_text(RC_ZSH)
    fifo = tmp / "pane.fifo"
    os.mkfifo(fifo)

    server = libtmux.Server(socket_name="osc133_async")
    session = server.new_session("rec", kill_session=True, window_command="zsh -f")
    pane = session.active_window.active_pane
    try:
        await asyncio.sleep(0.6)
        pane.send_keys(f"source {tmp / 'rc.zsh'}", enter=True)
        await asyncio.sleep(0.6)
        await record(pane, fifo, ["echo hi", "false", r"printf 'x\ny\n'"])
    finally:
        server.kill()


if __name__ == "__main__":
    asyncio.run(main())

It prints each record live as the command finishes:

live -> {'input': 'echo hi', 'output': 'hi\n', 'exit_code': 0}
live -> {'input': 'false', 'output': '', 'exit_code': 1}
live -> {'input': "printf 'x\\ny\\n'", 'output': 'x\ny\n', 'exit_code': 0}

Bonus: snapshot-only with capture_pane() (no piping)

If you only want where the prompt and output lines are (not exact bytes or exit codes), Pane.capture_pane(line_flags=True) exposes tmux's per-line OSC 133 flags - P for a prompt line (A), O for an output-start line (C). Requires libtmux ≥ 0.60.0 and tmux ≥ 3.7 (capture-pane -F). Note it loses B (input end) and D (exit code), so it's coarser than the streaming recorder.

import libtmux, time

server = libtmux.Server(socket_name="osc133_capture")
session = server.new_session("c", kill_session=True, window_command="zsh -f")
pane = session.active_window.active_pane
try:
    time.sleep(0.5)
    pane.send_keys(r"PS1=$'\e]133;A\aREADY$ '", enter=True); time.sleep(0.3)
    pane.send_keys(r"preexec() { print -rn $'\e]133;C\a' }", enter=True); time.sleep(0.3)
    pane.send_keys("clear", enter=True); time.sleep(0.3)  # scroll the setup lines off
    pane.send_keys("echo first; echo second", enter=True); time.sleep(0.5)
    for line in pane.capture_pane(line_flags=True):  # tmux 3.7+: capture-pane -F
        if line[2:].strip():
            print(repr(line))
finally:
    server.kill()

'P READY$ echo first; echo second'   # P = prompt line
'O first'                            # O = output begins
'- second'                           # - = continuation
'P READY$'

Versions

• Pane.pipe() - libtmux ≥ 0.56.0.
• Pane.capture_pane(line_flags=True) - libtmux ≥ 0.60.0 + tmux ≥ 3.7.
• Pane.capture_pane() (snapshot) - available throughout.
• The examples above were run on libtmux 0.60.0 / tmux 3.7.

Should this live in libtmux?

I'd keep the full recorder downstream. Command/prompt boundary detection has to come from the shell (OSC 133 or a precmd/preexec hook), not from tmux's ORM. libtmux already ships the pieces (Pane.pipe() for the live stream, Pane.capture_pane() for snapshots); the ~30-line OSC 133 segmenter above is the only glue, and it belongs in your tool (or, if there's interest, a docs/examples/ cookbook entry) rather than the core.

[ch1c0t]

Thank you. I appreciate you taking the time to answer.