feat: add notify parameter to cronjob tool for delivery control

Adds a 'notify' parameter to the cronjob tool that controls when
delivery happens:

- 'always' (default): deliver every run (current behavior)
- 'changes_only': the cron agent can respond with [SILENT] to
  suppress delivery when nothing new to report. The scheduler
  injects guidance into the prompt so the agent knows about this.
- 'never': skip delivery entirely, only save output locally

The notify field is stored in the job JSON, passed through create
and update, and shown in job listings. Output is always saved to
disk for audit regardless of notify mode. Failed jobs always
deliver regardless of the setting.

Changes:
- cron/jobs.py: accept and store notify parameter in create_job
- tools/cronjob_tools.py: add notify to function, schema, handler
  lambda, format_job, and update path
- cron/scheduler.py: SILENT_MARKER constant, notify-aware delivery
  logic in tick(), prompt injection for changes_only jobs
- tests: 14 new tests covering all three notify modes, prompt
  injection, case insensitivity, failure delivery, output saving

cron/jobs.py

@@ -303,6 +303,7 @@ def create_job(
     model: Optional[str] = None,
     provider: Optional[str] = None,
     base_url: Optional[str] = None,
+    notify: Optional[str] = None,
 ) -> Dict[str, Any]:
     """
     Create a new cron job.
@@ -319,6 +320,9 @@ def create_job(
         model: Optional per-job model override
         provider: Optional per-job provider override
         base_url: Optional per-job base URL override
+        notify: Delivery notification mode: "always" (default), "changes_only",
+                or "never". "changes_only" lets the cron agent suppress delivery
+                by responding with [SILENT]. "never" skips delivery entirely.
 
     Returns:
         The created job dict
@@ -371,6 +375,7 @@ def create_job(
         "last_error": None,
         # Delivery configuration
         "deliver": deliver,
+        "notify": notify or "always",
         "origin": origin,  # Tracks where job was created for "origin" delivery
     }
 

cron/scheduler.py

@@ -37,6 +37,11 @@ sys.path.insert(0, str(Path(__file__).parent.parent))
 
 from cron.jobs import get_due_jobs, mark_job_run, save_job_output
 
+# Sentinel: when a job has notify="changes_only", the cron agent can start
+# its response with this marker to suppress delivery.  Output is still saved
+# locally for audit.
+SILENT_MARKER = "[SILENT]"
+
 # Resolve Hermes home directory (respects HERMES_HOME override)
 _hermes_home = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes"))
 
@@ -179,11 +184,24 @@ def _deliver_result(job: dict, content: str) -> None:
 def _build_job_prompt(job: dict) -> str:
     """Build the effective prompt for a cron job, optionally loading one or more skills first."""
     prompt = job.get("prompt", "")
+    notify = job.get("notify", "always")
     skills = job.get("skills")
     if skills is None:
         legacy = job.get("skill")
         skills = [legacy] if legacy else []
 
+    # When notify=changes_only, prepend guidance so the cron agent knows
+    # it can suppress delivery by starting its response with [SILENT].
+    if notify == "changes_only":
+        silent_hint = (
+            "[SYSTEM: This job uses notify=changes_only. If you have nothing new "
+            "or noteworthy to report, respond with exactly \"[SILENT]\" (optionally "
+            "followed by a brief internal note). This suppresses delivery to the "
+            "user while still saving output locally. Only use [SILENT] when there "
+            "are genuinely no changes worth reporting.]\n\n"
+        )
+        prompt = silent_hint + prompt
+
     skill_names = [str(name).strip() for name in skills if str(name).strip()]
     if not skill_names:
         return prompt
@@ -481,8 +499,20 @@ def tick(verbose: bool = True) -> int:
                     logger.info("Output saved to: %s", output_file)
 
                 # Deliver the final response to the origin/target chat
+                notify_mode = job.get("notify", "always")
                 deliver_content = final_response if success else f"⚠️ Cron job '{job.get('name', job['id'])}' failed:\n{error}"
-                if deliver_content:
+
+                # Determine whether to suppress delivery based on notify mode
+                should_deliver = bool(deliver_content)
+                if should_deliver and success and notify_mode == "never":
+                    logger.info("Job '%s': notify=never — skipping delivery", job["id"])
+                    should_deliver = False
+                elif should_deliver and success and notify_mode == "changes_only":
+                    if deliver_content.strip().upper().startswith(SILENT_MARKER):
+                        logger.info("Job '%s': agent returned %s — skipping delivery", job["id"], SILENT_MARKER)
+                        should_deliver = False
+
+                if should_deliver:
                     try:
                         _deliver_result(job, deliver_content)
                     except Exception as de:

tests/cron/test_scheduler.py

@@ -7,7 +7,13 @@ from unittest.mock import AsyncMock, patch, MagicMock
 
 import pytest
 
-from cron.scheduler import _resolve_origin, _resolve_delivery_target, _deliver_result, run_job
+from cron.scheduler import (
+    _resolve_origin,
+    _resolve_delivery_target,
+    _deliver_result,
+    run_job,
+    SILENT_MARKER,
+)
 
 
 class TestResolveOrigin:
@@ -449,3 +455,154 @@ class TestRunJobSkillBacked:
         assert "Instructions for blogwatcher." in prompt_arg
         assert "Instructions for find-nearby." in prompt_arg
         assert "Combine the results." in prompt_arg
+
+
+class TestNotifyMode:
+    """Verify the notify parameter controls delivery behavior."""
+
+    def _make_job(self, notify="always"):
+        return {
+            "id": "monitor-job",
+            "name": "monitor",
+            "deliver": "origin",
+            "notify": notify,
+            "origin": {"platform": "telegram", "chat_id": "123"},
+        }
+
+    # -- notify=always (default) --
+
+    def test_always_delivers_normally(self):
+        with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job("always")]), \
+             patch("cron.scheduler.run_job", return_value=(True, "# output", "Results here", None)), \
+             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
+             patch("cron.scheduler._deliver_result") as deliver_mock, \
+             patch("cron.scheduler.mark_job_run"):
+            from cron.scheduler import tick
+            tick(verbose=False)
+        deliver_mock.assert_called_once()
+
+    def test_always_delivers_even_silent_response(self):
+        """With notify=always, [SILENT] in the response is just normal text."""
+        with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job("always")]), \
+             patch("cron.scheduler.run_job", return_value=(True, "# output", "[SILENT] nothing", None)), \
+             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
+             patch("cron.scheduler._deliver_result") as deliver_mock, \
+             patch("cron.scheduler.mark_job_run"):
+            from cron.scheduler import tick
+            tick(verbose=False)
+        deliver_mock.assert_called_once()
+
+    # -- notify=never --
+
+    def test_never_skips_delivery(self, caplog):
+        with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job("never")]), \
+             patch("cron.scheduler.run_job", return_value=(True, "# output", "Important results!", None)), \
+             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
+             patch("cron.scheduler._deliver_result") as deliver_mock, \
+             patch("cron.scheduler.mark_job_run"):
+            from cron.scheduler import tick
+            with caplog.at_level(logging.INFO, logger="cron.scheduler"):
+                tick(verbose=False)
+        deliver_mock.assert_not_called()
+        assert any("notify=never" in r.message for r in caplog.records)
+
+    def test_never_still_delivers_on_failure(self):
+        """Failed jobs always get delivered regardless of notify mode."""
+        with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job("never")]), \
+             patch("cron.scheduler.run_job", return_value=(False, "# output", "", "some error")), \
+             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
+             patch("cron.scheduler._deliver_result") as deliver_mock, \
+             patch("cron.scheduler.mark_job_run"):
+            from cron.scheduler import tick
+            tick(verbose=False)
+        deliver_mock.assert_called_once()
+
+    # -- notify=changes_only --
+
+    def test_changes_only_delivers_normal_response(self):
+        with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job("changes_only")]), \
+             patch("cron.scheduler.run_job", return_value=(True, "# output", "New data found!", None)), \
+             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
+             patch("cron.scheduler._deliver_result") as deliver_mock, \
+             patch("cron.scheduler.mark_job_run"):
+            from cron.scheduler import tick
+            tick(verbose=False)
+        deliver_mock.assert_called_once()
+
+    def test_changes_only_suppresses_silent(self, caplog):
+        with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job("changes_only")]), \
+             patch("cron.scheduler.run_job", return_value=(True, "# output", "[SILENT]", None)), \
+             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
+             patch("cron.scheduler._deliver_result") as deliver_mock, \
+             patch("cron.scheduler.mark_job_run"):
+            from cron.scheduler import tick
+            with caplog.at_level(logging.INFO, logger="cron.scheduler"):
+                tick(verbose=False)
+        deliver_mock.assert_not_called()
+        assert any(SILENT_MARKER in r.message for r in caplog.records)
+
+    def test_changes_only_suppresses_silent_with_note(self):
+        with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job("changes_only")]), \
+             patch("cron.scheduler.run_job", return_value=(True, "# output", "[SILENT] No changes detected", None)), \
+             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
+             patch("cron.scheduler._deliver_result") as deliver_mock, \
+             patch("cron.scheduler.mark_job_run"):
+            from cron.scheduler import tick
+            tick(verbose=False)
+        deliver_mock.assert_not_called()
+
+    def test_changes_only_case_insensitive(self):
+        with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job("changes_only")]), \
+             patch("cron.scheduler.run_job", return_value=(True, "# output", "[silent] nothing new", None)), \
+             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
+             patch("cron.scheduler._deliver_result") as deliver_mock, \
+             patch("cron.scheduler.mark_job_run"):
+            from cron.scheduler import tick
+            tick(verbose=False)
+        deliver_mock.assert_not_called()
+
+    def test_changes_only_still_delivers_on_failure(self):
+        with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job("changes_only")]), \
+             patch("cron.scheduler.run_job", return_value=(False, "# output", "", "error msg")), \
+             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
+             patch("cron.scheduler._deliver_result") as deliver_mock, \
+             patch("cron.scheduler.mark_job_run"):
+            from cron.scheduler import tick
+            tick(verbose=False)
+        deliver_mock.assert_called_once()
+
+    def test_output_saved_even_when_suppressed(self):
+        with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job("changes_only")]), \
+             patch("cron.scheduler.run_job", return_value=(True, "# full output", "[SILENT]", None)), \
+             patch("cron.scheduler.save_job_output") as save_mock, \
+             patch("cron.scheduler._deliver_result") as deliver_mock, \
+             patch("cron.scheduler.mark_job_run"):
+            save_mock.return_value = "/tmp/out.md"
+            from cron.scheduler import tick
+            tick(verbose=False)
+        save_mock.assert_called_once_with("monitor-job", "# full output")
+        deliver_mock.assert_not_called()
+
+
+class TestBuildJobPromptNotify:
+    """Verify _build_job_prompt injects [SILENT] guidance for changes_only jobs."""
+
+    def test_changes_only_injects_silent_hint(self):
+        from cron.scheduler import _build_job_prompt
+        job = {"prompt": "Check for updates", "notify": "changes_only"}
+        result = _build_job_prompt(job)
+        assert "[SILENT]" in result
+        assert "Check for updates" in result
+
+    def test_always_does_not_inject_hint(self):
+        from cron.scheduler import _build_job_prompt
+        job = {"prompt": "Check for updates", "notify": "always"}
+        result = _build_job_prompt(job)
+        assert "[SILENT]" not in result
+        assert result == "Check for updates"
+
+    def test_missing_notify_defaults_to_no_hint(self):
+        from cron.scheduler import _build_job_prompt
+        job = {"prompt": "Check for updates"}
+        result = _build_job_prompt(job)
+        assert "[SILENT]" not in result

tools/cronjob_tools.py

@@ -128,6 +128,7 @@ def _format_job(job: Dict[str, Any]) -> Dict[str, Any]:
         "schedule": job.get("schedule_display"),
         "repeat": _repeat_display(job),
         "deliver": job.get("deliver", "local"),
+        "notify": job.get("notify", "always"),
         "next_run_at": job.get("next_run_at"),
         "last_run_at": job.get("last_run_at"),
         "last_status": job.get("last_status"),
@@ -146,6 +147,7 @@ def cronjob(
     name: Optional[str] = None,
     repeat: Optional[int] = None,
     deliver: Optional[str] = None,
+    notify: Optional[str] = None,
     include_disabled: bool = False,
     skill: Optional[str] = None,
     skills: Optional[List[str]] = None,
@@ -178,6 +180,7 @@ def cronjob(
                 name=name,
                 repeat=repeat,
                 deliver=deliver,
+                notify=notify,
                 origin=_origin_from_env(),
                 skills=canonical_skills,
                 model=_normalize_optional_job_value(model),
@@ -255,6 +258,8 @@ def cronjob(
                 updates["name"] = name
             if deliver is not None:
                 updates["deliver"] = deliver
+            if notify is not None:
+                updates["notify"] = notify
             if skills is not None or skill is not None:
                 canonical_skills = _canonical_skills(skill, skills)
                 updates["skills"] = canonical_skills
@@ -374,6 +379,11 @@ Important safety rule: cron-run sessions should not recursively schedule more cr
                 "type": "string",
                 "description": "Delivery target: origin, local, telegram, discord, signal, sms, or platform:chat_id"
             },
+            "notify": {
+                "type": "string",
+                "enum": ["always", "changes_only", "never"],
+                "description": "When to notify: 'always' (default) delivers every run, 'changes_only' lets the cron agent suppress delivery when nothing new to report, 'never' saves output locally only"
+            },
             "model": {
                 "type": "string",
                 "description": "Optional per-job model override used when the cron job runs"
@@ -444,6 +454,7 @@ registry.register(
         name=args.get("name"),
         repeat=args.get("repeat"),
         deliver=args.get("deliver"),
+        notify=args.get("notify"),
         include_disabled=args.get("include_disabled", False),
         skill=args.get("skill"),
         skills=args.get("skills"),