fix(bay): prevent duplicate startup runs and GC deletes

Delay the first background scheduler cycle when startup execution has
already run so GC and browser learning do not execute twice
immediately.

Also skip orphan cargo deletion when a managed runtime still references
the cargo, avoiding premature cleanup during runtime teardown.

Add regression tests for scheduler startup timing, orphan cargo
protection, and warm pool reconciliation with a live runtime.

Author: RC-CHN

pkgs/bay/app/services/gc/scheduler.py

@@ -182,15 +182,27 @@ async def stop(self) -> None:
         self._log.info("gc.scheduler.stopped")
 
     async def _background_loop(self) -> None:
-        """Internal background loop."""
+        """Internal background loop.
+
+        Note:
+        - If run_on_startup is enabled, lifecycle already executed one cycle.
+          Sleep before the first loop cycle to avoid immediate duplicate execution.
+        """
+        first_iteration = True
+
         while self._running:
+            should_sleep = (first_iteration and self._config.run_on_startup) or (
+                not first_iteration
+            )
+            if should_sleep:
+                try:
+                    await asyncio.sleep(self._config.interval_seconds)
+                except asyncio.CancelledError:
+                    break
+
+            first_iteration = False
+
             try:
                 await self.run_once()
             except Exception as e:
                 self._log.exception("gc.scheduler.cycle_error", error=str(e))
-
-            # Sleep until next cycle (or until cancelled)
-            try:
-                await asyncio.sleep(self._config.interval_seconds)
-            except asyncio.CancelledError:
-                break

pkgs/bay/app/services/gc/tasks/orphan_cargo.py

@@ -63,6 +63,14 @@ async def run(self) -> GCResult:
 
         for cargo_id in orphans:
             try:
