Add sync command to discover running services and update state

The sync command queries all hosts to find where services are actually
running and updates the state file to match reality. Supports --dry-run
to preview changes without modifying state. Useful for initial setup
or after manual changes.

src/compose_farm/cli.py

@@ -14,15 +14,20 @@ from .config import Config, load_config
 from .logs import snapshot_services
 from .ssh import (
     CommandResult,
+    check_service_running,
+    run_compose,
+    run_compose_on_host,
     run_on_services,
     run_sequential_on_services,
 )
+from .state import get_service_host, load_state, remove_service, save_state, set_service_host
 
 if TYPE_CHECKING:
     from collections.abc import Coroutine
 
 T = TypeVar("T")
 
+
 def _version_callback(value: bool) -> None:
     """Print version and exit."""
     if value:
@@ -101,15 +106,51 @@ LogPathOption = Annotated[
 ]
 
 
+async def _up_with_migration(
+    cfg: Config,
+    services: list[str],
+) -> list[CommandResult]:
+    """Start services with automatic migration if host changed."""
+    results: list[CommandResult] = []
+
+    for service in services:
+        target_host = cfg.services[service]
+        current_host = get_service_host(service)
+
+        # If service is deployed elsewhere, migrate it
+        if current_host and current_host != target_host:
+            if current_host in cfg.hosts:
+                typer.echo(f"[{service}] Migrating from {current_host} to {target_host}...")
+                down_result = await run_compose_on_host(cfg, service, current_host, "down")
+                if not down_result.success:
+                    results.append(down_result)
+                    continue
+            else:
+                typer.echo(
+                    f"[{service}] Warning: was on {current_host} (not in config), skipping down",
+                    err=True,
+                )
+
+        # Start on target host
+        up_result = await run_compose(cfg, service, "up -d")
+        results.append(up_result)
+
+        # Update state on success
+        if up_result.success:
+            set_service_host(service, target_host)
+
+    return results
+
+
 @app.command()
 def up(
     services: ServicesArg = None,
     all_services: AllOption = False,
     config: ConfigOption = None,
 ) -> None:
-    """Start services (docker compose up -d)."""
+    """Start services (docker compose up -d). Auto-migrates if host changed."""
     svc_list, cfg = _get_services(services or [], all_services, config)
-    results = _run_async(run_on_services(cfg, svc_list, "up -d"))
+    results = _run_async(_up_with_migration(cfg, svc_list))
     _report_results(results)
 
 
@@ -122,6 +163,12 @@ def down(
     """Stop services (docker compose down)."""
     svc_list, cfg = _get_services(services or [], all_services, config)
     results = _run_async(run_on_services(cfg, svc_list, "down"))
+
+    # Remove from state on success
+    for result in results:
+        if result.success:
+            remove_service(result.service)
+
     _report_results(results)
 
 
@@ -243,5 +290,98 @@ def traefik_file(
         typer.echo(warning, err=True)
 
 
+async def _discover_running_services(cfg: Config) -> dict[str, str]:
+    """Discover which services are running on which hosts.
+
+    Returns a dict mapping service names to host names for running services.
+    """
+    discovered: dict[str, str] = {}
+
+    for service, assigned_host in cfg.services.items():
+        # Check assigned host first (most common case)
+        if await check_service_running(cfg, service, assigned_host):
+            discovered[service] = assigned_host
+            continue
+
+        # Check other hosts in case service was migrated but state is stale
+        for host_name in cfg.hosts:
+            if host_name == assigned_host:
+                continue
+            if await check_service_running(cfg, service, host_name):
+                discovered[service] = host_name
+                break
+
+    return discovered
+
+
+def _report_sync_changes(
+    added: list[str],
+    removed: list[str],
+    changed: list[tuple[str, str, str]],
+    discovered: dict[str, str],
+    current_state: dict[str, str],
+) -> None:
+    """Report sync changes to the user."""
+    if added:
+        typer.echo(f"\nNew services found ({len(added)}):")
+        for service in sorted(added):
+            typer.echo(f"  + {service} on {discovered[service]}")
+
+    if changed:
+        typer.echo(f"\nServices on different hosts ({len(changed)}):")
+        for service, old_host, new_host in sorted(changed):
+            typer.echo(f"  ~ {service}: {old_host} -> {new_host}")
+
+    if removed:
+        typer.echo(f"\nServices no longer running ({len(removed)}):")
+        for service in sorted(removed):
+            typer.echo(f"  - {service} (was on {current_state[service]})")
+
+
+@app.command()
+def sync(
+    config: ConfigOption = None,
+    dry_run: Annotated[
+        bool,
+        typer.Option("--dry-run", "-n", help="Show what would be synced without writing"),
+    ] = False,
+) -> None:
+    """Discover running services and update state file.
+
+    Queries all hosts to find where services are actually running and
+    updates the state file to match reality. Useful after manual changes
+    or when first setting up compose-farm on an existing deployment.
+    """
+    cfg = load_config(config)
+    current_state = load_state()
+
+    typer.echo("Discovering running services...")
+    discovered = _run_async(_discover_running_services(cfg))
+
+    # Calculate changes
+    added = [s for s in discovered if s not in current_state]
+    removed = [s for s in current_state if s not in discovered]
+    changed = [
+        (s, current_state[s], discovered[s])
+        for s in discovered
+        if s in current_state and current_state[s] != discovered[s]
+    ]
+
+    # Report findings
+    if not (added or removed or changed):
+        typer.echo("State is already in sync.")
+        return
+
+    _report_sync_changes(added, removed, changed, discovered, current_state)
+
+    if dry_run:
+        typer.echo("\n(dry-run: no changes made)")
+        return
+
+    # Apply changes
+    save_state(discovered)
+    typer.echo(f"\nState updated: {len(discovered)} services tracked.")
+
+
 if __name__ == "__main__":
     app()

src/compose_farm/config.py

@@ -41,8 +41,22 @@ class Config(BaseModel):
         return self.hosts[self.services[service]]
 
     def get_compose_path(self, service: str) -> Path:
-        """Get compose file path for a service."""
-        return self.compose_dir / service / "docker-compose.yml"
+        """Get compose file path for a service.
+
+        Tries compose.yaml first, then docker-compose.yml.
+        """
+        service_dir = self.compose_dir / service
+        for filename in (
+            "compose.yaml",
+            "compose.yml",
+            "docker-compose.yml",
+            "docker-compose.yaml",
+        ):
+            candidate = service_dir / filename
+            if candidate.exists():
+                return candidate
+        # Default to compose.yaml if none exist (will error later)
+        return service_dir / "compose.yaml"
 
 
 def _parse_hosts(raw_hosts: dict[str, str | dict[str, str | int]]) -> dict[str, Host]:

src/compose_farm/ssh.py

@@ -167,6 +167,25 @@ async def run_compose(
     return await run_command(host, command, service, stream=stream)
 
 
+async def run_compose_on_host(
+    config: Config,
+    service: str,
+    host_name: str,
+    compose_cmd: str,
+    *,
+    stream: bool = True,
+) -> CommandResult:
+    """Run a docker compose command for a service on a specific host.
+
+    Used for migration - running 'down' on the old host before 'up' on new host.
+    """
+    host = config.hosts[host_name]
+    compose_path = config.get_compose_path(service)
+
+    command = f"docker compose -f {compose_path} {compose_cmd}"
+    return await run_command(host, command, service, stream=stream)
+
+
 async def run_on_services(
     config: Config,
     services: list[str],
@@ -206,3 +225,20 @@ async def run_sequential_on_services(
         run_sequential_commands(config, service, commands, stream=stream) for service in services
     ]
     return await asyncio.gather(*tasks)
+
+
+async def check_service_running(
+    config: Config,
+    service: str,
+    host_name: str,
+) -> bool:
+    """Check if a service has running containers on a specific host."""
+    host = config.hosts[host_name]
+    compose_path = config.get_compose_path(service)
+
+    # Use ps --status running to check for running containers
+    command = f"docker compose -f {compose_path} ps --status running -q"
+    result = await run_command(host, command, service, stream=False)
+
+    # If command succeeded and has output, containers are running
+    return result.success and bool(result.stdout.strip())

src/compose_farm/state.py

@@ -0,0 +1,58 @@
+"""State tracking for deployed services."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+
+def _get_state_path() -> Path:
+    """Get the path to the state file."""
+    state_dir = Path.home() / ".config" / "compose-farm"
+    state_dir.mkdir(parents=True, exist_ok=True)
+    return state_dir / "state.yaml"
+
+
+def load_state() -> dict[str, str]:
+    """Load the current deployment state.
+
+    Returns a dict mapping service names to host names.
+    """
+    state_path = _get_state_path()
+    if not state_path.exists():
+        return {}
+
+    with state_path.open() as f:
+        data: dict[str, Any] = yaml.safe_load(f) or {}
+
+    deployed: dict[str, str] = data.get("deployed", {})
+    return deployed
+
+
+def save_state(deployed: dict[str, str]) -> None:
+    """Save the deployment state."""
+    state_path = _get_state_path()
+    with state_path.open("w") as f:
+        yaml.safe_dump({"deployed": deployed}, f, sort_keys=False)
+
+
+def get_service_host(service: str) -> str | None:
+    """Get the host where a service is currently deployed."""
+    state = load_state()
+    return state.get(service)
+
+
+def set_service_host(service: str, host: str) -> None:
+    """Record that a service is deployed on a host."""
+    state = load_state()
+    state[service] = host
+    save_state(state)
+
+
+def remove_service(service: str) -> None:
+    """Remove a service from the state (after down)."""
+    state = load_state()
+    state.pop(service, None)
+    save_state(state)

tests/test_state.py

@@ -0,0 +1,134 @@
+"""Tests for state module."""
+
+from pathlib import Path
+
+import pytest
+
+from compose_farm import state as state_module
+from compose_farm.state import (
+    get_service_host,
+    load_state,
+    remove_service,
+    save_state,
+    set_service_host,
+)
+
+
+@pytest.fixture
+def state_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    """Create a temporary state directory and patch _get_state_path."""
+    state_path = tmp_path / ".config" / "compose-farm"
+    state_path.mkdir(parents=True)
+
+    def mock_get_state_path() -> Path:
+        return state_path / "state.yaml"
+
+    monkeypatch.setattr(state_module, "_get_state_path", mock_get_state_path)
+    return state_path
+
+
+class TestLoadState:
+    """Tests for load_state function."""
+
+    def test_load_state_empty(self, state_dir: Path) -> None:
+        """Returns empty dict when state file doesn't exist."""
+        _ = state_dir  # Fixture activates the mock
+        result = load_state()
+        assert result == {}
+
+    def test_load_state_with_data(self, state_dir: Path) -> None:
+        """Loads existing state from file."""
+        state_file = state_dir / "state.yaml"
+        state_file.write_text("deployed:\n  plex: nas01\n  jellyfin: nas02\n")
+
+        result = load_state()
+        assert result == {"plex": "nas01", "jellyfin": "nas02"}
+
+    def test_load_state_empty_file(self, state_dir: Path) -> None:
+        """Returns empty dict for empty file."""
+        state_file = state_dir / "state.yaml"
+        state_file.write_text("")
+
+        result = load_state()
+        assert result == {}
+
+
+class TestSaveState:
+    """Tests for save_state function."""
+
+    def test_save_state(self, state_dir: Path) -> None:
+        """Saves state to file."""
+        save_state({"plex": "nas01", "jellyfin": "nas02"})
+
+        state_file = state_dir / "state.yaml"
+        assert state_file.exists()
+        content = state_file.read_text()
+        assert "plex: nas01" in content
+        assert "jellyfin: nas02" in content
+
+
+class TestGetServiceHost:
+    """Tests for get_service_host function."""
+
+    def test_get_existing_service(self, state_dir: Path) -> None:
+        """Returns host for existing service."""
+        state_file = state_dir / "state.yaml"
+        state_file.write_text("deployed:\n  plex: nas01\n")
+
+        host = get_service_host("plex")
+        assert host == "nas01"
+
+    def test_get_nonexistent_service(self, state_dir: Path) -> None:
+        """Returns None for service not in state."""
+        state_file = state_dir / "state.yaml"
+        state_file.write_text("deployed:\n  plex: nas01\n")
+
+        host = get_service_host("unknown")
+        assert host is None
+
+
+class TestSetServiceHost:
+    """Tests for set_service_host function."""
+
+    def test_set_new_service(self, state_dir: Path) -> None:
+        """Adds new service to state."""
+        _ = state_dir  # Fixture activates the mock
+        set_service_host("plex", "nas01")
+
+        result = load_state()
+        assert result["plex"] == "nas01"
+
+    def test_update_existing_service(self, state_dir: Path) -> None:
+        """Updates host for existing service."""
+        state_file = state_dir / "state.yaml"
+        state_file.write_text("deployed:\n  plex: nas01\n")
+
+        set_service_host("plex", "nas02")
+
+        result = load_state()
+        assert result["plex"] == "nas02"
+
+
+class TestRemoveService:
+    """Tests for remove_service function."""
+
+    def test_remove_existing_service(self, state_dir: Path) -> None:
+        """Removes service from state."""
+        state_file = state_dir / "state.yaml"
+        state_file.write_text("deployed:\n  plex: nas01\n  jellyfin: nas02\n")
+
+        remove_service("plex")
+
+        result = load_state()
+        assert "plex" not in result
+        assert result["jellyfin"] == "nas02"
+
+    def test_remove_nonexistent_service(self, state_dir: Path) -> None:
+        """Removing nonexistent service doesn't error."""
+        state_file = state_dir / "state.yaml"
+        state_file.write_text("deployed:\n  plex: nas01\n")
+
+        remove_service("unknown")  # Should not raise
+
+        result = load_state()
+        assert result["plex"] == "nas01"

tests/test_sync.py

@@ -0,0 +1,174 @@
+"""Tests for sync command and related functions."""
+
+from pathlib import Path
+from typing import Any
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from compose_farm import cli as cli_module
+from compose_farm import ssh as ssh_module
+from compose_farm import state as state_module
+from compose_farm.config import Config, Host
+from compose_farm.ssh import CommandResult, check_service_running
+
+
+@pytest.fixture
+def mock_config(tmp_path: Path) -> Config:
+    """Create a mock config for testing."""
+    compose_dir = tmp_path / "stacks"
+    compose_dir.mkdir()
+
+    # Create service directories with compose files
+    for service in ["plex", "jellyfin", "sonarr"]:
+        svc_dir = compose_dir / service
+        svc_dir.mkdir()
+        (svc_dir / "compose.yaml").write_text(f"# {service} compose file\n")
+
+    return Config(
+        compose_dir=compose_dir,
+        hosts={
+            "nas01": Host(address="192.168.1.10", user="admin", port=22),
+            "nas02": Host(address="192.168.1.11", user="admin", port=22),
+        },
+        services={
+            "plex": "nas01",
+            "jellyfin": "nas01",
+            "sonarr": "nas02",
+        },
+    )
+
+
+@pytest.fixture
+def state_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    """Create a temporary state directory and patch _get_state_path."""
+    state_path = tmp_path / ".config" / "compose-farm"
+    state_path.mkdir(parents=True)
+
+    def mock_get_state_path() -> Path:
+        return state_path / "state.yaml"
+
+    monkeypatch.setattr(state_module, "_get_state_path", mock_get_state_path)
+    return state_path
+
+
+class TestCheckServiceRunning:
+    """Tests for check_service_running function."""
+
+    @pytest.mark.asyncio
+    async def test_service_running(self, mock_config: Config) -> None:
+        """Returns True when service has running containers."""
+        with patch.object(ssh_module, "run_command", new_callable=AsyncMock) as mock_run:
+            mock_run.return_value = CommandResult(
+                service="plex",
+                exit_code=0,
+                success=True,
+                stdout="abc123\ndef456\n",
+            )
+            result = await check_service_running(mock_config, "plex", "nas01")
+            assert result is True
+
+    @pytest.mark.asyncio
+    async def test_service_not_running(self, mock_config: Config) -> None:
+        """Returns False when service has no running containers."""
+        with patch.object(ssh_module, "run_command", new_callable=AsyncMock) as mock_run:
+            mock_run.return_value = CommandResult(
+                service="plex",
+                exit_code=0,
+                success=True,
+                stdout="",
+            )
+            result = await check_service_running(mock_config, "plex", "nas01")
+            assert result is False
+
+    @pytest.mark.asyncio
+    async def test_command_failed(self, mock_config: Config) -> None:
+        """Returns False when command fails."""
+        with patch.object(ssh_module, "run_command", new_callable=AsyncMock) as mock_run:
+            mock_run.return_value = CommandResult(
+                service="plex",
+                exit_code=1,
+                success=False,
+            )
+            result = await check_service_running(mock_config, "plex", "nas01")
+            assert result is False
+
+
+class TestDiscoverRunningServices:
+    """Tests for _discover_running_services function."""
+
+    @pytest.mark.asyncio
+    async def test_discovers_on_assigned_host(self, mock_config: Config) -> None:
+        """Discovers service running on its assigned host."""
+        with patch.object(
+            cli_module, "check_service_running", new_callable=AsyncMock
+        ) as mock_check:
+            # plex running on nas01, jellyfin not running, sonarr on nas02
+            async def check_side_effect(_cfg: Any, service: str, host: str) -> bool:
+                return (service == "plex" and host == "nas01") or (
+                    service == "sonarr" and host == "nas02"
+                )
+
+            mock_check.side_effect = check_side_effect
+
+            result = await cli_module._discover_running_services(mock_config)
+            assert result == {"plex": "nas01", "sonarr": "nas02"}
+
+    @pytest.mark.asyncio
+    async def test_discovers_on_different_host(self, mock_config: Config) -> None:
+        """Discovers service running on non-assigned host (after migration)."""
+        with patch.object(
+            cli_module, "check_service_running", new_callable=AsyncMock
+        ) as mock_check:
+            # plex migrated to nas02
+            async def check_side_effect(_cfg: Any, service: str, host: str) -> bool:
+                return service == "plex" and host == "nas02"
+
+            mock_check.side_effect = check_side_effect
+
+            result = await cli_module._discover_running_services(mock_config)
+            assert result == {"plex": "nas02"}
+
+
+class TestReportSyncChanges:
+    """Tests for _report_sync_changes function."""
+
+    def test_reports_added(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """Reports newly discovered services."""
+        cli_module._report_sync_changes(
+            added=["plex", "jellyfin"],
+            removed=[],
+            changed=[],
+            discovered={"plex": "nas01", "jellyfin": "nas02"},
+            current_state={},
+        )
+        captured = capsys.readouterr()
+        assert "New services found (2)" in captured.out
+        assert "+ plex on nas01" in captured.out
+        assert "+ jellyfin on nas02" in captured.out
+
+    def test_reports_removed(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """Reports services that are no longer running."""
+        cli_module._report_sync_changes(
+            added=[],
+            removed=["sonarr"],
+            changed=[],
+            discovered={},
+            current_state={"sonarr": "nas01"},
+        )
+        captured = capsys.readouterr()
+        assert "Services no longer running (1)" in captured.out
+        assert "- sonarr (was on nas01)" in captured.out
+
+    def test_reports_changed(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """Reports services that moved to a different host."""
+        cli_module._report_sync_changes(
+            added=[],
+            removed=[],
+            changed=[("plex", "nas01", "nas02")],
+            discovered={"plex": "nas02"},
+            current_state={"plex": "nas01"},
+        )
+        captured = capsys.readouterr()
+        assert "Services on different hosts (1)" in captured.out
+        assert "~ plex: nas01 -> nas02" in captured.out