354 lines
15 KiB
Python
354 lines
15 KiB
Python
"""Unpinned cron jobs run on their creation snapshot (#44585 follow-up).
|
|
|
|
Background: an UNPINNED cron job used to follow the live global default provider/model. A
|
|
temporary switch to a paid provider made every unpinned job silently inherit it on its next
|
|
tick (the $7.73 incident). The first fix failed closed on any drift, which instead killed every
|
|
unpinned job whenever the operator changed models — silently, for days.
|
|
|
|
Current contract:
|
|
- create_job() snapshots the provider/model resolution WOULD pick at creation into
|
|
job["provider_snapshot"] / job["model_snapshot"] (unpinned, agent-backed jobs only).
|
|
- run_job() treats the snapshot as the effective pin: an unpinned axis runs on its snapshot
|
|
even after the global default moved. Explicit per-job pins and the cron.model /
|
|
cron.model_provider fleet defaults still win; a job with no snapshot follows the global
|
|
default as before.
|
|
|
|
These tests exercise the full run_job path (real imports, mocked AIAgent +
|
|
resolve_runtime_provider against a temp HERMES_HOME) and the create_job snapshot capture.
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
# Ensure project root is importable.
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
|
|
|
from cron.scheduler import run_job
|
|
|
|
|
|
def _base_job(**overrides):
|
|
job = {
|
|
"id": "pin-test",
|
|
"name": "pin test",
|
|
"prompt": "hello",
|
|
"model": None,
|
|
"provider": None,
|
|
"provider_snapshot": None,
|
|
"base_url": None,
|
|
}
|
|
job.update(overrides)
|
|
return job
|
|
|
|
|
|
def _run(job, tmp_path, *, current_provider="openrouter", current_model=None, cron_model=None,
|
|
cron_model_provider=None):
|
|
"""Drive run_job against a temp config.yaml whose ``model.default`` / ``model.provider`` are
|
|
the CURRENT global defaults. Returns ``(success, error, agent_kwargs, resolve_kwargs)`` where
|
|
the last two are the kwargs AIAgent / resolve_runtime_provider were called with (None when
|
|
never called)."""
|
|
config_yaml = ""
|
|
if current_model or current_provider:
|
|
config_yaml += "model:\n"
|
|
if current_model:
|
|
config_yaml += f" default: {current_model}\n"
|
|
if current_provider:
|
|
config_yaml += f" provider: {current_provider}\n"
|
|
cron_lines = []
|
|
if cron_model is not None:
|
|
cron_lines.append(f" model: {cron_model}")
|
|
if cron_model_provider is not None:
|
|
cron_lines.append(f" model_provider: {cron_model_provider}")
|
|
if cron_lines:
|
|
config_yaml += "cron:\n" + "\n".join(cron_lines) + "\n"
|
|
(tmp_path / "config.yaml").write_text(config_yaml)
|
|
|
|
resolve_kwargs = {}
|
|
|
|
def _resolve(**kwargs):
|
|
resolve_kwargs.update(kwargs)
|
|
return {
|
|
"api_key": "test-key",
|
|
"base_url": "https://example.invalid/v1",
|
|
"provider": kwargs.get("requested") or current_provider,
|
|
"api_mode": "chat_completions",
|
|
}
|
|
|
|
fake_db = MagicMock()
|
|
with patch("cron.scheduler._hermes_home", tmp_path), \
|
|
patch("cron.scheduler._get_hermes_home", return_value=tmp_path), \
|
|
patch("cron.scheduler_delivery._resolve_origin", return_value=None), \
|
|
patch("hermes_cli.env_loader.load_hermes_dotenv"), \
|
|
patch("hermes_cli.env_loader.reset_secret_source_cache"), \
|
|
patch("hermes_state_registry.acquire", return_value=fake_db), \
|
|
patch("hermes_cli.runtime_provider.resolve_runtime_provider", side_effect=_resolve), \
|
|
patch("run_agent.AIAgent") as mock_agent_cls:
|
|
mock_agent = MagicMock()
|
|
mock_agent.run_conversation.return_value = {"final_response": "ok"}
|
|
mock_agent_cls.return_value = mock_agent
|
|
success, _output, _final, error = run_job(job)
|
|
agent_kwargs = mock_agent_cls.call_args.kwargs if mock_agent_cls.called else None
|
|
return success, error, agent_kwargs, (resolve_kwargs or None)
|
|
|
|
|
|
class TestSnapshotIsTheEffectivePin:
|
|
def test_unpinned_job_runs_on_snapshot_after_global_default_moved(self, tmp_path):
|
|
"""Global default moved old-provider/old-model -> new-provider/new-model; the unpinned job
|
|
still runs, on what it was created under. Neither a skip nor a silent inherit."""
|
|
job = _base_job(provider_snapshot="old-provider", model_snapshot="old-model")
|
|
success, error, agent_kwargs, resolve_kwargs = _run(
|
|
job, tmp_path, current_provider="new-provider", current_model="new-model")
|
|
|
|
assert success is True, error
|
|
assert agent_kwargs["model"] == "old-model"
|
|
assert resolve_kwargs["requested"] == "old-provider"
|
|
assert resolve_kwargs["target_model"] == "old-model"
|
|
|
|
def test_explicit_job_pin_beats_snapshot(self, tmp_path):
|
|
job = _base_job(
|
|
provider="pinned-provider", model="pinned-model",
|
|
provider_snapshot="old-provider", model_snapshot="old-model")
|
|
success, error, agent_kwargs, resolve_kwargs = _run(
|
|
job, tmp_path, current_provider="new-provider", current_model="new-model",
|
|
cron_model="fleet-model")
|
|
|
|
assert success is True, error
|
|
assert agent_kwargs["model"] == "pinned-model"
|
|
assert resolve_kwargs["requested"] == "pinned-provider"
|
|
|
|
def test_cron_fleet_default_beats_snapshot(self, tmp_path):
|
|
"""cron.model / cron.model_provider deliberately route the whole unpinned fleet."""
|
|
job = _base_job(provider_snapshot="old-provider", model_snapshot="old-model")
|
|
success, error, agent_kwargs, resolve_kwargs = _run(
|
|
job, tmp_path, current_provider="new-provider", current_model="new-model",
|
|
cron_model="fleet-model", cron_model_provider="fleet-provider")
|
|
|
|
assert success is True, error
|
|
assert agent_kwargs["model"] == "fleet-model"
|
|
assert resolve_kwargs["requested"] == "fleet-provider"
|
|
|
|
def test_job_without_snapshot_follows_global_default(self, tmp_path):
|
|
"""Legacy record (keys absent) keeps tracking the live global default."""
|
|
job = _base_job()
|
|
job.pop("provider_snapshot", None)
|
|
success, error, agent_kwargs, resolve_kwargs = _run(
|
|
job, tmp_path, current_provider="new-provider", current_model="new-model")
|
|
|
|
assert success is True, error
|
|
assert agent_kwargs["model"] == "new-model"
|
|
assert resolve_kwargs["requested"] is None
|
|
|
|
def test_missing_model_guides_to_user_owned_cli(self, tmp_path, monkeypatch):
|
|
"""A missing-model failure cannot advertise agent-owned pinning."""
|
|
monkeypatch.delenv("HERMES_MODEL", raising=False)
|
|
success, error, agent_kwargs, _ = _run(
|
|
_base_job(), tmp_path, current_provider="openrouter", current_model=None)
|
|
|
|
assert success is False
|
|
assert agent_kwargs is None
|
|
assert "hermes cron edit pin-test --model <name>" in error
|
|
assert "cronjob action=update" not in error
|
|
|
|
|
|
class TestCreateJobSnapshot:
|
|
"""create_job captures provider_snapshot for unpinned agent jobs only."""
|
|
|
|
@staticmethod
|
|
def _isolate_storage(monkeypatch):
|
|
"""Patch cron.jobs storage so create_job never touches the real store."""
|
|
import contextlib
|
|
import cron.jobs as jobs
|
|
|
|
@contextlib.contextmanager
|
|
def _noop_lock():
|
|
yield
|
|
|
|
monkeypatch.setattr(jobs, "_jobs_lock", _noop_lock, raising=True)
|
|
monkeypatch.setattr(jobs, "load_jobs", lambda: [], raising=True)
|
|
monkeypatch.setattr(jobs, "save_jobs", lambda j: None, raising=True)
|
|
return jobs
|
|
|
|
def test_unpinned_job_captures_snapshot(self, monkeypatch):
|
|
jobs = self._isolate_storage(monkeypatch)
|
|
|
|
with patch(
|
|
"hermes_cli.runtime_provider.resolve_runtime_provider",
|
|
return_value={"provider": "openrouter"},
|
|
):
|
|
job = jobs.create_job(prompt="do a thing", schedule="every 1 hour")
|
|
|
|
assert job["provider"] is None
|
|
assert job["provider_snapshot"] == "openrouter"
|
|
|
|
def test_pinned_job_skips_snapshot(self, monkeypatch):
|
|
jobs = self._isolate_storage(monkeypatch)
|
|
|
|
resolver = MagicMock(return_value={"provider": "openrouter"})
|
|
with patch("hermes_cli.runtime_provider.resolve_runtime_provider", resolver):
|
|
job = jobs.create_job(
|
|
prompt="do a thing", schedule="every 1 hour", provider="nous"
|
|
)
|
|
|
|
# Explicit provider → pinned → no snapshot needed, and resolution skipped.
|
|
assert job["provider"] == "nous"
|
|
assert job["provider_snapshot"] is None
|
|
resolver.assert_not_called()
|
|
|
|
def test_snapshot_resolution_error_fails_open_to_none(self, monkeypatch):
|
|
"""If resolution raises at creation, snapshot is None — creation never breaks."""
|
|
jobs = self._isolate_storage(monkeypatch)
|
|
|
|
with patch(
|
|
"hermes_cli.runtime_provider.resolve_runtime_provider",
|
|
side_effect=RuntimeError("no creds"),
|
|
):
|
|
job = jobs.create_job(prompt="do a thing", schedule="every 1 hour")
|
|
|
|
assert job["provider_snapshot"] is None
|
|
|
|
|
|
class TestRuntimeResolutionTargetModel:
|
|
"""run_job must resolve the primary provider against the model the job will actually run
|
|
(per-job pin > cron.model > snapshot > config default), so providers with model-specific
|
|
api_mode routing pick the mode for that model instead of the stale persisted default."""
|
|
|
|
def test_primary_resolution_passes_effective_model(self, tmp_path):
|
|
job = _base_job(model="my-pinned-model", provider="openrouter")
|
|
success, error, _agent_kwargs, resolve_kwargs = _run(
|
|
job, tmp_path, current_provider="openrouter", current_model="other-model")
|
|
|
|
assert success is True, error
|
|
assert resolve_kwargs["target_model"] == "my-pinned-model"
|
|
assert resolve_kwargs["requested"] == "openrouter"
|
|
|
|
|
|
class TestResnapshot:
|
|
"""resnapshot_job / resnapshot_all_unpinned — 'adopt the current global
|
|
default without pinning' (#44585 companion). These refresh an unpinned
|
|
job's snapshot(s) to the CURRENT global resolution while leaving the job
|
|
unpinned, so it keeps tracking future global changes."""
|
|
|
|
@staticmethod
|
|
def _install_store(monkeypatch, initial_jobs):
|
|
"""Install an in-memory cron job store backed by a real list so
|
|
resnapshot functions can load/save against it."""
|
|
import contextlib
|
|
import cron.jobs as jobs
|
|
|
|
store = [dict(j) for j in initial_jobs] # deep-ish copy per job
|
|
|
|
@contextlib.contextmanager
|
|
def _lock():
|
|
yield
|
|
|
|
monkeypatch.setattr(jobs, "_jobs_lock", _lock, raising=True)
|
|
monkeypatch.setattr(jobs, "load_jobs", lambda: [dict(j) for j in store], raising=True)
|
|
|
|
def _save(job_list):
|
|
store[:] = [dict(j) for j in job_list]
|
|
|
|
monkeypatch.setattr(jobs, "save_jobs", _save, raising=True)
|
|
return jobs, store
|
|
|
|
def _make_job(self, job_id, **overrides):
|
|
job = {
|
|
"id": job_id,
|
|
"name": f"job {job_id}",
|
|
"prompt": "do a thing",
|
|
"model": None,
|
|
"provider": None,
|
|
"model_snapshot": "old-model",
|
|
"provider_snapshot": "old-provider",
|
|
"base_url": None,
|
|
"no_agent": False,
|
|
}
|
|
job.update(overrides)
|
|
return job
|
|
|
|
def test_resnapshot_unpinned_refreshes_to_current(self, monkeypatch, tmp_path):
|
|
jobs_mod, store = self._install_store(
|
|
monkeypatch, [self._make_job("j1", model_snapshot="old-model")]
|
|
)
|
|
(tmp_path / "config.yaml").write_text("model:\n default: new-model\n")
|
|
monkeypatch.setattr("cron.jobs.get_hermes_home", lambda: tmp_path, raising=True)
|
|
with patch(
|
|
"hermes_cli.runtime_provider.resolve_runtime_provider",
|
|
return_value={"provider": "openrouter"},
|
|
):
|
|
updated = jobs_mod.resnapshot_job("j1")
|
|
|
|
assert updated is not None
|
|
assert updated["model"] is None, "job must stay unpinned"
|
|
assert updated["model_snapshot"] == "new-model"
|
|
assert updated["provider_snapshot"] == "openrouter"
|
|
# Persisted too.
|
|
assert store[0]["model_snapshot"] == "new-model"
|
|
assert store[0]["provider_snapshot"] == "openrouter"
|
|
|
|
def test_resnapshot_pinned_job_keeps_none_snapshot(self, monkeypatch, tmp_path):
|
|
# A fully-pinned job already carries None snapshots; resnapping must not
|
|
# clobber them into a global default.
|
|
jobs_mod, store = self._install_store(
|
|
monkeypatch,
|
|
[
|
|
self._make_job(
|
|
"j1",
|
|
model="my-pinned-model",
|
|
provider="openrouter",
|
|
model_snapshot=None,
|
|
provider_snapshot=None,
|
|
)
|
|
],
|
|
)
|
|
(tmp_path / "config.yaml").write_text("model:\n default: new-model\n")
|
|
monkeypatch.setattr("cron.jobs.get_hermes_home", lambda: tmp_path, raising=True)
|
|
with patch(
|
|
"hermes_cli.runtime_provider.resolve_runtime_provider",
|
|
return_value={"provider": "openrouter"},
|
|
):
|
|
updated = jobs_mod.resnapshot_job("j1")
|
|
|
|
# Pinned axes stay pinned: model unchanged, snapshots still None.
|
|
assert updated["model"] == "my-pinned-model"
|
|
assert updated["model_snapshot"] is None
|
|
assert updated["provider_snapshot"] is None
|
|
|
|
def test_resnapshot_missing_job_returns_none(self, monkeypatch, tmp_path):
|
|
jobs_mod, _store = self._install_store(monkeypatch, [])
|
|
assert jobs_mod.resnapshot_job("nope") is None
|
|
|
|
def test_resnapshot_all_skips_no_agent_and_fully_pinned(self, monkeypatch, tmp_path):
|
|
jobs_mod, store = self._install_store(
|
|
monkeypatch,
|
|
[
|
|
# unpinned, model-only → should be refreshed (provider axis too)
|
|
self._make_job("j1", model_snapshot="old", provider_snapshot="old"),
|
|
# no_agent → skipped entirely
|
|
self._make_job("j2", no_agent=True, model_snapshot="old", provider_snapshot="old"),
|
|
# fully pinned → skipped (nothing unpinned)
|
|
self._make_job(
|
|
"j3",
|
|
model="pm", provider="pp",
|
|
model_snapshot=None, provider_snapshot=None,
|
|
),
|
|
],
|
|
)
|
|
(tmp_path / "config.yaml").write_text("model:\n default: new-model\n")
|
|
monkeypatch.setattr("cron.jobs.get_hermes_home", lambda: tmp_path, raising=True)
|
|
with patch(
|
|
"hermes_cli.runtime_provider.resolve_runtime_provider",
|
|
return_value={"provider": "openrouter"},
|
|
):
|
|
updated = jobs_mod.resnapshot_all_unpinned()
|
|
|
|
ids = [j["id"] for j in updated]
|
|
assert ids == ["j1"], "only the unpinned agent job is refreshed"
|
|
by_id = {j["id"]: j for j in store}
|
|
assert by_id["j1"]["model_snapshot"] == "new-model"
|
|
assert by_id["j1"]["provider_snapshot"] == "openrouter"
|
|
# no_agent job keeps its (irrelevant) old snapshot untouched.
|
|
assert by_id["j2"]["model_snapshot"] == "old"
|
|
# pinned job keeps None.
|
|
assert by_id["j3"]["model_snapshot"] is None
|
|
|