Do not count a stopped VM against HOST_LIMITS

docs/api.md

@@ -41,8 +41,9 @@ warm host. Callers cannot set it. An `Idempotency-Key` belongs to the
 service account that first used it. Another one reusing it gets `409`.
 
 `POST /hosts` returns `503` with `CAPACITY` and `Retry-After` when the
-provider has `HOST_LIMITS` live hosts. A live host is not in `error` and
-has not expired.
+provider has `HOST_LIMITS` live hosts. A live host is not in `error`, has
+not expired, and is not stopped by its provider (docker-sbx stops an idle
+sandbox).
 
 `POST /hosts` returns `422` with `SECRET_MISSING` when a secret is missing.
 The detail is the issuer's reason, for example that the GitHub App is not

docs/deploy.md

@@ -598,7 +598,7 @@ Core, optional:
 | `LEASE_DEFAULT_TTL` | `86400` | Lease TTL in seconds for hosts created without an explicit `expires_at`, and the extension applied by an empty `POST /hosts/{id}/renew`. An explicit `expires_at: null` at create time opts out of expiry entirely. |
 | `IDEMPOTENCY_KEY_TTL_HOURS` | `24` | Retention period for successful `Idempotency-Key` mappings. |
 | `POOL_SIZES` | `{}` | Warm hosts to keep ready per provider, as JSON (e.g. `{"exe": 2, "hetzner": 1}`). Overrides `POOL_SIZE` for the providers it names. |
-| `HOST_LIMITS` | `{}` | Most live hosts per provider, as JSON (e.g. `{"exe": 28, "docker": 6}`). A create at the limit returns `503` with `CAPACITY`. Warm pool hosts count. A provider it does not name has no limit. |
+| `HOST_LIMITS` | `{}` | Most live hosts per provider, as JSON (e.g. `{"exe": 28, "docker": 6}`). A create at the limit returns `503` with `CAPACITY`. Warm pool hosts count; a sandbox that docker-sbx stopped does not. A provider it does not name has no limit. |
 | `POOL_SIZE` | `0` | Warm hosts to keep ready for the default provider. `0` disables its pool. |
 | `POOL_HOST_MAX_AGE_HOURS` | `4` | Max age before the janitor reaps an unclaimed pool host. |
 | `POOL_MAX_CREATES_PER_TICK` | `2` | Upper bound on pool provisions per tick, across all providers; caps over-provision blast radius when ticks overlap. |

src/hosts/service.py

@@ -12,20 +12,15 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from uuid6 import uuid7
 
+import hosts.exceptions
+import providers.exceptions
 from core.database import async_session_factory
 from core.exceptions import ResourceNotFoundError
 from core.settings import Settings, get_settings
 from gateway.settings import GatewaySettings
 from host_secrets import catalog
 from host_secrets.exceptions import SecretsProxyNotConfiguredError, SecretStaticError
 from host_secrets.placeholder import Placeholder
-from hosts.exceptions import (
-    HostCapacityError,
-    HostKeyScanError,
-    HostStateError,
-    IdempotencyKeyConflictError,
-    ProvisioningFailedError,
-)
 from hosts.models import Host, HostStatus, IdempotencyKey
 from networking.tailscale import (
     DeviceDiscoveryTimeoutError,
@@ -34,12 +29,6 @@
 )
 from providers.base import VMProvider
 from providers.environment import get_persist
-from providers.exceptions import (
-    ProviderError,
-    ProviderNotFoundError,
-    UnknownProviderError,
-    UnsupportedSizingError,
-)
 from providers.registry import get_provider_names, get_vm_provider
 from secrets_exchange.client import SecretsExchange
 from secrets_exchange.secrets import IssuerError, MissingSecretError, Secret
