fix: normalize cron repeat updates

cron/jobs.py

@@ -363,6 +363,73 @@ def _apply_skill_fields(job: Dict[str, Any]) -> Dict[str, Any]:
     return normalized
 
 
+def _is_plain_int(value: Any) -> bool:
+    """Return True for integers, excluding bool's int subclass."""
+    return isinstance(value, int) and not isinstance(value, bool)
+
+
+def _repeat_completed_count(repeat_state: Any) -> int:
+    """Extract a valid completed count from an existing repeat state."""
+    if isinstance(repeat_state, dict):
+        completed = repeat_state.get("completed", 0)
+        if _is_plain_int(completed) and completed >= 0:
+            return completed
+    return 0
+
+
+def _normalize_repeat_state(
+    repeat: Any,
+    existing_repeat: Optional[Any] = None,
+) -> Dict[str, Optional[int]]:
+    """Normalize API/scalar repeat values to the persisted scheduler shape."""
+    completed = _repeat_completed_count(existing_repeat)
+    if isinstance(repeat, dict):
+        times = repeat.get("times")
+        completed = repeat.get("completed", completed)
+        if not _is_plain_int(completed) or completed < 0:
+            raise ValueError("Repeat completed count must be a non-negative integer")
+    else:
+        times = repeat
+
+    if times is None:
+        normalized_times = None
+    elif _is_plain_int(times) and times > 0:
+        normalized_times = times
+    elif isinstance(repeat, dict) and _is_plain_int(times) and times <= 0:
+        normalized_times = None
+    else:
+        raise ValueError("Repeat must be a positive integer or null")
+
+    return {"times": normalized_times, "completed": completed}
+
+
+def _repair_repeat_state(job: Dict[str, Any]) -> bool:
+    """Repair a persisted legacy repeat value in place; return whether it changed."""
+    if "repeat" not in job:
+        return False
+    raw_repeat = job.get("repeat")
+    try:
+        normalized = _normalize_repeat_state(raw_repeat, raw_repeat)
+    except ValueError as exc:
+        schedule = job.get("schedule")
+        schedule_kind = schedule.get("kind") if isinstance(schedule, dict) else None
+        logger.warning(
+            "Job '%s' (%s) had invalid repeat state %r; repairing to safe default: %s",
+            job.get("name", job.get("id", "unknown")),
+            job.get("id", "unknown"),
+            raw_repeat,
+            exc,
+        )
+        normalized = {
+            "times": 1 if schedule_kind == "once" else None,
+            "completed": 0,
+        }
+    if raw_repeat == normalized:
+        return False
+    job["repeat"] = normalized
+    return True
+
+
 def _coerce_job_text(value: Any, fallback: str = "") -> str:
     """Coerce legacy/hand-edited nullable cron fields to strings for readers."""
     if value is None:
@@ -1309,6 +1376,11 @@ def update_job(job_id: str, updates: Dict[str, Any]) -> Optional[Dict[str, Any]]
                 else:
                     updates["workdir"] = _normalize_workdir(_wd)
 
+            if "repeat" in updates:
+                updates["repeat"] = _normalize_repeat_state(
+                    updates["repeat"], job.get("repeat")
+                )
+
             previous_inference_axes = _normalized_inference_axes(job)
             updated = _apply_skill_fields({**job, **updates})
             schedule_changed = "schedule" in updates
