feat: deliver installable local alpha sessions

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc

src/tamq/cli.py

@@ -67,7 +67,21 @@ def ensure_service() -> bool:
             if asyncio.run(ping()):
                 return True
             time.sleep(0.05)
-        return False
+    return False
+
+
+def preflight_runtime_paths() -> None:
+    """Ensure local state/runtime parents are available before tmux mutation."""
+    parents = {db_path().parent, socket_path().parent, pid_path().parent, lock_path().parent}
+    for parent in parents:
+        parent.mkdir(parents=True, exist_ok=True)
+        if not os.access(parent, os.W_OK | os.X_OK):
+            raise OSError(f"runtime directory is not writable: {parent}")
+
+
+def attach_session(session: str) -> int:
+    command = ["tmux", "switch-client" if os.environ.get("TMUX") else "attach-session", "-t", session]
+    return subprocess.run(command, check=False).returncode
 
 
 def build_parser() -> argparse.ArgumentParser:
@@ -83,9 +97,9 @@ def build_parser() -> argparse.ArgumentParser:
 
     start = subparsers.add_parser("start", help="start or reuse a tamq endpoint")
     start.add_argument("repos", nargs="*", help="gita-registered repository slugs")
-    start.add_argument("--cmd", default="codex", help="agent command for new windows")
+    start.add_argument("--command", "--cmd", dest="agent_command", default="codex", help="agent command for new windows (default: codex)")
     start.add_argument("--detach", action="store_true", help="detach after startup")
-    start.add_argument("--no-service", action="store_true", help="skip service startup")
+    start.add_argument("--no-service", action="store_true", help="open tmux windows without service registration or messaging")
     attach = subparsers.add_parser("attach", help="attach to the managed tmux session")
     attach.add_argument("--session", default="tamq", help="tmux session name")
 
@@ -168,8 +182,7 @@ def main(argv: list[str] | None = None) -> int:
         except KeyboardInterrupt: return 0
         return 0
     if args.command == "attach":
-        result = subprocess.run(["tmux", "attach-session", "-t", args.session], check=False)
-        return result.returncode
+        return attach_session(args.session)
     if args.command == "completion":
         print(completion_script(args.shell), end="")
         return 0
@@ -189,30 +202,56 @@ def main(argv: list[str] | None = None) -> int:
         finally:
             store.close()
     if args.command == "start":
+        manager = TmuxManager()
+        try:
+            preflight_runtime_paths()
+            launch_plan = manager.preflight(args.repos, args.agent_command)
+        except (TmuxError, OSError) as exc:
+            print(f"tamq: {exc}", file=sys.stderr)
+            return 2
         if not args.no_service and not ensure_service():
             print("tamq service failed to start; use --no-service to open repos without messaging", file=sys.stderr)
             return 1
         try:
-            endpoint = TmuxManager().ensure(args.repos, args.cmd)
+            endpoint = manager.ensure_plan(launch_plan)
         except (TmuxError, OSError) as exc:
             print(f"tamq: {exc}", file=sys.stderr)
             return 2
-        registration = asyncio.run(request({"op": "register", "endpoint_id": endpoint.endpoint_id, "instance_id": endpoint.instance_key, "pid": endpoint.pid, "session": endpoint.session, "repos": endpoint.repos}))
-        if not registration.get("ok"):
-            print(f"tamq: endpoint registration failed: {registration.get('error', 'unknown error')}", file=sys.stderr)
-            return 1
-        control = ControlModeClient(endpoint.session)
-        queue_store = Store(db_path())
-        try:
-            control.start()
-            delivered = InputBroker(queue_store, BrokerIdentity(endpoint.instance_key, endpoint.repos[0])).deliver_pending(
-                control, window_for_repo=lambda repo: f"{endpoint.session}:{repo}"
-            )
-        finally:
-            control.close()
-            queue_store.close()
-        print(json.dumps({"endpoint_id": endpoint.endpoint_id, "instance_id": endpoint.instance_key, "session": endpoint.session, "repos": endpoint.repos, "delivered": delivered}))
-        return 0
+        delivered = 0
+        registered = False
+        if not args.no_service:
+            try:
+                registration = asyncio.run(request({"op": "register", "endpoint_id": endpoint.endpoint_id, "instance_id": endpoint.instance_key, "pid": endpoint.pid, "session": endpoint.session, "repos": endpoint.repos}))
+                if not registration.get("ok"):
+                    raise RuntimeError(f"endpoint registration failed: {registration.get('error', 'unknown error')}")
+                registered = True
+                control = ControlModeClient(endpoint.session)
+                queue_store = Store(db_path())
+                try:
+                    control.start()
+                    delivered = InputBroker(queue_store, BrokerIdentity(endpoint.instance_key, endpoint.repos[0])).deliver_pending(
+                        control, window_for_repo=lambda repo: f"{endpoint.session}:{repo}"
+                    )
+                finally:
+                    control.close()
+                    queue_store.close()
+            except (OSError, RuntimeError, ValueError, json.JSONDecodeError) as exc:
+                manager.rollback(endpoint)
+                print(f"tamq: {exc}", file=sys.stderr)
+                return 1
+        summary = {
+            "endpoint_id": endpoint.endpoint_id,
+            "instance_id": endpoint.instance_key,
+            "session": endpoint.session,
+            "repos": endpoint.repos,
+            "delivered": delivered,
+            "service": not args.no_service,
+            "messaging": registered,
+        }
+        print(json.dumps(summary), flush=True)
+        if args.detach:
+            return 0
+        return attach_session(endpoint.session)
     store = Store(db_path())
     try:
         advisory = history_advisory(store)

