Rust backend

.github/workflows/tests.yml

@@ -12,6 +12,11 @@ jobs:
       matrix:
         python-version: ['3.14']
         tmux-version: ['3.2a', '3.3a', '3.4', '3.5', '3.6', 'master']
+        backend: ['default']
+        include:
+          - python-version: '3.14'
+            tmux-version: '3.6'
+            backend: 'rust-control'
     steps:
       - uses: actions/checkout@v6
 
@@ -78,6 +83,24 @@ jobs:
           export PATH=$HOME/tmux-builds/tmux-${{ matrix.tmux-version }}/bin:$PATH
           ls $HOME/tmux-builds/tmux-${{ matrix.tmux-version }}/bin
           tmux -V
+          backend="${{ matrix.backend }}"
+          if [ "$backend" != "default" ]; then
+            if ! uv run python - <<'PY'
+import importlib.util
+import sys
+sys.exit(0 if importlib.util.find_spec("vibe_tmux") else 1)
+PY
+            then
+              echo "vibe_tmux not installed; skipping rust backend tests"
+              exit 0
+            fi
+            export LIBTMUX_BACKEND=rust
+            export LIBTMUX_RUST_CONNECTION_KIND=protocol
+            if [ "$backend" = "rust-control" ]; then
+              export LIBTMUX_RUST_CONTROL_MODE=1
+              export LIBTMUX_RUST_CONTROL_AUTOSTART=1
+            fi
+          fi
           uv run py.test --cov=./ --cov-append --cov-report=xml -n auto --verbose
         env:
           COV_CORE_SOURCE: .

conftest.py

@@ -16,6 +16,7 @@
 import pytest
 from _pytest.doctest import DoctestItem
 
+from libtmux._internal import trace as libtmux_trace
 from libtmux.pane import Pane
 from libtmux.pytest_plugin import USING_ZSH
 from libtmux.server import Server
@@ -28,6 +29,34 @@
 pytest_plugins = ["pytester"]
 
 
