wip: feat: workspaces tmux support

workspaces.py

@@ -10,6 +10,7 @@
 # ]
 # ///
 
+from __future__ import annotations
 import os
 import re
 import shutil
@@ -18,7 +19,9 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import List, Optional, Tuple
 
+
 import typer
+
 from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from rich.console import Console
@@ -233,7 +236,7 @@ def read_workspace_readme(ws_dir: Path) -> Tuple[str, str]:
         return ws_dir.name, ""
 
     # First non-empty line must be '# ...' per spec
-    first_non_empty_idx = next((i for i, l in enumerate(lines) if l.strip()), None)
+    first_non_empty_idx = next((i for i, line in enumerate(lines) if line.strip()), None)
     if first_non_empty_idx is None:
         return ws_dir.name, ""
 
@@ -1087,5 +1090,262 @@ def status_workspace(
     if desc:
         console.print(Panel(desc, title="Workspace description"))
 
+def not_in_tmux() -> bool:
+    """Return True if not inside tmux or zellij."""
+    return not os.environ.get("TMUX") and not os.environ.get("ZELLIJ")
+
+
+def run_tmux(args: list[str], *, clear_tmux_env: bool = False) -> subprocess.CompletedProcess:
+    """Run a tmux command."""
+    env = os.environ.copy()
+    if clear_tmux_env:
+        env["TMUX"] = ""
+    return subprocess.run(
+        ["tmux", *args],
+        env=env,
+        check=False,
+    )
+
+
+def session_exists(session_name: str) -> bool:
+    """Check if the tmux session exists."""
+    result = run_tmux(["has-session", "-t", f"={session_name}"])
+    return result.returncode == 0
+
+
+def pick_project(base_dir: Path) -> Optional[str]:
+    """Use fzf to pick a subdirectory (project) from base_dir."""
+    if not base_dir.is_dir():
+        typer.echo(f"[ta] {base_dir} is not a directory.", err=True)
+        return None
+
+    subdirs = sorted(
+        [p.name for p in base_dir.iterdir() if p.is_dir()]
+    )
+    if not subdirs:
+        typer.echo(f"[ta] No subdirectories found in {base_dir}.", err=True)
+        return None
+
+    if not shutil.which("fzf"):
+        typer.echo("[ta] fzf not found in PATH.", err=True)
+        return None
+
+    proc = subprocess.run(
+        ["fzf", "--reverse", f"--header=Select project from {base_dir.name} >"],
+        input="\n".join(subdirs),
+        text=True,
+        capture_output=True,
+    )
+
+    if proc.returncode != 0:
+        # Cancelled or failed
+        return None
+
+    choice = proc.stdout.strip()
+    return choice or None
+
+
+def create_detached_session(
+    session_name: str,
+    path_name: Path,
+    start_mode: bool,
+) -> bool:
+    """
+    Create a detached session.
+
+    - If start_mode: just a single window.
+    - Else: split layout with nvim on top.
+    """
+    # Run tmux as if not in tmux (clear TMUX env)
+    if start_mode:
+        r = run_tmux(
+            ["new-session", "-Ad", "-s", session_name, "-c", str(path_name)],
+            clear_tmux_env=True,
+        )
+        return r.returncode == 0
+    else:
+        r = run_tmux(
+            ["new-session", "-Ad", "-s", session_name, "-c", str(path_name)],
+            clear_tmux_env=True,
+        )
+        if r.returncode != 0:
+            return False
+
+        r = run_tmux(
+            [
+                "split-window",
+                "-vb",
+                "-t",
+                session_name,
+                "-c",
+                str(path_name),
+                "-p",
+                "70",
+            ],
+            clear_tmux_env=True,
+        )
+        if r.returncode != 0:
+            return False
+
+        r = run_tmux(
+            [
+                "send-keys",
+                "-t",
+                session_name,
+                "nvim '+Telescope find_files'",
+                "Enter",
+            ],
+            clear_tmux_env=True,
+        )
+        return r.returncode == 0
+
+
+def create_if_needed_and_attach(
+    session_name: str,
+    path_name: Path,
+    start_mode: bool,
+) -> bool:
+    """
+    Mimic the bash logic:
+
+    - If not in tmux: `tmux new-session -As ... -c path_name`
+    - Else:
+        - If session doesn't exist: create_detached_session()
+        - Then `tmux switch-client -t session_name`
+    """
+    if not_in_tmux():
+        r = run_tmux(
+            ["new-session", "-As", session_name, "-c", str(path_name)],
+        )
+        return r.returncode == 0
+
+    # Inside tmux
+    if not session_exists(session_name):
+        if not create_detached_session(session_name, path_name, start_mode):
+            return False
+
+    r = run_tmux(["switch-client", "-t", session_name])
+    return r.returncode == 0
+
+
+def attach_to_first_session() -> None:
+    """Fallback: attach to first tmux session and open choose-tree."""
+    # Attach to the first listed session
+    list_proc = subprocess.run(
+        ["tmux", "list-sessions", "-F", "#{session_name}"],
+        text=True,
+        capture_output=True,
+    )
+    if list_proc.returncode != 0 or not list_proc.stdout.strip():
+        typer.echo("[ta] No tmux sessions found to attach to.", err=True)
+        raise typer.Exit(1)
+
+    first_session = list_proc.stdout.strip().splitlines()[0]
+
+    attach_proc = run_tmux(["attach-session", "-t", first_session])
+    if attach_proc.returncode != 0:
+        raise typer.Exit(attach_proc.returncode)
+
+    # After attach, show choose-tree (this will run in the attached session)
+    run_tmux(["choose-tree", "-Za"])
+
+tmux_app = typer.Typer(
+    name="tmux",
+    help="tmux commands",
+    add_completion=False,
+)
+
+@tmux_app.command('attach')
+def attach(
+    ctx: typer.Context,
+    workspace: Optional[str] = typer.Option(
+        None,
+        "--workspace",
+        "-w",
+        help=(
+            "Workspace directory name to show status for. "
+            "If omitted, uses the workspace containing the current directory."
+        ),
+    ),
+):
+    """
+    Attach or create a session for a repo in a workspace.
+    """
+    # ws_dir = get_workspace_dir(workspace)
+    _settings, _repos_dir, workspaces_dir = get_ctx_paths(ctx)
+    ws_dir = find_workspace_dir(workspaces_dir, workspace)
+    if not ws_dir:
+        console.print(f"[red]Workspace '{workspace}' does not exist at {ws_dir}[/red]")
+        raise typer.Exit(1)
+    # pick repo in workspace
+    repo = pick_repo_with_iterfzf([dir for dir in ws_dir.iterdir() if dir.is_dir()])
+    if not repo:
+        console.print("[red]No repo selected. Exiting.[/red]")
+        raise typer.Exit(1)
+    print(ws_dir, repo)
+    session_name = f'{ws_dir.name}|{repo.name}'
+    print(f'{ws_dir.name}|{repo.name}')
+    create_if_needed_and_attach(session_name, repo, False)
+
+
+app.add_typer(tmux_app)
+
+# @app.command()
+# def main(
+#     dir: Optional[Path] = typer.Argument(
+#         None,
+#         help="Base directory containing projects. If omitted, auto-attach or --start behavior.",
+#     ),
+#     start: bool = typer.Option(
+#         False,
+#         "--start",
+#         help="Start a new session based on the current directory.",
+#     ),
+# ):
+#     # Replicate initial "DIR=$1" and attach-or-start logic
+#     # If no args and no --start
+#     if dir is None and not start:
+#         if not_in_tmux():
+#             # Try to attach to an existing session
+#             attach_proc = run_tmux(["attach"])
+#             if attach_proc.returncode == 0:
+#                 # Successfully attached; exit like the bash script
+#                 raise typer.Exit(1)
+#             # If attach failed, fall through to start mode
+#             start = True
+#         else:
+#             # In tmux and no dir/start: nothing to do
+#             raise typer.Exit(1)
+#
+#     # Figure out session_name and path_name
+#     if start:
+#         path_name = Path.cwd()
+#         session_name = path_name.name.replace(".", "_")
+#     else:
+#         if dir is None:
+#             typer.echo("[ta] DIR argument is required unless --start is used.", err=True)
+#             raise typer.Exit(1)
+#
+#         dir = dir.expanduser().resolve()
+#         project = pick_project(dir)
+#         if not project:
+#             # cancelled or error
+#             raise typer.Exit(1)
+#
+#         session_name = project.replace(".", "_")
+#         path_name = (dir / project).resolve()
+#
+#     typer.echo(f'session name is "{session_name}"')
+#     typer.echo(f"path name is {path_name}")
+#
+#     if not session_name:
+#         raise typer.Exit(1)
+#
+#     # Try main attach/create flow; on failure, fall back
+#     ok = create_if_needed_and_attach(session_name, path_name, start_mode=start)
+#     if not ok:
+#         attach_to_first_session()
+#
+
 if __name__ == "__main__":
     app()