@@ -1494,6 +1566,8 @@ def mark_job_run(job_id: str, success: bool, error: Optional[str] = None,
                 # is claimable again. No-op if the job never carried a claim.
                 if job.get("run_claim") is not None:
                     job["run_claim"] = None
+
+                _repair_repeat_state(job)
 
                 # Increment completed count.  Finite one-shot jobs are
                 # pre-claimed by claim_dispatch() BEFORE the side effect runs
@@ -1582,13 +1656,20 @@ def claim_dispatch(job_id: str) -> bool:
         for i, job in enumerate(jobs):
             if job["id"] != job_id:
                 continue
-            if job.get("schedule", {}).get("kind") != "once":
+            repeat_repaired = _repair_repeat_state(job)
+            schedule = job.get("schedule")
+            schedule_kind = schedule.get("kind") if isinstance(schedule, dict) else None
+            if schedule_kind != "once":
+                if repeat_repaired:
+                    save_jobs(jobs)
                 return True  # recurring jobs use advance_next_run(), not dispatch claims
             repeat = job.get("repeat")
             if not repeat:
                 return True  # no repeat limit — always dispatch
             times = repeat.get("times")
             if times is None or times <= 0:
+                if repeat_repaired:
+                    save_jobs(jobs)
                 return True  # infinite — always dispatch
             completed = repeat.get("completed", 0)
             if completed >= times:
@@ -1795,6 +1876,10 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
             rj["id"] = rj.pop("job_id", None) or uuid.uuid4().hex[:12]
             needs_save = True
 
+    for rj in raw_jobs:
+        if _repair_repeat_state(rj):
+            needs_save = True
+
     jobs = [_apply_skill_fields(j) for j in copy.deepcopy(raw_jobs)]
     due = []
 

gateway/platforms/api_server.py

@@ -3705,6 +3705,12 @@ async def _handle_update_job(self, request: "web.Request") -> "web.Response":
                 return web.json_response(
                     {"error": f"Prompt must be ≤ {self._MAX_PROMPT_LENGTH} characters"}, status=400,
                 )
+            if "repeat" in sanitized:
+                repeat = sanitized["repeat"]
+                if repeat is not None and (
+                    not isinstance(repeat, int) or isinstance(repeat, bool) or repeat < 1
+                ):
+                    return web.json_response({"error": "Repeat must be a positive integer or null"}, status=400)
             if sanitized.get("prompt") and _scan_cron_prompt is not None:
                 scan_error = _scan_cron_prompt(sanitized["prompt"])
                 if scan_error:

tests/cron/test_jobs.py

@@ -257,6 +257,7 @@ def test_list_jobs(self, tmp_cron_dir):
         jobs = list_jobs()
         assert len(jobs) == 2
 
+
     def test_list_jobs_normalizes_partial_legacy_records(self, tmp_cron_dir):
         save_jobs([
             {
@@ -414,6 +415,47 @@ def test_update_nonexistent_returns_none(self, tmp_cron_dir):
         result = update_job("nonexistent_id", {"name": "X"})
         assert result is None
 
+    def test_update_repeat_scalar_preserves_repeat_shape_and_completed_count(self, tmp_cron_dir):
+        job = create_job(prompt="Repeat me", schedule="every 1h")
+        mark_job_run(job["id"], success=True)
+        assert get_job(job["id"])["repeat"]["completed"] == 1
+
+        updated = update_job(job["id"], {"repeat": 3})
+
+        assert updated["repeat"] == {"times": 3, "completed": 1}
+        mark_job_run(job["id"], success=True)
+        assert get_job(job["id"])["repeat"] == {"times": 3, "completed": 2}
+
+    def test_update_repeat_none_preserves_repeat_shape_and_clears_limit(self, tmp_cron_dir):
+        job = create_job(prompt="Repeat me", schedule="every 1h", repeat=2)
+        mark_job_run(job["id"], success=True)
+        assert get_job(job["id"])["repeat"]["completed"] == 1
+
+        updated = update_job(job["id"], {"repeat": None})
+
+        assert updated["repeat"] == {"times": None, "completed": 1}
+        mark_job_run(job["id"], success=True)
+        assert get_job(job["id"]) is not None
+        assert get_job(job["id"])["repeat"] == {"times": None, "completed": 2}
+
+    def test_mark_job_run_normalizes_legacy_scalar_repeat(self, tmp_cron_dir):
+        job = create_job(prompt="Legacy repeat", schedule="every 1h")
+        jobs = load_jobs()
+        jobs[0]["repeat"] = 3
+        save_jobs(jobs)
+
+        mark_job_run(job["id"], success=True)
+
+        updated = get_job(job["id"])
+        assert updated["repeat"] == {"times": 3, "completed": 1}
+
+    @pytest.mark.parametrize("bad_repeat", [0, -1, "3", 2.5, True])
+    def test_update_repeat_rejects_invalid_values(self, tmp_cron_dir, bad_repeat):
+        job = create_job(prompt="Bad repeat", schedule="every 1h")
+
+        with pytest.raises(ValueError, match="Repeat"):
+            update_job(job["id"], {"repeat": bad_repeat})
+
     def test_update_rejects_id_change(self, tmp_cron_dir):
         """Job IDs are filesystem path components — must be immutable."""
         job = create_job(prompt="Original", schedule="every 1h")
@@ -1802,6 +1844,31 @@ def test_claim_increments_and_persists(self, tmp_cron_dir):
         # Persisted BEFORE any side effect — survives a crash.
         assert load_jobs()[0]["repeat"]["completed"] == 1
 
+    def test_claim_normalizes_legacy_scalar_repeat(self, tmp_cron_dir):
+        job = self._oneshot()
+        job["repeat"] = 3
+        save_jobs([job])
+
+        assert claim_dispatch("os1") is True
+        assert load_jobs()[0]["repeat"] == {"times": 3, "completed": 1}
+
+    def test_claim_repairs_malformed_oneshot_repeat_as_finite(self, tmp_cron_dir):
+        job = self._oneshot()
+        job["repeat"] = "bad"
+        save_jobs([job])
+
+        assert claim_dispatch("os1") is True
+        assert load_jobs()[0]["repeat"] == {"times": 1, "completed": 1}
+
+    def test_claim_repairs_repeat_when_schedule_is_malformed(self, tmp_cron_dir):
+        job = self._oneshot()
+        job["schedule"] = None
+        job["repeat"] = "bad"
+        save_jobs([job])
+
+        assert claim_dispatch("os1") is True
+        assert load_jobs()[0]["repeat"] == {"times": None, "completed": 0}
+
     def test_already_dispatched_oneshot_is_removed(self, tmp_cron_dir):
         # A prior tick claimed (completed==times) then died before mark_job_run
         # could remove the job.  The next claim must refuse AND clean up.
@@ -1859,6 +1926,19 @@ def test_mark_job_run_still_increments_recurring(self, tmp_cron_dir):
         mark_job_run("rec", success=True)
         assert load_jobs()[0]["repeat"]["completed"] == 2
 
+    def test_get_due_jobs_normalizes_legacy_scalar_repeat(self, tmp_cron_dir):
+        past = (datetime.now(timezone.utc) - timedelta(seconds=5)).isoformat()
+        job = self._oneshot()
+        job["schedule"]["run_at"] = past
+        job["next_run_at"] = past
+        job["repeat"] = 3
+        save_jobs([job])
+
+        due = get_due_jobs()
+
+        assert [item["id"] for item in due] == ["os1"]
+        assert load_jobs()[0]["repeat"] == {"times": 3, "completed": 0}
+
     def test_get_due_jobs_removes_stale_maxed_oneshot(self, tmp_cron_dir):
         # A claimed one-shot whose tick died leaves completed>=times with
         # last_run_at still unset, so the recovery helper re-arms it as due.
@@ -1893,6 +1973,7 @@ def test_bad_schedule_does_not_crash_or_block_sibling_jobs(self, tmp_cron_dir):
             "name": "bad",
             "enabled": True,
             "schedule": None,  # poison: not a dict
+            "repeat": "bad",
             "next_run_at": future,  # not due
         }
         good = {

tests/gateway/test_api_server_jobs.py

@@ -383,6 +383,77 @@ async def test_update_job_no_valid_fields(self, adapter):
                 data = await resp.json()
                 assert "No valid fields" in data["error"]
 
+    @pytest.mark.asyncio
+    async def test_update_repeat_persists_shape_before_oneshot_claim(
+        self, adapter, tmp_path
+    ):
+        """PATCH scalar repeat survives storage and the pre-run one-shot claim."""
+        from cron.jobs import claim_dispatch, create_job, load_jobs, use_cron_store
+
+        app = _create_app(adapter)
+        with use_cron_store(tmp_path):
+            job = create_job(
+                prompt="one-shot",
+                schedule="1h",
+                repeat=1,
+                deliver="local",
+            )
+            async with TestClient(TestServer(app)) as cli:
+                with patch(
+                    f"{_MOD}._CRON_AVAILABLE", True
+                ), patch(
+                    f"{_MOD}._notify_cron_provider_jobs_changed"
+                ):
+                    resp = await cli.patch(
+                        f"/api/jobs/{job['id']}",
+                        json={"repeat": 3},
+                    )
+                    assert resp.status == 200
+
+            assert load_jobs()[0]["repeat"] == {"times": 3, "completed": 0}
+            assert claim_dispatch(job["id"]) is True
+            assert load_jobs()[0]["repeat"] == {"times": 3, "completed": 1}
+
+    @pytest.mark.asyncio
+    async def test_update_job_accepts_repeat_null_to_clear_limit(self, adapter):
+        """PATCH /api/jobs/{id} accepts repeat=null to clear the run limit."""
+        app = _create_app(adapter)
+        updated_job = {**SAMPLE_JOB, "repeat": {"times": None, "completed": 0}}
+        mock_update = MagicMock(return_value=updated_job)
+        async with TestClient(TestServer(app)) as cli:
+            with patch(
+                f"{_MOD}._CRON_AVAILABLE", True
+            ), patch(
+                f"{_MOD}._cron_update", mock_update
+            ):
+                resp = await cli.patch(
+                    f"/api/jobs/{VALID_JOB_ID}",
+                    json={"repeat": None},
+                )
+                assert resp.status == 200
+                mock_update.assert_called_once_with(VALID_JOB_ID, {"repeat": None})
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("bad_repeat", [0, -1, "3", 2.5, True])
+    async def test_update_job_rejects_invalid_repeat(self, adapter, bad_repeat):
+        """PATCH /api/jobs/{id} rejects repeat values that corrupt scheduler state."""
+        app = _create_app(adapter)
+        mock_update = MagicMock(return_value=SAMPLE_JOB)
+        async with TestClient(TestServer(app)) as cli:
+            with patch(
+                f"{_MOD}._CRON_AVAILABLE", True
+            ), patch(
+                f"{_MOD}._cron_update", mock_update
+            ):
+                resp = await cli.patch(
+                    f"/api/jobs/{VALID_JOB_ID}",
+                    json={"repeat": bad_repeat},
+                )
+                assert resp.status == 400
+                data = await resp.json()
+                assert "repeat" in data["error"].lower()
+                mock_update.assert_not_called()
+
 
 # ---------------------------------------------------------------------------
 # 13. test_delete_job