+                if await self._has_runtime_references(cargo_id):
+                    result.skipped_count += 1
+                    self._log.info(
+                        "gc.orphan_cargo.skip.runtime_in_use",
+                        cargo_id=cargo_id,
+                    )
+                    continue
+
                 await self._cargo_mgr.delete_internal_by_id(cargo_id)
                 result.cleaned_count += 1
                 self._log.info(
@@ -79,6 +87,20 @@ async def run(self) -> GCResult:
 
         return result
 
+    async def _has_runtime_references(self, cargo_id: str) -> bool:
+        """Check whether any runtime instance still references the cargo.
+
+        Conservative by design: if any runtime instance still carries the cargo label,
+        skip deletion for this GC cycle and let a later cycle retry after runtime cleanup.
+        """
+        instances = await self._driver.list_runtime_instances(
+            labels={
+                "bay.cargo_id": cargo_id,
+                "bay.managed": "true",
+            }
+        )
+        return len(instances) > 0
+
     async def _find_orphans(self) -> list[str]:
         """Find orphan managed cargo IDs."""
         orphan_ids: list[str] = []

pkgs/bay/app/services/skills/scheduler.py

@@ -466,12 +466,27 @@ async def stop(self) -> None:
         self._log.info("skills.browser.scheduler.stopped")
 
     async def _background_loop(self) -> None:
+        """Internal background loop.
+
+        Note:
+        - If run_on_startup is enabled, lifecycle already executed one cycle.
+          Sleep before the first loop cycle to avoid immediate duplicate execution.
+        """
+        first_iteration = True
+
         while self._running:
+            should_sleep = (first_iteration and self._config.run_on_startup) or (
+                not first_iteration
+            )
+            if should_sleep:
+                try:
+                    await asyncio.sleep(self._config.interval_seconds)
+                except asyncio.CancelledError:
+                    break
+
+            first_iteration = False
+
             try:
                 await self.run_once()
             except Exception as exc:
                 self._log.exception("skills.browser.scheduler.cycle_failed", error=str(exc))
-            try:
-                await asyncio.sleep(self._config.interval_seconds)
-            except asyncio.CancelledError:
-                break

pkgs/bay/tests/unit/gc/test_gc_scheduler.py

@@ -213,6 +213,31 @@ async def test_stop_idempotent(self, gc_config):
 
         assert not scheduler.is_running
 
+    @pytest.mark.asyncio
+    async def test_background_loop_sleeps_first_when_startup_cycle_already_ran(self, gc_config):
+        """When run_on_startup is enabled, background loop should not immediately run again."""
+        task = FakeGCTask("task1")
+        gc_config.run_on_startup = True
+        gc_config.interval_seconds = 0.2
+
+        scheduler = GCScheduler(
+            tasks=[task],
+            config=gc_config,
+        )
+
+        # Simulate lifecycle startup run happening before background loop starts.
+        await scheduler.run_once()
+        assert task.run_count == 1
+
+        await scheduler.start()
+        await asyncio.sleep(0.05)
+
+        # Background loop should still be in initial sleep window.
+        assert task.run_count == 1
+
+        await scheduler.stop()
+        assert not scheduler.is_running
+
 
 class TestNoopCoordinator:
     """Tests for NoopCoordinator."""

pkgs/bay/tests/unit/gc/test_gc_tasks.py

@@ -413,11 +413,12 @@ class TestOrphanCargoGC:
 
     @pytest.mark.asyncio
     async def test_finds_orphan_cargos(self):
-        """Should find managed cargos without a valid sandbox."""
+        """Should delete orphan cargos when no runtime still references them."""
         from app.services.gc.tasks.orphan_cargo import OrphanCargoGC
         from tests.fakes import FakeDriver
 
         driver = FakeDriver()
+        driver.list_runtime_instances = AsyncMock(return_value=[])
         db_session = AsyncMock()
 
         # Mock cargo manager
@@ -429,14 +430,49 @@ async def test_finds_orphan_cargos(self):
         result = await task.run()
 
         assert result.cleaned_count == 2
+        assert result.skipped_count == 0
         assert task._cargo_mgr.delete_internal_by_id.call_count == 2
 
+    @pytest.mark.asyncio
+    async def test_orphan_cargo_skips_when_runtime_still_references_volume(self):
+        """Should skip deleting orphan cargo when runtime instances still reference it."""
+        from app.services.gc.tasks.orphan_cargo import OrphanCargoGC
+        from tests.fakes import FakeDriver
+
+        driver = FakeDriver()
+        driver.list_runtime_instances = AsyncMock(
+            return_value=[
+                RuntimeInstance(
+                    id="container-1",
+                    name="bay-session-sess-1",
+                    labels={
+                        "bay.cargo_id": "ws-orphan-1",
+                        "bay.managed": "true",
+                    },
+                    state="running",
+                )
+            ]
+        )
+        db_session = AsyncMock()
+
+        task = OrphanCargoGC(driver, db_session)
+        task._find_orphans = AsyncMock(return_value=["ws-orphan-1"])
+        task._cargo_mgr = MagicMock()
+        task._cargo_mgr.delete_internal_by_id = AsyncMock()
+
+        result = await task.run()
+
+        assert result.cleaned_count == 0
+        assert result.skipped_count == 1
+        task._cargo_mgr.delete_internal_by_id.assert_not_called()
+
     @pytest.mark.asyncio
     async def test_orphan_cargo_no_orphans(self):
         """Should handle case when no orphan cargos are found."""
         from tests.fakes import FakeDriver
 
         driver = FakeDriver()
+        driver.list_runtime_instances = AsyncMock(return_value=[])
         db_session = AsyncMock()
 
         from app.services.gc.tasks.orphan_cargo import OrphanCargoGC

pkgs/bay/tests/unit/managers/test_browser_learning_scheduler.py

@@ -602,3 +602,34 @@ async def fake_background_loop() -> None:
 
     # stop again should be noop
     await scheduler.stop()
+
+
+@pytest.mark.asyncio
+async def test_scheduler_background_loop_sleeps_first_when_startup_cycle_already_ran(
+    monkeypatch: pytest.MonkeyPatch,
+    learning_config: BrowserLearningConfig,
+):
+    config = learning_config.model_copy(update={"run_on_startup": True, "interval_seconds": 0.2})
+    scheduler = scheduler_module.BrowserLearningScheduler(config=config)
+
+    run_once_calls = 0
+
+    async def fake_run_once():
+        nonlocal run_once_calls
+        run_once_calls += 1
+        return scheduler_module.BrowserLearningCycleResult()
+
+    monkeypatch.setattr(scheduler, "run_once", fake_run_once)
+
+    # Simulate lifecycle startup run happening before background loop starts.
+    await scheduler.run_once()
+    assert run_once_calls == 1
+
+    await scheduler.start()
+    await asyncio.sleep(0.05)
+
+    # Background loop should still be in its initial sleep window.
+    assert run_once_calls == 1
+
+    await scheduler.stop()
+    assert scheduler.is_running is False

pkgs/bay/tests/unit/warm_pool/test_warm_pool_manager.py

@@ -448,3 +448,70 @@ async def __aexit__(self, exc_type, exc, tb):
         updated_sandbox = sandbox_result.scalars().first()
         assert updated_sandbox is not None
         assert updated_sandbox.warm_state == WarmState.AVAILABLE.value
+
+    @pytest.mark.asyncio
+    async def test_reconcile_keeps_available_sandbox_when_single_runtime_alive(
+        self,
+        db_session,
+        driver,
+        monkeypatch,
+    ):
+        sandbox_mgr = SandboxManager(driver=driver, db_session=db_session)
+        sandbox = await sandbox_mgr.create_warm_sandbox(profile_id="python-default")
+        await sandbox_mgr.mark_warm_available(sandbox.id)
+
+        session = Session(
+            id="sess-warm-single",
+            sandbox_id=sandbox.id,
+            profile_id="python-default",
+            container_id="live-container",
+            endpoint="http://live-runtime",
+            observed_state=SessionStatus.RUNNING,
+            desired_state=SessionStatus.RUNNING,
+        )
+        db_session.add(session)
+        sandbox.current_session_id = session.id
+        await db_session.commit()
+
+        queue = _FakeWarmupQueue()
+        scheduler = WarmPoolScheduler(
+            config=SimpleNamespace(interval_seconds=60, run_on_startup=False),
+            warmup_queue=queue,
+        )
+
+        driver.set_status_override(
+            "live-container",
+            ContainerInfo(
+                container_id="live-container",
+                status=ContainerStatus.RUNNING,
+                endpoint="http://live-runtime",
+            ),
+        )
+
+        class _SessionFactory:
+            async def __aenter__(self):
+                return db_session
+
+            async def __aexit__(self, exc_type, exc, tb):
+                return False
+
+        monkeypatch.setattr(
+            "app.db.session.get_async_session",
+            lambda: _SessionFactory(),
+        )
+        monkeypatch.setattr("app.api.dependencies.get_driver", lambda: driver)
+
+        reconciled = await scheduler._reconcile_profile_runtime_state("python-default")
+        assert reconciled == 0
+        assert queue.enqueued == []
+
+        sandbox_result = await db_session.execute(select(Sandbox).where(Sandbox.id == sandbox.id))
+        updated_sandbox = sandbox_result.scalars().first()
+        assert updated_sandbox is not None
+        assert updated_sandbox.warm_state == WarmState.AVAILABLE.value
+
+        session_result = await db_session.execute(select(Session).where(Session.id == session.id))
+        updated_session = session_result.scalars().first()
+        assert updated_session is not None
+        assert updated_session.observed_state == SessionStatus.RUNNING
+        assert updated_session.endpoint == "http://live-runtime"