fix(sandbox): run auto-stop on every admin pod

rock/sandbox/base_manager.py

@@ -83,9 +83,15 @@ def _setup_job_check_scheduler(self):
             )
             logger.info("auto_transition and reconcile jobs registered (primary pod)")
         else:
-            logger.info("auto_transition and reconcile jobs skipped (non-primary pod)")
+            self.scheduler.add_job(
+                func=self._auto_stop_expired,
+                trigger=IntervalTrigger(seconds=self._auto_transition_interval),
+                id="auto_stop_expired",
+                name="Sandbox Auto Stop Expired",
+            )
+            logger.info("auto_stop_expired job registered (non-primary pod); other lifecycle jobs skipped")
         self.scheduler.start()
-        logger.info("APScheduler started for auto_transition and reconcile")
+        logger.info("APScheduler started for lifecycle jobs")
 
     async def _collect_and_report_metrics(self):
         start_time = time.time()
@@ -169,6 +175,9 @@ async def _collect_sandbox_meta(self) -> tuple[int, dict[str, dict[str, str]]]:
     @abstractmethod
     async def _auto_transition(self): ...
 
+    @abstractmethod
+    async def _auto_stop_expired(self): ...
+
     @abstractmethod
     async def _reconcile(self): ...
 

tests/unit/sandbox/test_base_manager_scheduler.py

@@ -0,0 +1,50 @@
+from unittest.mock import MagicMock, patch
+
+from rock.sandbox.base_manager import BaseManager
+
+
+class _ConcreteManager(BaseManager):
+    async def _auto_transition(self): ...
+
+    async def _auto_stop_expired(self): ...
+
+    async def _reconcile(self): ...
+
+
+def _manager() -> _ConcreteManager:
+    manager = _ConcreteManager.__new__(_ConcreteManager)
+    manager._auto_transition_interval = 180
+    manager._reconcile_interval = 30
+    return manager
+
+
+@patch("rock.sandbox.base_manager.AsyncIOScheduler")
+@patch("rock.sandbox.base_manager.is_primary_pod", return_value=True)
+def test_primary_registers_full_lifecycle_and_reconcile(mock_is_primary, mock_scheduler_cls):
+    scheduler = MagicMock()
+    mock_scheduler_cls.return_value = scheduler
+
+    manager = _manager()
+    manager._setup_job_check_scheduler()
+
+    assert [call.kwargs["id"] for call in scheduler.add_job.call_args_list] == ["auto_transition", "reconcile"]
+    assert scheduler.add_job.call_args_list[0].kwargs["func"] == manager._auto_transition
+    scheduler.start.assert_called_once_with()
+    mock_is_primary.assert_called_once_with()
+
+
+@patch("rock.sandbox.base_manager.AsyncIOScheduler")
+@patch("rock.sandbox.base_manager.is_primary_pod", return_value=False)
+def test_non_primary_registers_only_auto_stop_expired(mock_is_primary, mock_scheduler_cls):
+    scheduler = MagicMock()
+    mock_scheduler_cls.return_value = scheduler
+
+    manager = _manager()
+    manager._setup_job_check_scheduler()
+
+    scheduler.add_job.assert_called_once()
+    job = scheduler.add_job.call_args.kwargs
+    assert job["id"] == "auto_stop_expired"
+    assert job["func"] == manager._auto_stop_expired
+    scheduler.start.assert_called_once_with()
+    mock_is_primary.assert_called_once_with()

tests/unit/sandbox/test_collect_sandbox_meta.py

@@ -37,11 +37,11 @@ def base_manager(meta_store):
     from rock.sandbox.base_manager import BaseManager
 
     class _ConcreteManager(BaseManager):
-        async def _auto_transition(self):
-            ...
+        async def _auto_transition(self): ...
 
-        async def _reconcile(self):
-            ...
+        async def _auto_stop_expired(self): ...
+
+        async def _reconcile(self): ...
 
     with patch.object(BaseManager, "__init__", lambda self, *a, **kw: None):
         mgr = _ConcreteManager.__new__(_ConcreteManager)