@@ -111,7 +100,9 @@ async def get_or_create_host(
             registered = get_provider_names()
             if provider not in registered:
                 available = ", ".join(sorted(registered))
-                raise UnknownProviderError(f"unknown provider {provider!r}; available: {available}")
+                raise providers.exceptions.UnknownProviderError(
+                    f"unknown provider {provider!r}; available: {available}"
+                )
 
         if idempotency_key and (
             existing := await self._lookup_idempotency_key(idempotency_key, service_account)
@@ -154,7 +145,7 @@ async def get_or_create_host(
             await self._release_idempotency_loser(host)
             winner = await self._lookup_idempotency_key(idempotency_key, service_account)
             if not winner:
-                raise HostStateError("idempotency race could not be resolved")
+                raise hosts.exceptions.HostStateError("idempotency race could not be resolved")
             return winner
         return host
 
@@ -223,11 +214,11 @@ async def create_host(
         # never claims its own members.
         vm = get_vm_provider(provider)
         if instance_type and not vm.supports_instance_type:
-            raise UnsupportedSizingError(
+            raise providers.exceptions.UnsupportedSizingError(
                 f"provider {vm.name!r} does not support a per-request instance_type"
             )
         if disk_gb and not vm.supports_disk_gb:
-            raise UnsupportedSizingError(
+            raise providers.exceptions.UnsupportedSizingError(
                 f"provider {vm.name!r} does not support a per-request disk_gb"
             )
         proxy = self.settings.secrets_proxy_url and self.settings.secrets_proxy_ca_file
@@ -272,6 +263,10 @@ async def create_host(
             pool_member=pool_member,
         )
         if vm.name in self.settings.host_limits:
+            try:
+                stopped = await vm.stopped_vms()
+            except providers.exceptions.ProviderError as exc:
+                raise hosts.exceptions.ProvisioningFailedError(str(exc)) from exc
             # The lock serializes this provider's creates until the insert
             # commits, so two requests cannot both pass the count. SQLite has
             # no advisory locks; it serves single-process quickstarts.
@@ -285,11 +280,14 @@ async def create_host(
                     .select_from(Host)
                     .where(Host.provider == vm.name)
                     .where(Host.status != HostStatus.ERROR.value)
+                    .where(Host.name.not_in(stopped))
                     .where(or_(Host.expires_at.is_(None), Host.expires_at > now))
                 )
             ).scalar_one()
             if live >= self.settings.host_limits[vm.name]:
-                raise HostCapacityError(f"provider {vm.name!r} is at its host limit")
+                raise hosts.exceptions.HostCapacityError(
+                    f"provider {vm.name!r} is at its host limit"
+                )
         self.session.add(host)
         await self.session.commit()
         await self.session.refresh(host)
@@ -298,7 +296,7 @@ async def create_host(
         await self.session.refresh(host)
 
         if host.status == HostStatus.ERROR.value:
-            raise ProvisioningFailedError(host.last_error)
+            raise hosts.exceptions.ProvisioningFailedError(host.last_error)
 
         # A dedicated session, so self.session's transaction stays short for
         # callers that hold an advisory lock. The match on the in-flight value
@@ -326,7 +324,7 @@ async def _lookup_idempotency_key(self, key: str, service_account: str | None) -
             )
             if host:
                 if host.service_account != service_account:
-                    raise IdempotencyKeyConflictError(
+                    raise hosts.exceptions.IdempotencyKeyConflictError(
                         f"idempotency key {key} belongs to another service account"
                     )
                 return host
@@ -402,10 +400,12 @@ async def renew_host(self, host_id: uuid.UUID, *, expires_at: datetime | None =
             raise ResourceNotFoundError("host not found")
 
         if host.pool_member and not host.claimed_at:
-            raise HostStateError("unclaimed pool host is managed by pool maintenance")
+            raise hosts.exceptions.HostStateError(
+                "unclaimed pool host is managed by pool maintenance"
+            )
 
         if host.status not in RENEWABLE_STATUSES:
-            raise HostStateError(f"cannot renew a host in status {host.status}")
+            raise hosts.exceptions.HostStateError(f"cannot renew a host in status {host.status}")
 
         host.expires_at = get_vm_provider(host.provider).lease_end(
             host.created_at, expires_at or self._default_lease_expires_at()
@@ -452,7 +452,7 @@ async def delete_host(
             return False
 
         if not force and host.status in DELETE_BLOCKED_STATUSES:
-            raise HostStateError("host is still provisioning")
+            raise hosts.exceptions.HostStateError("host is still provisioning")
 
         if force or host.status in VM_BACKED_STATUSES:
             # An abandoned provision stranded in CREATING_VM may already have a
@@ -470,7 +470,7 @@ async def delete_host(
             await vm.secrets.delete_secrets(vm=host.name)
             try:
                 await vm.delete_vm(host.name)
-            except ProviderNotFoundError:
+            except providers.exceptions.ProviderNotFoundError:
                 # Evicted by the provider, or a previous delete got this far.
                 logger.warning(
                     "host VM already absent at provider during teardown: "
@@ -492,7 +492,7 @@ async def provision(self, host: Host) -> None:
         if vm.tailnet and not self.tailscale:
             await self.mark_failed(
                 host,
-                ProvisioningFailedError(
+                hosts.exceptions.ProvisioningFailedError(
                     f"provider {vm.name!r} joins the tailnet; set TAILSCALE_TAILNET, "
                     "TAILSCALE_AUTH_TAGS, TAILSCALE_OAUTH_CLIENT_ID and "
                     "TAILSCALE_OAUTH_CLIENT_SECRET"
@@ -520,7 +520,7 @@ async def provision(self, host: Host) -> None:
         except MissingSecretError as exc:
             await self.mark_failed(host, exc)
             raise
-        except (IssuerError, ProviderError, ValueError) as exc:
+        except (IssuerError, providers.exceptions.ProviderError, ValueError) as exc:
             await self.mark_failed(host, exc)
             return
 
@@ -533,7 +533,7 @@ async def provision(self, host: Host) -> None:
             # Without a gateway address the host's SSH coordinates would be dead.
             await self.mark_failed(
                 host,
-                ProvisioningFailedError(
+                hosts.exceptions.ProvisioningFailedError(
                     f"provider {vm.name!r} requires the SSH gateway; set GATEWAY_SSH_HOST"
                 ),
             )
@@ -547,7 +547,7 @@ async def provision(self, host: Host) -> None:
                 instance_type=host.instance_type,
                 disk_gb=host.disk_gb,
             )
-        except ProviderError as exc:
+        except providers.exceptions.ProviderError as exc:
             await self.mark_failed(host, exc)
             return
 
@@ -587,7 +587,7 @@ async def provision(self, host: Host) -> None:
 
         try:
             known_hosts_data = await self.scan_known_hosts(host)
-        except HostKeyScanError as exc:
+        except hosts.exceptions.HostKeyScanError as exc:
             await self.mark_failed(host, exc)
             return
 
@@ -661,7 +661,9 @@ async def scan_known_hosts(self, host: Host) -> bytes:
                         stderr=asyncio.subprocess.PIPE,
                     )
                 except OSError as error:
-                    raise HostKeyScanError(f"could not run ssh-keyscan: {error}") from error
+                    raise hosts.exceptions.HostKeyScanError(
+                        f"could not run ssh-keyscan: {error}"
+                    ) from error
                 stdout, stderr = await process.communicate()
                 collected += stdout
                 errors.append(stderr.decode().strip())
@@ -671,5 +673,7 @@ async def scan_known_hosts(self, host: Host) -> bytes:
             # read nothing during the gap.
             if time.monotonic() >= deadline:
                 detail = "; ".join(filter(None, errors)) or "empty output"
-                raise HostKeyScanError(f"ssh-keyscan never returned host keys: {detail}")
+                raise hosts.exceptions.HostKeyScanError(
+                    f"ssh-keyscan never returned host keys: {detail}"
+                )
             await asyncio.sleep(0.5)

src/hosts/tests/test_pool.py

@@ -224,6 +224,19 @@ async def test_create_host_refuses_a_provider_at_its_host_limit(monkeypatch, stu
     assert {provider: count for provider, count in counted.all()} == {"exe": 1, "stub": 1}
 
 
+async def test_create_host_gives_the_slot_of_a_stopped_vm(monkeypatch, stub_provider):
+    monkeypatch.setenv("HOST_LIMITS", '{"stub": 1}')
+    settings = Settings()  # pyright: ignore[reportCallIssue]
+    await _seed_pool_host(name="lb-stopped", provider="stub", pool_member=False)
+    monkeypatch.setattr(stub_provider, "stopped_vms", AsyncMock(return_value={"lb-stopped"}))
+    monkeypatch.setattr("hosts.service.HostService.provision", AsyncMock())
+
+    async with async_session_factory() as session:
+        await HostService(session, settings=settings).create_host(
+            env={}, image=None, provider="stub"
+        )
+
+
 async def test_create_host_skips_pool_when_sized(multi_pool_settings, monkeypatch, stub_provider):
     # Pool members are warmed at the provider's default size, so a sized
     # request must provision fresh even when its provider has a warm host.

src/providers/base.py

@@ -116,6 +116,11 @@ def lease_end(self, created_at: datetime, expires_at: datetime | None) -> dateti
             return min(expires_at, stop) if expires_at else stop
         return expires_at
 
+    async def stopped_vms(self) -> set[str]:
+        """Names of the VMs the provider stopped while idle. A stopped VM
+        holds no memory, so it takes no HOST_LIMITS slot."""
+        return set()
+
     async def open_gateway_tunnel(self, name: str) -> asyncssh.SSHClientConnection:
         """Open an SSH tunnel into a gateway host's sandbox. The gateway
         forwards TCP channels through it."""

src/providers/docker_sbx/api.py

@@ -2,6 +2,7 @@
 import json
 import re
 from pathlib import Path
+from typing import Any
 
 from .exceptions import DockerSbxNotFoundError, DockerSbxTransportError
 
@@ -125,11 +126,11 @@ async def load_template(self, archive: Path) -> None:
     async def remove_template(self, template: str) -> None:
         await self._run("template", "rm", "--force", template)
 
-    async def sandbox_count(self) -> int:
+    async def sandboxes(self) -> list[dict[str, Any]]:
         output = await self._run("ls", "--json")
         try:
             # Go writes an empty list as null.
-            return len(json.loads(output) or [])
+            return json.loads(output) or []
         except json.JSONDecodeError as error:
             raise DockerSbxTransportError(
                 f"sbx ls returned an unreadable sandbox list: {output.strip()!r}"

src/providers/docker_sbx/provider.py

@@ -214,10 +214,18 @@ async def delete_template_image(self, image: str) -> None:
         except DockerSbxProviderError as exc:
             raise ProviderTransportError(str(exc)) from exc
 
+    async def stopped_vms(self) -> set[str]:
+        # sandboxd stops a sandbox with no session and starts it on the next exec.
+        try:
+            sandboxes = await self.api.sandboxes()
+        except DockerSbxProviderError as exc:
+            raise ProviderTransportError(str(exc)) from exc
+        return {sandbox["name"] for sandbox in sandboxes if sandbox["status"] == "stopped"}
+
     async def diagnose(self) -> str:
-        count = await self.api.sandbox_count()
+        sandboxes = await self.api.sandboxes()
         await self.api.check_ssh_endpoint()
-        return f"sandboxd reachable, {count} sandbox(es), SSH endpoint ready"
+        return f"sandboxd reachable, {len(sandboxes)} sandbox(es), SSH endpoint ready"
 
     async def aclose(self) -> None:
         await self.docker.aclose()

src/providers/docker_sbx/tests/test_api.py

@@ -167,21 +167,21 @@ async def fake_exec(*args, **kwargs):
 
 
 @pytest.mark.asyncio
-async def test_sandbox_count_reads_the_listing(monkeypatch):
+async def test_sandboxes_reads_the_listing(monkeypatch):
     listing = b'[{"name": "sb-a"}, {"name": "sb-b"}]'
     create = AsyncMock(return_value=_process(stdout=listing))
     monkeypatch.setattr("providers.docker_sbx.api.asyncio.create_subprocess_exec", create)
 
-    assert await SbxCLI().sandbox_count() == 2
+    assert await SbxCLI().sandboxes() == [{"name": "sb-a"}, {"name": "sb-b"}]
 
 
 @pytest.mark.asyncio
-async def test_sandbox_count_treats_a_null_list_as_empty(monkeypatch):
+async def test_sandboxes_treats_a_null_list_as_empty(monkeypatch):
     # Go writes an empty list as null. An unused daemon shows null.
     create = AsyncMock(return_value=_process(stdout=b"null"))
     monkeypatch.setattr("providers.docker_sbx.api.asyncio.create_subprocess_exec", create)
 
-    assert await SbxCLI().sandbox_count() == 0
+    assert await SbxCLI().sandboxes() == []
 
 
 @pytest.mark.asyncio
@@ -304,4 +304,4 @@ async def test_missing_sbx_binary_maps_to_transport_error(monkeypatch):
     monkeypatch.setattr("providers.docker_sbx.api.asyncio.create_subprocess_exec", create)
 
     with pytest.raises(DockerSbxTransportError, match="could not be started"):
-        await SbxCLI().sandbox_count()
+        await SbxCLI().sandboxes()

src/providers/docker_sbx/tests/test_diagnose.py

@@ -14,7 +14,7 @@ def _provider(api: MagicMock) -> DockerSbxProvider:
 @pytest.mark.asyncio
 async def test_diagnose_reports_daemon_reachability():
     api = MagicMock()
-    api.sandbox_count = AsyncMock(return_value=2)
+    api.sandboxes = AsyncMock(return_value=[{}, {}])
     api.check_ssh_endpoint = AsyncMock()
 
     assert (
@@ -25,7 +25,7 @@ async def test_diagnose_reports_daemon_reachability():
 @pytest.mark.asyncio
 async def test_diagnose_raises_when_the_ssh_endpoint_is_unreachable():
     api = MagicMock()
-    api.sandbox_count = AsyncMock(return_value=2)
+    api.sandboxes = AsyncMock(return_value=[{}, {}])
     api.check_ssh_endpoint = AsyncMock(
         side_effect=DockerSbxTransportError("the sandboxd SSH endpoint is disabled")
     )
@@ -37,9 +37,7 @@ async def test_diagnose_raises_when_the_ssh_endpoint_is_unreachable():
 @pytest.mark.asyncio
 async def test_diagnose_raises_so_doctor_can_classify_the_failure():
     api = MagicMock()
-    api.sandbox_count = AsyncMock(
-        side_effect=DockerSbxTransportError("Not authenticated to Docker")
-    )
+    api.sandboxes = AsyncMock(side_effect=DockerSbxTransportError("Not authenticated to Docker"))
 
     with pytest.raises(DockerSbxTransportError):
         await _provider(api).diagnose()