feat(cron): per-job max_turns override + record budget stops as PARTIAL

cron/jobs.py

@@ -537,6 +537,7 @@ def create_job(
     enabled_toolsets: Optional[List[str]] = None,
     workdir: Optional[str] = None,
     no_agent: bool = False,
+    max_turns: Optional[int] = None,
 ) -> Dict[str, Any]:
     """
     Create a new cron job.
@@ -581,6 +582,10 @@ def create_job(
                 and deliver its stdout directly. Empty stdout = silent (no
                 delivery). Requires ``script`` to be set. Ideal for classic
                 watchdogs and periodic alerts that don't need LLM reasoning.
+        max_turns: Optional per-job agent iteration budget. When a positive int,
+                   overrides the global config default for this job's runs.
+                   Values that are not a positive non-boolean int are ignored
+                   (field is omitted from the record).
 
     Returns:
         The created job dict
@@ -615,6 +620,14 @@ def create_job(
     normalized_toolsets = normalized_toolsets or None
     normalized_workdir = _normalize_workdir(workdir)
     normalized_no_agent = bool(no_agent)
+    # max_turns: only store when it is a positive non-boolean int; omit otherwise
+    # so existing records without the field are not affected.
+    _max_turns_valid = (
+        isinstance(max_turns, int)
+        and not isinstance(max_turns, bool)
+        and max_turns > 0
+    )
+    normalized_max_turns = max_turns if _max_turns_valid else None
 
     # no_agent jobs are meaningless without a script — the script IS the job.
     # Surface this as a clear ValueError at create time so bad configs never
@@ -669,6 +682,8 @@ def create_job(
         "enabled_toolsets": normalized_toolsets,
         "workdir": normalized_workdir,
     }
+    if normalized_max_turns is not None:
+        job["max_turns"] = normalized_max_turns
 
     jobs = load_jobs()
     jobs.append(job)

cron/scheduler.py

@@ -1635,8 +1635,22 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
                     logger.warning("Job '%s': failed to parse prefill messages file '%s': %s", job_id, pfpath, e)
                     prefill_messages = None
 
-        # Max iterations
-        max_iterations = _cfg.get("agent", {}).get("max_turns") or _cfg.get("max_turns") or 90
+        # Max iterations (a job-record "max_turns" overrides the global config —
+        # long-form worker jobs need a bigger budget than interactive turns).
+        # `bool` is a subclass of `int`, so exclude it explicitly: `max_turns:
+        # true` must fall back to the default, not silently set a 1-turn budget.
+        _job_max_turns = job.get("max_turns")
+        _job_max_turns_valid = (
+            isinstance(_job_max_turns, int)
+            and not isinstance(_job_max_turns, bool)
+            and _job_max_turns > 0
+        )
+        max_iterations = (
+            (_job_max_turns if _job_max_turns_valid else None)
+            or _cfg.get("agent", {}).get("max_turns")
+            or _cfg.get("max_turns")
+            or 90
+        )
 
         # Provider routing
         pr = _cfg.get("provider_routing", {})
@@ -1858,12 +1872,41 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
         # job's `last_status` set to "ok". Raise so the except handler below
         # builds the proper failure tuple. (issue #17855)
         if result.get("failed") is True or result.get("completed") is False:
-            _err_text = (
-                result.get("error")
-                or (result.get("final_response") or "").strip()
-                or "agent reported failure"
-            )
-            raise RuntimeError(_err_text)
+            _exit_reason = str(result.get("turn_exit_reason") or "")
+            _final_text = (result.get("final_response") or "").strip()
+            # Iteration-budget stops that still produced a substantive report
+            # are partial work, not failures. Wrapping them in RuntimeError
+            # buried multi-KB handoff briefs in `last_error` and made them
+            # indistinguishable from a turn-1 crash in the dashboard.
+            # The >=300-char floor rejects the short turn-completion explainer
+            # boilerplate that finalize_turn injects when the post-budget
+            # summary call itself fails — that's a real failure (no work
+            # record), not a partial.
+            # ("budget_exhausted" is normally rewritten to
+            # "max_iterations_reached(...)" by finalize_turn before it reaches
+            # here; it is matched too so a directly-surfaced budget stop is
+            # still treated as partial.)
+            if (
+                result.get("failed") is not True
+                and _exit_reason.startswith(("max_iterations_reached", "budget_exhausted"))
+                and len(_final_text) >= 300
+            ):
+                logger.warning(
+                    "Job '%s' hit iteration budget (%s) — recording as partial success",
+                    job_name, _exit_reason,
+                )
+                result["final_response"] = (
+                    f"⚠️ PARTIAL — iteration budget hit ({_exit_reason}). "
+                    "Work may be incomplete; treat the report below as a handoff brief.\n\n"
+                    + _final_text
+                )
+            else:
+                _err_text = (
+                    result.get("error")
+                    or _final_text
+                    or "agent reported failure"
+                )
+                raise RuntimeError(_err_text)
 
         final_response = result.get("final_response", "") or ""
         # Strip leaked placeholder text that upstream may inject on empty completions.

tests/cron/test_jobs.py

@@ -849,6 +849,44 @@ def test_broken_interval_without_next_run_is_recovered(self, tmp_cron_dir, monke
         assert recovered_dt > now
 
 
+class TestMaxTurns:
+    def test_max_turns_stored_when_positive_int(self, tmp_cron_dir):
+        job = create_job(prompt="worker", schedule="every 1h", max_turns=150)
+        assert job["max_turns"] == 150
+
+    def test_max_turns_persisted(self, tmp_cron_dir):
+        job = create_job(prompt="worker", schedule="every 1h", max_turns=150)
+        fetched = get_job(job["id"])
+        assert fetched["max_turns"] == 150
+
+    def test_max_turns_omitted_when_not_provided(self, tmp_cron_dir):
+        job = create_job(prompt="worker", schedule="every 1h")
+        assert "max_turns" not in job
+
+    def test_max_turns_omitted_for_zero(self, tmp_cron_dir):
+        job = create_job(prompt="worker", schedule="every 1h", max_turns=0)
+        assert "max_turns" not in job
+
+    def test_max_turns_omitted_for_negative(self, tmp_cron_dir):
+        job = create_job(prompt="worker", schedule="every 1h", max_turns=-5)
+        assert "max_turns" not in job
+
+    def test_max_turns_omitted_for_bool_true(self, tmp_cron_dir):
+        # bool is a subclass of int — True == 1 but must be excluded
+        job = create_job(prompt="worker", schedule="every 1h", max_turns=True)
+        assert "max_turns" not in job
+
+    def test_max_turns_omitted_for_bool_false(self, tmp_cron_dir):
+        job = create_job(prompt="worker", schedule="every 1h", max_turns=False)
+        assert "max_turns" not in job
+
+    def test_max_turns_updated_via_update_job(self, tmp_cron_dir):
+        job = create_job(prompt="worker", schedule="every 1h")
+        update_job(job["id"], {"max_turns": 200})
+        fetched = get_job(job["id"])
+        assert fetched["max_turns"] == 200
+
+
 class TestEnabledToolsets:
     def test_enabled_toolsets_stored(self, tmp_cron_dir):
         job = create_job(prompt="monitor", schedule="every 1h", enabled_toolsets=["web", "terminal"])