src/tamq/control.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import os
 import subprocess
+from collections.abc import Sequence
 from dataclasses import dataclass
 
 
@@ -11,13 +13,19 @@ class ControlModeError(RuntimeError):
 @dataclass
 class ControlModeClient:
     session: str
+    tmux_command: Sequence[str] | None = None
     process: subprocess.Popen[str] | None = None
 
     def start(self) -> None:
         if self.process is not None:
             return
+        socket_name = os.environ.get("TAMQ_TMUX_SOCKET")
+        command = tuple(
+            self.tmux_command
+            or (["tmux", "-L", socket_name] if socket_name else ["tmux"])
+        )
         self.process = subprocess.Popen(
-            ["tmux", "-C", "attach-session", "-t", self.session],
+            [*command, "-C", "attach-session", "-t", self.session],
             stdin=subprocess.PIPE,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
@@ -37,7 +45,13 @@ class ControlModeClient:
 
     def close(self) -> None:
         if self.process is not None:
-            self.process.terminate()
+            process = self.process
+            process.terminate()
+            try:
+                process.wait(timeout=1)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait(timeout=1)
             self.process = None
 
 

src/tamq/service.py

@@ -62,7 +62,7 @@ class Service:
         """Continuously inject pending messages into registered live endpoints."""
         while not stopped.is_set():
             try:
-                await asyncio.to_thread(self._deliver_once)
+                self._deliver_once()
             except Exception:
                 # A disappearing tmux session must not take down the queue.
                 pass

src/tamq/store.py

@@ -57,6 +57,11 @@ class Store:
 
     def register_endpoint(self, endpoint_id: str, pid: int, session: str, repos: list[str]) -> None:
         import json
+        self.db.execute(
+            "UPDATE endpoints SET disconnected_at=strftime('%s','now') "
+            "WHERE pid=? AND session=? AND endpoint_id<>? AND disconnected_at IS NULL",
+            (pid, session, endpoint_id),
+        )
         self.db.execute(
             "INSERT INTO endpoints(endpoint_id,pid,session,repos,connected_at,disconnected_at) VALUES(?,?,?,?,strftime('%s','now'),NULL) "
             "ON CONFLICT(endpoint_id) DO UPDATE SET pid=excluded.pid, session=excluded.session, repos=excluded.repos, connected_at=excluded.connected_at, disconnected_at=NULL",
@@ -73,17 +78,21 @@ class Store:
         self.db.commit()
 
     def endpoints(self) -> list[sqlite3.Row]:
-        return list(self.db.execute("SELECT * FROM endpoints ORDER BY endpoint_id"))
+        return list(
+            self.db.execute(
+                "SELECT * FROM endpoints WHERE disconnected_at IS NULL ORDER BY endpoint_id"
+            )
+        )
 
     def endpoint(self, endpoint_id: str) -> sqlite3.Row | None:
         row = self.db.execute("SELECT * FROM endpoints WHERE endpoint_id=? AND disconnected_at IS NULL", (endpoint_id,)).fetchone()
         if row is not None:
             return row
         if endpoint_id.startswith("tmux-amq-"):
-            try:
-                pid = int(endpoint_id.removeprefix("tmux-amq-").split("-", 1)[0])
-            except ValueError:
+            visible_pid = endpoint_id.removeprefix("tmux-amq-")
+            if not visible_pid.isdigit():
                 return None
+            pid = int(visible_pid)
             return self.db.execute("SELECT * FROM endpoints WHERE pid=? AND disconnected_at IS NULL ORDER BY connected_at DESC LIMIT 1", (pid,)).fetchone()
         return None
 

src/tamq/tmux.py

@@ -2,9 +2,12 @@ from __future__ import annotations
 
 import os
 import shlex
+import shutil
 from uuid import uuid4
 import subprocess
+from collections.abc import Sequence
 from dataclasses import dataclass
+from pathlib import Path
 
 from .registry import repository_paths, validate_targets
 
@@ -19,6 +22,8 @@ class Endpoint:
     pid: int
     repos: list[str]
     instance_id: str = ""
+    created_session: bool = False
+    created_windows: tuple[str, ...] = ()
 
     @property
     def endpoint_id(self) -> str:
@@ -29,48 +34,154 @@ class Endpoint:
         return self.instance_id or self.endpoint_id
 
 
+@dataclass(frozen=True)
+class LaunchPlan:
+    repos: tuple[str, ...]
+    paths: dict[str, str]
+    command: tuple[str, ...]
+
+
 class TmuxManager:
-    def __init__(self, session: str = "tamq"):
+    def __init__(
+        self,
+        session: str = "tamq",
+        *,
+        tmux_command: Sequence[str] | None = None,
+        tamq_command: Sequence[str] | None = None,
+    ):
         self.session = session
+        socket_name = os.environ.get("TAMQ_TMUX_SOCKET")
+        self.tmux_command = tuple(tmux_command or (["tmux", "-L", socket_name] if socket_name else ["tmux"]))
+        self.tamq_command = tuple(tamq_command or ["tamq"])
 
     def _run(self, *args: str, check: bool = True) -> str:
-        result = subprocess.run(["tmux", *args], text=True, capture_output=True, check=False)
+        result = subprocess.run([*self.tmux_command, *args], text=True, capture_output=True, check=False)
         if check and result.returncode:
             raise TmuxError(result.stderr.strip() or "tmux command failed")
         return result.stdout.strip()
 
-    def ensure(self, repos: list[str], command: str = "codex") -> Endpoint:
-        if not repos:
-            raise TmuxError("at least one repository is required")
-        try:
-            validate_targets(repos)
-        except Exception as exc:
-            raise TmuxError(str(exc)) from exc
-        result = subprocess.run(["tmux", "has-session", "-t", self.session], capture_output=True)
+    def _existing_session_identity(self) -> tuple[int, str] | None:
+        result = subprocess.run(
+            [*self.tmux_command, "has-session", "-t", self.session],
+            capture_output=True,
+        )
         if result.returncode != 0:
-            self._run("new-session", "-d", "-s", self.session, "-n", "__tamq_boot", "-c", os.getcwd(), "sleep", "infinity")
+            return None
         pid_text = self._run("display-message", "-p", "-t", self.session, "#{pid}")
         try:
             pid = int(pid_text)
         except ValueError as exc:
             raise TmuxError(f"unable to determine tmux PID: {pid_text!r}") from exc
-        endpoint = Endpoint(self.session, pid, list(dict.fromkeys(repos)), f"tmux-amq-{pid}-{uuid4().hex[:12]}")
+        instance_id = self._run(
+            "show-options", "-v", "-t", self.session, "@tamq_instance_id", check=False,
+        )
+        managed = self._run(
+            "show-options", "-v", "-t", self.session, "@tamq_managed", check=False,
+        )
+        if managed != "1" and not instance_id.startswith(f"tmux-amq-{pid}-"):
+            raise TmuxError(
+                f"tmux session already exists and is not managed by tamq: {self.session}"
+            )
+        return pid, instance_id
+
+    def preflight(self, repos: list[str], command: str = "codex") -> LaunchPlan:
+        if not repos:
+            raise TmuxError("at least one repository is required")
+        if not self.tmux_command or shutil.which(self.tmux_command[0]) is None:
+            raise TmuxError("tmux is required; install it and ensure 'tmux' is on PATH")
+        if shutil.which("gita") is None:
+            raise TmuxError("gita is required; install it and ensure 'gita' is on PATH")
+        try:
+            command_parts = tuple(shlex.split(command))
+        except ValueError as exc:
+            raise TmuxError(f"invalid agent command: {exc}") from exc
+        if not command_parts:
+            raise TmuxError("agent command must not be empty")
+        if shutil.which(command_parts[0]) is None:
+            raise TmuxError(f"agent command not found on PATH: {command_parts[0]}")
+        unique_repos = tuple(dict.fromkeys(repos))
+        try:
+            validate_targets(list(unique_repos))
+        except Exception as exc:
+            raise TmuxError(str(exc)) from exc
         paths = repository_paths()
-        windows = self._run("list-windows", "-t", self.session, "-F", "#{window_name}").splitlines()
-        for index, repo in enumerate(endpoint.repos):
-            if repo not in windows:
+        selected_paths: dict[str, str] = {}
+        for repo in unique_repos:
+            try:
+                repo_path = paths[repo]
+            except KeyError as exc:
+                raise TmuxError(f"gita path missing for registered repository: {repo}") from exc
+            if not Path(repo_path).is_dir():
+                raise TmuxError(f"gita path is not a directory for registered repository {repo}: {repo_path}")
+            selected_paths[repo] = repo_path
+        self._existing_session_identity()
+        return LaunchPlan(unique_repos, selected_paths, command_parts)
+
+    def ensure(self, repos: list[str], command: str = "codex") -> Endpoint:
+        return self.ensure_plan(self.preflight(repos, command))
+
+    def ensure_plan(self, plan: LaunchPlan) -> Endpoint:
+        created_session = False
+        created_windows: list[str] = []
+        try:
+            identity = self._existing_session_identity()
+            if identity is None:
+                first_repo = plan.repos[0]
+                self._run(
+                    "new-session", "-d", "-s", self.session, "-n", "__tamq_boot",
+                    "-c", plan.paths[first_repo],
+                )
+                created_session = True
+                pid_text = self._run("display-message", "-p", "-t", self.session, "#{pid}")
+                try:
+                    pid = int(pid_text)
+                except ValueError as exc:
+                    raise TmuxError(f"unable to determine tmux PID: {pid_text!r}") from exc
+                instance_id = ""
+            else:
+                pid, instance_id = identity
+            endpoint_id = f"tmux-amq-{pid}"
+            if not instance_id.startswith(f"{endpoint_id}-"):
+                instance_id = f"{endpoint_id}-{uuid4().hex[:12]}"
+                self._run("set-option", "-t", self.session, "@tamq_instance_id", instance_id)
+            self._run("set-option", "-t", self.session, "@tamq_managed", "1")
+            windows = self._run("list-windows", "-t", self.session, "-F", "#{window_name}").splitlines()
+            for index, repo in enumerate(plan.repos):
+                if repo in windows:
+                    continue
                 if index == 0 and "__tamq_boot" in windows:
+                    if not created_session:
+                        self._run("respawn-pane", "-k", "-t", f"{self.session}:__tamq_boot", "-c", plan.paths[repo])
                     self._run("rename-window", "-t", f"{self.session}:__tamq_boot", repo)
                     windows[windows.index("__tamq_boot")] = repo
                 else:
-                    try:
-                        repo_path = paths[repo]
-                    except KeyError as exc:
-                        raise TmuxError(f"gita path missing for registered repository: {repo}") from exc
-                    self._run("new-window", "-d", "-t", self.session, "-n", repo, "-c", repo_path, "sleep", "infinity")
-                tap = ["tamq", "tap", "--repo", repo, "--endpoint", endpoint.endpoint_id, "--", command]
+                    self._run("new-window", "-d", "-t", self.session, "-n", repo, "-c", plan.paths[repo])
+                    windows.append(repo)
+                created_windows.append(repo)
+                tap = [
+                    *self.tamq_command,
+                    "tap", "--repo", repo, "--endpoint", endpoint_id, "--", *plan.command,
+                ]
                 self._run("send-keys", "-t", f"{self.session}:{repo}", shell_join(tap), "C-m")
-        return endpoint
+            self._run("select-window", "-t", f"{self.session}:{plan.repos[0]}")
+            return Endpoint(
+                self.session,
+                pid,
+                list(plan.repos),
+                instance_id,
+                created_session,
+                tuple(created_windows),
+            )
+        except Exception:
+            self.rollback(Endpoint(self.session, 0, list(plan.repos), created_session=created_session, created_windows=tuple(created_windows)))
+            raise
+
+    def rollback(self, endpoint: Endpoint) -> None:
+        if endpoint.created_session:
+            self._run("kill-session", "-t", endpoint.session, check=False)
+            return
+        for window in reversed(endpoint.created_windows):
+            self._run("kill-window", "-t", f"{endpoint.session}:{window}", check=False)
 
 
 def shell_join(parts: list[str]) -> str: