diff --git a/cron/jobs.py b/cron/jobs.py
index 52d9367ff84e2..b366e455d4cab 100644
--- a/cron/jobs.py
+++ b/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)
diff --git a/cron/scheduler.py b/cron/scheduler.py
index 3590699661950..cafc28e21b03a 100644
--- a/cron/scheduler.py
+++ b/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.
diff --git a/tests/cron/test_jobs.py b/tests/cron/test_jobs.py
index d044f051ff14d..f1d033d5c8412 100644
--- a/tests/cron/test_jobs.py
+++ b/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"])
diff --git a/tests/cron/test_run_job_partial_budget.py b/tests/cron/test_run_job_partial_budget.py
new file mode 100644
index 0000000000000..aa5b35cd7bc6b
--- /dev/null
+++ b/tests/cron/test_run_job_partial_budget.py
@@ -0,0 +1,369 @@
+"""Tests for run_job partial-success handling and per-job max_turns override.
+
+Covers the two behaviours added in the PR under test:
+
+1. Per-job max_turns override: job["max_turns"] (positive int) takes precedence
+   over _cfg["agent"]["max_turns"] / _cfg["max_turns"] / 90 default.
+
+2. Partial-success path: when failed is NOT True AND turn_exit_reason starts
+   with "max_iterations_reached" or "budget_exhausted" AND the final_response
+   has >= 300 chars, run_job records PARTIAL success (prepends warning banner,
+   returns success=True, no RuntimeError) instead of raising.
+
+Mocking pattern mirrors TestRunJobSessionPersistence in test_scheduler.py:
+  - patch cron.scheduler._hermes_home   → tmp_path (avoids real FS)
+  - patch cron.scheduler._resolve_origin → None
+  - patch dotenv.load_dotenv             → no-op
+  - patch hermes_state.SessionDB        → MagicMock fake_db
+  - patch hermes_cli.runtime_provider.resolve_runtime_provider → minimal dict
+  - patch run_agent.AIAgent             → MagicMock whose run_conversation
+                                           returns a controlled result dict
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from cron.scheduler import run_job
+
+# ---------------------------------------------------------------------------
+# Shared helpers
+# ---------------------------------------------------------------------------
+
+_FAKE_RUNTIME = {
+    "api_key": "test-key",
+    "base_url": "https://example.invalid/v1",
+    "provider": "openrouter",
+    "api_mode": "chat_completions",
+}
+
+# A report long enough to satisfy the >= 300 char floor.
+_LONG_REPORT = (
+    "Detailed handoff brief. " * 20  # 480 chars
+)
+assert len(_LONG_REPORT.strip()) >= 300, "fixture is too short — adjust"
+
+# A report deliberately shorter than 300 chars.
+_SHORT_REPORT = "Brief summary. " * 10  # 150 chars
+assert len(_SHORT_REPORT.strip()) < 300, "fixture is too long — adjust"
+
+
+def _make_job(**extra):
+    """Minimal valid job dict, merging any extra keys."""
+    base = {
+        "id": "partial-budget-test",
+        "name": "partial budget test",
+        "prompt": "do work",
+    }
+    base.update(extra)
+    return base
+
+
+def _run_job_with_agent_result(tmp_path, agent_result, job_extra=None):
+    """Run run_job with a controlled agent result dict.
+
+    Returns (success, output, final_response, error) plus the
+    mock_agent_cls so callers can inspect call_args.
+    """
+    job = _make_job(**(job_extra or {}))
+    fake_db = MagicMock()
+
+    with patch("cron.scheduler._hermes_home", tmp_path), \
+         patch("cron.scheduler._resolve_origin", return_value=None), \
+         patch("dotenv.load_dotenv"), \
+         patch("hermes_state.SessionDB", return_value=fake_db), \
+         patch(
+             "hermes_cli.runtime_provider.resolve_runtime_provider",
+             return_value=_FAKE_RUNTIME,
+         ), \
+         patch("run_agent.AIAgent") as mock_agent_cls:
+        mock_agent = MagicMock()
+        mock_agent.run_conversation.return_value = agent_result
+        mock_agent_cls.return_value = mock_agent
+
+        result = run_job(job)
+
+    return result, mock_agent_cls
+
+
+# ---------------------------------------------------------------------------
+# Case (a) — partial success: budget_exhausted with long report
+# ---------------------------------------------------------------------------
+
+class TestPartialSuccessBudgetExhausted:
+    def test_budget_exhausted_long_report_succeeds(self, tmp_path):
+        """failed=False, completed=False, turn_exit_reason='budget_exhausted',
+        final_response >= 300 chars → success=True, no error, banner present."""
+        agent_result = {
+            "failed": False,
+            "completed": False,
+            "turn_exit_reason": "budget_exhausted",
+            "final_response": _LONG_REPORT,
+        }
+        (success, output, final_response, error), _ = _run_job_with_agent_result(
+            tmp_path, agent_result
+        )
+
+        assert success is True, "Expected partial success to be truthy"
+        assert not error, f"Expected error to be falsy, got: {error!r}"
+        assert "⚠️ PARTIAL" in final_response, (
+            "Expected PARTIAL banner in final_response"
+        )
+        assert "budget_exhausted" in final_response, (
+            "Expected exit reason in final_response banner"
+        )
+        # The original report text must be preserved below the banner.
+        assert _LONG_REPORT.strip() in final_response, (
+            "Original report text should be included after the banner"
+        )
+
+    def test_max_iterations_reached_long_report_succeeds(self, tmp_path):
+        """turn_exit_reason='max_iterations_reached(90/90)' also triggers partial."""
+        exit_reason = "max_iterations_reached(90/90)"
+        agent_result = {
+            "failed": False,
+            "completed": False,
+            "turn_exit_reason": exit_reason,
+            "final_response": _LONG_REPORT,
+        }
+        (success, output, final_response, error), _ = _run_job_with_agent_result(
+            tmp_path, agent_result
+        )
+
+        assert success is True
+        assert not error
+        assert "⚠️ PARTIAL" in final_response
+        assert exit_reason in final_response
+        assert _LONG_REPORT.strip() in final_response
+
+
+# ---------------------------------------------------------------------------
+# Case (b) — short report does NOT qualify for partial path
+# ---------------------------------------------------------------------------
+
+class TestShortReportStillFails:
+    def test_budget_exhausted_short_report_fails(self, tmp_path):
+        """Same turn_exit_reason but final_response < 300 chars → failure, not partial."""
+        agent_result = {
+            "failed": False,
+            "completed": False,
+            "turn_exit_reason": "budget_exhausted",
+            "final_response": _SHORT_REPORT,
+        }
+        (success, output, final_response, error), _ = _run_job_with_agent_result(
+            tmp_path, agent_result
+        )
+
+        assert not success, "Short-report partial should NOT succeed"
+        assert error, "Expected error to be set"
+
+    def test_max_iterations_reached_short_report_fails(self, tmp_path):
+        """max_iterations_reached with < 300 char report → failure."""
+        agent_result = {
+            "failed": False,
+            "completed": False,
+            "turn_exit_reason": "max_iterations_reached(5/5)",
+            "final_response": _SHORT_REPORT,
+        }
+        (success, output, final_response, error), _ = _run_job_with_agent_result(
+            tmp_path, agent_result
+        )
+
+        assert not success
+        assert error
+
+    def test_empty_response_with_budget_exit_fails(self, tmp_path):
+        """Empty final_response cannot be partial — must fail."""
+        agent_result = {
+            "failed": False,
+            "completed": False,
+            "turn_exit_reason": "budget_exhausted",
+            "final_response": "",
+        }
+        (success, output, final_response, error), _ = _run_job_with_agent_result(
+            tmp_path, agent_result
+        )
+
+        assert not success
+        assert error
+
+
+# ---------------------------------------------------------------------------
+# Case (c) — failed=True always propagates as failure
+# ---------------------------------------------------------------------------
+
+class TestGenuineFailureNotPartial:
+    def test_failed_true_with_long_response_still_fails(self, tmp_path):
+        """failed=True must NOT be treated as partial even with a long response."""
+        agent_result = {
+            "failed": True,
+            "completed": False,
+            "turn_exit_reason": "budget_exhausted",
+            "final_response": _LONG_REPORT,
+        }
+        (success, output, final_response, error), _ = _run_job_with_agent_result(
+            tmp_path, agent_result
+        )
+
+        assert not success, "failed=True should always report failure"
+        assert error, "Expected error to be set when failed=True"
+        # Banner must NOT appear — this is a genuine failure, not a partial.
+        assert "⚠️ PARTIAL" not in (final_response or ""), (
+            "PARTIAL banner must not appear for a genuine failure"
+        )
+
+    def test_failed_true_with_max_iterations_reason_still_fails(self, tmp_path):
+        """Even when turn_exit_reason looks like a budget stop, failed=True wins."""
+        agent_result = {
+            "failed": True,
+            "completed": False,
+            "turn_exit_reason": "max_iterations_reached(90/90)",
+            "final_response": _LONG_REPORT,
+        }
+        (success, output, final_response, error), _ = _run_job_with_agent_result(
+            tmp_path, agent_result
+        )
+
+        assert not success
+        assert error
+
+    def test_plain_api_failure_reports_error(self, tmp_path):
+        """A normal API-exhaustion failure dict is surfaced as failure."""
+        agent_result = {
+            "final_response": "API call failed after 3 retries: Request timed out.",
+            "failed": True,
+            "completed": False,
+            "error": "API call failed after 3 retries: Request timed out.",
+        }
+        (success, output, final_response, error), _ = _run_job_with_agent_result(
+            tmp_path, agent_result
+        )
+
+        assert not success
+        assert error
+        assert "API call failed" in error
+
+
+# ---------------------------------------------------------------------------
+# Case (d) — per-job max_turns override
+# ---------------------------------------------------------------------------
+
+class TestPerJobMaxTurnsOverride:
+    def test_job_max_turns_passed_to_agent_as_max_iterations(self, tmp_path):
+        """job['max_turns'] (positive int) must be forwarded to AIAgent as
+        max_iterations, overriding the global config default of 90.
+
+        The existing harness exposes AIAgent's constructor kwargs via
+        mock_agent_cls.call_args.kwargs — the same seam used by e.g.
+        test_run_job_passes_enabled_toolsets_to_agent.  We inspect
+        kwargs['max_iterations'] directly.
+        """
+        job_extra = {"max_turns": 150}
+        agent_result = {"final_response": "done"}
+
+        (success, *_), mock_agent_cls = _run_job_with_agent_result(
+            tmp_path, agent_result, job_extra=job_extra
+        )
+
+        assert success is True
+        kwargs = mock_agent_cls.call_args.kwargs
+        assert kwargs["max_iterations"] == 150, (
+            f"Expected max_iterations=150 (from job['max_turns']), "
+            f"got {kwargs['max_iterations']!r}"
+        )
+
+    def test_zero_max_turns_falls_back_to_global_default(self, tmp_path):
+        """job['max_turns']=0 is not a positive int → falls back to config/default."""
+        job_extra = {"max_turns": 0}
+        agent_result = {"final_response": "done"}
+
+        (success, *_), mock_agent_cls = _run_job_with_agent_result(
+            tmp_path, agent_result, job_extra=job_extra
+        )
+
+        assert success is True
+        kwargs = mock_agent_cls.call_args.kwargs
+        # With no config.yaml in tmp_path, the fallback is 90.
+        assert kwargs["max_iterations"] == 90, (
+            f"Expected fallback max_iterations=90, got {kwargs['max_iterations']!r}"
+        )
+
+    def test_negative_max_turns_falls_back_to_global_default(self, tmp_path):
+        """job['max_turns']=-1 is also not positive → falls back."""
+        job_extra = {"max_turns": -1}
+        agent_result = {"final_response": "done"}
+
+        (success, *_), mock_agent_cls = _run_job_with_agent_result(
+            tmp_path, agent_result, job_extra=job_extra
+        )
+
+        assert success is True
+        kwargs = mock_agent_cls.call_args.kwargs
+        assert kwargs["max_iterations"] == 90
+
+    def test_string_max_turns_falls_back_to_global_default(self, tmp_path):
+        """job['max_turns']='150' is not an int → falls back (type guard)."""
+        job_extra = {"max_turns": "150"}
+        agent_result = {"final_response": "done"}
+
+        (success, *_), mock_agent_cls = _run_job_with_agent_result(
+            tmp_path, agent_result, job_extra=job_extra
+        )
+
+        assert success is True
+        kwargs = mock_agent_cls.call_args.kwargs
+        assert kwargs["max_iterations"] == 90
+
+    def test_bool_max_turns_falls_back_to_global_default(self, tmp_path):
+        """job['max_turns']=True must NOT be treated as 1: bool is an int
+        subclass, so it's excluded explicitly and falls back to the default."""
+        job_extra = {"max_turns": True}
+        agent_result = {"final_response": "done"}
+
+        (success, *_), mock_agent_cls = _run_job_with_agent_result(
+            tmp_path, agent_result, job_extra=job_extra
+        )
+
+        assert success is True
+        kwargs = mock_agent_cls.call_args.kwargs
+        assert kwargs["max_iterations"] == 90, (
+            f"bool max_turns must fall back to 90, got {kwargs['max_iterations']!r}"
+        )
+
+    def test_max_turns_config_yaml_used_when_no_job_override(self, tmp_path):
+        """When no job['max_turns'], the config.yaml agent.max_turns is used."""
+        (tmp_path / "config.yaml").write_text(
+            "agent:\n  max_turns: 42\n",
+            encoding="utf-8",
+        )
+        agent_result = {"final_response": "done"}
+
+        (success, *_), mock_agent_cls = _run_job_with_agent_result(
+            tmp_path, agent_result
+        )
+
+        assert success is True
+        kwargs = mock_agent_cls.call_args.kwargs
+        assert kwargs["max_iterations"] == 42, (
+            f"Expected config.yaml max_turns=42 to be used, "
+            f"got {kwargs['max_iterations']!r}"
+        )
+
+    def test_job_max_turns_beats_config_yaml(self, tmp_path):
+        """job['max_turns'] wins over config.yaml agent.max_turns."""
+        (tmp_path / "config.yaml").write_text(
+            "agent:\n  max_turns: 42\n",
+            encoding="utf-8",
+        )
+        job_extra = {"max_turns": 200}
+        agent_result = {"final_response": "done"}
+
+        (success, *_), mock_agent_cls = _run_job_with_agent_result(
+            tmp_path, agent_result, job_extra=job_extra
+        )
+
+        assert success is True
+        kwargs = mock_agent_cls.call_args.kwargs
+        assert kwargs["max_iterations"] == 200, (
+            "job['max_turns'] must override config.yaml agent.max_turns"
+        )
diff --git a/tests/tools/test_cronjob_tools.py b/tests/tools/test_cronjob_tools.py
index 1ca877064a7a5..287e2243d4c6a 100644
--- a/tests/tools/test_cronjob_tools.py
+++ b/tests/tools/test_cronjob_tools.py
@@ -453,6 +453,62 @@ def test_update_normalizes_list_form_deliver(self):
         stored = get_job(created["job_id"])
         assert stored["deliver"] == "telegram"
 
