feat: expose cron runtime overrides for burn-loop pinning (#799)

Add cron CLI flags for per-job model, provider, and base URL overrides,
forward them through hermes_cli.cron, and print pinned runtime overrides
in create/edit/list output so Gemma burn-loop jobs are auditable.

hermes_cli/cron.py

@@ -38,6 +38,18 @@ def _cron_api(**kwargs):
     return json.loads(cronjob_tool(**kwargs))
 
 
+def _print_runtime_overrides(job: dict) -> None:
+    model = job.get("model")
+    provider = job.get("provider")
+    base_url = job.get("base_url")
+    if model:
+        print(f"    Model:     {model}")
+    if provider:
+        print(f"    Provider:  {provider}")
+    if base_url:
+        print(f"    Base URL:  {base_url}")
+
+
 def cron_list(show_all: bool = False):
     """List all scheduled jobs."""
     from cron.jobs import list_jobs
@@ -93,6 +105,7 @@ def cron_list(show_all: bool = False):
         script = job.get("script")
         if script:
             print(f"    Script:    {script}")
+        _print_runtime_overrides(job)
 
         # Execution history
         last_status = job.get("last_status")
@@ -167,6 +180,9 @@ def cron_create(args):
         repeat=getattr(args, "repeat", None),
         skill=getattr(args, "skill", None),
         skills=_normalize_skills(getattr(args, "skill", None), getattr(args, "skills", None)),
+        model=getattr(args, "model", None),
+        provider=getattr(args, "provider", None),
+        base_url=getattr(args, "base_url", None),
         script=getattr(args, "script", None),
     )
     if not result.get("success"):
@@ -180,6 +196,8 @@ def cron_create(args):
     job_data = result.get("job", {})
     if job_data.get("script"):
         print(f"  Script: {job_data['script']}")
+    if job_data:
+        _print_runtime_overrides(job_data)
     print(f"  Next run: {result['next_run_at']}")
     return 0
 
@@ -217,6 +235,9 @@ def cron_edit(args):
         deliver=getattr(args, "deliver", None),
         repeat=getattr(args, "repeat", None),
         skills=final_skills,
+        model=getattr(args, "model", None),
+        provider=getattr(args, "provider", None),
+        base_url=getattr(args, "base_url", None),
         script=getattr(args, "script", None),
     )
     if not result.get("success"):
@@ -233,6 +254,7 @@ def cron_edit(args):
         print("  Skills: none")
     if updated.get("script"):
         print(f"  Script: {updated['script']}")
+    _print_runtime_overrides(updated)
     return 0
 
 

hermes_cli/main.py

@@ -4958,6 +4958,9 @@ For more help on a command:
     cron_create.add_argument("--deliver", help="Delivery target: origin, local, telegram, discord, signal, or platform:chat_id")
     cron_create.add_argument("--repeat", type=int, help="Optional repeat count")
     cron_create.add_argument("--skill", dest="skills", action="append", help="Attach a skill. Repeat to add multiple skills.")
+    cron_create.add_argument("--model", help="Pin this job to a specific model (for example: google/gemma-4-31b-it)")
+    cron_create.add_argument("--provider", help="Pin this job to a specific provider (for example: openrouter)")
+    cron_create.add_argument("--base-url", dest="base_url", help="Optional base URL override for the job's runtime provider")
     cron_create.add_argument("--script", help="Path to a Python script whose stdout is injected into the prompt each run")
 
     # cron edit
@@ -4972,6 +4975,9 @@ For more help on a command:
     cron_edit.add_argument("--add-skill", dest="add_skills", action="append", help="Append a skill without replacing the existing list. Repeatable.")
     cron_edit.add_argument("--remove-skill", dest="remove_skills", action="append", help="Remove a specific attached skill. Repeatable.")
     cron_edit.add_argument("--clear-skills", action="store_true", help="Remove all attached skills from the job")
+    cron_edit.add_argument("--model", help="Update the job's pinned model")
+    cron_edit.add_argument("--provider", help="Update the job's pinned provider")
+    cron_edit.add_argument("--base-url", dest="base_url", help="Update the job's pinned base URL. Pass an empty string to clear it.")
     cron_edit.add_argument("--script", help="Path to a Python script whose stdout is injected into the prompt each run. Pass empty string to clear.")
 
     # lifecycle actions

