diff --git a/CHANGES b/CHANGES index f7ba0efba..7abdff9b1 100644 --- a/CHANGES +++ b/CHANGES @@ -45,6 +45,27 @@ $ uvx --from 'libtmux' --prerelease allow python _Notes on the upcoming release will go here._ +libtmux 0.62.x adds pure Python workspace archive helpers inspired by +tmux-resurrect and tmux-continuum. Projects can capture, restore, rotate, +autosave, and startup-restore tmux workspaces through libtmux APIs without +installing TPM plugins. + +### What's new + +#### Headless workspace archives and autosave (#701, #702) + +The new {mod}`libtmux.resurrect` package captures tmux workspaces into typed +JSON archives and restores them without shelling through tmux plugin scripts. +Archives include session, window, pane, layout, focus, grouped-session, +pane-title, zoom, and `automatic-rename` state, with idempotent reuse behavior +for existing sessions. + +The package also includes conservative process restore policies, optional full +process command capture providers, tmux-resurrect tab-file import/export, +timestamped snapshot rotation with a portable `last.json` pointer, and +tmux-continuum-style autosave and startup-restore guards. See +{doc}`topics/resurrect` for usage. + ## libtmux 0.61.0 (2026-07-04) libtmux 0.61.0 hardens support for the tmux 3.7 patch line. It fixes diff --git a/docs/api/index.md b/docs/api/index.md index 23cd9043b..57569705c 100644 --- a/docs/api/index.md +++ b/docs/api/index.md @@ -120,6 +120,12 @@ Format strings and constants. Exception hierarchy. ::: +:::{grid-item-card} Resurrect +:link: libtmux.resurrect +:link-type: doc +Headless workspace archive, restore, autosave, and startup helpers. +::: + :::: ## Testing @@ -178,4 +184,5 @@ Options Hooks Constants Exceptions +Resurrect ``` diff --git a/docs/api/libtmux.resurrect.md b/docs/api/libtmux.resurrect.md new file mode 100644 index 000000000..4e47717b4 --- /dev/null +++ b/docs/api/libtmux.resurrect.md @@ -0,0 +1,21 @@ +# Resurrect + +```{eval-rst} +.. automodule:: libtmux.resurrect + :members: + +.. automodule:: libtmux.resurrect.archives + :members: + +.. automodule:: libtmux.resurrect.continuum + :members: + +.. automodule:: libtmux.resurrect.processes + :members: + +.. automodule:: libtmux.resurrect.resurrect_file + :members: + +.. automodule:: libtmux.resurrect.storage + :members: +``` diff --git a/docs/topics/index.md b/docs/topics/index.md index cf9ac2c42..718b722f1 100644 --- a/docs/topics/index.md +++ b/docs/topics/index.md @@ -42,6 +42,12 @@ Create and position floating (overlay) panes on tmux 3.7+. Create sessions, windows, and panes programmatically. ::: +:::{grid-item-card} Workspace Archives +:link: resurrect +:link-type: doc +Capture, restore, rotate, and autosave tmux workspaces headlessly. +::: + :::{grid-item-card} Automation Patterns :link: automation_patterns :link-type: doc @@ -86,6 +92,7 @@ filtering pane_interaction floating_panes workspace_setup +resurrect automation_patterns context_managers options_and_hooks diff --git a/docs/topics/resurrect.md b/docs/topics/resurrect.md new file mode 100644 index 000000000..9ac61421f --- /dev/null +++ b/docs/topics/resurrect.md @@ -0,0 +1,183 @@ +# Workspace Archives + +libtmux includes pure Python helpers for capturing and restoring tmux +workspaces without installing tmux plugins. The API is modeled on +tmux-resurrect and tmux-continuum, but it uses libtmux calls and JSON archives +so it can run in tests, background jobs, and MCP servers. + +## Capture and Restore + +Capture the current server into a typed archive: + +```python +from pathlib import Path + +from libtmux import Server +from libtmux.resurrect import capture_archive, write_archive + +server = Server() +archive = capture_archive(server) +write_archive(archive, Path("workspace.json")) +``` + +Restore an archive into a fresh or reusable server: + +```python +from pathlib import Path + +from libtmux import Server +from libtmux.resurrect import restore_archive + +server = Server() +restore_archive(Path("workspace.json"), server, on_exists="reuse") +``` + +Archives preserve sessions, windows, panes, working directories, layouts, +active panes, active and alternate windows, grouped sessions, pane titles, +zoom flags, `automatic-rename`, and attached-client session focus when tmux can +replay it. + +## Process Commands + +By default, process restore uses a conservative tmux-resurrect-style allow-list +for interactive commands such as `vim`, `less`, `tail`, and `top`. Shell panes +are recreated at their saved working directory without replaying a command. + +Use a policy when you want to add commands or restore everything: + +```python +from pathlib import Path + +from libtmux import Server +from libtmux.resurrect import ProcessRestorePolicy, restore_archive + +server = Server() +policy = ProcessRestorePolicy.from_options("'python->uv run python *' 'git log'") +restore_archive(Path("workspace.json"), server, process_policy=policy) +``` + +Full command capture is explicit because command lines can contain sensitive +arguments. Pass a provider when you want to save process arguments: + +```python +from libtmux import Server +from libtmux.resurrect import capture_archive, default_process_command_provider + +server = Server() +archive = capture_archive(server, process_provider=default_process_command_provider()) +``` + +The default provider reads Linux procfs and leaves `full_command` empty when +procfs cannot resolve the foreground process. Use +{class}`~libtmux.resurrect.PsProcessCommandProvider` explicitly when you want a +`ps`-based provider. + +## tmux-resurrect Files + +Existing tmux-resurrect save files can be imported without running TPM or the +plugin scripts: + +```python +from pathlib import Path + +from libtmux.resurrect import archive_from_resurrect_file, write_archive + +archive = archive_from_resurrect_file(Path("last").read_text(encoding="utf-8")) +write_archive(archive, Path("workspace.json")) +``` + +You can also export a libtmux archive back to tmux-resurrect tab rows: + +```python +from pathlib import Path + +from libtmux.resurrect import archive_to_resurrect_file, read_archive + +archive = read_archive(Path("workspace.json")) +Path("last").write_text(archive_to_resurrect_file(archive), encoding="utf-8") +``` + +## Autosave and Rotation + +Autosave helpers provide tmux-continuum-style interval checks and socket-aware +paths. Snapshot storage adds timestamped archives, retention rotation, and a +`last.json` pointer: + +```python +from pathlib import Path + +from libtmux import Server +from libtmux.resurrect import autosave_once, default_autosave_paths + +server = Server(socket_name="main") +paths = default_autosave_paths(server, Path("archives")) +autosave_once(server, archive_path=paths.archive_path, state_path=paths.state_path) +``` + +```python +from pathlib import Path + +from libtmux.resurrect import capture_archive, write_archive_snapshot + +archive = capture_archive(server) +write_archive_snapshot(archive, Path("archives"), keep=10, portable_last=True) +``` + +Use `portable_last=True` when a filesystem or operating system does not allow +symlink creation. Without it, libtmux writes a relative symlink when possible +and falls back to a copy when symlink creation fails. + +## Startup Restore + +Startup restore is split into a decision helper and a one-shot restore helper. +This makes downstream service wrappers explicit about when restore is allowed: + +```python +from pathlib import Path + +from libtmux import Server +from libtmux.resurrect import startup_restore_once + +server = Server(socket_name="main") +result = startup_restore_once( + server, + Path("archives/last.json"), + enabled=True, + halt_file=Path("archives/restore-halt"), + another_server_running=False, +) +``` + +The helper skips restore when disabled, when a halt file exists, when sessions +already exist, when another server owns the restore window, or when the startup +grace period has elapsed. `result.reason` contains a stable reason string for +service logs. + +For a systemd user service, run a small Python wrapper that calls +`startup_restore_once()` after starting tmux: + +```ini +[Unit] +Description=Restore tmux workspace + +[Service] +Type=oneshot +ExecStart=python -m your_restore_module + +[Install] +WantedBy=default.target +``` + +For launchd, use the same wrapper from a `ProgramArguments` entry: + +```xml +ProgramArguments + + python + -m + your_restore_module + +``` + +The wrapper should compute `another_server_running` using the service manager +or deployment environment that owns tmux startup. diff --git a/src/libtmux/resurrect/__init__.py b/src/libtmux/resurrect/__init__.py new file mode 100644 index 000000000..f2446f10f --- /dev/null +++ b/src/libtmux/resurrect/__init__.py @@ -0,0 +1,101 @@ +"""Headless tmux workspace archive helpers.""" + +from __future__ import annotations + +from .archives import ( + CAPTURED_CAPABILITIES, + DEFAULT_SHELL_COMMANDS, + FORMAT_VERSION, + PaneArchive, + RestorePolicy, + SessionArchive, + WindowArchive, + WorkspaceArchive, + capture_archive, + read_archive, + restore_archive, + write_archive, +) +from .continuum import ( + DEFAULT_AUTOSAVE_INTERVAL, + DEFAULT_STARTUP_RESTORE_GRACE, + STATE_FORMAT_VERSION, + AutosavePaths, + AutosaveResult, + AutosaveState, + StartupRestoreDecision, + StartupRestoreResult, + autosave_once, + default_autosave_paths, + next_autosave_at, + read_autosave_state, + should_autosave, + should_restore_on_startup, + startup_restore_once, + write_autosave_state, +) +from .processes import ( + DEFAULT_PROCESS_RESTORE_POLICY, + DEFAULT_RESTORE_PROGRAMS, + CompositeProcessCommandProvider, + ProcessCommandProvider, + ProcessRestorePolicy, + ProcessRestoreRule, + ProcfsProcessCommandProvider, + PsProcessCommandProvider, + default_process_command_provider, +) +from .resurrect_file import ( + archive_from_resurrect_file, + archive_to_resurrect_file, +) +from .storage import ( + ArchiveSnapshot, + LastPointerKind, + write_archive_snapshot, +) + +__all__ = ( + "CAPTURED_CAPABILITIES", + "DEFAULT_AUTOSAVE_INTERVAL", + "DEFAULT_PROCESS_RESTORE_POLICY", + "DEFAULT_RESTORE_PROGRAMS", + "DEFAULT_SHELL_COMMANDS", + "DEFAULT_STARTUP_RESTORE_GRACE", + "FORMAT_VERSION", + "STATE_FORMAT_VERSION", + "ArchiveSnapshot", + "AutosavePaths", + "AutosaveResult", + "AutosaveState", + "CompositeProcessCommandProvider", + "LastPointerKind", + "PaneArchive", + "ProcessCommandProvider", + "ProcessRestorePolicy", + "ProcessRestoreRule", + "ProcfsProcessCommandProvider", + "PsProcessCommandProvider", + "RestorePolicy", + "SessionArchive", + "StartupRestoreDecision", + "StartupRestoreResult", + "WindowArchive", + "WorkspaceArchive", + "archive_from_resurrect_file", + "archive_to_resurrect_file", + "autosave_once", + "capture_archive", + "default_autosave_paths", + "default_process_command_provider", + "next_autosave_at", + "read_archive", + "read_autosave_state", + "restore_archive", + "should_autosave", + "should_restore_on_startup", + "startup_restore_once", + "write_archive", + "write_archive_snapshot", + "write_autosave_state", +) diff --git a/src/libtmux/resurrect/archives.py b/src/libtmux/resurrect/archives.py new file mode 100644 index 000000000..0c003957d --- /dev/null +++ b/src/libtmux/resurrect/archives.py @@ -0,0 +1,1289 @@ +"""Pure Python tmux workspace archive helpers. + +This module intentionally does not depend on tmux plugin manager state. It +captures tmux state through the libtmux ``Server`` API and stores a typed JSON +archive that can be restored headlessly. +""" + +from __future__ import annotations + +import datetime +import json +import os +import pathlib +import typing as t +from dataclasses import dataclass + +from libtmux._internal.types import StrPath +from libtmux.common import raise_if_stderr +from libtmux.formats import FORMAT_SEPARATOR +from libtmux.resurrect.processes import ( + DEFAULT_PROCESS_RESTORE_POLICY, + ProcessCommandProvider, + ProcessRestorePolicy, +) + +if t.TYPE_CHECKING: + from libtmux.server import Server + from libtmux.session import Session + from libtmux.window import Window + +FORMAT_VERSION = "libtmux.resurrect.archive.v1" +"""Archive format identifier.""" + +CAPTURED_CAPABILITIES = ( + "sessions", + "windows", + "panes", + "window-order", + "pane-order", + "working-directories", + "layouts", + "active-windows", + "active-panes", + "pane-current-command", + "pane-full-command", + "pane-titles", + "window-flags", + "automatic-rename", + "grouped-sessions", + "alternate-windows", + "active-sessions", + "alternate-sessions", + "history-size", +) +"""tmux-resurrect parity features captured by :func:`capture_archive`.""" + +DEFAULT_SHELL_COMMANDS = frozenset({"bash", "dash", "fish", "ksh", "sh", "zsh"}) +"""Commands treated as shells during restore.""" + +RestorePolicy: t.TypeAlias = t.Literal["error", "replace", "reuse"] +"""How restore handles sessions that already exist.""" + +_FIELD_SEPARATOR = FORMAT_SEPARATOR +_PANE_FIELDS = ( + "#{session_name}", + "#{window_index}", + "#{window_name}", + "#{window_layout}", + "#{window_active}", + "#{window_flags}", + "#{pane_index}", + "#{pane_active}", + "#{pane_pid}", + "#{pane_current_command}", + "#{pane_current_path}", + "#{pane_title}", + "#{history_size}", +) +_PANE_FORMAT = "".join(f"{field}{_FIELD_SEPARATOR}" for field in _PANE_FIELDS) +_SESSION_FORMAT = "".join( + f"{field}{_FIELD_SEPARATOR}" + for field in ( + "#{session_name}", + "#{session_grouped}", + "#{session_group}", + ) +) +_CLIENT_FORMAT = "".join( + f"{field}{_FIELD_SEPARATOR}" + for field in ( + "#{client_session}", + "#{client_last_session}", + ) +) + + +@dataclass(frozen=True, slots=True) +class PaneArchive: + """Serialized tmux pane state.""" + + index: int + active: bool + current_command: str + current_path: str + title: str = "" + full_command: str = "" + history_size: int = 0 + contents: tuple[str, ...] = () + + +@dataclass(frozen=True, slots=True) +class WindowArchive: + """Serialized tmux window state.""" + + index: int + name: str + layout: str + active: bool + panes: tuple[PaneArchive, ...] + flags: str = "" + automatic_rename: bool | None = None + + +@dataclass(frozen=True, slots=True) +class SessionArchive: + """Serialized tmux session state.""" + + name: str + windows: tuple[WindowArchive, ...] + group_name: str | None = None + active_window_index: int | None = None + alternate_window_index: int | None = None + + +@dataclass(frozen=True, slots=True) +class WorkspaceArchive: + """Serialized tmux workspace state.""" + + saved_at: datetime.datetime + sessions: tuple[SessionArchive, ...] + format_version: str = FORMAT_VERSION + capabilities: tuple[str, ...] = CAPTURED_CAPABILITIES + active_session_name: str | None = None + alternate_session_name: str | None = None + + +@dataclass(frozen=True, slots=True) +class _PaneRow: + """Flattened list-panes row before grouping.""" + + session_name: str + window_index: int + window_name: str + window_layout: str + window_active: bool + window_flags: str + pane_index: int + pane_active: bool + pane_pid: int + pane_current_command: str + pane_current_path: str + pane_title: str + history_size: int + + +def capture_archive( + server: Server, + *, + process_provider: ProcessCommandProvider | None = None, + saved_at: datetime.datetime | None = None, +) -> WorkspaceArchive: + """Capture all panes from a tmux server into a workspace archive. + + Examples + -------- + >>> archive = capture_archive(server) + >>> archive.format_version + 'libtmux.resurrect.archive.v1' + """ + proc = server.cmd("list-panes", "-a", "-F", _PANE_FORMAT) + raise_if_stderr(proc, "list-panes") + + rows = tuple(_parse_pane_row(line) for line in proc.stdout) + active_session_name, alternate_session_name = _capture_client_state(server) + return _archive_from_rows( + rows, + active_session_name=active_session_name, + alternate_session_name=alternate_session_name, + automatic_renames=_capture_automatic_renames(server, rows), + process_commands=_capture_process_commands(rows, process_provider), + saved_at=saved_at, + session_groups=_capture_session_groups(server), + ) + + +def write_archive(archive: WorkspaceArchive, path: StrPath) -> pathlib.Path: + """Write an archive as stable JSON using an atomic replace. + + Examples + -------- + >>> import pathlib + >>> target = pathlib.Path(request.getfixturevalue("tmp_path")) / "tmux.json" + >>> saved = write_archive(capture_archive(server), target) + >>> saved == target + True + """ + destination = pathlib.Path(path) + destination.parent.mkdir(parents=True, exist_ok=True) + tmp_path = destination.with_name(f".{destination.name}.{os.getpid()}.tmp") + tmp_path.write_text( + json.dumps(_archive_to_dict(archive), indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + tmp_path.replace(destination) + return destination + + +def read_archive(path: StrPath) -> WorkspaceArchive: + """Read an archive written by :func:`write_archive`. + + Examples + -------- + >>> import pathlib + >>> target = pathlib.Path(request.getfixturevalue("tmp_path")) / "tmux.json" + >>> _ = write_archive(capture_archive(server), target) + >>> read_archive(target).format_version + 'libtmux.resurrect.archive.v1' + """ + source = pathlib.Path(path) + return _archive_from_dict(json.loads(source.read_text(encoding="utf-8"))) + + +def restore_archive( + archive: WorkspaceArchive | StrPath, + server: Server, + *, + on_exists: RestorePolicy = "error", + process_policy: ProcessRestorePolicy | None = DEFAULT_PROCESS_RESTORE_POLICY, + shell_commands: t.Collection[str] = DEFAULT_SHELL_COMMANDS, +) -> list[Session]: + """Restore sessions from an archive into a tmux server. + + Shell panes are recreated at their recorded working directory. Non-shell + panes are started with their recorded command. + + Examples + -------- + >>> archive = capture_archive(server) + >>> restored = restore_archive(archive, server, on_exists="replace") + >>> isinstance(restored, list) + True + """ + if on_exists not in {"error", "replace", "reuse"}: + msg = f"unknown restore policy: {on_exists!r}" + raise ValueError(msg) + + resolved_archive = ( + archive if isinstance(archive, WorkspaceArchive) else read_archive(archive) + ) + sessions: list[Session] = [] + group_representatives = _group_representatives(resolved_archive.sessions) + + for session_archive in _ordered_session_archives( + resolved_archive.sessions, + group_representatives=group_representatives, + ): + group_target = _group_target(session_archive, group_representatives) + if server.has_session(session_archive.name): + if on_exists == "reuse": + _restore_missing_session_topology( + server, + session_archive, + process_policy=process_policy, + shell_commands=shell_commands, + ) + _restore_session_focus(server, session_archive) + continue + if on_exists == "error": + msg = f"session already exists: {session_archive.name}" + raise FileExistsError(msg) + proc = server.cmd("kill-session", target=session_archive.name) + raise_if_stderr(proc, "kill-session") + + if group_target is not None and server.has_session(group_target): + session = _restore_grouped_session( + server, + session_archive, + target_session=group_target, + ) + if session is not None: + sessions.append(session) + continue + + session = _restore_session( + server, + session_archive, + process_policy=process_policy, + shell_commands=shell_commands, + ) + sessions.append(session) + + _restore_workspace_focus(server, resolved_archive) + + return sessions + + +def _archive_from_rows( + rows: tuple[_PaneRow, ...], + *, + active_session_name: str | None = None, + alternate_session_name: str | None = None, + automatic_renames: t.Mapping[tuple[str, int], bool | None] | None = None, + process_commands: t.Mapping[int, str] | None = None, + saved_at: datetime.datetime | None = None, + session_groups: t.Mapping[str, str] | None = None, +) -> WorkspaceArchive: + grouped: dict[str, dict[int, list[_PaneRow]]] = {} + for row in sorted(rows, key=lambda item: (item.session_name, item.window_index)): + grouped.setdefault(row.session_name, {}).setdefault( + row.window_index, [] + ).append( + row, + ) + + sessions: list[SessionArchive] = [] + for session_name, windows_by_index in grouped.items(): + windows: list[WindowArchive] = [] + for window_index, pane_rows in sorted(windows_by_index.items()): + first = pane_rows[0] + panes = tuple( + PaneArchive( + index=row.pane_index, + active=row.pane_active, + current_command=row.pane_current_command, + current_path=row.pane_current_path, + full_command=(process_commands or {}).get(row.pane_pid, ""), + title=row.pane_title, + history_size=row.history_size, + ) + for row in sorted(pane_rows, key=lambda item: item.pane_index) + ) + windows.append( + WindowArchive( + index=window_index, + name=first.window_name, + layout=first.window_layout, + active=first.window_active, + panes=panes, + flags=first.window_flags, + automatic_rename=(automatic_renames or {}).get( + (first.session_name, first.window_index) + ), + ), + ) + active_window_index = next( + (window.index for window in windows if window.active), + None, + ) + alternate_window_index = next( + (window.index for window in windows if "-" in window.flags), + None, + ) + sessions.append( + SessionArchive( + name=session_name, + windows=tuple(windows), + group_name=(session_groups or {}).get(session_name), + active_window_index=active_window_index, + alternate_window_index=alternate_window_index, + ), + ) + + return WorkspaceArchive( + active_session_name=active_session_name, + alternate_session_name=alternate_session_name, + saved_at=_coerce_saved_at(saved_at), + sessions=tuple(sessions), + ) + + +def _restore_session( + server: Server, + session_archive: SessionArchive, + *, + process_policy: ProcessRestorePolicy | None, + shell_commands: t.Collection[str], +) -> Session: + first_window = session_archive.windows[0] if session_archive.windows else None + first_pane = first_window.panes[0] if first_window and first_window.panes else None + + session = server.new_session( + session_name=session_archive.name, + start_directory=_pane_path(first_pane), + window_name=_name_or_none(first_window.name) if first_window else None, + window_command=_pane_command( + first_pane, + process_policy=process_policy, + shell_commands=shell_commands, + ), + ) + + if first_window is not None: + active_window = session.active_window + _move_initial_window( + server, + session_archive=session_archive, + window=active_window, + window_archive=first_window, + ) + _restore_window( + server, + session_archive=session_archive, + window=active_window, + window_archive=first_window, + process_policy=process_policy, + shell_commands=shell_commands, + skip_first_pane=True, + ) + + for window_archive in session_archive.windows[1:]: + first_window_pane = window_archive.panes[0] if window_archive.panes else None + window = session.new_window( + window_name=_name_or_none(window_archive.name), + start_directory=_pane_path(first_window_pane), + window_index=str(window_archive.index), + window_shell=_pane_command( + first_window_pane, + process_policy=process_policy, + shell_commands=shell_commands, + ), + attach=False, + ) + _restore_window( + server, + session_archive=session_archive, + window=window, + window_archive=window_archive, + process_policy=process_policy, + shell_commands=shell_commands, + skip_first_pane=True, + ) + + _restore_session_focus(server, session_archive, session=session) + + return session + + +def _restore_grouped_session( + server: Server, + session_archive: SessionArchive, + *, + target_session: str, +) -> Session | None: + proc = server.cmd( + "new-session", + "-d", + "-s", + session_archive.name, + "-t", + target_session, + ) + raise_if_stderr(proc, "new-session") + _restore_session_focus(server, session_archive) + return _resolve_session(server, session_archive.name) + + +def _restore_missing_session_topology( + server: Server, + session_archive: SessionArchive, + *, + process_policy: ProcessRestorePolicy | None, + shell_commands: t.Collection[str], +) -> None: + existing_windows = _existing_window_indexes(server, session_archive.name) + for window_archive in session_archive.windows: + if window_archive.index not in existing_windows: + _create_missing_window( + server, + session_archive=session_archive, + window_archive=window_archive, + process_policy=process_policy, + shell_commands=shell_commands, + ) + for pane_archive in window_archive.panes[1:]: + _create_missing_pane( + server, + session_archive=session_archive, + window_archive=window_archive, + pane_archive=pane_archive, + process_policy=process_policy, + shell_commands=shell_commands, + ) + _restore_reused_window_state( + server, + session_archive=session_archive, + window_archive=window_archive, + ) + continue + + existing_panes = _existing_pane_indexes( + server, + session_archive.name, + window_archive.index, + ) + for pane_archive in window_archive.panes: + if pane_archive.index in existing_panes: + continue + _create_missing_pane( + server, + session_archive=session_archive, + window_archive=window_archive, + pane_archive=pane_archive, + process_policy=process_policy, + shell_commands=shell_commands, + ) + _restore_reused_window_state( + server, + session_archive=session_archive, + window_archive=window_archive, + ) + + +def _create_missing_window( + server: Server, + *, + session_archive: SessionArchive, + window_archive: WindowArchive, + process_policy: ProcessRestorePolicy | None, + shell_commands: t.Collection[str], +) -> None: + first_pane = window_archive.panes[0] if window_archive.panes else None + args = [ + "-d", + "-t", + f"{session_archive.name}:{window_archive.index}", + "-n", + window_archive.name, + ] + path = _pane_path(first_pane) + if path is not None: + args.extend(("-c", path)) + command = _pane_command( + first_pane, + process_policy=process_policy, + shell_commands=shell_commands, + ) + if command is not None: + args.append(command) + + proc = server.cmd("new-window", *args) + raise_if_stderr(proc, "new-window") + + +def _create_missing_pane( + server: Server, + *, + session_archive: SessionArchive, + window_archive: WindowArchive, + pane_archive: PaneArchive, + process_policy: ProcessRestorePolicy | None, + shell_commands: t.Collection[str], +) -> None: + args = ["-d", "-t", f"{session_archive.name}:{window_archive.index}"] + path = _path_or_none(pane_archive.current_path) + if path is not None: + args.extend(("-c", path)) + command = _pane_command( + pane_archive, + process_policy=process_policy, + shell_commands=shell_commands, + ) + if command is not None: + args.append(command) + + proc = server.cmd("split-window", *args) + raise_if_stderr(proc, "split-window") + + +def _existing_window_indexes(server: Server, session_name: str) -> set[int]: + proc = server.cmd("list-windows", "-t", session_name, "-F", "#{window_index}") + if proc.stderr: + return set() + return {_tmux_int(line) for line in proc.stdout} + + +def _existing_pane_indexes( + server: Server, + session_name: str, + window_index: int, +) -> set[int]: + proc = server.cmd( + "list-panes", + "-t", + f"{session_name}:{window_index}", + "-F", + "#{pane_index}", + ) + if proc.stderr: + return set() + return {_tmux_int(line) for line in proc.stdout} + + +def _restored_pane_targets( + server: Server, + *, + session_archive: SessionArchive, + window_archive: WindowArchive, +) -> dict[int, str]: + proc = server.cmd( + "list-panes", + "-t", + _target_window(session_archive, window_archive), + "-F", + "#{pane_id}", + ) + if proc.stderr: + return {} + + targets = tuple(line for line in proc.stdout if line.startswith("%")) + if len(targets) != len(window_archive.panes): + return {} + + return { + pane_archive.index: target + for pane_archive, target in zip(window_archive.panes, targets, strict=True) + } + + +def _restore_window( + server: Server, + *, + session_archive: SessionArchive, + window: Window, + window_archive: WindowArchive, + process_policy: ProcessRestorePolicy | None, + shell_commands: t.Collection[str], + skip_first_pane: bool, +) -> None: + pane_archives = ( + window_archive.panes[1:] if skip_first_pane else window_archive.panes + ) + for pane_archive in pane_archives: + window.split( + start_directory=_path_or_none(pane_archive.current_path), + shell=_pane_command( + pane_archive, + process_policy=process_policy, + shell_commands=shell_commands, + ), + attach=pane_archive.active, + ) + + if window_archive.layout: + window.select_layout(window_archive.layout) + + pane_targets = _restored_pane_targets( + server, + session_archive=session_archive, + window_archive=window_archive, + ) + _restore_window_metadata( + server, + session_archive=session_archive, + window_archive=window_archive, + pane_targets=pane_targets, + ) + _restore_active_pane( + server, + session_archive=session_archive, + window_archive=window_archive, + pane_targets=pane_targets, + ) + + +def _restore_reused_window_state( + server: Server, + *, + session_archive: SessionArchive, + window_archive: WindowArchive, +) -> None: + target_window = _target_window(session_archive, window_archive) + if window_archive.layout: + proc = server.cmd("select-layout", "-t", target_window, window_archive.layout) + raise_if_stderr(proc, "select-layout") + + pane_targets = _restored_pane_targets( + server, + session_archive=session_archive, + window_archive=window_archive, + ) + _restore_window_metadata( + server, + session_archive=session_archive, + window_archive=window_archive, + pane_targets=pane_targets, + ) + _restore_active_pane( + server, + session_archive=session_archive, + window_archive=window_archive, + pane_targets=pane_targets, + ) + + +def _restore_window_metadata( + server: Server, + *, + session_archive: SessionArchive, + window_archive: WindowArchive, + pane_targets: t.Mapping[int, str], +) -> None: + target_window = _target_window(session_archive, window_archive) + if window_archive.name: + proc = server.cmd("rename-window", "-t", target_window, window_archive.name) + raise_if_stderr(proc, "rename-window") + + if window_archive.automatic_rename is not None: + proc = server.cmd( + "set-window-option", + "-t", + target_window, + "automatic-rename", + "on" if window_archive.automatic_rename else "off", + ) + raise_if_stderr(proc, "set-window-option") + + for pane_archive in window_archive.panes: + if not pane_archive.title: + continue + proc = server.cmd( + "select-pane", + "-T", + pane_archive.title, + target=_target_pane( + session_archive, + window_archive, + pane_archive, + pane_targets=pane_targets, + ), + ) + raise_if_stderr(proc, "select-pane") + + if "Z" in window_archive.flags: + proc = server.cmd("resize-pane", "-Z", target=target_window) + raise_if_stderr(proc, "resize-pane") + + +def _restore_active_pane( + server: Server, + *, + session_archive: SessionArchive, + window_archive: WindowArchive, + pane_targets: t.Mapping[int, str], +) -> None: + active_pane_archive = _active_pane(window_archive) + if active_pane_archive is not None: + proc = server.cmd( + "select-pane", + target=_target_pane( + session_archive, + window_archive, + active_pane_archive, + pane_targets=pane_targets, + ), + ) + raise_if_stderr(proc, "select-pane") + + +def _restore_session_focus( + server: Server, + session_archive: SessionArchive, + *, + session: Session | None = None, +) -> None: + if session_archive.alternate_window_index is not None: + proc = server.cmd( + "select-window", + "-t", + f"{session_archive.name}:{session_archive.alternate_window_index}", + ) + raise_if_stderr(proc, "select-window") + + active_window_index = session_archive.active_window_index + if active_window_index is None: + active_window_archive = _active_window(session_archive) + active_window_index = ( + active_window_archive.index if active_window_archive is not None else None + ) + + if active_window_index is None: + return + + if session is not None: + session.select_window(active_window_index) + return + + proc = server.cmd( + "select-window", + "-t", + f"{session_archive.name}:{active_window_index}", + ) + raise_if_stderr(proc, "select-window") + + +def _restore_workspace_focus(server: Server, archive: WorkspaceArchive) -> None: + for session_name in ( + archive.alternate_session_name, + archive.active_session_name, + ): + if session_name is None: + continue + server.cmd("switch-client", "-t", session_name) + + +def _target_window( + session_archive: SessionArchive, + window_archive: WindowArchive, +) -> str: + return f"{session_archive.name}:{window_archive.index}" + + +def _target_pane( + session_archive: SessionArchive, + window_archive: WindowArchive, + pane_archive: PaneArchive, + *, + pane_targets: t.Mapping[int, str], +) -> str: + return pane_targets.get( + pane_archive.index, + f"{_target_window(session_archive, window_archive)}.{pane_archive.index}", + ) + + +def _group_representatives( + sessions: t.Iterable[SessionArchive], +) -> dict[str, str]: + members_by_group: dict[str, list[str]] = {} + for session_archive in sessions: + if session_archive.group_name is None: + continue + members_by_group.setdefault(session_archive.group_name, []).append( + session_archive.name, + ) + + return { + group_name: group_name if group_name in members else members[0] + for group_name, members in members_by_group.items() + } + + +def _ordered_session_archives( + sessions: tuple[SessionArchive, ...], + *, + group_representatives: t.Mapping[str, str], +) -> tuple[SessionArchive, ...]: + return tuple( + session_archive + for _, session_archive in sorted( + enumerate(sessions), + key=lambda item: ( + _is_group_follower(item[1], group_representatives), + item[0], + ), + ) + ) + + +def _is_group_follower( + session_archive: SessionArchive, + group_representatives: t.Mapping[str, str], +) -> bool: + if session_archive.group_name is None: + return False + return group_representatives.get(session_archive.group_name) != session_archive.name + + +def _group_target( + session_archive: SessionArchive, + group_representatives: t.Mapping[str, str], +) -> str | None: + if not _is_group_follower(session_archive, group_representatives): + return None + return group_representatives[session_archive.group_name or ""] + + +def _resolve_session(server: Server, session_name: str) -> Session | None: + try: + return server.sessions.get(default=None, session_name=session_name) + except AttributeError: + return None + + +def _move_initial_window( + server: Server, + *, + session_archive: SessionArchive, + window: Window, + window_archive: WindowArchive, +) -> None: + current_index = str(window.window_index) + if current_index == str(window_archive.index): + return + + proc = server.cmd( + "move-window", + "-s", + f"{session_archive.name}:{current_index}", + "-t", + f"{session_archive.name}:{window_archive.index}", + ) + raise_if_stderr(proc, "move-window") + + +def _active_window(session: SessionArchive) -> WindowArchive | None: + return next((window for window in session.windows if window.active), None) + + +def _active_pane(window: WindowArchive) -> PaneArchive | None: + return next((pane for pane in window.panes if pane.active), None) + + +def _pane_path(pane: PaneArchive | None) -> str | None: + if pane is None: + return None + return _path_or_none(pane.current_path) + + +def _path_or_none(path: str) -> str | None: + return path or None + + +def _name_or_none(name: str) -> str | None: + return name or None + + +def _pane_command( + pane: PaneArchive | None, + *, + process_policy: ProcessRestorePolicy | None, + shell_commands: t.Collection[str], +) -> str | None: + if pane is None: + return None + if not pane.current_command or pane.current_command in shell_commands: + return None + if process_policy is None: + return None + return process_policy.resolve_command(pane.full_command or pane.current_command) + + +def _parse_pane_row(row: str) -> _PaneRow: + parts = row.split(_FIELD_SEPARATOR) + if parts and parts[-1] == "": + parts.pop() + + if len(parts) not in {9, 12, 13}: + msg = f"expected 9, 12, or 13 list-panes fields, got {len(parts)}" + raise ValueError(msg) + + if len(parts) == 9: + parts = [ + *parts[:5], + "", + parts[5], + parts[6], + "0", + parts[7], + parts[8], + "", + "0", + ] + elif len(parts) == 12: + parts = [ + *parts[:8], + "0", + *parts[8:], + ] + + return _PaneRow( + session_name=parts[0], + window_index=_tmux_int(parts[1]), + window_name=parts[2], + window_layout=parts[3], + window_active=_tmux_bool(parts[4]), + window_flags=parts[5], + pane_index=_tmux_int(parts[6]), + pane_active=_tmux_bool(parts[7]), + pane_pid=_tmux_int(parts[8]), + pane_current_command=parts[9], + pane_current_path=parts[10], + pane_title=parts[11], + history_size=_tmux_int(parts[12]), + ) + + +def _tmux_bool(value: str) -> bool: + return value == "1" + + +def _tmux_int(value: str) -> int: + try: + return int(value) + except ValueError: + return 0 + + +def _capture_session_groups(server: Server) -> dict[str, str]: + proc = server.cmd("list-sessions", "-F", _SESSION_FORMAT) + if proc.stderr: + return {} + + groups: dict[str, str] = {} + for line in proc.stdout: + parts = _split_tmux_row(line) + if len(parts) != 3: + continue + session_name, session_grouped, session_group = parts + if _tmux_bool(session_grouped) and session_group: + groups[session_name] = session_group + return groups + + +def _capture_client_state(server: Server) -> tuple[str | None, str | None]: + proc = server.cmd("list-clients", "-F", _CLIENT_FORMAT) + if proc.stderr or not proc.stdout: + return None, None + + parts = _split_tmux_row(proc.stdout[0]) + if len(parts) != 2: + return None, None + return parts[0] or None, parts[1] or None + + +def _capture_automatic_renames( + server: Server, + rows: tuple[_PaneRow, ...], +) -> dict[tuple[str, int], bool | None]: + window_keys = sorted({(row.session_name, row.window_index) for row in rows}) + return { + window_key: _capture_automatic_rename(server, *window_key) + for window_key in window_keys + } + + +def _capture_automatic_rename( + server: Server, + session_name: str, + window_index: int, +) -> bool | None: + proc = server.cmd( + "show-window-options", + "-v", + "-t", + f"{session_name}:{window_index}", + "automatic-rename", + ) + if proc.stderr or not proc.stdout: + return None + value = proc.stdout[0] + if value == "on": + return True + if value == "off": + return False + return None + + +def _capture_process_commands( + rows: tuple[_PaneRow, ...], + process_provider: ProcessCommandProvider | None, +) -> dict[int, str]: + if process_provider is None: + return {} + + commands: dict[int, str] = {} + for pane_pid in sorted({row.pane_pid for row in rows if row.pane_pid > 0}): + command = process_provider.capture(pane_pid) + if command: + commands[pane_pid] = command + return commands + + +def _split_tmux_row(row: str) -> list[str]: + parts = row.split(_FIELD_SEPARATOR) + if parts and parts[-1] == "": + parts.pop() + return parts + + +def _coerce_saved_at(value: datetime.datetime | None) -> datetime.datetime: + if value is None: + return datetime.datetime.now(datetime.timezone.utc) + if value.tzinfo is None: + return value.replace(tzinfo=datetime.timezone.utc) + return value.astimezone(datetime.timezone.utc) + + +def _archive_to_dict(archive: WorkspaceArchive) -> dict[str, object]: + return { + "active_session_name": archive.active_session_name, + "alternate_session_name": archive.alternate_session_name, + "capabilities": list(archive.capabilities), + "format_version": archive.format_version, + "saved_at": _coerce_saved_at(archive.saved_at).isoformat(), + "sessions": [ + { + "active_window_index": session.active_window_index, + "alternate_window_index": session.alternate_window_index, + "group_name": session.group_name, + "name": session.name, + "windows": [ + { + "active": window.active, + "automatic_rename": window.automatic_rename, + "flags": window.flags, + "index": window.index, + "layout": window.layout, + "name": window.name, + "panes": [ + { + "active": pane.active, + "contents": list(pane.contents), + "current_command": pane.current_command, + "current_path": pane.current_path, + "full_command": pane.full_command, + "history_size": pane.history_size, + "index": pane.index, + "title": pane.title, + } + for pane in window.panes + ], + } + for window in session.windows + ], + } + for session in archive.sessions + ], + } + + +def _archive_from_dict(data: object) -> WorkspaceArchive: + archive_data = _expect_mapping(data, "archive") + format_version = _expect_str(archive_data, "format_version") + if format_version != FORMAT_VERSION: + msg = f"unsupported archive format: {format_version}" + raise ValueError(msg) + + return WorkspaceArchive( + active_session_name=_optional_str(archive_data, "active_session_name"), + alternate_session_name=_optional_str(archive_data, "alternate_session_name"), + capabilities=_optional_str_tuple( + archive_data, + "capabilities", + default=CAPTURED_CAPABILITIES, + ), + format_version=format_version, + saved_at=datetime.datetime.fromisoformat(_expect_str(archive_data, "saved_at")), + sessions=tuple( + _session_from_dict(session) + for session in _expect_list(archive_data, "sessions") + ), + ) + + +def _session_from_dict(data: object) -> SessionArchive: + session_data = _expect_mapping(data, "session") + return SessionArchive( + active_window_index=_optional_int(session_data, "active_window_index"), + alternate_window_index=_optional_int(session_data, "alternate_window_index"), + group_name=_optional_str(session_data, "group_name"), + name=_expect_str(session_data, "name"), + windows=tuple( + _window_from_dict(window) + for window in _expect_list(session_data, "windows") + ), + ) + + +def _window_from_dict(data: object) -> WindowArchive: + window_data = _expect_mapping(data, "window") + return WindowArchive( + active=_expect_bool(window_data, "active"), + automatic_rename=_optional_bool(window_data, "automatic_rename"), + flags=_optional_str(window_data, "flags") or "", + index=_expect_int(window_data, "index"), + layout=_expect_str(window_data, "layout"), + name=_expect_str(window_data, "name"), + panes=tuple( + _pane_from_dict(pane) for pane in _expect_list(window_data, "panes") + ), + ) + + +def _pane_from_dict(data: object) -> PaneArchive: + pane_data = _expect_mapping(data, "pane") + return PaneArchive( + active=_expect_bool(pane_data, "active"), + contents=_optional_str_tuple(pane_data, "contents", default=()), + current_command=_expect_str(pane_data, "current_command"), + current_path=_expect_str(pane_data, "current_path"), + full_command=_optional_str(pane_data, "full_command") or "", + history_size=_optional_int(pane_data, "history_size") or 0, + index=_expect_int(pane_data, "index"), + title=_optional_str(pane_data, "title") or "", + ) + + +def _expect_mapping(data: object, name: str) -> t.Mapping[str, object]: + if not isinstance(data, dict): + msg = f"{name} must be an object" + raise TypeError(msg) + return t.cast("t.Mapping[str, object]", data) + + +def _expect_list(data: t.Mapping[str, object], key: str) -> list[object]: + value = data[key] + if not isinstance(value, list): + msg = f"{key} must be a list" + raise TypeError(msg) + return value + + +def _optional_str_tuple( + data: t.Mapping[str, object], + key: str, + *, + default: tuple[str, ...], +) -> tuple[str, ...]: + value = data.get(key) + if value is None: + return default + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + msg = f"{key} must be a list of strings" + raise TypeError(msg) + return tuple(value) + + +def _optional_str(data: t.Mapping[str, object], key: str) -> str | None: + value = data.get(key) + if value is None: + return None + if not isinstance(value, str): + msg = f"{key} must be a string or null" + raise TypeError(msg) + return value + + +def _optional_int(data: t.Mapping[str, object], key: str) -> int | None: + value = data.get(key) + if value is None: + return None + if not isinstance(value, int): + msg = f"{key} must be an integer or null" + raise TypeError(msg) + return value + + +def _optional_bool(data: t.Mapping[str, object], key: str) -> bool | None: + value = data.get(key) + if value is None: + return None + if not isinstance(value, bool): + msg = f"{key} must be a boolean or null" + raise TypeError(msg) + return value + + +def _expect_str(data: t.Mapping[str, object], key: str) -> str: + value = data[key] + if not isinstance(value, str): + msg = f"{key} must be a string" + raise TypeError(msg) + return value + + +def _expect_int(data: t.Mapping[str, object], key: str) -> int: + value = data[key] + if not isinstance(value, int): + msg = f"{key} must be an integer" + raise TypeError(msg) + return value + + +def _expect_bool(data: t.Mapping[str, object], key: str) -> bool: + value = data[key] + if not isinstance(value, bool): + msg = f"{key} must be a boolean" + raise TypeError(msg) + return value diff --git a/src/libtmux/resurrect/continuum.py b/src/libtmux/resurrect/continuum.py new file mode 100644 index 000000000..9b34272f8 --- /dev/null +++ b/src/libtmux/resurrect/continuum.py @@ -0,0 +1,453 @@ +"""Headless tmux-continuum style autosave helpers.""" + +from __future__ import annotations + +import datetime +import hashlib +import json +import os +import pathlib +import typing as t +from dataclasses import dataclass + +from libtmux._internal.types import StrPath +from libtmux.resurrect.archives import ( + RestorePolicy, + capture_archive, + restore_archive, + write_archive, +) + +if t.TYPE_CHECKING: + from libtmux.server import Server + +STATE_FORMAT_VERSION = "libtmux.resurrect.autosave-state.v1" +"""Autosave state format identifier.""" + +DEFAULT_AUTOSAVE_INTERVAL = datetime.timedelta(minutes=15) +"""Default interval between autosaves.""" + +DEFAULT_STARTUP_RESTORE_GRACE = datetime.timedelta(seconds=10) +"""Default window in which startup restore is considered fresh.""" + + +@dataclass(frozen=True, slots=True) +class AutosaveState: + """Persisted autosave state.""" + + last_saved_at: datetime.datetime | None = None + last_archive_path: pathlib.Path | None = None + save_count: int = 0 + format_version: str = STATE_FORMAT_VERSION + + +@dataclass(frozen=True, slots=True) +class AutosavePaths: + """Socket-aware archive and state paths.""" + + archive_path: pathlib.Path + state_path: pathlib.Path + + +@dataclass(frozen=True, slots=True) +class AutosaveResult: + """Result returned by :func:`autosave_once`.""" + + saved: bool + reason: str + archive_path: pathlib.Path + state: AutosaveState + state_path: pathlib.Path | None = None + + +@dataclass(frozen=True, slots=True) +class StartupRestoreDecision: + """Decision returned by :func:`should_restore_on_startup`.""" + + allowed: bool + reason: str + + +@dataclass(frozen=True, slots=True) +class StartupRestoreResult: + """Result returned by :func:`startup_restore_once`.""" + + restored: bool + reason: str + archive_path: pathlib.Path + decision: StartupRestoreDecision + + +def default_autosave_paths( + server: Server, + directory: StrPath, + *, + prefix: str = "workspace", +) -> AutosavePaths: + """Return deterministic, socket-aware autosave paths. + + The generated filenames are keyed by the tmux socket identity, but socket + paths are hashed so local filesystem details are not embedded in filenames. + + Examples + -------- + >>> import pathlib + >>> paths = default_autosave_paths( + ... Server(socket_name="demo"), + ... pathlib.Path("/tmp"), + ... ) + >>> paths.archive_path.name.startswith("workspace-name-") + True + >>> paths.state_path.name.endswith(".state.json") + True + """ + base_dir = pathlib.Path(directory) + stem = _autosave_stem(server, prefix=prefix) + return AutosavePaths( + archive_path=base_dir / f"{stem}.json", + state_path=base_dir / f"{stem}.state.json", + ) + + +def next_autosave_at( + last_saved_at: datetime.datetime | None, + *, + interval: datetime.timedelta = DEFAULT_AUTOSAVE_INTERVAL, +) -> datetime.datetime | None: + """Return when the next autosave is due. + + Examples + -------- + >>> import datetime + >>> saved = datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc) + >>> next_autosave_at(saved) + datetime.datetime(2026, 7, 4, 12, 15, tzinfo=datetime.timezone.utc) + >>> next_autosave_at(None) is None + True + """ + if last_saved_at is None: + return None + return _coerce_datetime(last_saved_at) + interval + + +def should_autosave( + *, + last_saved_at: datetime.datetime | None, + now: datetime.datetime | None = None, + interval: datetime.timedelta = DEFAULT_AUTOSAVE_INTERVAL, +) -> bool: + """Return True when an autosave should run. + + Examples + -------- + >>> should_autosave(last_saved_at=None) + True + """ + if last_saved_at is None: + return True + + due_at = next_autosave_at(last_saved_at, interval=interval) + return due_at is not None and _coerce_datetime(now) >= due_at + + +def read_autosave_state(path: StrPath) -> AutosaveState: + """Read persisted autosave state, returning an empty state if missing. + + Examples + -------- + >>> import pathlib + >>> target = pathlib.Path(request.getfixturevalue("tmp_path")) / "state.json" + >>> read_autosave_state(target).save_count + 0 + """ + source = pathlib.Path(path) + if not source.exists(): + return AutosaveState() + return _state_from_dict(json.loads(source.read_text(encoding="utf-8"))) + + +def write_autosave_state(state: AutosaveState, path: StrPath) -> pathlib.Path: + """Write autosave state using an atomic replace. + + Examples + -------- + >>> import pathlib + >>> target = pathlib.Path(request.getfixturevalue("tmp_path")) / "state.json" + >>> saved = write_autosave_state(AutosaveState(save_count=1), target) + >>> saved == target + True + """ + destination = pathlib.Path(path) + destination.parent.mkdir(parents=True, exist_ok=True) + tmp_path = destination.with_name(f".{destination.name}.{os.getpid()}.tmp") + tmp_path.write_text( + json.dumps(_state_to_dict(state), indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + tmp_path.replace(destination) + return destination + + +def autosave_once( + server: Server, + *, + archive_path: StrPath, + state_path: StrPath | None = None, + now: datetime.datetime | None = None, + interval: datetime.timedelta = DEFAULT_AUTOSAVE_INTERVAL, + force: bool = False, +) -> AutosaveResult: + """Capture and write one archive if autosave is due. + + Examples + -------- + >>> import pathlib + >>> target = pathlib.Path(request.getfixturevalue("tmp_path")) / "tmux.json" + >>> result = autosave_once(server, archive_path=target, force=True) + >>> result.saved + True + """ + resolved_archive_path = pathlib.Path(archive_path) + resolved_state_path = pathlib.Path(state_path) if state_path is not None else None + previous_state = ( + read_autosave_state(resolved_state_path) + if resolved_state_path is not None + else AutosaveState() + ) + resolved_now = _coerce_datetime(now) + + if not force and not should_autosave( + last_saved_at=previous_state.last_saved_at, + now=resolved_now, + interval=interval, + ): + return AutosaveResult( + saved=False, + reason="interval_not_elapsed", + archive_path=resolved_archive_path, + state=previous_state, + state_path=resolved_state_path, + ) + + archive = capture_archive(server, saved_at=resolved_now) + write_archive(archive, resolved_archive_path) + next_state = AutosaveState( + last_saved_at=resolved_now, + last_archive_path=resolved_archive_path, + save_count=previous_state.save_count + 1, + ) + if resolved_state_path is not None: + write_autosave_state(next_state, resolved_state_path) + + return AutosaveResult( + saved=True, + reason="saved", + archive_path=resolved_archive_path, + state=next_state, + state_path=resolved_state_path, + ) + + +def should_restore_on_startup( + *, + enabled: bool, + halt_file: StrPath | None, + session_count: int, + another_server_running: bool, + tmux_started_at: datetime.datetime | None = None, + now: datetime.datetime | None = None, + startup_grace: datetime.timedelta = DEFAULT_STARTUP_RESTORE_GRACE, +) -> StartupRestoreDecision: + """Return whether startup restore should run. + + Examples + -------- + >>> import datetime + >>> now = datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc) + >>> should_restore_on_startup( + ... enabled=True, + ... halt_file=None, + ... session_count=0, + ... another_server_running=False, + ... tmux_started_at=now, + ... now=now, + ... ) + StartupRestoreDecision(allowed=True, reason='restore_allowed') + """ + if not enabled: + return StartupRestoreDecision(allowed=False, reason="restore_disabled") + if halt_file is not None and pathlib.Path(halt_file).exists(): + return StartupRestoreDecision(allowed=False, reason="halt_file_present") + if another_server_running: + return StartupRestoreDecision(allowed=False, reason="another_server_running") + if session_count > 0: + return StartupRestoreDecision(allowed=False, reason="sessions_exist") + if tmux_started_at is not None: + elapsed = _coerce_datetime(now) - _coerce_datetime(tmux_started_at) + if elapsed > startup_grace: + return StartupRestoreDecision( + allowed=False, + reason="startup_window_elapsed", + ) + return StartupRestoreDecision(allowed=True, reason="restore_allowed") + + +def startup_restore_once( + server: Server, + archive_path: StrPath, + *, + enabled: bool, + halt_file: StrPath | None = None, + another_server_running: bool = False, + tmux_started_at: datetime.datetime | None = None, + now: datetime.datetime | None = None, + startup_grace: datetime.timedelta = DEFAULT_STARTUP_RESTORE_GRACE, + on_exists: RestorePolicy = "reuse", +) -> StartupRestoreResult: + """Restore an archive once when startup guards allow it. + + Examples + -------- + >>> import pathlib + >>> result = startup_restore_once( + ... server, + ... pathlib.Path("missing.json"), + ... enabled=False, + ... ) + >>> result.restored + False + """ + resolved_archive_path = pathlib.Path(archive_path) + decision = should_restore_on_startup( + enabled=enabled, + halt_file=halt_file, + session_count=len(server.sessions), + another_server_running=another_server_running, + tmux_started_at=tmux_started_at, + now=now, + startup_grace=startup_grace, + ) + if not decision.allowed: + return StartupRestoreResult( + restored=False, + reason=decision.reason, + archive_path=resolved_archive_path, + decision=decision, + ) + + restore_archive(resolved_archive_path, server, on_exists=on_exists) + return StartupRestoreResult( + restored=True, + reason="restored", + archive_path=resolved_archive_path, + decision=decision, + ) + + +def _coerce_datetime(value: datetime.datetime | None) -> datetime.datetime: + if value is None: + return datetime.datetime.now(datetime.timezone.utc) + if value.tzinfo is None: + return value.replace(tzinfo=datetime.timezone.utc) + return value.astimezone(datetime.timezone.utc) + + +def _autosave_stem(server: Server, *, prefix: str) -> str: + safe_prefix = _safe_path_component(prefix) or "workspace" + socket_path = getattr(server, "socket_path", None) + socket_name = getattr(server, "socket_name", None) + if socket_path is not None: + identity = f"path:{os.fspath(socket_path)}" + identity_kind = "path" + elif socket_name is not None: + identity = f"name:{socket_name}" + identity_kind = "name" + else: + identity = "default" + identity_kind = "default" + + digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()[:12] + return f"{safe_prefix}-{identity_kind}-{digest}" + + +def _safe_path_component(value: str) -> str: + return "".join( + character if character.isalnum() or character in {"-", "_"} else "-" + for character in value + ).strip("-") + + +def _state_to_dict(state: AutosaveState) -> dict[str, object]: + return { + "format_version": state.format_version, + "last_archive_path": ( + str(state.last_archive_path) if state.last_archive_path else None + ), + "last_saved_at": ( + _coerce_datetime(state.last_saved_at).isoformat() + if state.last_saved_at is not None + else None + ), + "save_count": state.save_count, + } + + +def _state_from_dict(data: object) -> AutosaveState: + state_data = _expect_mapping(data, "state") + format_version = _expect_str(state_data, "format_version") + if format_version != STATE_FORMAT_VERSION: + msg = f"unsupported autosave state format: {format_version}" + raise ValueError(msg) + + return AutosaveState( + format_version=format_version, + last_archive_path=_optional_path(state_data, "last_archive_path"), + last_saved_at=_optional_datetime(state_data, "last_saved_at"), + save_count=_expect_int(state_data, "save_count"), + ) + + +def _optional_path(data: t.Mapping[str, object], key: str) -> pathlib.Path | None: + value = data[key] + if value is None: + return None + if not isinstance(value, str): + msg = f"{key} must be a string or null" + raise TypeError(msg) + return pathlib.Path(value) + + +def _optional_datetime( + data: t.Mapping[str, object], + key: str, +) -> datetime.datetime | None: + value = data[key] + if value is None: + return None + if not isinstance(value, str): + msg = f"{key} must be a string or null" + raise TypeError(msg) + return _coerce_datetime(datetime.datetime.fromisoformat(value)) + + +def _expect_mapping(data: object, name: str) -> t.Mapping[str, object]: + if not isinstance(data, dict): + msg = f"{name} must be an object" + raise TypeError(msg) + return t.cast("t.Mapping[str, object]", data) + + +def _expect_str(data: t.Mapping[str, object], key: str) -> str: + value = data[key] + if not isinstance(value, str): + msg = f"{key} must be a string" + raise TypeError(msg) + return value + + +def _expect_int(data: t.Mapping[str, object], key: str) -> int: + value = data[key] + if not isinstance(value, int): + msg = f"{key} must be an integer" + raise TypeError(msg) + return value diff --git a/src/libtmux/resurrect/processes.py b/src/libtmux/resurrect/processes.py new file mode 100644 index 000000000..bba6276d5 --- /dev/null +++ b/src/libtmux/resurrect/processes.py @@ -0,0 +1,498 @@ +"""Pure Python process restore policy helpers.""" + +from __future__ import annotations + +import pathlib +import shlex +import subprocess +import typing as t +from dataclasses import dataclass + +DEFAULT_RESTORE_PROGRAMS = ( + "vi", + "vim", + "view", + "nvim", + "emacs", + "man", + "less", + "more", + "tail", + "top", + "htop", + "irssi", + "weechat", + "mutt", +) +"""Conservative default process allow-list from tmux-resurrect.""" + + +@dataclass(frozen=True, slots=True) +class ProcessRestoreRule: + """A single process restore allow-list entry.""" + + match: str + relaxed: bool = False + command_template: str | None = None + + @classmethod + def parse(cls, value: str) -> ProcessRestoreRule: + """Parse a tmux-resurrect-style process rule. + + Examples + -------- + >>> ProcessRestoreRule.parse('vim') + ProcessRestoreRule(match='vim', relaxed=False, command_template=None) + + >>> rule = ProcessRestoreRule.parse('python->uv run python *') + >>> rule.match + 'python' + >>> rule.command_template + 'uv run python *' + """ + relaxed = value.startswith("~") + raw = value[1:] if relaxed else value + if "->" in raw: + match, command_template = raw.split("->", 1) + else: + match, command_template = raw, None + return cls( + match=match, + relaxed=relaxed, + command_template=command_template or None, + ) + + def matches(self, full_command: str) -> bool: + """Return True when this rule matches a saved command. + + Examples + -------- + >>> ProcessRestoreRule('vim').matches('vim pyproject.toml') + True + + >>> ProcessRestoreRule('server', relaxed=True).matches('node server.js') + True + + >>> ProcessRestoreRule('git log').matches('git log --oneline') + True + """ + if self.relaxed: + return self.match in full_command + return _command_matches(full_command, self.match) + + def resolve(self, full_command: str) -> str: + """Return the replay command for this rule. + + Examples + -------- + >>> ProcessRestoreRule('vim').resolve('vim pyproject.toml') + 'vim pyproject.toml' + + >>> ProcessRestoreRule('python', command_template='uv run python *').resolve( + ... 'python -m http.server 8000', + ... ) + 'uv run python -m http.server 8000' + + >>> ProcessRestoreRule( + ... 'rails server', + ... relaxed=True, + ... command_template='rails server *', + ... ).resolve('/rubies/bin/ruby script/rails server -p 3000') + 'rails server -p 3000' + """ + if self.command_template is None: + return full_command + arguments = _arguments_after_match( + full_command, + self.match, + relaxed=self.relaxed, + ) + return self.command_template.replace("*", arguments).strip() + + +@dataclass(frozen=True, slots=True) +class ProcessRestorePolicy: + """Process replay policy independent from tmux plugin options.""" + + enabled: bool = True + restore_all: bool = False + rules: tuple[ProcessRestoreRule, ...] = tuple( + ProcessRestoreRule(program) for program in DEFAULT_RESTORE_PROGRAMS + ) + + @classmethod + def from_options( + cls, + user_processes: str | None, + *, + default_processes: tuple[str, ...] = DEFAULT_RESTORE_PROGRAMS, + ) -> ProcessRestorePolicy: + """Build a policy from tmux-resurrect-style process options. + + Examples + -------- + >>> policy = ProcessRestorePolicy.from_options("'python->uv run python *'") + >>> policy.resolve_command('python -m http.server 8000') + 'uv run python -m http.server 8000' + + >>> ProcessRestorePolicy.from_options('false').resolve_command('vim') + """ + if user_processes == "false": + return cls(enabled=False, rules=()) + if user_processes == ":all:": + return cls(restore_all=True, rules=()) + + rules = [ProcessRestoreRule.parse(program) for program in default_processes] + if user_processes: + rules.extend( + ProcessRestoreRule.parse(program) + for program in shlex.split(user_processes) + ) + return cls(rules=tuple(rules)) + + def resolve_command(self, full_command: str) -> str | None: + """Return a replay command, or None when policy skips this process. + + Examples + -------- + >>> ProcessRestorePolicy.from_options(None).resolve_command('vim README.md') + 'vim README.md' + + >>> ProcessRestorePolicy.from_options(None).resolve_command('node server.js') + + >>> policy = ProcessRestorePolicy.from_options('"git log"') + >>> policy.resolve_command('git log --oneline') + 'git log --oneline' + """ + if not self.enabled or not full_command: + return None + if self.restore_all: + return full_command + + for rule in self.rules: + if rule.matches(full_command): + return rule.resolve(full_command) + return None + + +DEFAULT_PROCESS_RESTORE_POLICY = ProcessRestorePolicy() +"""Default conservative process replay policy.""" + + +class ProcessCommandProvider(t.Protocol): + """Provider that resolves a pane process id to a full command line.""" + + def capture(self, pid: int) -> str | None: + """Return a shell command for a process id. + + Examples + -------- + >>> class MissingProvider: + ... def capture(self, pid: int) -> str | None: + ... return None + >>> MissingProvider().capture(1234) is None + True + """ + ... + + +@dataclass(frozen=True, slots=True) +class _ProcfsProcess: + """Parsed procfs process metadata.""" + + pid: int + ppid: int + pgrp: int + tpgid: int + command: tuple[str, ...] + + +@dataclass(frozen=True, slots=True) +class ProcfsProcessCommandProvider: + """Capture full process commands from Linux procfs.""" + + proc_root: pathlib.Path = pathlib.Path("/proc") + + def capture(self, pid: int) -> str | None: + """Return the foreground command for a tmux pane process id. + + Examples + -------- + >>> import pathlib + >>> provider = ProcfsProcessCommandProvider(pathlib.Path('/missing-proc')) + >>> provider.capture(1234) is None + True + """ + if pid <= 0: + return None + + try: + proc_dirs = tuple(self.proc_root.iterdir()) + except OSError: + proc_dirs = () + + processes = { + process.pid: process + for process in (_read_procfs_process(proc_dir) for proc_dir in proc_dirs) + if process is not None + } + root = processes.get(pid) or _read_procfs_process(self.proc_root / str(pid)) + if root is None: + return None + processes[root.pid] = root + + process = _select_foreground_process(root, processes) + if process is None: + return None + return shlex.join(process.command) + + +def _read_procfs_process(proc_dir: pathlib.Path) -> _ProcfsProcess | None: + """Return parsed process metadata from a procfs PID directory. + + Examples + -------- + >>> _read_procfs_process(pathlib.Path('/missing-proc/1234')) is None + True + """ + if not proc_dir.name.isdecimal(): + return None + + pid = int(proc_dir.name) + command = _read_procfs_cmdline(proc_dir / "cmdline") + stat = _read_procfs_stat(proc_dir / "stat") + if stat is None and not command: + return None + + ppid, pgrp, tpgid = stat or (0, 0, 0) + return _ProcfsProcess( + pid=pid, + ppid=ppid, + pgrp=pgrp, + tpgid=tpgid, + command=command, + ) + + +def _read_procfs_cmdline(path: pathlib.Path) -> tuple[str, ...]: + """Return decoded argv from a procfs ``cmdline`` file. + + Examples + -------- + >>> _read_procfs_cmdline(pathlib.Path('/missing-proc/cmdline')) + () + """ + try: + raw_cmdline = path.read_bytes() + except OSError: + return () + + return tuple( + part.decode("utf-8", "backslashreplace") + for part in raw_cmdline.split(b"\0") + if part + ) + + +def _read_procfs_stat(path: pathlib.Path) -> tuple[int, int, int] | None: + r"""Return ``(ppid, pgrp, tpgid)`` from a procfs ``stat`` file. + + Examples + -------- + >>> tmp_path = pathlib.Path(request.getfixturevalue("tmp_path")) + >>> stat_path = tmp_path / "stat" + >>> _ = stat_path.write_text("123 (vim) S 1 200 0 34816 200 0\\n") + >>> _read_procfs_stat(stat_path) + (1, 200, 200) + """ + try: + stat = path.read_text(encoding="utf-8") + except OSError: + return None + + try: + suffix = stat.rsplit(")", 1)[1].split() + return int(suffix[1]), int(suffix[2]), int(suffix[5]) + except (IndexError, ValueError): + return None + + +def _select_foreground_process( + root: _ProcfsProcess, + processes: t.Mapping[int, _ProcfsProcess], +) -> _ProcfsProcess | None: + """Return the foreground descendant process, falling back to ``root``. + + Examples + -------- + >>> shell = _ProcfsProcess(100, 1, 100, 200, ("-zsh",)) + >>> vim = _ProcfsProcess(200, 100, 200, 200, ("vim", "README.md")) + >>> _select_foreground_process(shell, {100: shell, 200: vim}) == vim + True + """ + children_by_parent: dict[int, list[_ProcfsProcess]] = {} + for process in processes.values(): + children_by_parent.setdefault(process.ppid, []).append(process) + + descendants: list[tuple[int, _ProcfsProcess]] = [] + seen = {root.pid} + stack = [(1, child) for child in children_by_parent.get(root.pid, [])] + while stack: + depth, process = stack.pop() + if process.pid in seen: + continue + seen.add(process.pid) + descendants.append((depth, process)) + stack.extend( + (depth + 1, child) for child in children_by_parent.get(process.pid, []) + ) + + foreground = [ + (depth, process) + for depth, process in descendants + if root.tpgid > 0 and process.pgrp == root.tpgid and process.command + ] + if foreground: + return max(foreground, key=lambda item: (item[0], item[1].pid))[1] + + with_command = [ + (depth, process) for depth, process in descendants if process.command + ] + if with_command: + return max(with_command, key=lambda item: (item[0], item[1].pid))[1] + + return root if root.command else None + + +@dataclass(frozen=True, slots=True) +class PsProcessCommandProvider: + """Capture full process commands with ``ps``.""" + + ps_bin: str = "ps" + + def capture(self, pid: int) -> str | None: + """Return a command from ``ps -p -o command=``. + + Examples + -------- + >>> PsProcessCommandProvider(ps_bin='/missing-ps').capture(1234) is None + True + """ + if pid <= 0: + return None + try: + proc = subprocess.run( + (self.ps_bin, "-p", str(pid), "-o", "command="), + capture_output=True, + check=False, + text=True, + ) + except OSError: + return None + if proc.returncode != 0: + return None + command = proc.stdout.strip() + return command or None + + +@dataclass(frozen=True, slots=True) +class CompositeProcessCommandProvider: + """Try multiple process command providers in order.""" + + providers: tuple[ProcessCommandProvider, ...] + + def capture(self, pid: int) -> str | None: + """Return the first command captured by a child provider. + + Examples + -------- + >>> class Provider: + ... def capture(self, pid: int) -> str | None: + ... return 'vim README.md' + >>> CompositeProcessCommandProvider((Provider(),)).capture(1234) + 'vim README.md' + """ + for provider in self.providers: + command = provider.capture(pid) + if command: + return command + return None + + +def default_process_command_provider() -> ProcessCommandProvider: + """Return the default full-command provider chain. + + Examples + -------- + >>> provider = default_process_command_provider() + >>> hasattr(provider, 'capture') + True + """ + return CompositeProcessCommandProvider( + (ProcfsProcessCommandProvider(),), + ) + + +def _command_matches(full_command: str, match: str) -> bool: + """Return True when a full command matches an exact restore rule. + + Examples + -------- + >>> _command_matches('vim pyproject.toml', 'vim') + True + + >>> _command_matches('git log --oneline', 'git log') + True + + >>> _command_matches('vimdiff file.py', 'vim') + False + """ + return full_command == match or full_command.startswith(f"{match} ") + + +def _arguments_after_match( + full_command: str, + match: str, + *, + relaxed: bool = False, +) -> str: + """Return shell-quoted arguments after an exact command match. + + Examples + -------- + >>> _arguments_after_match('python -m http.server 8000', 'python') + '-m http.server 8000' + + >>> _arguments_after_match('node server.js', 'python') + '' + + >>> _arguments_after_match( + ... '/rubies/bin/ruby script/rails server -p 3000', + ... 'rails server', + ... relaxed=True, + ... ) + '-p 3000' + """ + if relaxed: + match_index = full_command.find(match) + if match_index == -1: + return "" + argument_index = match_index + len(match) + while ( + argument_index < len(full_command) + and not full_command[argument_index].isspace() + ): + argument_index += 1 + return full_command[argument_index:].lstrip() + + try: + parts = shlex.split(full_command) + except ValueError: + return "" + try: + match_parts = shlex.split(match) + except ValueError: + return "" + if parts[: len(match_parts)] != match_parts: + return "" + return shlex.join(parts[len(match_parts) :]) diff --git a/src/libtmux/resurrect/resurrect_file.py b/src/libtmux/resurrect/resurrect_file.py new file mode 100644 index 000000000..14ca29269 --- /dev/null +++ b/src/libtmux/resurrect/resurrect_file.py @@ -0,0 +1,350 @@ +"""tmux-resurrect tab-file import and export helpers.""" + +from __future__ import annotations + +import datetime +import typing as t + +from libtmux.resurrect.archives import ( + PaneArchive, + SessionArchive, + WindowArchive, + WorkspaceArchive, +) + +_DELIMITER = "\t" + + +def archive_to_resurrect_file(archive: WorkspaceArchive) -> str: + """Export a native archive to tmux-resurrect-style rows. + + Examples + -------- + >>> archive = WorkspaceArchive( + ... saved_at=datetime.datetime(2026, 7, 4, tzinfo=datetime.timezone.utc), + ... sessions=(), + ... ) + >>> archive_to_resurrect_file(archive) + '' + """ + rows: list[str] = [] + group_representatives = _group_representatives(archive.sessions) + + for session in archive.sessions: + group_target = _group_target(session, group_representatives) + if group_target is None: + continue + rows.append(_grouped_session_row(session, group_target)) + + for session in archive.sessions: + if _group_target(session, group_representatives) is not None: + continue + for window in session.windows: + rows.extend(_pane_row(session, window, pane) for pane in window.panes) + + for session in archive.sessions: + if _group_target(session, group_representatives) is not None: + continue + rows.extend(_window_row(session, window) for window in session.windows) + + if archive.active_session_name or archive.alternate_session_name: + rows.append( + _DELIMITER.join( + ( + "state", + archive.active_session_name or "", + archive.alternate_session_name or "", + ), + ), + ) + + return "\n".join(rows) + ("\n" if rows else "") + + +def archive_from_resurrect_file( + text: str, + *, + saved_at: datetime.datetime | None = None, +) -> WorkspaceArchive: + r"""Import tmux-resurrect-style rows into a native archive. + + Examples + -------- + >>> text = "window\talpha\t0\t:editor\t1\t:*\ttiled\ton\n" + >>> archive = archive_from_resurrect_file( + ... text, + ... saved_at=datetime.datetime(2026, 7, 4, tzinfo=datetime.timezone.utc), + ... ) + >>> archive.sessions[0].windows[0].name + 'editor' + """ + active_session_name: str | None = None + alternate_session_name: str | None = None + grouped_sessions: dict[str, tuple[str, int | None, int | None]] = {} + panes: dict[tuple[str, int], list[PaneArchive]] = {} + windows: dict[tuple[str, int], WindowArchive] = {} + + for raw_line in text.splitlines(): + if not raw_line: + continue + fields = raw_line.split(_DELIMITER) + line_type = _field(fields, 0) + if line_type == "grouped_session": + grouped_sessions[_field(fields, 1)] = ( + _field(fields, 2), + _unprefixed_int(_field(fields, 3)), + _unprefixed_int(_field(fields, 4)), + ) + elif line_type == "pane": + session_name = _field(fields, 1) + window_index = _int_field(fields, 2) + panes.setdefault((session_name, window_index), []).append( + _pane_from_row(fields), + ) + elif line_type == "window": + session_name = _field(fields, 1) + window = _window_from_row(fields) + windows[(session_name, window.index)] = window + elif line_type == "state": + active_session_name = _empty_to_none(_field(fields, 1)) + alternate_session_name = _empty_to_none(_field(fields, 2)) + + session_names = ( + {session_name for session_name, _ in windows} + | {session_name for session_name, _ in panes} + | set(grouped_sessions) + | {target for target, _, _ in grouped_sessions.values() if target} + ) + grouped_targets = {target for target, _, _ in grouped_sessions.values() if target} + + sessions = tuple( + _session_from_rows( + session_name, + grouped_sessions=grouped_sessions, + grouped_targets=grouped_targets, + panes=panes, + windows=windows, + ) + for session_name in sorted(session_names) + ) + + return WorkspaceArchive( + active_session_name=active_session_name, + alternate_session_name=alternate_session_name, + saved_at=saved_at or datetime.datetime.now(datetime.timezone.utc), + sessions=sessions, + ) + + +def _session_from_rows( + session_name: str, + *, + grouped_sessions: t.Mapping[str, tuple[str, int | None, int | None]], + grouped_targets: set[str], + panes: t.Mapping[tuple[str, int], list[PaneArchive]], + windows: t.Mapping[tuple[str, int], WindowArchive], +) -> SessionArchive: + session_windows = tuple( + WindowArchive( + active=window.active, + automatic_rename=window.automatic_rename, + flags=window.flags, + index=window.index, + layout=window.layout, + name=window.name, + panes=tuple( + sorted( + panes.get((session_name, window_index), ()), + key=lambda pane: pane.index, + ), + ), + ) + for (window_session_name, window_index), window in sorted(windows.items()) + if window_session_name == session_name + ) + group_name: str | None = None + alternate_window_index: int | None = None + active_window_index: int | None = _active_window_index(session_windows) + if session_name in grouped_sessions: + group_name, alternate_window_index, active_window_index = grouped_sessions[ + session_name + ] + elif session_name in grouped_targets: + group_name = session_name + + return SessionArchive( + name=session_name, + group_name=group_name, + alternate_window_index=alternate_window_index, + active_window_index=active_window_index, + windows=session_windows, + ) + + +def _grouped_session_row(session: SessionArchive, group_target: str) -> str: + return _DELIMITER.join( + ( + "grouped_session", + session.name, + group_target, + _prefixed(session.alternate_window_index), + _prefixed(_session_active_window_index(session)), + ), + ) + + +def _window_row(session: SessionArchive, window: WindowArchive) -> str: + if window.automatic_rename is None: + automatic_rename = ":" + else: + automatic_rename = "on" if window.automatic_rename else "off" + return _DELIMITER.join( + ( + "window", + session.name, + str(window.index), + f":{window.name}", + _bool_field(window.active), + f":{window.flags}", + window.layout, + automatic_rename, + ), + ) + + +def _pane_row( + session: SessionArchive, + window: WindowArchive, + pane: PaneArchive, +) -> str: + return _DELIMITER.join( + ( + "pane", + session.name, + str(window.index), + _bool_field(window.active), + f":{window.flags}", + str(pane.index), + pane.title, + f":{_escape_path(pane.current_path)}", + _bool_field(pane.active), + pane.current_command, + f":{pane.full_command}", + ), + ) + + +def _window_from_row(fields: list[str]) -> WindowArchive: + return WindowArchive( + active=_bool_value(_field(fields, 4)), + automatic_rename=_auto_rename_value(_field(fields, 7)), + flags=_remove_prefix(_field(fields, 5)), + index=_int_field(fields, 2), + layout=_field(fields, 6), + name=_remove_prefix(_field(fields, 3)), + panes=(), + ) + + +def _pane_from_row(fields: list[str]) -> PaneArchive: + return PaneArchive( + active=_bool_value(_field(fields, 8)), + current_command=_field(fields, 9), + current_path=_unescape_path(_remove_prefix(_field(fields, 7))), + full_command=_remove_prefix(_field(fields, 10)), + index=_int_field(fields, 5), + title=_field(fields, 6), + ) + + +def _group_representatives( + sessions: t.Iterable[SessionArchive], +) -> dict[str, str]: + members_by_group: dict[str, list[str]] = {} + for session_archive in sessions: + if session_archive.group_name is None: + continue + members_by_group.setdefault(session_archive.group_name, []).append( + session_archive.name, + ) + + return { + group_name: group_name if group_name in members else members[0] + for group_name, members in members_by_group.items() + } + + +def _group_target( + session: SessionArchive, + group_representatives: t.Mapping[str, str], +) -> str | None: + if session.group_name is None: + return None + group_target = group_representatives.get(session.group_name) + if group_target is None or group_target == session.name: + return None + return group_target + + +def _bool_field(value: bool) -> str: + return "1" if value else "0" + + +def _bool_value(value: str) -> bool: + return value == "1" + + +def _prefixed(value: int | None) -> str: + return ":" if value is None else f":{value}" + + +def _unprefixed_int(value: str) -> int | None: + raw = _remove_prefix(value) + return int(raw) if raw else None + + +def _remove_prefix(value: str) -> str: + return value[1:] if value.startswith(":") else value + + +def _empty_to_none(value: str) -> str | None: + return value or None + + +def _auto_rename_value(value: str) -> bool | None: + if value == "on": + return True + if value == "off": + return False + return None + + +def _active_window_index(windows: t.Sequence[WindowArchive]) -> int | None: + active = next((window for window in windows if window.active), None) + return active.index if active else None + + +def _session_active_window_index(session: SessionArchive) -> int | None: + if session.active_window_index is not None: + return session.active_window_index + return _active_window_index(session.windows) + + +def _field(fields: list[str], index: int, default: str = "") -> str: + try: + return fields[index] + except IndexError: + return default + + +def _int_field(fields: list[str], index: int) -> int: + value = _field(fields, index) + return int(value) if value else 0 + + +def _escape_path(value: str) -> str: + return value.replace(" ", "\\ ") + + +def _unescape_path(value: str) -> str: + return value.replace("\\ ", " ") diff --git a/src/libtmux/resurrect/storage.py b/src/libtmux/resurrect/storage.py new file mode 100644 index 000000000..9bc3869bc --- /dev/null +++ b/src/libtmux/resurrect/storage.py @@ -0,0 +1,120 @@ +"""Archive history, rotation, and last-pointer helpers.""" + +from __future__ import annotations + +import pathlib +import typing as t +from dataclasses import dataclass + +from libtmux._internal.types import StrPath +from libtmux.resurrect.archives import WorkspaceArchive, write_archive + +LastPointerKind: t.TypeAlias = t.Literal["copy", "symlink"] +"""How the ``last`` pointer was written.""" + + +@dataclass(frozen=True, slots=True) +class ArchiveSnapshot: + """Paths touched while writing an archive snapshot.""" + + archive_path: pathlib.Path + last_path: pathlib.Path + last_kind: LastPointerKind + removed_paths: tuple[pathlib.Path, ...] + + +def write_archive_snapshot( + archive: WorkspaceArchive, + directory: StrPath, + *, + basename: str = "workspace", + keep: int = 5, + portable_last: bool = False, +) -> ArchiveSnapshot: + """Write a timestamped archive, rotate old copies, and update ``last``. + + Examples + -------- + >>> import datetime + >>> import pathlib + >>> from libtmux.resurrect.archives import WorkspaceArchive + >>> archive = WorkspaceArchive( + ... saved_at=datetime.datetime(2026, 7, 4, tzinfo=datetime.timezone.utc), + ... sessions=(), + ... ) + >>> target = pathlib.Path(request.getfixturevalue("tmp_path")) + >>> snapshot = write_archive_snapshot(archive, target, portable_last=True) + >>> snapshot.archive_path.name + 'workspace-20260704T000000Z.json' + >>> snapshot.last_kind + 'copy' + """ + base_dir = pathlib.Path(directory) + base_dir.mkdir(parents=True, exist_ok=True) + timestamp = archive.saved_at.strftime("%Y%m%dT%H%M%SZ") + archive_path = write_archive(archive, base_dir / f"{basename}-{timestamp}.json") + last_path = base_dir / "last.json" + last_kind = _write_last_pointer( + archive, + archive_path=archive_path, + last_path=last_path, + portable_last=portable_last, + ) + removed = _rotate( + base_dir, + basename=basename, + keep=keep, + protected=archive_path, + ) + return ArchiveSnapshot( + archive_path=archive_path, + last_path=last_path, + last_kind=last_kind, + removed_paths=removed, + ) + + +def _write_last_pointer( + archive: WorkspaceArchive, + *, + archive_path: pathlib.Path, + last_path: pathlib.Path, + portable_last: bool, +) -> LastPointerKind: + if portable_last: + write_archive(archive, last_path) + return "copy" + + try: + _replace_relative_symlink(archive_path, last_path) + except OSError: + write_archive(archive, last_path) + return "copy" + return "symlink" + + +def _replace_relative_symlink(source: pathlib.Path, link: pathlib.Path) -> None: + tmp_link = link.with_name(f".{link.name}.tmp") + tmp_link.unlink(missing_ok=True) + tmp_link.symlink_to(source.name) + tmp_link.replace(link) + + +def _rotate( + directory: pathlib.Path, + *, + basename: str, + keep: int, + protected: pathlib.Path | None = None, +) -> tuple[pathlib.Path, ...]: + if keep <= 0: + return () + archives = sorted(directory.glob(f"{basename}-*.json")) + retained = keep + if protected is not None and protected in archives: + archives.remove(protected) + retained -= 1 + expired = archives if retained <= 0 else archives[:-retained] + for path in expired: + path.unlink(missing_ok=True) + return tuple(expired) diff --git a/tests/test_resurrect_archives.py b/tests/test_resurrect_archives.py new file mode 100644 index 000000000..b29a4dad2 --- /dev/null +++ b/tests/test_resurrect_archives.py @@ -0,0 +1,1248 @@ +"""Tests for tmux-resurrect style workspace archives.""" + +from __future__ import annotations + +import datetime +import pathlib +import typing as t + +import pytest + +from libtmux.formats import FORMAT_SEPARATOR +from libtmux.resurrect.archives import ( + PaneArchive, + SessionArchive, + WindowArchive, + WorkspaceArchive, + capture_archive, + read_archive, + restore_archive, + write_archive, +) +from libtmux.resurrect.processes import ProcessRestorePolicy + +if t.TYPE_CHECKING: + from libtmux.server import Server + + +class _Cmd: + """Small command result test double.""" + + def __init__(self, stdout: list[str] | None = None) -> None: + self.stdout = stdout or [] + self.stderr: list[str] = [] + self.returncode = 0 + + +class _FakePane: + """Pane double used by restore tests.""" + + def __init__(self, calls: list[tuple[str, tuple[object, ...], dict[str, object]]]): + self._calls = calls + + def split( + self, + *, + start_directory: str | None = None, + shell: str | None = None, + attach: bool = False, + ) -> _FakePane: + """Record pane splits.""" + self._calls.append( + ( + "pane.split", + (), + { + "attach": attach, + "shell": shell, + "start_directory": start_directory, + }, + ), + ) + return _FakePane(self._calls) + + +class _FakeWindow: + """Window double used by restore tests.""" + + def __init__( + self, + calls: list[tuple[str, tuple[object, ...], dict[str, object]]], + *, + window_index: str = "0", + ) -> None: + self._calls = calls + self.window_index = window_index + self.active_pane = _FakePane(calls) + + def split( + self, + *, + start_directory: str | None = None, + shell: str | None = None, + attach: bool = False, + ) -> _FakePane: + """Record window splits.""" + self._calls.append( + ( + "window.split", + (), + { + "attach": attach, + "shell": shell, + "start_directory": start_directory, + }, + ), + ) + return _FakePane(self._calls) + + def select_layout(self, layout: str) -> _FakeWindow: + """Record layout selection.""" + self._calls.append(("window.select_layout", (layout,), {})) + return self + + +class _FakeSession: + """Session double used by restore tests.""" + + def __init__( + self, + calls: list[tuple[str, tuple[object, ...], dict[str, object]]], + ) -> None: + self._calls = calls + self.active_window = _FakeWindow(calls) + + def new_window( + self, + *, + window_name: str | None = None, + start_directory: str | None = None, + window_index: str = "", + window_shell: str | None = None, + attach: bool = False, + ) -> _FakeWindow: + """Record window creation.""" + self._calls.append( + ( + "session.new_window", + (), + { + "attach": attach, + "start_directory": start_directory, + "window_index": window_index, + "window_name": window_name, + "window_shell": window_shell, + }, + ), + ) + return _FakeWindow(self._calls, window_index=window_index) + + def select_window(self, target_window: str | int) -> _FakeWindow: + """Record active window selection.""" + self._calls.append(("session.select_window", (target_window,), {})) + return _FakeWindow(self._calls, window_index=str(target_window)) + + +class _FakeServer: + """Server double used by archive tests.""" + + def __init__( + self, + stdout: list[str] | None = None, + stdout_by_cmd: dict[str, list[str]] | None = None, + ) -> None: + self.stdout = stdout or [] + self.stdout_by_cmd = stdout_by_cmd or {} + self.calls: list[tuple[str, tuple[object, ...], dict[str, object]]] = [] + self.existing_sessions: set[str] = set() + + def cmd(self, cmd: str, *args: object, **kwargs: object) -> _Cmd: + """Record tmux commands.""" + self.calls.append((cmd, args, kwargs)) + if cmd == "new-session": + session_name = _arg_after(args, "-s") + if session_name is not None: + self.existing_sessions.add(session_name) + if cmd in self.stdout_by_cmd: + return _Cmd(self.stdout_by_cmd[cmd]) + return _Cmd(self.stdout if cmd == "list-panes" else []) + + def has_session(self, target_session: str) -> bool: + """Return configured session existence.""" + self.calls.append(("has_session", (target_session,), {})) + return target_session in self.existing_sessions + + def new_session( + self, + *, + session_name: str | None = None, + start_directory: str | None = None, + window_name: str | None = None, + window_command: str | None = None, + ) -> _FakeSession: + """Record session creation.""" + if session_name is not None: + self.existing_sessions.add(session_name) + self.calls.append( + ( + "new_session", + (), + { + "session_name": session_name, + "start_directory": start_directory, + "window_command": window_command, + "window_name": window_name, + }, + ), + ) + return _FakeSession(self.calls) + + +class _ProcessProvider: + """Process command provider test double.""" + + def __init__(self, commands: dict[int, str]) -> None: + self.commands = commands + self.captured_pids: list[int] = [] + + def capture(self, pid: int) -> str | None: + """Record pid capture and return a configured command.""" + self.captured_pids.append(pid) + return self.commands.get(pid) + + +class RestoreReuseMissingWindowCase(t.NamedTuple): + """Case for restoring an archived window missing from a reused session.""" + + test_id: str + panes: tuple[PaneArchive, ...] + expected_split_count: int + expected_active_target: str + + +class RestoreReuseWindowNameCase(t.NamedTuple): + """Case for restoring an archived name onto an existing reused window.""" + + test_id: str + window_name: str + expected_target: str + + +class RestoreAutomaticRenameCase(t.NamedTuple): + """Case for preserving automatic rename while restoring window names.""" + + test_id: str + automatic_rename: bool + expected_option: str + + +class RestorePaneTargetMapCase(t.NamedTuple): + """Case for mapping archived panes onto restored tmux pane targets.""" + + test_id: str + archived_indexes: tuple[int, int] + restored_pane_ids: tuple[str, str] + expected_title_targets: tuple[str, str] + expected_active_target: str + + +RESTORE_REUSE_MISSING_WINDOW_CASES = ( + RestoreReuseMissingWindowCase( + test_id="single_pane", + panes=( + PaneArchive( + index=0, + active=True, + current_command="tail", + current_path="/logs", + title="main", + ), + ), + expected_split_count=0, + expected_active_target="alpha:2.0", + ), + RestoreReuseMissingWindowCase( + test_id="multi_pane", + panes=( + PaneArchive( + index=0, + active=False, + current_command="tail", + current_path="/logs", + title="main", + ), + PaneArchive( + index=1, + active=True, + current_command="less", + current_path="/logs", + full_command="less error.log", + title="secondary", + ), + ), + expected_split_count=1, + expected_active_target="alpha:2.1", + ), +) + +RESTORE_REUSE_WINDOW_NAME_CASES = ( + RestoreReuseWindowNameCase( + test_id="existing_window", + window_name="logs", + expected_target="alpha:0", + ), +) + +RESTORE_AUTOMATIC_RENAME_CASES = ( + RestoreAutomaticRenameCase( + test_id="automatic_rename_on", + automatic_rename=True, + expected_option="on", + ), +) + +RESTORE_PANE_TARGET_MAP_CASES = ( + RestorePaneTargetMapCase( + test_id="pane_base_index_mismatch", + archived_indexes=(1, 2), + restored_pane_ids=("%10", "%11"), + expected_title_targets=("%10", "%11"), + expected_active_target="%11", + ), +) + + +def _arg_after(args: tuple[object, ...], flag: str) -> str | None: + try: + value = args[args.index(flag) + 1] + except (ValueError, IndexError): + return None + return str(value) + + +def test_capture_archive_groups_tmux_pane_rows() -> None: + """capture_archive() groups list-panes rows by session and window.""" + saved_at = datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc) + server = _FakeServer( + [ + FORMAT_SEPARATOR.join( + [ + "alpha", + "0", + "editor", + "tiled", + "1", + "0", + "1", + "vim", + "/workspace", + ], + ), + FORMAT_SEPARATOR.join( + [ + "alpha", + "0", + "editor", + "tiled", + "1", + "1", + "0", + "bash", + "/workspace", + ], + ), + ], + ) + + archive = capture_archive(t.cast("Server", server), saved_at=saved_at) + + assert archive.saved_at == saved_at + assert len(archive.sessions) == 1 + assert archive.sessions[0].name == "alpha" + assert archive.sessions[0].windows[0].name == "editor" + assert archive.sessions[0].windows[0].panes == ( + PaneArchive( + index=0, + active=True, + current_command="vim", + current_path="/workspace", + ), + PaneArchive( + index=1, + active=False, + current_command="bash", + current_path="/workspace", + ), + ) + assert server.calls[0][0] == "list-panes" + + +def test_capture_archive_records_captured_capabilities() -> None: + """capture_archive() declares the parity features captured in the archive.""" + archive = capture_archive(t.cast("Server", _FakeServer())) + + assert archive.capabilities == ( + "sessions", + "windows", + "panes", + "window-order", + "pane-order", + "working-directories", + "layouts", + "active-windows", + "active-panes", + "pane-current-command", + "pane-full-command", + "pane-titles", + "window-flags", + "automatic-rename", + "grouped-sessions", + "alternate-windows", + "active-sessions", + "alternate-sessions", + "history-size", + ) + + +def test_capture_archive_uses_configured_libtmux_separator() -> None: + """capture_archive() uses a tmux-version-safe separator.""" + server = _FakeServer() + + capture_archive(t.cast("Server", server)) + + format_arg = server.calls[0][1][2] + assert isinstance(format_arg, str) + assert server.calls[0] == ("list-panes", ("-a", "-F", format_arg), {}) + assert FORMAT_SEPARATOR in format_arg + if FORMAT_SEPARATOR != "\x1f": + assert "\x1f" not in format_arg + + +def test_capture_archive_accepts_trailing_format_separator() -> None: + """capture_archive() accepts the output shape used by neo format strings.""" + saved_at = datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc) + server = _FakeServer( + [ + FORMAT_SEPARATOR.join( + [ + "alpha", + "0", + "editor", + "tiled", + "1", + "0", + "1", + "vim", + "/workspace", + "", + ], + ), + ], + ) + + archive = capture_archive(t.cast("Server", server), saved_at=saved_at) + + assert archive.sessions[0].windows[0].panes == ( + PaneArchive( + index=0, + active=True, + current_command="vim", + current_path="/workspace", + ), + ) + + +def test_capture_archive_records_focus_and_window_metadata() -> None: + """capture_archive() records focus, grouping, pane, and window metadata.""" + saved_at = datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc) + server = _FakeServer( + stdout_by_cmd={ + "list-clients": [ + FORMAT_SEPARATOR.join(["alpha", "beta", ""]), + ], + "list-panes": [ + FORMAT_SEPARATOR.join( + [ + "alpha", + "1", + "editor", + "tiled", + "1", + "*Z", + "0", + "1", + "vim", + "/workspace", + "src", + "42", + "", + ], + ), + ], + "list-sessions": [ + FORMAT_SEPARATOR.join(["alpha", "1", "work", ""]), + ], + "show-window-options": ["off"], + }, + ) + + archive = capture_archive(t.cast("Server", server), saved_at=saved_at) + + assert archive.active_session_name == "alpha" + assert archive.alternate_session_name == "beta" + session = archive.sessions[0] + assert session.group_name == "work" + assert session.active_window_index == 1 + assert session.alternate_window_index is None + window = session.windows[0] + assert window.flags == "*Z" + assert window.automatic_rename is False + pane = window.panes[0] + assert pane.title == "src" + assert pane.history_size == 42 + + +def test_capture_archive_uses_process_command_provider() -> None: + """capture_archive() records provider-supplied full process commands.""" + saved_at = datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc) + provider = _ProcessProvider({123: "vim pyproject.toml"}) + server = _FakeServer( + stdout_by_cmd={ + "list-panes": [ + FORMAT_SEPARATOR.join( + [ + "alpha", + "0", + "editor", + "tiled", + "1", + "*", + "0", + "1", + "123", + "vim", + "/workspace", + "src", + "42", + "", + ], + ), + ], + }, + ) + + archive = capture_archive( + t.cast("Server", server), + process_provider=provider, + saved_at=saved_at, + ) + + pane = archive.sessions[0].windows[0].panes[0] + assert pane.full_command == "vim pyproject.toml" + assert provider.captured_pids == [123] + + +def test_write_read_archive_round_trips_json(tmp_path: pathlib.Path) -> None: + """write_archive() persists JSON that read_archive() loads back.""" + archive = WorkspaceArchive( + saved_at=datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc), + sessions=( + SessionArchive( + name="alpha", + windows=( + WindowArchive( + index=0, + name="editor", + layout="tiled", + active=True, + panes=( + PaneArchive( + index=0, + active=True, + current_command="vim", + current_path="/workspace", + ), + ), + ), + ), + ), + ), + ) + archive_path = tmp_path / "workspace.json" + + assert write_archive(archive, archive_path) == archive_path + assert read_archive(archive_path) == archive + + +def test_read_archive_accepts_existing_minimal_json(tmp_path: pathlib.Path) -> None: + """read_archive() remains compatible with pre-parity archive JSON.""" + archive_path = tmp_path / "minimal.json" + archive_path.write_text( + """{ + "capabilities": [ + "sessions", + "windows", + "panes" + ], + "format_version": "libtmux.resurrect.archive.v1", + "saved_at": "2026-07-04T12:00:00+00:00", + "sessions": [ + { + "name": "alpha", + "windows": [ + { + "active": true, + "index": 0, + "layout": "tiled", + "name": "editor", + "panes": [ + { + "active": true, + "current_command": "vim", + "current_path": "/workspace", + "index": 0 + } + ] + } + ] + } + ] +} +""", + encoding="utf-8", + ) + + archive = read_archive(archive_path) + + assert archive.active_session_name is None + assert archive.alternate_session_name is None + assert archive.sessions[0].group_name is None + assert archive.sessions[0].active_window_index is None + assert archive.sessions[0].alternate_window_index is None + assert archive.sessions[0].windows[0].flags == "" + assert archive.sessions[0].windows[0].automatic_rename is None + assert archive.sessions[0].windows[0].panes[0].title == "" + assert archive.sessions[0].windows[0].panes[0].full_command == "" + assert archive.sessions[0].windows[0].panes[0].history_size == 0 + assert archive.sessions[0].windows[0].panes[0].contents == () + + +def test_write_read_archive_preserves_extended_metadata( + tmp_path: pathlib.Path, +) -> None: + """write_archive() persists the richer parity metadata.""" + archive = WorkspaceArchive( + saved_at=datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc), + active_session_name="alpha", + alternate_session_name="beta", + sessions=( + SessionArchive( + name="alpha", + group_name="work", + active_window_index=1, + alternate_window_index=0, + windows=( + WindowArchive( + index=1, + name="editor", + layout="tiled", + active=True, + flags="*Z", + automatic_rename=False, + panes=( + PaneArchive( + index=0, + active=True, + current_command="vim", + current_path="/workspace", + title="src", + full_command="vim pyproject.toml", + history_size=42, + contents=("hello", "world"), + ), + ), + ), + ), + ), + ), + ) + archive_path = tmp_path / "extended.json" + + assert write_archive(archive, archive_path) == archive_path + assert read_archive(archive_path) == archive + + +def test_restore_archive_recreates_windows_panes_and_layout() -> None: + """restore_archive() recreates sessions, windows, panes, and selections.""" + archive = WorkspaceArchive( + saved_at=datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc), + sessions=( + SessionArchive( + name="alpha", + windows=( + WindowArchive( + index=0, + name="editor", + layout="tiled", + active=False, + panes=( + PaneArchive( + index=0, + active=True, + current_command="vim", + current_path="/workspace", + ), + PaneArchive( + index=1, + active=False, + current_command="bash", + current_path="/workspace", + ), + ), + ), + WindowArchive( + index=2, + name="logs", + layout="even-horizontal", + active=True, + panes=( + PaneArchive( + index=0, + active=True, + current_command="tail", + current_path="/logs", + ), + ), + ), + ), + ), + ), + ) + server = _FakeServer() + + restored = restore_archive(archive, t.cast("Server", server)) + + assert len(restored) == 1 + assert ( + "new_session", + (), + { + "session_name": "alpha", + "start_directory": "/workspace", + "window_command": "vim", + "window_name": "editor", + }, + ) in server.calls + assert ( + "window.split", + (), + { + "attach": False, + "shell": None, + "start_directory": "/workspace", + }, + ) in server.calls + assert ( + "session.new_window", + (), + { + "attach": False, + "start_directory": "/logs", + "window_index": "2", + "window_name": "logs", + "window_shell": "tail", + }, + ) in server.calls + assert ("window.select_layout", ("tiled",), {}) in server.calls + assert ("window.select_layout", ("even-horizontal",), {}) in server.calls + assert ("select-pane", (), {"target": "alpha:0.0"}) in server.calls + assert ("session.select_window", (2,), {}) in server.calls + + +def test_restore_archive_reuses_existing_topology() -> None: + """restore_archive(on_exists='reuse') creates only missing topology.""" + archive = WorkspaceArchive( + saved_at=datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc), + sessions=( + SessionArchive( + name="alpha", + windows=( + WindowArchive( + index=0, + name="editor", + layout="tiled", + active=True, + panes=( + PaneArchive( + index=0, + active=True, + current_command="vim", + current_path="/workspace", + ), + PaneArchive( + index=1, + active=False, + current_command="bash", + current_path="/workspace", + ), + ), + ), + WindowArchive( + index=2, + name="logs", + layout="even-horizontal", + active=False, + panes=( + PaneArchive( + index=0, + active=True, + current_command="tail", + current_path="/logs", + ), + ), + ), + ), + ), + ), + ) + server = _FakeServer( + stdout_by_cmd={ + "list-panes": ["0"], + "list-windows": ["0"], + }, + ) + server.existing_sessions.add("alpha") + + assert restore_archive(archive, t.cast("Server", server), on_exists="reuse") == [] + + assert not any(call[0] == "new_session" for call in server.calls) + assert [call[0] for call in server.calls].count("new-window") == 1 + assert [call[0] for call in server.calls].count("split-window") == 1 + assert ( + "new-window", + ("-d", "-t", "alpha:2", "-n", "logs", "-c", "/logs", "tail"), + {}, + ) in server.calls + assert ( + "split-window", + ("-d", "-t", "alpha:0", "-c", "/workspace"), + {}, + ) in server.calls + + +@pytest.mark.parametrize( + "case", + RESTORE_REUSE_WINDOW_NAME_CASES, + ids=[case.test_id for case in RESTORE_REUSE_WINDOW_NAME_CASES], +) +def test_restore_archive_reuses_existing_window_name( + case: RestoreReuseWindowNameCase, +) -> None: + """restore_archive(on_exists='reuse') renames existing reused windows.""" + archive = WorkspaceArchive( + saved_at=datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc), + sessions=( + SessionArchive( + name="alpha", + windows=( + WindowArchive( + index=0, + name=case.window_name, + layout="tiled", + active=True, + panes=( + PaneArchive( + index=0, + active=True, + current_command="tail", + current_path="/logs", + ), + ), + ), + ), + ), + ), + ) + server = _FakeServer(stdout_by_cmd={"list-panes": ["0"], "list-windows": ["0"]}) + server.existing_sessions.add("alpha") + + assert restore_archive(archive, t.cast("Server", server), on_exists="reuse") == [] + assert ( + "rename-window", + ("-t", case.expected_target, case.window_name), + {}, + ) in server.calls + + +@pytest.mark.parametrize( + "case", + RESTORE_AUTOMATIC_RENAME_CASES, + ids=[case.test_id for case in RESTORE_AUTOMATIC_RENAME_CASES], +) +def test_restore_archive_preserves_automatic_rename_after_name_restore( + case: RestoreAutomaticRenameCase, +) -> None: + """restore_archive() restores automatic rename after manual rename.""" + archive = WorkspaceArchive( + saved_at=datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc), + sessions=( + SessionArchive( + name="alpha", + windows=( + WindowArchive( + index=0, + name="logs", + layout="", + active=True, + automatic_rename=case.automatic_rename, + panes=( + PaneArchive( + index=0, + active=True, + current_command="tail", + current_path="/logs", + ), + ), + ), + ), + ), + ), + ) + server = _FakeServer() + + restore_archive(archive, t.cast("Server", server)) + rename_call: tuple[str, tuple[object, ...], dict[str, object]] = ( + "rename-window", + ("-t", "alpha:0", "logs"), + {}, + ) + option_call: tuple[str, tuple[object, ...], dict[str, object]] = ( + "set-window-option", + ("-t", "alpha:0", "automatic-rename", case.expected_option), + {}, + ) + assert rename_call in server.calls + assert option_call in server.calls + assert server.calls.index(rename_call) < server.calls.index(option_call) + + +@pytest.mark.parametrize( + "case", + RESTORE_PANE_TARGET_MAP_CASES, + ids=[case.test_id for case in RESTORE_PANE_TARGET_MAP_CASES], +) +def test_restore_archive_maps_pane_metadata_to_restored_targets( + case: RestorePaneTargetMapCase, +) -> None: + """restore_archive() maps archived pane indexes to restored pane targets.""" + archive = WorkspaceArchive( + saved_at=datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc), + sessions=( + SessionArchive( + name="alpha", + windows=( + WindowArchive( + index=0, + name="editor", + layout="", + active=True, + panes=( + PaneArchive( + index=case.archived_indexes[0], + active=False, + current_command="vim", + current_path="/workspace", + title="left", + ), + PaneArchive( + index=case.archived_indexes[1], + active=True, + current_command="less", + current_path="/workspace", + title="right", + ), + ), + ), + ), + ), + ), + ) + server = _FakeServer(stdout_by_cmd={"list-panes": list(case.restored_pane_ids)}) + + restore_archive(archive, t.cast("Server", server)) + + for title, target in zip( + ("left", "right"), + case.expected_title_targets, + strict=True, + ): + assert ("select-pane", ("-T", title), {"target": target}) in server.calls + assert ("select-pane", (), {"target": case.expected_active_target}) in server.calls + + +@pytest.mark.parametrize( + "case", + RESTORE_REUSE_MISSING_WINDOW_CASES, + ids=[case.test_id for case in RESTORE_REUSE_MISSING_WINDOW_CASES], +) +def test_restore_archive_reuses_missing_window_state( + case: RestoreReuseMissingWindowCase, +) -> None: + """restore_archive(on_exists='reuse') restores newly created window state.""" + archive = WorkspaceArchive( + saved_at=datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc), + sessions=( + SessionArchive( + name="alpha", + windows=( + WindowArchive( + index=2, + name="logs", + layout="even-horizontal", + active=True, + automatic_rename=False, + panes=case.panes, + ), + ), + ), + ), + ) + server = _FakeServer(stdout_by_cmd={"list-windows": ["0"]}) + server.existing_sessions.add("alpha") + + assert restore_archive(archive, t.cast("Server", server), on_exists="reuse") == [] + + assert ( + "new-window", + ("-d", "-t", "alpha:2", "-n", "logs", "-c", "/logs", "tail"), + {}, + ) in server.calls + assert ( + len( + [ + call + for call in server.calls + if call[0] == "split-window" and call[1][0:3] == ("-d", "-t", "alpha:2") + ], + ) + == case.expected_split_count + ) + assert ( + "select-layout", + ("-t", "alpha:2", "even-horizontal"), + {}, + ) in server.calls + assert ( + "set-window-option", + ("-t", "alpha:2", "automatic-rename", "off"), + {}, + ) in server.calls + for pane in case.panes: + assert ( + "select-pane", + ("-T", pane.title), + {"target": f"alpha:2.{pane.index}"}, + ) in server.calls + assert ("select-pane", (), {"target": case.expected_active_target}) in server.calls + + +def test_restore_archive_uses_process_restore_policy() -> None: + """restore_archive() only replays pane commands allowed by policy.""" + archive = WorkspaceArchive( + saved_at=datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc), + sessions=( + SessionArchive( + name="alpha", + windows=( + WindowArchive( + index=0, + name="server", + layout="", + active=True, + panes=( + PaneArchive( + index=0, + active=True, + current_command="node", + current_path="/workspace", + full_command="node server.js", + ), + ), + ), + ), + ), + ), + ) + default_server = _FakeServer() + all_server = _FakeServer() + + restore_archive(archive, t.cast("Server", default_server)) + restore_archive( + archive, + t.cast("Server", all_server), + process_policy=ProcessRestorePolicy.from_options(":all:"), + ) + + assert ( + "new_session", + (), + { + "session_name": "alpha", + "start_directory": "/workspace", + "window_command": None, + "window_name": "server", + }, + ) in default_server.calls + assert ( + "new_session", + (), + { + "session_name": "alpha", + "start_directory": "/workspace", + "window_command": "node server.js", + "window_name": "server", + }, + ) in all_server.calls + + +def test_restore_archive_replays_focus_and_window_metadata() -> None: + """restore_archive() replays focus, title, zoom, and window option metadata.""" + archive = WorkspaceArchive( + saved_at=datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc), + active_session_name="alpha", + alternate_session_name="beta", + sessions=( + SessionArchive( + name="alpha", + active_window_index=2, + alternate_window_index=0, + windows=( + WindowArchive( + index=0, + name="editor", + layout="tiled", + active=False, + flags="-", + automatic_rename=True, + panes=( + PaneArchive( + index=0, + active=True, + current_command="vim", + current_path="/workspace", + title="src", + ), + ), + ), + WindowArchive( + index=2, + name="logs", + layout="even-horizontal", + active=True, + flags="*Z", + automatic_rename=False, + panes=( + PaneArchive( + index=0, + active=True, + current_command="tail", + current_path="/logs", + title="log", + ), + ), + ), + ), + ), + ), + ) + server = _FakeServer() + + restore_archive(archive, t.cast("Server", server)) + + assert ( + "set-window-option", + ("-t", "alpha:0", "automatic-rename", "on"), + {}, + ) in server.calls + assert ( + "set-window-option", + ("-t", "alpha:2", "automatic-rename", "off"), + {}, + ) in server.calls + assert ("select-pane", ("-T", "src"), {"target": "alpha:0.0"}) in server.calls + assert ("select-pane", ("-T", "log"), {"target": "alpha:2.0"}) in server.calls + assert ("resize-pane", ("-Z",), {"target": "alpha:2"}) in server.calls + assert ("select-window", ("-t", "alpha:0"), {}) in server.calls + assert ("session.select_window", (2,), {}) in server.calls + assert ("switch-client", ("-t", "beta"), {}) in server.calls + assert ("switch-client", ("-t", "alpha"), {}) in server.calls + + +def test_restore_archive_recreates_grouped_sessions_once() -> None: + """restore_archive() recreates grouped sessions without duplicating windows.""" + windows = ( + WindowArchive( + index=1, + name="editor", + layout="tiled", + active=True, + panes=( + PaneArchive( + index=0, + active=True, + current_command="vim", + current_path="/workspace", + ), + ), + ), + ) + archive = WorkspaceArchive( + saved_at=datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc), + sessions=( + SessionArchive( + name="alpha", + group_name="shared", + active_window_index=1, + windows=windows, + ), + SessionArchive( + name="beta", + group_name="shared", + active_window_index=1, + windows=windows, + ), + ), + ) + server = _FakeServer() + + restore_archive(archive, t.cast("Server", server)) + + assert ( + "new_session", + (), + { + "session_name": "alpha", + "start_directory": "/workspace", + "window_command": "vim", + "window_name": "editor", + }, + ) in server.calls + assert ("new-session", ("-d", "-s", "beta", "-t", "alpha"), {}) in server.calls + assert ("session.select_window", (1,), {}) in server.calls + assert ("select-window", ("-t", "beta:1"), {}) in server.calls + assert [ + call + for call in server.calls + if call[0] in {"new_session", "session.new_window"} + ] == [ + ( + "new_session", + (), + { + "session_name": "alpha", + "start_directory": "/workspace", + "window_command": "vim", + "window_name": "editor", + }, + ), + ] diff --git a/tests/test_resurrect_continuum.py b/tests/test_resurrect_continuum.py new file mode 100644 index 000000000..be762ea0e --- /dev/null +++ b/tests/test_resurrect_continuum.py @@ -0,0 +1,338 @@ +"""Tests for headless tmux-continuum style autosave helpers.""" + +from __future__ import annotations + +import datetime +import pathlib +import typing as t + +from libtmux.formats import FORMAT_SEPARATOR +from libtmux.resurrect.archives import WorkspaceArchive, write_archive +from libtmux.resurrect.continuum import ( + DEFAULT_AUTOSAVE_INTERVAL, + AutosavePaths, + AutosaveState, + StartupRestoreDecision, + autosave_once, + default_autosave_paths, + next_autosave_at, + read_autosave_state, + should_autosave, + should_restore_on_startup, + startup_restore_once, + write_autosave_state, +) + +if t.TYPE_CHECKING: + from libtmux.server import Server + + +class _Cmd: + """Small command result test double.""" + + def __init__(self, stdout: list[str] | None = None) -> None: + self.stdout = stdout or [] + self.stderr: list[str] = [] + self.returncode = 0 + + +class _FakeServer: + """Server double that returns one captured pane.""" + + def __init__( + self, + *, + sessions: list[object] | None = None, + socket_name: str | None = None, + socket_path: pathlib.Path | None = None, + ) -> None: + self.sessions = sessions or [] + self.socket_name = socket_name + self.socket_path = socket_path + self.stdout = [ + FORMAT_SEPARATOR.join( + [ + "alpha", + "0", + "editor", + "tiled", + "1", + "0", + "1", + "vim", + "/workspace", + ], + ), + ] + self.calls: list[tuple[str, tuple[object, ...], dict[str, object]]] = [] + + def cmd(self, cmd: str, *args: object, **kwargs: object) -> _Cmd: + """Record tmux commands.""" + self.calls.append((cmd, args, kwargs)) + return _Cmd(self.stdout if cmd == "list-panes" else []) + + +def test_should_autosave_without_previous_timestamp() -> None: + """should_autosave() saves when there is no previous timestamp.""" + assert should_autosave(last_saved_at=None) + + +def test_should_autosave_after_interval_elapsed() -> None: + """should_autosave() saves only after the interval has elapsed.""" + last_saved_at = datetime.datetime( + 2026, + 7, + 4, + 12, + tzinfo=datetime.timezone.utc, + ) + now = last_saved_at + DEFAULT_AUTOSAVE_INTERVAL + + assert should_autosave(last_saved_at=last_saved_at, now=now) + assert not should_autosave( + last_saved_at=last_saved_at, + now=now - datetime.timedelta(seconds=1), + ) + + +def test_next_autosave_at_adds_interval() -> None: + """next_autosave_at() returns the next due timestamp.""" + last_saved_at = datetime.datetime( + 2026, + 7, + 4, + 12, + tzinfo=datetime.timezone.utc, + ) + + assert next_autosave_at(last_saved_at) == ( + last_saved_at + DEFAULT_AUTOSAVE_INTERVAL + ) + assert next_autosave_at(None) is None + + +def test_read_write_autosave_state_round_trips(tmp_path: pathlib.Path) -> None: + """write_autosave_state() persists state read_autosave_state() loads.""" + state_path = tmp_path / "state.json" + state = AutosaveState( + last_saved_at=datetime.datetime( + 2026, + 7, + 4, + 12, + tzinfo=datetime.timezone.utc, + ), + last_archive_path=tmp_path / "archive.json", + save_count=2, + ) + + assert write_autosave_state(state, state_path) == state_path + assert read_autosave_state(state_path) == state + + +def test_default_autosave_paths_are_socket_aware(tmp_path: pathlib.Path) -> None: + """default_autosave_paths() keeps independent tmux servers separate.""" + alpha = default_autosave_paths( + t.cast("Server", _FakeServer(socket_name="alpha")), + tmp_path, + ) + beta = default_autosave_paths( + t.cast("Server", _FakeServer(socket_name="beta")), + tmp_path, + ) + + assert isinstance(alpha, AutosavePaths) + assert alpha.archive_path.parent == tmp_path + assert alpha.state_path.parent == tmp_path + assert alpha.archive_path != alpha.state_path + assert alpha.archive_path != beta.archive_path + assert alpha.state_path != beta.state_path + + +def test_default_autosave_paths_hash_socket_paths(tmp_path: pathlib.Path) -> None: + """default_autosave_paths() does not embed raw socket paths in filenames.""" + socket_path = pathlib.Path("/tmp/tmux-user/private.sock") + + paths = default_autosave_paths( + t.cast("Server", _FakeServer(socket_path=socket_path)), + tmp_path, + ) + + assert "private.sock" not in paths.archive_path.name + assert "tmux-user" not in paths.archive_path.name + assert paths.archive_path.suffix == ".json" + assert paths.state_path.name.endswith(".state.json") + + +def test_autosave_once_skips_until_interval_elapses(tmp_path: pathlib.Path) -> None: + """autosave_once() skips before the configured interval has elapsed.""" + now = datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc) + state_path = tmp_path / "state.json" + archive_path = tmp_path / "archive.json" + write_autosave_state( + AutosaveState( + last_saved_at=now - datetime.timedelta(minutes=1), + last_archive_path=archive_path, + save_count=1, + ), + state_path, + ) + server = _FakeServer() + + result = autosave_once( + t.cast("Server", server), + archive_path=archive_path, + state_path=state_path, + now=now, + ) + + assert not result.saved + assert result.reason == "interval_not_elapsed" + assert not archive_path.exists() + assert server.calls == [] + + +def test_autosave_once_writes_archive_and_state(tmp_path: pathlib.Path) -> None: + """autosave_once() captures an archive and advances state when due.""" + now = datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc) + state_path = tmp_path / "state.json" + archive_path = tmp_path / "archive.json" + server = _FakeServer() + + result = autosave_once( + t.cast("Server", server), + archive_path=archive_path, + state_path=state_path, + now=now, + ) + + assert result.saved + assert result.reason == "saved" + assert archive_path.exists() + assert read_autosave_state(state_path) == AutosaveState( + last_saved_at=now, + last_archive_path=archive_path, + save_count=1, + ) + assert server.calls[0][0] == "list-panes" + + +def test_should_restore_on_startup_allows_fresh_opt_in_restore( + tmp_path: pathlib.Path, +) -> None: + """should_restore_on_startup() allows fresh, explicit startup restore.""" + now = datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc) + + decision = should_restore_on_startup( + enabled=True, + halt_file=tmp_path / "halt", + session_count=0, + another_server_running=False, + tmux_started_at=now - datetime.timedelta(seconds=2), + now=now, + ) + + assert decision == StartupRestoreDecision(allowed=True, reason="restore_allowed") + + +def test_should_restore_on_startup_reports_specific_vetoes( + tmp_path: pathlib.Path, +) -> None: + """should_restore_on_startup() reports why restore is skipped.""" + now = datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc) + halt_file = tmp_path / "halt" + halt_file.write_text("", encoding="utf-8") + + assert ( + should_restore_on_startup( + enabled=False, + halt_file=tmp_path / "missing", + session_count=0, + another_server_running=False, + tmux_started_at=now, + now=now, + ).reason + == "restore_disabled" + ) + assert ( + should_restore_on_startup( + enabled=True, + halt_file=halt_file, + session_count=0, + another_server_running=False, + tmux_started_at=now, + now=now, + ).reason + == "halt_file_present" + ) + assert ( + should_restore_on_startup( + enabled=True, + halt_file=None, + session_count=0, + another_server_running=True, + tmux_started_at=now, + now=now, + ).reason + == "another_server_running" + ) + assert ( + should_restore_on_startup( + enabled=True, + halt_file=None, + session_count=1, + another_server_running=False, + tmux_started_at=now, + now=now, + ).reason + == "sessions_exist" + ) + assert ( + should_restore_on_startup( + enabled=True, + halt_file=None, + session_count=0, + another_server_running=False, + tmux_started_at=now - datetime.timedelta(minutes=1), + now=now, + ).reason + == "startup_window_elapsed" + ) + + +def test_startup_restore_once_restores_when_allowed( + tmp_path: pathlib.Path, +) -> None: + """startup_restore_once() restores only after startup guards pass.""" + now = datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc) + archive_path = tmp_path / "workspace.json" + write_archive(WorkspaceArchive(saved_at=now, sessions=()), archive_path) + + result = startup_restore_once( + t.cast("Server", _FakeServer()), + archive_path, + enabled=True, + tmux_started_at=now, + now=now, + ) + + assert result.restored + assert result.reason == "restored" + assert result.decision == StartupRestoreDecision( + allowed=True, + reason="restore_allowed", + ) + + +def test_startup_restore_once_skips_when_guard_vetoes( + tmp_path: pathlib.Path, +) -> None: + """startup_restore_once() returns the guard reason without reading archive.""" + result = startup_restore_once( + t.cast("Server", _FakeServer(sessions=[object()])), + tmp_path / "missing.json", + enabled=True, + ) + + assert not result.restored + assert result.reason == "sessions_exist" diff --git a/tests/test_resurrect_file.py b/tests/test_resurrect_file.py new file mode 100644 index 000000000..21d998478 --- /dev/null +++ b/tests/test_resurrect_file.py @@ -0,0 +1,97 @@ +"""Tests for tmux-resurrect tab-file conversion.""" + +from __future__ import annotations + +import datetime + +from libtmux.resurrect.archives import ( + PaneArchive, + SessionArchive, + WindowArchive, + WorkspaceArchive, +) +from libtmux.resurrect.resurrect_file import ( + archive_from_resurrect_file, + archive_to_resurrect_file, +) + + +def test_resurrect_file_converter_round_trips_core_rows() -> None: + """tmux-resurrect tab rows convert to and from native archives.""" + saved_at = datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc) + archive = WorkspaceArchive( + saved_at=saved_at, + active_session_name="alpha", + alternate_session_name="beta", + sessions=( + SessionArchive( + name="alpha", + active_window_index=0, + windows=( + WindowArchive( + index=0, + name="editor", + layout="tiled", + active=True, + flags="*Z", + automatic_rename=False, + panes=( + PaneArchive( + index=0, + active=True, + current_command="vim", + current_path="/work space", + title="src", + full_command="vim pyproject.toml", + ), + ), + ), + ), + ), + ), + ) + + text = archive_to_resurrect_file(archive) + restored = archive_from_resurrect_file(text, saved_at=saved_at) + + assert "state\talpha\tbeta" in text + assert "pane\talpha\t0\t1\t:*Z\t0\tsrc\t:/work\\ space" in text + assert "window\talpha\t0\t:editor\t1\t:*Z\ttiled\toff" in text + assert restored == archive + + +def test_resurrect_file_imports_upstream_grouped_rows() -> None: + """tmux-resurrect grouped follower rows import without duplicated windows.""" + saved_at = datetime.datetime(2026, 7, 4, 12, tzinfo=datetime.timezone.utc) + text = "\n".join( + ( + "grouped_session\tbeta\talpha\t:0\t:1", + ( + "pane\talpha\t1\t1\t:*Z\t0\tsrc\t:/work\\ space\t1\tvim" + "\t:vim pyproject.toml" + ), + "window\talpha\t1\t:editor\t1\t:*Z\ttiled\toff", + "state\talpha\tbeta", + "", + ), + ) + + archive = archive_from_resurrect_file(text, saved_at=saved_at) + exported = archive_to_resurrect_file(archive) + + assert archive.active_session_name == "alpha" + assert archive.alternate_session_name == "beta" + alpha = archive.sessions[0] + beta = archive.sessions[1] + assert alpha.name == "alpha" + assert alpha.group_name == "alpha" + assert alpha.windows[0].name == "editor" + assert alpha.windows[0].panes[0].current_path == "/work space" + assert alpha.windows[0].panes[0].full_command == "vim pyproject.toml" + assert beta.name == "beta" + assert beta.group_name == "alpha" + assert beta.alternate_window_index == 0 + assert beta.active_window_index == 1 + assert beta.windows == () + assert "grouped_session\tbeta\talpha\t:0\t:1" in exported + assert "pane\tbeta" not in exported diff --git a/tests/test_resurrect_processes.py b/tests/test_resurrect_processes.py new file mode 100644 index 000000000..5ab02a8fe --- /dev/null +++ b/tests/test_resurrect_processes.py @@ -0,0 +1,225 @@ +"""Tests for tmux-resurrect style process restore policies.""" + +from __future__ import annotations + +import pathlib +import typing as t + +import pytest + +import libtmux.resurrect.processes as resurrect_processes +from libtmux.resurrect.processes import ( + CompositeProcessCommandProvider, + ProcessRestorePolicy, + ProcessRestoreRule, + ProcfsProcessCommandProvider, + PsProcessCommandProvider, + default_process_command_provider, +) + + +class _Provider: + """Process command provider test double.""" + + def __init__(self, command: str | None) -> None: + self.command = command + + def capture(self, pid: int) -> str | None: + """Return the configured command.""" + return self.command + + +class ProcfsEntry(t.NamedTuple): + """Procfs process entry test fixture.""" + + pid: int + ppid: int + pgrp: int + tpgid: int + command: tuple[str, ...] + + +class ProcfsCaptureCase(t.NamedTuple): + """Case for procfs process command capture.""" + + test_id: str + pid: int + entries: tuple[ProcfsEntry, ...] + expected_command: str | None + + +class DefaultProviderFallbackCase(t.NamedTuple): + """Case for default process provider fallback behavior.""" + + test_id: str + pid: int + ps_command: str + + +PROCFS_CAPTURE_CASES = ( + ProcfsCaptureCase( + test_id="direct_cmdline", + pid=123, + entries=( + ProcfsEntry( + pid=123, + ppid=1, + pgrp=123, + tpgid=123, + command=("python", "-m", "http.server"), + ), + ), + expected_command="python -m http.server", + ), + ProcfsCaptureCase( + test_id="foreground_child", + pid=100, + entries=( + ProcfsEntry( + pid=100, + ppid=1, + pgrp=100, + tpgid=200, + command=("-zsh",), + ), + ProcfsEntry( + pid=200, + ppid=100, + pgrp=200, + tpgid=200, + command=("vim", "README.md"), + ), + ), + expected_command="vim README.md", + ), + ProcfsCaptureCase( + test_id="missing_pid", + pid=456, + entries=(), + expected_command=None, + ), +) + +DEFAULT_PROVIDER_FALLBACK_CASES = ( + DefaultProviderFallbackCase( + test_id="pane_shell_from_ps", + pid=1234, + ps_command="-zsh", + ), +) + + +def _write_procfs_entry(proc_root: pathlib.Path, entry: ProcfsEntry) -> None: + proc_dir = proc_root / str(entry.pid) + proc_dir.mkdir() + (proc_dir / "cmdline").write_bytes( + b"\0".join(part.encode() for part in entry.command) + b"\0", + ) + (proc_dir / "stat").write_text( + ( + f"{entry.pid} ({entry.command[0]}) S {entry.ppid} {entry.pgrp} " + f"0 34816 {entry.tpgid} 0 0 0 0 0 0 0 0 20 0 1 0 0\n" + ), + encoding="utf-8", + ) + + +def test_process_restore_policy_matches_defaults_and_inline_rules() -> None: + """ProcessRestorePolicy parses tmux-resurrect-style process options.""" + default_policy = ProcessRestorePolicy.from_options(None) + custom_policy = ProcessRestorePolicy.from_options( + "'python->uv run python *' '~rails server->rails server *' 'git log'", + ) + + assert default_policy.resolve_command("vim pyproject.toml") == "vim pyproject.toml" + assert default_policy.resolve_command("node server.js") is None + assert custom_policy.resolve_command("python -m http.server 8000") == ( + "uv run python -m http.server 8000" + ) + assert custom_policy.resolve_command("git log --oneline") == "git log --oneline" + assert ( + custom_policy.resolve_command( + "/rubies/bin/ruby script/rails server -p 3000", + ) + == "rails server -p 3000" + ) + assert ( + ProcessRestorePolicy.from_options(":all:").resolve_command( + "node server.js", + ) + == "node server.js" + ) + assert ProcessRestorePolicy.from_options("false").resolve_command("vim") is None + + +def test_process_restore_rule_handles_unparseable_commands() -> None: + """ProcessRestoreRule treats invalid shell syntax as a non-match.""" + rule = ProcessRestoreRule("vim") + + assert rule.matches('"unterminated') is False + assert rule.resolve("vim notes.md") == "vim notes.md" + + +@pytest.mark.parametrize( + "case", + PROCFS_CAPTURE_CASES, + ids=[case.test_id for case in PROCFS_CAPTURE_CASES], +) +def test_procfs_process_command_provider_reads_cmdline( + case: ProcfsCaptureCase, + tmp_path: pathlib.Path, +) -> None: + """ProcfsProcessCommandProvider resolves pane PIDs to active commands.""" + for entry in case.entries: + _write_procfs_entry(tmp_path, entry) + + provider = ProcfsProcessCommandProvider(tmp_path) + + assert provider.capture(case.pid) == case.expected_command + assert provider.capture(0) is None + + +def test_composite_process_command_provider_uses_first_command() -> None: + """CompositeProcessCommandProvider falls through missing providers.""" + provider = CompositeProcessCommandProvider( + ( + _Provider(None), + _Provider("vim README.md"), + ), + ) + + assert provider.capture(123) == "vim README.md" + + +def test_default_process_command_provider_and_missing_ps() -> None: + """Default providers are headless and tolerate unavailable ps binaries.""" + provider = default_process_command_provider() + + assert hasattr(provider, "capture") + assert PsProcessCommandProvider(ps_bin="/missing-ps").capture(123) is None + + +@pytest.mark.parametrize( + "case", + DEFAULT_PROVIDER_FALLBACK_CASES, + ids=[case.test_id for case in DEFAULT_PROVIDER_FALLBACK_CASES], +) +def test_default_process_command_provider_skips_ps_fallback( + case: DefaultProviderFallbackCase, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """default_process_command_provider() does not archive ps shell fallback.""" + monkeypatch.setattr( + resurrect_processes, + "ProcfsProcessCommandProvider", + lambda: _Provider(None), + ) + monkeypatch.setattr( + resurrect_processes, + "PsProcessCommandProvider", + lambda: _Provider(case.ps_command), + ) + + provider = default_process_command_provider() + + assert provider.capture(case.pid) is None diff --git a/tests/test_resurrect_storage.py b/tests/test_resurrect_storage.py new file mode 100644 index 000000000..d81f095f9 --- /dev/null +++ b/tests/test_resurrect_storage.py @@ -0,0 +1,130 @@ +"""Tests for archive snapshot storage helpers.""" + +from __future__ import annotations + +import datetime +import pathlib +import typing as t + +import pytest + +from libtmux.resurrect.archives import WorkspaceArchive, read_archive +from libtmux.resurrect.storage import ( + ArchiveSnapshot, + write_archive_snapshot, +) + + +def _archive(minute: int) -> WorkspaceArchive: + return WorkspaceArchive( + saved_at=datetime.datetime( + 2026, + 7, + 4, + 12, + minute, + tzinfo=datetime.timezone.utc, + ), + sessions=(), + ) + + +class SnapshotRotationCase(t.NamedTuple): + """Case for timestamped archive rotation.""" + + test_id: str + existing_minutes: tuple[int, ...] + current_minute: int + keep: int + expected_names: tuple[str, ...] + + +SNAPSHOT_ROTATION_CASES = ( + SnapshotRotationCase( + test_id="older_current", + existing_minutes=(2,), + current_minute=0, + keep=1, + expected_names=("workspace-20260704T120000Z.json",), + ), +) + + +def test_write_archive_snapshot_updates_portable_last( + tmp_path: pathlib.Path, +) -> None: + """write_archive_snapshot() can write last.json as a portable copy.""" + archive = _archive(0) + + snapshot = write_archive_snapshot(archive, tmp_path, portable_last=True) + + assert isinstance(snapshot, ArchiveSnapshot) + assert snapshot.archive_path.name == "workspace-20260704T120000Z.json" + assert snapshot.archive_path.exists() + assert snapshot.last_path == tmp_path / "last.json" + assert snapshot.last_kind == "copy" + assert not snapshot.last_path.is_symlink() + assert read_archive(snapshot.last_path) == archive + + +def test_write_archive_snapshot_updates_last_pointer( + tmp_path: pathlib.Path, +) -> None: + """write_archive_snapshot() writes a relative symlink or portable fallback.""" + archive = _archive(1) + + snapshot = write_archive_snapshot(archive, tmp_path) + + assert snapshot.archive_path.exists() + assert snapshot.last_path.exists() + if snapshot.last_kind == "symlink": + assert snapshot.last_path.is_symlink() + assert snapshot.last_path.readlink() == pathlib.Path(snapshot.archive_path.name) + else: + assert read_archive(snapshot.last_path) == archive + + +def test_write_archive_snapshot_rotates_old_archives( + tmp_path: pathlib.Path, +) -> None: + """write_archive_snapshot() removes older timestamped snapshots.""" + first = write_archive_snapshot(_archive(0), tmp_path, keep=2, portable_last=True) + second = write_archive_snapshot(_archive(1), tmp_path, keep=2, portable_last=True) + third = write_archive_snapshot(_archive(2), tmp_path, keep=2, portable_last=True) + + assert first.archive_path in third.removed_paths + assert not first.archive_path.exists() + assert second.archive_path.exists() + assert third.archive_path.exists() + assert sorted(path.name for path in tmp_path.glob("workspace-*.json")) == [ + "workspace-20260704T120100Z.json", + "workspace-20260704T120200Z.json", + ] + + +@pytest.mark.parametrize( + "case", + SNAPSHOT_ROTATION_CASES, + ids=[case.test_id for case in SNAPSHOT_ROTATION_CASES], +) +def test_write_archive_snapshot_keeps_current_archive_during_rotation( + case: SnapshotRotationCase, + tmp_path: pathlib.Path, +) -> None: + """write_archive_snapshot() keeps the archive last.json points at.""" + for minute in case.existing_minutes: + write_archive_snapshot(_archive(minute), tmp_path, keep=case.keep) + + archive = _archive(case.current_minute) + snapshot = write_archive_snapshot(archive, tmp_path, keep=case.keep) + + assert snapshot.archive_path.exists() + assert snapshot.last_path.exists() + assert snapshot.archive_path not in snapshot.removed_paths + if snapshot.last_kind == "symlink": + assert snapshot.last_path.readlink() == pathlib.Path(snapshot.archive_path.name) + else: + assert read_archive(snapshot.last_path) == archive + assert sorted(path.name for path in tmp_path.glob("workspace-*.json")) == list( + case.expected_names, + )