fix(cron): retire persisted heartbeat/dream jobs when disabled

Disabling gateway.heartbeat (or agents.defaults.dream) only skipped
job registration on startup; the previously persisted system job in
<workspace>/cron/jobs.json kept firing, and remove_job refuses to
touch protected system jobs. Add CronService.remove_system_job for
startup reconciliation and call it from the gateway disabled branches
so the config toggle takes effect after restart.

nanobot/cli/gateway_runtime.py

@@ -785,6 +785,7 @@ def _run_gateway(
         console.print(f"[green]✓[/green] Dream: {dream_cfg.describe_schedule()}")
     else:
         console.print("[yellow]○[/yellow] Dream: disabled")
+        cron.remove_system_job("dream")
         _advance_dream_cursor_if_behind(agent.context.memory)
 
     # Register Heartbeat system job (idempotent on restart)
@@ -799,6 +800,10 @@ def _run_gateway(
             ),
             payload=CronPayload(kind="system_event"),
         ))
+    else:
+        # Retire any previously persisted heartbeat job so that disabling
+        # gateway.heartbeat in config takes effect after restart.
+        cron.remove_system_job("heartbeat")
 
     async def _open_browser_when_ready() -> None:
         """Wait for the gateway to bind, then point the user's browser at the webui."""

nanobot/cron/service.py

@@ -718,6 +718,25 @@ class CronService:
         logger.info("Cron: registered system job '{}' ({})", job.name, job.id)
         return job
 
+    def remove_system_job(self, job_id: str) -> bool:
+        """Remove an internal system job by id (startup reconciliation).
+
+        Unlike ``remove_job``, this bypasses the protected-system-job guard:
+        the gateway retires a persisted system job whose config has been
+        disabled, so e.g. ``gateway.heartbeat.enabled=false`` actually takes
+        effect after restart instead of the leftover job firing forever.
+        Returns True when a job was removed.
+        """
+        store = self._require_store()
+        before = len(store.jobs)
+        store.jobs = [j for j in store.jobs if j.id != job_id]
+        removed = len(store.jobs) < before
+        if removed:
+            self._save_store()
+            self._arm_timer()
+            logger.info("Cron: removed system job {}", job_id)
+        return removed
+
     def remove_job(self, job_id: str) -> Literal["removed", "protected", "not_found"]:
         """Remove a job by ID, unless it is a protected system job."""
         store = self._require_store()

tests/cli/test_commands.py

@@ -3221,6 +3221,9 @@ def test_gateway_local_trigger_queue_submits_agent_turns(
         def register_system_job(self, _job) -> None:
             return None
 
+        def remove_system_job(self, _job_id: str) -> bool:
+            return False
+
     class _FakeAgentLoop(_GatewayAgentContractStub):
         @classmethod
         def from_config(cls, config, bus=None, **extra):

tests/cron/test_cron_persistence.py

@@ -211,6 +211,7 @@ def test_load_store_falls_back_to_in_memory_on_corruption_after_start(
         ("enable_job", lambda service: service.enable_job("missing", enabled=False)),
         ("update_job", lambda service: service.update_job("missing", name="new name")),
         ("register_system_job", lambda service: service.register_system_job(_system_job())),
+        ("remove_system_job", lambda service: service.remove_system_job("heartbeat")),
     ],
 )
 def test_public_apis_raise_clear_error_for_unavailable_corrupt_store(

tests/cron/test_cron_service.py

@@ -785,6 +785,39 @@ def test_remove_job_refuses_system_jobs(tmp_path) -> None:
     assert service.get_job("dream") is not None
 
 
+def test_remove_system_job_retires_persisted_system_job(tmp_path) -> None:
+    """Disabling a system job (e.g. gateway.heartbeat.enabled=false) must
+    actually retire the previously persisted job, not just skip registration."""
+    store_path = tmp_path / "cron" / "jobs.json"
+    service = CronService(store_path)
+    service.register_system_job(CronJob(
+        id="heartbeat",
+        name="heartbeat",
+        schedule=CronSchedule(kind="every", every_ms=1_800_000, tz="UTC"),
+        payload=CronPayload(kind="system_event"),
+    ))
+    assert service.get_job("heartbeat") is not None
+
+    removed = service.remove_system_job("heartbeat")
+
+    assert removed is True
+    assert service.get_job("heartbeat") is None
+    # Removal must persist: a fresh instance (next gateway start) must not
+    # resurrect the job from the on-disk store.
+    assert CronService(store_path).get_job("heartbeat") is None
+    # Idempotent: removing a missing system job reports False without raising.
+    assert service.remove_system_job("heartbeat") is False
+    # User-facing removal still protects remaining system jobs.
+    other = CronService(store_path)
+    other.register_system_job(CronJob(
+        id="dream",
+        name="dream",
+        schedule=CronSchedule(kind="cron", expr="0 */2 * * *", tz="UTC"),
+        payload=CronPayload(kind="system_event"),
+    ))
+    assert other.remove_job("dream") == "protected"
+
+
 @pytest.mark.asyncio
 async def test_start_server_not_jobs(tmp_path):
     store_path = tmp_path / "cron" / "jobs.json"