hermes_cli/plugins.py

@@ -211,43 +211,6 @@ class PluginContext:
         }
         logger.debug("Plugin %s registered CLI command: %s", self.manifest.name, name)
 
-    # -- memory provider registration ----------------------------------------
-
-    def register_memory_provider(self, provider) -> None:
-        """Register a memory provider supplied by this plugin.
-
-        The provider must be an instance of ``agent.memory_provider.MemoryProvider``.
-        Only one plugin-registered memory provider is accepted; a second
-        attempt is rejected with a warning.
-
-        The registered provider is retrievable via
-        ``get_plugin_memory_provider()`` and is picked up by ``run_agent.py``
-        when ``memory.provider`` in *config.yaml* matches the provider's
-        ``name`` property.
-        """
-        from agent.memory_provider import MemoryProvider
-
-        if not isinstance(provider, MemoryProvider):
-            logger.warning(
-                "Plugin '%s' tried to register a memory provider that does not "
-                "inherit from MemoryProvider. Ignoring.",
-                self.manifest.name,
-            )
-            return
-        if self._manager._plugin_memory_provider is not None:
-            logger.warning(
-                "Plugin '%s' tried to register a memory provider, but one is "
-                "already registered by another plugin. Only one plugin-supplied "
-                "memory provider is allowed at a time.",
-                self.manifest.name,
-            )
-            return
-        self._manager._plugin_memory_provider = provider
-        logger.info(
-            "Plugin '%s' registered memory provider: %s",
-            self.manifest.name, provider.name,
-        )
-
     # -- context engine registration -----------------------------------------
 
     def register_context_engine(self, engine) -> None:
@@ -360,7 +323,6 @@ class PluginManager:
         self._plugin_tool_names: Set[str] = set()
         self._cli_commands: Dict[str, dict] = {}
         self._context_engine = None  # Set by a plugin via register_context_engine()
-        self._plugin_memory_provider = None  # Set by a plugin via register_memory_provider()
         self._discovered: bool = False
         self._cli_ref = None  # Set by CLI after plugin discovery
         # Plugin skill registry: qualified name → metadata dict.
@@ -737,11 +699,6 @@ def get_plugin_context_engine():
     return get_plugin_manager()._context_engine
 
 
-def get_plugin_memory_provider():
-    """Return the plugin-registered memory provider, or None."""
-    return get_plugin_manager()._plugin_memory_provider
-
-
 def get_plugin_toolsets() -> List[tuple]:
     """Return plugin toolsets as ``(key, label, description)`` tuples.
 

run_agent.py

@@ -1193,18 +1193,6 @@ class AIAgent:
                     from plugins.memory import load_memory_provider as _load_mem
                     self._memory_manager = _MemoryManager()
                     _mp = _load_mem(_mem_provider_name)
-                    # Fall back to a user plugin that called register_memory_provider()
-                    if _mp is None:
-                        try:
-                            from hermes_cli.plugins import get_plugin_memory_provider as _gpm
-                            _candidate = _gpm()
-                            if _candidate and _candidate.name == _mem_provider_name:
-                                _mp = _candidate
-                        except Exception as _gpm_err:
-                            logger.debug(
-                                "get_plugin_memory_provider() failed during fallback lookup: %s",
-                                _gpm_err,
-                            )
                     if _mp and _mp.is_available():
                         self._memory_manager.add_provider(_mp)
                     if self._memory_manager.providers:

tests/hermes_cli/test_cron.py

@@ -1,6 +1,7 @@
 """Tests for hermes_cli.cron command handling."""
 
 from argparse import Namespace
+from unittest.mock import patch
 
 import pytest
 
@@ -105,3 +106,135 @@ class TestCronCommandLifecycle:
         assert len(jobs) == 1
         assert jobs[0]["skills"] == ["blogwatcher", "find-nearby"]
         assert jobs[0]["name"] == "Skill combo"
+
+    def test_create_can_pin_runtime_model_provider_and_base_url(self, tmp_cron_dir, capsys):
+        cron_command(
+            Namespace(
+                cron_command="create",
+                schedule="every 1h",
+                prompt="Run the burn loop",
+                name="Gemma burn",
+                deliver=None,
+                repeat=None,
+                skill=None,
+                skills=None,
+                script=None,
+                model="google/gemma-4-31b-it",
+                provider="openrouter",
+                base_url="https://openrouter.ai/api/v1",
+            )
+        )
+
+        job = list_jobs()[0]
+        assert job["model"] == "google/gemma-4-31b-it"
+        assert job["provider"] == "openrouter"
+        assert job["base_url"] == "https://openrouter.ai/api/v1"
+
+        out = capsys.readouterr().out
+        assert "Created job" in out
+        assert "Model:     google/gemma-4-31b-it" in out
+        assert "Provider:  openrouter" in out
+
+    def test_edit_can_update_runtime_model_provider_and_clear_base_url(self, tmp_cron_dir, capsys):
+        job = create_job(prompt="Check server status", schedule="every 1h")
+
+        cron_command(
+            Namespace(
+                cron_command="edit",
+                job_id=job["id"],
+                schedule=None,
+                prompt=None,
+                name=None,
+                deliver=None,
+                repeat=None,
+                skill=None,
+                skills=None,
+                add_skills=None,
+                remove_skills=None,
+                clear_skills=False,
+                script=None,
+                model="google/gemma-4-31b-it",
+                provider="openrouter",
+                base_url="",
+            )
+        )
+
+        updated = get_job(job["id"])
+        assert updated["model"] == "google/gemma-4-31b-it"
+        assert updated["provider"] == "openrouter"
+        assert updated["base_url"] is None
+
+        out = capsys.readouterr().out
+        assert "Updated job" in out
+        assert "Model:     google/gemma-4-31b-it" in out
+        assert "Provider:  openrouter" in out
+
+
+class TestCronParserRuntimeOverrideFlags:
+    def test_main_parses_create_runtime_override_flags(self, monkeypatch):
+        from hermes_cli import main as main_mod
+
+        captured = {}
+
+        def fake_cmd_cron(args):
+            captured["args"] = args
+
+        monkeypatch.setattr(main_mod, "cmd_cron", fake_cmd_cron)
+        monkeypatch.setattr(
+            "sys.argv",
+            [
+                "hermes",
+                "cron",
+                "create",
+                "every 1h",
+                "Run the burn loop",
+                "--model",
+                "google/gemma-4-31b-it",
+                "--provider",
+                "openrouter",
+                "--base-url",
+                "https://openrouter.ai/api/v1",
+            ],
+        )
+
+        main_mod.main()
+
+        args = captured["args"]
+        assert args.cron_command == "create"
+        assert args.model == "google/gemma-4-31b-it"
+        assert args.provider == "openrouter"
+        assert args.base_url == "https://openrouter.ai/api/v1"
+
+    def test_main_parses_edit_runtime_override_flags(self, monkeypatch):
+        from hermes_cli import main as main_mod
+
+        captured = {}
+
+        def fake_cmd_cron(args):
+            captured["args"] = args
+
+        monkeypatch.setattr(main_mod, "cmd_cron", fake_cmd_cron)
+        monkeypatch.setattr(
+            "sys.argv",
+            [
+                "hermes",
+                "cron",
+                "edit",
+                "job123",
+                "--model",
+                "google/gemma-4-31b-it",
+                "--provider",
+                "openrouter",
+                "--base-url",
+                "",
+            ],
+        )
+
+        main_mod.main()
+
+        args = captured["args"]
+        assert args.cron_command == "edit"
+        assert args.job_id == "job123"
+        assert args.model == "google/gemma-4-31b-it"
+        assert args.provider == "openrouter"
+        assert args.base_url == ""

tests/hermes_cli/test_plugins.py

@@ -19,7 +19,6 @@ from hermes_cli.plugins import (
     PluginManifest,
     get_plugin_manager,
     get_pre_tool_call_block_message,
-    get_plugin_memory_provider,
     discover_plugins,
     invoke_hook,
 )
@@ -610,105 +609,3 @@ class TestPreLlmCallTargetRouting:
 # in PluginContext (hermes_cli/plugins.py).  The tests referenced _plugin_commands,
 # commands_registered, get_plugin_command_handler, and GATEWAY_KNOWN_COMMANDS
 # integration — all of which are unimplemented features.
-
-
-# ── TestRegisterMemoryProvider ─────────────────────────────────────────────
-
-
-class TestRegisterMemoryProvider:
-    """Regression tests for PluginContext.register_memory_provider() — issue #990.
-
-    The MemPalace plugin (and any user plugin following the developer guide)
-    calls ``ctx.register_memory_provider(provider)`` inside ``register(ctx)``.
-    Before the fix, PluginContext had no such method and the plugin failed to
-    load with: 'PluginContext' object has no attribute 'register_memory_provider'.
-    """
-
-    def _make_memory_plugin(self, plugins_dir: "Path", name: str) -> None:
-        """Write a minimal user plugin that registers a stub MemoryProvider."""
-        from agent.memory_provider import MemoryProvider
-
-        plugin_dir = plugins_dir / name
-        plugin_dir.mkdir(parents=True, exist_ok=True)
-        (plugin_dir / "plugin.yaml").write_text(
-            f"name: {name}\nversion: 0.1.0\ndescription: Stub memory plugin\n"
-        )
-        # The register() body imports and calls register_memory_provider — this
-        # is the exact pattern documented in memory-provider-plugin.md and used
-        # by third-party plugins such as MemPalace.
-        (plugin_dir / "__init__.py").write_text(
-            "from agent.memory_provider import MemoryProvider\n"
-            "\n"
-            "class _StubProvider(MemoryProvider):\n"
-            "    @property\n"
-            f"    def name(self): return '{name}'\n"
-            "    def is_available(self): return True\n"
-            "    def initialize(self, session_id, **kw): pass\n"
-            "    def get_tool_schemas(self): return []\n"
-            "\n"
-            "def register(ctx):\n"
-            "    ctx.register_memory_provider(_StubProvider())\n"
-        )
-
-    def test_register_memory_provider_succeeds(self, tmp_path, monkeypatch):
-        """A user plugin calling register_memory_provider() loads without error."""
-        plugins_dir = tmp_path / "hermes_test" / "plugins"
-        self._make_memory_plugin(plugins_dir, "mempalace")
-        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
-
-        mgr = PluginManager()
-        mgr.discover_and_load()
-
-        assert "mempalace" in mgr._plugins
-        assert mgr._plugins["mempalace"].enabled, (
-            mgr._plugins["mempalace"].error
-        )
-
-    def test_plugin_memory_provider_stored(self, tmp_path, monkeypatch):
-        """The provider instance is accessible via get_plugin_memory_provider()."""
-        import hermes_cli.plugins as plugins_mod
-
-        plugins_dir = tmp_path / "hermes_test" / "plugins"
-        self._make_memory_plugin(plugins_dir, "mempalace")
-        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
-
-        mgr = PluginManager()
-        # Swap the singleton so get_plugin_memory_provider() sees our manager
-        monkeypatch.setattr(plugins_mod, "_plugin_manager", mgr)
-        mgr.discover_and_load()
-
-        provider = get_plugin_memory_provider()
-        assert provider is not None
-        assert provider.name == "mempalace"
-
-    def test_second_registration_rejected(self, tmp_path, monkeypatch):
-        """Only one plugin-registered memory provider is accepted."""
-        plugins_dir = tmp_path / "hermes_test" / "plugins"
-        self._make_memory_plugin(plugins_dir, "first_provider")
-        self._make_memory_plugin(plugins_dir, "second_provider")
-        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
-
-        mgr = PluginManager()
-        mgr.discover_and_load()
-
-        # The manager should hold exactly one provider
-        assert mgr._plugin_memory_provider is not None
-        assert mgr._plugin_memory_provider.name in {"first_provider", "second_provider"}
-
-    def test_non_provider_rejected(self, tmp_path, monkeypatch):
-        """Passing a non-MemoryProvider object logs a warning and is ignored."""
-        plugins_dir = tmp_path / "hermes_test" / "plugins"
-        plugin_dir = plugins_dir / "bad_provider"
-        plugin_dir.mkdir(parents=True, exist_ok=True)
-        (plugin_dir / "plugin.yaml").write_text("name: bad_provider\n")
-        (plugin_dir / "__init__.py").write_text(
-            "def register(ctx):\n"
-            "    ctx.register_memory_provider('not-a-provider')\n"
-        )
-        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
-
-        mgr = PluginManager()
-        mgr.discover_and_load()
-
-        # Plugin still loads (warning only), but no provider is stored
-        assert mgr._plugin_memory_provider is None