fix: reconcile orphan containers on API startup

The periodic reaper only runs after a 5-minute sleep, leaving orphan
Docker containers running after API restarts. Add startup reconciliation
that cross-references Docker containers (by helprs.* labels) with DB
session records and cleans up mismatches: orphan containers are stopped,
stale DB sessions are marked FAILED/TIMEOUT, and stuck PENDING sessions
are resolved.

Author: mariuspruvot

apps/api/src/helprs/main.py

@@ -101,6 +101,37 @@ async def _run_webhook_reaper(app: FastAPI, *, interval_seconds: int = _REAPER_I
         raise
 
 
+async def _reconcile_containers_on_startup(app: FastAPI) -> None:
+    """Reconcile DB container sessions with actual Docker state at boot.
+
+    Mirrors the crash-replay pattern: bounded by a timeout, failures never
+    block startup.
+    """
+    from helprs.modules.container.service import AioDockerClient, reconcile_on_startup
+
+    try:
+        docker = AioDockerClient()
+        try:
+            async with app.state.session_factory() as db:
+                containers_removed, sessions_updated = await asyncio.wait_for(
+                    reconcile_on_startup(db, docker),
+                    timeout=_REPLAY_DISCOVERY_TIMEOUT_SECONDS,
+                )
+                await db.commit()
+                if containers_removed or sessions_updated:
+                    logger.info(
+                        "startup_container_reconciliation",
+                        containers_removed=containers_removed,
+                        sessions_updated=sessions_updated,
+                    )
+        finally:
+            await docker.close()
+    except TimeoutError:
+        logger.warning("container_reconciliation_timeout", timeout=_REPLAY_DISCOVERY_TIMEOUT_SECONDS)
+    except Exception:
+        logger.exception("container_reconciliation_failed")
+
+
 async def _run_container_cleanup(
     app: FastAPI,
     *,
@@ -164,6 +195,11 @@ async def lifespan(app: FastAPI):
             # server starts serving traffic immediately (AC #2).
             await _replay_pending_webhook_events(app)
 
+            # Container reconciliation: sync DB sessions with Docker state
+            # before the periodic reaper starts so orphans are cleaned
+            # immediately on restart.
+            await _reconcile_containers_on_startup(app)
+
             # Periodic reaper: handles rows that get stuck mid-run (e.g.
             # mark_processed commit failure) without waiting for the next
             # restart.

apps/api/src/helprs/modules/container/service.py

@@ -15,7 +15,8 @@
 import contextlib
 import json
 import os
-from datetime import UTC, datetime
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import TYPE_CHECKING, Protocol
 
@@ -51,6 +52,14 @@
 SKILLS_HOST_PATH = os.environ.get("SKILLS_HOST_PATH", str(SKILLS_BASE_PATH))
 
 
+@dataclass(frozen=True, slots=True)
+class ContainerInfo:
+    """Minimal container metadata returned by list_containers."""
+
+    container_id: str
+    labels: dict[str, str]
+
+
 class DockerClient(Protocol):
     """Protocol for the Docker client used by the container service.
 
@@ -91,6 +100,10 @@ async def wait_container(self, container_id: str) -> int:
         """Wait for a container to exit. Returns exit code."""
         ...
 
+    async def list_containers(self, label_filter: str) -> list[ContainerInfo]:
+        """List containers that have the given label key present."""
+        ...
+
 
 class AioDockerClient:
     """Production Docker client wrapping aiodocker."""
@@ -114,6 +127,7 @@ async def create_container(
             "Env": env_list,
             "Labels": labels,
             "OpenStdin": True,
+            "StopTimeout": 10,
             "HostConfig": {
                 "Binds": binds,
                 "Memory": 512 * 1024 * 1024,  # 512MB
@@ -162,6 +176,20 @@ async def wait_container(self, container_id: str) -> int:
         result = await container.wait()
         return result["StatusCode"]
 
+    async def list_containers(self, label_filter: str) -> list[ContainerInfo]:
+        """List containers that have the given label key present."""
+        raw = await self._docker.containers.list(
+            all=True,
+            filters=json.dumps({"label": [label_filter]}),
+        )
+        return [
+            ContainerInfo(
+                container_id=c.id,
+                labels=c["Labels"],
+            )
+            for c in raw
+        ]
+
     async def close(self) -> None:
         await self._docker.close()
 
@@ -493,3 +521,95 @@ async def cleanup_expired(
         await logger.ainfo("expired_sessions_cleaned", count=cleaned)
 
     return cleaned
+
+
+async def reconcile_on_startup(
+    db: AsyncSession,
+    docker: DockerClient,
+) -> tuple[int, int]:
+    """Reconcile DB sessions with actual Docker container state at API startup.
+
+    Three-way reconciliation:
+    1. Docker containers with helprs labels but no matching active DB session
+       -> orphan container: stop and remove.
+    2. DB sessions past TTL -> mark TIMEOUT, remove container if present.
+    3. DB sessions whose container vanished from Docker -> mark FAILED.
+
+    Returns (containers_removed, sessions_updated).
+    Never raises — logs errors and returns (0, 0) if Docker is unreachable.
+    """
+    try:
+        docker_containers = await docker.list_containers(label_filter="helprs.session_id")
+    except Exception:
+        await logger.aerror("reconcile_list_containers_failed")
+        return (0, 0)
+
+    # Build lookups from Docker state
+    docker_by_id: dict[str, ContainerInfo] = {c.container_id: c for c in docker_containers}
+    docker_session_ids: set[str] = set()
+    for c in docker_containers:
+        sid = c.labels.get("helprs.session_id", "")
+        if sid:
+            docker_session_ids.add(sid)
+
+    # Get all active DB sessions
+    result = await db.execute(
+        select(ContainerSession).where(
+            ContainerSession.status.in_([ContainerStatus.RUNNING, ContainerStatus.PENDING]),
+        )
+    )
+    active_sessions = list(result.scalars().all())
+    active_session_ids: set[str] = {str(cs.id) for cs in active_sessions}
+
+    now = datetime.now(UTC)
+    cutoff_dt = now - timedelta(seconds=CONTAINER_TTL_SECONDS)
+    containers_removed = 0
+    sessions_updated = 0
+
+    # Step 1: Orphan containers — in Docker but no matching active DB session
+    for container in docker_containers:
+        session_label = container.labels.get("helprs.session_id", "")
+        if session_label not in active_session_ids:
+            try:
+                await docker.stop_container(container.container_id)
+            except Exception:
+                await logger.awarning("reconcile_stop_orphan_failed", container_id=container.container_id)
+            try:
+                await docker.remove_container(container.container_id, force=True)
+            except Exception:
+                await logger.awarning("reconcile_remove_orphan_failed", container_id=container.container_id)
+            containers_removed += 1
+
+    # Step 2: Stale DB sessions — reconcile against Docker state
+    for cs in active_sessions:
+        if cs.created_at < cutoff_dt:
+            # Past TTL -> TIMEOUT, remove container if it exists
+            if cs.container_id and cs.container_id in docker_by_id:
+                with contextlib.suppress(Exception):
+                    await docker.stop_container(cs.container_id)
+                with contextlib.suppress(Exception):
+                    await docker.remove_container(cs.container_id, force=True)
+                containers_removed += 1
+            cs.status = ContainerStatus.TIMEOUT
+            cs.completed_at = now
+            sessions_updated += 1
+        elif cs.container_id and cs.container_id not in docker_by_id:
+            # Container vanished from Docker -> FAILED
+            cs.status = ContainerStatus.FAILED
+            cs.completed_at = now
+            sessions_updated += 1
+        elif not cs.container_id:
+            # Stuck PENDING with no container_id -> FAILED
+            cs.status = ContainerStatus.FAILED
+            cs.completed_at = now
+            sessions_updated += 1
+
+    if sessions_updated:
+        await db.flush()
+
+    await logger.ainfo(
+        "startup_reconciliation_complete",
+        containers_removed=containers_removed,
+        sessions_updated=sessions_updated,
+    )
+    return (containers_removed, sessions_updated)

apps/api/tests/modules/container/test_service.py

@@ -11,11 +11,13 @@
 from helprs.modules.container.models import ContainerStatus
 from helprs.modules.container.service import (
     CONTAINER_TTL_SECONDS,
+    ContainerInfo,
     cleanup_expired,
     create_session,
     get_session,
     get_session_or_404,
     mark_completed,
+    reconcile_on_startup,
     start_container,
     stop_container,
     stream_output,
@@ -40,12 +42,16 @@ def __init__(
         fail_on_create: bool = False,
         fail_on_stop: bool = False,
         log_lines: list[str] | None = None,
+        listed_containers: list[ContainerInfo] | None = None,
+        fail_on_list: bool = False,
     ):
         self._container_id = container_id
         self._exit_code = exit_code
         self._fail_on_create = fail_on_create
         self._fail_on_stop = fail_on_stop
+        self._fail_on_list = fail_on_list
         self._log_lines = log_lines or ["line 1", "line 2"]
+        self._listed_containers = listed_containers or []
         self.created: list[dict] = []
         self.started: list[str] = []
         self.stopped: list[str] = []
@@ -81,6 +87,11 @@ async def container_logs(self, container_id: str, follow: bool = False) -> Async
     async def wait_container(self, container_id: str) -> int:
         return self._exit_code
 
+    async def list_containers(self, label_filter: str) -> list[ContainerInfo]:
+        if self._fail_on_list:
+            raise RuntimeError("Docker daemon unreachable")
+        return self._listed_containers
+
 
 # ---------------------------------------------------------------------------
 # Fixtures
@@ -539,3 +550,143 @@ async def test_skips_completed_sessions(
 
         cleaned = await cleanup_expired(db=db, docker=docker)
         assert cleaned == 0
+
+
+# ---------------------------------------------------------------------------
+# Tests: reconcile_on_startup
+# ---------------------------------------------------------------------------
+
+
+class TestReconcileOnStartup:
+    async def test_removes_orphan_docker_containers(self, db: AsyncSession, installation: Installation):
+        """Docker container exists but has no matching active DB session."""
+        orphan = ContainerInfo(
+            container_id="orphan-container-123",
+            labels={"helprs.session_id": str(uuid.uuid4())},
+        )
+        docker = FakeDockerClient(listed_containers=[orphan])
+
+        containers_removed, sessions_updated = await reconcile_on_startup(db, docker)
+
+        assert containers_removed == 1
+        assert sessions_updated == 0
+        assert "orphan-container-123" in docker.stopped
+        assert "orphan-container-123" in docker.removed
+
+    async def test_marks_stale_sessions_failed(self, db: AsyncSession, installation: Installation):
+        """DB session is RUNNING but its container_id is not in Docker."""
+        cs = await create_session(
+            db=db,
+            installation_id=installation.id,
+            pr_number=1,
+            repo_full_name="org/repo",
+            skill_name="challenge-me",
+        )
+        cs.status = ContainerStatus.RUNNING
+        cs.container_id = "vanished-container-456"
+        await db.flush()
+
+        docker = FakeDockerClient(listed_containers=[])
+
+        containers_removed, sessions_updated = await reconcile_on_startup(db, docker)
+
+        assert sessions_updated == 1
+        refreshed = await get_session(db, cs.id)
+        assert refreshed is not None
+        assert refreshed.status == ContainerStatus.FAILED
+        assert refreshed.completed_at is not None
+
+    async def test_marks_expired_sessions_timeout(self, db: AsyncSession, installation: Installation):
+        """DB session is RUNNING and past TTL, with container still in Docker."""
+        from datetime import UTC, datetime, timedelta
+
+        cs = await create_session(
+            db=db,
+            installation_id=installation.id,
+            pr_number=1,
+            repo_full_name="org/repo",
+            skill_name="challenge-me",
+        )
+        cs.status = ContainerStatus.RUNNING
+        cs.container_id = "expired-container-789"
+        cs.created_at = datetime.now(UTC) - timedelta(seconds=CONTAINER_TTL_SECONDS + 120)
+        await db.flush()
+
+        container_info = ContainerInfo(
+            container_id="expired-container-789",
+            labels={"helprs.session_id": str(cs.id)},
+        )
+        docker = FakeDockerClient(listed_containers=[container_info])
+
+        containers_removed, sessions_updated = await reconcile_on_startup(db, docker)
+
+        assert containers_removed == 1
+        assert sessions_updated == 1
+        refreshed = await get_session(db, cs.id)
+        assert refreshed is not None
+        assert refreshed.status == ContainerStatus.TIMEOUT
+        assert "expired-container-789" in docker.stopped
+
+    async def test_marks_stuck_pending_failed(self, db: AsyncSession, installation: Installation):
+        """DB session stuck in PENDING with no container_id."""
+        cs = await create_session(
+            db=db,
+            installation_id=installation.id,
+            pr_number=1,
+            repo_full_name="org/repo",
+            skill_name="challenge-me",
+        )
+        # create_session already sets PENDING and container_id=None
+        docker = FakeDockerClient(listed_containers=[])
+
+        containers_removed, sessions_updated = await reconcile_on_startup(db, docker)
+
+        assert sessions_updated == 1
+        refreshed = await get_session(db, cs.id)
+        assert refreshed is not None
+        assert refreshed.status == ContainerStatus.FAILED
+
+    async def test_leaves_valid_sessions_untouched(self, db: AsyncSession, installation: Installation):
+        """RUNNING session with recent created_at and matching Docker container."""
+        cs = await create_session(
+            db=db,
+            installation_id=installation.id,
+            pr_number=1,
+            repo_full_name="org/repo",
+            skill_name="challenge-me",
+        )
+        cs.status = ContainerStatus.RUNNING
+        cs.container_id = "active-container-abc"
+        await db.flush()
+
+        container_info = ContainerInfo(
+            container_id="active-container-abc",
+            labels={"helprs.session_id": str(cs.id)},
+        )
+        docker = FakeDockerClient(listed_containers=[container_info])
+
+        containers_removed, sessions_updated = await reconcile_on_startup(db, docker)
+
+        assert containers_removed == 0
+        assert sessions_updated == 0
+        refreshed = await get_session(db, cs.id)
+        assert refreshed is not None
+        assert refreshed.status == ContainerStatus.RUNNING
+
+    async def test_handles_docker_unreachable(self, db: AsyncSession):
+        """list_containers failure returns (0, 0) without crashing."""
+        docker = FakeDockerClient(fail_on_list=True)
+
+        containers_removed, sessions_updated = await reconcile_on_startup(db, docker)
+
+        assert containers_removed == 0
+        assert sessions_updated == 0
+
+    async def test_noop_on_empty_state(self, db: AsyncSession):
+        """No sessions, no containers -> (0, 0)."""
+        docker = FakeDockerClient(listed_containers=[])
+
+        containers_removed, sessions_updated = await reconcile_on_startup(db, docker)
+
+        assert containers_removed == 0
+        assert sessions_updated == 0