fix(cron): surface initial scheduler registration failures

cron/scheduler.py

@@ -4254,6 +4254,42 @@ def _notify_provider_jobs_changed() -> None:
         logger.debug("on_jobs_changed notify failed: %s", e)
 
 
+class CronSchedulerRegistrationError(RuntimeError):
+    """A job was persisted but its first external trigger was not registered."""
+
+    def __init__(self, job: dict, cause: Exception) -> None:
+        self.job = job
+        self.cause = cause
+        super().__init__(
+            f"Cron job '{job['id']}' was saved, but its first scheduler "
+            f"registration failed ({type(cause).__name__}). Do not create a "
+            "duplicate."
+        )
+
+    def to_dict(self) -> dict:
+        """Return the public partial-failure contract without provider details."""
+        return {
+            "error": str(self),
+            "job_id": self.job["id"],
+            "job_saved": True,
+            "scheduler_registered": False,
+            "retry_create": False,
+        }
+
+
+def create_job_with_scheduler_registration(*args, **kwargs) -> dict:
+    """Persist one job and register its first trigger with the active provider."""
+    from cron.jobs import create_job
+    from cron.scheduler_provider import resolve_cron_scheduler
+
+    job = create_job(*args, **kwargs)
+    try:
+        resolve_cron_scheduler().register_job(job)
+    except Exception as exc:
+        raise CronSchedulerRegistrationError(job, exc) from exc
+    return job
+
+
 def tick(
     verbose: bool = True,
     adapters=None,

cron/scheduler_provider.py

@@ -82,6 +82,16 @@ class CronScheduler(ABC):
         Built-in: no-op (it re-reads jobs.json on every tick)."""
         return None
 
+    def register_job(self, job: dict[str, Any]) -> None:
+        """Register the first external trigger for one newly persisted job.
+
+        The built-in provider reads the local store on every tick, so its
+        default is a no-op. External providers override this when creating a
+        job requires a remote registration before callers can honestly report
+        that the job is scheduled.
+        """
+        return None
+
     def recover_interrupted(self) -> int:
         """Run profile-local attempt recovery for every provider lifecycle."""
         from cron.executions import recover_interrupted_executions

cron/suggestions.py

@@ -232,13 +232,22 @@ def accept_suggestion(ref: str, *, origin: Optional[Dict[str, Any]] = None) -> O
     if not s or s.get("status") != _STATUS_PENDING:
         return None
 
-    from cron.jobs import create_job
+    from cron.scheduler import (
+        CronSchedulerRegistrationError,
+        create_job_with_scheduler_registration,
+    )
 
     spec = dict(s.get("job_spec") or {})
     if origin is not None and "origin" not in spec:
         spec["origin"] = origin
 
-    job = create_job(**spec)
+    try:
+        job = create_job_with_scheduler_registration(**spec)
+    except CronSchedulerRegistrationError:
+        # The job is already durable. Resolve the suggestion so retrying the
+        # same acceptance cannot create another local copy.
+        _set_status(s["id"], _STATUS_ACCEPTED)
+        raise
     _set_status(s["id"], _STATUS_ACCEPTED)
     return job
 

gateway/platforms/api_server.py

@@ -1284,13 +1284,16 @@ try:
     from cron.jobs import (
         list_jobs as _cron_list,
         get_job as _cron_get,
-        create_job as _cron_create,
         update_job as _cron_update,
         remove_job as _cron_remove,
         pause_job as _cron_pause,
         resume_job as _cron_resume,
         trigger_job as _cron_trigger,
     )
+    from cron.scheduler import (
+        CronSchedulerRegistrationError as _CronSchedulerRegistrationError,
+        create_job_with_scheduler_registration as _cron_create,
+    )
     _CRON_AVAILABLE = True
 except ImportError:
     _cron_list = None
@@ -1302,6 +1305,9 @@ except ImportError:
     _cron_resume = None
     _cron_trigger = None
 
+    class _CronSchedulerRegistrationError(RuntimeError):
+        pass
+
 
 def _notify_cron_provider_jobs_changed() -> None:
     """Tell the active cron scheduler provider the job set changed after a REST
@@ -5573,8 +5579,9 @@ class APIServerAdapter(BasePlatformAdapter):
                 kwargs["repeat"] = repeat
 
             job = _cron_create(**kwargs)
-            _notify_cron_provider_jobs_changed()
             return web.json_response({"job": job})
+        except _CronSchedulerRegistrationError as e:
+            return web.json_response(e.to_dict(), status=424)
         except Exception as e:
             return web.json_response({"error": _redact_api_error_text(e)}, status=500)
 

hermes_cli/blueprint_cmd.py

@@ -303,9 +303,14 @@ def handle_blueprint_command(
         )
 
     try:
-        from cron.jobs import create_job
+        from cron.scheduler import (
+            CronSchedulerRegistrationError,
+            create_job_with_scheduler_registration,
+        )
 
-        job = create_job(**spec)
+        job = create_job_with_scheduler_registration(**spec)
+    except CronSchedulerRegistrationError as e:
+        return BlueprintCommandResult(str(e))
     except Exception as e:
         logger.debug("blueprint create_job failed: %s", e)
         return BlueprintCommandResult(f"Failed to create the job: {e}")

hermes_cli/suggestions_cmd.py

@@ -97,7 +97,12 @@ def handle_suggestions_command(
     if sub in ("accept", "add", "schedule"):
         if not rest:
             return "Usage: /suggestions accept <number|id>"
-        job = store.accept_suggestion(rest, origin=origin)
+        from cron.scheduler import CronSchedulerRegistrationError
+
+        try:
+            job = store.accept_suggestion(rest, origin=origin)
+        except CronSchedulerRegistrationError as e:
+            return str(e)
         if job is None:
             return f"No pending suggestion matches '{rest}'. Run /suggestions to list them."
         sched = job.get("schedule_display") or (job.get("job_spec", {}) or {}).get("schedule", "")

hermes_cli/web_routers/cron.py

@@ -239,5 +239,12 @@ async def instantiate_blueprint(body: AutomationBlueprintInstantiate, profile: s
     except HTTPException:
         raise
     except Exception as e:
+        from cron.scheduler import CronSchedulerRegistrationError
+
+        if isinstance(e, CronSchedulerRegistrationError):
+            raise HTTPException(
+                status_code=424,
+                detail=e.to_dict(),
+            ) from e
         _log.exception("POST /api/cron/blueprints/instantiate failed")
         raise HTTPException(status_code=400, detail=str(e))

hermes_cli/web_server.py

@@ -11743,7 +11743,12 @@ def _call_cron_for_profile(target_profile: Optional[str], func_name: str, *args,
     token = set_hermes_home_override(str(home))
     try:
         with cron_jobs.use_cron_store(home):
-            result = getattr(cron_jobs, func_name)(*args, **kwargs)
+            if func_name == "create_job":
+                from cron.scheduler import create_job_with_scheduler_registration
+
+                result = create_job_with_scheduler_registration(*args, **kwargs)
+            else:
+                result = getattr(cron_jobs, func_name)(*args, **kwargs)
     finally:
         reset_hermes_home_override(token)
 
@@ -11905,6 +11910,13 @@ def _create_cron_job_sync(body: CronJobCreate, profile: Optional[str] = None):
     except HTTPException:
         raise
     except Exception as e:
+        from cron.scheduler import CronSchedulerRegistrationError
+
+        if isinstance(e, CronSchedulerRegistrationError):
+            raise HTTPException(
+                status_code=424,
+                detail=e.to_dict(),
+            ) from e
         _log.exception("POST /api/cron/jobs failed")
         raise HTTPException(status_code=400, detail=str(e))
 

plugins/cron_providers/chronos/__init__.py

@@ -127,6 +127,15 @@ class ChronosCronScheduler(CronScheduler):
         except Exception as e:
             logger.debug("Chronos on_jobs_changed reconcile failed: %s", e)
 
+    def register_job(self, job: Dict[str, Any]) -> None:
+        """Arm the first one-shot for a newly persisted job.
+
+        Unlike full reconciliation, this operation is allowed to raise so the
+        creation surface can report that the local job exists but its external
+        trigger was not registered.
+        """
+        self._arm_one_shot(job)
+
     # -- arming -----------------------------------------------------------
 
     def _arm_one_shot(self, job: Dict[str, Any]) -> None:

tests/cron/test_jobs_changed_notify.py

@@ -47,18 +47,121 @@ def test_builtin_notify_is_harmless(monkeypatch):
     sched._notify_provider_jobs_changed()
 
 
-def test_tool_create_notifies_provider(temp_home, monkeypatch):
-    """Creating a job via the cronjob tool path invokes on_jobs_changed."""
+def test_create_registers_first_trigger_with_active_provider(temp_home, monkeypatch):
+    """A successful create is not reported until the provider sees the job."""
+    import cron.scheduler_provider as sp
     import cron.scheduler as sched
-    calls = []
-    monkeypatch.setattr(sched, "_notify_provider_jobs_changed",
-                        lambda: calls.append("changed"))
+
+    registered = []
+
+    class Spy(sp.CronScheduler):
+        @property
+        def name(self):
+            return "spy"
+
+        def start(self, stop_event, **kw):
+            pass
+
+        def register_job(self, job):
+            registered.append(job)
+
+    monkeypatch.setattr(sp, "resolve_cron_scheduler", lambda: Spy())
+
+    job = sched.create_job_with_scheduler_registration(
+        prompt="echo hi",
+        schedule="every 5m",
+        name="w",
+    )
+
+    assert registered == [job]
+
+
+def test_create_failure_preserves_job_and_hides_provider_details(temp_home, monkeypatch):
+    """Registration failure is explicit without losing the durable local job."""
+    import cron.jobs as jobs
+    import cron.scheduler_provider as sp
+    import cron.scheduler as sched
+
+    class FailingProvider(sp.CronScheduler):
+        @property
+        def name(self):
+            return "failing"
+
+        def start(self, stop_event, **kw):
+            pass
+
+        def register_job(self, job):
+            raise RuntimeError("private callback URL and token")
+
+    monkeypatch.setattr(sp, "resolve_cron_scheduler", lambda: FailingProvider())
+
+    with pytest.raises(sched.CronSchedulerRegistrationError) as exc_info:
+        sched.create_job_with_scheduler_registration(
+            prompt="echo hi",
+            schedule="every 5m",
+            name="w",
+        )
+
+    error = exc_info.value
+    assert jobs.get_job(error.job["id"]) == error.job
+    assert "private callback URL and token" not in str(error)
+    assert "Do not create a duplicate" in str(error)
+
+
+def test_tool_create_registers_provider_before_reporting_success(temp_home, monkeypatch):
+    """The model-tool success response includes a provider-registered job."""
+    import cron.scheduler_provider as sp
+
+    registered = []
+
+    class RecordingProvider(sp.CronScheduler):
+        @property
+        def name(self):
+            return "recording"
+
+        def start(self, stop_event, **kw):
+            pass
+
+        def register_job(self, job):
+            registered.append(job)
+
+    monkeypatch.setattr(sp, "resolve_cron_scheduler", lambda: RecordingProvider())
+
+    from tools.cronjob_tools import cronjob
+    import json
+
+    out = json.loads(
+        cronjob(action="create", prompt="echo hi", schedule="every 5m", name="w")
+    )
+
+    assert out["success"] is True
+    assert [job["id"] for job in registered] == [out["job_id"]]
+
+
+def test_tool_create_reports_partial_registration_failure(temp_home, monkeypatch):
+    """The model tool must not claim a remotely unregistered job succeeded."""
+    import cron.scheduler_provider as sp
+
+    class FailingProvider(sp.CronScheduler):
+        @property
+        def name(self):
+            return "failing"
+
+        def start(self, stop_event, **kw):
+            pass
+
+        def register_job(self, job):
+            raise RuntimeError("private callback URL and token")
+
+    monkeypatch.setattr(sp, "resolve_cron_scheduler", lambda: FailingProvider())
 
     from tools.cronjob_tools import cronjob
     import json
 
     out = json.loads(cronjob(action="create", prompt="echo hi", schedule="every 5m", name="w"))
-    assert out["success"] is True
-    assert calls == ["changed"]
-
-
+    assert out["success"] is False
+    assert out["job_saved"] is True
+    assert out["scheduler_registered"] is False
+    assert out["retry_create"] is False
+    assert out["job_id"]
+    assert "private callback URL and token" not in out["error"]

tests/cron/test_suggestions.py

@@ -104,6 +104,24 @@ class TestStore:
         # And accepting again is a no-op (not pending anymore).
         assert store.accept_suggestion("acc") is None
 
+    def test_registration_failure_marks_suggestion_accepted(self, store):
+        """Retrying an acceptance must not create a duplicate durable job."""
+        from cron.scheduler import CronSchedulerRegistrationError
+
+        rec = _add(store, key="registration-failed", title="My Job")
+        job = {"id": "job123", "name": "My Job"}
+        failure = CronSchedulerRegistrationError(job, RuntimeError("private detail"))
+
+        with patch(
+            "cron.scheduler.create_job_with_scheduler_registration",
+            side_effect=failure,
+        ):
+            with pytest.raises(CronSchedulerRegistrationError):
+                store.accept_suggestion(rec["id"])
+
+        assert store.list_pending() == []
+        assert store.accept_suggestion(rec["id"]) is None
+
     def test_get_by_id_and_index_and_title(self, store):
         rec = _add(store, key="byref", title="Findable")
         assert store.get_suggestion(rec["id"])["id"] == rec["id"]

tests/gateway/test_api_server_jobs.py

@@ -140,6 +140,35 @@ class TestCreateJob:
                 assert call_kwargs["origin"]["user_agent"] == "cron-client"
 
 
+    @pytest.mark.asyncio
+    async def test_create_job_reports_saved_but_unregistered(self, adapter):
+        """A failed external registration is a structured partial failure."""
+        from cron.scheduler import CronSchedulerRegistrationError
+
+        app = _create_app(adapter)
+        failure = CronSchedulerRegistrationError(
+            SAMPLE_JOB,
+            RuntimeError("private callback URL and token"),
+        )
+        async with TestClient(TestServer(app)) as cli:
+            with patch(f"{_MOD}._CRON_AVAILABLE", True), patch(
+                f"{_MOD}._cron_create", side_effect=failure
+            ):
+                resp = await cli.post("/api/jobs", json={
+                    "name": "test-job",
+                    "schedule": "*/5 * * * *",
+                    "prompt": "do something",
+                })
+
+                assert resp.status == 424
+                data = await resp.json()
+                assert data["job_id"] == SAMPLE_JOB["id"]
+                assert data["job_saved"] is True
+                assert data["scheduler_registered"] is False
+                assert data["retry_create"] is False
+                assert "private callback URL and token" not in data["error"]
+
+
     @pytest.mark.asyncio
     async def test_create_job_prompt_too_long(self, adapter):
         """POST /api/jobs with prompt > 5000 chars returns 400."""
@@ -469,4 +498,3 @@ class TestCronPromptScanParity:
                 assert "Blocked" in data["error"] or "threat" in data["error"].lower()
                 mock_create.assert_not_called()
 
-

tests/hermes_cli/test_web_server_cron_profiles.py

@@ -82,6 +82,83 @@ def test_fire_cron_job_scopes_store_and_runtime_home_together(
         reset_hermes_home_override(outer_token)
 
 
+def test_create_registers_scheduler_inside_target_profile(
+    isolated_profiles,
+    monkeypatch,
+):
+    """Dashboard create must resolve and register under the selected profile."""
+    from cron import jobs as cron_jobs
+    from hermes_cli import web_server
+    from hermes_constants import get_hermes_home
+
+    worker_home = isolated_profiles["worker_alpha"]
+    captured = {}
+
+    class RecordingProvider:
+        def register_job(self, job):
+            captured["job"] = job
+            captured["runtime_home"] = get_hermes_home()
+            captured["jobs_file"] = cron_jobs._current_cron_store().jobs_file
+
+    monkeypatch.setattr(
+        "cron.scheduler_provider.resolve_cron_scheduler",
+        lambda: RecordingProvider(),
+    )
+
+    job = web_server._call_cron_for_profile(
+        "worker_alpha",
+        "create_job",
+        prompt="managed by named profile",
+        schedule="every 1h",
+        name="named-profile-job",
+    )
+
+    assert captured["job"]["id"] == job["id"]
+    assert captured["runtime_home"] == worker_home
+    assert captured["jobs_file"] == worker_home / "cron" / "jobs.json"
+    assert job["profile"] == "worker_alpha"
+
+
+def test_dashboard_create_reports_saved_but_unregistered(
+    isolated_profiles,
+    monkeypatch,
+):
+    """Dashboard callers can distinguish persistence from remote registration."""
+    from cron.scheduler import CronSchedulerRegistrationError
+    from hermes_cli import web_server
+
+    job = {"id": "saved-job", "name": "saved job"}
+    failure = CronSchedulerRegistrationError(
+        job,
+        RuntimeError("private callback URL and token"),
+    )
+
+    def fail_create(*args, **kwargs):
+        raise failure
+
+    monkeypatch.setattr(web_server, "_call_cron_for_profile", fail_create)
+
+    with pytest.raises(HTTPException) as exc_info:
+        web_server._create_cron_job_sync(
+            web_server.CronJobCreate(
+                prompt="managed by named profile",
+                schedule="every 1h",
+                name="named-profile-job",
+            ),
+            profile="worker_alpha",
+        )
+
+    assert exc_info.value.status_code == 424
+    assert exc_info.value.detail == {
+        "error": str(failure),
+        "job_id": "saved-job",
+        "job_saved": True,
+        "scheduler_registered": False,
+        "retry_create": False,
+    }
+    assert "private callback URL and token" not in str(exc_info.value.detail)
+
+
 def test_profile_call_cannot_retarget_ticker_store_mid_write(
     isolated_profiles,
     monkeypatch,
@@ -242,7 +319,3 @@ async def test_dashboard_cron_rejects_missing_context_from(isolated_profiles):
 
 
 
-
-
-
-

tests/plugins/test_chronos_cron.py

@@ -75,6 +75,29 @@ def test_arm_one_shot_sends_provision(chronos):
     assert p["agent_callback_url"] == "https://agent.example/"
 
 
+def test_register_job_arms_only_the_created_job(chronos):
+    prov, fake = chronos
+    job = {"id": "created", "next_run_at": "2026-06-18T12:00:00+00:00"}
+
+    prov.register_job(job)
+
+    assert [p["job_id"] for p in fake.provisions] == ["created"]
+
+
+def test_register_job_propagates_provision_failure(chronos):
+    prov, fake = chronos
+
+    def fail_provision(**kwargs):
+        raise RuntimeError("provision rejected")
+
+    fake.provision = fail_provision
+
+    with pytest.raises(RuntimeError, match="provision rejected"):
+        prov.register_job(
+            {"id": "created", "next_run_at": "2026-06-18T12:00:00+00:00"}
+        )
+
+
 # -- reconcile ----------------------------------------------------------------
 
 def test_reconcile_arms_all_enabled(temp_home, chronos, monkeypatch):
@@ -105,4 +128,3 @@ def test_fire_due_rearms_next_oneshot(chronos, monkeypatch):
     assert [p["job_id"] for p in fake.provisions] == ["j1"]
     assert fake.provisions[0]["fire_at"] == "2026-06-18T12:05:00+00:00"
 
-

tools/blueprints.py

@@ -206,12 +206,12 @@ def create_blueprint_job(
     optional ``prompt`` becomes the task instruction. Delivery, model, and
     toolsets carry through. Returns the created job dict.
     """
-    from cron.jobs import create_job
+    from cron.scheduler import create_job_with_scheduler_registration
 
     job_spec = blueprint_to_job_spec(spec, name=name)
     if origin is not None:
         job_spec["origin"] = origin
-    return create_job(**job_spec)
+    return create_job_with_scheduler_registration(**job_spec)
 
 
 def register_blueprint_suggestion(spec: BlueprintSpec) -> Optional[Dict[str, Any]]:

tools/cronjob_tools.py

@@ -40,7 +40,6 @@ sys.path.insert(0, str(Path(__file__).parent.parent))
 from cron.jobs import (
     AmbiguousJobReference,
     claim_job_for_fire,
-    create_job,
     get_job,
     list_jobs,
     mark_job_run,
@@ -1099,25 +1098,38 @@ def cronjob(
                             success=False,
                         )
 
-            job = create_job(
-                prompt=prompt or "",
-                schedule=schedule,
-                name=name,
-                repeat=repeat,
-                deliver=_normalize_deliver_param(deliver),
-                origin=_origin_from_env(),
-                skills=canonical_skills,
-                model=_normalize_optional_job_value(model),
-                provider=_normalize_optional_job_value(provider),
-                base_url=_normalize_optional_job_value(base_url, strip_trailing_slash=True),
-                script=_normalize_optional_job_value(script),
-                context_from=context_from,
-                enabled_toolsets=enabled_toolsets or None,
-                workdir=_normalize_optional_job_value(workdir),
-                no_agent=_no_agent,
-                attach_to_session=attach_to_session,
+            from cron.scheduler import (
+                CronSchedulerRegistrationError,
+                create_job_with_scheduler_registration,
             )
-            _notify_provider_jobs_changed_safe()
+
+            try:
+                job = create_job_with_scheduler_registration(
+                    prompt=prompt or "",
+                    schedule=schedule,
+                    name=name,
+                    repeat=repeat,
+                    deliver=_normalize_deliver_param(deliver),
+                    origin=_origin_from_env(),
+                    skills=canonical_skills,
+                    model=_normalize_optional_job_value(model),
+                    provider=_normalize_optional_job_value(provider),
+                    base_url=_normalize_optional_job_value(base_url, strip_trailing_slash=True),
+                    script=_normalize_optional_job_value(script),
+                    context_from=context_from,
+                    enabled_toolsets=enabled_toolsets or None,
+                    workdir=_normalize_optional_job_value(workdir),
+                    no_agent=_no_agent,
+                    attach_to_session=attach_to_session,
+                )
+            except CronSchedulerRegistrationError as exc:
+                return json.dumps(
+                    {
+                        "success": False,
+                        **exc.to_dict(),
+                    },
+                    indent=2,
+                )
             _create_message = f"Cron job '{job['name']}' created."
             _local_notice = _local_delivery_notice(job, _normalize_deliver_param(deliver))
             if _local_notice: