diff --git a/cron/jobs.py b/cron/jobs.py index 90c318742e6fc..2e554556044cc 100644 --- a/cron/jobs.py +++ b/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 = [] diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 59deed4e7f67b..337e7f8fb21bb 100644 --- a/gateway/platforms/api_server.py +++ b/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: diff --git a/tests/cron/test_jobs.py b/tests/cron/test_jobs.py index d7d16e551341d..59b1a7a9f2324 100644 --- a/tests/cron/test_jobs.py +++ b/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 = { diff --git a/tests/gateway/test_api_server_jobs.py b/tests/gateway/test_api_server_jobs.py index 082ab6cf16710..83c98753a0e20 100644 --- a/tests/gateway/test_api_server_jobs.py +++ b/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