Compare commits

..

1 Commits

Author SHA1 Message Date
Alexander Whitestone
77f10fa611 feat: expose cron runtime overrides for burn-loop pinning (#799)
All checks were successful
Lint / lint (pull_request) Successful in 16s
Add cron CLI flags for per-job model, provider, and base URL overrides,
forward them through hermes_cli.cron, and print pinned runtime overrides
in create/edit/list output so Gemma burn-loop jobs are auditable.
2026-04-22 11:02:47 -04:00
6 changed files with 168 additions and 41 deletions

View File

@@ -38,6 +38,18 @@ def _cron_api(**kwargs):
return json.loads(cronjob_tool(**kwargs))
def _print_runtime_overrides(job: dict) -> None:
model = job.get("model")
provider = job.get("provider")
base_url = job.get("base_url")
if model:
print(f" Model: {model}")
if provider:
print(f" Provider: {provider}")
if base_url:
print(f" Base URL: {base_url}")
def cron_list(show_all: bool = False):
"""List all scheduled jobs."""
from cron.jobs import list_jobs
@@ -93,6 +105,7 @@ def cron_list(show_all: bool = False):
script = job.get("script")
if script:
print(f" Script: {script}")
_print_runtime_overrides(job)
# Execution history
last_status = job.get("last_status")
@@ -167,6 +180,9 @@ def cron_create(args):
repeat=getattr(args, "repeat", None),
skill=getattr(args, "skill", None),
skills=_normalize_skills(getattr(args, "skill", None), getattr(args, "skills", None)),
model=getattr(args, "model", None),
provider=getattr(args, "provider", None),
base_url=getattr(args, "base_url", None),
script=getattr(args, "script", None),
)
if not result.get("success"):
@@ -180,6 +196,8 @@ def cron_create(args):
job_data = result.get("job", {})
if job_data.get("script"):
print(f" Script: {job_data['script']}")
if job_data:
_print_runtime_overrides(job_data)
print(f" Next run: {result['next_run_at']}")
return 0
@@ -217,6 +235,9 @@ def cron_edit(args):
deliver=getattr(args, "deliver", None),
repeat=getattr(args, "repeat", None),
skills=final_skills,
model=getattr(args, "model", None),
provider=getattr(args, "provider", None),
base_url=getattr(args, "base_url", None),
script=getattr(args, "script", None),
)
if not result.get("success"):
@@ -233,6 +254,7 @@ def cron_edit(args):
print(" Skills: none")
if updated.get("script"):
print(f" Script: {updated['script']}")
_print_runtime_overrides(updated)
return 0

View File

@@ -4958,6 +4958,9 @@ For more help on a command:
cron_create.add_argument("--deliver", help="Delivery target: origin, local, telegram, discord, signal, or platform:chat_id")
cron_create.add_argument("--repeat", type=int, help="Optional repeat count")
cron_create.add_argument("--skill", dest="skills", action="append", help="Attach a skill. Repeat to add multiple skills.")
cron_create.add_argument("--model", help="Pin this job to a specific model (for example: google/gemma-4-31b-it)")
cron_create.add_argument("--provider", help="Pin this job to a specific provider (for example: openrouter)")
cron_create.add_argument("--base-url", dest="base_url", help="Optional base URL override for the job's runtime provider")
cron_create.add_argument("--script", help="Path to a Python script whose stdout is injected into the prompt each run")
# cron edit
@@ -4972,6 +4975,9 @@ For more help on a command:
cron_edit.add_argument("--add-skill", dest="add_skills", action="append", help="Append a skill without replacing the existing list. Repeatable.")
cron_edit.add_argument("--remove-skill", dest="remove_skills", action="append", help="Remove a specific attached skill. Repeatable.")
cron_edit.add_argument("--clear-skills", action="store_true", help="Remove all attached skills from the job")
cron_edit.add_argument("--model", help="Update the job's pinned model")
cron_edit.add_argument("--provider", help="Update the job's pinned provider")
cron_edit.add_argument("--base-url", dest="base_url", help="Update the job's pinned base URL. Pass an empty string to clear it.")
cron_edit.add_argument("--script", help="Path to a Python script whose stdout is injected into the prompt each run. Pass empty string to clear.")
# lifecycle actions

View File

@@ -250,16 +250,12 @@ _PROVIDER_MODELS: dict[str, list[str]] = {
"big-pickle",
],
"opencode-go": [
"kimi-k2.6",
"kimi-k2.5",
"glm-5.1",
"glm-5",
"kimi-k2.5",
"mimo-v2-pro",
"mimo-v2-omni",
"minimax-m2.7",
"minimax-m2.5",
"qwen3.6-plus",
"qwen3.5-plus",
],
"ai-gateway": [
"anthropic/claude-opus-4.6",

View File

@@ -105,7 +105,7 @@ _DEFAULT_PROVIDER_MODELS = {
"ai-gateway": ["anthropic/claude-opus-4.6", "anthropic/claude-sonnet-4.6", "openai/gpt-5", "google/gemini-3-flash"],
"kilocode": ["anthropic/claude-opus-4.6", "anthropic/claude-sonnet-4.6", "openai/gpt-5.4", "google/gemini-3-pro-preview", "google/gemini-3-flash-preview"],
"opencode-zen": ["gpt-5.4", "gpt-5.3-codex", "claude-sonnet-4-6", "gemini-3-flash", "glm-5", "kimi-k2.5", "minimax-m2.7"],
"opencode-go": ["kimi-k2.6", "kimi-k2.5", "glm-5.1", "glm-5", "mimo-v2-pro", "mimo-v2-omni", "minimax-m2.5", "minimax-m2.7", "qwen3.6-plus", "qwen3.5-plus"],
"opencode-go": ["glm-5", "kimi-k2.5", "mimo-v2-pro", "mimo-v2-omni", "minimax-m2.5", "minimax-m2.7"],
"huggingface": [
"Qwen/Qwen3.5-397B-A17B", "Qwen/Qwen3-235B-A22B-Thinking-2507",
"Qwen/Qwen3-Coder-480B-A35B-Instruct", "deepseek-ai/DeepSeek-R1-0528",

View File

@@ -1,6 +1,7 @@
"""Tests for hermes_cli.cron command handling."""
from argparse import Namespace
from unittest.mock import patch
import pytest
@@ -105,3 +106,135 @@ class TestCronCommandLifecycle:
assert len(jobs) == 1
assert jobs[0]["skills"] == ["blogwatcher", "find-nearby"]
assert jobs[0]["name"] == "Skill combo"
def test_create_can_pin_runtime_model_provider_and_base_url(self, tmp_cron_dir, capsys):
cron_command(
Namespace(
cron_command="create",
schedule="every 1h",
prompt="Run the burn loop",
name="Gemma burn",
deliver=None,
repeat=None,
skill=None,
skills=None,
script=None,
model="google/gemma-4-31b-it",
provider="openrouter",
base_url="https://openrouter.ai/api/v1",
)
)
job = list_jobs()[0]
assert job["model"] == "google/gemma-4-31b-it"
assert job["provider"] == "openrouter"
assert job["base_url"] == "https://openrouter.ai/api/v1"
out = capsys.readouterr().out
assert "Created job" in out
assert "Model: google/gemma-4-31b-it" in out
assert "Provider: openrouter" in out
def test_edit_can_update_runtime_model_provider_and_clear_base_url(self, tmp_cron_dir, capsys):
job = create_job(prompt="Check server status", schedule="every 1h")
cron_command(
Namespace(
cron_command="edit",
job_id=job["id"],
schedule=None,
prompt=None,
name=None,
deliver=None,
repeat=None,
skill=None,
skills=None,
add_skills=None,
remove_skills=None,
clear_skills=False,
script=None,
model="google/gemma-4-31b-it",
provider="openrouter",
base_url="",
)
)
updated = get_job(job["id"])
assert updated["model"] == "google/gemma-4-31b-it"
assert updated["provider"] == "openrouter"
assert updated["base_url"] is None
out = capsys.readouterr().out
assert "Updated job" in out
assert "Model: google/gemma-4-31b-it" in out
assert "Provider: openrouter" in out
class TestCronParserRuntimeOverrideFlags:
def test_main_parses_create_runtime_override_flags(self, monkeypatch):
from hermes_cli import main as main_mod
captured = {}
def fake_cmd_cron(args):
captured["args"] = args
monkeypatch.setattr(main_mod, "cmd_cron", fake_cmd_cron)
monkeypatch.setattr(
"sys.argv",
[
"hermes",
"cron",
"create",
"every 1h",
"Run the burn loop",
"--model",
"google/gemma-4-31b-it",
"--provider",
"openrouter",
"--base-url",
"https://openrouter.ai/api/v1",
],
)
main_mod.main()
args = captured["args"]
assert args.cron_command == "create"
assert args.model == "google/gemma-4-31b-it"
assert args.provider == "openrouter"
assert args.base_url == "https://openrouter.ai/api/v1"
def test_main_parses_edit_runtime_override_flags(self, monkeypatch):
from hermes_cli import main as main_mod
captured = {}
def fake_cmd_cron(args):
captured["args"] = args
monkeypatch.setattr(main_mod, "cmd_cron", fake_cmd_cron)
monkeypatch.setattr(
"sys.argv",
[
"hermes",
"cron",
"edit",
"job123",
"--model",
"google/gemma-4-31b-it",
"--provider",
"openrouter",
"--base-url",
"",
],
)
main_mod.main()
args = captured["args"]
assert args.cron_command == "edit"
assert args.job_id == "job123"
assert args.model == "google/gemma-4-31b-it"
assert args.provider == "openrouter"
assert args.base_url == ""

View File

@@ -4,62 +4,32 @@ import os
from unittest.mock import patch
from hermes_cli.model_switch import list_authenticated_providers
from hermes_cli.models import curated_models_for_provider
@patch.dict(os.environ, {"OPENCODE_GO_API_KEY": "test-key"}, clear=False)
def test_opencode_go_appears_when_api_key_set():
"""opencode-go should appear in list_authenticated_providers when OPENCODE_GO_API_KEY is set."""
providers = list_authenticated_providers(current_provider="openrouter")
# Find opencode-go in results
opencode_go = next((p for p in providers if p["slug"] == "opencode-go"), None)
assert opencode_go is not None, "opencode-go should appear when OPENCODE_GO_API_KEY is set"
assert opencode_go["models"] == [
"kimi-k2.6",
"kimi-k2.5",
"glm-5.1",
"glm-5",
"mimo-v2-pro",
"mimo-v2-omni",
"minimax-m2.7",
"minimax-m2.5",
]
assert opencode_go["total_models"] == 10
assert opencode_go["models"] == ["glm-5", "kimi-k2.5", "mimo-v2-pro", "mimo-v2-omni", "minimax-m2.7", "minimax-m2.5"]
# opencode-go can appear as "built-in" (from PROVIDER_TO_MODELS_DEV when
# models.dev is reachable) or "hermes" (from HERMES_OVERLAYS fallback when
# the API is unavailable, e.g. in CI).
assert opencode_go["source"] in ("built-in", "hermes")
@patch("hermes_cli.models.provider_model_ids", return_value=[])
def test_opencode_go_curated_fallback_includes_new_models(_mock_provider_model_ids):
"""Fallback catalog should include Kimi K2.6 and both Qwen Plus models."""
model_ids = [model_id for model_id, _ in curated_models_for_provider("opencode-go")]
assert model_ids == [
"kimi-k2.6",
"kimi-k2.5",
"glm-5.1",
"glm-5",
"mimo-v2-pro",
"mimo-v2-omni",
"minimax-m2.7",
"minimax-m2.5",
"qwen3.6-plus",
"qwen3.5-plus",
]
def test_opencode_go_not_appears_when_no_creds():
"""opencode-go should NOT appear when no credentials are set."""
# Ensure OPENCODE_GO_API_KEY is not set
env_without_key = {k: v for k, v in os.environ.items() if k != "OPENCODE_GO_API_KEY"}
with patch.dict(os.environ, env_without_key, clear=True):
providers = list_authenticated_providers(current_provider="openrouter")
# opencode-go should not be in results
opencode_go = next((p for p in providers if p["slug"] == "opencode-go"), None)
assert opencode_go is None, "opencode-go should not appear without credentials"