Compare commits

...

1 Commits

Author SHA1 Message Date
Alexander Whitestone
07bd9c51f9 fix: inject runtime context into cron prompts, detect provider mismatches (#372)
Some checks failed
Forge CI / smoke-and-build (pull_request) Failing after 1m2s
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
2026-04-13 19:23:55 -04:00
2 changed files with 161 additions and 5 deletions

View File

@@ -544,8 +544,57 @@ 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."""
# 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_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")
@@ -577,9 +626,21 @@ def _build_job_prompt(job: dict) -> str:
# 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. "
@@ -666,12 +727,10 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
job_id = job["id"]
job_name = job["name"]
prompt = _build_job_prompt(job)
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.
@@ -762,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,

View File

@@ -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