fix(container): scope shutdown and boot reconciliation to the owning worker

Production runs `--workers 4` against one shared Docker socket, and both
lifecycle hooks operated on "every unfinished session". So whenever any one
worker restarted or crashed, its shutdown hook CANCELLED every live session
on the host and its boot hook marked the survivors FAILED. Unlike the
reaper, these were not idempotent -- they were actively destructive to
peers. Neither function had a single test, which is how it survived.

Shutdown is now scoped by ownership. Every runner carries a `helprs.boot_id`
label identifying the process that started it, and `cleanup_own_running`
asks Docker which containers carry this process's label. Docker is the
authority on what is actually running; the database only records what was
intended.

Boot reconciliation stops guessing. `reconcile_stale_sessions` asks
`container_is_running` per session instead of treating "unfinished" as
evidence of death -- at boot, most RUNNING rows belong to a peer that never
stopped and is still streaming to a user.

`current_boot_id` recomputes when the PID changes rather than caching once,
because uvicorn's --workers mode may fork after import and two workers
sharing a boot id would each stop the other's containers -- reintroducing
the bug through the mechanism meant to fix it.

Adds `container_is_running` and `list_runners` to the DockerClient protocol
and its doubles, plus `RunnerContainer` so the boundary returns a typed pair
rather than a dict.

Eight new tests, run against the previous behaviour first: four of them fail
on it.

Author: mariuspruvot

CLAUDE.md

@@ -126,7 +126,7 @@ Key additions for production: `ENVIRONMENT=production`, `ADMIN_PASSWORD`, `CORS_
 - **Worktrees and node_modules**: `npm install` must be run in each worktree separately — `node_modules` aren't shared from the main tree
 - **Shallow clone + `gh pr checkout --detach`**: `gh pr checkout` (without `--detach`) fails on `--depth=1` clones because git can't set up tracking branches from shallow refs. Always use `--detach` — containers don't need tracking branches, just files on disk.
 - **Installation IDs in URLs**: frontend routes (`/installations/:id`) use `github_installation_id` (integer, e.g. `123093268`), NOT the internal UUID. All API endpoints (installation AND container session creation) expect the GitHub integer ID. The service layer resolves to internal UUID via `get_installation_by_github_id()`.
-- **Multi-worker background tasks**: with `--workers N`, each uvicorn worker runs its own lifespan (webhook reaper + container cleanup). Both are idempotent: reaper uses atomic row-level claim (`mark_processing`), cleanup suppresses double-stop exceptions.
+- **Multi-worker background tasks**: with `--workers N`, each uvicorn worker runs its own lifespan. The reaper claims rows atomically (`mark_processing`). The container hooks are scoped to the process that owns the work: every runner carries a `helprs.boot_id` label (`container/service.py:current_boot_id`), shutdown stops only containers with that label, and boot reconciliation asks Docker `container_is_running` instead of assuming every unfinished row is dead. Both were previously unscoped and destroyed peers' live sessions on any single worker restart.
 - **Run migrations after checkout**: `docker exec helprs-api-1 uv run alembic current` vs `alembic heads` — if they differ, run `make migrate`. Missing columns cause 500s that surface as browser CORS errors (response lacks CORS headers on unhandled exceptions).
 - **API rebuild invalidates tokens**: re-authenticate after an API restart if the deploy changed `SECRET_KEY`. Nothing in the code generates one — `Settings.SECRET_KEY` is required and the app refuses to boot without it.
 - **Coolify `--project-directory`**: Coolify sets `--project-directory` to the repo root, not the compose file location. Relative paths in the compose (`context`, `volumes`) must be relative to the repo root (`./apps/api`, not `../../apps/api`).

apps/api/src/helprs/main.py

@@ -22,6 +22,7 @@
 from helprs.core.exceptions import DomainError, domain_exception_handler
 from helprs.core.middleware import configure_logging, configure_sentry, setup_middleware
 from helprs.modules.container import cleanup as container_cleanup
+from helprs.modules.container import service as container_service
 from helprs.modules.container.docker_client import AioDockerClient
 from helprs.modules.container.router import router as container_router
 from helprs.modules.identity.router import router as identity_router
@@ -129,25 +130,43 @@ async def _run_container_cleanup(
 
 
 async def _reconcile_stale_sessions(session_factory: SessionFactory) -> None:
-    """Mark sessions left RUNNING/PENDING by a previous run as FAILED."""
+    """Fail sessions whose container is no longer running.
+
+    Takes a Docker client because "unfinished" alone is not evidence of death:
+    with several workers, most RUNNING rows at boot belong to a peer that is
+    still streaming them.
+    """
     try:
-        async with session_factory() as db:
-            await container_cleanup.reconcile_stale_sessions(db)
-            await db.commit()
+        docker = AioDockerClient()
+        try:
+            async with session_factory() as db:
+                await container_cleanup.reconcile_stale_sessions(db, docker)
+                await db.commit()
+        finally:
+            await docker.close()
     except Exception:
         logger.exception("session_reconciliation_failed")
 
 
-async def _stop_running_containers(session_factory: SessionFactory) -> None:
-    """Stop every container still running before the engine goes away."""
+async def _stop_own_containers(session_factory: SessionFactory) -> None:
+    """Stop the containers THIS process started, before the engine goes away.
+
+    Not every container on the host: with ``--workers N`` they share a Docker
+    socket, and stopping the lot cancelled peers' live sessions on any single
+    worker restart.
+    """
     try:
         docker = AioDockerClient()
-        async with session_factory() as db:
-            stopped = await container_cleanup.cleanup_all_running(db, docker)
-            await db.commit()
-            if stopped:
-                logger.info("shutdown_containers_stopped", count=stopped)
-        await docker.close()
+        try:
+            async with session_factory() as db:
+                stopped = await container_cleanup.cleanup_own_running(
+                    db, docker, boot_id=container_service.current_boot_id()
+                )
+                await db.commit()
+                if stopped:
+                    logger.info("shutdown_containers_stopped", count=stopped)
+        finally:
+            await docker.close()
     except Exception:
         logger.exception("shutdown_container_cleanup_failed")
 
@@ -188,7 +207,7 @@ async def _lifespan(app: FastAPI) -> AsyncIterator[None]:
         await _replay_pending_webhook_events(app)
         await _reconcile_stale_sessions(session_factory)
 
-        stack.push_async_callback(_stop_running_containers, session_factory)
+        stack.push_async_callback(_stop_own_containers, session_factory)
         stack.push_async_callback(_drain_replay_tasks, app)
 
         for loop in (_run_webhook_reaper(app), _run_container_cleanup(app)):

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

@@ -44,28 +44,47 @@ async def cleanup_expired(db: AsyncSession, docker: DockerClient, *, ttl_seconds
     return len(expired)
 
 
-async def cleanup_all_running(db: AsyncSession, docker: DockerClient) -> int:
-    """Stop every live session — used on graceful shutdown."""
-    sessions = await repository.list_unfinished(db)
-
-    for cs in sessions:
+async def cleanup_own_running(db: AsyncSession, docker: DockerClient, *, boot_id: str) -> int:
+    """Stop the sessions THIS process started — used on graceful shutdown.
+
+    Scoped by boot id rather than "everything unfinished". Several uvicorn
+    workers share one Docker socket, so the unscoped version cancelled every
+    peer's live sessions whenever any one worker restarted. Docker is asked
+    which containers carry this process's label, because it is the authority
+    on what is actually running; the database only records what was intended.
+    """
+    stopped = 0
+    for runner in await docker.list_runners(boot_id=boot_id):
+        cs = await repository.get(db, runner.session_id)
+        if cs is None or cs.completed_at is not None:
+            continue
         await _destroy(docker, cs)
         cs.status = ContainerStatus.CANCELLED
         cs.completed_at = datetime.now(UTC)
+        stopped += 1
 
-    if sessions:
+    if stopped:
         await db.flush()
-        await logger.ainfo("shutdown_sessions_stopped", count=len(sessions))
-    return len(sessions)
+        await logger.ainfo("shutdown_sessions_stopped", count=stopped, boot_id=boot_id)
+    return stopped
+
 
+async def reconcile_stale_sessions(db: AsyncSession, docker: DockerClient) -> int:
+    """Fail sessions whose container is gone.
 
-async def reconcile_stale_sessions(db: AsyncSession) -> int:
-    """Fail sessions left RUNNING by a crashed process.
+    Asks Docker rather than assuming, because "unfinished at boot" is not the
+    same as "dead": with several workers a RUNNING row usually belongs to a
+    peer that is still streaming it, and marking those FAILED killed live
+    sessions every time one worker restarted.
 
-    Their containers died with the host process, so waiting for the TTL would
-    only leave the dashboard lying for 15 minutes.
+    A row with no container id never got that far, so nothing is running for
+    it either way.
     """
-    stale = await repository.list_unfinished(db)
+    stale = []
+    for cs in await repository.list_unfinished(db):
+        if cs.container_id and await docker.container_is_running(cs.container_id):
+            continue
+        stale.append(cs)
 
     for cs in stale:
         cs.status = ContainerStatus.FAILED

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

@@ -5,11 +5,30 @@
 """
 
 import base64
+import contextlib
 from collections.abc import AsyncIterator
+from dataclasses import dataclass
 from typing import Protocol
+from uuid import UUID
 
 CLAUDE_RUNNER_IMAGE = "claude-runner:latest"
 
+# Labels stamped on every runner. ``boot_id`` identifies the API process that
+# spawned it, which is how a shutting-down worker tells its own containers
+# apart from a peer's: with several uvicorn workers sharing one Docker socket,
+# "every running container" is not the same set as "mine".
+LABEL_SESSION_ID = "helprs.session_id"
+LABEL_BOOT_ID = "helprs.boot_id"
+
+
+@dataclass(frozen=True)
+class RunnerContainer:
+    """A live runner, identified by both ids the cleanup paths need."""
+
+    container_id: str
+    session_id: UUID
+
+
 CONTAINER_MEMORY_BYTES = 512 * 1024 * 1024
 CONTAINER_NANO_CPUS = 1_000_000_000
 # Generous for a git clone plus a node process, tight enough that a runaway
@@ -54,6 +73,19 @@ async def wait_container(self, container_id: str) -> int:
         """Block until the container exits; return its exit code."""
         ...
 
+    async def container_is_running(self, container_id: str) -> bool:
+        """Whether the container still exists and is running.
+
+        The honest answer to "is this session actually alive?", which the DB
+        cannot give: a row says RUNNING whether its container is streaming or
+        died with the process that started it.
+        """
+        ...
+
+    async def list_runners(self, *, boot_id: str) -> list[RunnerContainer]:
+        """Live runners started by one API process, newest state from Docker."""
+        ...
+
     async def close(self) -> None: ...
 
 
@@ -96,6 +128,31 @@ async def create_container(
         )
         return container.id
 
+    async def container_is_running(self, container_id: str) -> bool:
+        try:
+            container = await self._docker.containers.get(container_id)
+            state = (await container.show()).get("State") or {}
+        except Exception:
+            # Gone, unreachable, or never existed: for every caller here that
+            # means "not alive", and treating an error as "still running"
+            # would leave sessions stuck RUNNING forever.
+            return False
+        return bool(state.get("Running"))
+
+    async def list_runners(self, *, boot_id: str) -> list[RunnerContainer]:
+        containers = await self._docker.containers.list(
+            filters={"label": [f"{LABEL_BOOT_ID}={boot_id}"]},
+        )
+        runners = []
+        for container in containers:
+            # A runner whose session label is missing or unparseable is not
+            # ours to reason about; skipping beats guessing which row it
+            # belongs to and cancelling the wrong session.
+            with contextlib.suppress(KeyError, TypeError, ValueError):
+                labels = container["Labels"] or {}
+                runners.append(RunnerContainer(container_id=container.id, session_id=UUID(labels[LABEL_SESSION_ID])))
+        return runners
+
     async def start_container(self, container_id: str) -> None:
         container = await self._docker.containers.get(container_id)
         await container.start()

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

@@ -10,7 +10,7 @@
 import os
 from datetime import UTC, datetime
 from pathlib import Path
-from uuid import UUID
+from uuid import UUID, uuid4
 
 import structlog
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -19,7 +19,12 @@
 from helprs.core.database import get_db_context
 from helprs.core.exceptions import ConflictError, ExternalServiceError, NotFoundError
 from helprs.modules.container import repository
-from helprs.modules.container.docker_client import CLAUDE_RUNNER_IMAGE, DockerClient
+from helprs.modules.container.docker_client import (
+    CLAUDE_RUNNER_IMAGE,
+    LABEL_BOOT_ID,
+    LABEL_SESSION_ID,
+    DockerClient,
+)
 from helprs.modules.container.models import ContainerSession, ContainerStatus, SessionEvent
 from helprs.modules.container.pr_comment import build_session_url, extract_score_card, format_pr_comment
 from helprs.modules.container.scorecard import extract_scorecard
@@ -43,6 +48,29 @@
 # --- Session records -------------------------------------------------------
 
 
+# Identity of this API process, stamped on every container it starts.
+#
+# Several uvicorn workers share one Docker socket, so "every running
+# container" is not the same set as "the ones I started". Without this a
+# worker's shutdown hook cancelled its peers' live sessions.
+_boot_id: tuple[int, str] | None = None
+
+
+def current_boot_id() -> str:
+    """A value unique to this process and stable for its lifetime.
+
+    Recomputed when the PID changes, so a forked worker never inherits its
+    parent's id: uvicorn's ``--workers`` mode may fork after import, and two
+    workers sharing a boot id would each stop the other's containers -- the
+    exact bug this exists to prevent.
+    """
+    global _boot_id
+    pid = os.getpid()
+    if _boot_id is None or _boot_id[0] != pid:
+        _boot_id = (pid, f"{pid}-{uuid4().hex[:8]}")
+    return _boot_id[1]
+
+
 async def create_session(
     db: AsyncSession,
     installation_id: UUID,
@@ -147,7 +175,8 @@ async def start_container(
             },
             volumes=[f"{SKILLS_HOST_PATH}/{cs.skill_name}:/skills/{cs.skill_name}:ro"],
             labels={
-                "helprs.session_id": str(cs.id),
+                LABEL_SESSION_ID: str(cs.id),
+                LABEL_BOOT_ID: current_boot_id(),
                 "helprs.skill": cs.skill_name,
                 "helprs.repo": cs.repo_full_name,
             },

apps/api/tests/integration/test_container_flow.py

@@ -27,6 +27,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
 from helprs.core.database import Base
+from helprs.modules.container.docker_client import RunnerContainer
 from helprs.modules.container.models import ContainerSession, ContainerStatus
 from helprs.modules.installation.models import BYOKConfig, Installation
 
@@ -84,6 +85,12 @@ async def container_logs(self, container_id: str, follow: bool = False) -> Async
     async def wait_container(self, container_id: str) -> int:
         return 0
 
+    async def container_is_running(self, container_id: str) -> bool:
+        return container_id in self.started and container_id not in self.removed
+
+    async def list_runners(self, *, boot_id: str) -> list[RunnerContainer]:
+        return []
+
     async def close(self) -> None:
         pass