asyncio: Asynchronous libtmux

[tony]

Docs

Asyncio

• https://docs.python.org/3/library/asyncio.html

Adding Asyncio Support to libtmux (Python 3.9+)

Key Functions Requiring Asynchronous Execution

Several core functions in libtmux currently execute tmux commands synchronously (blocking the main thread). These need refactoring for async operation:

• libtmux.common.tmux_cmd – This helper (class/function) runs tmux commands via subprocess.Popen and waits for completion. It blocks while reading tmux’s output, so making it async is crucial.
• libtmux.neo module – Helper functions (e.g. fetch_obj, list/query functions for sessions, windows, panes) call tmux’s list-* commands and parse results synchronously, blocking the thread. These should be awaitable to avoid blocking.
• libtmux.server.Server methods – Methods like Server.cmd(...), Server.new_session(...), Server.attach_session(...), and Server.kill_server() execute tmux commands synchronously. These must be adapted for asynchronous operation.
• libtmux.session.Session methods – Methods such as Session.new_window(...), Session.rename_session(...), and Session.kill_session() invoke synchronous tmux commands internally and should provide async equivalents or become fully non-blocking.

By identifying these functions, we can target them for asyncio integration so managing tmux sessions and windows no longer halts the main thread.

Approach 1: Using asyncio.subprocess for Non-Blocking Tmux Calls

Python’s asyncio provides native support for spawning subprocesses without blocking the event loop. This would involve rewriting libtmux’s command execution to an async def style. For example:

import asyncio

