feat(cron): add configurable max_repeat safeguard for cron job repeats (#59333)

cron/jobs.py

@@ -1220,6 +1220,15 @@ def create_job(
     if normalized_attach is not None:
         job["attach_to_session"] = normalized_attach
 
+    # Apply max_repeat safeguard: clamp repeat.times to the configured
+    # maximum (default 1000). Lazy-import to avoid circular dependency —
+    # scheduler.py imports from this module.
+    try:
+        from cron.scheduler import _apply_max_repeat_to_job
+        _apply_max_repeat_to_job(job)
+    except Exception:
+        pass
+
     with _jobs_lock():
         jobs = load_jobs()
         jobs.append(job)

cron/scheduler.py

@@ -209,6 +209,55 @@ def _resolve_cron_enabled_toolsets(job: dict, cfg: dict) -> list[str] | None:
     "qqbot", "yuanbao",
 })
 
+# Maximum repeat count for cron jobs — a safeguard against accidental
+# high values.  Configurable via cron.max_repeat in config.yaml.
+_MAX_REPEAT_DEFAULT = 1000
+
+
+def _get_max_repeat(cfg: Optional[dict] = None) -> int:
+    """Read ``max_repeat`` from the cron config section.
+
+    Returns the configured value cast to int, or ``_MAX_REPEAT_DEFAULT``
+    (1000) if unset or on any parse error.  Accepts an optional pre-loaded
+    ``cfg`` dict to avoid re-reading config.yaml when the caller already
+    has one.
+    """
+    try:
+        if cfg is None:
+            cfg = load_config() or {}
+        cron_cfg = (cfg.get("cron", {}) or {}) if isinstance(cfg, dict) else {}
+        value = cron_cfg.get("max_repeat", _MAX_REPEAT_DEFAULT)
+        return int(value) if value is not None else _MAX_REPEAT_DEFAULT
+    except Exception:
+        return _MAX_REPEAT_DEFAULT
+
+
+def _apply_max_repeat_to_job(job: dict, max_repeat: Optional[int] = None) -> bool:
+    """Clamp ``repeat.times`` on a job dict to ``max_repeat``.
+
+    If ``max_repeat`` is None, reads it from config via ``_get_max_repeat``.
+    Infinite repeats (``times`` is None) are left untouched.
+
+    Returns True if the value was clamped, False if unchanged.
+    """
+    if max_repeat is None:
+        max_repeat = _get_max_repeat()
+    repeat = job.get("repeat")
+    if not isinstance(repeat, dict):
+        return False
+    times = repeat.get("times")
+    if times is None:
+        return False  # infinite — intentional, don't clamp
+    if times > max_repeat:
+        repeat["times"] = max_repeat
+        logger.warning(
+            "Job '%s': repeat count %d exceeds max_repeat=%d — clamped to %d",
+            job.get("name", job.get("id", "?")), times, max_repeat, max_repeat,
+        )
+        return True
+    return False
+
+
 # Platforms that support a configured cron/notification home target, mapped to
 # the environment variable used by gateway setup/runtime config.
 _HOME_TARGET_ENV_VARS = {

tests/cron/test_scheduler.py

@@ -4868,3 +4868,87 @@ def test_all_targets_fail_returns_combined_errors(self):
         assert "a@example.com" in result
         assert "b@example.com" in result
         assert mock_pool.submit.call_count == 2
+
+
+class TestMaxRepeatSafeguard:
+    """Tests for the ``max_repeat`` safeguard in cron job creation.
+
+    ``_get_max_repeat`` reads the configured maximum from the cron config
+    section.  ``_apply_max_repeat_to_job`` clamps ``repeat.times`` on a job
+    dict to that maximum and logs a warning when clamping.
+    """
+
+    # ------------------------------------------------------------------
+    # _get_max_repeat
+    # ------------------------------------------------------------------
+
+    def test_default_max_repeat(self):
+        """Default max_repeat is 1000 when config has no setting."""
+        from cron.scheduler import _get_max_repeat
+
+        assert _get_max_repeat(cfg={}) == 1000
+        assert _get_max_repeat(cfg=None) == 1000
+
+    def test_max_repeat_from_config(self):
+        """Reads max_repeat from cron config section."""
+        from cron.scheduler import _get_max_repeat
+
+        assert _get_max_repeat(cfg={"cron": {"max_repeat": 500}}) == 500
+
+    def test_max_repeat_invalid_config_returns_default(self):
+        """Non-integer or missing values fall back to default."""
+        from cron.scheduler import _get_max_repeat
+
+        assert _get_max_repeat(cfg={"cron": {"max_repeat": "not-a-number"}}) == 1000
+        assert _get_max_repeat(cfg={"cron": {"max_repeat": None}}) == 1000
+
+    def test_max_repeat_zero_or_negative_is_accepted(self):
+        """Zero and negative values from config are accepted as-is
+        (they disable repeats, which is a valid policy choice)."""
+        from cron.scheduler import _get_max_repeat
+
+        assert _get_max_repeat(cfg={"cron": {"max_repeat": 0}}) == 0
+        assert _get_max_repeat(cfg={"cron": {"max_repeat": -1}}) == -1
+
+    # ------------------------------------------------------------------
+    # _apply_max_repeat_to_job
+    # ------------------------------------------------------------------
+
+    def test_clamp_high_repeat(self):
+        """repeat.times > max_repeat is clamped."""
+        from cron.scheduler import _apply_max_repeat_to_job
+
+        job = {"id": "j1", "name": "test", "repeat": {"times": 5000, "completed": 0}}
+        clamped = _apply_max_repeat_to_job(job, max_repeat=1000)
+        assert clamped is True
+        assert job["repeat"]["times"] == 1000
+
+    def test_no_clamp_when_below_max(self):
+        """repeat.times <= max_repeat is left unchanged."""
+        from cron.scheduler import _apply_max_repeat_to_job
+
+        job = {"id": "j2", "repeat": {"times": 50, "completed": 0}}
+        clamped = _apply_max_repeat_to_job(job, max_repeat=1000)
+        assert clamped is False
+        assert job["repeat"]["times"] == 50
+
+    def test_no_clamp_for_infinite_repeat(self):
+        """None (infinite) repeat is never clamped."""
+        from cron.scheduler import _apply_max_repeat_to_job
+
+        job = {"id": "j3", "name": "forever", "repeat": {"times": None, "completed": 0}}
+        clamped = _apply_max_repeat_to_job(job, max_repeat=1000)
+        assert clamped is False
+        assert job["repeat"]["times"] is None
+
+    def test_clamp_logs_warning(self, caplog):
+        """A warning is logged when clamping occurs."""
+        from cron.scheduler import _apply_max_repeat_to_job
+
+        caplog.set_level(logging.WARNING)
+
+        job = {"id": "j4", "name": "boom", "repeat": {"times": 9999, "completed": 0}}
+        _apply_max_repeat_to_job(job, max_repeat=1000)
+
+        assert "repeat count 9999 exceeds max_repeat=1000" in caplog.text
+        assert "boom" in caplog.text