+@pytest.fixture(autouse=True, scope="session")
+def trace_session() -> None:
+    """Initialize trace collection when enabled."""
+    if not libtmux_trace.TRACE_ENABLED:
+        return
+    if libtmux_trace.TRACE_RESET:
+        libtmux_trace.reset_trace()
+
+
+def pytest_terminal_summary(terminalreporter: pytest.TerminalReporter) -> None:
+    """Print trace summary in pytest's terminal summary."""
+    if not libtmux_trace.TRACE_ENABLED:
+        return
+    terminalreporter.section("libtmux trace")
+    terminalreporter.write_line(libtmux_trace.summarize())
+
+
+@pytest.fixture(autouse=True)
+def trace_test_context(request: pytest.FixtureRequest) -> t.Iterator[None]:
+    """Attach the current pytest node id to trace events."""
+    if not libtmux_trace.TRACE_ENABLED:
+        yield
+        return
+    libtmux_trace.set_test_context(request.node.nodeid)
+    yield
+    libtmux_trace.set_test_context(None)
+
+
 @pytest.fixture(autouse=True)
 def add_doctest_fixtures(
     request: pytest.FixtureRequest,

pyproject.toml

@@ -80,6 +80,12 @@ dev = [
   "mypy",
 ]
 
+otel = [
+  "opentelemetry-api>=1.20.0",
+  "opentelemetry-sdk>=1.20.0",
+  "opentelemetry-exporter-otlp>=1.20.0",
+]
+
 docs = [
   "sphinx<9",
   "furo",

src/libtmux/_internal/trace.py

@@ -0,0 +1,147 @@
+"""Lightweight tracing for libtmux timing audits."""
+
+from __future__ import annotations
+
+import contextlib
+import contextvars
+import itertools
+import json
+import os
+import pathlib
+import threading
+import time
+import typing as t
+
+TRACE_PATH = os.getenv("LIBTMUX_TRACE_PATH", "/tmp/libtmux-trace.jsonl")
+
+
+def _env_flag(name: str) -> bool:
+    value = os.getenv(name)
+    if value is None:
+        return False
+    return value not in {"", "0", "false", "False", "no", "NO"}
+
+
+TRACE_ENABLED = _env_flag("LIBTMUX_TRACE")
+TRACE_RESET = _env_flag("LIBTMUX_TRACE_RESET")
+
+_TRACE_TEST: contextvars.ContextVar[str | None] = contextvars.ContextVar(
+    "libtmux_trace_test", default=None
+)
+_TRACE_STACK: contextvars.ContextVar[tuple[int, ...]] = contextvars.ContextVar(
+    "libtmux_trace_stack", default=()
+)
+_TRACE_COUNTER = itertools.count(1)
+
+
+def set_test_context(name: str | None) -> None:
+    _TRACE_TEST.set(name)
+
+
+def reset_trace(path: str | None = None) -> None:
+    if not TRACE_ENABLED:
+        return
+    target = path or TRACE_PATH
+    with pathlib.Path(target).open("w", encoding="utf-8") as handle:
+        handle.write("")
+
+
+def _write_event(event: dict[str, t.Any]) -> None:
+    if not TRACE_ENABLED:
+        return
+    event["pid"] = os.getpid()
+    event["thread"] = threading.get_ident()
+    test_name = _TRACE_TEST.get()
+    if test_name:
+        event["test"] = test_name
+    with pathlib.Path(TRACE_PATH).open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(event, sort_keys=False))
+        handle.write("\n")
+
+
+@contextlib.contextmanager
+def span(name: str, **fields: t.Any) -> t.Iterator[None]:
+    if not TRACE_ENABLED:
+        yield
+        return
+    span_id = next(_TRACE_COUNTER)
+    stack = _TRACE_STACK.get()
+    parent_id = stack[-1] if stack else None
+    _TRACE_STACK.set((*stack, span_id))
+    start_ns = time.perf_counter_ns()
+    try:
+        yield
+    finally:
+        duration_ns = time.perf_counter_ns() - start_ns
+        _TRACE_STACK.set(stack)
+        event = {
+            "event": name,
+            "span_id": span_id,
+            "parent_id": parent_id,
+            "depth": len(stack),
+            "start_ns": start_ns,
+            "duration_ns": duration_ns,
+        }
+        event.update(fields)
+        _write_event(event)
+
+
+def point(name: str, **fields: t.Any) -> None:
+    if not TRACE_ENABLED:
+        return
+    event = {"event": name, "point": True, "ts_ns": time.perf_counter_ns()}
+    event.update(fields)
+    _write_event(event)
+
+
+def summarize(path: str | None = None, limit: int = 20) -> str:
+    target = path or TRACE_PATH
+    if not pathlib.Path(target).exists():
+        return "libtmux trace: no data collected"
+
+    totals: dict[str, dict[str, int]] = {}
+    slowest: list[tuple[int, str]] = []
+
+    with pathlib.Path(target).open(encoding="utf-8") as handle:
+        for line in handle:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                event = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            if event.get("point"):
+                continue
+            name = str(event.get("event", "unknown"))
+            duration = int(event.get("duration_ns", 0))
+            entry = totals.setdefault(name, {"count": 0, "total_ns": 0, "max_ns": 0})
+            entry["count"] += 1
+            entry["total_ns"] += duration
+            if duration > entry["max_ns"]:
+                entry["max_ns"] = duration
+            slowest.append((duration, json.dumps(event, sort_keys=False)))
+
+    if not totals:
+        return "libtmux trace: no span data collected"
+
+    sorted_totals = sorted(
+        totals.items(), key=lambda item: item[1]["total_ns"], reverse=True
+    )
+    sorted_slowest = sorted(slowest, key=lambda item: item[0], reverse=True)[:limit]
+
+    lines = ["libtmux trace summary (ns):"]
+    for name, stats in sorted_totals[:limit]:
+        avg = stats["total_ns"] // max(stats["count"], 1)
+        lines.append(
+            f"- {name}: count={stats['count']} total={stats['total_ns']} avg={avg} "
+            f"max={stats['max_ns']}"
+        )
+    lines.append("libtmux trace slowest spans:")
+    for duration, payload in sorted_slowest:
+        lines.append(f"- {duration} {payload}")
+    return "\n".join(lines)
+
+
+if TRACE_ENABLED and TRACE_RESET:
+    reset_trace()

src/libtmux/_rust.py

@@ -0,0 +1,46 @@
+"""Rust-backed libtmux bindings.
+
+This module is intentionally thin: it re-exports Rust types from the
+vibe-tmux extension so libtmux can opt into the Rust backend.
+"""
+
+from __future__ import annotations
+
+import importlib
+from typing import Any
+
+_EXPORTS = ("Server",)
+_NATIVE: Any | None = None
+
+
+class RustBackendImportError(ImportError):
+    """Raise when the Rust backend cannot be imported."""
+
+    def __init__(self) -> None:
+        super().__init__(
+            "libtmux rust backend requires the vibe_tmux extension to be installed"
+        )
+
+
+def _load_native() -> Any:
+    global _NATIVE
+    if _NATIVE is None:
+        try:
+            _NATIVE = importlib.import_module("vibe_tmux")
+        except Exception as exc:  # pragma: no cover - import path is env-dependent
+            raise RustBackendImportError() from exc
+    return _NATIVE
+
+
+def __getattr__(name: str) -> Any:
+    if name in _EXPORTS:
+        native = _load_native()
+        return getattr(native, name)
+    raise AttributeError(name)
+
+
+def __dir__() -> list[str]:
+    return sorted(list(globals().keys()) + list(_EXPORTS))
+
+
+__all__ = _EXPORTS