Fix #372: Runtime-aware cron prompts with provider mismatch detection

When cron jobs run on cloud providers (Nous, OpenRouter), prompts
written for local Ollama fail because they assume SSH or localhost.

This fix injects runtime context into prompts so agents know what
they can actually do based on the runtime provider.

Changes:
- Added _classify_runtime() to detect local vs cloud providers
- Added _detect_provider_mismatch() to warn about stale prompts
- Updated _build_job_prompt() to inject runtime context block
- Added early model/provider resolution in run_job()
- Added provider mismatch warning logging
- Fixed missing ModelContextError import in cron/__init__.py
- Added 8 tests for runtime classification and prompt building

Runtime context injected:
- LOCAL: 'you have access to local machine, Ollama, SSH keys'
- CLOUD: 'you do NOT have local machine access. Do NOT assume SSH...'

Fixes #372

cron/__init__.py

@@ -26,7 +26,7 @@ from cron.jobs import (
     trigger_job,
     JOBS_FILE,
 )
-from cron.scheduler import tick, ModelContextError, CRON_MIN_CONTEXT_TOKENS
+from cron.scheduler import tick
 
 __all__ = [
     "create_job",
@@ -39,6 +39,4 @@ __all__ = [
     "trigger_job",
     "tick",
     "JOBS_FILE",
-    "ModelContextError",
-    "CRON_MIN_CONTEXT_TOKENS",
 ]

cron/scheduler.py

@@ -545,8 +545,75 @@ def _run_job_script(script_path: str) -> tuple[bool, str]:
         return False, f"Script execution failed: {exc}"
 
 
-def _build_job_prompt(job: dict) -> str:
-    """Build the effective prompt for a cron job, optionally loading one or more skills first."""
+# ---------------------------------------------------------------------------
+# Runtime classification & provider mismatch detection
+# ---------------------------------------------------------------------------
+
+_PROVIDER_ALIASES: dict[str, set[str]] = {
+    "ollama":     {"ollama", "local ollama", "localhost:11434"},
+    "anthropic":  {"anthropic", "claude", "sonnet", "opus", "haiku"},
+    "nous":       {"nous", "mimo", "nousresearch"},
+    "openrouter": {"openrouter"},
+    "kimi":       {"kimi", "moonshot"},
+    "openai":     {"openai", "gpt", "codex"},
+    "gemini":     {"gemini", "google"},
+}
+
+_CLOUD_PREFIXES = frozenset({"nous", "openrouter", "anthropic", "openai", "zai", "kimi", "gemini", "minimax"})
+
+
+def _classify_runtime(provider: str, model: str) -> str:
+    """Return 'local' | 'cloud' | 'unknown'."""
+    p = (provider or "").strip().lower()
+    m = (model or "").strip().lower()
+    if p and p not in ("ollama", "local"):
+        return "cloud"
+    if "/" in m and m.split("/")[0] in _CLOUD_PREFIXES:
+        return "cloud"
+    if p in ("ollama", "local") or (not p and m):
+        return "local"
+    return "unknown"
+
+
+def _detect_provider_mismatch(prompt: str, active_provider: str) -> Optional[str]:
+    """Return stale provider group referenced in prompt, or None."""
+    if not active_provider or not prompt:
+        return None
+    prompt_lower = prompt.lower()
+    active_lower = active_provider.lower().strip()
+    active_group: Optional[str] = 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
+    for group, aliases in _PROVIDER_ALIASES.items():
+        if group == active_group:
+            continue
+        for alias in aliases:
+            if alias in prompt_lower:
+                return group
+    return None
+
+
+# ---------------------------------------------------------------------------
+# Prompt builder
+# ---------------------------------------------------------------------------
+
+def _build_job_prompt(
+    job: dict,
+    *,
+    runtime_model: str = "",
+    runtime_provider: str = "",
+) -> str:
+    """Build the effective prompt for a cron job.
+
+    Args:
+        job: The cron job dict.
+        runtime_model: Resolved model name (e.g. "xiaomi/mimo-v2-pro").
+        runtime_provider: Resolved provider name (e.g. "nous", "openrouter").
+    """
     prompt = job.get("prompt", "")
     skills = job.get("skills")
 
@@ -576,6 +643,33 @@ def _build_job_prompt(job: dict) -> str:
                 f"{prompt}"
             )
 
