diff --git a/cron/scheduler.py b/cron/scheduler.py index 0b3bf7465a..a2d6559f5a 100644 --- a/cron/scheduler.py +++ b/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, diff --git a/cron/scheduler_provider.py b/cron/scheduler_provider.py index 4c20db01c6..db3641a8c9 100644 --- a/cron/scheduler_provider.py +++ b/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 diff --git a/cron/suggestions.py b/cron/suggestions.py index 6826c2d8d3..dbffea0452 100644 --- a/cron/suggestions.py +++ b/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 diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 1b8c4af9c4..d7bf533266 100644 --- a/gateway/platforms/api_server.py +++ b/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) diff --git a/hermes_cli/blueprint_cmd.py b/hermes_cli/blueprint_cmd.py index e4a3afbf3d..a2f91bc8e1 100644 --- a/hermes_cli/blueprint_cmd.py +++ b/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}") diff --git a/hermes_cli/suggestions_cmd.py b/hermes_cli/suggestions_cmd.py index 0c0fd5a3e3..17381921b8 100644 --- a/hermes_cli/suggestions_cmd.py +++ b/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 " - 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", "") diff --git a/hermes_cli/web_routers/cron.py b/hermes_cli/web_routers/cron.py index 611926466d..7e7ad3bcba 100644 --- a/hermes_cli/web_routers/cron.py +++ b/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)) diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 7fc50719a3..997047e59e 100644 --- a/hermes_cli/web_server.py +++ b/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)) diff --git a/plugins/cron_providers/chronos/__init__.py b/plugins/cron_providers/chronos/__init__.py index 45e0fcf33e..b46d1d6dc5 100644 --- a/plugins/cron_providers/chronos/__init__.py +++ b/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: diff --git a/tests/cron/test_jobs_changed_notify.py b/tests/cron/test_jobs_changed_notify.py index 9275e0b71b..699afce836 100644 --- a/tests/cron/test_jobs_changed_notify.py +++ b/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"] diff --git a/tests/cron/test_suggestions.py b/tests/cron/test_suggestions.py index 9db7d2eeb4..605686f52c 100644 --- a/tests/cron/test_suggestions.py +++ b/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"] diff --git a/tests/gateway/test_api_server_jobs.py b/tests/gateway/test_api_server_jobs.py index 5f791650d4..416b3c5a1d 100644 --- a/tests/gateway/test_api_server_jobs.py +++ b/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() - diff --git a/tests/hermes_cli/test_web_server_cron_profiles.py b/tests/hermes_cli/test_web_server_cron_profiles.py index 59a5326462..93709203dd 100644 --- a/tests/hermes_cli/test_web_server_cron_profiles.py +++ b/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): - - - - diff --git a/tests/plugins/test_chronos_cron.py b/tests/plugins/test_chronos_cron.py index a1b77311b6..bf0cdca797 100644 --- a/tests/plugins/test_chronos_cron.py +++ b/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" - diff --git a/tools/blueprints.py b/tools/blueprints.py index 58427b7a7f..c236a32a8b 100644 --- a/tools/blueprints.py +++ b/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]]: diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index 3607288081..aeb2b39ae6 100644 --- a/tools/cronjob_tools.py +++ b/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: