feat(cron): create paused jobs without a scheduling race

Persist paused state, timestamp, reason and no first trigger in the original
locked creation write. Forward the same boolean contract across CLI, tool,
gateway API and dashboard API, validating at the store boundary. Preserve
explicit operator force-run behavior and normal enabled creation.

The live CLI probe also caught the command shim dropping failure return codes;
forward them so invalid creation reports exit 1 rather than success.

Credit earlier atomic-creation work in #78935 and #94952 and the focused
implementation in #104578. The broader manifest staging layer is not imported.

Co-authored-by: Konstantin Khlopkov <konstantin.khlopkov93@gmail.com>
Co-authored-by: Chloé DuPont <321112755+misschloedupont@users.noreply.github.com>

contributors/emails/321112755+misschloedupont@users.noreply.github.com

@@ -0,0 +1,2 @@
+misschloedupont
+# Earlier paused creation #94952

cron/jobs.py

@@ -1703,6 +1703,8 @@ def create_job(
     monitor_url: Optional[str] = None,
     reasoning_effort: Optional[str] = None,
     failure_deliver: Optional[str] = None,
+    paused: bool = False,
+    paused_reason: Optional[str] = None,
 ) -> Dict[str, Any]:
     """Create a new cron job and return the stored record.
 
@@ -1712,6 +1714,12 @@ def create_job(
     injected. workdir: absolute cwd for tools/scripts. monitor_script/monitor_url: cheap monitor
     source run FIRST each tick; unchanged output suppresses the agent run (mutually exclusive,
     incompatible with ``no_agent``). reasoning_effort: per-job pin; capability NOT validated."""
+    if not isinstance(paused, bool):
+        raise ValueError("paused must be a boolean.")
+    if paused_reason is not None and not isinstance(paused_reason, str):
+        raise ValueError("paused_reason must be a string.")
+    if paused_reason is not None and not paused:
+        raise ValueError("paused_reason requires paused=True.")
     parsed_schedule = parse_schedule(schedule)
     # Normalize repeat: treat 0 or negative values as None (infinite). String forms
     # ('forever'/'once'/numeric) coerce via normalize_repeat_value — the shared chokepoint with update paths
@@ -1769,12 +1777,12 @@ def create_job(
         "schedule": parsed_schedule,
         "schedule_display": parsed_schedule.get("display", schedule),
         "repeat": {"times": repeat, "completed": 0},  # times None = forever
-        "enabled": True,
-        "state": "scheduled",
-        "paused_at": None,
-        "paused_reason": None,
+        "enabled": not paused,
+        "state": "paused" if paused else "scheduled",
+        "paused_at": now if paused else None,
+        "paused_reason": ((paused_reason or "").strip() or "Created paused; awaiting operator approval.") if paused else None,
         "created_at": now,
-        "next_run_at": next_run_at,
+        "next_run_at": None if paused else next_run_at,
         "last_run_at": None,
         "last_status": None,
         "last_error": None,

cron/scheduler.py

@@ -3415,6 +3415,8 @@ def create_job_with_scheduler_registration(**kwargs) -> dict:
     from cron.scheduler_provider import resolve_cron_scheduler
 
     job = create_job(**kwargs)
+    if not job.get("enabled", True):
+        return job
     try:
         resolve_cron_scheduler().register_job(job)
     except Exception as exc:

gateway/platforms/api_server.py

@@ -3347,6 +3347,9 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter):
                 "prompt": prompt, "schedule": schedule, "name": name,
                 "deliver": body.get("deliver", "local"),
                 "origin": self._cron_origin_from_request(request)}
+            for key in ("paused", "paused_reason"):
+                if key in body:
+                    kwargs[key] = body[key]
             if skills:
                 kwargs["skills"] = skills
             if repeat is not None:
@@ -3354,6 +3357,8 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter):
             return web.json_response({"job": _cron_create(**kwargs)})
         except _CronSchedulerRegistrationError as e:
             return web.json_response(e.to_dict(), status=424)
+        except ValueError as e:
+            return web.json_response({"error": str(e)}, status=400)
         except Exception as e:
             return self._cron_error_response(e)
 

