fix(cron): guard legacy agent init kwargs

cron/scheduler.py

@@ -15,6 +15,7 @@ import logging
 import os
 import subprocess
 import sys
+import inspect
 
 # fcntl is Unix-only; on Windows use msvcrt for file locking
 try:
@@ -593,29 +594,51 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
             },
         )
 
-        agent = AIAgent(
-            model=turn_route["model"],
-            api_key=turn_route["runtime"].get("api_key"),
-            base_url=turn_route["runtime"].get("base_url"),
-            provider=turn_route["runtime"].get("provider"),
-            api_mode=turn_route["runtime"].get("api_mode"),
-            acp_command=turn_route["runtime"].get("command"),
-            acp_args=turn_route["runtime"].get("args"),
-            max_iterations=max_iterations,
-            reasoning_config=reasoning_config,
-            prefill_messages=prefill_messages,
-            providers_allowed=pr.get("only"),
-            providers_ignored=pr.get("ignore"),
-            providers_order=pr.get("order"),
-            provider_sort=pr.get("sort"),
-            disabled_toolsets=["cronjob", "messaging", "clarify"],
-            tool_choice="required",
-            quiet_mode=True,
-            skip_memory=True,  # Cron system prompts would corrupt user representations
-            platform="cron",
-            session_id=_cron_session_id,
-            session_db=_session_db,
-        )
+        agent_kwargs = {
+            "model": turn_route["model"],
+            "api_key": turn_route["runtime"].get("api_key"),
+            "base_url": turn_route["runtime"].get("base_url"),
+            "provider": turn_route["runtime"].get("provider"),
+            "api_mode": turn_route["runtime"].get("api_mode"),
+            "acp_command": turn_route["runtime"].get("command"),
+            "acp_args": turn_route["runtime"].get("args"),
+            "max_iterations": max_iterations,
+            "reasoning_config": reasoning_config,
+            "prefill_messages": prefill_messages,
+            "providers_allowed": pr.get("only"),
+            "providers_ignored": pr.get("ignore"),
+            "providers_order": pr.get("order"),
+            "provider_sort": pr.get("sort"),
+            "disabled_toolsets": ["cronjob", "messaging", "clarify"],
+            "tool_choice": "required",
+            "quiet_mode": True,
+            "skip_memory": True,  # Cron system prompts would corrupt user representations
+            "platform": "cron",
+            "session_id": _cron_session_id,
+            "session_db": _session_db,
+        }
+        try:
+            _agent_sig = inspect.signature(AIAgent.__init__)
+            _agent_params = _agent_sig.parameters
+            _accepts_var_kw = any(
+                p.kind == inspect.Parameter.VAR_KEYWORD
+                for p in _agent_params.values()
+            )
+            if not _accepts_var_kw:
+                _supported = {name for name in _agent_params if name != "self"}
+                _dropped = [key for key in list(agent_kwargs) if key not in _supported]
+                if _dropped:
+                    logger.warning(
+                        "Job '%s': dropping unsupported AIAgent kwargs for compatibility: %s",
+                        job_id,
+                        ", ".join(sorted(_dropped)),
+                    )
+                    for key in _dropped:
+                        agent_kwargs.pop(key, None)
+        except (TypeError, ValueError):
+            pass
+
+        agent = AIAgent(**agent_kwargs)
         
         # Run the agent with an *inactivity*-based timeout: the job can run
         # for hours if it's actively calling tools / receiving stream tokens,

tests/cron/test_scheduler.py

@@ -371,6 +371,76 @@ class TestRunJobSessionPersistence:
         assert call_args[0][1] == "cron_complete"
         fake_db.close.assert_called_once()
 
+    def test_run_job_drops_unsupported_agent_kwargs_for_legacy_agent_signature(self, tmp_path):
+        job = {
+            "id": "legacy-agent-job",
+            "name": "legacy",
+            "prompt": "hello",
+        }
+        fake_db = MagicMock()
+        seen = {}
+
+        class LegacyAgent:
+            def __init__(
+                self,
+                *,
+                model=None,
+                api_key=None,
+                base_url=None,
+                provider=None,
+                api_mode=None,
+                acp_command=None,
+                acp_args=None,
+                max_iterations=None,
+                reasoning_config=None,
+                prefill_messages=None,
+                providers_allowed=None,
+                providers_ignored=None,
+                providers_order=None,
+                provider_sort=None,
+                disabled_toolsets=None,
+                quiet_mode=None,
+                skip_memory=None,
+                platform=None,
+                session_id=None,
+                session_db=None,
+            ):
+                seen.update({
+                    "model": model,
+                    "platform": platform,
+                    "session_id": session_id,
+                    "session_db": session_db,
+                })
+
+            def run_conversation(self, *_args, **_kwargs):
+                return {"final_response": "ok"}
+
+        with patch("cron.scheduler._hermes_home", tmp_path), \
+             patch("cron.scheduler._resolve_origin", return_value=None), \
+             patch("dotenv.load_dotenv"), \
+             patch("hermes_state.SessionDB", return_value=fake_db), \
+             patch(
+                 "hermes_cli.runtime_provider.resolve_runtime_provider",
+                 return_value={
+                     "api_key": "***",
+                     "base_url": "https://example.invalid/v1",
+                     "provider": "openrouter",
+                     "api_mode": "chat_completions",
+                 },
+             ), \
+             patch("run_agent.AIAgent", LegacyAgent):
+            success, output, final_response, error = run_job(job)
+
+        assert success is True
+        assert error is None
+        assert final_response == "ok"
+        assert "ok" in output
+        assert seen["model"] is not None
+        assert seen["platform"] == "cron"
+        assert seen["session_id"].startswith("cron_legacy-agent-job_")
+        assert seen["session_db"] is fake_db
+        fake_db.close.assert_called_once()
+
     def test_run_job_empty_response_returns_empty_not_placeholder(self, tmp_path):
         """Empty final_response should stay empty for delivery logic (issue #2234).