+    def test_create_with_max_turns_persists_value(self):
+        """create with max_turns=150 stores max_turns=150 on the job record."""
+        from cron.jobs import get_job
+
+        result = json.loads(
+            cronjob(action="create", prompt="long job", schedule="every 1h", max_turns=150)
+        )
+        assert result["success"] is True
+        stored = get_job(result["job_id"])
+        assert stored["max_turns"] == 150
+
+    def test_create_without_max_turns_omits_field(self):
+        """create without max_turns does not add the key to the job record."""
+        from cron.jobs import get_job
+
+        result = json.loads(
+            cronjob(action="create", prompt="normal job", schedule="every 1h")
+        )
+        assert result["success"] is True
+        stored = get_job(result["job_id"])
+        assert "max_turns" not in stored
+
+    def test_create_max_turns_zero_ignored(self):
+        """create with max_turns=0 is not a positive int — field must be absent."""
+        from cron.jobs import get_job
+
+        result = json.loads(
+            cronjob(action="create", prompt="job", schedule="every 1h", max_turns=0)
+        )
+        assert result["success"] is True
+        stored = get_job(result["job_id"])
+        assert "max_turns" not in stored
+
+    def test_create_max_turns_negative_ignored(self):
+        """create with max_turns=-1 is not positive — field must be absent."""
+        from cron.jobs import get_job
+
+        result = json.loads(
+            cronjob(action="create", prompt="job", schedule="every 1h", max_turns=-1)
+        )
+        assert result["success"] is True
+        stored = get_job(result["job_id"])
+        assert "max_turns" not in stored
+
+    def test_update_max_turns_sets_value(self):
+        """update with max_turns=75 stores max_turns=75 on the job record."""
+        from cron.jobs import get_job
+
+        created = json.loads(cronjob(action="create", prompt="job", schedule="every 1h"))
+        job_id = created["job_id"]
+
+        updated = json.loads(cronjob(action="update", job_id=job_id, max_turns=75))
+        assert updated["success"] is True
+        stored = get_job(job_id)
+        assert stored["max_turns"] == 75
+
 
 # =========================================================================
 # Per-job model/provider override resolution
diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py
index 7ec31b806c46a..a67b0db15fabc 100644
--- a/tools/cronjob_tools.py
+++ b/tools/cronjob_tools.py
@@ -482,6 +482,7 @@ def cronjob(
     enabled_toolsets: Optional[List[str]] = None,
     workdir: Optional[str] = None,
     no_agent: Optional[bool] = None,
+    max_turns: Optional[int] = None,
     task_id: str = None,
 ) -> str:
     """Unified cron job management tool."""
@@ -548,6 +549,7 @@ def cronjob(
                 enabled_toolsets=enabled_toolsets or None,
                 workdir=_normalize_optional_job_value(workdir),
                 no_agent=_no_agent,
+                max_turns=max_turns,
             )
             return json.dumps(
                 {
@@ -701,6 +703,14 @@ def cronjob(
                 repeat_state = dict(job.get("repeat") or {})
                 repeat_state["times"] = normalized_repeat
                 updates["repeat"] = repeat_state
+            if max_turns is not None:
+                # Only accept positive non-boolean ints; clear (None) otherwise.
+                _mt_valid = (
+                    isinstance(max_turns, int)
+                    and not isinstance(max_turns, bool)
+                    and max_turns > 0
+                )
+                updates["max_turns"] = max_turns if _mt_valid else None
             if schedule is not None:
                 parsed_schedule = parse_schedule(schedule)
                 updates["schedule"] = parsed_schedule
@@ -834,6 +844,10 @@ def cronjob(
                 "type": "string",
                 "description": "Optional absolute path to run the job from. When set, AGENTS.md / CLAUDE.md / .cursorrules from that directory are injected into the system prompt, and the terminal/file/code_exec tools use it as their working directory — useful for running a job inside a specific project repo. Must be an absolute path that exists. When unset (default), preserves the original behaviour: no project context files, tools use the scheduler's cwd. On update, pass an empty string to clear. Jobs with workdir run sequentially (not parallel) to keep per-job directories isolated."
             },
+            "max_turns": {
+                "type": "integer",
+                "description": "Maximum agent turns (iteration budget) for this job's runs; overrides the global default. Useful for long-running worker jobs. Must be a positive integer. Invalid values (0, negative, or non-integer) are ignored and the global default is used instead. On update, set to a positive integer to override or omit to leave unchanged."
+            },
         },
         "required": ["action"]
     }