feat(api): warm up sandbox after create

enqueue a warmup hook on successful sandbox creation so runtime
startup can begin without delaying API response completion.

skip warmup when returning an idempotency cache hit, and add unit
tests covering warmup scheduling and idempotent create paths.

Author: RC-CHN

pkgs/bay/app/api/v1/sandboxes.py

@@ -9,15 +9,17 @@
 from datetime import datetime
 
 import structlog
-from fastapi import APIRouter, Header, Query
+from fastapi import APIRouter, BackgroundTasks, Header, Query
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 
 from app.adapters.base import BaseAdapter
 from app.adapters.gull import GullAdapter
 from app.adapters.ship import ShipAdapter
-from app.api.dependencies import AuthDep, IdempotencyServiceDep, SandboxManagerDep
+from app.api.dependencies import AuthDep, IdempotencyServiceDep, SandboxManagerDep, get_driver
 from app.config import get_settings
+from app.db.session import get_async_session
+from app.managers.sandbox import SandboxManager
 from app.models.sandbox import Sandbox, SandboxStatus
 from app.models.session import Session
 from app.router.capability.adapter_pool import default_adapter_pool
@@ -276,9 +278,50 @@ async def _query_containers_status(
 # Endpoints
 
 
+async def _warmup_sandbox_runtime_impl(*, sandbox_id: str, owner: str) -> None:
+    """Perform sandbox warmup in an isolated DB session."""
+    try:
+        async with get_async_session() as db:
+            manager = SandboxManager(driver=get_driver(), db_session=db)
+            sandbox = await manager.get(sandbox_id, owner)
+            await manager.ensure_running(sandbox)
+    except Exception as exc:
+        _log.warning(
+            "sandbox.warmup_failed",
+            sandbox_id=sandbox_id,
+            owner=owner,
+            error=str(exc),
+        )
+
+
+async def _warmup_sandbox_runtime(*, sandbox_id: str, owner: str) -> None:
+    """Schedule sandbox warmup in a detached task and return immediately.
+
+    This keeps request completion fast even if invoked via BackgroundTasks.
+    """
+    task = asyncio.create_task(
+        _warmup_sandbox_runtime_impl(sandbox_id=sandbox_id, owner=owner),
+        name=f"warmup-{sandbox_id}",
+    )
+
+    def _on_warmup_done(t: asyncio.Task[None]) -> None:
+        try:
+            t.result()
+        except Exception as exc:  # pragma: no cover
+            _log.warning(
+                "sandbox.warmup_task_failed",
+                sandbox_id=sandbox_id,
+                owner=owner,
+                error=str(exc),
+            )
+
+    task.add_done_callback(_on_warmup_done)
+
+
 @router.post("", response_model=SandboxResponse, status_code=201)
 async def create_sandbox(
     request: CreateSandboxRequest,
+    background_tasks: BackgroundTasks,
     sandbox_mgr: SandboxManagerDep,
     idempotency_svc: IdempotencyServiceDep,
     owner: AuthDep,
@@ -331,6 +374,14 @@ async def create_sandbox(
             status_code=201,
         )
 
+    # 4. Enqueue warmup hook. The hook itself detaches actual warmup work,
+    # so this does not block request completion on keep-alive connections.
+    background_tasks.add_task(
+        _warmup_sandbox_runtime,
+        sandbox_id=sandbox.id,
+        owner=owner,
+    )
+
     return response
 
 

pkgs/bay/tests/unit/api/test_sandbox_create_warmup.py

@@ -0,0 +1,105 @@
+"""Unit tests for sandbox create endpoint warmup behavior."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from fastapi import BackgroundTasks
+from fastapi.responses import JSONResponse
+
+from app.api.v1.sandboxes import CreateSandboxRequest, create_sandbox
+from app.models.sandbox import Sandbox
+
+
+@pytest.mark.asyncio
+async def test_create_sandbox_schedules_background_warmup_when_created():
+    """Fresh create should enqueue exactly one warmup background task."""
+    request = CreateSandboxRequest(profile="python-default", ttl=300)
+    background_tasks = BackgroundTasks()
+
+    sandbox_mgr = AsyncMock()
+    sandbox_mgr.create.return_value = Sandbox(
+        id="sandbox-abc123",
+        owner="user-1",
+        profile_id="python-default",
+        cargo_id="cargo-1",
+    )
+
+    idempotency_svc = AsyncMock()
+    idempotency_svc.check.return_value = None
+
+    resp = await create_sandbox(
+        request=request,
+        background_tasks=background_tasks,
+        sandbox_mgr=sandbox_mgr,
+        idempotency_svc=idempotency_svc,
+        owner="user-1",
+        idempotency_key=None,
+    )
+
+    assert resp.id == "sandbox-abc123"
+    assert len(background_tasks.tasks) == 1
+    task = background_tasks.tasks[0]
+    assert task.func.__name__ == "_warmup_sandbox_runtime"
+    assert task.kwargs == {"sandbox_id": "sandbox-abc123", "owner": "user-1"}
+
+
+@pytest.mark.asyncio
+async def test_create_sandbox_idempotency_cache_does_not_enqueue_warmup():
+    """Idempotency cache hit should return cached JSONResponse without warmup task."""
+    request = CreateSandboxRequest(profile="python-default", ttl=300)
+    background_tasks = BackgroundTasks()
+
+    sandbox_mgr = AsyncMock()
+
+    idempotency_svc = AsyncMock()
+    idempotency_svc.check.return_value = SimpleNamespace(
+        response={"id": "sandbox-cached", "status": "idle"},
+        status_code=201,
+    )
+
+    resp = await create_sandbox(
+        request=request,
+        background_tasks=background_tasks,
+        sandbox_mgr=sandbox_mgr,
+        idempotency_svc=idempotency_svc,
+        owner="user-1",
+        idempotency_key="idem-key-1",
+    )
+
+    assert isinstance(resp, JSONResponse)
+    assert len(background_tasks.tasks) == 0
+    sandbox_mgr.create.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_create_sandbox_with_idempotency_save_and_enqueue_warmup():
+    """Non-cached idempotent create should save key and enqueue warmup task."""
+    request = CreateSandboxRequest(profile="python-default", ttl=600)
+    background_tasks = BackgroundTasks()
+
+    sandbox_mgr = AsyncMock()
+    sandbox_mgr.create.return_value = Sandbox(
+        id="sandbox-new-idem",
+        owner="user-1",
+        profile_id="python-default",
+        cargo_id="cargo-2",
+    )
+
+    idempotency_svc = AsyncMock()
+    idempotency_svc.check.return_value = None
+
+    resp = await create_sandbox(
+        request=request,
+        background_tasks=background_tasks,
+        sandbox_mgr=sandbox_mgr,
+        idempotency_svc=idempotency_svc,
+        owner="user-1",
+        idempotency_key="idem-key-2",
+    )
+
+    assert resp.id == "sandbox-new-idem"
+    idempotency_svc.save.assert_awaited_once()
+    assert len(background_tasks.tasks) == 1