feat(memory): protect Dream cron and polish migration UX

nanobot/agent/tools/cron.py

@@ -6,7 +6,7 @@ from typing import Any
 
 from nanobot.agent.tools.base import Tool
 from nanobot.cron.service import CronService
-from nanobot.cron.types import CronJobState, CronSchedule
+from nanobot.cron.types import CronJob, CronJobState, CronSchedule
 
 
 class CronTool(Tool):
@@ -219,6 +219,12 @@ class CronTool(Tool):
             lines.append(f"  Next run: {self._format_timestamp(state.next_run_at_ms, display_tz)}")
         return lines
 
+    @staticmethod
+    def _system_job_purpose(job: CronJob) -> str:
+        if job.name == "dream":
+            return "Dream memory consolidation for long-term memory."
+        return "System-managed internal job."
+
     def _list_jobs(self) -> str:
         jobs = self._cron.list_jobs()
         if not jobs:
@@ -227,6 +233,9 @@ class CronTool(Tool):
         for j in jobs:
             timing = self._format_timing(j.schedule)
             parts = [f"- {j.name} (id: {j.id}, {timing})"]
+            if j.payload.kind == "system_event":
+                parts.append(f"  Purpose: {self._system_job_purpose(j)}")
+                parts.append("  Protected: visible for inspection, but cannot be removed.")
             parts.extend(self._format_state(j.state, j.schedule))
             lines.append("\n".join(parts))
         return "Scheduled jobs:\n" + "\n".join(lines)
@@ -234,6 +243,19 @@ class CronTool(Tool):
     def _remove_job(self, job_id: str | None) -> str:
         if not job_id:
             return "Error: job_id is required for remove"
-        if self._cron.remove_job(job_id):
+        result = self._cron.remove_job(job_id)
+        if result == "removed":
             return f"Removed job {job_id}"
+        if result == "protected":
+            job = self._cron.get_job(job_id)
+            if job and job.name == "dream":
+                return (
+                    "Cannot remove job `dream`.\n"
+                    "This is a system-managed Dream memory consolidation job for long-term memory.\n"
+                    "It remains visible so you can inspect it, but it cannot be removed."
+                )
+            return (
+                f"Cannot remove job `{job_id}`.\n"
+                "This is a protected system-managed cron job."
+            )
         return f"Job {job_id} not found"

nanobot/cron/service.py

@@ -6,7 +6,7 @@ import time
 import uuid
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Callable, Coroutine
+from typing import Any, Callable, Coroutine, Literal
 
 from loguru import logger
 
@@ -365,9 +365,16 @@ class CronService:
         logger.info("Cron: registered system job '{}' ({})", job.name, job.id)
         return job
 
-    def remove_job(self, job_id: str) -> bool:
-        """Remove a job by ID."""
+    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._load_store()
+        job = next((j for j in store.jobs if j.id == job_id), None)
+        if job is None:
+            return "not_found"
+        if job.payload.kind == "system_event":
+            logger.info("Cron: refused to remove protected system job {}", job_id)
+            return "protected"
+
         before = len(store.jobs)
         store.jobs = [j for j in store.jobs if j.id != job_id]
         removed = len(store.jobs) < before
@@ -376,8 +383,9 @@ class CronService:
             self._save_store()
             self._arm_timer()
             logger.info("Cron: removed job {}", job_id)
+            return "removed"
 
-        return removed
+        return "not_found"
 
     def enable_job(self, job_id: str, enabled: bool = True) -> CronJob | None:
         """Enable or disable a job."""

tests/cron/test_cron_service.py

@@ -4,7 +4,7 @@ import json
 import pytest
 
 from nanobot.cron.service import CronService
-from nanobot.cron.types import CronSchedule
+from nanobot.cron.types import CronJob, CronPayload, CronSchedule
 
 
 def test_add_job_rejects_unknown_timezone(tmp_path) -> None:
@@ -141,3 +141,18 @@ async def test_running_service_honors_external_disable(tmp_path) -> None:
         assert called == []
     finally:
         service.stop()
+
+
+def test_remove_job_refuses_system_jobs(tmp_path) -> None:
+    service = CronService(tmp_path / "cron" / "jobs.json")
+    service.register_system_job(CronJob(
+        id="dream",
+        name="dream",
+        schedule=CronSchedule(kind="cron", expr="0 */2 * * *", tz="UTC"),
+        payload=CronPayload(kind="system_event"),
+    ))
+
+    result = service.remove_job("dream")
+
+    assert result == "protected"
+    assert service.get_job("dream") is not None

tests/cron/test_cron_tool_list.py

@@ -4,7 +4,7 @@ from datetime import datetime, timezone
 
 from nanobot.agent.tools.cron import CronTool
 from nanobot.cron.service import CronService
-from nanobot.cron.types import CronJobState, CronSchedule
+from nanobot.cron.types import CronJob, CronJobState, CronPayload, CronSchedule
 
 
 def _make_tool(tmp_path) -> CronTool:
@@ -262,6 +262,39 @@ def test_list_shows_next_run(tmp_path) -> None:
     assert "(UTC)" in result
 
 
+def test_list_includes_protected_dream_system_job_with_memory_purpose(tmp_path) -> None:
+    tool = _make_tool(tmp_path)
+    tool._cron.register_system_job(CronJob(
+        id="dream",
+        name="dream",
+        schedule=CronSchedule(kind="cron", expr="0 */2 * * *", tz="UTC"),
+        payload=CronPayload(kind="system_event"),
+    ))
+
+    result = tool._list_jobs()
+
+    assert "- dream (id: dream, cron: 0 */2 * * * (UTC))" in result
+    assert "Dream memory consolidation for long-term memory." in result
+    assert "cannot be removed" in result
+
+
+def test_remove_protected_dream_job_returns_clear_feedback(tmp_path) -> None:
+    tool = _make_tool(tmp_path)
+    tool._cron.register_system_job(CronJob(
+        id="dream",
+        name="dream",
+        schedule=CronSchedule(kind="cron", expr="0 */2 * * *", tz="UTC"),
+        payload=CronPayload(kind="system_event"),
+    ))
+
+    result = tool._remove_job("dream")
+
+    assert "Cannot remove job `dream`." in result
+    assert "Dream memory consolidation job for long-term memory" in result
+    assert "cannot be removed" in result
+    assert tool._cron.get_job("dream") is not None
+
+
 def test_add_cron_job_defaults_to_tool_timezone(tmp_path) -> None:
     tool = _make_tool_with_tz(tmp_path, "Asia/Shanghai")
     tool.set_context("telegram", "chat-1")