feat(cron): per-job max_turns override (CLI-only, mirrors reasoning_effort)

cron/jobs.py

@@ -2101,6 +2101,31 @@ def _normalize_reasoning_effort(value: Any) -> Optional[str]:
     return text
 
 
+def _normalize_job_max_turns(value: Any) -> Optional[int]:
+    """Validate a per-job agent turn budget at the storage choke point.
+
+    Returns None for unset (None, empty string, or 0 — the CLI's "clear"
+    spelling), the positive int for valid input, and raises ValueError for
+    anything else so a typo never persists and then surfaces hours later as an
+    unexplained mid-run stop.
+    """
+    if value is None:
+        return None
+    if isinstance(value, str) and not value.strip():
+        return None
+    try:
+        turns = int(value)
+    except (TypeError, ValueError):
+        raise ValueError(
+            f"Invalid max_turns {value!r}: pass a positive integer (0 or empty clears the override)."
+        )
+    if turns == 0:
+        return None
+    if turns < 0:
+        raise ValueError(f"Invalid max_turns {turns}: pass a positive integer (0 clears the override).")
+    return turns
+
+
 def _compute_provider_model_snapshots(
     *,
     provider: Any,
@@ -2204,6 +2229,7 @@ def create_job(
     monitor_script: Optional[str] = None,
     monitor_url: Optional[str] = None,
     reasoning_effort: Optional[str] = None,
+    max_turns: Optional[int] = None,
 ) -> Dict[str, Any]:
     """
     Create a new cron job.
@@ -2261,6 +2287,12 @@ def create_job(
         monitor_url: Optional http(s) URL used as the monitor source instead
                 of a script — fetched with a bounded GET each tick. Same
                 hash-suppression semantics as ``monitor_script``.
+        max_turns: Optional per-job agent turn budget, overriding the global
+                ``agent.max_turns`` for this job only. Long multi-phase jobs
+                (browse → plan → generate → QA) legitimately need more turns
+                than the shared default, and raising the global value for
+                them would spend the same budget on every other job. Absent
+                or 0 means follow config. Ignored when ``no_agent=True``.
         reasoning_effort: Optional per-job reasoning effort pin. One of the
                 canonical Hermes levels (none|minimal|low|medium|high|xhigh|
                 max|ultra, case-insensitive). When set, it wins over BOTH the
@@ -2306,6 +2338,7 @@ def create_job(
     normalized_no_agent = bool(no_agent)
     normalized_attach = attach_to_session if isinstance(attach_to_session, bool) else None
     normalized_reasoning_effort = _normalize_reasoning_effort(reasoning_effort)
+    normalized_max_turns = _normalize_job_max_turns(max_turns)
     normalized_monitor_script = str(monitor_script).strip() if isinstance(monitor_script, str) else None
     normalized_monitor_script = normalized_monitor_script or None
     normalized_monitor_url = str(monitor_url).strip() if isinstance(monitor_url, str) else None
@@ -2422,6 +2455,9 @@ def create_job(
     # absent key = job follows config resolution (pre-feature behavior).
     if normalized_reasoning_effort is not None:
         job["reasoning_effort"] = normalized_reasoning_effort
+    # And for the per-job turn budget: absent key = global agent.max_turns.
+    if normalized_max_turns is not None:
+        job["max_turns"] = normalized_max_turns
 
     with _jobs_lock():
         jobs = load_jobs()
@@ -2536,6 +2572,10 @@ def update_job(job_id: str, updates: Dict[str, Any]) -> Optional[Dict[str, Any]]
                 updates["reasoning_effort"] = _normalize_reasoning_effort(
                     updates["reasoning_effort"]
                 )
+            # Same for the per-job turn budget: positive int, 0/empty clears,
+            # anything else raises before the merge.
+            if "max_turns" in updates:
+                updates["max_turns"] = _normalize_job_max_turns(updates["max_turns"])
 
             # Normalize repeat the same way create_job does. Callers pass
             # either the stored dict shape ({"times": N, "completed": M}) or

cron/scheduler.py

@@ -5418,6 +5418,33 @@ def _call():
         return _bounded
 
 
+def _resolve_job_turn_limit(job: dict, config_limit: int) -> int:
+    """Per-job ``max_turns`` wins over the global ``agent.max_turns``.
+
+    Long multi-phase jobs (browse → plan → generate → QA) legitimately need
+    more turns than the shared default, and raising the global value to suit
+    them would spend the same budget on every other job. The value is
+    validated at the store choke point (``cron/jobs.py::_normalize_job_max_turns``);
+    a garbage value in a hand-edited store warns and follows config instead
+    of killing the tick.
+    """
+    raw = job.get("max_turns")
+    if raw is None or (isinstance(raw, str) and not raw.strip()):
+        return config_limit
+    job_name = job.get("name") or job.get("id") or "cron job"
+    try:
+        turns = int(raw)
+    except (TypeError, ValueError):
+        turns = 0
+    if turns <= 0:
+        logger.warning(
+            "Job '%s': ignoring invalid max_turns %r; using %s", job_name, raw, config_limit
+        )
+        return config_limit
+    logger.info("Job '%s': per-job max_turns override -> %s", job_name, turns)
+    return turns
+
+
 def run_job(
     job: dict,
     *,
@@ -6025,6 +6052,7 @@ def run_job(
         if _mt is None:
             _mt = _cfg.get("max_turns")
         max_iterations = _resolve_turn_limit(_mt)
+        max_iterations = _resolve_job_turn_limit(job, max_iterations)
 
         # Provider routing
         pr = _cfg.get("provider_routing") or {}

hermes_cli/cron.py

@@ -545,6 +545,7 @@ def cron_create(args):
         monitor_url=getattr(args, "monitor_url", None),
         continuity=getattr(args, "continuity", None),
         reasoning_effort=getattr(args, "reasoning_effort", None),
+        max_turns=getattr(args, "max_turns", None),
     )
     if not result.get("success"):
         print(color(f"Failed to create job: {result.get('error', 'unknown error')}", Colors.RED))
@@ -555,6 +556,8 @@ def cron_create(args):
     if result.get("skills"):
         print(f"  Skills: {', '.join(result['skills'])}")
     job_data = result.get("job", {})
+    if job_data.get("max_turns"):
+        print(f"  Max turns: {job_data['max_turns']}")
     if job_data.get("script"):
         print(f"  Script: {job_data['script']}")
     if job_data.get("monitor_script"):
@@ -620,6 +623,7 @@ def cron_edit(args):
         monitor_url=getattr(args, "monitor_url", None),
         continuity=getattr(args, "continuity", None),
         reasoning_effort=getattr(args, "reasoning_effort", None),
+        max_turns=getattr(args, "max_turns", None),
     )
     if not result.get("success"):
         print(color(f"Failed to update job: {result.get('error', 'unknown error')}", Colors.RED))
@@ -633,6 +637,8 @@ def cron_edit(args):
         print(f"  Skills: {', '.join(updated['skills'])}")
     else:
         print("  Skills: none")
+    if updated.get("max_turns"):
+        print(f"  Max turns: {updated['max_turns']}")
     if updated.get("script"):
         print(f"  Script: {updated['script']}")
     if updated.get("monitor_script"):

hermes_cli/subcommands/cron.py

@@ -119,6 +119,16 @@ def build_cron_parser(subparsers, *, cmd_cron: Callable) -> None:
             "clamped by the provider at request time. Omit to follow config."
         ),
     )
+    cron_create.add_argument(
+        "--max-turns",
+        dest="max_turns",
+        type=int,
+        help=(
+            "Per-job agent turn budget, overriding agent.max_turns for this job "
+            "only. For long multi-phase jobs that need more turns than the "
+            "shared default. Omit to follow config."
+        ),
+    )
     cron_create.add_argument(
         "--continuity",
         dest="continuity",
@@ -254,6 +264,15 @@ def build_cron_parser(subparsers, *, cmd_cron: Callable) -> None:
             "the pin and follow config resolution."
         ),
     )
+    cron_edit.add_argument(
+        "--max-turns",
+        dest="max_turns",
+        type=int,
+        help=(
+            "Per-job agent turn budget, overriding agent.max_turns for this job "
+            "only. Pass 0 to clear the override and follow config."
+        ),
+    )
 
     # lifecycle actions
     cron_pause = cron_subparsers.add_parser("pause", help="Pause a scheduled job")

tests/cron/test_cron_job_max_turns.py

@@ -0,0 +1,106 @@
+"""Per-job max_turns override: store contract + scheduler precedence.
+
+A cron job may carry its own agent turn budget, independent of the global
+``agent.max_turns``. Long multi-phase jobs (browse → plan → generate → QA)
+legitimately need more turns than the shared default, and raising the global
+value to suit them would spend the same budget on every other job.
+
+Contract under test:
+
+- Job store (cron/jobs.py): validated at the storage choke point — a positive
+  int is stored, 0/empty clears, garbage raises and nothing persists. An
+  absent field keeps the record byte-identical to pre-feature behavior.
+- Scheduler resolution (cron/scheduler.py::_resolve_job_turn_limit): the
+  job's value wins over the config-resolved limit; an absent field yields
+  the config limit unchanged; a garbage value in a hand-edited store warns
+  and follows config instead of killing the tick.
+"""
+
+import pytest
+
+from cron.jobs import create_job, load_jobs, update_job
+
+
+@pytest.fixture()
+def tmp_cron_dir(tmp_path, monkeypatch):
+    """Isolate the cron store (same pattern as tests/cron/test_jobs.py)."""
+    monkeypatch.setattr("cron.jobs.CRON_DIR", tmp_path / "cron")
+    monkeypatch.setattr("cron.jobs.JOBS_FILE", tmp_path / "cron" / "jobs.json")
+    monkeypatch.setattr("cron.jobs.OUTPUT_DIR", tmp_path / "cron" / "output")
+    return tmp_path / "cron"
+
+
+def _create(**kw):
+    kw.setdefault("prompt", "say hi")
+    kw.setdefault("schedule", "every 1h")
+    return create_job(**kw)
+
+
+class TestJobStoreMaxTurns:
+    def test_absent_field_keeps_the_record_shape(self, tmp_cron_dir):
+        job = _create()
+        assert "max_turns" not in job
+        assert "max_turns" not in load_jobs()[0]
+
+    @pytest.mark.parametrize("value, expected", [(160, 160), ("160", 160), (1, 1)])
+    def test_positive_values_stored_as_int(self, tmp_cron_dir, value, expected):
+        job = _create(max_turns=value)
+        assert job["max_turns"] == expected
+        assert load_jobs()[0]["max_turns"] == expected
+
+    @pytest.mark.parametrize("empty", [None, "", "  ", 0, "0"])
+    def test_empty_or_zero_means_follow_config(self, tmp_cron_dir, empty):
+        job = _create(max_turns=empty)
+        assert "max_turns" not in job
+
+    @pytest.mark.parametrize("garbage", ["lots", "12.5", -1, "-3", [160]])
+    def test_garbage_rejected_nothing_persisted(self, tmp_cron_dir, garbage):
+        with pytest.raises(ValueError, match="max_turns"):
+            _create(max_turns=garbage)
+        assert load_jobs() == []
+
+    def test_update_sets_field(self, tmp_cron_dir):
+        job = _create()
+        updated = update_job(job["id"], {"max_turns": "200"})
+        assert updated["max_turns"] == 200
+        assert load_jobs()[0]["max_turns"] == 200
+
+    def test_update_zero_clears(self, tmp_cron_dir):
+        job = _create(max_turns=160)
+        updated = update_job(job["id"], {"max_turns": 0})
+        assert updated["max_turns"] is None
+        assert load_jobs()[0]["max_turns"] is None
+
+    def test_update_garbage_rejected_stored_value_untouched(self, tmp_cron_dir):
+        job = _create(max_turns=160)
+        with pytest.raises(ValueError, match="max_turns"):
+            update_job(job["id"], {"max_turns": "many"})
+        assert load_jobs()[0]["max_turns"] == 160
+
+
+class TestSchedulerTurnLimit:
+    def _resolve(self, job, config_limit=60):
+        from cron.scheduler import _resolve_job_turn_limit
+
+        return _resolve_job_turn_limit(job, config_limit)
+
+    def test_job_value_wins_over_config(self):
+        assert self._resolve({"name": "plan", "max_turns": 160}) == 160
+
+    def test_numeric_string_from_a_hand_edited_store_is_honored(self):
+        assert self._resolve({"name": "plan", "max_turns": "160"}) == 160
+
+    @pytest.mark.parametrize("absent", [{}, {"max_turns": None}, {"max_turns": ""}])
+    def test_absent_field_follows_config(self, absent):
+        assert self._resolve({"name": "plan", **absent}, config_limit=60) == 60
+
+    @pytest.mark.parametrize("garbage", ["many", 0, -5, "-5", 12.0 * 0])
+    def test_garbage_warns_and_follows_config(self, garbage, caplog):
+        with caplog.at_level("WARNING", logger="cron.scheduler"):
+            assert self._resolve({"name": "plan", "max_turns": garbage}, config_limit=60) == 60
+        assert "ignoring invalid max_turns" in caplog.text
+
+    def test_the_override_is_logged_so_operators_can_verify_it_applied(self, caplog):
+        with caplog.at_level("INFO", logger="cron.scheduler"):
+            self._resolve({"name": "plan", "max_turns": 160})
+        assert "per-job max_turns override -> 160" in caplog.text

tools/cronjob_tools.py

@@ -781,6 +781,8 @@ def _format_job(job: Dict[str, Any]) -> Dict[str, Any]:
         result["script"] = job["script"]
     if job.get("reasoning_effort"):
         result["reasoning_effort"] = job["reasoning_effort"]
+    if job.get("max_turns"):
+        result["max_turns"] = job["max_turns"]
     if job.get("monitor_script"):
         result["monitor_script"] = job["monitor_script"]
     if job.get("monitor_url"):
@@ -1482,6 +1484,7 @@ def cronjob(
     monitor_script: Optional[str] = None,
     monitor_url: Optional[str] = None,
     reasoning_effort: Optional[str] = None,
+    max_turns: Optional[int] = None,
     task_id: str = None,
     session_id: Optional[str] = None,
 ) -> str:
@@ -1598,6 +1601,9 @@ def cronjob(
                     # dispatch below: models do not make model-config
                     # decisions (standing policy).
                     reasoning_effort=reasoning_effort,
+                    # max_turns is the same CLI-only lane: a turn budget is a
+                    # spend decision, so the model never sets it.
+                    max_turns=max_turns,
                 )
             except CronSchedulerRegistrationError as exc:
                 _partial = exc.to_dict()
@@ -1814,6 +1820,10 @@ def cronjob(
                 # CLI-only lane (see create above): update_job validates
                 # against the canonical grammar; empty string clears the pin.
                 updates["reasoning_effort"] = reasoning_effort
+            if max_turns is not None:
+                # CLI-only lane like reasoning_effort: update_job validates;
+                # 0 clears the override.
+                updates["max_turns"] = max_turns
             # Re-validate the EFFECTIVE provider/base_url on EVERY update, not
             # only when this update supplies provider/base_url. A job persisted
             # before this guard (or written directly to the jobs store) may