feat(cron): pre-dispatch configuration validation (blocked_config + alert-once)
Validate a job's configuration BEFORE any agent machinery is constructed: - missing provider API key (AuthError from a read-only resolve_runtime_provider probe; skipped when a fallback_providers chain is configured, since auth-fallback may rescue the run) - attached skill not ready (skill_view readiness_status=setup_needed — missing required env vars / commands / credential files) - delivery platform unknown or unconnected (deliver=local/origin/all are never checked; gateway-config load failures fail open) On a failing check run_job returns a [blocked_config]-marked error without constructing AIAgent/MCP/etc, so a misconfigured job never burns an LLM call. run_one_job records last_status='blocked_config' and delivers the alert exactly ONCE across ticks (persisted preflight_alerted bit — the alert-once shape from the #73506 dead-pin auto-pause); the next healthy run clears the marker so a future break re-alerts. Every preflight check fails open: only an affirmative misconfiguration verdict blocks. Config: cron.preflight (default true); `cron.preflight: false` restores the old fail-during-run behavior. Documented in the cron user guide and config defaults. mark_job_run gains an optional status= override (unblocked call shape unchanged) and drops preflight_alerted on any successful run. Tests: tests/cron/test_preflight_config.py (blocked_config + no agent + single alert across two ticks, healthy job unaffected, recovery clears dedup, fallback-chain rescue, opt-out restores old behavior, skill readiness miss, unknown delivery platform, deliver=local never loads gateway config). Full tests/cron/ + cronjob tool suite green (525 tests). Ported from: paperclipai/paperclip execution-semantics §5 (MIT); in-repo precedent: #27948, #73506
This commit is contained in:
Teknium
parent
04e8a661f2
commit
ed903f953e
51
cron/jobs.py
51
cron/jobs.py
|
|
@ -1974,8 +1974,44 @@ def remove_job(job_id: str) -> bool:
|
|||
return False
|
||||
|
||||
|
||||
def _set_preflight_alerted(job_id: str, value: bool) -> bool:
|
||||
"""Set/clear the preflight alert-dedup marker; return the PRIOR value.
|
||||
|
||||
The marker records that the operator was already alerted about this
|
||||
job's blocked configuration, so the scheduler alerts exactly once and
|
||||
stays silent on subsequent ticks until the config heals (same
|
||||
alert-once shape as the dead-pin auto-pause in #73506). Persisted on
|
||||
the job record so the dedup survives gateway restarts.
|
||||
"""
|
||||
with _jobs_lock():
|
||||
jobs = load_jobs()
|
||||
for i, job in enumerate(jobs):
|
||||
if job["id"] == job_id:
|
||||
prior = bool(job.get("preflight_alerted"))
|
||||
if value:
|
||||
job["preflight_alerted"] = True
|
||||
else:
|
||||
job.pop("preflight_alerted", None)
|
||||
if prior != value:
|
||||
jobs[i] = job
|
||||
save_jobs(jobs)
|
||||
return prior
|
||||
return False
|
||||
|
||||
|
||||
def mark_preflight_alerted(job_id: str) -> bool:
|
||||
"""Mark the job as preflight-alerted; return True if it already was."""
|
||||
return _set_preflight_alerted(job_id, True)
|
||||
|
||||
|
||||
def clear_preflight_alerted(job_id: str) -> None:
|
||||
"""Clear the preflight alert-dedup marker (config validates again)."""
|
||||
_set_preflight_alerted(job_id, False)
|
||||
|
||||
|
||||
def mark_job_run(job_id: str, success: bool, error: Optional[str] = None,
|
||||
delivery_error: Optional[str] = None):
|
||||
delivery_error: Optional[str] = None,
|
||||
status: Optional[str] = None):
|
||||
"""
|
||||
Mark a job as having been run.
|
||||
|
||||
|
|
@ -1984,6 +2020,12 @@ def mark_job_run(job_id: str, success: bool, error: Optional[str] = None,
|
|||
|
||||
``delivery_error`` is tracked separately from the agent error — a job
|
||||
can succeed (agent produced output) but fail delivery (platform down).
|
||||
|
||||
``status`` overrides the derived ``last_status`` ("ok"/"error") with a
|
||||
specific terminal status for this run — e.g. ``"blocked_config"`` when
|
||||
the pre-dispatch configuration validation refused to run the agent
|
||||
(T1-26), so `cronjob list` distinguishes "your config is broken" from
|
||||
"the run itself failed".
|
||||
"""
|
||||
with _jobs_lock():
|
||||
jobs = load_jobs()
|
||||
|
|
@ -1991,8 +2033,13 @@ def mark_job_run(job_id: str, success: bool, error: Optional[str] = None,
|
|||
if job["id"] == job_id:
|
||||
now = _hermes_now().isoformat()
|
||||
job["last_run_at"] = now
|
||||
job["last_status"] = "ok" if success else "error"
|
||||
job["last_status"] = status or ("ok" if success else "error")
|
||||
job["last_error"] = error if not success else None
|
||||
# A healthy run means the configuration validates again — drop
|
||||
# the preflight alert-dedup marker so a FUTURE config break
|
||||
# re-alerts instead of being silently swallowed.
|
||||
if success:
|
||||
job.pop("preflight_alerted", None)
|
||||
# Track delivery failures separately — cleared on successful delivery
|
||||
job["last_delivery_error"] = delivery_error
|
||||
# Clear any external-fire claim so a re-armed recurring job can
|
||||
|
|
|
|||
|
|
@ -2905,6 +2905,217 @@ def _guard_job_credential_exfil(job: dict) -> None:
|
|||
raise RuntimeError(f"Cron job '{job_id}' blocked for safety: {err}")
|
||||
|
||||
|
||||
# Marker prefix stamped into the error string returned by ``run_job`` when the
|
||||
# pre-dispatch configuration validation (T1-26) refuses to run the agent.
|
||||
# ``run_one_job`` keys off it to record ``last_status='blocked_config'`` and to
|
||||
# apply the alert-once dedup. The ``:silent`` variant means "already alerted on
|
||||
# a previous tick — do not deliver again".
|
||||
BLOCKED_CONFIG_MARKER = "[blocked_config]"
|
||||
BLOCKED_CONFIG_SILENT_MARKER = "[blocked_config:silent]"
|
||||
|
||||
|
||||
def _cron_preflight_enabled(cfg: dict) -> bool:
|
||||
"""Whether cron pre-dispatch configuration validation is enabled.
|
||||
|
||||
Default ON; only the literal boolean ``false`` under ``cron.preflight``
|
||||
opts out (mirrors ``cron_model_drift_guard_enabled`` semantics).
|
||||
"""
|
||||
cron_cfg = (cfg or {}).get("cron")
|
||||
if not isinstance(cron_cfg, dict):
|
||||
return True
|
||||
return cron_cfg.get("preflight", True) is not False
|
||||
|
||||
|
||||
def _preflight_check_provider_key(job: dict, cfg: dict) -> Optional[str]:
|
||||
"""READ-ONLY probe: would provider resolution fail for lack of a key?
|
||||
|
||||
Mirrors the effective requested-provider computation from run_job's
|
||||
resolution block without any side effects on the run. When a fallback
|
||||
chain is configured the check is skipped entirely — the existing
|
||||
auth-fallback path may legitimately rescue a missing primary key, so
|
||||
blocking here would break that contract (and burning zero LLM calls is
|
||||
already guaranteed by the fallback resolution being config-local).
|
||||
"""
|
||||
try:
|
||||
if get_fallback_chain(cfg):
|
||||
return None
|
||||
except Exception:
|
||||
return None # fail-open: never block on a preflight-internal error
|
||||
|
||||
_cron_cfg = cfg.get("cron") if isinstance(cfg.get("cron"), dict) else {}
|
||||
requested = (
|
||||
job.get("provider")
|
||||
or str((_cron_cfg or {}).get("model_provider") or "").strip()
|
||||
or None
|
||||
)
|
||||
model = job.get("model") or os.getenv("HERMES_MODEL") or ""
|
||||
|
||||
from hermes_cli.auth import AuthError
|
||||
|
||||
try:
|
||||
from hermes_cli.runtime_provider import resolve_runtime_provider
|
||||
|
||||
kwargs = {"requested": requested, "target_model": model}
|
||||
if job.get("base_url"):
|
||||
kwargs["explicit_base_url"] = job.get("base_url")
|
||||
resolve_runtime_provider(**kwargs)
|
||||
except AuthError as exc:
|
||||
return (
|
||||
f"provider credential missing: {exc}. "
|
||||
"Set the provider API key in .env (or `hermes setup`), or pin a "
|
||||
"working provider via `cronjob action=update job_id="
|
||||
f"{job.get('id')} provider=<p>`."
|
||||
)
|
||||
except Exception:
|
||||
# Non-auth resolution errors (bad config shapes, network probes,
|
||||
# import issues) are NOT a missing-credential condition — let the
|
||||
# real resolution path handle and report them as before.
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def _preflight_check_delivery(job: dict) -> Optional[str]:
|
||||
"""Check the job's delivery target(s) resolve to configured platforms.
|
||||
|
||||
``local``/``origin`` (and the ``all`` routing token) need no gateway
|
||||
credentials and are never checked — a deliver=local job must not pay a
|
||||
gateway-config load. For concrete platform targets, an unknown platform
|
||||
always blocks; a known platform additionally blocks when the gateway
|
||||
config is loadable and reports it unconnected (enabled + credentials —
|
||||
the same source `cron_delivery_targets` uses). Gateway-config load
|
||||
failures fail OPEN so a transient config hiccup never wedges delivery
|
||||
that would have worked.
|
||||
"""
|
||||
deliver_value = _normalize_deliver_value(job.get("deliver", "local"))
|
||||
platform_parts: list[str] = []
|
||||
for part in deliver_value.split(","):
|
||||
part = part.strip()
|
||||
if not part or part.lower() in {"local", "origin", "all"}:
|
||||
continue
|
||||
platform_parts.append(part.split(":", 1)[0].strip())
|
||||
if not platform_parts:
|
||||
return None
|
||||
|
||||
connected: Optional[set] = None
|
||||
for platform_name in platform_parts:
|
||||
if not _is_known_delivery_platform(platform_name):
|
||||
return (
|
||||
f"delivery platform '{platform_name}' is not a known cron "
|
||||
"delivery target. Fix the job's `deliver` value or configure "
|
||||
"the platform's gateway credentials."
|
||||
)
|
||||
if connected is None:
|
||||
try:
|
||||
from gateway.config import load_gateway_config
|
||||
|
||||
gateway_config = load_gateway_config()
|
||||
connected = {
|
||||
p.value for p in gateway_config.get_connected_platforms()
|
||||
}
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"preflight: gateway config unavailable — skipping "
|
||||
"delivery credential check", exc_info=True,
|
||||
)
|
||||
return None # fail-open
|
||||
if platform_name.lower() not in connected:
|
||||
return (
|
||||
f"delivery platform '{platform_name}' has no gateway "
|
||||
"credentials configured (not connected). Configure it via "
|
||||
"`hermes setup` or change the job's `deliver` target."
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def _preflight_check_skills(job: dict) -> Optional[str]:
|
||||
"""Check attached skills report ready (no missing required env/commands).
|
||||
|
||||
Consults the same ``readiness_status`` payload ``skill_view`` computes
|
||||
for interactive use. Skills that fail to load at all are left to the
|
||||
existing skipped-skill handling in ``_build_job_prompt`` (fail-open):
|
||||
this check only blocks on an affirmative "setup needed" verdict, i.e.
|
||||
the skill exists but its required environment is missing — a run that
|
||||
is guaranteed to misfire.
|
||||
"""
|
||||
skills = job.get("skills")
|
||||
if skills is None:
|
||||
legacy = job.get("skill")
|
||||
skills = [legacy] if legacy else []
|
||||
elif isinstance(skills, str):
|
||||
skills = [skills]
|
||||
skill_names = [str(name).strip() for name in skills if str(name).strip()]
|
||||
if not skill_names:
|
||||
return None
|
||||
|
||||
from tools.skills_tool import skill_view
|
||||
|
||||
for skill_name in skill_names:
|
||||
try:
|
||||
payload = json.loads(skill_view(skill_name))
|
||||
except Exception:
|
||||
continue # unreadable/missing skill → existing skip handling
|
||||
if not isinstance(payload, dict) or not payload.get("success"):
|
||||
continue
|
||||
if (
|
||||
payload.get("setup_needed")
|
||||
or payload.get("readiness_status") == "setup_needed"
|
||||
):
|
||||
missing = [
|
||||
f"env ${name}"
|
||||
for name in payload.get(
|
||||
"missing_required_environment_variables"
|
||||
) or []
|
||||
]
|
||||
missing += [
|
||||
f"command '{name}'"
|
||||
for name in payload.get("missing_required_commands") or []
|
||||
]
|
||||
missing += [
|
||||
f"credential file {name}"
|
||||
for name in payload.get("missing_credential_files") or []
|
||||
]
|
||||
detail = ", ".join(missing) or "required setup incomplete"
|
||||
return (
|
||||
f"attached skill '{skill_name}' is not ready: missing "
|
||||
f"{detail}. Provide the missing prerequisites or detach the "
|
||||
"skill from this job."
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def _preflight_job_config(job: dict, cfg: dict) -> Optional[str]:
|
||||
"""Pre-dispatch configuration validation (T1-26).
|
||||
|
||||
Returns a human-readable reason when the job's configuration cannot
|
||||
produce a successful run — missing provider API key, unconfigured
|
||||
delivery platform, or an attached skill with missing required env —
|
||||
so the caller can refuse the run BEFORE any agent machinery is
|
||||
constructed and no LLM call is burned. Returns ``None`` when the
|
||||
configuration validates (or when a check cannot be evaluated: every
|
||||
check fails open, so preflight can only ever block on an affirmative
|
||||
misconfiguration verdict).
|
||||
|
||||
Same fail-before-spend spirit as the #44585 drift guard and the
|
||||
fail-loud-on-hidden-tools direction in #27948; alert dedup follows the
|
||||
alert-once pattern from the dead-pin auto-pause (#73506).
|
||||
"""
|
||||
for name, check in (
|
||||
("provider_key", lambda: _preflight_check_provider_key(job, cfg)),
|
||||
("skills", lambda: _preflight_check_skills(job)),
|
||||
("delivery", lambda: _preflight_check_delivery(job)),
|
||||
):
|
||||
try:
|
||||
reason = check()
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"preflight check %s raised — failing open", name, exc_info=True
|
||||
)
|
||||
continue
|
||||
if reason:
|
||||
return reason
|
||||
return None
|
||||
|
||||
|
||||
def run_job(
|
||||
job: dict, *, defer_agent_teardown: Optional[list] = None,
|
||||
extra_prompt: Optional[str] = None,
|
||||
|
|
@ -3539,6 +3750,75 @@ def run_job(
|
|||
# off-host call is ever made with a stored key.
|
||||
_guard_job_credential_exfil(job)
|
||||
|
||||
# ---------------------------------------------------------------
|
||||
# Pre-dispatch configuration validation (T1-26).
|
||||
#
|
||||
# A job whose configuration cannot possibly produce a successful
|
||||
# run — missing provider API key (no fallback chain), unready
|
||||
# attached skill, unconfigured delivery platform — is refused HERE,
|
||||
# before AIAgent is constructed and before the resolution below can
|
||||
# feed a doomed runtime into it, so a misconfigured job never burns
|
||||
# an LLM call. run_one_job keys off the BLOCKED_CONFIG_MARKER in
|
||||
# the returned error to record last_status='blocked_config' and
|
||||
# alert exactly once (dedup persisted via the job's
|
||||
# `preflight_alerted` bit — the #73506 alert-once shape).
|
||||
# Runs after the wake-gate/prompt build so silent script ticks stay
|
||||
# silent. Opt-out: `cron.preflight: false` in config.yaml.
|
||||
# ---------------------------------------------------------------
|
||||
_pf_reason = None
|
||||
try:
|
||||
if _cron_preflight_enabled(_cfg):
|
||||
_pf_reason = _preflight_job_config(job, _cfg)
|
||||
if not _pf_reason and job.get("preflight_alerted"):
|
||||
# Configuration validates again — clear the alert-once
|
||||
# marker so a FUTURE config break re-alerts.
|
||||
try:
|
||||
from cron.jobs import clear_preflight_alerted
|
||||
clear_preflight_alerted(job_id)
|
||||
except Exception:
|
||||
pass
|
||||
except Exception:
|
||||
# The validator must never take down a runnable job — fail open.
|
||||
logger.debug(
|
||||
"Job '%s': preflight validation errored — failing open",
|
||||
job_id, exc_info=True,
|
||||
)
|
||||
_pf_reason = None
|
||||
|
||||
if _pf_reason:
|
||||
logger.warning(
|
||||
"Job '%s' (ID: %s): BLOCKED by pre-dispatch config "
|
||||
"validation — %s (no LLM call was made)",
|
||||
job_name, job_id, _pf_reason,
|
||||
)
|
||||
already_alerted = False
|
||||
try:
|
||||
from cron.jobs import mark_preflight_alerted
|
||||
already_alerted = mark_preflight_alerted(job_id)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"Job '%s': could not persist preflight alert marker",
|
||||
job_id, exc_info=True,
|
||||
)
|
||||
marker = (
|
||||
BLOCKED_CONFIG_SILENT_MARKER if already_alerted
|
||||
else BLOCKED_CONFIG_MARKER
|
||||
)
|
||||
blocked_doc = (
|
||||
f"# Cron Job: {job_name}\n\n"
|
||||
f"**Job ID:** {job_id}\n"
|
||||
f"**Run Time:** {_hermes_now().strftime('%Y-%m-%d %H:%M:%S')}\n"
|
||||
f"**Status:** BLOCKED (configuration)\n\n"
|
||||
"Pre-dispatch validation found a configuration problem and "
|
||||
"the agent was NOT run (no tokens spent).\n\n"
|
||||
f"**Reason:** {_pf_reason}\n\n"
|
||||
"The job will stay blocked (without re-alerting) until the "
|
||||
"configuration is fixed; the next healthy run clears this "
|
||||
"state. Set `cron.preflight: false` in config.yaml to "
|
||||
"disable this validation."
|
||||
)
|
||||
return False, blocked_doc, "", f"{marker} {_pf_reason}"
|
||||
|
||||
primary_model_for_drift = model
|
||||
configured_provider_for_drift = (
|
||||
str(_model_cfg.get("provider") or "").strip().lower()
|
||||
|
|
@ -4223,6 +4503,7 @@ def run_one_job(
|
|||
# deferred agent is still torn down. Otherwise the outer `except` would
|
||||
# swallow the error and leak the agent's subprocesses/clients (#10200).
|
||||
delivery_error = None
|
||||
blocked_config = False
|
||||
try:
|
||||
output_file = save_job_output(job["id"], output)
|
||||
if verbose:
|
||||
|
|
@ -4245,11 +4526,41 @@ def run_one_job(
|
|||
# Deliver the final response to the origin/target chat.
|
||||
# If the agent responded with [SILENT], skip delivery (but
|
||||
# output is already saved above). Failed jobs always deliver.
|
||||
deliver_content = final_response if success else _summarize_cron_failure_for_delivery(job, error)
|
||||
#
|
||||
# Exception: a run blocked by pre-dispatch config validation
|
||||
# (T1-26) alerts exactly ONCE — the silent marker means the
|
||||
# operator was already told on a previous tick, so re-delivering
|
||||
# the same alert every tick would be spam (#73506 alert-once
|
||||
# shape).
|
||||
blocked_config_silent = (
|
||||
bool(error) and BLOCKED_CONFIG_SILENT_MARKER in str(error)
|
||||
)
|
||||
blocked_config = blocked_config_silent or (
|
||||
bool(error) and BLOCKED_CONFIG_MARKER in str(error)
|
||||
)
|
||||
if blocked_config and not success:
|
||||
# Blocked-config alert: bypass the generic failure summarizer
|
||||
# (whose auth/timeout heuristics would mislabel this as a
|
||||
# provider runtime failure) — say plainly that config
|
||||
# validation blocked the run and nothing was spent.
|
||||
_pf_text = re.sub(
|
||||
r"\[blocked_config[^\]]*\]\s*", "", str(error)
|
||||
).strip()
|
||||
deliver_content = (
|
||||
f"⛔ Cron '{job.get('name') or job['id']}' blocked by "
|
||||
f"configuration validation (no LLM call was made): "
|
||||
f"{_pf_text} "
|
||||
"This alert is sent once; the job stays blocked until "
|
||||
"the configuration is fixed."
|
||||
)
|
||||
else:
|
||||
deliver_content = final_response if success else _summarize_cron_failure_for_delivery(job, error)
|
||||
# Treat whitespace-only final responses the same as empty
|
||||
# responses: do not deliver a blank message, and let the
|
||||
# empty-response guard below mark the run as a soft failure.
|
||||
should_deliver = bool(deliver_content.strip())
|
||||
if blocked_config_silent:
|
||||
should_deliver = False
|
||||
unresolved_origin = False
|
||||
# Cron silence suppression — see _is_cron_silence_response. Replaces the
|
||||
# old `SILENT_MARKER in ...upper()` substring check, which both leaked
|
||||
|
|
@ -4286,7 +4597,13 @@ def run_one_job(
|
|||
error = "Agent completed but produced empty response (model error, timeout, or misconfiguration)"
|
||||
|
||||
if not _consume_interrupted_flag(job["id"]):
|
||||
mark_job_run(job["id"], success, error, delivery_error=delivery_error)
|
||||
if blocked_config:
|
||||
mark_job_run(
|
||||
job["id"], success, error, delivery_error=delivery_error,
|
||||
status="blocked_config",
|
||||
)
|
||||
else:
|
||||
mark_job_run(job["id"], success, error, delivery_error=delivery_error)
|
||||
normalized_deliver = _normalize_deliver_value(job.get("deliver", "local"))
|
||||
if delivery_error:
|
||||
delivery_outcome = "failed"
|
||||
|
|
|
|||
|
|
@ -2196,6 +2196,14 @@ DEFAULT_CONFIG = {
|
|||
},
|
||||
|
||||
"cron": {
|
||||
# Pre-dispatch configuration validation (T1-26): before constructing
|
||||
# any agent machinery for a job, verify the provider API key resolves
|
||||
# (unless a fallback chain is configured), attached skills are ready
|
||||
# (required env/commands present), and delivery platforms are
|
||||
# configured. A failing job is recorded as last_status=blocked_config
|
||||
# with ONE alert (no re-alert every tick) and NO LLM call is made.
|
||||
# Set to false to restore the old behavior (fail during the run).
|
||||
"preflight": True,
|
||||
# Fail closed when an unpinned job's current global model/provider
|
||||
# differs from its creation-time snapshot. This prevents unattended
|
||||
# jobs from silently inheriting a paid default. Set to false only when
|
||||
|
|
|
|||
|
|
@ -0,0 +1,342 @@
|
|||
"""Cron pre-dispatch configuration validation (T1-26).
|
||||
|
||||
A job whose configuration cannot possibly produce a successful run — missing
|
||||
provider API key, unready attached skill (missing required env), unknown
|
||||
delivery platform — must be blocked BEFORE any agent machinery is constructed:
|
||||
|
||||
- ``last_status`` becomes ``blocked_config`` (not a generic ``error``),
|
||||
- exactly ONE alert is delivered (no re-alert every tick — same
|
||||
alert-once spirit as the dead-pin auto-pause in #73506),
|
||||
- the agent is NEVER constructed, so no LLM call is burned.
|
||||
|
||||
``cron.preflight: false`` in config.yaml restores the old behavior (the run
|
||||
proceeds to resolution and fails loudly every tick).
|
||||
|
||||
Related precedent: #27948 (fail-loud for hidden tools — same fail-before-run
|
||||
spirit, different check) and #44585 (drift guard: skip-run-no-spend shape).
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
||||
|
||||
import cron.jobs as cron_jobs
|
||||
from cron.scheduler import run_job
|
||||
import cron.scheduler as sched
|
||||
|
||||
|
||||
_RUNTIME = {
|
||||
"api_key": "test-key",
|
||||
"base_url": "https://example.invalid/v1",
|
||||
"provider": "openrouter",
|
||||
"api_mode": "chat_completions",
|
||||
}
|
||||
|
||||
|
||||
def _job(**overrides):
|
||||
job = {
|
||||
"id": "pf-test",
|
||||
"name": "preflight test",
|
||||
"prompt": "hello",
|
||||
"enabled": True,
|
||||
"state": "scheduled",
|
||||
"schedule": {"kind": "interval", "minutes": 5, "display": "every 5m"},
|
||||
"deliver": "local",
|
||||
"model": None,
|
||||
"provider": None,
|
||||
"base_url": None,
|
||||
}
|
||||
job.update(overrides)
|
||||
return job
|
||||
|
||||
|
||||
class _AuthErrorFactory:
|
||||
"""Raise a real AuthError from hermes_cli.auth."""
|
||||
|
||||
def __call__(self, **kwargs):
|
||||
from hermes_cli.auth import AuthError
|
||||
|
||||
raise AuthError("No API key configured for provider 'openrouter'")
|
||||
|
||||
|
||||
def _run_job_patched(job, tmp_path, *, resolve=None, skill_view=None):
|
||||
"""Drive run_job with the standard cron-test seams patched.
|
||||
|
||||
Returns (success, output, final_response, error, agent_constructed).
|
||||
"""
|
||||
fake_db = MagicMock()
|
||||
patches = [
|
||||
patch("cron.scheduler._hermes_home", tmp_path),
|
||||
patch("cron.scheduler._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.SessionDB", return_value=fake_db),
|
||||
patch("tools.mcp_tool.discover_mcp_tools", return_value=[]),
|
||||
]
|
||||
if resolve is None:
|
||||
patches.append(
|
||||
patch(
|
||||
"hermes_cli.runtime_provider.resolve_runtime_provider",
|
||||
return_value=dict(_RUNTIME),
|
||||
)
|
||||
)
|
||||
else:
|
||||
patches.append(
|
||||
patch(
|
||||
"hermes_cli.runtime_provider.resolve_runtime_provider",
|
||||
side_effect=resolve,
|
||||
)
|
||||
)
|
||||
if skill_view is not None:
|
||||
patches.append(patch("tools.skills_tool.skill_view", side_effect=skill_view))
|
||||
|
||||
with 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
|
||||
from contextlib import ExitStack
|
||||
|
||||
with ExitStack() as stack:
|
||||
for p in patches:
|
||||
stack.enter_context(p)
|
||||
success, output, final_response, error = run_job(job)
|
||||
agent_constructed = mock_agent_cls.called
|
||||
return success, output, final_response, error, agent_constructed
|
||||
|
||||
|
||||
class TestMissingProviderKeyBlocks:
|
||||
def test_missing_key_blocked_config_no_agent(self, tmp_path):
|
||||
"""Missing provider key (AuthError, no fallback chain) → blocked_config,
|
||||
agent never constructed, no LLM run burned."""
|
||||
job = _job()
|
||||
with cron_jobs.use_cron_store(tmp_path):
|
||||
cron_jobs.save_jobs([job])
|
||||
success, output, final_response, error, agent_constructed = \
|
||||
_run_job_patched(job, tmp_path, resolve=_AuthErrorFactory())
|
||||
|
||||
assert agent_constructed is False
|
||||
assert success is False
|
||||
assert error is not None
|
||||
assert "[blocked_config]" in error
|
||||
assert "blocked" in output.lower() or "BLOCKED" in output
|
||||
|
||||
def test_single_alert_across_two_ticks_and_blocked_status(self, tmp_path):
|
||||
"""Two ticks of a blocked job through run_one_job deliver exactly ONE
|
||||
alert and persist last_status='blocked_config'."""
|
||||
job = _job()
|
||||
deliveries = []
|
||||
|
||||
def fake_deliver(job, content, adapters=None, loop=None):
|
||||
deliveries.append(content)
|
||||
return None
|
||||
|
||||
with cron_jobs.use_cron_store(tmp_path):
|
||||
cron_jobs.save_jobs([job])
|
||||
fake_db = MagicMock()
|
||||
for _tick in range(2):
|
||||
fresh = [j for j in cron_jobs.load_jobs() if j["id"] == job["id"]][0]
|
||||
with patch("cron.scheduler._hermes_home", tmp_path), \
|
||||
patch("cron.scheduler._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.SessionDB", return_value=fake_db), \
|
||||
patch("tools.mcp_tool.discover_mcp_tools", return_value=[]), \
|
||||
patch("hermes_cli.runtime_provider.resolve_runtime_provider",
|
||||
side_effect=_AuthErrorFactory()), \
|
||||
patch.object(sched, "_deliver_result", side_effect=fake_deliver), \
|
||||
patch("run_agent.AIAgent") as mock_agent_cls:
|
||||
ok = sched.run_one_job(fresh)
|
||||
assert ok is True
|
||||
assert mock_agent_cls.called is False
|
||||
|
||||
stored = [j for j in cron_jobs.load_jobs() if j["id"] == job["id"]][0]
|
||||
|
||||
assert stored["last_status"] == "blocked_config"
|
||||
assert len(deliveries) == 1, (
|
||||
f"expected exactly one alert across two ticks, got {len(deliveries)}: "
|
||||
f"{deliveries!r}"
|
||||
)
|
||||
assert "blocked" in deliveries[0].lower()
|
||||
|
||||
def test_fallback_chain_rescues_missing_primary_key(self, tmp_path):
|
||||
"""A configured fallback chain means a missing primary key does NOT
|
||||
block — the existing auth-fallback path handles it."""
|
||||
(tmp_path / "config.yaml").write_text(
|
||||
"fallback_providers:\n"
|
||||
" - provider: openrouter\n"
|
||||
" model: z-ai/glm-5.2\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
calls = []
|
||||
|
||||
def resolve(**kwargs):
|
||||
calls.append(kwargs.get("requested"))
|
||||
if kwargs.get("requested") in (None, ""):
|
||||
from hermes_cli.auth import AuthError
|
||||
|
||||
raise AuthError("no key")
|
||||
return {**_RUNTIME, "provider": "openrouter"}
|
||||
|
||||
job = _job()
|
||||
with cron_jobs.use_cron_store(tmp_path):
|
||||
cron_jobs.save_jobs([job])
|
||||
success, output, final_response, error, agent_constructed = \
|
||||
_run_job_patched(job, tmp_path, resolve=resolve)
|
||||
|
||||
assert agent_constructed is True
|
||||
assert success is True
|
||||
assert error is None
|
||||
|
||||
|
||||
class TestHealthyJobUnaffected:
|
||||
def test_healthy_job_runs_normally(self, tmp_path):
|
||||
job = _job()
|
||||
with cron_jobs.use_cron_store(tmp_path):
|
||||
cron_jobs.save_jobs([job])
|
||||
success, output, final_response, error, agent_constructed = \
|
||||
_run_job_patched(job, tmp_path)
|
||||
|
||||
assert success is True
|
||||
assert error is None
|
||||
assert final_response == "ok"
|
||||
assert agent_constructed is True
|
||||
|
||||
def test_recovery_clears_alert_marker(self, tmp_path):
|
||||
"""After a blocked tick, a healthy tick clears the alert-dedup marker
|
||||
so a FUTURE config break re-alerts."""
|
||||
job = _job()
|
||||
with cron_jobs.use_cron_store(tmp_path):
|
||||
cron_jobs.save_jobs([job])
|
||||
# Tick 1: blocked.
|
||||
_run_job_patched(job, tmp_path, resolve=_AuthErrorFactory())
|
||||
stored = [j for j in cron_jobs.load_jobs() if j["id"] == job["id"]][0]
|
||||
assert stored.get("preflight_alerted")
|
||||
# Tick 2: key restored → healthy run clears the marker.
|
||||
fresh = [j for j in cron_jobs.load_jobs() if j["id"] == job["id"]][0]
|
||||
success, *_rest, agent_constructed = _run_job_patched(fresh, tmp_path)
|
||||
assert success is True
|
||||
assert agent_constructed is True
|
||||
stored = [j for j in cron_jobs.load_jobs() if j["id"] == job["id"]][0]
|
||||
assert not stored.get("preflight_alerted")
|
||||
|
||||
|
||||
class TestOptOut:
|
||||
def test_preflight_false_restores_old_behavior(self, tmp_path):
|
||||
"""cron.preflight: false → job proceeds to resolution and fails the
|
||||
old way (error status, re-alerts every tick, no blocked_config)."""
|
||||
(tmp_path / "config.yaml").write_text(
|
||||
"cron:\n preflight: false\n", encoding="utf-8"
|
||||
)
|
||||
job = _job()
|
||||
deliveries = []
|
||||
|
||||
def fake_deliver(job, content, adapters=None, loop=None):
|
||||
deliveries.append(content)
|
||||
return None
|
||||
|
||||
with cron_jobs.use_cron_store(tmp_path):
|
||||
cron_jobs.save_jobs([job])
|
||||
fake_db = MagicMock()
|
||||
for _tick in range(2):
|
||||
fresh = [j for j in cron_jobs.load_jobs() if j["id"] == job["id"]][0]
|
||||
with patch("cron.scheduler._hermes_home", tmp_path), \
|
||||
patch("cron.scheduler._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.SessionDB", return_value=fake_db), \
|
||||
patch("tools.mcp_tool.discover_mcp_tools", return_value=[]), \
|
||||
patch("hermes_cli.runtime_provider.resolve_runtime_provider",
|
||||
side_effect=_AuthErrorFactory()), \
|
||||
patch.object(sched, "_deliver_result", side_effect=fake_deliver), \
|
||||
patch("run_agent.AIAgent") as mock_agent_cls:
|
||||
sched.run_one_job(fresh)
|
||||
assert mock_agent_cls.called is False
|
||||
|
||||
stored = [j for j in cron_jobs.load_jobs() if j["id"] == job["id"]][0]
|
||||
|
||||
assert stored["last_status"] == "error"
|
||||
assert len(deliveries) == 2 # old behavior: alert every tick
|
||||
|
||||
|
||||
class TestSkillReadiness:
|
||||
def test_unready_skill_blocks(self, tmp_path):
|
||||
"""An attached skill whose readiness_status is setup_needed (missing
|
||||
required env) blocks the run before the agent is constructed."""
|
||||
payload = json.dumps(
|
||||
{
|
||||
"success": True,
|
||||
"content": "# needy skill\nbody",
|
||||
"readiness_status": "setup_needed",
|
||||
"setup_needed": True,
|
||||
"missing_required_environment_variables": ["NEEDY_API_KEY"],
|
||||
"missing_required_commands": [],
|
||||
}
|
||||
)
|
||||
|
||||
def fake_skill_view(name, *args, **kwargs):
|
||||
return payload
|
||||
|
||||
job = _job(skills=["needy-skill"])
|
||||
with cron_jobs.use_cron_store(tmp_path):
|
||||
cron_jobs.save_jobs([job])
|
||||
success, output, final_response, error, agent_constructed = \
|
||||
_run_job_patched(job, tmp_path, skill_view=fake_skill_view)
|
||||
|
||||
assert agent_constructed is False
|
||||
assert success is False
|
||||
assert error is not None and "[blocked_config]" in error
|
||||
assert "NEEDY_API_KEY" in f"{error} {output}"
|
||||
|
||||
def test_ready_skill_runs(self, tmp_path):
|
||||
payload = json.dumps(
|
||||
{
|
||||
"success": True,
|
||||
"content": "# ready skill\nbody",
|
||||
"readiness_status": "available",
|
||||
"setup_needed": False,
|
||||
"missing_required_environment_variables": [],
|
||||
}
|
||||
)
|
||||
|
||||
def fake_skill_view(name, *args, **kwargs):
|
||||
return payload
|
||||
|
||||
job = _job(skills=["ready-skill"])
|
||||
with cron_jobs.use_cron_store(tmp_path):
|
||||
cron_jobs.save_jobs([job])
|
||||
success, output, final_response, error, agent_constructed = \
|
||||
_run_job_patched(job, tmp_path, skill_view=fake_skill_view)
|
||||
|
||||
assert success is True
|
||||
assert agent_constructed is True
|
||||
|
||||
|
||||
class TestDeliveryPlatform:
|
||||
def test_unknown_delivery_platform_blocks(self, tmp_path):
|
||||
job = _job(deliver="notaplatform")
|
||||
with cron_jobs.use_cron_store(tmp_path):
|
||||
cron_jobs.save_jobs([job])
|
||||
with patch("cron.scheduler._is_known_delivery_platform",
|
||||
return_value=False):
|
||||
success, output, final_response, error, agent_constructed = \
|
||||
_run_job_patched(job, tmp_path)
|
||||
|
||||
assert agent_constructed is False
|
||||
assert success is False
|
||||
assert error is not None and "[blocked_config]" in error
|
||||
assert "notaplatform" in f"{error} {output}"
|
||||
|
||||
def test_local_delivery_never_touches_gateway_config(self, tmp_path):
|
||||
"""deliver=local jobs must not load gateway config in preflight."""
|
||||
job = _job(deliver="local")
|
||||
with cron_jobs.use_cron_store(tmp_path):
|
||||
cron_jobs.save_jobs([job])
|
||||
with patch("gateway.config.load_gateway_config",
|
||||
side_effect=AssertionError("gateway config loaded")):
|
||||
success, *_rest, agent_constructed = _run_job_patched(job, tmp_path)
|
||||
|
||||
assert success is True
|
||||
assert agent_constructed is True
|
||||
|
|
@ -31,7 +31,7 @@ def _patch_pipeline(monkeypatch, *, success=True, output="out", final="final res
|
|||
calls.append(("deliver", job["id"]))
|
||||
return None
|
||||
|
||||
def fake_mark(jid, ok, err=None, delivery_error=None):
|
||||
def fake_mark(jid, ok, err=None, delivery_error=None, **_kw):
|
||||
calls.append(("mark", jid, ok))
|
||||
|
||||
monkeypatch.setattr(s, "run_job", fake_run_job)
|
||||
|
|
|
|||
|
|
@ -67,6 +67,33 @@ Every morning at 9am, check Hacker News for AI news and send me a summary on Tel
|
|||
|
||||
Hermes will use the unified `cronjob` tool internally.
|
||||
|
||||
## Pre-dispatch configuration validation
|
||||
|
||||
Before constructing any agent machinery for a scheduled run, the scheduler
|
||||
validates that the job's configuration can actually produce a successful run:
|
||||
|
||||
- the provider API key resolves (skipped when a `fallback_providers` chain is
|
||||
configured, since the fallback path may rescue a missing primary key),
|
||||
- attached skills are ready (no missing required environment variables,
|
||||
commands, or credential files),
|
||||
- delivery platform targets are known and have gateway credentials configured
|
||||
(`local`/`origin` targets are never checked).
|
||||
|
||||
When validation fails, the job's `last_status` becomes `blocked_config`, ONE
|
||||
alert is delivered (it is not repeated every tick), and **no LLM call is
|
||||
made** — a misconfigured job never spends tokens. The next healthy run clears
|
||||
the blocked state so a future configuration break alerts again.
|
||||
|
||||
To disable the validation and restore the old behavior (the run proceeds and
|
||||
fails during execution):
|
||||
|
||||
```yaml
|
||||
cron:
|
||||
preflight: false
|
||||
```
|
||||
|
||||
Or: `hermes config set cron.preflight false`
|
||||
|
||||
## Letting unpinned jobs track global defaults
|
||||
|
||||
The model/provider drift guard is enabled by default. If your unpinned cron
|
||||
|
|
|
|||
Loading…
Reference in New Issue