feat(web): Allow reconnection to running tasks after navigation (#38)

* fix(web): Add PATH for self-update SSH command

Non-interactive SSH sessions don't source profile files, so `cf` isn't
found when installed in ~/.local/bin. Prepend common install locations
to PATH before running the remote command.

* fix(web): Enable TTY for self-update SSH to show progress bars

README.md

@@ -407,7 +407,7 @@ Full `--help` output for each command. See the [Usage](#usage) table above for a
 ╰──────────────────────────────────────────────────────────────────────────────╯
 ╭─ Monitoring ─────────────────────────────────────────────────────────────────╮
 │ logs           Show service logs.                                            │
-│ ps             Show status of all services.                                  │
+│ ps             Show status of services.                                      │
 │ stats          Show overview statistics for hosts and services.              │
 ╰──────────────────────────────────────────────────────────────────────────────╯
 ╭─ Server ─────────────────────────────────────────────────────────────────────╮
@@ -904,11 +904,19 @@ Full `--help` output for each command. See the [Usage](#usage) table above for a
 <!-- ⚠️ This content is auto-generated by `markdown-code-runner`. -->
 ```yaml
 
- Usage: cf ps [OPTIONS]
+ Usage: cf ps [OPTIONS] [SERVICES]...
 
- Show status of all services.
+ Show status of services.
 
+ Without arguments: shows all services (same as --all). With service names:
+ shows only those services. With --host: shows services on that host.
+
+╭─ Arguments ──────────────────────────────────────────────────────────────────╮
+│   services      [SERVICES]...  Services to operate on                        │
+╰──────────────────────────────────────────────────────────────────────────────╯
 ╭─ Options ────────────────────────────────────────────────────────────────────╮
+│ --all     -a            Run on all services                                  │
+│ --host    -H      TEXT  Filter to services on this host                      │
 │ --config  -c      PATH  Path to config file                                  │
 │ --help    -h            Show this message and exit.                          │
 ╰──────────────────────────────────────────────────────────────────────────────╯

src/compose_farm/cli/common.py

@@ -144,17 +144,45 @@ def get_services(
     services: list[str],
     all_services: bool,
     config_path: Path | None,
+    *,
+    host: str | None = None,
+    default_all: bool = False,
 ) -> tuple[list[str], Config]:
     """Resolve service list and load config.
 
+    Handles three mutually exclusive selection methods:
+    - Explicit service names
+    - --all flag
+    - --host filter
+
+    Args:
+        services: Explicit service names
+        all_services: Whether --all was specified
+        config_path: Path to config file
+        host: Filter to services on this host
+        default_all: If True, default to all services when nothing specified (for ps)
+
     Supports "." as shorthand for the current directory name.
+
     """
+    validate_service_selection(services, all_services, host)
     config = load_config_or_exit(config_path)
 
+    if host is not None:
+        validate_host(config, host)
+        svc_list = [s for s in config.services if host in config.get_hosts(s)]
+        if not svc_list:
+            print_warning(f"No services configured for host [magenta]{host}[/]")
+            raise typer.Exit(0)
+        return svc_list, config
+
     if all_services:
         return list(config.services.keys()), config
+
     if not services:
-        print_error("Specify services or use [bold]--all[/]")
+        if default_all:
+            return list(config.services.keys()), config
+        print_error("Specify services or use [bold]--all[/] / [bold]--host[/]")
         raise typer.Exit(1)
 
     # Resolve "." to current directory name
@@ -286,6 +314,22 @@ def validate_host_for_service(cfg: Config, service: str, host: str) -> None:
         raise typer.Exit(1)
 
 
+def validate_service_selection(
+    services: list[str] | None,
+    all_services: bool,
+    host: str | None,
+) -> None:
+    """Validate that only one service selection method is used.
+
+    The three selection methods (explicit services, --all, --host) are mutually
+    exclusive. This ensures consistent behavior across all commands.
+    """
+    methods = sum([bool(services), all_services, host is not None])
+    if methods > 1:
+        print_error("Use only one of: service names, [bold]--all[/], or [bold]--host[/]")
+        raise typer.Exit(1)
+
+
 def run_host_operation(
     cfg: Config,
     svc_list: list[str],

src/compose_farm/cli/lifecycle.py

@@ -21,19 +21,16 @@ from compose_farm.cli.common import (
     maybe_regenerate_traefik,
     report_results,
     run_async,
-    run_host_operation,
 )
 from compose_farm.console import MSG_DRY_RUN, console, print_error, print_success
 from compose_farm.executor import run_on_services, run_sequential_on_services
 from compose_farm.operations import stop_orphaned_services, up_services
 from compose_farm.state import (
-    add_service_to_host,
     get_orphaned_services,
     get_service_host,
     get_services_needing_migration,
     get_services_not_in_state,
     remove_service,
-    remove_service_from_host,
 )
 
 
@@ -45,14 +42,7 @@ def up(
     config: ConfigOption = None,
 ) -> None:
     """Start services (docker compose up -d). Auto-migrates if host changed."""
-    svc_list, cfg = get_services(services or [], all_services, config)
-
-    # Per-host operation: run on specific host only
-    if host:
-        run_host_operation(cfg, svc_list, host, "up -d", "Starting", add_service_to_host)
-        return
-
-    # Normal operation: use up_services with migration logic
+    svc_list, cfg = get_services(services or [], all_services, config, host=host)
     results = run_async(up_services(cfg, svc_list, raw=True))
     maybe_regenerate_traefik(cfg, results)
     report_results(results)
@@ -72,7 +62,7 @@ def down(
     config: ConfigOption = None,
 ) -> None:
     """Stop services (docker compose down)."""
-    # Handle --orphaned flag
+    # Handle --orphaned flag (mutually exclusive with other selection methods)
     if orphaned:
         if services or all_services or host:
             print_error(
@@ -95,14 +85,7 @@ def down(
         report_results(results)
         return
 
-    svc_list, cfg = get_services(services or [], all_services, config)
-
-    # Per-host operation: run on specific host only
-    if host:
-        run_host_operation(cfg, svc_list, host, "down", "Stopping", remove_service_from_host)
-        return
-
-    # Normal operation
+    svc_list, cfg = get_services(services or [], all_services, config, host=host)
     raw = len(svc_list) == 1
     results = run_async(run_on_services(cfg, svc_list, "down", raw=raw))
 

src/compose_farm/cli/monitoring.py

@@ -20,9 +20,8 @@ from compose_farm.cli.common import (
     report_results,
     run_async,
     run_parallel_with_progress,
-    validate_host,
 )
-from compose_farm.console import console, print_error, print_warning
+from compose_farm.console import console
 from compose_farm.executor import run_command, run_on_services
 from compose_farm.state import get_services_needing_migration, group_services_by_host, load_state
 
@@ -127,22 +126,7 @@ def logs(
     config: ConfigOption = None,
 ) -> None:
     """Show service logs."""
-    if all_services and host is not None:
-        print_error("Cannot combine [bold]--all[/] and [bold]--host[/]")
-        raise typer.Exit(1)
-
-    cfg = load_config_or_exit(config)
-
-    # Determine service list based on options
-    if host is not None:
-        validate_host(cfg, host)
-        # Include services where host is in the list of configured hosts
-        svc_list = [s for s in cfg.services if host in cfg.get_hosts(s)]
-        if not svc_list:
-            print_warning(f"No services configured for host [magenta]{host}[/]")
-            return
-    else:
-        svc_list, cfg = get_services(services or [], all_services, config)
+    svc_list, cfg = get_services(services or [], all_services, config, host=host)
 
     # Default to fewer lines when showing multiple services
     many_services = all_services or host is not None or len(svc_list) > 1
@@ -156,11 +140,19 @@ def logs(
 
 @app.command(rich_help_panel="Monitoring")
 def ps(
+    services: ServicesArg = None,
+    all_services: AllOption = False,
+    host: HostOption = None,
     config: ConfigOption = None,
 ) -> None:
-    """Show status of all services."""
-    cfg = load_config_or_exit(config)
-    results = run_async(run_on_services(cfg, list(cfg.services.keys()), "ps"))
+    """Show status of services.
+
+    Without arguments: shows all services (same as --all).
+    With service names: shows only those services.
+    With --host: shows services on that host.
+    """
+    svc_list, cfg = get_services(services or [], all_services, config, host=host, default_all=True)
+    results = run_async(run_on_services(cfg, svc_list, "ps"))
     report_results(results)
 
 

src/compose_farm/compose.py

@@ -7,14 +7,14 @@ from __future__ import annotations
 
 import os
 import re
+import stat
 from dataclasses import dataclass
+from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 import yaml
 
 if TYPE_CHECKING:
-    from pathlib import Path
-
     from .config import Config
 
 # Port parsing constants
@@ -141,23 +141,42 @@ def _resolve_host_path(host_path: str, compose_dir: Path) -> str | None:
     return None  # Named volume
 
 
+def _is_socket(path: str) -> bool:
+    """Check if a path is a socket (e.g., SSH agent socket)."""
+    try:
+        return stat.S_ISSOCK(Path(path).stat().st_mode)
+    except (FileNotFoundError, PermissionError, OSError):
+        return False
+
+
 def _parse_volume_item(
     item: str | dict[str, Any],
     env: dict[str, str],
     compose_dir: Path,
 ) -> str | None:
-    """Parse a single volume item and return host path if it's a bind mount."""
+    """Parse a single volume item and return host path if it's a bind mount.
+
+    Skips socket paths (e.g., SSH_AUTH_SOCK) since they're machine-local
+    and shouldn't be validated on remote hosts.
+    """
+    host_path: str | None = None
+
     if isinstance(item, str):
         interpolated = _interpolate(item, env)
         parts = interpolated.split(":")
         if len(parts) >= _MIN_VOLUME_PARTS:
-            return _resolve_host_path(parts[0], compose_dir)
+            host_path = _resolve_host_path(parts[0], compose_dir)
     elif isinstance(item, dict) and item.get("type") == "bind":
         source = item.get("source")
         if source:
             interpolated = _interpolate(str(source), env)
-            return _resolve_host_path(interpolated, compose_dir)
-    return None
+            host_path = _resolve_host_path(interpolated, compose_dir)
+
+    # Skip sockets - they're machine-local (e.g., SSH agent)
+    if host_path and _is_socket(host_path):
+        return None
+
+    return host_path
 
 
 def parse_host_volumes(config: Config, service: str) -> list[str]:

src/compose_farm/web/app.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import sys
 from contextlib import asynccontextmanager, suppress
 from typing import TYPE_CHECKING
@@ -12,19 +13,35 @@ from pydantic import ValidationError
 
 from compose_farm.web.deps import STATIC_DIR, get_config
 from compose_farm.web.routes import actions, api, pages
+from compose_farm.web.streaming import TASK_TTL_SECONDS, cleanup_stale_tasks
 
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator
 
 
+async def _task_cleanup_loop() -> None:
+    """Periodically clean up stale completed tasks."""
+    while True:
+        await asyncio.sleep(TASK_TTL_SECONDS // 2)  # Run every 5 minutes
+        cleanup_stale_tasks()
+
+
 @asynccontextmanager
 async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     """Application lifespan handler."""
     # Startup: pre-load config (ignore errors - handled per-request)
     with suppress(ValidationError, FileNotFoundError):
         get_config()
+
+    # Start background cleanup task
+    cleanup_task = asyncio.create_task(_task_cleanup_loop())
+
     yield
-    # Shutdown: nothing to clean up
+
+    # Shutdown: cancel cleanup task
+    cleanup_task.cancel()
+    with suppress(asyncio.CancelledError):
+        await cleanup_task
 
 
 def create_app() -> FastAPI:

src/compose_farm/web/static/app.js

@@ -17,6 +17,10 @@ const editors = {};
 let monacoLoaded = false;
 let monacoLoading = false;
 
+// LocalStorage key prefix for active tasks (scoped by page)
+const TASK_KEY_PREFIX = 'cf_task:';
+const getTaskKey = () => TASK_KEY_PREFIX + window.location.pathname;
+
 // Language detection from file path
 const LANGUAGE_MAP = {
     'yaml': 'yaml', 'yml': 'yaml',
@@ -131,11 +135,18 @@ function initTerminal(elementId, taskId) {
     const { term, fitAddon } = createTerminal(container);
     const ws = createWebSocket(`/ws/terminal/${taskId}`);
 
+    const taskKey = getTaskKey();
     ws.onopen = () => {
         term.write(`${ANSI.DIM}[Connected]${ANSI.RESET}${ANSI.CRLF}`);
         setTerminalLoading(true);
+        localStorage.setItem(taskKey, taskId);
+    };
+    ws.onmessage = (event) => {
+        term.write(event.data);
+        if (event.data.includes('[Done]') || event.data.includes('[Failed]')) {
+            localStorage.removeItem(taskKey);
+        }
     };
-    ws.onmessage = (event) => term.write(event.data);
     ws.onclose = () => setTerminalLoading(false);
     ws.onerror = (error) => {
         term.write(`${ANSI.RED}[WebSocket Error]${ANSI.RESET}${ANSI.CRLF}`);
@@ -407,11 +418,33 @@ function initPage() {
     initSaveButton();
 }
 
+/**
+ * Attempt to reconnect to an active task from localStorage
+ */
+function tryReconnectToTask() {
+    const taskId = localStorage.getItem(getTaskKey());
+    if (!taskId) return;
+
+    // Wait for xterm to be loaded
+    const tryInit = (attempts) => {
+        if (typeof Terminal !== 'undefined' && typeof FitAddon !== 'undefined') {
+            expandTerminal();
+            initTerminal('terminal-output', taskId);
+        } else if (attempts > 0) {
+            setTimeout(() => tryInit(attempts - 1), 100);
+        }
+    };
+    tryInit(20);
+}
+
 // Initialize on page load
 document.addEventListener('DOMContentLoaded', function() {
     initPage();
     initKeyboardShortcuts();
 
+    // Try to reconnect to any active task
+    tryReconnectToTask();
+
     // Handle ?action= parameter (from command palette navigation)
     const params = new URLSearchParams(window.location.search);
     const action = params.get('action');
@@ -427,6 +460,8 @@ document.addEventListener('DOMContentLoaded', function() {
 document.body.addEventListener('htmx:afterSwap', function(evt) {
     if (evt.detail.target.id === 'main-content') {
         initPage();
+        // Try to reconnect when navigating back to dashboard
+        tryReconnectToTask();
     }
 });
 

src/compose_farm/web/streaming.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import os
+import time
 from typing import TYPE_CHECKING, Any
 
 from compose_farm.executor import build_ssh_command
@@ -25,6 +26,25 @@ CRLF = "\r\n"
 # In-memory task registry
 tasks: dict[str, dict[str, Any]] = {}
 
+# How long to keep completed tasks (10 minutes)
+TASK_TTL_SECONDS = 600
+
+
+def cleanup_stale_tasks() -> int:
+    """Remove tasks that completed more than TASK_TTL_SECONDS ago.
+
+    Returns the number of tasks removed.
+    """
+    cutoff = time.time() - TASK_TTL_SECONDS
+    stale = [
+        tid
+        for tid, task in tasks.items()
+        if task.get("completed_at") and task["completed_at"] < cutoff
+    ]
+    for tid in stale:
+        tasks.pop(tid, None)
+    return len(stale)
+
 
 async def stream_to_task(task_id: str, message: str) -> None:
     """Send a message to a task's output buffer."""
@@ -77,10 +97,12 @@ async def run_cli_streaming(
 
         exit_code = await process.wait()
         tasks[task_id]["status"] = "completed" if exit_code == 0 else "failed"
+        tasks[task_id]["completed_at"] = time.time()
 
     except Exception as e:
         await stream_to_task(task_id, f"{RED}Error: {e}{RESET}{CRLF}")
         tasks[task_id]["status"] = "failed"
+        tasks[task_id]["completed_at"] = time.time()
 
 
 def _is_self_update(service: str, command: str) -> bool:
@@ -108,21 +130,23 @@ async def _run_cli_via_ssh(
         # Get the host for the web service
         host = config.get_host(CF_WEB_SERVICE)
 
-        # Build the remote command
-        remote_cmd = f"cf {' '.join(args)} --config={config.config_path}"
+        # Build the remote command - prepend common install locations to PATH
+        # since non-interactive SSH doesn't source profile files
+        cf_cmd = f"cf {' '.join(args)} --config={config.config_path}"
+        remote_cmd = f"PATH=$HOME/.local/bin:/usr/local/bin:$PATH {cf_cmd}"
 
-        # Show what we're doing
+        # Show what we're doing (display the cf command, not the bash wrapper)
         await stream_to_task(
             task_id,
-            f"{DIM}$ ssh {host.user}@{host.address} {remote_cmd}{RESET}{CRLF}",
+            f"{DIM}$ ssh {host.user}@{host.address} {cf_cmd}{RESET}{CRLF}",
         )
         await stream_to_task(
             task_id,
             f"{GREEN}Running via SSH (self-update protection){RESET}{CRLF}",
         )
 
-        # Build SSH command using shared helper
-        ssh_args = build_ssh_command(host, remote_cmd)
+        # Build SSH command using shared helper (tty=True for progress bars)
+        ssh_args = build_ssh_command(host, remote_cmd, tty=True)
 
         # Set up environment with SSH agent
         env = {**os.environ, "FORCE_COLOR": "1", "TERM": "xterm-256color"}
@@ -147,10 +171,12 @@ async def _run_cli_via_ssh(
 
         exit_code = await process.wait()
         tasks[task_id]["status"] = "completed" if exit_code == 0 else "failed"
+        tasks[task_id]["completed_at"] = time.time()
 
     except Exception as e:
         await stream_to_task(task_id, f"{RED}Error: {e}{RESET}{CRLF}")
         tasks[task_id]["status"] = "failed"
+        tasks[task_id]["completed_at"] = time.time()
 
 
 async def run_compose_streaming(

src/compose_farm/web/ws.py

@@ -285,5 +285,4 @@ async def terminal_websocket(websocket: WebSocket, task_id: str) -> None:
             await asyncio.sleep(0.05)
     except WebSocketDisconnect:
         pass
-    finally:
-        tasks.pop(task_id, None)
+    # Task stays in memory for reconnection; cleanup_stale_tasks() handles expiry

tests/test_cli_logs.py

@@ -150,7 +150,7 @@ class TestLogsHostFilter:
         mock_run_async, _ = _mock_run_async_factory(["svc1", "svc2"])
 
         with (
-            patch("compose_farm.cli.monitoring.load_config_or_exit", return_value=cfg),
+            patch("compose_farm.cli.common.load_config_or_exit", return_value=cfg),
             patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
             patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
         ):
@@ -174,7 +174,7 @@ class TestLogsHostFilter:
         mock_run_async, _ = _mock_run_async_factory(["svc1", "svc2"])
 
         with (
-            patch("compose_farm.cli.monitoring.load_config_or_exit", return_value=cfg),
+            patch("compose_farm.cli.common.load_config_or_exit", return_value=cfg),
             patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
             patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
         ):