From f5ba2cc1aac2acddfd6f67cb30a4de2587705538 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 28 Jul 2026 11:12:38 +0200 Subject: [PATCH] Cron hot-reload: CronService.reload + POST /api/cron/reload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Editing jobs.yaml or system.yaml now applies to the running daemon. The reload diffs the new job set against the scheduler and adds, removes or reschedules only what changed, leaving source runners and internal jobs alone. Gate plugins are re-imported, so an edited or deleted plugin takes effect even for jobs whose YAML did not change — those keep their timer and have their gate objects swapped, so editing a gate never resets a schedule. The reload is all-or-nothing: the whole change set is planned, and every trigger built, before the scheduler is touched, so a failure leaves every job on its existing schedule instead of applying half of the change. Reserved job ids are refused at load time, with the reservation shared with the CLI so one definition governs both. Schedule strings are parsed strictly. A 5-field crontab whose fields the scheduler rejects raised the same ValueError as "this is not a crontab", so "99 * * * *" fell through to the interval parser and silently became its 2-hour default. The two cases are now distinct exceptions, answered per call site: reload refuses the whole change set, startup skips just that job and logs it (raising would reach a blanket handler and cost every other cron, every source runner and the reload endpoint), and catch-up will not fire a job startup declined to schedule. Intervals are matched against the whole string, so "0.5h" is 1800s rather than the five hours an unanchored scan read out of the digits next to the unit. Co-Authored-By: Claude --- docs/cron.md | 93 +++- nerve/cli.py | 24 +- nerve/cron/gate_plugins.py | 102 +++- nerve/cron/gates.py | 24 +- nerve/cron/jobs.py | 71 ++- nerve/cron/service.py | 363 +++++++++++-- nerve/gateway/routes/cron.py | 16 + nerve/sources/runner.py | 8 +- tests/conftest.py | 18 + tests/test_cron.py | 116 +++++ tests/test_cron_gate_plugins.py | 360 +++++++++++-- tests/test_cron_gates.py | 23 +- tests/test_cron_reload.py | 866 ++++++++++++++++++++++++++++++++ 13 files changed, 1981 insertions(+), 103 deletions(-) create mode 100644 tests/test_cron_reload.py diff --git a/docs/cron.md b/docs/cron.md index 0872889c..f44f9063 100644 --- a/docs/cron.md +++ b/docs/cron.md @@ -26,6 +26,43 @@ Both files use the same format. On startup, CronService loads and merges both: Running `nerve init` on an existing install regenerates `system.yaml` (e.g., to pick up updated prompts from a Nerve update) without touching `jobs.yaml`. +### Hot-reload (no restart) + +`POST /api/cron/reload` re-reads `jobs.yaml` and `system.yaml` and applies them +to the running daemon, so a job you add, remove, reschedule, enable or disable +takes effect **without a restart**. The same call re-reads the gate-plugins +directory, so an added, edited or deleted plugin file lands with it. A +`prompt_file` needs no reload: its contents are read on every run. + +A reload rebuilds every enabled job from the file, and leaves source runners and +internal cleanup/wakeup jobs alone. + +Notes: +- Rebuilding a job normally leaves its next fire time alone: a crontab is + absolute, and an interval is measured from the job's last successful run. An + interval job that has never succeeded has no such anchor, so each reload + restarts its countdown. On a fresh install, where nothing has run yet, editing + the file repeatedly keeps every interval job's wait starting over — they + settle once each has run for the first time. +- A **malformed** `jobs.yaml`/`system.yaml` (bad YAML, or a job that doesn't + parse) is refused: reload returns `400` and the running schedule is left + untouched, so a typo can't wipe your crons. +- A reload is **all-or-nothing**. The whole change set is computed (files parsed, + every trigger built) before the scheduler is touched, so a reload that fails + for any reason leaves every job on its existing schedule instead of applying + half the change. Gate plugins are covered too: a refused reload leaves the + registered gates as they were, so a deleted plugin file only takes its gate + away once a reload succeeds. +- A job holding a [reserved id](#job-fields) is **skipped**, at reload and at + startup alike. The reload itself still succeeds; only that job is dropped. +- An **invalid schedule** (a crontab whose fields the scheduler rejects, e.g. + `99 * * * *`) is refused the same way at reload: `400`, nothing applied. + Startup is the one case that differs — the daemon logs an error naming the job + and comes up without it, rather than taking every other cron down over one + typo. `GET /api/cron/jobs` still lists the job, with a null `next_run`. +- `show_session_label` is **restart-only**: changing it and reloading has no + effect until the daemon restarts. + ## Job Definition ```yaml @@ -83,8 +120,8 @@ prompt definition. | Field | Type | Required | Description | |-------|------|----------|-------------| -| `id` | string | yes | Unique job identifier | -| `schedule` | string | yes | Crontab expression or interval (`2h`, `30m`) | +| `id` | string | yes | Unique job identifier. `cleanup`, `wakeup_sweep` and anything starting with `source:` are **reserved by the daemon** — a job using one is skipped (with a warning naming it in the daemon log) and never scheduled, so rename it | +| `schedule` | string | yes | Crontab expression or interval (`2h`, `30m`, `1h30m`, `0.5h`) — see [Interval syntax](#interval-syntax). A 5-field crontab with an out-of-range field (`99 * * * *`) is **rejected**, never reinterpreted as an interval — reload returns `400` and startup skips that one job with an error in the daemon log | | `prompt` | string | yes* | Message sent to the agent | | `prompt_file` | string | yes* | Path to a file containing the prompt (relative to the YAML's directory). Read fresh each run; shareable between jobs. *One of `prompt`/`prompt_file` is required | | `description` | string | no | Human-readable description | @@ -101,6 +138,31 @@ prompt definition. | `run_if` | list | no | Run gates — preconditions that must all hold for the job to fire. See [Run Gates](#run-gates) | | `workflow` | map | yes* | Launch a budget-capped [workflow run](workflow-runs.md) instead of a prompt: `{engine, prompt, budget_usd[, title, model, effort, cwd]}`. Takes precedence over `prompt`/`prompt_file`; the cron job only launches the run (fire-and-forget) — the run notifies on its own | +### Interval syntax + +A `schedule` that isn't a 5-field crontab is read as an interval: a run of +`` tokens, where the unit is `h`, `m` or `s`. Tokens add up, and +fractions are allowed. + +| Schedule | Interval | +|----------|----------| +| `4h` | 4 hours | +| `30m` | 30 minutes | +| `90s` | 90 seconds | +| `1h30m` | 90 minutes (`1h 30m` works too) | +| `0.5h` | 30 minutes | +| `10.5m` | 10 minutes 30 seconds | + +A fractional interval is rounded to the nearest whole second (`1.333m` → 80s). + +Anything that isn't a whole string of such tokens is **not an interval** and +silently falls back to **every 2 hours** — `hourly`, `@daily`, `4x`, `1h junk`, +and a zero interval (`0h`, or a fraction that rounds down to zero seconds), which +would otherwise mean "fire as fast as you can". The daemon keeps running on a +conservative cadence rather than refusing to start over one mistyped field, so a +job firing every 2 hours when you asked for something else means the schedule +string didn't parse. + ## Run Gates A **run gate** is a precondition evaluated right before a job fires. It answers @@ -134,9 +196,17 @@ When multiple gates are listed, the job runs only if every one passes: sources: [gmail, github] ``` -Gates are **fail-open**: if a gate errors while checking (e.g. a transient DB -issue), the run proceeds rather than being skipped — an occasional wasted run -beats a cron that silently never fires. +Gates are **fail-open while checking**: if a gate raises during evaluation (e.g. +a transient DB issue), the run proceeds rather than being skipped — an occasional +wasted run beats a cron that silently never fires. + +A gate that can't be *built* is a different matter and is **refused**. If a +`run_if` entry names a type nothing provides — a typo, or a gate plugin that was +deleted or fails to import — the job does not load: reload returns `400` naming +it, and at startup the daemon logs an error and comes up without that job. The +alternative would be to drop the gate and run the job anyway, which turns "only +when this holds" into "every time" — a precondition removed by accident is not a +precondition. ### Gate types @@ -230,7 +300,8 @@ run_if: A gate must implement the same three methods as a built-in (`is_satisfied`, `describe`, `from_config`). -**Rules** (all fail-safe — a bad plugin never crashes the daemon): +**Rules** (a bad plugin never crashes the daemon — though it does cost the jobs +that name its gate, see below): - Files whose name starts with `_` (and `__pycache__`) are ignored. - A plugin whose `type` collides with an already-registered gate is skipped @@ -238,8 +309,14 @@ A gate must implement the same three methods as a built-in (`is_satisfied`, loaded (filename-sorted) wins**. - Any import error in a plugin file is logged (naming the file) and that file is skipped; the rest still load. -- **No hot-reload:** adding or changing a plugin requires a daemon restart — - the same as every other piece of cron config. +- **Hot-reloadable:** `POST /api/cron/reload` re-reads the directory from + scratch, so an added, edited, deleted or renamed plugin takes effect without a + daemon restart. Built-in gates are never dropped. +- A plugin that stops registering its gate — deleted, renamed, or newly failing + to import — takes the jobs that name that gate with it: the reload is refused + with a `400`, and the running schedule is left as it was. Restore the file (or + remove the `run_if` entry) and reload again. Deleting a plugin is not a way to + switch a gate off; it stops those jobs instead of ungating them. > **Context is DB-only.** A gate receives `GateContext{job_id, db}`, which is > enough for DB-driven conditions (task counts, source cursors, age filters). A diff --git a/nerve/cli.py b/nerve/cli.py index 7b447429..43d02438 100644 --- a/nerve/cli.py +++ b/nerve/cli.py @@ -955,7 +955,7 @@ def doctor_report(config, config_source: str = "", check_api: bool = False) -> s # Check cron files (merge like the scheduler does: user jobs override system by ID) try: - from nerve.cron.jobs import load_jobs + from nerve.cron.jobs import describe_reserved_job_ids, is_reserved_job_id, load_jobs system_jobs = load_jobs(config.cron.system_file) if config.cron.system_file.exists() else [] user_jobs = load_jobs(config.cron.jobs_file) if config.cron.jobs_file.exists() else [] @@ -966,6 +966,18 @@ def doctor_report(config, config_source: str = "", check_api: bool = False) -> s for j in user_jobs: merged[j.id] = j + # Jobs on a reserved id are dropped by the scheduler, so counting them + # would report crons that can never fire. Call them out instead. + reserved_ids = sorted(jid for jid in merged if is_reserved_job_id(jid)) + for jid in reserved_ids: + del merged[jid] + if reserved_ids: + warnings.append( + f"[WARN] Cron job(s) {', '.join(reserved_ids)} use ids reserved " + f"by the daemon ({describe_reserved_job_ids()}) and are never " + f"scheduled — rename them" + ) + all_jobs = list(merged.values()) enabled = sum(1 for j in all_jobs if j.enabled) @@ -1314,7 +1326,7 @@ async def _run(): await cron_svc.run_job(job_id) else: click.echo("Available jobs:") - from nerve.cron.jobs import load_jobs + from nerve.cron.jobs import is_reserved_job_id, load_jobs # Load from both files, show provenance system_jobs = load_jobs(config.cron.system_file) @@ -1331,7 +1343,13 @@ async def _run(): all_jobs.append(("system", j)) for source, job in all_jobs: - status = "enabled" if job.enabled else "disabled" + # A reserved id is never scheduled by the daemon. Show the + # job rather than hiding it — the reason it isn't running is + # the whole thing the reader is here to find out. + if is_reserved_job_id(job.id): + status = "RESERVED ID — never scheduled, rename it" + else: + status = "enabled" if job.enabled else "disabled" click.echo( f" [{source:6s}] {job.id}: " f"{job.description or job.schedule} ({status})" diff --git a/nerve/cron/gate_plugins.py b/nerve/cron/gate_plugins.py index c67b7ca2..006022cd 100644 --- a/nerve/cron/gate_plugins.py +++ b/nerve/cron/gate_plugins.py @@ -45,8 +45,17 @@ def from_config(cls, spec: dict) -> "StaleTasksGate": * A gate gets only ``GateContext{job_id, db}`` (DB-only). A liveness/registry based gate is out of scope for this loader — it would need the context widened, a separate change. -* No hot-reload: adding or changing a plugin requires a daemon restart, the - same as every other piece of cron config. +* Hot-reload: ``load_gate_plugins(dir, replace=True)`` (used by + :meth:`nerve.cron.service.CronService.reload`) re-reads the directory from + scratch — a *new* file is registered, an *edited* file's new code replaces the + old class, and a *deleted* file's gate is unregistered. Reload then rebuilds + every job's gates from the refreshed registry and swaps the rebuilt job into + the scheduler, so the change reaches the running job and not just the + registry. Without ``replace`` the first-registered class wins (used on fresh + startup, and by validation so a candidate bundle can't swap the live + registry). A reload that is then refused rolls the registry back, so the same + rule holds for it: only a change that is actually applied reaches the live + registry. **Trust model.** Files in the gate-plugins directory are imported (i.e. executed) at daemon startup. This is the same trust model as ``config.yaml``, @@ -65,13 +74,36 @@ def from_config(cls, spec: dict) -> "StaleTasksGate": logger = logging.getLogger(__name__) +#: Module-name prefix stamped on every plugin-loaded module (see +#: :func:`_import_module`). It's how we tell a plugin-registered gate class apart +#: from a built-in (whose ``__module__`` is ``nerve.cron.gates``) — used by +#: ``replace=True`` to drop only plugin-owned entries on reload. +_PLUGIN_MODULE_PREFIX = "nerve_cron_gate_plugin_" -def load_gate_plugins(plugins_dir: Path) -> int: + +def load_gate_plugins( + plugins_dir: Path, *, replace: bool = False, warn_vanished: bool = True, +) -> int: """Discover and register :class:`CronGate` subclasses from *plugins_dir*. Returns the number of gate classes newly registered into :data:`GATE_REGISTRY`. A missing directory is a no-op (returns ``0``). + With ``replace=True`` the directory is re-read from scratch: every + plugin-owned entry is dropped from the registry first, so an *edited* + plugin's new code takes effect, a *deleted* plugin's gate is unregistered, + and a *renamed* type is moved — the registry ends up reflecting exactly the + files on disk. This is the hot-reload path (:meth:`CronService.reload`). + Built-ins are never dropped. The default (``replace=False``) is + first-registered-wins and is used on fresh startup and by config validation + (so validating a candidate bundle can't mutate the live registry). + + ``warn_vanished=False`` suppresses the gate-disappeared warning. A caller + that can still abandon the load — :meth:`CronService.reload` rolls the + registry back if the rest of the reload is refused — passes ``False`` here + and calls :func:`warn_vanished_gates` once the change is committed, so the + warning describes what happened rather than what was attempted. + Never raises: a broken plugin file is logged and skipped so it can't take down daemon startup (mirrors :func:`nerve.cron.gates.build_gates`' existing tolerance of a bad spec). @@ -82,8 +114,15 @@ def load_gate_plugins(plugins_dir: Path) -> int: logger.warning("Invalid cron gate plugins dir %r: %s", plugins_dir, e) return 0 + # Drop previously plugin-loaded gates before rescanning so edits/deletes are + # reflected. Do this even if the dir has since disappeared (deleted plugins + # must still unregister). Built-ins stay put. + removed_types = _unregister_plugin_gates() if replace else set() + if not plugins_dir.is_dir(): # Missing dir is the normal case — most installs have no custom gates. + if warn_vanished: + _warn_vanished(removed_types) return 0 registered = 0 @@ -97,9 +136,60 @@ def load_gate_plugins(plugins_dir: Path) -> int: logger.info( "Loaded %d custom cron gate(s) from %s", registered, plugins_dir, ) + if replace and warn_vanished: + _warn_vanished(removed_types) return registered +def warn_vanished_gates(before: dict[str, type[CronGate]]) -> None: + """Warn about gate types in *before* that are no longer registered. + + The deferred half of ``load_gate_plugins(..., warn_vanished=False)``: pass + the registry snapshot taken before the load and call this once the reload + that prompted it has committed. Built-ins are never dropped, so including + them in the snapshot cannot produce a spurious warning. + """ + _warn_vanished(set(before)) + + +def _unregister_plugin_gates() -> set[str]: + """Remove every plugin-loaded gate from :data:`GATE_REGISTRY`; keep built-ins. + + Returns the set of gate types removed, so the caller can warn about any that + fail to come back (a plugin that was deleted or no longer imports cleanly). + """ + removed: set[str] = set() + for gate_type, cls in list(GATE_REGISTRY.items()): + if getattr(cls, "__module__", "").startswith(_PLUGIN_MODULE_PREFIX): + del GATE_REGISTRY[gate_type] + removed.add(gate_type) + return removed + + +def _warn_vanished(removed_types: set[str]) -> None: + """Warn about gate types that were unregistered and did not re-register. + + A gate that disappears on reload means its plugin file was deleted or now + fails to import. A job whose ``run_if`` names it no longer builds at all — + ``build_gates`` raises on a type nothing registered — which refuses that + reload before it can commit, so such a reload never reaches this warning. Its + 400 is the signal instead, and it names the job. + + What reaches here is the other case: the plugin went away and no job + referenced it, so the reload committed. That is harmless right now, and still + worth a line, because the next job to name that type will be refused and the + cause will be this pull rather than that edit. + """ + for gate_type in sorted(removed_types - set(GATE_REGISTRY)): + logger.warning( + "Cron gate %r is no longer registered after reload — its plugin file " + "was removed or failed to re-import. No job uses it, or this reload " + "would have been refused; one that names it will be refused until " + "the plugin is back", + gate_type, + ) + + def _load_file(path: Path) -> int: """Import one plugin file and register its gate classes. Returns the count.""" module = _import_module(path) @@ -146,7 +236,11 @@ def _import_module(path: Path): throwaway namespace whose only purpose is to surface the gate classes it defines, so it never pollutes the global module table. """ - mod_name = f"nerve_cron_gate_plugin_{path.stem}" + # Derived from the shared constant, never spelled out again: the prefix is + # the only link between a module loaded here and the registry entries + # ``_unregister_plugin_gates`` is allowed to drop, so two copies drifting + # apart would silently turn hot-reload into "plugin gates never unregister". + mod_name = f"{_PLUGIN_MODULE_PREFIX}{path.stem}" try: spec = importlib.util.spec_from_file_location(mod_name, path) if spec is None or spec.loader is None: diff --git a/nerve/cron/gates.py b/nerve/cron/gates.py index f859a9cc..0b62287d 100644 --- a/nerve/cron/gates.py +++ b/nerve/cron/gates.py @@ -461,17 +461,21 @@ def build_gate(spec: dict) -> CronGate: def build_gates(specs: list[dict]) -> list[CronGate]: """Build gates from a list of config specs. - Invalid specs are logged and skipped rather than raising, so one bad - gate can't take down the whole cron service at load time. A job whose - gates all fail to build behaves as if it has no gates (runs normally). + Raises :class:`GateConfigError` if any spec cannot be built — an unknown + ``type`` above all, which is what a deleted or unimportable gate plugin looks + like from here. This runs from ``CronJob.__post_init__``, so it takes the job + with it: skipped with an error at startup, and a ``400`` that refuses the + whole change set at reload. + + Skipping the spec instead would be worse than losing the job. A gate is a + *precondition*, so dropping one makes the job run more often, not less: a + job that asked to run "only when the inbox is busy" starts running every + time, and the config that said otherwise is still sitting there. With + hot-reload that arrives unattended, from a workspace pull that removed a + plugin file. Refusing is the visible failure of the two — the reload names + the job in its 400, startup logs it, and no job silently changes meaning. """ - gates: list[CronGate] = [] - for spec in specs or []: - try: - gates.append(build_gate(spec)) - except GateConfigError as e: - logger.warning("Ignoring invalid cron gate %s: %s", spec, e) - return gates + return [build_gate(spec) for spec in specs or []] async def evaluate_gates( diff --git a/nerve/cron/jobs.py b/nerve/cron/jobs.py index 7d7703b2..cfa9aae7 100644 --- a/nerve/cron/jobs.py +++ b/nerve/cron/jobs.py @@ -19,6 +19,29 @@ logger = logging.getLogger(__name__) +# Scheduler ids the daemon owns; a job defined in system.yaml/jobs.yaml may not +# claim one. `source:` is reserved as a whole namespace rather than as the set of +# runners that happen to be live: source runners register as `source:`, and +# which ones exist depends on the sync config, so reserving only the live ones +# would let a job take `source:telegram` while telegram is switched off and then +# be silently replaced the moment it is switched back on. +RESERVED_JOB_IDS = frozenset({"cleanup", "wakeup_sweep"}) +RESERVED_JOB_ID_PREFIX = "source:" + + +def is_reserved_job_id(job_id: str) -> bool: + """True if *job_id* belongs to the daemon and can't be used by a cron job.""" + return job_id in RESERVED_JOB_IDS or job_id.startswith(RESERVED_JOB_ID_PREFIX) + + +def describe_reserved_job_ids() -> str: + """Human-readable summary of the reservation, for logs and CLI output.""" + return ( + f"{', '.join(sorted(RESERVED_JOB_IDS))}, " + f"and anything starting with '{RESERVED_JOB_ID_PREFIX}'" + ) + + @dataclass class CronJob: """A cron job definition.""" @@ -141,8 +164,15 @@ def resolve_prompt(self) -> str: return self.prompt def _build_gates(self) -> list["CronGate"]: - """Construct gate objects from run_if plus the legacy shorthand.""" - from nerve.cron.gates import build_gates + """Construct gate objects from run_if plus the legacy shorthand. + + A spec that cannot be built raises out of __post_init__ and takes the + whole job with it — see :func:`nerve.cron.gates.build_gates` for why that + beats quietly dropping the gate. The job id is added to the message here + because the spec alone doesn't say which YAML entry to go and fix, and + this text is what reaches the reload's 400. + """ + from nerve.cron.gates import GateConfigError, build_gates specs: list[dict] = list(self.run_if) # Translate the legacy skip_when_idle shorthand into a messages gate @@ -153,7 +183,10 @@ def _build_gates(self) -> list["CronGate"]: "sources": list(self.skip_when_idle), "consumer": self.idle_consumer, }) - return build_gates(specs) + try: + return build_gates(specs) + except GateConfigError as e: + raise GateConfigError(f"Cron job {self.id!r}: {e}") from e @classmethod def from_dict(cls, d: dict, base_dir: Path | None = None) -> CronJob: @@ -191,8 +224,16 @@ def from_dict(cls, d: dict, base_dir: Path | None = None) -> CronJob: return job -def load_jobs(jobs_file: Path) -> list[CronJob]: - """Load cron jobs from a YAML file.""" +def load_jobs(jobs_file: Path, strict: bool = False) -> list[CronJob]: + """Load cron jobs from a YAML file. + + By default this is *tolerant*: a YAML parse failure or an invalid job entry + is logged and skipped (so a typo never takes down daemon startup). With + ``strict=True`` any such failure raises :class:`nerve.config.ConfigError` + instead — used by hot-reload so a malformed file is refused rather than + silently unscheduling every job. A missing file is never an error (returns + ``[]``) in either mode. + """ if not jobs_file.exists(): logger.info("No cron jobs file at %s", jobs_file) return [] @@ -202,11 +243,20 @@ def load_jobs(jobs_file: Path) -> list[CronJob]: data = yaml.safe_load(f) or {} except Exception as e: logger.error("Failed to load cron jobs from %s: %s", jobs_file, e) + if strict: + from nerve.config import ConfigError + raise ConfigError(f"Failed to parse cron file {jobs_file}: {e}") from e return [] - jobs_data = data.get("jobs", []) - if isinstance(data, list): - jobs_data = data + jobs_data = data.get("jobs", []) if isinstance(data, dict) else data + if not isinstance(jobs_data, list): + if strict: + from nerve.config import ConfigError + raise ConfigError( + f"Cron file {jobs_file} must contain a 'jobs:' list" + ) + logger.error("Cron file %s has no valid 'jobs' list — ignoring", jobs_file) + return [] jobs = [] for item in jobs_data: @@ -214,6 +264,11 @@ def load_jobs(jobs_file: Path) -> list[CronJob]: jobs.append(CronJob.from_dict(item, base_dir=jobs_file.parent)) except (KeyError, TypeError, ValueError) as e: logger.warning("Invalid cron job definition: %s — %s", item, e) + if strict: + from nerve.config import ConfigError + raise ConfigError( + f"Invalid cron job in {jobs_file}: {item!r} — {e}" + ) from e logger.info("Loaded %d cron jobs from %s", len(jobs), jobs_file) return jobs diff --git a/nerve/cron/service.py b/nerve/cron/service.py index bafa48c3..8d501a94 100644 --- a/nerve/cron/service.py +++ b/nerve/cron/service.py @@ -17,11 +17,17 @@ from apscheduler.triggers.interval import IntervalTrigger from nerve.agent.engine import AgentEngine -from nerve.config import NerveConfig -from nerve.cron.jobs import CronJob, load_jobs +from nerve.config import ConfigError, NerveConfig +from nerve.cron.jobs import ( + CronJob, + describe_reserved_job_ids, + is_reserved_job_id, + load_jobs, +) from nerve.db import Database if TYPE_CHECKING: + from nerve.cron.gates import CronGate from nerve.sources.runner import SourceRunner logger = logging.getLogger(__name__) @@ -45,20 +51,89 @@ def _resolve_wakeup_prompt(prompt: str) -> str: return _WAKEUP_SENTINEL_PROMPT if prompt.strip() in _WAKEUP_SENTINELS else prompt +def _drop_reserved(jobs: list[CronJob]) -> list[CronJob]: + """Drop jobs holding a reserved id, warning once per offender. + + Nothing but a hand-edited YAML file can produce one of these, and a job that + simply never runs is near-impossible to diagnose from the outside, so the + warning has to name the job and the way out of it. + """ + kept: list[CronJob] = [] + for job in jobs: + if is_reserved_job_id(job.id): + logger.warning( + "Cron job '%s' (from %s) will not be scheduled: its id is " + "reserved by the daemon (%s). Rename the job to schedule it.", + job.id, + job.metadata.get("_source", "user"), + describe_reserved_job_ids(), + ) + else: + kept.append(job) + return kept + + +class NotCrontabError(ValueError): + """A schedule string that isn't a crontab expression at all. + + Signals "try the interval parser instead" — as opposed to + :class:`InvalidScheduleError`, which means the string *is* a crontab and + the interval parser must never see it. + """ + + +class InvalidScheduleError(ConfigError): + """A 5-field crontab expression the scheduler rejects. + + A ``ConfigError`` because that is exactly what it is — an operator typo in + a schedule — and because the reload route turns ConfigError into a 400 + naming the offending job instead of a bare 500. + """ + + def _parse_interval(interval: str) -> int: - """Parse an interval string like '2h', '30m', '1h30m' into seconds.""" + """Parse an interval string like '2h', '30m', '1h30m', '0.5h' into seconds. + + The whole string has to be a run of ```` tokens. Anything + else — ``hourly``, ``@daily``, ``???``, ``1h junk`` — is not an interval at + all and gets the 2h default, because the daemon has to keep running: a job + on a conservative cadence beats a scheduler that refused to start. + """ import re - total = 0 - parts = re.findall(r"(\d+)([hms])", interval.lower()) - for value, unit in parts: - v = int(value) + # One token, and the same expression used to check that the string is + # *only* tokens. An unanchored scan (the obvious `findall`) matches just the + # digits touching each unit, so it reads "0.5h" as 5 hours and "1.5h" as 5 + # hours with the leading 1 dropped entirely — a partial match silently + # mistaken for a whole one. fullmatch makes that impossible: leftovers mean + # the string isn't an interval, not that the parser gets to keep the part + # it liked. Whitespace between tokens is allowed ("1h 30m"); a 5-field + # crontab has already been ruled out by the time we get here. + # + # The number is spelled out the long way instead of as `\d*\.?\d+` so that + # each token matches exactly one way. Two ways to match a multi-digit number + # would make the repetition backtrack through every combination of them + # before giving up on a string that doesn't match — seconds of CPU for a + # schedule of twenty-odd tokens, which is a config typo away. + token = r"(\d+(?:\.\d+)?|\.\d+)([hms])" + text = interval.strip().lower() + if not re.fullmatch(rf"(?:{token}\s*)+", text): + return 7200 # Not an interval at all → default 2h + total = 0.0 + for value, unit in re.findall(token, text): + v = float(value) if unit == "h": total += v * 3600 elif unit == "m": total += v * 60 elif unit == "s": total += v - return total or 7200 # Default 2h + # Fractions mean what they say ("0.5h" is 1800s), rounded to the nearest + # whole second because IntervalTrigger counts seconds and half a second of + # drift on a cadence measured in minutes is noise. Zero takes the default + # too — written ("0h") or rounded down to it ("0.4s") — since + # IntervalTrigger(seconds=0) is a fire-as-fast-as-you-can loop rather than + # a schedule. + return round(total) or 7200 # Default 2h # Unix crontab day-of-week numbering is 0=Sun..6=Sat (7 also means Sun). @@ -110,25 +185,39 @@ def _crontab_to_trigger( Drop-in replacement for ``CronTrigger.from_crontab`` that fixes the day-of-week off-by-one (see ``_UNIX_DOW_TO_NAME``). Only the DOW field is treated differently; the other four fields and the no-explicit-timezone - behaviour are identical to ``from_crontab``. Raises ``ValueError`` for - anything that is not a 5-field expression, so interval strings like ``4h`` - keep falling through to the IntervalTrigger path. + behaviour are identical to ``from_crontab``. + + The two failure modes are different exceptions, because callers have to + tell them apart: :class:`NotCrontabError` for anything that is not a + 5-field expression, so interval strings like ``4h`` keep falling through + to the IntervalTrigger path, and :class:`InvalidScheduleError` when it is + a crontab whose fields the scheduler rejects. Both subclass ``ValueError``. """ fields = schedule.split() if len(fields) != 5: - raise ValueError(f"Not a 5-field crontab expression: {schedule!r}") + raise NotCrontabError(f"Not a 5-field crontab expression: {schedule!r}") minute, hour, day, month, day_of_week = fields remapped_dow = ",".join( _remap_dow_atom(atom) for atom in day_of_week.split(",") ) - return CronTrigger( - minute=minute, - hour=hour, - day=day, - month=month, - day_of_week=remapped_dow, - timezone=timezone, - ) + try: + return CronTrigger( + minute=minute, + hour=hour, + day=day, + month=month, + day_of_week=remapped_dow, + timezone=timezone, + ) + except ValueError as e: + # Five fields and one of them is bad ("99 * * * *"): a typo in a + # crontab, not an interval. Re-raised as its own type so no caller can + # mistake it for "not a crontab" and hand it to _parse_interval, which + # finds no h/m/s token and returns its 2h default — turning the typo + # into a job that quietly runs on a cadence nobody asked for. + raise InvalidScheduleError( + f"Invalid crontab expression {schedule!r}: {e}", + ) from e def _parse_timestamp(ts: str) -> datetime: @@ -170,7 +259,26 @@ async def start(self) -> None: if not job.enabled: continue - trigger = await self._make_trigger(job) + try: + trigger = await self._make_trigger(job) + except InvalidScheduleError as e: + # Deliberately the opposite of reload(), which refuses the whole + # change set: there the running schedule survives untouched and + # a synchronous caller gets a 400 naming the job. Nothing can be + # handed back that way here — gateway/server.py wraps this call + # in `except Exception: logger.warning(...)`, so raising over one + # typo would cost every other cron, every source runner and the + # reload endpoint itself (503 with no service) until someone + # noticed one warning line and restarted the daemon. Skip the + # offender loudly instead; it stays in _jobs, so /api/cron/jobs + # keeps listing it with a null next run rather than dropping it + # out of sight. + # _make_trigger's message already names the job; don't say it + # twice. + logger.error( + "%s. Not scheduled — fix the schedule and reload.", e, + ) + continue self.scheduler.add_job( self._run_job_wrapper, @@ -202,11 +310,21 @@ async def start(self) -> None: trigger = _crontab_to_trigger( schedule_str, timezone=self.timezone, ) - except ValueError: + except NotCrontabError: seconds = _parse_interval(schedule_str) trigger = IntervalTrigger( seconds=seconds, timezone=self.timezone, ) + except InvalidScheduleError as e: + # Same call as the cron loop above, same answer — skip this + # one, keep the rest. Letting it reach the enclosing + # `except Exception` would abandon every source after it in + # the list under a single "failed to register source + # runners" warning that names the wrong problem. + logger.error( + "Source '%s' will not be scheduled: %s", source_name, e, + ) + continue self.scheduler.add_job( self._run_source_wrapper, @@ -250,6 +368,169 @@ async def start(self) -> None: # Catch up missed jobs in background (don't block startup) asyncio.create_task(self._catchup_missed_jobs()) + async def reload(self) -> dict: + """Re-read cron config and apply changes to the running scheduler. + + Picks up added / removed / rescheduled / enabled-toggled jobs without a + daemon restart, mirroring the MCP-config reload pattern: + + * jobs removed or newly disabled → unscheduled + * every other enabled job → rebuilt from the new definition + + Every enabled job is rebuilt, not only the ones whose YAML changed. + Rebuilding a trigger keeps the next fire time as it was for a crontab + job, and for an interval job anchored to a successful run; an interval + job that has never succeeded has no anchor and restarts its countdown, + which is the only thing this costs (documented in docs/cron.md). + + Source runners and the fixed cleanup/wakeup jobs are left untouched. + Custom gate plugins are re-read from scratch (replace=True): a new file + registers, an edited file's code takes effect, and a deleted file's gate + unregisters (see nerve/cron/gate_plugins.py). Rebuilding every job is + what puts those gates in front of the jobs that run: the scheduler + executes the CronJob object it holds, and gates are built into that + object when it is constructed. + + All-or-nothing: the complete change set — every trigger included — is + built before the scheduler is touched, so a reload the daemon cannot + carry out raises with the running schedule exactly as it was. That + covers the gate registry too; see :meth:`_reload_from_disk`. + + Returns a summary dict: ``{"added", "removed", "updated", "enabled"}``, + where "updated" is every enabled job that already had a trigger. It says + what the reload rescheduled, not which jobs the file changed. + """ + from nerve.cron.gates import GATE_REGISTRY + + # The one piece of reload state that isn't local: GATE_REGISTRY is + # process-global and has to be replaced *before* jobs are rebuilt, + # because a CronJob builds its gates from it at construction time. That + # puts it ahead of every check that can still refuse the reload, so it + # needs undoing by hand — leaving the scheduler untouched is not enough + # when a deleted plugin's gate has already been unregistered. The next + # CronJob built from disk (run_job, rotate_session) would then fail to + # build for want of a gate type, off a reload that returned 400: the + # instance would lose jobs it had never agreed to change. + gates_before = dict(GATE_REGISTRY) + try: + return await self._reload_from_disk(gates_before) + except BaseException: + # BaseException, not Exception: a cancelled reload has to put the + # registry back too. + GATE_REGISTRY.clear() + GATE_REGISTRY.update(gates_before) + raise + + async def _reload_from_disk( + self, gates_before: dict[str, type["CronGate"]], + ) -> dict: + """The body of :meth:`reload`. Only call it through there. + + Takes the pre-reload registry snapshot so it can report which gates + vanished once the change is committed, and so its caller can restore + that snapshot if this raises. + """ + from nerve.cron.gate_plugins import load_gate_plugins, warn_vanished_gates + + # Re-read drop-in gate plugins before jobs are rebuilt (their gates are + # constructed at CronJob build time). replace=True so an edited/deleted + # plugin's code is picked up, not just newly-added files. The loader's own + # "gate vanished" warning is held back until the reload commits, since a + # refused one leaves those jobs holding the gate after all. + load_gate_plugins( + self.config.cron.gate_plugins_dir, replace=True, warn_vanished=False, + ) + + # -- Plan: everything that can fail, before anything is applied -------- + # + # Both the strict load below and the trigger construction further down + # can raise: the load on a malformed file, _make_trigger on a schedule + # the scheduler rejects (InvalidScheduleError → 400 — unlike at startup, + # where the offending job is skipped and the daemon comes up anyway, + # because here the running schedule survives and someone is waiting on + # an answer) or a failing DB read (interval jobs anchor their timer to + # the last successful run). Neither may run after a scheduler + # mutation. Unscheduling first and only then discovering a job whose + # trigger won't build would leave the daemon half-reloaded — part of the + # crons unscheduled, part still on their old triggers, and nothing to + # tell an operator which is which. Refusing the whole reload is the only + # outcome anyone can reason about, and the only one the API's 400 and + # the docs actually promise. + # + # Load strictly so a malformed jobs.yaml/system.yaml raises (and the + # route returns 400) instead of silently reading as "all jobs removed" + # and unscheduling everything. + # The loader already drops (and warns about) reserved ids, so nothing + # here can replace or remove the internal cleanup/wakeup/source + # triggers. _jobs is re-filtered in case it was seeded directly. + new_jobs = self._load_merged_jobs(strict=True) + new_by_id = {j.id: j for j in new_jobs} + old_by_id = {j.id: j for j in self._jobs if not is_reserved_job_id(j.id)} + + added: list[str] = [] + removed: list[str] = [] + updated: list[str] = [] + + unschedule: list[str] = [] + reschedule: list[tuple[CronJob, CronTrigger | IntervalTrigger]] = [] + + # Jobs that disappeared or became disabled → unschedule. + for jid, old in old_by_id.items(): + gone = jid not in new_by_id + disabled = (not gone) and (not new_by_id[jid].enabled) + if (gone or disabled) and self.scheduler.get_job(jid) is not None: + unschedule.append(jid) + if gone: + removed.append(jid) + + # Every enabled job is rebuilt from the file, whether or not its YAML + # changed. The scheduler lookup here still sees the pre-reload schedule, + # which is what it is meant to report against; the two loops never visit + # the same id anyway (this one skips disabled jobs and only walks ids + # present in the new config). + for jid, job in new_by_id.items(): + if not job.enabled: + continue + existing = self.scheduler.get_job(jid) is not None + reschedule.append((job, await self._make_trigger(job))) + # Report against the scheduler, not against _jobs: a job that was in + # _jobs but had no trigger is newly scheduled, not re-scheduled. + if existing: + updated.append(jid) + else: + added.append(jid) + + # -- Apply: scheduler bookkeeping only -------------------------------- + # No parsing, no DB, no awaits past this point, so the whole diff lands + # in one event-loop step and no job can fire against a half-applied set. + for jid in unschedule: + self.scheduler.remove_job(jid) + for job, trigger in reschedule: + self.scheduler.add_job( + self._run_job_wrapper, + trigger, + args=[job], + id=job.id, + name=job.description or job.id, + replace_existing=True, + ) + + self._jobs = new_jobs + # Committed, so the gates that disappeared really are gone from the jobs + # that are now live — say so (the loader was told to keep quiet above). + warn_vanished_gates(gates_before) + enabled = sum(1 for j in new_by_id.values() if j.enabled) + logger.info( + "Cron reloaded: +%d added, ~%d updated, -%d removed (%d enabled)", + len(added), len(updated), len(removed), enabled, + ) + return { + "added": added, + "removed": removed, + "updated": updated, + "enabled": enabled, + } + async def stop(self) -> None: """Stop the scheduler.""" self.scheduler.shutdown(wait=False) @@ -262,11 +543,20 @@ async def _make_trigger(self, job: CronJob) -> CronTrigger | IntervalTrigger: For interval schedules, anchors to the last successful run so the cadence survives restarts (persistent timer). + + Raises :class:`InvalidScheduleError` for a crontab the scheduler + rejects; each caller decides what refusing means (see start() and + reload()). """ try: return _crontab_to_trigger(job.schedule, timezone=self.timezone) - except ValueError: - pass + except NotCrontabError: + pass # not a crontab → an interval string like "4h" + except InvalidScheduleError as e: + # Name the job: the schedule alone doesn't say which YAML entry to + # go and fix, and this message is what an operator sees, either in + # the startup log or in the reload's 400. + raise InvalidScheduleError(f"Cron job '{job.id}': {e}") from e seconds = _parse_interval(job.schedule) last_run = await self.db.get_last_successful_cron_run(job.id) @@ -328,9 +618,16 @@ def _is_overdue( ) next_fire = trigger.get_next_fire_time(last_run, last_run) return next_fire is not None and next_fire < now - except ValueError: + except NotCrontabError: seconds = _parse_interval(job.schedule) return (now - last_run).total_seconds() >= seconds + except InvalidScheduleError: + # start() refused to schedule this job, so it has no fire times it + # could have missed. Catching it up would run, once per restart, + # precisely the job the daemon declined to schedule. start() already + # logged the typo, so this stays quiet rather than reporting it + # twice per boot. + return False # -- End persistent timers --------------------------------------------- @@ -536,18 +833,24 @@ async def _resolve_persistent_session(self, job: CronJob) -> tuple[str, bool]: # -- End persistent session generations ---------------------------------- - def _load_merged_jobs(self) -> list[CronJob]: + def _load_merged_jobs(self, strict: bool = False) -> list[CronJob]: """Load and merge jobs from system.yaml and jobs.yaml. System jobs come from system.yaml (managed by `nerve init`). User jobs come from jobs.yaml (user-defined, never touched by Nerve). If a user job has the same ID as a system job, the user version wins. + + Jobs holding an id the daemon reserves for itself are dropped here, so + no caller can schedule, catch up or hand-run one. + + With ``strict=True`` a malformed file raises ConfigError instead of + being silently treated as empty (used by reload()). """ system_file = self.config.cron.system_file jobs_file = self.config.cron.jobs_file - system_jobs = load_jobs(system_file) - user_jobs = load_jobs(jobs_file) + system_jobs = load_jobs(system_file, strict=strict) + user_jobs = load_jobs(jobs_file, strict=strict) if not system_jobs and user_jobs: # Backward compat: old install with everything in jobs.yaml @@ -558,7 +861,7 @@ def _load_merged_jobs(self) -> list[CronJob]: # Tag all as user-sourced (no system file yet) for j in user_jobs: j.metadata["_source"] = "user" - return user_jobs + return _drop_reserved(user_jobs) # Tag sources for display in CLI for j in system_jobs: @@ -579,7 +882,7 @@ def _load_merged_jobs(self) -> list[CronJob]: for j in user_jobs: jobs_by_id[j.id] = j - return list(jobs_by_id.values()) + return _drop_reserved(list(jobs_by_id.values())) async def _run_job_wrapper(self, job: CronJob) -> None: """Wrapper to run a cron job with logging and optional lock.""" diff --git a/nerve/gateway/routes/cron.py b/nerve/gateway/routes/cron.py index 067653ae..49a9f1f0 100644 --- a/nerve/gateway/routes/cron.py +++ b/nerve/gateway/routes/cron.py @@ -22,6 +22,22 @@ async def list_cron_jobs(user: dict = Depends(require_auth)): return {"jobs": jobs} +@router.post("/api/cron/reload") +async def reload_cron_jobs(user: dict = Depends(require_auth)): + """Re-read cron config and apply changes to the scheduler without a restart.""" + from nerve.config import ConfigError + from nerve.gateway.server import _cron_service + + if not _cron_service: + raise HTTPException(status_code=503, detail="Cron service not available") + + try: + result = await _cron_service.reload() + except ConfigError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + return {"reloaded": True, **result} + + @router.post("/api/cron/jobs/{job_id}/trigger") async def trigger_cron_job(job_id: str, user: dict = Depends(require_auth)): """Manually trigger a specific cron job or source runner.""" diff --git a/nerve/sources/runner.py b/nerve/sources/runner.py index c4e3d91e..5fd6ed6e 100644 --- a/nerve/sources/runner.py +++ b/nerve/sources/runner.py @@ -16,6 +16,7 @@ from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING, Any, Callable +from nerve.cron.jobs import RESERVED_JOB_ID_PREFIX from nerve.sources.models import IngestResult, SourceRecord if TYPE_CHECKING: @@ -110,7 +111,6 @@ class SourceRunner: source: The data source to fetch from. db: Database for cursor persistence and logging. batch_size: Max records per fetch. - job_id: Cron job ID for logging (default: "source:"). condense: Enable LLM condensation for long records. condense_model: Model name for condensation (e.g. "claude-haiku-4-5-20251001"). condense_prompt: Custom system prompt for condensation (uses default if empty). @@ -129,7 +129,6 @@ def __init__( source: Source, db: Database, batch_size: int = 50, - job_id: str = "", condense: bool = False, condense_model: str = "", condense_prompt: str = "", @@ -140,7 +139,10 @@ def __init__( self.source = source self.db = db self.batch_size = batch_size - self.job_id = job_id or f"source:{source.source_name}" + # Always derived, never passed in: the scheduler reserves the whole + # `source:` namespace for these runners (see nerve/cron/jobs.py), and a + # runner sitting outside it could be displaced by a user cron job. + self.job_id = f"{RESERVED_JOB_ID_PREFIX}{source.source_name}" self.condense = condense self.condense_model = condense_model self.condense_prompt = condense_prompt diff --git a/tests/conftest.py b/tests/conftest.py index 0f4ced7a..69dfc8f4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -78,6 +78,24 @@ def _isolate_nerve_state_files(tmp_path, monkeypatch): _repoint_import_time_state_paths(monkeypatch) +@pytest.fixture +def clean_registry(): + """Snapshot ``GATE_REGISTRY`` and restore it after the test. + + The gate-plugin loader mutates the process-global registry; without this, + gates registered by one test would leak into the others (and into + test_cron_gates.py, which asserts on the exact built-in set). + """ + from nerve.cron.gates import GATE_REGISTRY + + saved = dict(GATE_REGISTRY) + try: + yield + finally: + GATE_REGISTRY.clear() + GATE_REGISTRY.update(saved) + + @pytest_asyncio.fixture async def db(tmp_path): """Create a fresh in-memory-like database for each test.""" diff --git a/tests/test_cron.py b/tests/test_cron.py index c2fea377..c44d30f6 100644 --- a/tests/test_cron.py +++ b/tests/test_cron.py @@ -13,6 +13,8 @@ from nerve.cron.jobs import CronJob from nerve.cron.service import ( CronService, + InvalidScheduleError, + NotCrontabError, _crontab_to_trigger, _parse_interval, _parse_timestamp, @@ -137,6 +139,48 @@ def test_seconds(self): def test_default_on_garbage(self): assert _parse_interval("???") == 7200 + @pytest.mark.parametrize( + ("interval", "expected"), + [ + ("0.5h", 1800), + ("1.5h", 5400), + ("10.5m", 630), + ("0.25h", 900), + ("1.5h30s", 5430), + ], + ) + def test_fractions_mean_what_they_say(self, interval, expected): + # An unanchored scan reads only the digits touching the unit: "0.5h" + # becomes 5 hours, and "1.5h" becomes 5 hours with the leading 1 gone. + assert _parse_interval(interval) == expected + + def test_fraction_rounds_to_whole_seconds(self): + # IntervalTrigger counts whole seconds; never hand it a float. + result = _parse_interval("1.333m") + assert result == 80 + assert isinstance(result, int) + + @pytest.mark.parametrize( + "interval", + ["hourly", "@daily", "4x", "0.5.3h", "1h junk", "5.h", ""], + ) + def test_not_an_interval_takes_the_default(self, interval): + # Leftover text means the string isn't an interval, not that the parser + # keeps the part it liked ("1h junk" used to come back as 1h). + assert _parse_interval(interval) == 7200 + + @pytest.mark.parametrize("interval", ["0h", "0m", "0s", "0.0h", "0.4s"]) + def test_zero_interval_takes_the_default(self, interval): + # IntervalTrigger(seconds=0) is a hot loop, not a schedule. + assert _parse_interval(interval) == 7200 + + def test_whitespace_between_tokens(self): + assert _parse_interval("1h 30m") == 5400 + assert _parse_interval(" 4h ") == 14400 + + def test_uppercase_units(self): + assert _parse_interval("30M") == 1800 + # --------------------------------------------------------------------------- # Configured timezone @@ -225,6 +269,29 @@ def test_non_crontab_raises_value_error(self, schedule): with pytest.raises(ValueError): _crontab_to_trigger(schedule) + @pytest.mark.parametrize("schedule", ["4h", "30m", "1h30m", "???", ""]) + def test_non_crontab_raises_not_crontab_error(self, schedule): + """And raises the subclass the interval fallback actually catches.""" + with pytest.raises(NotCrontabError): + _crontab_to_trigger(schedule) + + @pytest.mark.parametrize("schedule", [ + "99 * * * *", # minute out of range + "0 99 * * *", # hour out of range + "* * * * 9", # day-of-week out of range + "nonsense * * * *", # not a number at all + ]) + def test_invalid_crontab_field_is_not_a_fallthrough(self, schedule): + """Five fields and one is bad: a typo, never an interval string. + + Must not be a NotCrontabError, or the callers' interval fallback + swallows it and _parse_interval turns the typo into its 2h default. + """ + with pytest.raises(InvalidScheduleError) as ei: + _crontab_to_trigger(schedule) + assert not isinstance(ei.value, NotCrontabError) + assert schedule in str(ei.value) + # --------------------------------------------------------------------------- # _is_overdue @@ -292,6 +359,17 @@ def test_weekly_overdue_after_exactly_one_week(self): job, last_run, last_run + timedelta(days=7, minutes=1), ) is True + def test_invalid_crontab_is_never_overdue(self): + """start() refused to schedule it, so catch-up must not fire it either. + + Without this the invalid crontab fell through to the interval parser's + 2h default and every restart ran a job that has no schedule at all. + """ + job = _make_job(schedule="99 * * * *") + assert CronService._is_overdue( + job, _utc_now() - timedelta(days=3), _utc_now(), + ) is False + # --------------------------------------------------------------------------- # _make_trigger (interval alignment) @@ -334,6 +412,18 @@ async def test_interval_no_last_run(self, cron_service): # Should be close to 4h from now assert timedelta(hours=3.5) < delta < timedelta(hours=4.5) + @pytest.mark.asyncio + async def test_fractional_interval(self, cron_service): + """"0.5h" is half an hour, and IntervalTrigger gets whole seconds.""" + cron_service.db.get_last_successful_cron_run.return_value = None + + job = _make_job(schedule="0.5h") + trigger = await cron_service._make_trigger(job) + + from apscheduler.triggers.interval import IntervalTrigger + assert isinstance(trigger, IntervalTrigger) + assert trigger.interval == timedelta(minutes=30) + @pytest.mark.asyncio async def test_crontab_unchanged(self, cron_service): """Crontab triggers are returned as-is (already absolute).""" @@ -363,6 +453,32 @@ async def test_interval_uses_configured_timezone(self): assert isinstance(trigger, IntervalTrigger) assert str(trigger.timezone) == "America/Los_Angeles" + @pytest.mark.asyncio + async def test_invalid_crontab_refused_not_turned_into_2h(self, cron_service): + """A typo'd crontab used to come back as a silent 2h interval. + + Both ValueErrors out of _crontab_to_trigger looked alike, so an invalid + field fell through to _parse_interval, which finds no h/m/s token and + returns its 2h default — a job running on a cadence nobody asked for, + with nothing logged. + """ + job = _make_job(id="typo", schedule="99 * * * *") + + with pytest.raises(InvalidScheduleError) as ei: + await cron_service._make_trigger(job) + + # The message has to name the job: the schedule alone doesn't say which + # YAML entry to fix. + assert "typo" in str(ei.value) and "99 * * * *" in str(ei.value) + + @pytest.mark.asyncio + async def test_invalid_crontab_raises_config_error(self, cron_service): + """reload()'s route maps ConfigError to 400; a bare ValueError 500s.""" + from nerve.config import ConfigError + + with pytest.raises(ConfigError): + await cron_service._make_trigger(_make_job(schedule="0 99 * * *")) + # --------------------------------------------------------------------------- # _catchup_missed_jobs diff --git a/tests/test_cron_gate_plugins.py b/tests/test_cron_gate_plugins.py index b2e11820..53bcccde 100644 --- a/tests/test_cron_gate_plugins.py +++ b/tests/test_cron_gate_plugins.py @@ -8,10 +8,12 @@ import pytest +from nerve.config import ConfigError from nerve.cron.gate_plugins import load_gate_plugins from nerve.cron.gates import ( GATE_REGISTRY, CronGate, + GateConfigError, GateContext, build_gate, evaluate_gates, @@ -23,22 +25,6 @@ # Fixtures / helpers # --------------------------------------------------------------------------- -@pytest.fixture -def clean_registry(): - """Snapshot GATE_REGISTRY and restore it after the test. - - The loader mutates the process-global registry; without this, gates - registered by one test would leak into the others (and into - test_cron_gates.py, which asserts on the exact built-in set). - """ - saved = dict(GATE_REGISTRY) - try: - yield - finally: - GATE_REGISTRY.clear() - GATE_REGISTRY.update(saved) - - # A valid plugin: a gate that is always satisfied, registered as "always_test". _VALID_PLUGIN = ''' from nerve.cron.gates import CronGate @@ -174,16 +160,17 @@ def test_abstract_gate_not_registered(self, tmp_path, clean_registry, caplog): assert "always_test" in GATE_REGISTRY assert "half.py" in caplog.text - def test_abstract_gate_does_not_crash_job_build(self, tmp_path, clean_registry): - # The crash vector itself: a job referencing the abstract gate's type - # must build without raising (the unknown type is dropped, fail-open). + def test_abstract_gate_is_refused_at_job_build(self, tmp_path, clean_registry): + # The loader skips the abstract class, so nothing registers its type, and + # a job naming it is refused. Containment means the daemon survives a bad + # plugin — not that a job whose precondition is missing runs anyway. _write(tmp_path, "half.py", _ABSTRACT_PLUGIN) load_gate_plugins(tmp_path) - job = CronJob( - id="j", schedule="1h", prompt="p", - run_if=[{"type": "half_test"}], - ) - assert job.gates == [] + with pytest.raises(GateConfigError, match="half_test"): + CronJob( + id="j", schedule="1h", prompt="p", + run_if=[{"type": "half_test"}], + ) def test_sys_exit_at_import_is_contained( self, tmp_path, clean_registry, caplog, @@ -279,6 +266,312 @@ def test_tilde_path_expanded(self, tmp_path, clean_registry, monkeypatch): assert load_gate_plugins(Path("~/nope/gates")) == 0 +# --------------------------------------------------------------------------- +# Hot-reload (replace=True) +# --------------------------------------------------------------------------- + +# Same type as _VALID_PLUGIN ("always_test") but the OPPOSITE behaviour: never +# satisfied. Simulates an operator editing a plugin's code in place. +_EDITED_PLUGIN = ''' +from nerve.cron.gates import CronGate + + +class AlwaysGate(CronGate): + type = "always_test" + + async def is_satisfied(self, ctx): + return False + + def describe(self): + return "never (edited test plugin)" + + @classmethod + def from_config(cls, spec): + return cls() +''' + + +class TestHotReloadReplace: + @pytest.mark.asyncio + async def test_edited_plugin_code_takes_effect_on_replace( + self, tmp_path, clean_registry, + ): + # Load v1 (always satisfied), then overwrite the file with v2 (never) + # and reload with replace=True — the new behaviour must win. + f = _write(tmp_path, "always.py", _VALID_PLUGIN) + load_gate_plugins(tmp_path) + gate = build_gate({"type": "always_test"}) + assert (await evaluate_gates([gate], _ctx())).should_run is True + + f.write_text(_EDITED_PLUGIN, encoding="utf-8") + assert load_gate_plugins(tmp_path, replace=True) == 1 + gate = build_gate({"type": "always_test"}) + assert (await evaluate_gates([gate], _ctx())).should_run is False + + def test_without_replace_edit_is_ignored(self, tmp_path, clean_registry): + # The old (asymmetric) behaviour: without replace, a re-run keeps the + # incumbent class, so an edit does NOT take effect. + f = _write(tmp_path, "always.py", _VALID_PLUGIN) + load_gate_plugins(tmp_path) + original = GATE_REGISTRY["always_test"] + f.write_text(_EDITED_PLUGIN, encoding="utf-8") + assert load_gate_plugins(tmp_path) == 0 # collision → kept + assert GATE_REGISTRY["always_test"] is original + + def test_deleted_plugin_unregisters_on_replace( + self, tmp_path, clean_registry, caplog, + ): + f = _write(tmp_path, "always.py", _VALID_PLUGIN) + load_gate_plugins(tmp_path) + assert "always_test" in GATE_REGISTRY + f.unlink() + with caplog.at_level(logging.WARNING): + assert load_gate_plugins(tmp_path, replace=True) == 0 + assert "always_test" not in GATE_REGISTRY # gate is gone + assert "always_test" in caplog.text # and it was surfaced + + def test_replace_when_dir_deleted_still_unregisters( + self, tmp_path, clean_registry, + ): + # The whole gates dir vanishing (not just a file) must still unregister. + import shutil + + gates_dir = tmp_path / "gates" + gates_dir.mkdir() + _write(gates_dir, "always.py", _VALID_PLUGIN) + load_gate_plugins(gates_dir) + assert "always_test" in GATE_REGISTRY + shutil.rmtree(gates_dir) + assert load_gate_plugins(gates_dir, replace=True) == 0 + assert "always_test" not in GATE_REGISTRY + + def test_replace_preserves_builtins(self, tmp_path, clean_registry): + _write(tmp_path, "always.py", _VALID_PLUGIN) + load_gate_plugins(tmp_path) + builtins_before = { + t: c for t, c in GATE_REGISTRY.items() + if not c.__module__.startswith("nerve_cron_gate_plugin_") + } + # A reload against an empty dir drops the plugin but keeps every built-in. + empty = tmp_path / "empty" + empty.mkdir() + load_gate_plugins(empty, replace=True) + assert "always_test" not in GATE_REGISTRY + for t, c in builtins_before.items(): + assert GATE_REGISTRY[t] is c + + def test_replace_swaps_type_across_files(self, tmp_path, clean_registry): + # Rename the gate's type by editing the file: old type gone, new present. + f = _write(tmp_path, "g.py", _VALID_PLUGIN) + load_gate_plugins(tmp_path) + assert "always_test" in GATE_REGISTRY + f.write_text( + _VALID_PLUGIN.replace('"always_test"', '"renamed_test"'), + encoding="utf-8", + ) + load_gate_plugins(tmp_path, replace=True) + assert "always_test" not in GATE_REGISTRY + assert "renamed_test" in GATE_REGISTRY + + @pytest.mark.asyncio + async def test_edit_takes_effect_through_cron_reload( + self, tmp_path, clean_registry, + ): + """The real path: CronService.reload() re-reads edited plugin code.""" + from unittest.mock import AsyncMock, MagicMock + + from nerve.cron.service import CronService + + gates_dir = tmp_path / "gates" + gates_dir.mkdir() + f = _write(gates_dir, "always.py", _VALID_PLUGIN) + + config = MagicMock() + config.timezone = "UTC" + config.cron.gate_plugins_dir = gates_dir + config.cron.system_file = tmp_path / "system.yaml" + config.cron.jobs_file = tmp_path / "jobs.yaml" + + svc = CronService(config, AsyncMock(), AsyncMock()) + svc.scheduler.start(paused=True) + svc._load_merged_jobs = MagicMock(return_value=[]) + + try: + await svc.reload() + assert ( + await evaluate_gates([build_gate({"type": "always_test"})], _ctx()) + ).should_run is True + f.write_text(_EDITED_PLUGIN, encoding="utf-8") + await svc.reload() + assert ( + await evaluate_gates([build_gate({"type": "always_test"})], _ctx()) + ).should_run is False + finally: + svc.scheduler.shutdown(wait=False) + + @pytest.mark.asyncio + async def test_edited_gate_reaches_the_scheduled_job( + self, tmp_path, clean_registry, + ): + """Registry-only isn't enough: the job APScheduler holds must be rebuilt. + + Editing a .py gate changes nothing in jobs.yaml, and the scheduler fires + the CronJob object it was handed, whose gates were built when that object + was constructed. Re-registering the class without replacing the object + would leave the job gating on the pre-edit class forever, while the API + reported the new one. + """ + import yaml + from unittest.mock import AsyncMock, MagicMock + + from nerve.cron.service import CronService + + gates_dir = tmp_path / "gates" + gates_dir.mkdir() + f = _write(gates_dir, "always.py", _VALID_PLUGIN) + + jobs_file = tmp_path / "jobs.yaml" + jobs_file.write_text(yaml.safe_dump({"jobs": [{ + "id": "j1", "schedule": "1h", "prompt": "x", + "run_if": [{"type": "always_test"}], + }]}), encoding="utf-8") + + config = MagicMock() + config.timezone = "UTC" + config.cron.gate_plugins_dir = gates_dir + config.cron.system_file = tmp_path / "system.yaml" + config.cron.jobs_file = jobs_file + + db = AsyncMock() + db.get_last_successful_cron_run = AsyncMock(return_value=None) + svc = CronService(config, AsyncMock(), db) + svc.scheduler.start(paused=True) + + def scheduled_gates(): + return svc.scheduler.get_job("j1").args[0].gates + + try: + await svc.reload() + assert (await evaluate_gates(scheduled_gates(), _ctx())).should_run is True + + f.write_text(_EDITED_PLUGIN, encoding="utf-8") + await svc.reload() + + # The object the scheduler will actually run now holds the new gate. + assert [g.describe() for g in scheduled_gates()] == [ + "never (edited test plugin)" + ] + assert (await evaluate_gates(scheduled_gates(), _ctx())).should_run is False + # ... and it agrees with what the API reports from _jobs. + assert [g.describe() for j in svc._jobs for g in j.gates] == [ + "never (edited test plugin)" + ] + finally: + svc.scheduler.shutdown(wait=False) + + @pytest.mark.asyncio + async def test_deleted_gate_stops_gating_the_scheduled_job( + self, tmp_path, clean_registry, + ): + """Deleting a plugin refuses the reload instead of ungating the job. + + This is the case the whole vanished-gate apparatus exists for. A synced + pull that removes a gate file must not turn "only when the plugin says + so" into "every time" on a live daemon, so the job stops building and the + reload is refused with the running schedule intact. + """ + import yaml + from unittest.mock import AsyncMock, MagicMock + + from nerve.cron.service import CronService + + gates_dir = tmp_path / "gates" + gates_dir.mkdir() + f = _write(gates_dir, "always.py", _EDITED_PLUGIN) # gate says "no" + + jobs_file = tmp_path / "jobs.yaml" + jobs_file.write_text(yaml.safe_dump({"jobs": [{ + "id": "j1", "schedule": "1h", "prompt": "x", + "run_if": [{"type": "always_test"}], + }]}), encoding="utf-8") + + config = MagicMock() + config.timezone = "UTC" + config.cron.gate_plugins_dir = gates_dir + config.cron.system_file = tmp_path / "system.yaml" + config.cron.jobs_file = jobs_file + + db = AsyncMock() + db.get_last_successful_cron_run = AsyncMock(return_value=None) + svc = CronService(config, AsyncMock(), db) + svc.scheduler.start(paused=True) + + try: + await svc.reload() + gates = svc.scheduler.get_job("j1").args[0].gates + assert (await evaluate_gates(gates, _ctx())).should_run is False + + f.unlink() + with pytest.raises(ConfigError, match="always_test"): + await svc.reload() + + # The running schedule is untouched, and the job it is still holding + # keeps the gate — so the job goes on being gated, not on firing. + gates = svc.scheduler.get_job("j1").args[0].gates + assert (await evaluate_gates(gates, _ctx())).should_run is False + # The registry was rolled back too, so the next load off disk works + # the moment the plugin is restored. + _write(gates_dir, "always.py", _EDITED_PLUGIN) + await svc.reload() + assert (await evaluate_gates( + svc.scheduler.get_job("j1").args[0].gates, _ctx(), + )).should_run is False + finally: + svc.scheduler.shutdown(wait=False) + + def test_warn_vanished_can_be_deferred(self, tmp_path, clean_registry, caplog): + """The unregister still happens; only the announcement is held back. + + A caller that may yet abandon the load asks for silence here and calls + warn_vanished_gates() itself once it has committed. + """ + from nerve.cron.gate_plugins import warn_vanished_gates + + f = _write(tmp_path, "always.py", _VALID_PLUGIN) + load_gate_plugins(tmp_path) + before = dict(GATE_REGISTRY) + + f.unlink() + with caplog.at_level(logging.WARNING): + load_gate_plugins(tmp_path, replace=True, warn_vanished=False) + assert "always_test" not in GATE_REGISTRY # dropped all the same + assert "always_test" not in caplog.text # but not announced + + with caplog.at_level(logging.WARNING): + warn_vanished_gates(before) + assert "always_test" in caplog.text + + def test_deferred_warning_ignores_builtins_in_the_snapshot( + self, tmp_path, clean_registry, caplog, + ): + """The snapshot holds built-ins too, and they are never dropped. + + warn_vanished_gates() takes the whole registry rather than the set the + loader removed, so it has to stay quiet about the built-ins in it. + """ + from nerve.cron.gate_plugins import warn_vanished_gates + + _write(tmp_path, "always.py", _VALID_PLUGIN) + load_gate_plugins(tmp_path) + before = dict(GATE_REGISTRY) + assert "tasks" in before, "expected a built-in in the snapshot" + + load_gate_plugins(tmp_path, replace=True, warn_vanished=False) + with caplog.at_level(logging.WARNING): + warn_vanished_gates(before) + assert caplog.text == "" + + # --------------------------------------------------------------------------- # End-to-end via CronJob.run_if # --------------------------------------------------------------------------- @@ -305,11 +598,14 @@ async def test_cronjob_plugin_gate_evaluates(self, tmp_path, clean_registry): decision = await evaluate_gates(job.gates, _ctx()) assert decision.should_run is True - def test_unknown_plugin_type_drops_gate_when_not_loaded(self, clean_registry): - # Without loading the plugin, an unknown type is dropped by build_gates - # (fail-open: the job ends up ungated) rather than raising. - job = CronJob( - id="j", schedule="1h", prompt="p", - run_if=[{"type": "never_loaded_gate"}], - ) - assert job.gates == [] + def test_unknown_plugin_type_refuses_the_job(self, clean_registry): + # Nothing registered the type — the plugin was never loaded, or never + # existed. Either way the job is refused, and the message names the job + # and the type so the YAML entry to fix is identifiable. + with pytest.raises(GateConfigError) as exc: + CronJob( + id="j", schedule="1h", prompt="p", + run_if=[{"type": "never_loaded_gate"}], + ) + assert "'j'" in str(exc.value) + assert "never_loaded_gate" in str(exc.value) diff --git a/tests/test_cron_gates.py b/tests/test_cron_gates.py index 8b4456f9..1dbedb0f 100644 --- a/tests/test_cron_gates.py +++ b/tests/test_cron_gates.py @@ -251,14 +251,27 @@ def test_non_dict_raises(self): with pytest.raises(GateConfigError): build_gate(["not", "a", "dict"]) # type: ignore[arg-type] - def test_build_gates_skips_invalid(self): - """An invalid spec is dropped; valid ones still build.""" + def test_build_gates_refuses_an_invalid_spec(self): + """One unbuildable spec fails the lot, rather than being left out. + + Silently dropping it would widen the job from "only when these hold" to + "whenever the schedule says", which is the opposite of what a gate is + for. The caller loses the job instead; see build_gates' docstring. + """ + with pytest.raises(GateConfigError, match="bogus"): + build_gates([ + {"type": "tasks", "status": "pending"}, + {"type": "bogus"}, + ]) + with pytest.raises(GateConfigError): + build_gates([{"type": "messages"}]) # no sources → invalid + + def test_build_gates_builds_every_valid_spec(self): gates = build_gates([ {"type": "tasks", "status": "pending"}, - {"type": "bogus"}, - {"type": "messages"}, # no sources → invalid + {"type": "messages", "sources": ["gmail"]}, ]) - assert len(gates) == 1 + assert len(gates) == 2 assert isinstance(gates[0], TasksGate) def test_build_gates_empty(self): diff --git a/tests/test_cron_reload.py b/tests/test_cron_reload.py new file mode 100644 index 00000000..06b10ea4 --- /dev/null +++ b/tests/test_cron_reload.py @@ -0,0 +1,866 @@ +"""Tests for cron hot-reload and the reserved job-id namespace (Story 5).""" + +from __future__ import annotations + +import logging +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock + +import pytest +import pytest_asyncio +import yaml + +from nerve.cron.service import CronService +from nerve.cron.jobs import CronJob, is_reserved_job_id + + +def _write_jobs(path: Path, jobs: list[dict]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(yaml.safe_dump({"jobs": jobs}), encoding="utf-8") + + +@pytest_asyncio.fixture +async def svc(tmp_path): + cron_dir = tmp_path / "cron" + cron_dir.mkdir(parents=True) + jobs_file = cron_dir / "jobs.yaml" + _write_jobs(jobs_file, []) + + config = MagicMock() + config.timezone = "UTC" + config.cron.jobs_file = jobs_file + config.cron.system_file = cron_dir / "system.yaml" # absent + config.cron.gate_plugins_dir = cron_dir / "gates" # absent → no-op + + db = AsyncMock() + db.get_last_successful_cron_run = AsyncMock(return_value=None) + + service = CronService(config, AsyncMock(), db) + # Start the scheduler paused so add/remove/replace_existing hit the real + # jobstore (as in production) without actually firing jobs. An unstarted + # scheduler keeps jobs in a "pending" list where replace_existing is a no-op. + service.scheduler.start(paused=True) + try: + yield service, jobs_file + finally: + service.scheduler.shutdown(wait=False) + + +def _job_dict(job_id="j1", schedule="1h", **kw): + return {"id": job_id, "schedule": schedule, "prompt": "do stuff", **kw} + + +# A gate plugin, for the reload paths that have to keep the registry in step with +# the schedule. It blocks, so a job that loses it goes from never running to +# running every time. +_GATE_PLUGIN = ''' +from nerve.cron.gates import CronGate + + +class BlockingGate(CronGate): + type = "reload_test" + + async def is_satisfied(self, ctx): + return False + + def describe(self): + return "blocking (test plugin)" + + @classmethod + def from_config(cls, spec): + return cls() +''' + + +def _write_gate_plugin(gates_dir: Path) -> Path: + """Drop the plugin above into *gates_dir* (the svc fixture's plugins dir).""" + gates_dir.mkdir(parents=True, exist_ok=True) + plugin = gates_dir / "blocking.py" + plugin.write_text(_GATE_PLUGIN, encoding="utf-8") + return plugin + + +def _gated_job(job_id: str, **kw) -> dict: + return _job_dict(job_id, run_if=[{"type": "reload_test"}], **kw) + + +class TestReload: + @pytest.mark.asyncio + async def test_add_job(self, svc): + service, jobs_file = svc + result = await service.reload() # empty → nothing + assert result["added"] == [] and result["enabled"] == 0 + + _write_jobs(jobs_file, [_job_dict("j1")]) + result = await service.reload() + assert result["added"] == ["j1"] + assert service.scheduler.get_job("j1") is not None + assert result["enabled"] == 1 + + @pytest.mark.asyncio + async def test_remove_job(self, svc): + service, jobs_file = svc + _write_jobs(jobs_file, [_job_dict("j1")]) + await service.reload() + assert service.scheduler.get_job("j1") is not None + + _write_jobs(jobs_file, []) + result = await service.reload() + assert result["removed"] == ["j1"] + assert service.scheduler.get_job("j1") is None + + @pytest.mark.asyncio + async def test_disable_job_unschedules(self, svc): + service, jobs_file = svc + _write_jobs(jobs_file, [_job_dict("j1")]) + await service.reload() + + _write_jobs(jobs_file, [_job_dict("j1", enabled=False)]) + result = await service.reload() + # Still present in config (so not "removed"), but unscheduled + not enabled. + assert result["removed"] == [] + assert "j1" not in result["added"] and "j1" not in result["updated"] + assert service.scheduler.get_job("j1") is None + assert result["enabled"] == 0 + + @pytest.mark.asyncio + async def test_reenable_job(self, svc): + service, jobs_file = svc + _write_jobs(jobs_file, [_job_dict("j1", enabled=False)]) + await service.reload() + assert service.scheduler.get_job("j1") is None + + _write_jobs(jobs_file, [_job_dict("j1", enabled=True)]) + result = await service.reload() + assert result["added"] == ["j1"] + assert service.scheduler.get_job("j1") is not None + + @pytest.mark.asyncio + async def test_reschedule_job_reports_updated(self, svc): + service, jobs_file = svc + _write_jobs(jobs_file, [_job_dict("j1", schedule="1h")]) + await service.reload() + first_trigger = str(service.scheduler.get_job("j1").trigger) + + _write_jobs(jobs_file, [_job_dict("j1", schedule="0 3 * * *")]) + result = await service.reload() + assert result["updated"] == ["j1"] + assert str(service.scheduler.get_job("j1").trigger) != first_trigger + + @pytest.mark.asyncio + async def test_identical_file_still_reschedules(self, svc): + """A reload rebuilds every enabled job rather than diffing against the + previous config, so an unchanged file reports its jobs as updated.""" + service, jobs_file = svc + _write_jobs(jobs_file, [_job_dict("j1")]) + await service.reload() + + result = await service.reload() # identical file + assert result == { + "added": [], "removed": [], "updated": ["j1"], "enabled": 1, + } + assert service.scheduler.get_job("j1") is not None + + +class TestReloadSafety: + @pytest.mark.asyncio + async def test_new_arrives_disabled_is_noop(self, svc): + service, jobs_file = svc + _write_jobs(jobs_file, [_job_dict("j1", enabled=False)]) + result = await service.reload() + assert result == {"added": [], "removed": [], "updated": [], "enabled": 0} + assert service.scheduler.get_job("j1") is None + + @pytest.mark.asyncio + async def test_disabled_stays_disabled_no_remove(self, svc): + service, jobs_file = svc + _write_jobs(jobs_file, [_job_dict("j1", enabled=False)]) + await service.reload() + # Reload again, still disabled — must not try to remove a job that was + # never scheduled (the get_job guard), and must not report it removed. + result = await service.reload() + assert result["removed"] == [] + + @pytest.mark.asyncio + async def test_edited_definition_reaches_the_scheduler(self, svc): + """The scheduler runs the CronJob object it holds, so a reload has to + replace that object and not just the copy in ``_jobs``. Uses a workflow + block because it decides what the run actually does — engine, prompt and + budget — and none of it is re-read at fire time.""" + service, jobs_file = svc + w = {"engine": "claude", "prompt": "go", "budget_usd": 1.0} + _write_jobs(jobs_file, [_job_dict("j1", workflow=w)]) + await service.reload() + _write_jobs( + jobs_file, [_job_dict("j1", workflow={**w, "budget_usd": 25.0})], + ) + result = await service.reload() + assert result["updated"] == ["j1"] + assert service.scheduler.get_job("j1").args[0].workflow["budget_usd"] == 25.0 + assert service._jobs[0].workflow["budget_usd"] == 25.0 + + @pytest.mark.asyncio + async def test_reserved_id_collision_ignored(self, svc): + service, jobs_file = svc + # Register a fake source runner id + rely on cleanup/wakeup_sweep. + fake_runner = MagicMock() + fake_runner.job_id = "source:gmail" + service._source_runners = [fake_runner] + # Simulate the internal cleanup job already occupying the slot. + service.scheduler.add_job( + lambda: None, "interval", seconds=3600, id="cleanup", + ) + + _write_jobs(jobs_file, [ + _job_dict("cleanup", schedule="1h"), + _job_dict("source:gmail", schedule="1h"), + _job_dict("legit", schedule="1h"), + ]) + result = await service.reload() + # Only the non-reserved job is scheduled; reserved ids are ignored. + assert result["added"] == ["legit"] + assert service.scheduler.get_job("legit") is not None + # The internal cleanup job is untouched (still the lambda interval job). + assert service.scheduler.get_job("cleanup") is not None + + @pytest.mark.asyncio + async def test_reserved_removal_does_not_delete_internal_job(self, svc): + service, jobs_file = svc + # User previously had a 'cleanup' cron in self._jobs (as start would keep). + service._jobs = [CronJob(id="cleanup", schedule="1h", prompt="x")] + service.scheduler.add_job( + lambda: None, "interval", seconds=3600, id="cleanup", + ) + _write_jobs(jobs_file, []) # user removed their cleanup cron + result = await service.reload() + # Internal cleanup job must survive; not reported removed. + assert "cleanup" not in result["removed"] + assert service.scheduler.get_job("cleanup") is not None + + @pytest.mark.asyncio + async def test_source_namespace_reserved_while_runner_is_down(self, svc): + service, jobs_file = svc + # No source runners registered at all — the whole `source:` namespace is + # still off limits, otherwise the job would be silently replaced the + # moment that source is turned back on. + assert service._source_runners == [] + + _write_jobs(jobs_file, [ + _job_dict("source:telegram", schedule="1h"), + _job_dict("legit", schedule="1h"), + ]) + result = await service.reload() + assert result["added"] == ["legit"] + assert result["enabled"] == 1 + assert service.scheduler.get_job("source:telegram") is None + assert "source:telegram" not in [j.id for j in service._jobs] + + @pytest.mark.asyncio + async def test_reserved_dropped_on_the_merged_path(self, svc): + """The system+user merge — every install after `nerve init` — filters too.""" + service, jobs_file = svc + system_file = service.config.cron.system_file + _write_jobs(system_file, [ + _job_dict("wakeup_sweep", schedule="1h"), + _job_dict("sys-job", schedule="1h"), + ]) + _write_jobs(jobs_file, [ + _job_dict("source:github", schedule="1h"), + _job_dict("user-job", schedule="1h"), + ]) + result = await service.reload() + assert sorted(result["added"]) == ["sys-job", "user-job"] + assert sorted(j.id for j in service._jobs) == ["sys-job", "user-job"] + assert service.scheduler.get_job("source:github") is None + # The daemon's own wakeup sweep is never displaced by a same-named job. + assert service.scheduler.get_job("wakeup_sweep") is None + + @pytest.mark.asyncio + async def test_reserved_rejection_is_logged(self, svc, caplog): + service, jobs_file = svc + _write_jobs(jobs_file, [_job_dict("source:gmail", schedule="1h")]) + with caplog.at_level(logging.WARNING, logger="nerve.cron.service"): + await service.reload() + # An operator who hand-edited jobs.yaml must be able to find out why. + assert any( + "source:gmail" in r.getMessage() and "reserved" in r.getMessage() + for r in caplog.records + ) + + @pytest.mark.asyncio + async def test_unchanged_job_with_missing_trigger_is_rescheduled(self, svc): + service, jobs_file = svc + _write_jobs(jobs_file, [_job_dict("j1")]) + await service.reload() + + # Something dropped the trigger behind reload's back. An unchanged job + # must not be left stranded while the summary keeps calling it enabled. + service.scheduler.remove_job("j1") + result = await service.reload() + # Newly scheduled, so "added" — the summary reports what the scheduler + # did, not what changed in the file. + assert result["added"] == ["j1"] + assert result["updated"] == [] + assert result["enabled"] == 1 + assert service.scheduler.get_job("j1") is not None + + @pytest.mark.asyncio + async def test_jobs_refreshed_without_scheduler_report_as_added(self, svc): + service, jobs_file = svc + # run_job()/rotate_session() refresh _jobs from disk without touching the + # scheduler. The next reload schedules those jobs for the first time, so + # they are added, not updated. + _write_jobs(jobs_file, [_job_dict("j1"), _job_dict("j2")]) + service._jobs = service._load_merged_jobs() + assert service.scheduler.get_job("j1") is None + + result = await service.reload() + assert sorted(result["added"]) == ["j1", "j2"] + assert result["updated"] == [] + + @pytest.mark.asyncio + async def test_malformed_yaml_refused(self, svc): + from nerve.config import ConfigError + + service, jobs_file = svc + _write_jobs(jobs_file, [_job_dict("j1")]) + await service.reload() + + # Corrupt the file — reload must refuse and leave the schedule intact. + jobs_file.write_text("jobs: [ this is: not valid: yaml\n", encoding="utf-8") + with pytest.raises(ConfigError): + await service.reload() + assert service.scheduler.get_job("j1") is not None # still scheduled + + @pytest.mark.asyncio + async def test_trigger_build_failure_leaves_schedule_intact(self, svc): + """A job whose trigger won't build must not take the others down with it. + + Interval jobs anchor their timer to the last successful run, so building + a trigger reads the database — and that read can fail long after the + config parsed fine. If the reload had already unscheduled the removed + jobs by then, the daemon would be left half-reloaded with no way for an + operator to tell which crons are still live. + """ + service, jobs_file = svc + _write_jobs(jobs_file, [ + _job_dict("keep-a", schedule="1h"), + _job_dict("keep-b", schedule="30m"), + _job_dict("doomed", schedule="2h"), + ]) + await service.reload() + before = { + jid: str(service.scheduler.get_job(jid).trigger) + for jid in ("keep-a", "keep-b", "doomed") + } + + # A reload that touches all three paths: keep-a rescheduled, keep-b + # removed, doomed's trigger unbuildable. + _write_jobs(jobs_file, [ + _job_dict("keep-a", schedule="15m"), + _job_dict("doomed", schedule="3h"), + ]) + + async def flaky_last_run(job_id): + if job_id == "doomed": + raise RuntimeError("database is locked") + return None + + service.db.get_last_successful_cron_run = AsyncMock( + side_effect=flaky_last_run, + ) + + with pytest.raises(RuntimeError): + await service.reload() + + # Every job that was running still runs, on its original trigger. + live = {jid: service.scheduler.get_job(jid) for jid in before} + assert [jid for jid, j in live.items() if j is None] == [] + assert {jid: str(j.trigger) for jid, j in live.items()} == before + # And the recorded job list still matches the live schedule. + assert sorted(j.id for j in service._jobs) == [ + "doomed", "keep-a", "keep-b", + ] + + @pytest.mark.asyncio + async def test_invalid_job_entry_refused(self, svc): + from nerve.config import ConfigError + + service, jobs_file = svc + _write_jobs(jobs_file, [_job_dict("j1")]) + await service.reload() + # A job missing required fields (no prompt/prompt_file) → strict raise. + jobs_file.write_text( + yaml.safe_dump({"jobs": [{"id": "bad", "schedule": "1h"}]}), + encoding="utf-8", + ) + with pytest.raises(ConfigError): + await service.reload() + assert service.scheduler.get_job("j1") is not None + + @pytest.mark.asyncio + async def test_invalid_schedule_refused(self, svc): + """A crontab the scheduler rejects refuses the reload, like bad YAML. + + It used to be accepted as a 2h interval, so the reload "succeeded" and + the job ran on a cadence the operator never wrote down. + """ + from nerve.config import ConfigError + + service, jobs_file = svc + _write_jobs(jobs_file, [_job_dict("j1", schedule="1h")]) + await service.reload() + before = str(service.scheduler.get_job("j1").trigger) + + _write_jobs(jobs_file, [ + _job_dict("j1", schedule="30m"), + _job_dict("typo", schedule="99 * * * *"), + ]) + with pytest.raises(ConfigError) as ei: + await service.reload() + + assert "typo" in str(ei.value) + # All-or-nothing: j1 keeps its old trigger, the typo is never scheduled. + assert str(service.scheduler.get_job("j1").trigger) == before + assert service.scheduler.get_job("typo") is None + + +class TestRefusedReloadRestoresGates: + """All-or-nothing has to cover GATE_REGISTRY, which is process-global. + + Plugin gates must be replaced before jobs are rebuilt — a CronJob builds its + gates at construction time — which puts the registry ahead of every check + that can still refuse the reload. Leaving the scheduler alone is then not + enough on its own: the registry has to be put back by hand. + """ + + @pytest.mark.asyncio + async def test_registry_is_restored(self, svc, clean_registry): + from nerve.config import ConfigError + from nerve.cron.gates import GATE_REGISTRY + + service, jobs_file = svc + plugin = _write_gate_plugin(jobs_file.parent / "gates") + _write_jobs(jobs_file, [_gated_job("j1")]) + await service.reload() + registered = GATE_REGISTRY["reload_test"] + + # One edit deletes the plugin and breaks the YAML. + plugin.unlink() + jobs_file.write_text("jobs: [ this is: not valid: yaml\n", encoding="utf-8") + with pytest.raises(ConfigError): + await service.reload() + + # The same class object, not just the same type name: a rollback puts the + # snapshot back, where a re-import would install an equivalent new class. + assert GATE_REGISTRY["reload_test"] is registered + + @pytest.mark.asyncio + async def test_rollback_covers_a_failure_after_the_load( + self, svc, clean_registry, + ): + """The guard spans the planning pass, not just the strict load. + + A trigger that won't build raises at the far end of it, long after the + plugins have been replaced. + """ + from nerve.cron.gates import GATE_REGISTRY + + service, jobs_file = svc + plugin = _write_gate_plugin(jobs_file.parent / "gates") + _write_jobs(jobs_file, [_gated_job("j1")]) + await service.reload() + + # The plugin and the job that used it go away together. Removing only the + # plugin would have the strict load refuse this reload before the planning + # pass is reached, which is a different failure than the one under test. + plugin.unlink() + _write_jobs(jobs_file, [_job_dict("doomed")]) + + async def flaky_last_run(job_id): + if job_id == "doomed": + raise RuntimeError("database is locked") + return None + + service.db.get_last_successful_cron_run = AsyncMock( + side_effect=flaky_last_run, + ) + with pytest.raises(RuntimeError): + await service.reload() + + assert "reload_test" in GATE_REGISTRY + + @pytest.mark.asyncio + async def test_no_vanished_gate_is_announced(self, svc, clean_registry, caplog): + """A refused reload must not announce a gate that is still registered. + + The warning reports that a gate stopped being available. A refused reload + puts it back, so saying so would describe a regression that this very + reload prevented — and the operator would go looking for a plugin that is + sitting where it always was. + """ + from nerve.config import ConfigError + + service, jobs_file = svc + plugin = _write_gate_plugin(jobs_file.parent / "gates") + _write_jobs(jobs_file, [_gated_job("j1")]) + await service.reload() + + plugin.unlink() + jobs_file.write_text("jobs: [ this is: not valid: yaml\n", encoding="utf-8") + with caplog.at_level(logging.WARNING): + with pytest.raises(ConfigError): + await service.reload() + + assert "reload_test" not in caplog.text + # And the claim it would have made is indeed false — j1 still has its gate. + assert len(next(j for j in service._jobs if j.id == "j1").gates) == 1 + + @pytest.mark.asyncio + async def test_a_manual_run_afterwards_keeps_its_gate(self, svc, clean_registry): + """What the rollback is for: run_job goes back to disk for a missing id. + + Rebuilding a job there takes its gates from the live registry, so a + registry left stripped by a refused reload makes every gated job fail to + build — and since run_job replaces the whole of _jobs, they all drop out + at once. The instance would lose jobs off a reload that returned 400, + having agreed to change nothing. + """ + from nerve.config import ConfigError + + service, jobs_file = svc + plugin = _write_gate_plugin(jobs_file.parent / "gates") + _write_jobs(jobs_file, [_gated_job("gated")]) + await service.reload() + + # The operator adds a second gated job, deletes the plugin, and mistypes + # a schedule badly enough to be refused — all in one sync. + plugin.unlink() + _write_jobs(jobs_file, [ + _gated_job("gated"), + _gated_job("added"), + _job_dict("typo", schedule="99 * * * *"), + ]) + with pytest.raises(ConfigError): + await service.reload() + + service._run_job_wrapper = AsyncMock() + await service.run_job("added") # not in _jobs → reloads from disk + + rebuilt = {j.id: len(j.gates) for j in service._jobs} + assert rebuilt["added"] == 1 + assert rebuilt["gated"] == 1 + + @pytest.mark.asyncio + async def test_a_committed_reload_still_announces_it( + self, svc, clean_registry, caplog, + ): + """Holding the warning back until the commit must not amount to losing it. + + The reload has to be one that commits, which now means the plugin and the + last job using it went away together — removing the plugin alone refuses + the reload, and its 400 does the telling. This is the case where the + warning is the only notice that a gate stopped being available. + """ + from nerve.cron.gates import GATE_REGISTRY + + service, jobs_file = svc + plugin = _write_gate_plugin(jobs_file.parent / "gates") + _write_jobs(jobs_file, [_gated_job("j1")]) + await service.reload() + + plugin.unlink() + _write_jobs(jobs_file, [_job_dict("j1")]) # same job, no longer gated + with caplog.at_level(logging.WARNING): + await service.reload() # valid config → applies + + assert "reload_test" not in GATE_REGISTRY + assert "reload_test" in caplog.text + + +class TestInvalidScheduleAtStartup: + """Startup answers the same error differently from reload(), on purpose.""" + + @pytest.mark.asyncio + async def test_start_skips_only_the_offending_job( + self, tmp_path, monkeypatch, caplog, + ): + """One typo must not cost the other jobs their daemon.""" + import nerve.sources.registry as registry + + monkeypatch.setattr(registry, "build_source_runners", lambda *a, **k: []) + + cron_dir = tmp_path / "cron" + jobs_file = cron_dir / "jobs.yaml" + _write_jobs(jobs_file, [ + _job_dict("typo", schedule="99 * * * *"), + _job_dict("legit", schedule="1h"), + ]) + + config = MagicMock() + config.timezone = "UTC" + config.cron.jobs_file = jobs_file + config.cron.system_file = cron_dir / "system.yaml" # absent + config.cron.gate_plugins_dir = cron_dir / "gates" # absent → no-op + + db = AsyncMock() + db.get_last_successful_cron_run = AsyncMock(return_value=None) + + service = CronService(config, AsyncMock(), db) + with caplog.at_level(logging.ERROR, logger="nerve.cron.service"): + await service.start() # must not raise + try: + assert service.scheduler.get_job("legit") is not None + assert service.scheduler.get_job("typo") is None + # Kept in _jobs so list_jobs still shows it — with no next run — + # rather than dropping it out of sight. + assert sorted(j.id for j in service._jobs) == ["legit", "typo"] + finally: + service.scheduler.shutdown(wait=False) + + logged = " ".join(r.getMessage() for r in caplog.records) + assert "typo" in logged and "99 * * * *" in logged + + @pytest.mark.asyncio + async def test_start_skips_only_the_offending_source( + self, tmp_path, monkeypatch, caplog, + ): + """Same for sync..schedule, and the later sources still register. + + The registration loop sits inside a blanket `except Exception`, so an + error escaping one runner would silently abandon every runner after it. + """ + import nerve.sources.registry as registry + + def _runner(name): + runner = MagicMock() + runner.source.source_name = name + runner.job_id = f"source:{name}" + return runner + + # Bad one first: whatever follows it is what a swallowed error costs. + monkeypatch.setattr( + registry, "build_source_runners", + lambda *a, **k: [_runner("gmail"), _runner("slack")], + ) + + cron_dir = tmp_path / "cron" + jobs_file = cron_dir / "jobs.yaml" + _write_jobs(jobs_file, []) + + config = MagicMock() + config.timezone = "UTC" + config.cron.jobs_file = jobs_file + config.cron.system_file = cron_dir / "system.yaml" # absent + config.cron.gate_plugins_dir = cron_dir / "gates" # absent → no-op + config.sync.gmail.schedule = "0 99 * * *" + config.sync.slack.schedule = "*/15 * * * *" + + db = AsyncMock() + db.get_last_successful_cron_run = AsyncMock(return_value=None) + + service = CronService(config, AsyncMock(), db) + with caplog.at_level(logging.ERROR, logger="nerve.cron.service"): + await service.start() + try: + assert service.scheduler.get_job("source:gmail") is None + assert service.scheduler.get_job("source:slack") is not None + finally: + service.scheduler.shutdown(wait=False) + + logged = " ".join(r.getMessage() for r in caplog.records) + assert "gmail" in logged and "0 99 * * *" in logged + + +class TestReloadRoute: + @pytest.mark.asyncio + async def test_503_when_no_service(self, monkeypatch): + import nerve.gateway.server as srv + from fastapi import HTTPException + + from nerve.gateway.routes.cron import reload_cron_jobs + + monkeypatch.setattr(srv, "_cron_service", None, raising=False) + with pytest.raises(HTTPException) as ei: + await reload_cron_jobs(user={}) + assert ei.value.status_code == 503 + + @pytest.mark.asyncio + async def test_returns_summary(self, monkeypatch): + import nerve.gateway.server as srv + + from nerve.gateway.routes.cron import reload_cron_jobs + + fake = MagicMock() + fake.reload = AsyncMock( + return_value={"added": ["a"], "removed": [], "updated": [], "enabled": 1} + ) + monkeypatch.setattr(srv, "_cron_service", fake, raising=False) + result = await reload_cron_jobs(user={}) + assert result["reloaded"] is True + assert result["added"] == ["a"] + + @pytest.mark.asyncio + async def test_400_on_config_error(self, monkeypatch): + import nerve.gateway.server as srv + from fastapi import HTTPException + + from nerve.config import ConfigError + from nerve.gateway.routes.cron import reload_cron_jobs + + fake = MagicMock() + fake.reload = AsyncMock(side_effect=ConfigError("bad cron file")) + monkeypatch.setattr(srv, "_cron_service", fake, raising=False) + with pytest.raises(HTTPException) as ei: + await reload_cron_jobs(user={}) + assert ei.value.status_code == 400 + assert "bad cron file" in ei.value.detail + + @pytest.mark.asyncio + async def test_400_on_invalid_schedule(self, monkeypatch): + """A typo'd schedule is a bad request, not a server error.""" + import nerve.gateway.server as srv + from fastapi import HTTPException + + from nerve.cron.service import InvalidScheduleError + from nerve.gateway.routes.cron import reload_cron_jobs + + fake = MagicMock() + fake.reload = AsyncMock( + side_effect=InvalidScheduleError("Cron job 'typo': bad minute"), + ) + monkeypatch.setattr(srv, "_cron_service", fake, raising=False) + with pytest.raises(HTTPException) as ei: + await reload_cron_jobs(user={}) + assert ei.value.status_code == 400 + assert "typo" in ei.value.detail + + +class TestReservedIds: + @pytest.mark.parametrize("job_id", [ + "cleanup", "wakeup_sweep", + "source:gmail", "source:telegram", "source:gmail:me@example.com", + "source:", + ]) + def test_reserved(self, job_id): + assert is_reserved_job_id(job_id) + + @pytest.mark.parametrize("job_id", [ + "cleanup-inbox", "my_wakeup_sweep", "sources:gmail", "source-gmail", + "morning-briefing", + ]) + def test_not_reserved(self, job_id): + assert not is_reserved_job_id(job_id) + + @pytest.mark.asyncio + async def test_start_drops_reserved_jobs(self, tmp_path, monkeypatch, caplog): + """Startup must reject reserved ids too, not just reload.""" + import nerve.sources.registry as registry + + monkeypatch.setattr(registry, "build_source_runners", lambda *a, **k: []) + + cron_dir = tmp_path / "cron" + jobs_file = cron_dir / "jobs.yaml" + _write_jobs(jobs_file, [ + _job_dict("cleanup", schedule="1h"), + _job_dict("source:gmail", schedule="1h"), + _job_dict("legit", schedule="1h"), + ]) + + config = MagicMock() + config.timezone = "UTC" + config.cron.jobs_file = jobs_file + config.cron.system_file = cron_dir / "system.yaml" # absent + config.cron.gate_plugins_dir = cron_dir / "gates" # absent → no-op + + db = AsyncMock() + db.get_last_successful_cron_run = AsyncMock(return_value=None) + + service = CronService(config, AsyncMock(), db) + with caplog.at_level(logging.WARNING, logger="nerve.cron.service"): + await service.start() + try: + assert [j.id for j in service._jobs] == ["legit"] + # No user job in the source namespace, and the daemon's own cleanup + # job — not the user's — owns the 'cleanup' slot. + assert service.scheduler.get_job("source:gmail") is None + assert service.scheduler.get_job("cleanup").name == "Cleanup expired data" + assert service.scheduler.get_job("legit") is not None + finally: + service.scheduler.shutdown(wait=False) + + logged = " ".join(r.getMessage() for r in caplog.records) + assert "cleanup" in logged and "source:gmail" in logged + + def test_source_runner_ids_are_always_inside_the_namespace(self): + """The reservation only holds if every runner really lives under it.""" + import inspect + + from nerve.sources.runner import SourceRunner + + # No caller may hand a runner an id outside the reserved namespace. + assert "job_id" not in inspect.signature(SourceRunner.__init__).parameters + + source = MagicMock() + source.source_name = "gmail:me@example.com" + runner = SourceRunner(source=source, db=AsyncMock()) + assert runner.job_id == "source:gmail:me@example.com" + assert is_reserved_job_id(runner.job_id) + + +class TestReservedIdsInCli: + """The daemon-log warning never reaches someone running the CLI.""" + + @staticmethod + def _config_dir(tmp_path, system_jobs, user_jobs): + cron_dir = tmp_path / "cron" + _write_jobs(cron_dir / "system.yaml", system_jobs) + _write_jobs(cron_dir / "jobs.yaml", user_jobs) + (tmp_path / "config.yaml").write_text( + f"cron:\n" + f" system_file: {cron_dir / 'system.yaml'}\n" + f" jobs_file: {cron_dir / 'jobs.yaml'}\n", + encoding="utf-8", + ) + return tmp_path + + def test_cron_listing_flags_reserved_jobs(self, tmp_path, monkeypatch): + from click.testing import CliRunner + + import nerve.agent.engine as engine_mod + import nerve.db as db_mod + from nerve.cli import main + + # The listing itself needs neither an engine nor a database. + monkeypatch.setattr(db_mod, "init_db", AsyncMock(return_value=AsyncMock())) + monkeypatch.setattr(db_mod, "close_db", AsyncMock()) + monkeypatch.setattr( + engine_mod, "AgentEngine", MagicMock(return_value=AsyncMock()), + ) + + cfg = self._config_dir( + tmp_path, [], [_job_dict("source:rss"), _job_dict("legit")], + ) + result = CliRunner().invoke(main, ["-c", str(cfg), "cron"]) + assert result.exit_code == 0, result.output + # The job is listed (so the reader can see it was read at all) but is + # never described as enabled. + assert "source:rss" in result.output + assert "RESERVED ID" in result.output + assert "source:rss: 1h (enabled)" not in result.output + assert "legit: 1h (enabled)" in result.output + + def test_doctor_excludes_reserved_from_the_enabled_count(self, tmp_path): + from click.testing import CliRunner + + from nerve.cli import main + + cfg = self._config_dir( + tmp_path, [], [_job_dict("source:rss"), _job_dict("legit")], + ) + # doctor exits non-zero on this bare config (no API key, no workspace); + # only its cron reporting is under test here. + result = CliRunner().invoke(main, ["-c", str(cfg), "doctor"]) + assert "Cron jobs: 1/1 enabled" in result.output + assert "source:rss" in result.output + assert "reserved" in result.output