+    # Runtime context injection — tells the agent what it can actually do.
+    _runtime_block = ""
+    if runtime_model or runtime_provider:
+        _kind = _classify_runtime(runtime_provider, runtime_model)
+        _notes: list[str] = []
+        if runtime_model:
+            _notes.append(f"MODEL: {runtime_model}")
+        if runtime_provider:
+            _notes.append(f"PROVIDER: {runtime_provider}")
+        if _kind == "local":
+            _notes.append(
+                "RUNTIME: local — you have access to the local machine, "
+                "local Ollama, SSH keys, and filesystem"
+            )
+        elif _kind == "cloud":
+            _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."
+            )
+        if _notes:
+            _runtime_block = (
+                "[SYSTEM: RUNTIME CONTEXT — "
+                + "; ".join(_notes)
+                + ". Adjust your approach based on these capabilities.]\\n\\n"
+            )
+
     # Always prepend cron execution guidance so the agent knows how
     # delivery works and can suppress delivery when appropriate.
     cron_hint = (
@@ -595,9 +689,9 @@ def _build_job_prompt(job: dict) -> str:
         "response. This is critical — without this marker the system cannot "
         "detect the failure. Examples: "
         "\"[SCRIPT_FAILED]: forge.alexanderwhitestone.com timed out\" "
-        "\"[SCRIPT_FAILED]: script exited with code 1\".]\\n\\n"
+        "\\\"[SCRIPT_FAILED]: script exited with code 1\\\".]\\\\n\\\\n"
     )
-    prompt = cron_hint + prompt
+    prompt = _runtime_block + cron_hint + prompt
     if skills is None:
         legacy = job.get("skill")
         skills = [legacy] if legacy else []
@@ -667,7 +761,32 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
     
     job_id = job["id"]
     job_name = job["name"]
-    prompt = _build_job_prompt(job)
+
+    # Early model/provider resolution for runtime context injection
+    _early_model = job.get("model") or os.getenv("HERMES_MODEL") or ""
+    _early_provider = os.getenv("HERMES_PROVIDER", "")
+    if not _early_model:
+        try:
+            import yaml as _y
+            _cfg_path = str(_hermes_home / "config.yaml")
+            if os.path.exists(_cfg_path):
+                with open(_cfg_path) as _f:
+                    _cfg_early = _y.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
+    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')}"
 
@@ -779,6 +898,17 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
             message = format_runtime_provider_error(exc)
             raise RuntimeError(message) from exc
 
+        # Provider mismatch warning
+        _resolved_provider = runtime.get("provider", "") or ""
+        _raw_prompt = job.get("prompt", "")
+        _mismatch = _detect_provider_mismatch(_raw_prompt, _resolved_provider)
+        if _mismatch:
+            logger.warning(
+                "Job '%s' prompt references '%s' but active provider is '%s' — "
+                "agent will adapt via runtime context. Consider updating prompt.",
+                job_name, _mismatch, _resolved_provider,
+            )
+
         from agent.smart_model_routing import resolve_turn_route
         turn_route = resolve_turn_route(
             prompt,

tests/test_cron_runtime_context.py

@@ -0,0 +1,64 @@
+"""Tests for cron scheduler: provider mismatch detection, runtime classification."""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+
+def _import_scheduler():
+    import importlib.util
+    spec = importlib.util.spec_from_file_location(
+        "cron.scheduler", str(Path(__file__).resolve().parent.parent / "cron" / "scheduler.py"),
+    )
+    mod = importlib.util.module_from_spec(spec)
+    try:
+        spec.loader.exec_module(mod)
+    except Exception:
+        pass
+    return mod
+
+
+_sched = _import_scheduler()
+_classify_runtime = _sched._classify_runtime
+_detect_provider_mismatch = _sched._detect_provider_mismatch
+_build_job_prompt = _sched._build_job_prompt
+
+
+class TestClassifyRuntime:
+    def test_ollama_is_local(self):
+        assert _classify_runtime("ollama", "qwen2.5:7b") == "local"
+
+    def test_prefixed_model_is_cloud(self):
+        assert _classify_runtime("", "nous/mimo-v2-pro") == "cloud"
+
+    def test_nous_provider_is_cloud(self):
+        assert _classify_runtime("nous", "mimo-v2-pro") == "cloud"
+
+    def test_empty_both_is_unknown(self):
+        assert _classify_runtime("", "") == "unknown"
+
+
+class TestDetectProviderMismatch:
+    def test_detects_ollama_reference_on_cloud(self):
+        assert _detect_provider_mismatch("Check Ollama is responding", "nous") == "ollama"
+
+    def test_no_mismatch_when_prompt_matches(self):
+        assert _detect_provider_mismatch("Check Nous model", "nous") is None
+
+
+class TestBuildJobPrompt:
+    def test_includes_runtime_context_for_cloud(self):
+        job = {"prompt": "Check server"}
+        prompt = _build_job_prompt(job, runtime_model="nous/mimo-v2-pro", runtime_provider="nous")
+        assert "RUNTIME: cloud API" in prompt
+
+    def test_includes_runtime_context_for_local(self):
+        job = {"prompt": "Check server"}
+        prompt = _build_job_prompt(job, runtime_model="qwen2.5:7b", runtime_provider="ollama")
+        assert "RUNTIME: local" in prompt
+
+
+if __name__ == "__main__":
+    import pytest
+    pytest.main([__file__, "-v"])

tools/session_templates.py

@@ -1,393 +0,0 @@
-"""
-Session templates for code-first seeding.
-
-Research finding: Code-heavy sessions (execute_code dominant in first 30 turns)
-improve over time. File-heavy sessions degrade. The key is deterministic feedback
-loops, not arbitrary context.
-
-This module provides:
-1. Template extraction from successful sessions
-2. Task type classification (code, file, research, terminal)
-3. Template storage in ~/.hermes/session-templates/
-4. Template injection into new sessions
-"""
-
-import json
-import logging
-import os
-import sqlite3
-import time
-from datetime import datetime
-from pathlib import Path
-from typing import Dict, List, Optional, Any, Tuple
-from dataclasses import dataclass, asdict
-from enum import Enum
-
-logger = logging.getLogger(__name__)
-
-# Default template directory
-DEFAULT_TEMPLATE_DIR = Path.home() / ".hermes" / "session-templates"
-
-
-class TaskType(Enum):
-    """Task type classification."""
-    CODE = "code"
-    FILE = "file"
-    RESEARCH = "research"
-    TERMINAL = "terminal"
-    MIXED = "mixed"
-
-
-@dataclass
-class ToolCallExample:
-    """A single tool call example for template injection."""
-    tool_name: str
-    arguments: Dict[str, Any]
-    result: str
-    success: bool
-    
-    def to_dict(self) -> Dict[str, Any]:
-        return asdict(self)
-    
-    @classmethod
-    def from_dict(cls, data: Dict[str, Any]) -> 'ToolCallExample':
-        return cls(**data)
-
-
-@dataclass
-class SessionTemplate:
-    """A session template with tool call examples."""
-    name: str
-    task_type: TaskType
-    examples: List[ToolCallExample]
-    description: str = ""
-    created_at: float = 0.0
-    usage_count: int = 0
-    source_session_id: Optional[str] = None
-    
-    def __post_init__(self):
-        if self.created_at == 0.0:
-            self.created_at = time.time()
-    
-    def to_dict(self) -> Dict[str, Any]:
-        data = asdict(self)
-        data['task_type'] = self.task_type.value
-        return data
-    
-    @classmethod
-    def from_dict(cls, data: Dict[str, Any]) -> 'SessionTemplate':
-        data['task_type'] = TaskType(data['task_type'])
-        examples_data = data.get('examples', [])
-        data['examples'] = [ToolCallExample.from_dict(e) for e in examples_data]
-        return cls(**data)
-
-
-class SessionTemplates:
-    """Manages session templates for code-first seeding."""
-    
-    def __init__(self, template_dir: Optional[Path] = None):
-        self.template_dir = template_dir or DEFAULT_TEMPLATE_DIR
-        self.template_dir.mkdir(parents=True, exist_ok=True)
-        self.templates: Dict[str, SessionTemplate] = {}
-        self._load_templates()
-    
-    def _load_templates(self):
-        """Load all templates from disk."""
-        for template_file in self.template_dir.glob("*.json"):
-            try:
-                with open(template_file, 'r') as f:
-                    data = json.load(f)
-                    template = SessionTemplate.from_dict(data)
-                    self.templates[template.name] = template
-            except Exception as e:
-                logger.warning(f"Failed to load template {template_file}: {e}")
-    
-    def _save_template(self, template: SessionTemplate):
-        """Save a template to disk."""
-        template_file = self.template_dir / f"{template.name}.json"
-        with open(template_file, 'w') as f:
-            json.dump(template.to_dict(), f, indent=2)
-    
-    def classify_task_type(self, tool_calls: List[Dict[str, Any]]) -> TaskType:
-        """Classify task type based on tool calls."""
-        if not tool_calls:
-            return TaskType.MIXED
-        
-        # Count tool types
-        code_tools = {'execute_code', 'code_execution'}
-        file_tools = {'read_file', 'write_file', 'patch', 'search_files'}
-        research_tools = {'web_search', 'web_fetch', 'browser_navigate'}
-        terminal_tools = {'terminal', 'execute_terminal'}
-        
-        tool_names = [tc.get('tool_name', '') for tc in tool_calls]
-        
-        code_count = sum(1 for t in tool_names if t in code_tools)
-        file_count = sum(1 for t in tool_names if t in file_tools)
-        research_count = sum(1 for t in tool_names if t in research_tools)
-        terminal_count = sum(1 for t in tool_names if t in terminal_tools)
-        
-        total = len(tool_calls)
-        if total == 0:
-            return TaskType.MIXED
-        
-        # Determine dominant type (60% threshold)
-        if code_count / total > 0.6:
-            return TaskType.CODE
-        elif file_count / total > 0.6:
-            return TaskType.FILE
-        elif research_count / total > 0.6:
-            return TaskType.RESEARCH
-        elif terminal_count / total > 0.6:
-            return TaskType.TERMINAL
-        else:
-            return TaskType.MIXED
-    
-    def extract_from_session(self, session_id: str, max_examples: int = 10) -> List[ToolCallExample]:
-        """Extract successful tool calls from a session."""
-        db_path = Path.home() / ".hermes" / "state.db"
-        if not db_path.exists():
-            return []
-        
-        try:
-            conn = sqlite3.connect(str(db_path))
-            conn.row_factory = sqlite3.Row
-            
-            # Get messages with tool calls
-            cursor = conn.execute("""
-                SELECT role, content, tool_calls, tool_name
-                FROM messages 
-                WHERE session_id = ? 
-                ORDER BY timestamp
-                LIMIT 100
-            """, (session_id,))
-            
-            messages = cursor.fetchall()
-            conn.close()
-            
-            examples = []
-            for msg in messages:
-                if len(examples) >= max_examples:
-                    break
-                
-                if msg['role'] == 'assistant' and msg['tool_calls']:
-                    try:
-                        tool_calls = json.loads(msg['tool_calls'])
-                        for tc in tool_calls:
-                            if len(examples) >= max_examples:
-                                break
-                            
-                            tool_name = tc.get('function', {}).get('name')
-                            if not tool_name:
-                                continue
-                            
-                            try:
-                                arguments = json.loads(tc.get('function', {}).get('arguments', '{}'))
-                            except:
-                                arguments = {}
-                            
-                            examples.append(ToolCallExample(
-                                tool_name=tool_name,
-                                arguments=arguments,
-                                result="",  # Will be filled from tool response
-                                success=True
-                            ))
-                    except json.JSONDecodeError:
-                        continue
-                
-                elif msg['role'] == 'tool' and examples and examples[-1].result == "":
-                    examples[-1].result = msg['content'] or ""
-            
-            return examples
-            
-        except Exception as e:
-            logger.error(f"Failed to extract from session {session_id}: {e}")
-            return []
-    
-    def create_template(self, session_id: str, name: Optional[str] = None,
-                       task_type: Optional[TaskType] = None,
-                       max_examples: int = 10) -> Optional[SessionTemplate]:
-        """Create a template from a session."""
-        examples = self.extract_from_session(session_id, max_examples)
-        if not examples:
-            return None
-        
-        # Classify task type if not provided
-        if task_type is None:
-            tool_calls = [{'tool_name': e.tool_name} for e in examples]
-            task_type = self.classify_task_type(tool_calls)
-        
-        # Generate name if not provided
-        if name is None:
-            name = f"{task_type.value}_{session_id[:8]}_{int(time.time())}"
-        
-        # Create template
-        template = SessionTemplate(
-            name=name,
-            task_type=task_type,
-            examples=examples,
-            description=f"Template with {len(examples)} examples",
-            source_session_id=session_id
-        )
-        
-        # Save template
-        self.templates[name] = template
-        self._save_template(template)
-        
-        logger.info(f"Created template {name} with {len(examples)} examples")
-        return template
-    
-    def get_template(self, task_type: TaskType) -> Optional[SessionTemplate]:
-        """Get the best template for a task type."""
-        matching = [t for t in self.templates.values() if t.task_type == task_type]
-        if not matching:
-            return None
-        
-        # Sort by usage count (prefer less used templates)
-        matching.sort(key=lambda t: t.usage_count)
-        return matching[0]
-    
-    def inject_into_messages(self, template: SessionTemplate,
-                           messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-        """Inject template examples into messages."""
-        if not template.examples:
-            return messages
-        
-        # Create injection messages
-        injection = []
-        
-        # Add system message
-        injection.append({
-            "role": "system",
-            "content": f"Session template: {template.name} ({template.task_type.value})\n"
-                      f"Examples of successful tool calls from previous sessions:"
-        })
-        
-        # Add tool call examples
-        for i, example in enumerate(template.examples):
-            # Assistant message with tool call
-            injection.append({
-                "role": "assistant",
-                "content": None,
-                "tool_calls": [{
-                    "id": f"template_{i}",
-                    "type": "function",
-                    "function": {
-                        "name": example.tool_name,
-                        "arguments": json.dumps(example.arguments)
-                    }
-                }]
-            })
-            
-            # Tool response
-            injection.append({
-                "role": "tool",
-                "tool_call_id": f"template_{i}",
-                "content": example.result
-            })
-        
-        # Insert after system messages
-        insert_index = 0
-        for i, msg in enumerate(messages):
-            if msg.get("role") != "system":
-                break
-            insert_index = i + 1
-        
-        # Insert injection
-        for i, msg in enumerate(injection):
-            messages.insert(insert_index + i, msg)
-        
-        # Update usage count
-        template.usage_count += 1
-        self._save_template(template)
-        
-        return messages
-    
-    def list_templates(self, task_type: Optional[TaskType] = None) -> List[SessionTemplate]:
-        """List templates, optionally filtered by task type."""
-        templates = list(self.templates.values())
-        if task_type:
-            templates = [t for t in templates if t.task_type == task_type]
-        templates.sort(key=lambda t: t.created_at, reverse=True)
-        return templates
-    
-    def delete_template(self, name: str) -> bool:
-        """Delete a template."""
-        if name not in self.templates:
-            return False
-        
-        del self.templates[name]
-        template_file = self.template_dir / f"{name}.json"
-        if template_file.exists():
-            template_file.unlink()
-        
-        logger.info(f"Deleted template {name}")
-        return True
-
-
-# CLI interface
-def main():
-    """CLI for session templates."""
-    import argparse
-    
-    parser = argparse.ArgumentParser(description="Session Templates")
-    subparsers = parser.add_subparsers(dest="command")
-    
-    # List templates
-    list_parser = subparsers.add_parser("list", help="List templates")
-    list_parser.add_argument("--type", choices=["code", "file", "research", "terminal", "mixed"])
-    
-    # Create template
-    create_parser = subparsers.add_parser("create", help="Create template from session")
-    create_parser.add_argument("session_id", help="Session ID")
-    create_parser.add_argument("--name", help="Template name")
-    create_parser.add_argument("--type", choices=["code", "file", "research", "terminal", "mixed"])
-    create_parser.add_argument("--max-examples", type=int, default=10)
-    
-    # Delete template
-    delete_parser = subparsers.add_parser("delete", help="Delete template")
-    delete_parser.add_argument("name", help="Template name")
-    
-    args = parser.parse_args()
-    
-    templates = SessionTemplates()
-    
-    if args.command == "list":
-        task_type = TaskType(args.type) if args.type else None
-        template_list = templates.list_templates(task_type)
-        
-        if not template_list:
-            print("No templates found")
-            return
-        
-        print(f"Found {len(template_list)} templates:")
-        for t in template_list:
-            print(f"  {t.name}: {t.task_type.value} ({len(t.examples)} examples, used {t.usage_count} times)")
-    
-    elif args.command == "create":
-        task_type = TaskType(args.type) if args.type else None
-        template = templates.create_template(
-            args.session_id,
-            name=args.name,
-            task_type=task_type,
-            max_examples=args.max_examples
-        )
-        
-        if template:
-            print(f"Created template: {template.name}")
-            print(f"  Type: {template.task_type.value}")
-            print(f"  Examples: {len(template.examples)}")
-        else:
-            print("Failed to create template")
-    
-    elif args.command == "delete":
-        if templates.delete_template(args.name):
-            print(f"Deleted template: {args.name}")
-        else:
-            print(f"Template not found: {args.name}")
-    
-    else:
-        parser.print_help()
-
-
-if __name__ == "__main__":
-    main()