async def run_tmux_command(args:list[str]) -> list[str]:
    proc = await asyncio.create_subprocess_exec(
        "tmux", *args,
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await proc.communicate()
    if proc.returncode != 0:
        err_msg = stderr.decode().strip()
        raise RuntimeError(f"tmux error: {err_msg}")
    return stdout.decode().splitlines()

In a Server class, we could add an async method:

class Server:
    async def cmd_async(self, *args: str) -> list[str]:
        cmd_args = [] 
        if self.socket_name:
            cmd_args += ["-L", self.socket_name]
        cmd_args += args
        return await run_tmux_command(cmd_args)

Using this async API, multiple tmux commands can run concurrently without blocking:

sessions_task = asyncio.create_task(server.cmd_async("list-sessions"))
new_session_task = asyncio.create_task(server.cmd_async("new-session", "-s", "test"))
await asyncio.gather(sessions_task, new_session_task)

Pros:

• Simple, idiomatic asyncio code
• High-level concurrency

Cons:

• May require new async methods, potentially breaking backward compatibility
• Process creation overhead for frequent commands

---

Approach 2: Async Wrappers using Threads or Process Pools

Maintain existing synchronous implementations, adding asynchronous wrappers via threads (asyncio.to_thread) or executors:

Example (Pane.send_keys adapted to async):

class Pane:
    def send_keys(self, keys: str):
        # blocking call that runs `tmux send-keys`
        ...

    async def asend_keys(self, keys: str):
        await asyncio.to_thread(self.send_keys, keys)

Now, await pane.asend_keys("ls -la") executes asynchronously, without blocking the event loop.

Pros:

• Easy, incremental implementation
• Preserves backward compatibility
• Low risk; uses existing code

Cons:

• Thread overhead (usually minimal for tmux)
• Potential thread-safety issues if shared state is mutated

---

Approach 3: Persistent Tmux Session with Async I/O (Control Mode)

Use tmux’s control mode (tmux -C) to maintain a persistent process for asynchronous command execution:

proc = await asyncio.create_subprocess_exec(
    "tmux", "-C", "attach-session", "-t", "my_session",
    stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE
)

# Async send command example
proc.stdin.write(b"list-windows -t my_session\n")
await proc.stdin.drain()
response = await proc.stdout.readline()
print("Tmux replied:", response.decode().strip())

Pros:

• Minimal overhead per command after initial setup
• Enables high-throughput and event-driven communication

Cons:

• Most complex implementation
• Requires robust parsing and command-response correlation logic
• Higher maintenance burden

---

Trade-Offs Between Approaches

Criterion	Asyncio subprocess	Threads/Executors	Persistent Control Mode
Performance	Good, minor overhead per command	Good, small threading overhead	Excellent, minimal overhead
Concurrency	High, native asyncio concurrency	High, via thread pools	Moderate, sequential commands per connection
Ease of implementation	Moderate, refactoring to async	Easiest, minimal refactoring	Complex, requires full async protocol handling
Compatibility with existing API	Moderate, introduces async-only methods	Excellent, preserves existing sync methods	Low, significant structural changes
Maintenance burden	Moderate, clearer async model	Low, minimal additional complexity	High, protocol handling and async streams

---

Recommendation

A combined approach might be ideal:

• Threads/Executors (Approach 2): Quick win for async compatibility with minimal effort and backward compatibility. Good starting point for incremental asyncio support.
• Asyncio Subprocess (Approach 1): Longer-term solution providing clean, idiomatic async APIs, facilitating future concurrency benefits.
• Persistent Control Mode (Approach 3): Ideal if high-performance or real-time event handling is required, albeit with significantly more complexity and maintenance overhead.

Each option involves trade-offs between performance, implementation complexity, and compatibility. Selection depends on libtmux’s roadmap priorities.

---

Sources

• Python asyncio subprocess documentation:
  https://docs.python.org/3/library/asyncio-subprocess.html

• Running blocking functions asynchronously with executors:
  https://docs.python.org/3/library/asyncio-eventloop.html#running-in-executor

• Tmux control mode documentation:
  https://github.com/tmux/tmux/wiki/Control-Mode

• Current synchronous implementation in libtmux:
  https://github.com/tmux-python/libtmux

See also

• pytest-asyncio incompatibility
  • pytest-asyncio incompatibility git-pull/gp-libs#43
  • Not compatible with --doctest-modules --doctest-ignore-import-errors pytest-dev/pytest-asyncio#797

Summary by Sourcery

Introduce asynchronous support for libtmux using asyncio.

New Features:

• Add AsyncTmuxCmd to run tmux commands asynchronously.
• Add acmd method to Server, Session, Window, and Pane objects for asynchronous command execution.

Tests:

• Add tests for asynchronous operations.

[codecov]

Codecov Report

Attention: Patch coverage is 64.61538% with 23 lines in your changes missing coverage. Please review.

Project coverage is 81.02%. Comparing base (7db6426) to head (327b098).

Files with missing lines	Patch %	Lines
src/libtmux/server.py	35.29%	7 Missing and 4 partials ⚠️
src/libtmux/common.py	73.52%	7 Missing and 2 partials ⚠️
src/libtmux/pane.py	80.00%	0 Missing and 1 partial ⚠️
src/libtmux/session.py	75.00%	0 Missing and 1 partial ⚠️
src/libtmux/window.py	80.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #554      +/-   ##
==========================================
- Coverage   81.39%   81.02%   -0.38%     
==========================================
  Files          37       37              
  Lines        2430     2493      +63     
  Branches      368      380      +12     
==========================================
+ Hits         1978     2020      +42     
- Misses        310      322      +12     
- Partials      142      151       +9     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[tony]

@sourcery-ai review

[sourcery-ai]

Reviewer's Guide by Sourcery

This pull request introduces asynchronous functionality to libtmux using Python's asyncio library. It adds a new AsyncTmuxCmd class for running tmux commands asynchronously and integrates it into the existing Server, Session, Window, and Pane classes. It also updates the project dependencies and adds asynchronous tests.

Sequence diagram for async tmux command execution

sequenceDiagram
    participant C as Client Code
    participant A as AsyncTmuxCmd
    participant S as Subprocess
    participant T as tmux

    C->>+A: run(*args)
    A->>A: validate tmux binary
    A->>+S: create_subprocess_exec
    S->>+T: execute command
    T-->>-S: command output
    S-->>-A: stdout, stderr, returncode
    A->>A: process output
    A-->>-C: AsyncTmuxCmd instance

Loading

Class diagram for new AsyncTmuxCmd and its integration

classDiagram
    class AsyncTmuxCmd {
        +list[str] cmd
        +list[str] stdout
        +list[str] stderr
        +int returncode
        +__init__(cmd, stdout, stderr, returncode)
        +run(*args) AsyncTmuxCmd
    }

    class Server {
        +cmd(cmd, *args, target)
        +acmd(cmd, *args, target) AsyncTmuxCmd
    }

    class Session {
        +cmd(cmd, *args, target)
        +acmd(cmd, *args, target) AsyncTmuxCmd
    }

    class Window {
        +cmd(cmd, *args, target)
        +acmd(cmd, *args, target) AsyncTmuxCmd
    }

    class Pane {
        +cmd(cmd, *args, target)
        +acmd(cmd, *args, target) AsyncTmuxCmd
    }

    Server --> AsyncTmuxCmd : uses
    Session --> Server : uses
    Window --> Server : uses
    Pane --> Server : uses

    note for AsyncTmuxCmd "New async class for tmux commands"
    note for Server "Added async support via acmd"

Loading

File-Level Changes

Change	Details	Files
Added AsyncTmuxCmd class for asynchronous command execution.	Introduced the AsyncTmuxCmd class to encapsulate asynchronous execution of tmux commands. Implemented the run method to handle subprocess creation and communication. Added utility functions for string conversion and logging. Included a workaround for the tmux has-session command behavior.	src/libtmux/common.py
Integrated asynchronous command execution into core classes.	Added acmd methods to the Server, Session, Window, and Pane classes for asynchronous command execution. Updated existing methods to use the new asynchronous functionality. Added examples and documentation for the new asynchronous methods.	src/libtmux/server.py src/libtmux/session.py src/libtmux/window.py src/libtmux/pane.py
Updated project dependencies and added tests.	Added pytest-asyncio to the project dependencies for asynchronous testing. Created a new test file tests/test_async.py with asynchronous test cases. Updated the lock file to reflect the dependency changes.	pyproject.toml tests/test_async.py
Updated documentation.	Added documentation for the new AsyncTmuxCmd class and its usage. Updated the documentation for the Server class to include the new asynchronous methods.	src/libtmux/common.py src/libtmux/server.py

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time. You can also use
  this command to specify where the summary should be inserted.

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

Hey @tony - I've reviewed your changes - here's some feedback:

Overall Comments:

• Consider adding more test cases to cover error conditions and different tmux command scenarios in test_async.py

Here's what I looked at during the review

• 🟢 General issues: all looks good
• 🟢 Security: all looks good
• 🟡 Testing: 2 issues found
• 🟢 Complexity: all looks good
• 🟢 Documentation: all looks good

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

suggestion (testing): Test more aspects of the asynchronous functionality.

The current test only verifies the type of the returned session object. It would be beneficial to interact with the session asynchronously, for example, by creating a window or pane, executing a command within the session, and then verifying the results. This would provide more comprehensive coverage of the asynchronous API.

Suggested implementation:

@pytest.mark.asyncio
async def test_asyncio(server: Server) -> None:
    """Test basic asyncio usage."""
    # Create new session
    result = await server.acmd("new-session", "-d", "-P", "-F#{session_id}")
    session_id = result.stdout[0]
    session = Session.from_session_id(
        session_id=session_id,
        server=server,
    )
    assert isinstance(session, Session)

    # Create new window
    window = await session.new_window(window_name="test_window")
    assert window.name == "test_window"

    # Create new pane and execute command
    pane = await window.split_window()
    await pane.send_keys("echo 'Hello, tmux!'")

    # Allow time for command execution
    await asyncio.sleep(0.5)

    # Capture and verify output
    output = await pane.capture_pane()
    assert "Hello, tmux!" in '\n'.join(output)

    # Clean up
    await window.kill_window()

You may need to:

1. Import asyncio if it's not already imported
2. Adjust the sleep duration (0.5s) based on your system's performance
3. Ensure the Session class has the async methods new_window(), and Window class has split_window() and kill_window()
4. Ensure Pane class has send_keys() and capture_pane() async methods

[sourcery-ai]

suggestion (testing): Add tests for error handling.

Include tests to verify the behavior of the acmd function when the tmux command fails, for example, by trying to create a session with an already existing name. Check that the appropriate exceptions are raised and handled correctly.

Suggested implementation:

@pytest.mark.asyncio
async def test_asyncio(server: Server) -> None:
    """Test basic asyncio usage."""
    result = await server.acmd("new-session", "-d", "-P", "-F#{session_id}")
    session_id = result.stdout[0]
    session = Session.from_session_id(
        session_id=session_id,
        server=server,
    )
    assert isinstance(session, Session)


@pytest.mark.asyncio
async def test_asyncio_duplicate_session(server: Server) -> None:
    """Test error handling when creating duplicate sessions."""
    # Create first session
    session_name = "test_duplicate"
    await server.acmd("new-session", "-d", "-s", session_name)

    # Attempt to create second session with same name
    with pytest.raises(LibTmuxException) as excinfo:
        await server.acmd("new-session", "-d", "-s", session_name)

    assert "duplicate session" in str(excinfo.value).lower()


@pytest.mark.asyncio
async def test_asyncio_invalid_command(server: Server) -> None:
    """Test error handling with invalid tmux commands."""
    with pytest.raises(LibTmuxException) as excinfo:
        await server.acmd("invalid-command")

    assert "unknown command" in str(excinfo.value).lower()

You'll need to:

1. Import LibTmuxException if not already imported (likely from libtmux.exc)
2. Ensure any existing sessions are cleaned up in test teardown to prevent test interference