hermes_cli/cron.py

@@ -571,7 +571,10 @@ def cron_create(args):
         action="create", schedule=args.schedule, prompt=args.prompt,
         skill=getattr(args, "skill", None),
         skills=_normalize_skills(getattr(args, "skill", None), getattr(args, "skills", None)),
-        no_agent=getattr(args, "no_agent", False) or None, **_job_api_kwargs(args))
+        no_agent=getattr(args, "no_agent", False) or None,
+        **({"paused": args.paused, "paused_reason": getattr(args, "paused_reason", None)}
+           if getattr(args, "paused", False) or getattr(args, "paused_reason", None) is not None else {}),
+        **_job_api_kwargs(args))
     if not result.get("success"):
         print(color(f"Failed to create job: {result.get('error', 'unknown error')}", Colors.RED))
         return 1
@@ -580,7 +583,10 @@ def cron_create(args):
     if result.get("skills"):
         print(f"  Skills: {', '.join(result['skills'])}")
     _print_job_details(result.get("job", {}))
-    print(f"  Next run: {result['next_run_at']}")
+    if not result.get("job", {}).get("enabled", True):
+        print("  Created PAUSED — resume to schedule, or explicitly run now.")
+    else:
+        print(f"  Next run: {result['next_run_at']}")
     _warn_if_gateway_not_running()
     return 0
 

hermes_cli/main.py

@@ -1776,7 +1776,7 @@ cmd_login = _forward_command("cmd_login", "hermes_cli.auth", "login_command", do
 cmd_logout = _forward_command("cmd_logout", "hermes_cli.auth", "logout_command", doc='Clear provider authentication.')
 cmd_auth = _forward_command("cmd_auth", "hermes_cli.auth_commands", "auth_command", doc='Manage pooled credentials.')
 cmd_status = _forward_command("cmd_status", "hermes_cli.status", "show_status", doc='Show status of all components.')
-cmd_cron = _forward_command("cmd_cron", "hermes_cli.cron", "cron_command", doc='Cron job management.')
+cmd_cron = _forward_command("cmd_cron", "hermes_cli.cron", "cron_command", forward_return=True, doc='Cron job management.')
 cmd_webhook = _forward_command("cmd_webhook", "hermes_cli.webhook", "webhook_command", doc='Webhook subscription management.')
 cmd_kanban = _forward_command("cmd_kanban", "hermes_cli.kanban", "kanban_command", forward_return=True, doc='Multi-profile collaboration board.')
 cmd_project = _forward_command("cmd_project", "hermes_cli.projects_cmd", "projects_command", forward_return=True, doc='Manage projects (named, multi-folder workspaces).')

hermes_cli/subcommands/cron.py

@@ -81,6 +81,10 @@ def build_cron_parser(subparsers, *, cmd_cron: Callable) -> None:
             "reported and continue where the last run left off (scouts, "
             "monitors, incremental digests). First run is unchanged.")
 
+    _flag(cron_create, "--paused", default=False,
+        help="Create disabled in one write; resume to schedule, or explicitly run now.")
+    cron_create.add_argument("--paused-reason", help="Auditable reason; requires --paused.")
+
     cron_edit = cron_subparsers.add_parser("edit", help="Edit an existing scheduled job")
     cron_edit.add_argument("job_id", help="Job ID to edit")
     cron_edit.add_argument("--schedule", help="New schedule")

hermes_cli/web_models.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import math
 from typing import Any, Dict, List, Literal, Optional
 
-from pydantic import BaseModel, SecretStr, field_validator
+from pydantic import BaseModel, SecretStr, StrictBool, field_validator
 
 
 class ConfigUpdate(BaseModel):
@@ -278,6 +278,8 @@ class SessionPrune(BaseModel):
     dry_run: bool = False
 
 class CronJobCreate(BaseModel):
+    paused: StrictBool = False
+    paused_reason: Optional[str] = None
     prompt: str = ""
     schedule: str
     name: str = ""

hermes_cli/web_server_cron.py

