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