refactor(cron): polish registration partial-failure surfaces

Follow-up to the salvaged registration contract:
- share one _raise_if_cron_registration_error() helper for the two
  byte-identical dashboard 424 except-blocks (web_server + cron router,
  via the existing late() seam)
- add endpoint-level 424 coverage for /api/cron/blueprints/instantiate
  (previously only the sync worker was tested)
- give chat/CLI surfaces a human-facing user_message() (job name, no
  exception class name) and add a recovery hint (pause/resume or update
  re-registers via provider reconcile) to the model/REST message
- consolidate five inline provider test doubles into one ABC-subclassing
  make_cron_provider conftest factory; the web_server test double now
  subclasses CronScheduler so an ABC rename fails loudly
- narrow the wrapper facade to keyword-only (**kwargs) and route the
  tool's partial-failure return through tool_error()

cron/scheduler.py

@@ -4263,7 +4263,16 @@ class CronSchedulerRegistrationError(RuntimeError):
         super().__init__(
             f"Cron job '{job['id']}' was saved, but its first scheduler "
             f"registration failed ({type(cause).__name__}). Do not create a "
-            "duplicate."
+            "duplicate. Pause/resume or update the job to retry registration."
+        )
+
+    def user_message(self) -> str:
+        """Human-facing variant for chat/CLI surfaces (no exception class name)."""
+        label = self.job.get("name") or self.job["id"]
+        return (
+            f"Saved cron job '{label}', but couldn't register it with the "
+            "external scheduler yet. The job is kept — don't re-create it; "
+            "pause/resume or edit it (e.g. via /cron) to retry registration."
         )
 
     def to_dict(self) -> dict:
@@ -4277,12 +4286,12 @@ class CronSchedulerRegistrationError(RuntimeError):
         }
 
 
-def create_job_with_scheduler_registration(*args, **kwargs) -> dict:
+def create_job_with_scheduler_registration(**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)
+    job = create_job(**kwargs)
     try:
         resolve_cron_scheduler().register_job(job)
     except Exception as exc:

hermes_cli/blueprint_cmd.py