@@ -258,7 +258,9 @@ def _create_cron_job_sync(body: CronJobCreate, profile: Optional[str] = None):
             context_from=context_from,
             enabled_toolsets=_cron_string_list(body.enabled_toolsets),
             workdir=_cron_optional_text(body.workdir),
-            no_agent=no_agent)
+            no_agent=no_agent,
+            **{key: getattr(body, key) for key in ("paused", "paused_reason")
+               if key in body.model_fields_set})
     except HTTPException:
         raise
     except Exception as e:

tests/cron/test_atomic_paused_creation.py

@@ -0,0 +1,48 @@
+"""Creation is either armed normally or durably paused before registration."""
+import json
+
+from cron import jobs
+from tools import cronjob_tools
+
+
+def test_paused_creation_is_inert_until_operator_action(tmp_path, monkeypatch, make_cron_provider):
+    registered, writes = [], []
+    provider = make_cron_provider(register_job=lambda job: registered.append(job["id"]))
+    monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", lambda: provider)
+    save = jobs.save_jobs
+
+    def observe(rows):
+        writes.append(json.loads(json.dumps(rows)))
+        return save(rows)
+
+    monkeypatch.setattr(jobs, "save_jobs", observe)
+    with jobs.use_cron_store(tmp_path / "cron"):
+        def create(**kwargs):
+            return json.loads(cronjob_tools.registry.dispatch("cronjob_manage", {
+                "action": "create", "schedule": "every 1h", "prompt": "canary", **kwargs}))
+
+        active = create()
+        assert active["success"] and active["job"]["enabled"]
+        paused = create(paused=True)
+        assert paused["success"], paused
+        job_id = paused["job_id"]
+        first = next(row for rows in writes for row in rows if row["id"] == job_id)
+        assert first["enabled"] is False and first["state"] == "paused"
+        assert first["paused_at"] and first["paused_reason"] and first["next_run_at"] is None
+        assert registered == [active["job_id"]]
+        assert job_id not in {row["id"] for row in jobs.get_due_jobs()}
+        assert jobs.claim_job_for_fire(job_id) is False
+        resumed = jobs.resume_job(job_id)
+        assert resumed["enabled"] and resumed["next_run_at"] and resumed["paused_reason"] is None
+        forced = create(paused=True, paused_reason="canary review")
+        assert jobs.claim_job_for_fire(forced["job_id"], force=True) is True
+
+
+def test_invalid_creation_is_rejected_without_writes(tmp_path):
+    with jobs.use_cron_store(tmp_path / "cron"):
+        for flags in ({"paused": "yes"}, {"paused": None}, {"paused": 1},
+                      {"paused_reason": "orphan"}, {"paused": True, "paused_reason": 123}):
+            result = json.loads(cronjob_tools.registry.dispatch("cronjob_manage", {
+                "action": "create", "schedule": "every 1h", "prompt": "invalid", **flags}))
+            assert result["success"] is False, (flags, result)
+            assert jobs.load_jobs() == []

tools/cronjob_tools.py

@@ -572,11 +572,15 @@ def _action_create(a: Dict[str, Any]) -> str:
             monitor_url=_normalize_optional_job_value(a["monitor_url"]),
             # CLI-only lane: absent from CRONJOB_SCHEMA and the model dispatch (models don't pick models).
             reasoning_effort=a["reasoning_effort"],
-            failure_deliver=_resolve_cron_context_deliver(_normalize_deliver_param(a["failure_deliver"])))
+            failure_deliver=_resolve_cron_context_deliver(_normalize_deliver_param(a["failure_deliver"])),
+            **({"paused": a["paused"], "paused_reason": a["paused_reason"]}
+               if a["paused"] is not False or a["paused_reason"] is not None else {}))
     except CronSchedulerRegistrationError as exc:
         _partial = exc.to_dict()
         return tool_error(_partial.pop("error"), success=False, **_partial)
-    _create_message = " ".join(filter(None, (f"Cron job '{job['name']}' created.", _local_delivery_notice(job, deliver))))
+    _create_message = " ".join(filter(None, (f"Cron job '{job['name']}' created.",
+        "Created PAUSED — resume to schedule, or explicitly run now." if not job.get("enabled", True) else None,
+        _local_delivery_notice(job, deliver))))
     # The builtin ticker lives in the gateway process: with no gateway running the job is stored
     # but never fires — tell the model (the CLI already warns).
     _result = {
@@ -872,7 +876,9 @@ def cronjob(
     reasoning_effort: Optional[str] = None,
     failure_deliver: Optional[Union[str, List[str]]] = None,
     task_id: str = None,
-    session_id: Optional[str] = None) -> str:
+    session_id: Optional[str] = None,
+    paused: bool = False,
+    paused_reason: Optional[str] = None) -> str:
     """Unified cron job management tool."""
     a = dict(locals())
     del a["task_id"]  # unused but kept for handler signature compatibility
@@ -904,6 +910,8 @@ Jobs run in a fresh session with no current-chat context, so prompts must be sel
     "parameters": {
         "type": "object",
         "properties": {
+            "paused": {"type": "boolean", "description": "Create only: persist disabled atomically. Resume to schedule; explicit run remains available. Default false."},
+            "paused_reason": {"type": "string", "description": "Create only: auditable reason; requires paused=true."},
             "action": {
                 "type": "string",
                 "description": "One of: create, list, update, pause, resume, remove, run. When action=create, the 'schedule' and 'prompt' fields are REQUIRED."
@@ -1000,7 +1008,8 @@ def check_cronjob_requirements() -> bool:
 # different model. Programmatic callers of cronjob() itself retain the parameters.
 _HANDLER_FORWARDED_ARGS = (
     "job_id", "prompt", "schedule", "name", "repeat", "deliver", "failure_deliver", "skill", "skills", "reason",
-    "script", "context_from", "continuity", "enabled_toolsets", "workdir", "no_agent", "attach_to_session")
+    "script", "context_from", "continuity", "enabled_toolsets", "workdir", "no_agent", "attach_to_session",
+    "paused_reason")
 
 
 def _cronjob_handler(args, **kw):
@@ -1014,6 +1023,7 @@ def _cronjob_handler(args, **kw):
         monitor_url=_mon_url,
         task_id=kw.get("task_id"),
         session_id=kw.get("session_id"),
+        paused=args.get("paused", False),
         **{key: args.get(key) for key in _HANDLER_FORWARDED_ARGS},
     )
 

website/docs/user-guide/features/cron.md

@@ -264,6 +264,28 @@ What they do:
 
 **Name-based lookup.** All four mutating verbs (`pause`, `resume`, `run`, `remove`, `edit`) plus the agent's `cronjob` tool now accept a job **name** (case-insensitive) in place of the hex ID. The agent and CLI both prefer an exact ID match if one exists; ambiguous name matches (multiple jobs sharing the same name) are refused with the full list of candidate IDs so you can pick one explicitly. Names are not unique, so this guard is load-bearing — it prevents silently mutating the wrong job when two share a name.
 
+### Creating a job paused (safe canary)
+
+Create a canary without a create-then-pause scheduling race:
+
+```bash
+hermes cron create "every 1h" "Post the digest" --paused --paused-reason "Awaiting review"
+hermes cron resume <job_id>
+```
+
+`--paused` stores `enabled: false`, `state: paused`, `next_run_at: null`, a pause
+timestamp and an auditable reason in the first locked write, without registering a
+trigger. Omit the reason to store "Created paused; awaiting operator approval."
+Omit `--paused` to retain normal enabled creation. `--paused-reason` requires
+`--paused`; invalid values are rejected before persistence.
+
+The same `paused` boolean and optional `paused_reason` string are accepted by
+`cron.jobs.create_job`, the cron management tool's `create` action, the gateway
+`POST /api/jobs`, and the dashboard `POST /api/cron/jobs`. Resume schedules the next
+future run. Pausing prevents automatic fires, not operator overrides: existing
+explicit **Run now** / force-run behavior remains available and can resume and run
+the job. It is not a security boundary against an operator who can run jobs.
+
 ## Agent-managed scheduling (cron jobs that manage cron jobs)
 
 By default, agents launched *by* the scheduler cannot use the `cronjob` tool —