412ee7329a
Some checks failed
Forge CI / smoke-and-build (pull_request) Failing after 1m7s
After provider migration (Ollama -> Nous/mimo-v2-pro), cron jobs with provider-specific prompts ran on the wrong provider without knowing it. Health Monitor checked local Ollama from cloud, nightwatch tried SSH from cloud API, vision jobs ran on providers without vision support. Changes to cron/scheduler.py: 1. _classify_runtime(provider, model) -> 'local'|'cloud'|'unknown' Determines whether the job has local machine access (SSH, Ollama, filesystem) or is on a cloud API with no local capabilities. 2. _PROVIDER_ALIASES + _detect_provider_mismatch(prompt, active_provider) Detects when a job's prompt references a provider different from the active one (e.g. 'ollama' in prompt when running on 'nous'). Logs a warning so operators know which prompts need updating. 3. _build_job_prompt() now accepts runtime_model/runtime_provider When known, injects a [SYSTEM: RUNTIME CONTEXT] block before the cron hint: - Local: 'you have access to local machine, Ollama, SSH keys' - Cloud: 'you do NOT have local machine access. Do NOT SSH, etc.' 4. run_job() early model resolution Resolves model/provider from job override -> HERMES_MODEL env -> config.yaml model.default, derives provider from model prefix. Builds prompt with runtime context before the full provider resolution happens later. 5. Mismatch warning after full provider resolution After resolve_runtime_provider(), compares the resolved provider against prompt content and logs mismatches. Supersedes #403 (early resolution only) and #427 (mismatch detection only). Combines both approaches with local/cloud capability awareness. Closes #372
130 lines
4.8 KiB
Python
130 lines
4.8 KiB
Python
"""Tests for cron scheduler: provider mismatch detection, runtime classification,
|
|
and capability-aware prompt building."""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
|
|
def _import_scheduler():
|
|
"""Import the scheduler module, bypassing __init__.py re-exports that may
|
|
reference symbols not yet merged upstream."""
|
|
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 # some top-level imports may fail in CI; functions are still defined
|
|
return mod
|
|
|
|
|
|
_sched = _import_scheduler()
|
|
_classify_runtime = _sched._classify_runtime
|
|
_detect_provider_mismatch = _sched._detect_provider_mismatch
|
|
_build_job_prompt = _sched._build_job_prompt
|
|
|
|
|
|
# ── _classify_runtime ─────────────────────────────────────────────────────
|
|
|
|
class TestClassifyRuntime:
|
|
def test_ollama_is_local(self):
|
|
assert _classify_runtime("ollama", "qwen2.5:7b") == "local"
|
|
|
|
def test_empty_provider_is_local(self):
|
|
assert _classify_runtime("", "my-local-model") == "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_openrouter_is_cloud(self):
|
|
assert _classify_runtime("openrouter", "anthropic/claude-sonnet-4") == "cloud"
|
|
|
|
def test_empty_both_is_unknown(self):
|
|
assert _classify_runtime("", "") == "unknown"
|
|
|
|
|
|
# ── _detect_provider_mismatch ─────────────────────────────────────────────
|
|
|
|
class TestDetectProviderMismatch:
|
|
def test_no_mismatch_when_not_mentioned(self):
|
|
assert _detect_provider_mismatch("Check system health", "nous") is None
|
|
|
|
def test_detects_ollama_when_nous_active(self):
|
|
assert _detect_provider_mismatch("Check Ollama is responding", "nous") == "ollama"
|
|
|
|
def test_detects_anthropic_when_nous_active(self):
|
|
assert _detect_provider_mismatch("Use Claude to analyze", "nous") == "anthropic"
|
|
|
|
def test_no_mismatch_same_provider(self):
|
|
assert _detect_provider_mismatch("Check Ollama models", "ollama") is None
|
|
|
|
def test_empty_prompt(self):
|
|
assert _detect_provider_mismatch("", "nous") is None
|
|
|
|
def test_empty_provider(self):
|
|
assert _detect_provider_mismatch("Check Ollama", "") is None
|
|
|
|
def test_detects_kimi_when_openrouter(self):
|
|
assert _detect_provider_mismatch("Use Kimi for coding", "openrouter") == "kimi"
|
|
|
|
def test_detects_glm_when_nous(self):
|
|
assert _detect_provider_mismatch("Use GLM for analysis", "nous") == "zai"
|
|
|
|
|
|
# ── _build_job_prompt ─────────────────────────────────────────────────────
|
|
|
|
class TestBuildJobPrompt:
|
|
def _job(self, prompt="Do something"):
|
|
return {"prompt": prompt, "skills": []}
|
|
|
|
def test_no_runtime_no_block(self):
|
|
result = _build_job_prompt(self._job())
|
|
assert "Do something" in result
|
|
assert "RUNTIME CONTEXT" not in result
|
|
|
|
def test_cloud_runtime_injected(self):
|
|
result = _build_job_prompt(
|
|
self._job(),
|
|
runtime_model="xiaomi/mimo-v2-pro",
|
|
runtime_provider="nous",
|
|
)
|
|
assert "MODEL: xiaomi/mimo-v2-pro" in result
|
|
assert "PROVIDER: nous" in result
|
|
assert "cloud API" in result
|
|
assert "Do NOT assume you can SSH" in result
|
|
|
|
def test_local_runtime_injected(self):
|
|
result = _build_job_prompt(
|
|
self._job(),
|
|
runtime_model="qwen2.5:7b",
|
|
runtime_provider="ollama",
|
|
)
|
|
assert "RUNTIME: local" in result
|
|
assert "SSH keys" in result
|
|
|
|
def test_empty_runtime_no_block(self):
|
|
result = _build_job_prompt(self._job(), runtime_model="", runtime_provider="")
|
|
assert "RUNTIME CONTEXT" not in result
|
|
|
|
def test_cron_hint_always_present(self):
|
|
result = _build_job_prompt(self._job())
|
|
assert "scheduled cron job" in result
|
|
assert "[SYSTEM:" in result
|
|
|
|
def test_runtime_block_before_cron_hint(self):
|
|
result = _build_job_prompt(
|
|
self._job("Check Ollama"),
|
|
runtime_model="mimo-v2-pro",
|
|
runtime_provider="nous",
|
|
)
|
|
runtime_pos = result.index("RUNTIME CONTEXT")
|
|
cron_pos = result.index("scheduled cron job")
|
|
assert runtime_pos < cron_pos
|