@@ -310,7 +310,7 @@ def handle_blueprint_command(
 
         job = create_job_with_scheduler_registration(**spec)
     except CronSchedulerRegistrationError as e:
-        return BlueprintCommandResult(str(e))
+        return BlueprintCommandResult(e.user_message())
     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

@@ -102,7 +102,7 @@ def handle_suggestions_command(
         try:
             job = store.accept_suggestion(rest, origin=origin)
         except CronSchedulerRegistrationError as e:
-            return str(e)
+            return e.user_message()
         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

@@ -44,6 +44,7 @@ _delete_cron_job_sync = late("_delete_cron_job_sync")
 _find_cron_job_profile = late("_find_cron_job_profile")
 _fire_cron_job_for_profile = late("_fire_cron_job_for_profile")
 _call_cron_for_profile = late("_call_cron_for_profile")
+_raise_if_cron_registration_error = late("_raise_if_cron_registration_error")
 load_config = late("load_config")
 cfg_get = late("cfg_get")
 
@@ -239,12 +240,6 @@ 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
+        _raise_if_cron_registration_error(e)
         _log.exception("POST /api/cron/blueprints/instantiate failed")
         raise HTTPException(status_code=400, detail=str(e))

hermes_cli/web_server.py

@@ -11797,6 +11797,19 @@ async def _run_cron_dashboard_io(func, *args, **kwargs):
     return result
 
 
+def _raise_if_cron_registration_error(e: Exception) -> None:
+    """Re-raise a cron partial-failure (job saved, external scheduler
+    registration failed) as HTTP 424 with the structured envelope.
+
+    Shared by every dashboard cron-create surface so the contract can't
+    drift between copies. The lazy import keeps cron out of module import.
+    """
+    from cron.scheduler import CronSchedulerRegistrationError
+
+    if isinstance(e, CronSchedulerRegistrationError):
+        raise HTTPException(status_code=424, detail=e.to_dict()) from e
+
+
 from hermes_cli.web_routers import cron as _cron_routes  # noqa: E402
 
 app.include_router(_cron_routes.router)
@@ -11910,13 +11923,7 @@ 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
+        _raise_if_cron_registration_error(e)
         _log.exception("POST /api/cron/jobs failed")
         raise HTTPException(status_code=400, detail=str(e))
 

tests/cron/conftest.py

@@ -14,6 +14,37 @@ inside the test, which overrides this fixture's value for that scope.
 import pytest
 
 
+@pytest.fixture()
+def make_cron_provider():
+    """Factory for minimal CronScheduler test doubles.
+
+    ``make_cron_provider(register_job=...)`` returns a real ``CronScheduler``
+    subclass instance whose ``register_job`` is the given callable — so tests
+    exercising the creation-registration contract share one stub instead of
+    redefining inline spy/failing classes, and an ABC rename breaks them
+    loudly instead of silently passing a duck-type.
+    """
+    from cron.scheduler_provider import CronScheduler
+
+    def _make(register_job=None, name="stub"):
+        class _StubProvider(CronScheduler):
+            @property
+            def name(self):  # pragma: no cover - trivial
+                return name
+
+            def start(self, stop_event, **kw):  # pragma: no cover - unused
+                pass
+
+            def register_job(self, job):
+                if register_job is not None:
+                    return register_job(job)
+                return None
+
+        return _StubProvider()
+
+    return _make
+
+
 @pytest.fixture(autouse=True)
 def _default_cron_test_model(monkeypatch):
     """Pin a default HERMES_MODEL so cron run_job tests have a resolvable model."""

tests/cron/test_jobs_changed_notify.py

@@ -9,6 +9,10 @@ the default path is unchanged.
 import pytest
 
 
+def _fail_registration(job):
+    raise RuntimeError("private callback URL and token")
+
+
 @pytest.fixture
 def temp_home(tmp_path, monkeypatch):
     monkeypatch.setenv("HERMES_HOME", str(tmp_path))
@@ -47,25 +51,16 @@ def test_builtin_notify_is_harmless(monkeypatch):
     sched._notify_provider_jobs_changed()
 
 
-def test_create_registers_first_trigger_with_active_provider(temp_home, monkeypatch):
+def test_create_registers_first_trigger_with_active_provider(
+    temp_home, monkeypatch, make_cron_provider
+):
     """A successful create is not reported until the provider sees the job."""
     import cron.scheduler_provider as sp
     import cron.scheduler as sched
 
     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())
+    provider = make_cron_provider(register_job=registered.append)
+    monkeypatch.setattr(sp, "resolve_cron_scheduler", lambda: provider)
 
     job = sched.create_job_with_scheduler_registration(
         prompt="echo hi",
@@ -76,24 +71,16 @@ def test_create_registers_first_trigger_with_active_provider(temp_home, monkeypa
     assert registered == [job]
 
 
-def test_create_failure_preserves_job_and_hides_provider_details(temp_home, monkeypatch):
+def test_create_failure_preserves_job_and_hides_provider_details(
+    temp_home, monkeypatch, make_cron_provider
+):
     """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())
+    provider = make_cron_provider(register_job=_fail_registration, name="failing")
+    monkeypatch.setattr(sp, "resolve_cron_scheduler", lambda: provider)
 
     with pytest.raises(sched.CronSchedulerRegistrationError) as exc_info:
         sched.create_job_with_scheduler_registration(
@@ -106,26 +93,20 @@ def test_create_failure_preserves_job_and_hides_provider_details(temp_home, monk
     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)
+    # Human-facing variant hides the exception class name and names the job.
+    assert "RuntimeError" not in error.user_message()
+    assert "'w'" in error.user_message()
 
 
-def test_tool_create_registers_provider_before_reporting_success(temp_home, monkeypatch):
+def test_tool_create_registers_provider_before_reporting_success(
+    temp_home, monkeypatch, make_cron_provider
+):
     """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())
+    provider = make_cron_provider(register_job=registered.append, name="recording")
+    monkeypatch.setattr(sp, "resolve_cron_scheduler", lambda: provider)
 
     from tools.cronjob_tools import cronjob
     import json
@@ -138,22 +119,14 @@ def test_tool_create_registers_provider_before_reporting_success(temp_home, monk
     assert [job["id"] for job in registered] == [out["job_id"]]
 
 
-def test_tool_create_reports_partial_registration_failure(temp_home, monkeypatch):
+def test_tool_create_reports_partial_registration_failure(
+    temp_home, monkeypatch, make_cron_provider
+):
     """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())
+    provider = make_cron_provider(register_job=_fail_registration, name="failing")
+    monkeypatch.setattr(sp, "resolve_cron_scheduler", lambda: provider)
 
     from tools.cronjob_tools import cronjob
     import json

tests/hermes_cli/test_cron_dashboard_off_loop.py

@@ -85,3 +85,45 @@ def test_blueprint_instantiate_create_job_off_loop(monkeypatch, loop_probe):
     assert ("call", False) in seen, (
         f"_call_cron_for_profile must run off the event loop; proof: {seen}"
     )
+
+
+def test_blueprint_instantiate_reports_saved_but_unregistered(monkeypatch):
+    """The instantiate endpoint maps a registration partial-failure to 424.
+
+    Endpoint-level guard for the shared ``_raise_if_cron_registration_error``
+    seam — the unit tests cover ``_create_cron_job_sync``; this proves the
+    blueprint route surfaces the same structured envelope through FastAPI.
+    """
+    from cron.scheduler import CronSchedulerRegistrationError
+
+    failure = CronSchedulerRegistrationError(
+        {"id": "bp-saved-job", "name": "bp job"},
+        RuntimeError("private callback URL and token"),
+    )
+
+    def fail_call(profile, fn, *args, **kwargs):
+        raise failure
+
+    monkeypatch.setattr(web_server, "_call_cron_for_profile", fail_call)
+    monkeypatch.setattr(web_server, "_has_valid_session_token", lambda req: True)
+
+    import cron.blueprint_catalog as bc
+    monkeypatch.setattr(bc, "get_blueprint", lambda key: object())
+    monkeypatch.setattr(
+        bc,
+        "fill_blueprint",
+        lambda bp, vals: {"name": "t", "schedule": "0 9 * * *", "prompt": "hi"},
+    )
+
+    client = TestClient(web_server.app)
+    resp = client.post(
+        "/api/cron/blueprints/instantiate",
+        json={"blueprint": "morning-brief", "values": {}},
+    )
+    assert resp.status_code == 424
+    detail = resp.json()["detail"]
+    assert detail["job_id"] == "bp-saved-job"
+    assert detail["job_saved"] is True
+    assert detail["scheduler_registered"] is False
+    assert detail["retry_create"] is False
+    assert "private callback URL and token" not in detail["error"]

tests/hermes_cli/test_web_server_cron_profiles.py

@@ -88,13 +88,21 @@ def test_create_registers_scheduler_inside_target_profile(
 ):
     """Dashboard create must resolve and register under the selected profile."""
     from cron import jobs as cron_jobs
+    from cron.scheduler_provider import CronScheduler
     from hermes_cli import web_server
     from hermes_constants import get_hermes_home
 
     worker_home = isolated_profiles["worker_alpha"]
     captured = {}
 
-    class RecordingProvider:
+    class RecordingProvider(CronScheduler):
+        @property
+        def name(self):
+            return "recording"
+
+        def start(self, stop_event, **kw):
+            pass
+
         def register_job(self, job):
             captured["job"] = job
             captured["runtime_home"] = get_hermes_home()

tools/cronjob_tools.py

@@ -1123,13 +1123,8 @@ def cronjob(
                     attach_to_session=attach_to_session,
                 )
             except CronSchedulerRegistrationError as exc:
-                return json.dumps(
-                    {
-                        "success": False,
-                        **exc.to_dict(),
-                    },
-                    indent=2,
-                )
+                _partial = exc.to_dict()
+                return tool_error(_partial.pop("error"), success=False, **_partial)
             _create_message = f"Cron job '{job['name']}' created."
             _local_notice = _local_delivery_notice(job, _normalize_deliver_param(deliver))
             if _local_notice: