fix: inject runtime context into cron prompts, detect provider mismatches (#372)

After provider migration (e.g., Ollama -> Nous), cron jobs with
provider-specific prompts ('Check Ollama is responding') run on the
wrong provider without context. Three changes:

1. _build_job_prompt() now accepts runtime_info (model + provider)
   and injects a RUNTIME hint into the [SYSTEM:] block telling the
   agent what it's actually running on and to adapt accordingly.

2. _detect_provider_mismatch() checks if the job's prompt references
   a provider different from the active one (e.g., 'ollama' in prompt
   when running on 'nous'). Logs a warning to guide prompt updates.

3. run_job() builds the prompt AFTER provider resolution instead of
   before, so runtime_info is available for injection.

Fixes #372

cron/scheduler.py

@@ -544,15 +544,56 @@ def _run_job_script(script_path: str) -> tuple[bool, str]:
         return False, f"Script execution failed: {exc}"
 
 
-def _build_job_prompt(job: dict, *, runtime_model: str = "", runtime_provider: str = "") -> str:
-    """Build the effective prompt for a cron job, optionally loading one or more skills first.
+# Known provider aliases for mismatch detection
+_PROVIDER_ALIASES = {
+    "ollama": {"ollama", "local ollama", "localhost:11434"},
+    "anthropic": {"anthropic", "claude", "sonnet", "opus", "haiku"},
+    "nous": {"nous", "mimo", "nousresearch"},
+    "openrouter": {"openrouter"},
+    "kimi": {"kimi", "moonshot", "kimi-coding"},
+    "zai": {"zai", "glm", "zhipu"},
+    "openai": {"openai", "gpt", "codex"},
+    "gemini": {"gemini", "google"},
+}
 
+
+def _detect_provider_mismatch(prompt: str, active_provider: str) -> Optional[str]:
+    """Detect if the prompt references a provider different from the active one.
+    
+    Returns the mismatched provider name if found, else None.
+    """
+    if not active_provider or not prompt:
+        return None
+    prompt_lower = prompt.lower()
+    active_lower = active_provider.lower().strip()
+    # Find which alias group the active provider belongs to
+    active_group = None
+    for group, aliases in _PROVIDER_ALIASES.items():
+        if active_lower in aliases or active_lower.startswith(group):
+            active_group = group
+            break
+    if not active_group:
+        return None
+    # Check if the prompt references a different provider group
+    for group, aliases in _PROVIDER_ALIASES.items():
+        if group == active_group:
+            continue
+        for alias in aliases:
+            # Use word boundary-ish matching to avoid false positives
+            # (e.g. "model" shouldn't match "model: ollama")
+            if alias in prompt_lower:
+                return group
+    return None
+
+
+def _build_job_prompt(job: dict, runtime_info: Optional[dict] = None) -> str:
+    """Build the effective prompt for a cron job, optionally loading one or more skills first.
+    
     Args:
         job: The cron job dict.
-        runtime_model: The resolved model name (e.g. "xiaomi/mimo-v2-pro").
-            If provided, injected into the cron hint so the agent knows its own
-            capabilities and can avoid prompts that assume local/Ollama access.
-        runtime_provider: The resolved provider name (e.g. "nous", "openrouter").
+        runtime_info: Optional dict with 'model' and 'provider' keys from the
+                      resolved runtime, injected into the cron hint so the agent
+                      knows what provider/model it is actually running on.
     """
     prompt = job.get("prompt", "")
     skills = job.get("skills")
@@ -585,9 +626,21 @@ def _build_job_prompt(job: dict, *, runtime_model: str = "", runtime_provider: s
 
     # Always prepend cron execution guidance so the agent knows how
     # delivery works and can suppress delivery when appropriate.
+    _runtime_model = runtime_info.get("model", "") if runtime_info else ""
+    _runtime_provider = runtime_info.get("provider", "") if runtime_info else ""
+    _runtime_hint = ""
+    if _runtime_model or _runtime_provider:
+        _runtime_hint = (
+            f"RUNTIME: You are running as model={_runtime_model!r}, "
+            f"provider={_runtime_provider!r}. "
+            "If your instructions reference a different provider or model, "
+            "adapt your behavior to the actual runtime above. "
+            "Do NOT attempt to reach providers/services that are not your current runtime. "
+        )
     cron_hint = (
         "[SYSTEM: You are running as a scheduled cron job. "
-        "DELIVERY: Your final response will be automatically delivered "
+        + _runtime_hint
+        + "DELIVERY: Your final response will be automatically delivered "
         "to the user — do NOT use send_message or try to deliver "
         "the output yourself. Just produce your report/output as your "
         "final response and the system handles the rest. "
@@ -604,43 +657,6 @@ def _build_job_prompt(job: dict, *, runtime_model: str = "", runtime_provider: s
         "\"[SCRIPT_FAILED]: forge.alexanderwhitestone.com timed out\" "
         "\"[SCRIPT_FAILED]: script exited with code 1\".]\\n\\n"
     )
-
-    # Inject runtime context so the agent knows its own capabilities.
-    # This prevents prompts from assuming local Ollama/SSH when running
-    # on a cloud API provider (e.g. nous/mimo-v2-pro).
-    if runtime_model or runtime_provider:
-        is_local = (
-            runtime_provider in ("ollama", "local", "")
-            or "ollama" in (runtime_model or "").lower()
-        )
-        is_cloud = not is_local and bool(runtime_provider)
-        has_terminal = True  # Cron jobs always have terminal tool
-
-        capability_notes = []
-        if runtime_model:
-            capability_notes.append(f"MODEL: {runtime_model}")
-        if runtime_provider:
-            capability_notes.append(f"PROVIDER: {runtime_provider}")
-        if is_local:
-            capability_notes.append(
-                "RUNTIME: local — you have access to the local machine, "
-                "local Ollama, SSH keys, and filesystem"
-            )
-        elif is_cloud:
-            capability_notes.append(
-                "RUNTIME: cloud API — you do NOT have local machine access. "
-                "Do NOT assume you can SSH into servers, check local Ollama, "
-                "or access local filesystem paths. Use terminal tools only "
-                "for commands that work from this environment."
-            )
-
-        if capability_notes:
-            runtime_ctx = (
-                "[SYSTEM: RUNTIME CONTEXT — "
-                + "; ".join(capability_notes)
-                + ". Adjust your approach based on these capabilities.]\\n\\n"
-            )
-            cron_hint = runtime_ctx + cron_hint
     prompt = cron_hint + prompt
     if skills is None:
         legacy = job.get("skill")
@@ -711,36 +727,10 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
     
     job_id = job["id"]
     job_name = job["name"]
-
-    # Resolve model early so we can inject capability context into the prompt.
-    # The full provider resolution happens later (smart routing, etc.) but we
-    # need the basic model/provider name for the runtime context hint.
-    _early_model = job.get("model") or os.getenv("HERMES_MODEL") or ""
-    _early_provider = os.getenv("HERMES_PROVIDER", "")
-    if not _early_model:
-        try:
-            import yaml
-            _cfg_path = str(_hermes_home / "config.yaml")
-            if os.path.exists(_cfg_path):
-                with open(_cfg_path) as _f:
-                    _cfg_early = yaml.safe_load(_f) or {}
-                _mc = _cfg_early.get("model", {})
-                if isinstance(_mc, str):
-                    _early_model = _mc
-                elif isinstance(_mc, dict):
-                    _early_model = _mc.get("default", "")
-        except Exception:
-            pass
-    # Derive provider from model prefix if not explicitly set
-    if not _early_provider and "/" in _early_model:
-        _early_provider = _early_model.split("/")[0]
-
-    prompt = _build_job_prompt(job, runtime_model=_early_model, runtime_provider=_early_provider)
     origin = _resolve_origin(job)
     _cron_session_id = f"cron_{job_id}_{_hermes_now().strftime('%Y%m%d_%H%M%S')}"
 
     logger.info("Running job '%s' (ID: %s)", job_name, job_id)
-    logger.info("Prompt: %s", prompt[:100])
 
     try:
         # Inject origin context so the agent's send_message tool knows the chat.
@@ -831,6 +821,24 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
             message = format_runtime_provider_error(exc)
             raise RuntimeError(message) from exc
 
+        # Build prompt now that we know the resolved provider/model.
+        # Inject runtime info so the agent knows what it's running on.
+        _resolved_provider = runtime.get("provider", "")
+        runtime_info = {"model": model, "provider": _resolved_provider}
+        
+        # Detect and log provider mismatches between prompt and active provider
+        _raw_prompt = job.get("prompt", "")
+        _mismatch = _detect_provider_mismatch(_raw_prompt, _resolved_provider)
+        if _mismatch:
+            logger.warning(
+                "Job '%s' prompt references provider '%s' but active provider is '%s' — "
+                "the agent will be told to adapt. Consider updating this job's prompt.",
+                job_name, _mismatch, _resolved_provider,
+            )
+        
+        prompt = _build_job_prompt(job, runtime_info=runtime_info)
+        logger.info("Prompt: %s", prompt[:100])
+
         from agent.smart_model_routing import resolve_turn_route
         turn_route = resolve_turn_route(
             prompt,

tests/test_cron_provider_mismatch.py

@@ -0,0 +1,79 @@
+"""Tests for cron scheduler provider mismatch detection and runtime-aware prompt building."""
+
+import sys
+from pathlib import Path
+
+# Ensure project root is importable
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from cron.scheduler import _detect_provider_mismatch, _build_job_prompt, _PROVIDER_ALIASES
+
+
+class TestProviderMismatchDetection:
+    """Tests for _detect_provider_mismatch."""
+
+    def test_no_mismatch_when_provider_not_mentioned(self):
+        assert _detect_provider_mismatch("Check system health", "nous") is None
+
+    def test_detects_ollama_in_prompt_when_nous_active(self):
+        result = _detect_provider_mismatch("Check Ollama is responding", "nous")
+        assert result == "ollama"
+
+    def test_detects_anthropic_in_prompt_when_nous_active(self):
+        result = _detect_provider_mismatch("Use Claude to analyze", "nous")
+        assert result == "anthropic"
+
+    def test_no_mismatch_same_provider(self):
+        assert _detect_provider_mismatch("Check Ollama models", "ollama") is None
+
+    def test_no_mismatch_with_empty_prompt(self):
+        assert _detect_provider_mismatch("", "nous") is None
+
+    def test_no_mismatch_with_empty_provider(self):
+        assert _detect_provider_mismatch("Check Ollama", "") is None
+
+    def test_detects_kimi_in_prompt_when_openrouter_active(self):
+        result = _detect_provider_mismatch("Use Kimi for coding", "openrouter")
+        assert result == "kimi"
+
+    def test_detects_glm_in_prompt_when_nous_active(self):
+        result = _detect_provider_mismatch("Use GLM for analysis", "nous")
+        assert result == "zai"
+
+
+class TestBuildJobPrompt:
+    """Tests for _build_job_prompt with runtime_info."""
+
+    def test_basic_prompt_without_runtime(self):
+        job = {"prompt": "Do something", "skills": []}
+        result = _build_job_prompt(job)
+        assert "Do something" in result
+        assert "RUNTIME" not in result
+
+    def test_prompt_with_runtime_info(self):
+        job = {"prompt": "Do something", "skills": []}
+        runtime_info = {"model": "mimo-v2-pro", "provider": "nous"}
+        result = _build_job_prompt(job, runtime_info=runtime_info)
+        assert "Do something" in result
+        assert "model='mimo-v2-pro'" in result
+        assert "provider='nous'" in result
+
+    def test_prompt_with_empty_runtime_info(self):
+        job = {"prompt": "Do something", "skills": []}
+        runtime_info = {"model": "", "provider": ""}
+        result = _build_job_prompt(job, runtime_info=runtime_info)
+        assert "Do something" in result
+        assert "RUNTIME" not in result
+
+    def test_cron_hint_always_present(self):
+        job = {"prompt": "Test", "skills": []}
+        result = _build_job_prompt(job)
+        assert "scheduled cron job" in result
+        assert "[SYSTEM:" in result
+
+    def test_adapt_instruction_in_runtime_hint(self):
+        job = {"prompt": "Check Ollama health", "skills": []}
+        runtime_info = {"model": "mimo-v2-pro", "provider": "nous"}
+        result = _build_job_prompt(job, runtime_info=runtime_info)
+        assert "adapt your behavior" in result
+        assert "Do NOT attempt to reach providers" in result