diff --git a/docs/cron.md b/docs/cron.md index ab3b639c..93b81f8c 100644 --- a/docs/cron.md +++ b/docs/cron.md @@ -170,11 +170,77 @@ run_if: > `idle_consumer: ` fields still work — they are translated into an > equivalent `messages` gate at load time. Prefer `run_if` for new jobs. -### Adding a new gate type +### Adding a built-in gate type -Gates live in `nerve/cron/gates.py`. To add one: subclass `CronGate`, set its -`type`, implement `is_satisfied`, `describe`, and `from_config`, then register -the class in `GATE_REGISTRY`. It becomes usable from `run_if` immediately. +Built-in gates live in `nerve/cron/gates.py`. To add one: subclass `CronGate`, +set its `type`, implement `is_satisfied`, `describe`, and `from_config`, then +register the class in `GATE_REGISTRY`. It becomes usable from `run_if` +immediately. This is the right path for gates that ship with Nerve. + +### Custom gate plugins (drop-in) + +To add your **own** gate without editing core source, drop a `.py` file into +the gate-plugins directory — `~/.nerve/cron/gates/` by default (overridable via +the `cron.gate_plugins_dir` config key). On daemon startup Nerve imports each +file and registers every `CronGate` subclass it defines with a non-empty +`type`. After that, `run_if` can reference your gate by `type` exactly like a +built-in. Because this never touches `nerve/cron/gates.py`, your custom gates +don't conflict when you pull Nerve upstream. + +```python +# ~/.nerve/cron/gates/stale_tasks.py +from nerve.cron.gates import CronGate, GateContext + + +class StaleTasksGate(CronGate): + type = "stale_tasks" + + def __init__(self, min_age_minutes: int = 30): + self.min_age_minutes = min_age_minutes + + async def is_satisfied(self, ctx: GateContext) -> bool: + # ctx exposes {job_id, db} — DB-only (see note below). + ... + + def describe(self) -> str: + return f"stale tasks older than {self.min_age_minutes}m" + + @classmethod + def from_config(cls, spec: dict) -> "StaleTasksGate": + return cls(min_age_minutes=int(spec.get("min_age_minutes", 30))) +``` + +```yaml +# ~/.nerve/cron/jobs.yaml — reference it like any built-in gate +run_if: + - type: stale_tasks + min_age_minutes: 60 +``` + +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): + +- Files whose name starts with `_` (and `__pycache__`) are ignored. +- A plugin whose `type` collides with an already-registered gate is skipped + with a warning: a **built-in always wins**, and among two plugins the **first + 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. + +> **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 +> gate that needs live runtime state — e.g. which sessions are currently +> running — is **not** supported by this loader; that would require widening +> the gate context, a separate change. + +> **Trust note.** Files in the gate-plugins directory are imported (executed) +> at daemon startup. This is the same trust model as `config.yaml`, configured +> MCP servers, and cron prompt files — all user-controlled code/config the +> daemon already loads. Only place files you trust in this directory. ## Session Modes diff --git a/nerve/config.py b/nerve/config.py index b9a9bcbf..80c314d6 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -503,12 +503,16 @@ def from_dict(cls, d: dict) -> MemoryConfig: class CronConfig: jobs_file: Path = field(default_factory=lambda: Path("~/.nerve/cron/jobs.yaml")) system_file: Path = field(default_factory=lambda: Path("~/.nerve/cron/system.yaml")) + # Directory scanned at startup for drop-in custom gate plugins (.py files + # defining CronGate subclasses). See nerve/cron/gate_plugins.py. + gate_plugins_dir: Path = field(default_factory=lambda: Path("~/.nerve/cron/gates")) @classmethod def from_dict(cls, d: dict) -> CronConfig: return cls( jobs_file=_expand_path(d.get("jobs_file", "~/.nerve/cron/jobs.yaml")) or Path("~/.nerve/cron/jobs.yaml"), system_file=_expand_path(d.get("system_file", "~/.nerve/cron/system.yaml")) or Path("~/.nerve/cron/system.yaml"), + gate_plugins_dir=_expand_path(d.get("gate_plugins_dir", "~/.nerve/cron/gates")) or Path("~/.nerve/cron/gates"), ) diff --git a/nerve/cron/gate_plugins.py b/nerve/cron/gate_plugins.py new file mode 100644 index 00000000..fb55ce73 --- /dev/null +++ b/nerve/cron/gate_plugins.py @@ -0,0 +1,196 @@ +"""Drop-in cron gate plugins — auto-register custom gates without editing core. + +Built-in gates live in :mod:`nerve.cron.gates` and are registered directly in +:data:`nerve.cron.gates.GATE_REGISTRY`. To add a *custom* gate **without +editing core source**, drop a ``.py`` file into the gate-plugins directory +(``~/.nerve/cron/gates/`` by default, overridable via the ``cron.gate_plugins_dir`` +config key). On daemon startup each file is imported and every +:class:`~nerve.cron.gates.CronGate` subclass it defines with a non-empty +``type`` is registered into ``GATE_REGISTRY`` — after which ``jobs.yaml`` can +reference it via ``run_if: [{type: , ...}]`` exactly like a built-in. + +Because the loader never edits ``gates.py``, custom gates don't conflict when +pulling Nerve upstream; the loader itself is generic and upstreamable. + +A plugin file looks like any other module defining a gate:: + + # ~/.nerve/cron/gates/stale_tasks.py + from nerve.cron.gates import CronGate, GateContext + + class StaleTasksGate(CronGate): + type = "stale_tasks" + + def __init__(self, min_age_minutes: int = 30): + self.min_age_minutes = min_age_minutes + + async def is_satisfied(self, ctx: GateContext) -> bool: + ... # ctx gives {job_id, db} — DB-only + + def describe(self) -> str: + return f"stale tasks older than {self.min_age_minutes}m" + + @classmethod + def from_config(cls, spec: dict) -> "StaleTasksGate": + return cls(min_age_minutes=int(spec.get("min_age_minutes", 30))) + +Rules (all fail-safe — a bad plugin never crashes the daemon): + +* Files whose name starts with ``_`` (and ``__pycache__``) are skipped. +* A plugin ``type`` that collides with an already-registered gate is skipped + with a warning: a **built-in always wins**, and among two plugins the + **first loaded (filename-sorted) wins**. +* Any import/exec error in a plugin file is logged (naming the file) and the + file is skipped; the remaining files still load. +* 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. + +**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``, +configured MCP servers, and cron prompt files — all user-controlled code/config +the daemon already loads. Only place files you trust in this directory. +""" + +from __future__ import annotations + +import importlib.util +import inspect +import logging +from pathlib import Path + +from nerve.cron.gates import GATE_REGISTRY, CronGate + +logger = logging.getLogger(__name__) + + +def load_gate_plugins(plugins_dir: Path) -> 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``). + + 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). + """ + try: + plugins_dir = Path(plugins_dir).expanduser() + except Exception as e: # noqa: BLE001 — defensive; never block startup + logger.warning("Invalid cron gate plugins dir %r: %s", plugins_dir, e) + return 0 + + if not plugins_dir.is_dir(): + # Missing dir is the normal case — most installs have no custom gates. + return 0 + + registered = 0 + for path in sorted(plugins_dir.glob("*.py")): + if path.name.startswith("_") or not path.is_file(): + # Skip private modules and the odd case of a directory named "*.py". + continue + registered += _load_file(path) + + if registered: + logger.info( + "Loaded %d custom cron gate(s) from %s", registered, plugins_dir, + ) + return registered + + +def _load_file(path: Path) -> int: + """Import one plugin file and register its gate classes. Returns the count.""" + module = _import_module(path) + if module is None: + return 0 + + count = 0 + for name, obj in inspect.getmembers(module, inspect.isclass): + # Only classes *defined in this file* — skip imported symbols such as + # the CronGate base itself or any built-in gate the plugin imported. + if obj.__module__ != module.__name__: + continue + if not issubclass(obj, CronGate) or obj is CronGate: + continue + gate_type = getattr(obj, "type", "") or "" + if not gate_type: + logger.warning( + "Cron gate plugin %s: class %s has an empty 'type'; skipping", + path.name, name, + ) + continue + if inspect.isabstract(obj): + # A typed but still-abstract gate imports cleanly, yet raises + # TypeError when instantiated (in from_config). build_gates only + # catches GateConfigError, so that TypeError would propagate out of + # CronJob construction and crash daemon startup. Refuse it here, + # where the failure is contained, instead. + logger.warning( + "Cron gate plugin %s: gate %s (type %r) is abstract — " + "missing %s; skipping", + path.name, name, gate_type, + ", ".join(sorted(obj.__abstractmethods__)), + ) + continue + if _register(gate_type, obj, path): + count += 1 + return count + + +def _import_module(path: Path): + """Load a ``.py`` file as an isolated module. Returns ``None`` on any error. + + The module is intentionally not inserted into ``sys.modules`` — it is a + 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}" + try: + spec = importlib.util.spec_from_file_location(mod_name, path) + if spec is None or spec.loader is None: + logger.warning( + "Cron gate plugin %s: could not create an import spec; skipping", + path.name, + ) + return None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + except (Exception, SystemExit) as e: # noqa: BLE001 + # A bad plugin must never crash startup. SystemExit (e.g. a stray + # sys.exit() at import) subclasses BaseException, not Exception, so it + # is caught explicitly. KeyboardInterrupt is deliberately NOT caught, + # so an operator can still Ctrl-C a hung import. + logger.warning("Cron gate plugin %s failed to load: %s", path.name, e) + return None + + +def _register(gate_type: str, cls: type[CronGate], path: Path) -> bool: + """Register *cls* under *gate_type* unless it collides. Returns True if added. + + A collision keeps the incumbent (a built-in, or the first plugin loaded by + filename order) and warns. Re-importing the *same* plugin class — a fresh + object with an identical module+qualname, e.g. if the loader is ever run a + second time in one process — is an idempotent no-op rather than a (noisy, + spurious) collision, since each import produces a brand-new class object. + """ + existing = GATE_REGISTRY.get(gate_type) + if existing is not None: + is_same_plugin_class = ( + existing.__module__ == cls.__module__ + and existing.__qualname__ == cls.__qualname__ + ) + if not is_same_plugin_class: + logger.warning( + "Cron gate plugin %s: type %r already registered by %s; " + "keeping the existing gate and skipping the plugin", + path.name, gate_type, existing.__module__, + ) + return False + GATE_REGISTRY[gate_type] = cls + logger.info( + "Registered custom cron gate %r (%s) from %s", + gate_type, cls.__name__, path.name, + ) + return True diff --git a/nerve/cron/gates.py b/nerve/cron/gates.py index d0c4772b..3ff19c27 100644 --- a/nerve/cron/gates.py +++ b/nerve/cron/gates.py @@ -14,6 +14,8 @@ which mirrors the ``type:`` field in the YAML spec. * Adding a new gate = subclass :class:`CronGate`, set ``type``, implement the three abstract methods, and register the class in :data:`GATE_REGISTRY`. + (Out-of-tree gates can instead be dropped into the gate-plugins directory + and auto-registered at startup — see :mod:`nerve.cron.gate_plugins`.) Example config:: diff --git a/nerve/cron/service.py b/nerve/cron/service.py index f8dfd024..1c76c5ea 100644 --- a/nerve/cron/service.py +++ b/nerve/cron/service.py @@ -82,6 +82,13 @@ def __init__(self, config: NerveConfig, engine: AgentEngine, db: Database): async def start(self) -> None: """Load jobs and start the scheduler.""" + # Register drop-in custom gate plugins BEFORE jobs are parsed, so their + # `type` keys are present in GATE_REGISTRY when each job's run_if specs + # are built (CronJob builds its gates at construction time). + from nerve.cron.gate_plugins import load_gate_plugins + + load_gate_plugins(self.config.cron.gate_plugins_dir) + # Load job definitions from both files self._jobs = self._load_merged_jobs() diff --git a/tests/test_cron_gate_plugins.py b/tests/test_cron_gate_plugins.py new file mode 100644 index 00000000..b2e11820 --- /dev/null +++ b/tests/test_cron_gate_plugins.py @@ -0,0 +1,315 @@ +"""Tests for the drop-in cron gate plugin loader (nerve/cron/gate_plugins.py).""" + +from __future__ import annotations + +import logging +from pathlib import Path +from unittest.mock import AsyncMock + +import pytest + +from nerve.cron.gate_plugins import load_gate_plugins +from nerve.cron.gates import ( + GATE_REGISTRY, + CronGate, + GateContext, + build_gate, + evaluate_gates, +) +from nerve.cron.jobs import CronJob + + +# --------------------------------------------------------------------------- +# 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 + + +class AlwaysGate(CronGate): + type = "always_test" + + async def is_satisfied(self, ctx): + return True + + def describe(self): + return "always (test plugin)" + + @classmethod + def from_config(cls, spec): + return cls() +''' + + +# A plugin that imports cleanly but defines an *abstract* gate (it forgets +# is_satisfied/from_config). It must NOT be registered: instantiating it would +# raise TypeError, which build_gates does not catch — crashing job construction. +_ABSTRACT_PLUGIN = ''' +from nerve.cron.gates import CronGate + + +class HalfGate(CronGate): + type = "half_test" + + def describe(self): + return "half" + # is_satisfied and from_config intentionally left unimplemented → abstract. +''' + +# A plugin that calls sys.exit() at import time. SystemExit is a BaseException +# (not Exception), so the loader must catch it explicitly or it would escape +# and crash daemon startup. +_SYS_EXIT_PLUGIN = "import sys\nsys.exit(1)\n" + + +def _write(dirpath: Path, name: str, body: str) -> Path: + p = dirpath / name + p.write_text(body, encoding="utf-8") + return p + + +def _ctx() -> GateContext: + return GateContext(job_id="j", db=AsyncMock()) + + +# --------------------------------------------------------------------------- +# Happy path +# --------------------------------------------------------------------------- + +class TestHappyPath: + def test_registers_and_builds(self, tmp_path, clean_registry): + _write(tmp_path, "always.py", _VALID_PLUGIN) + assert load_gate_plugins(tmp_path) == 1 + assert "always_test" in GATE_REGISTRY + gate = build_gate({"type": "always_test"}) + assert isinstance(gate, CronGate) + assert gate.type == "always_test" + + @pytest.mark.asyncio + async def test_loaded_gate_evaluates(self, tmp_path, clean_registry): + _write(tmp_path, "always.py", _VALID_PLUGIN) + load_gate_plugins(tmp_path) + gate = build_gate({"type": "always_test"}) + decision = await evaluate_gates([gate], _ctx()) + assert decision.should_run is True + + def test_multiple_gates_in_one_file(self, tmp_path, clean_registry): + body = _VALID_PLUGIN + ''' + +class AlwaysGate2(CronGate): + type = "always_test_2" + + async def is_satisfied(self, ctx): + return True + + def describe(self): + return "always 2" + + @classmethod + def from_config(cls, spec): + return cls() +''' + _write(tmp_path, "multi.py", body) + assert load_gate_plugins(tmp_path) == 2 + assert {"always_test", "always_test_2"} <= set(GATE_REGISTRY) + + +# --------------------------------------------------------------------------- +# Fail-safe isolation +# --------------------------------------------------------------------------- + +class TestFailSafe: + def test_broken_plugin_does_not_block_valid_one( + self, tmp_path, clean_registry, caplog, + ): + _write(tmp_path, "broken.py", "this is not valid python !!!\n") + _write(tmp_path, "good.py", _VALID_PLUGIN) + with caplog.at_level(logging.WARNING): + n = load_gate_plugins(tmp_path) + assert n == 1 # only the good one + assert "always_test" in GATE_REGISTRY + assert "broken.py" in caplog.text # the failure named the file + + def test_import_error_at_module_level_isolated( + self, tmp_path, clean_registry, caplog, + ): + _write(tmp_path, "raises.py", "raise RuntimeError('boom at import')\n") + _write(tmp_path, "good.py", _VALID_PLUGIN) + with caplog.at_level(logging.WARNING): + n = load_gate_plugins(tmp_path) + assert n == 1 + assert "raises.py" in caplog.text + + def test_file_without_crongate_is_skipped(self, tmp_path, clean_registry): + _write(tmp_path, "nogate.py", "x = 1\ndef helper():\n return 2\n") + assert load_gate_plugins(tmp_path) == 0 + + def test_abstract_gate_not_registered(self, tmp_path, clean_registry, caplog): + # A typed-but-abstract gate imports fine but must not be registered — + # registering it would defer a TypeError crash to job-build time. + _write(tmp_path, "half.py", _ABSTRACT_PLUGIN) + _write(tmp_path, "good.py", _VALID_PLUGIN) + with caplog.at_level(logging.WARNING): + n = load_gate_plugins(tmp_path) + assert n == 1 # only the concrete gate + assert "half_test" not in GATE_REGISTRY + 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). + _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 == [] + + def test_sys_exit_at_import_is_contained( + self, tmp_path, clean_registry, caplog, + ): + # sys.exit() raises SystemExit (a BaseException). The loader must catch + # it — this call must NOT raise — and the valid plugin must still load. + _write(tmp_path, "exiter.py", _SYS_EXIT_PLUGIN) + _write(tmp_path, "good.py", _VALID_PLUGIN) + with caplog.at_level(logging.WARNING): + n = load_gate_plugins(tmp_path) # must not raise SystemExit + assert n == 1 + assert "always_test" in GATE_REGISTRY + assert "exiter.py" in caplog.text + + def test_empty_type_is_skipped(self, tmp_path, clean_registry, caplog): + body = _VALID_PLUGIN.replace('type = "always_test"', 'type = ""') + _write(tmp_path, "notype.py", body) + with caplog.at_level(logging.WARNING): + assert load_gate_plugins(tmp_path) == 0 + assert "always_test" not in GATE_REGISTRY + assert "notype.py" in caplog.text + + def test_underscore_prefixed_file_ignored(self, tmp_path, clean_registry): + _write(tmp_path, "_helper.py", _VALID_PLUGIN) + assert load_gate_plugins(tmp_path) == 0 + assert "always_test" not in GATE_REGISTRY + + def test_non_py_files_ignored(self, tmp_path, clean_registry): + _write(tmp_path, "always.txt", _VALID_PLUGIN) + _write(tmp_path, "readme.md", "# not a plugin\n") + assert load_gate_plugins(tmp_path) == 0 + + +# --------------------------------------------------------------------------- +# Collisions +# --------------------------------------------------------------------------- + +class TestCollisions: + def test_builtin_collision_keeps_builtin( + self, tmp_path, clean_registry, caplog, + ): + # A plugin claiming the built-in "tasks" type must not override it. + body = ( + _VALID_PLUGIN + .replace('"always_test"', '"tasks"') + .replace("AlwaysGate", "FakeTasksGate") + ) + _write(tmp_path, "collide.py", body) + before = GATE_REGISTRY["tasks"] + with caplog.at_level(logging.WARNING): + n = load_gate_plugins(tmp_path) + assert n == 0 + assert GATE_REGISTRY["tasks"] is before # built-in retained + assert "tasks" in caplog.text + + def test_two_plugins_same_type_first_wins( + self, tmp_path, clean_registry, caplog, + ): + first = _VALID_PLUGIN.replace("AlwaysGate", "FirstGate") + second = ( + _VALID_PLUGIN + .replace("AlwaysGate", "SecondGate") + .replace('"always (test plugin)"', '"second"') + ) + # Filenames sort so a_*.py loads before b_*.py → FirstGate wins. + _write(tmp_path, "a_first.py", first) + _write(tmp_path, "b_second.py", second) + with caplog.at_level(logging.WARNING): + n = load_gate_plugins(tmp_path) + assert n == 1 + assert GATE_REGISTRY["always_test"].__name__ == "FirstGate" + + +# --------------------------------------------------------------------------- +# Empty / missing directory +# --------------------------------------------------------------------------- + +class TestNoOpDirs: + def test_missing_dir_returns_zero(self, tmp_path, clean_registry): + assert load_gate_plugins(tmp_path / "does_not_exist") == 0 + + def test_empty_dir_returns_zero(self, tmp_path, clean_registry): + assert load_gate_plugins(tmp_path) == 0 + + def test_file_path_instead_of_dir_returns_zero(self, tmp_path, clean_registry): + f = _write(tmp_path, "always.py", _VALID_PLUGIN) + # Pointing at a file (not a dir) is treated as "no dir" — no crash. + assert load_gate_plugins(f) == 0 + + def test_tilde_path_expanded(self, tmp_path, clean_registry, monkeypatch): + # A "~/..." path is expanded; a non-existent one is a no-op (no raise). + monkeypatch.setenv("HOME", str(tmp_path)) + assert load_gate_plugins(Path("~/nope/gates")) == 0 + + +# --------------------------------------------------------------------------- +# End-to-end via CronJob.run_if +# --------------------------------------------------------------------------- + +class TestEndToEndViaConfig: + def test_cronjob_run_if_builds_plugin_gate(self, tmp_path, clean_registry): + _write(tmp_path, "always.py", _VALID_PLUGIN) + load_gate_plugins(tmp_path) + job = CronJob( + id="j", schedule="1h", prompt="p", + run_if=[{"type": "always_test"}], + ) + assert len(job.gates) == 1 + assert job.gates[0].type == "always_test" + + @pytest.mark.asyncio + async def test_cronjob_plugin_gate_evaluates(self, tmp_path, clean_registry): + _write(tmp_path, "always.py", _VALID_PLUGIN) + load_gate_plugins(tmp_path) + job = CronJob( + id="j", schedule="1h", prompt="p", + run_if=[{"type": "always_test"}], + ) + 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 == []