ed703493da
One gateway per host is the only supported topology (#100896). This is the convergence path that makes it true on an upgraded machine. `gateway migrate --multiplex` is now defined by TOPOLOGY, not by the config flag: a host is converged when no secondary profile owns a gateway process or a supervisor unit. A half-migrated host (flag flipped, a unit left behind, a crash between the two) therefore converges on the next run instead of reporting "already multiplexed" — that flag-only test made the re-run a no-op on exactly the host that needed it. The manifest is the resume record: present means UNFINISHED, so a confirmed convergence clears it and any manifest found on disk is resumed from rather than refused. Windows Scheduled Tasks (and the Startup-folder fallback `gateway install` writes when it cannot register a task) are now detected and removed like any other unit; Windows used to be a flat refusal with hand-migration instructions. s6 stays refused, because there the per-profile gateways are slots the container's own boot registers — and that boot now registers named slots DOWN unconditionally, which is the s6 leg of the convergence. It used to read `gateway.multiplex_profiles`, so the UNSET default (on) booted the slots anyway and the image shipped the opt-out topology by accident. A running gateway is never SIGTERMed silently: the plan names every pid it will signal and prints before anything is signalled, with a dry run that changes nothing. `--standalone` is gone. Reinstalling per-profile services is not a supported target, so there is no rollback command; the machinery survives only as the compensator inside a single failed apply, because the one outcome worse than a per-profile fleet is a profile with no gateway at all. `gateway.multiplex_profiles: false` is retired as a topology opt-out: it still parses and still carries the runtime mode every scoped code path reads, but it can no longer pin a second gateway process — it resolves like an unset key, warns, and points at the migration. The unset path is not optimistic (it refuses to multiplex while a real blocker holds), so a host that genuinely cannot fold still comes up standalone and says why. The eager Desktop activation stops reading it too: a stale `false` there made a multi-home host serve a second profile with the LAUNCH profile's credentials. The host gateway lock flips from observe-only to a refusal. `host_attach` already attaches/rescans/refuses before anything binds, but it reads a RECORD published a moment after the owner starts, so two gateways launched together can both see no owner. The lock is the only atomic arbiter of that race. The refusal names the owner, prints the migrate command, and exits 75 (EX_TEMPFAIL) so a supervisor RETRIES — never the parking 78 — and on the retry the record exists and the attach path resolves it properly. `--force`, `--replace` and an unopenable lock dir are not refusals.
911 lines
44 KiB
Python
911 lines
44 KiB
Python
"""Cron subcommand for hermes CLI."""
|
|
|
|
import contextlib
|
|
import json
|
|
import re
|
|
import sys
|
|
from datetime import timezone
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Iterable, List, Optional
|
|
|
|
PROJECT_ROOT = Path(__file__).parent.parent.resolve()
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from hermes_cli.colors import Colors, color
|
|
|
|
|
|
def _normalize_skills(single_skill=None, skills: Optional[Iterable[str]] = None) -> Optional[List[str]]:
|
|
"""Deduped, stripped skill names; None when neither argument was given."""
|
|
if skills is None and single_skill is None:
|
|
return None
|
|
normalized: List[str] = []
|
|
for item in list(skills) if skills is not None else [single_skill]:
|
|
text = str(item or "").strip()
|
|
if text and text not in normalized:
|
|
normalized.append(text)
|
|
return normalized
|
|
|
|
|
|
def _cron_api(**kwargs):
|
|
from tools.cronjob_tools import cronjob as cronjob_tool
|
|
return json.loads(cronjob_tool(**kwargs))
|
|
|
|
|
|
def _active_cron_provider_name() -> str:
|
|
"""Resolved cron scheduler provider name ('builtin', 'chronos', …); 'builtin' on failure."""
|
|
try:
|
|
from cron.scheduler_provider import resolve_cron_scheduler
|
|
return resolve_cron_scheduler().name or "builtin"
|
|
except Exception:
|
|
return "builtin"
|
|
|
|
|
|
def _builtin_gateway_liveness() -> Optional[bool]:
|
|
"""Tri-state scheduler readiness (None = probe failed).
|
|
|
|
Local gateways use process liveness; served satellites also require their own fresh heartbeat. External providers use their own machinery and are exempt.
|
|
"""
|
|
try:
|
|
if _active_cron_provider_name() != "builtin":
|
|
return True
|
|
# The runtime lock is held for exactly the gateway's lifetime — more reliable than PID
|
|
# scanning (find_gateway_pids transiently misses the gateway right after a restart, and
|
|
# inside the gateway it must never say "not running"). A crashing probe is "unknown".
|
|
with contextlib.suppress(Exception):
|
|
from gateway.status import is_gateway_runtime_lock_active
|
|
if is_gateway_runtime_lock_active():
|
|
return True
|
|
from hermes_cli.gateway import (
|
|
find_gateway_pids, named_profile_served_by_running_multiplexer)
|
|
if find_gateway_pids():
|
|
return True
|
|
if not named_profile_served_by_running_multiplexer():
|
|
return False
|
|
# List/create and status require a fresh heartbeat from the satellite's own store.
|
|
from cron.jobs import get_ticker_heartbeat_age
|
|
return _ticker_age_is_fresh(get_ticker_heartbeat_age())
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _warn_if_gateway_not_running() -> None:
|
|
"""Warn at create/list time when the scheduler is not ready; stay silent on an unknown probe result."""
|
|
if _builtin_gateway_liveness() is not False:
|
|
return
|
|
print(color(" ⚠ Scheduler is not ready: no gateway or no fresh profile heartbeat.", Colors.YELLOW))
|
|
print(color(" If no gateway is running: hermes gateway install\n"
|
|
" sudo hermes gateway install --system # Linux servers\n"
|
|
" Check status: hermes cron status", Colors.DIM))
|
|
|
|
|
|
def _format_lateness(seconds: float) -> str:
|
|
"""Render a lateness duration compactly: '31m', '2h 30m', '45s'."""
|
|
try:
|
|
seconds = max(0, int(seconds))
|
|
except (TypeError, ValueError):
|
|
return "?"
|
|
if seconds < 60:
|
|
return f"{seconds}s"
|
|
minutes, _ = divmod(seconds, 60)
|
|
hours, minutes = divmod(minutes, 60)
|
|
days, hours = divmod(hours, 24)
|
|
parts = [(days, "d"), (hours, "h"), (minutes if not days else 0, "m")]
|
|
return " ".join(f"{n}{unit}" for n, unit in parts if n) or "0m"
|
|
|
|
|
|
def _dispatch_kind_label(kind) -> Optional[str]:
|
|
return {"catch_up": "catch-up after missed fire", "late": "late"}.get(kind)
|
|
|
|
|
|
def _next_run_overdue_seconds(next_run_at: Any) -> Optional[float]:
|
|
"""Seconds the stored ``next_run_at`` is already in the past (negative while still
|
|
upcoming); None when it is not a parseable ISO timestamp.
|
|
|
|
Parses through the scheduler's own ``_parse_aware`` so the CLI and the ticker agree on
|
|
the instant (mixed UTC offsets, DST folds, legacy naive stamps read as system-local).
|
|
"""
|
|
from cron.jobs import _parse_aware
|
|
from hermes_time import now
|
|
dt = _parse_aware(next_run_at)
|
|
if dt is None:
|
|
return None
|
|
# Same-tzinfo subtraction is wall-clock arithmetic in Python; compare instants.
|
|
return (now().astimezone(timezone.utc) - dt.astimezone(timezone.utc)).total_seconds()
|
|
|
|
|
|
def _next_run_row(job: Dict[str, Any]) -> tuple[str, str]:
|
|
"""``("Next run" | "Overdue", value)`` for one job.
|
|
|
|
A stamp parked past `cron doctor`'s grace on a job that is supposed to fire is the only
|
|
user-visible trace of a dead scheduler; never present it as an upcoming run (#114309).
|
|
"""
|
|
stamp = job.get("next_run_at", "?")
|
|
overdue_s = _next_run_overdue_seconds(stamp)
|
|
if (overdue_s is None or overdue_s <= _OVERDUE_GRACE_SECONDS
|
|
or not job.get("enabled", True) or job.get("state") in {"paused", "completed"}):
|
|
return ("Next run", stamp)
|
|
return ("Overdue", color(f"{stamp} ({_format_lateness(overdue_s)} ago — the job has not fired; "
|
|
"is the scheduler running?)", Colors.YELLOW))
|
|
|
|
|
|
def _dispatch_display(dispatch: dict) -> Optional[str]:
|
|
"""One-line scheduled-vs-actual dispatch summary; None when the stamp is malformed.
|
|
|
|
On-time dispatches render dim; late/catch-up dispatches render loudly so a run fired long
|
|
after gateway downtime doesn't look like an ordinary success.
|
|
|
|
See #99879.
|
|
"""
|
|
if not isinstance(dispatch, dict):
|
|
return None
|
|
scheduled, actual, kind = (dispatch.get(k) for k in ("scheduled_at", "dispatched_at", "kind"))
|
|
if not scheduled or not actual or not kind:
|
|
return None
|
|
lateness = _format_lateness(dispatch.get("lateness_seconds", 0))
|
|
if kind == "on_time":
|
|
return color(f"on time (scheduled {scheduled})", Colors.DIM)
|
|
label = _dispatch_kind_label(kind) or "late"
|
|
return (color(f"⚠ {label}: ", Colors.YELLOW) + f"scheduled {scheduled}, ran {actual} "
|
|
+ color(f"({lateness} late)", Colors.YELLOW))
|
|
|
|
|
|
def _print_banner(title: str) -> None:
|
|
"""Boxed cyan section header shared by ``cron list`` and ``cron incidents``."""
|
|
print()
|
|
rule = "─" * 73
|
|
for line in (f"┌{rule}┐", "│" + " " * 25 + title.ljust(48) + "│", f"└{rule}┘"):
|
|
print(color(line, Colors.CYAN))
|
|
print()
|
|
|
|
|
|
def _unverified_targets(unverified) -> str:
|
|
return ", ".join(map(str, unverified)) if isinstance(unverified, list) else str(unverified)
|
|
|
|
|
|
_STATE_BADGES = {"paused": ("[paused]", Colors.YELLOW), "completed": ("[completed]", Colors.BLUE)}
|
|
|
|
|
|
def cron_list(show_all: bool = False):
|
|
"""List all scheduled jobs."""
|
|
from cron.jobs import effective_job_state, list_jobs
|
|
jobs = list_jobs(include_disabled=True)
|
|
if not show_all:
|
|
jobs = [
|
|
job for job in jobs
|
|
if job.get("enabled", True) or effective_job_state(job) == "paused"
|
|
]
|
|
|
|
if not jobs:
|
|
print(color("No scheduled jobs.\nCreate one with 'hermes cron create ...' "
|
|
"or the /cron command in chat.", Colors.DIM))
|
|
return
|
|
|
|
_print_banner("Scheduled Jobs")
|
|
|
|
for job in jobs:
|
|
# effective_job_state honours the scheduler flag — never [paused] when enabled=true.
|
|
badge = _STATE_BADGES.get(effective_job_state(job)) or (
|
|
("[active]", Colors.GREEN) if job.get("enabled", True) else ("[disabled]", Colors.RED))
|
|
print(f" {color(job.get('id', '?'), Colors.YELLOW)} {color(*badge)}")
|
|
for label, value in _job_rows(job):
|
|
print(f" {label + ':':<11}{value}")
|
|
for line in _job_warnings(job):
|
|
print(f" {line}")
|
|
print()
|
|
|
|
_warn_if_gateway_not_running()
|
|
|
|
|
|
def _last_run_display(job: Dict[str, Any]) -> str:
|
|
last_status = job["last_status"]
|
|
if last_status == "ok":
|
|
return color("ok", Colors.GREEN)
|
|
if last_status == "delivery_queued":
|
|
return color("finished; delivery is still in progress", Colors.YELLOW)
|
|
if last_status == "delivery_failed":
|
|
# Agent succeeded but the result never reached the user — not green; last_error is None.
|
|
return color(f"ran, but the result was not delivered ({_short_reason(job.get('last_delivery_error'))}). "
|
|
f"{_delivery_fix_hint(job)}", Colors.YELLOW)
|
|
display = color(f"{last_status}: {job.get('last_error', '?')}", Colors.RED)
|
|
streak = int(job.get("failure_streak") or 0)
|
|
if streak >= 2:
|
|
display += color(f" ({streak} failures in a row)", Colors.RED)
|
|
return display
|
|
|
|
|
|
def _job_rows(job: Dict[str, Any]) -> List[tuple[str, str]]:
|
|
"""``(label, value)`` detail rows for one job in ``cron list``."""
|
|
# `repeat` / `deliver` may be present-but-null (dict-default only covers a missing key).
|
|
repeat_info = job.get("repeat") or {}
|
|
repeat_times = repeat_info.get("times")
|
|
# `deliver` may be present-but-null in the job record (same pitfall as `repeat` above), so coalesce to
|
|
# the default rather than relying on the dict-default, which only applies to a missing key. A null value
|
|
# would otherwise reach `", ".join(None)` and crash the whole listing (#32896).
|
|
deliver = job.get("deliver") or ["local"]
|
|
skills = job.get("skills") or ([job["skill"]] if job.get("skill") else [])
|
|
monitor_source = job.get("monitor_script") or job.get("monitor_url")
|
|
mon_state = job.get("monitor_state") or {}
|
|
latest_execution = job.get("latest_execution") or {}
|
|
optional = [
|
|
("Skills", ", ".join(skills) if skills else ""),
|
|
("Script", job.get("script")),
|
|
("Monitor", f"{monitor_source} (agent runs only on output change)" if monitor_source
|
|
else ""),
|
|
("Changed", mon_state.get("last_changed_at") if monitor_source else ""),
|
|
("Mode", color("no-agent", Colors.DIM) + " (script stdout delivered directly)"
|
|
if job.get("no_agent") else ""),
|
|
("Workdir", job.get("workdir")),
|
|
("Last run", f"{job.get('last_run_at', '?')} {_last_run_display(job)}"
|
|
if job.get("last_status") else ""),
|
|
("Dispatch", _dispatch_display(job.get("last_dispatch"))),
|
|
("Execution", f"{latest_execution.get('status', '?')} {latest_execution.get('id', '?')}"
|
|
if latest_execution else "")]
|
|
return [
|
|
("Name", job.get("name", "(unnamed)")),
|
|
("Schedule", job.get("schedule_display", job.get("schedule", {}).get("value", "?"))),
|
|
("Repeat", f"{repeat_info.get('completed', 0)}/{repeat_times}" if repeat_times else "∞"),
|
|
_next_run_row(job),
|
|
("Deliver", deliver if isinstance(deliver, str) else ", ".join(deliver)),
|
|
] + [(label, value) for label, value in optional if value]
|
|
|
|
|
|
def _short_reason(text: Any, limit: int = 120) -> str:
|
|
"""First line of an adapter/error blob, whitespace-collapsed and capped, or 'no details'."""
|
|
first = str(text or "").strip().splitlines()
|
|
reason = " ".join(first[0].split()) if first else ""
|
|
return (reason[: limit - 1] + "…") if len(reason) > limit else (reason or "no details")
|
|
|
|
|
|
def _delivery_fix_hint(job: Dict[str, Any]) -> str:
|
|
return (f"Check the target with `hermes cron status` or change it with "
|
|
f"`hermes cron edit {job.get('id', '<id>')} --deliver <target>`.")
|
|
|
|
|
|
def _missed_fire_issue(job: Dict[str, Any], fire_err: Dict[str, Any]) -> str:
|
|
return (f"missed scheduled fire at {fire_err.get('at', '?')}: {_short_reason(fire_err['detail'])}. "
|
|
"The messaging gateway was unreachable. Run `hermes gateway restart`, then "
|
|
f"`hermes cron run {job.get('id', '<id>')}` to run it now.")
|
|
|
|
|
|
def _job_warnings(job: Dict[str, Any]) -> List[str]:
|
|
"""Delivery / fire warning lines for one job in ``cron list``."""
|
|
lines = []
|
|
if queued := job.get("last_delivery_queued"):
|
|
lines.append(f"Delivery still in progress (the result was handed off but not confirmed yet): {queued}")
|
|
if job.get("last_delivery_error"):
|
|
lines.append(f"{color('⚠ The result was not delivered:', Colors.YELLOW)} "
|
|
f"{_short_reason(job['last_delivery_error'])}. {_delivery_fix_hint(job)}")
|
|
# A live adapter acked the last send but returned no message_id / raw_response
|
|
# (Slack/Matrix/Mattermost shape): accepted as delivered, but say so here.
|
|
if unverified := job.get("last_delivery_unverified"):
|
|
lines.append(f"{color('⚠ Delivery UNVERIFIED:', Colors.YELLOW)} adapter acked "
|
|
f"{_unverified_targets(unverified)} without message_id/raw_response")
|
|
fire_err = job.get("last_fire_error")
|
|
if isinstance(fire_err, dict) and fire_err.get("detail"):
|
|
lines.append(color(f"⚠ {_missed_fire_issue(job, fire_err)}", Colors.RED))
|
|
return lines
|
|
|
|
|
|
def cron_tick():
|
|
"""Run due jobs once and exit."""
|
|
from cron.scheduler import CronTickYielded, tick
|
|
try:
|
|
tick(verbose=True)
|
|
except CronTickYielded as exc:
|
|
# Inert for a one-shot CLI (no boot fingerprint); report cleanly rather than traceback.
|
|
print(color(f"✗ {exc}", Colors.YELLOW))
|
|
print(" A fresher gateway process owns the runtime lock and will fire due jobs; this "
|
|
"stale process yielded its tick.")
|
|
return 1
|
|
except OSError as exc:
|
|
# Real lock-acquisition failures (EMFILE, EACCES) propagate; they are not contention.
|
|
# For the one-shot CLI surface, report cleanly instead of dumping a traceback; the gateway ticker
|
|
# loop handles its own retry. See #87644.
|
|
print(color(f"✗ Cron tick failed: {exc}", Colors.RED))
|
|
print(" Check `hermes cron status` and the gateway log for details.")
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def cron_runs(job_id: Optional[str] = None, limit: int = 20):
|
|
"""Show indexed durable cron execution history."""
|
|
from cron.executions import list_executions
|
|
records = list_executions(job_id=job_id, limit=limit)
|
|
if not records:
|
|
print("No cron execution attempts recorded.")
|
|
return
|
|
for record in records:
|
|
print(f"{record.get('id', '?')} {record.get('status', '?'):<9} "
|
|
f"job={record.get('job_id', '?')} source={record.get('source', '?')} "
|
|
f"{record.get('claimed_at', '?')}")
|
|
if record.get("error"):
|
|
print(f" {record['error']}")
|
|
|
|
|
|
_INCIDENT_STATE_COLORS = {"detected": Colors.RED, "alerted": Colors.YELLOW, "resolved": Colors.GREEN,
|
|
"closed": Colors.DIM}
|
|
|
|
|
|
def cron_incidents(args) -> int:
|
|
"""List (``[--state <s>]``) or ``ack <id>`` durable cron failure incidents.
|
|
|
|
Acking closes an incident so its failure ping stays silent until the error signature changes.
|
|
"""
|
|
from cron.incidents import ack_incident, list_incidents
|
|
action = getattr(args, "incident_action", "list")
|
|
if action == "ack":
|
|
incident_id = getattr(args, "incident_id", None)
|
|
if not incident_id:
|
|
print(color("✗ Incident ID required: hermes cron incidents ack <incident_id>", Colors.RED))
|
|
return 1
|
|
if ack_incident(incident_id):
|
|
print(color(f"✓ Incident {incident_id} acknowledged (closed).", Colors.GREEN))
|
|
else:
|
|
print(color(f"Incident {incident_id} not found or already closed.", Colors.YELLOW))
|
|
return 0
|
|
|
|
state = getattr(args, "state", None)
|
|
incidents = list_incidents(state=state)
|
|
if not incidents:
|
|
print(color("No cron failure incidents recorded.", Colors.DIM))
|
|
if state:
|
|
print(color(f" (filtered by state '{state}')", Colors.DIM))
|
|
return 0
|
|
|
|
_print_banner("Cron Failure Incidents")
|
|
for inc in incidents:
|
|
state_display = color(inc["state"], _INCIDENT_STATE_COLORS.get(inc["state"], Colors.DIM))
|
|
error_text = re.sub(r"\s+", " ", inc.get("error") or "").strip()
|
|
if len(error_text) > 160:
|
|
error_text = error_text[:157].rstrip() + "..."
|
|
rows = [("Job", inc["job_id"]), ("Type", inc.get("failure_type", "unknown")),
|
|
("First seen", inc.get("first_seen_at", "?")),
|
|
("Last seen", inc.get("last_seen_at", "?")), ("Error", error_text),
|
|
("Output", inc.get("output_file"))]
|
|
print(f" {color(inc['id'], Colors.YELLOW)} {state_display}")
|
|
for label, value in rows:
|
|
if label != "Output" or value:
|
|
print(f" {label + ':':<12}{value}")
|
|
print()
|
|
print(color(f" {len(incidents)} incident(s) | ack one with: hermes cron incidents ack <id>",
|
|
Colors.DIM))
|
|
return 0
|
|
|
|
|
|
_PERMISSION_HINT = (" Hint: jobs.json may be owned by another user (e.g. rewritten by a root "
|
|
"`docker exec hermes hermes cron ...`). Fix ownership to match the gateway "
|
|
"user, and prefer `docker exec -u <uid>:<gid>`.")
|
|
_FD_EXHAUSTION_HINT = (" Hint: the ticker hit file-descriptor exhaustion (EMFILE). The scheduler "
|
|
"now retries with backoff and attempts fd reclamation, but if the leak "
|
|
"persists, restart the gateway to recover scheduling.")
|
|
|
|
|
|
def _ticker_age_is_fresh(age: Optional[float]) -> bool:
|
|
from cron.jobs import TICKER_INTERVAL_SECONDS
|
|
return age is not None and age <= TICKER_INTERVAL_SECONDS * 3 + 20
|
|
|
|
|
|
def _print_ticker_health(pids: list, restart_command: str = "hermes gateway restart") -> None:
|
|
"""Report builtin-ticker liveness for a gateway process known to be alive.
|
|
|
|
The ticker THREAD can die silently or stay alive while every tick fails, so check both
|
|
the liveness heartbeat and the last-successful-tick marker before saying "will fire".
|
|
"""
|
|
# See #32612, #32895.
|
|
from cron.jobs import (
|
|
get_ticker_heartbeat_age, get_ticker_last_error, get_ticker_success_age)
|
|
from cron.scheduler import _is_fd_exhaustion_text as _cron_is_fd_exhaustion_text
|
|
from cron.scheduler import stale_code_yield_labels
|
|
hb_age = get_ticker_heartbeat_age()
|
|
ok_age = get_ticker_success_age()
|
|
last_error = get_ticker_last_error()
|
|
pid_line = f" PID: {', '.join(map(str, pids))}" if pids else None
|
|
|
|
def _warn(headline: str) -> None:
|
|
print(color(headline, Colors.YELLOW))
|
|
if pid_line:
|
|
print(pid_line)
|
|
|
|
if hb_age is None:
|
|
# Ticker never started (non-cron profile, gateway just started, or a config issue).
|
|
_warn("⚠ Gateway is running but the cron ticker has not reported a heartbeat.")
|
|
print(" Cron jobs will NOT fire until the ticker writes its first heartbeat.\n"
|
|
" If the gateway just started, wait ~60s and re-run `hermes cron status`.\n"
|
|
f" If heartbeat never appears, restart: {restart_command}")
|
|
elif not _ticker_age_is_fresh(hb_age): # ticker thread is gone
|
|
_warn("⚠ Gateway is running but the cron ticker looks STALLED — "
|
|
f"no heartbeat for {int(hb_age)}s (expected every ~60s).")
|
|
print(f" Cron jobs may NOT be firing. Restart: {restart_command}")
|
|
elif (skew := stale_code_yield_labels(last_error)) is not None:
|
|
# `hermes update` moved the checkout under a running gateway: its ticker yields every
|
|
# tick (heartbeat stays fresh, nothing dispatches) until the process is restarted (#117275).
|
|
_warn("⚠ Gateway is running STALE code — its cron ticker yields every tick and "
|
|
"fires NOTHING.")
|
|
print(color(f" Booted on {skew[0]}, checkout is now at {skew[1]} "
|
|
"(the code was updated under the running gateway).", Colors.RED))
|
|
print(f" Restart it onto the new code: {restart_command}")
|
|
elif (ok_age is not None and not _ticker_age_is_fresh(ok_age)) or (ok_age is None and last_error):
|
|
# Loop alive but every tick fails (or has never succeeded since boot).
|
|
_warn("⚠ Gateway and cron ticker are running, but no tick has "
|
|
f"succeeded {'in ' + str(int(ok_age)) + 's' if ok_age is not None else 'yet'} "
|
|
"— ticks may be failing.")
|
|
if last_error:
|
|
# WHY ticks fail: root-rewritten jobs.json (PermissionError) or fd exhaustion.
|
|
# Show WHY ticks fail — e.g. a root-rewritten jobs.json (PermissionError) that silently locked
|
|
# out the ticker's uid for ~14h in the field (#68483), or fd exhaustion (EMFILE) that used to
|
|
# stall the scheduler invisibly (#87644).
|
|
print(color(f" Last tick error: {last_error}", Colors.RED))
|
|
if "Permission denied" in last_error:
|
|
print(color(_PERMISSION_HINT, Colors.YELLOW))
|
|
elif _cron_is_fd_exhaustion_text(last_error):
|
|
print(color(_FD_EXHAUSTION_HINT, Colors.YELLOW))
|
|
print(" Check the gateway log for 'Cron tick error'.")
|
|
else:
|
|
print(color("✓ Gateway is running — cron jobs will fire automatically", Colors.GREEN))
|
|
if pid_line:
|
|
print(pid_line)
|
|
if hb_age is not None:
|
|
print(f" Ticker heartbeat: {int(hb_age)}s ago")
|
|
|
|
|
|
def cron_status():
|
|
"""Show cron execution status."""
|
|
from cron.jobs import list_jobs
|
|
from hermes_cli.gateway import find_gateway_pids, named_profile_served_by_running_multiplexer
|
|
from hermes_cli.profiles import get_active_profile_name
|
|
print()
|
|
|
|
provider = _active_cron_provider_name()
|
|
if provider != "builtin":
|
|
# External providers fire via webhook: no ticker thread / heartbeat file by design, so
|
|
# the liveness heuristics would always say "stalled".
|
|
print(color(f"✓ Cron provider: {provider} — jobs fire via the managed scheduler, "
|
|
"not the in-process ticker.", Colors.GREEN))
|
|
print(color(" (No ticker heartbeat is expected for an external provider; "
|
|
"due jobs are delivered by an authenticated webhook.)", Colors.DIM))
|
|
else:
|
|
from gateway.host_topology import host_gateway_serving
|
|
active = get_active_profile_name()
|
|
# FIRST question under multiplex-only: is the HOST gateway alive and does it tick THIS
|
|
# profile? Starting from find_gateway_pids() (argv `-p <name>`) made every served profile
|
|
# report "not running" and told the user to start a SECOND host process.
|
|
host = None
|
|
with contextlib.suppress(Exception):
|
|
host = host_gateway_serving(active)
|
|
pids = [] if host is not None else find_gateway_pids()
|
|
gateway_alive_via_lock = False
|
|
served_by_multiplexer = False
|
|
if host is None and not pids:
|
|
# The pid scan transiently misses a live gateway right after a restart; the runtime
|
|
# lock proves the process is alive. Declare "not running" only when both agree.
|
|
with contextlib.suppress(Exception):
|
|
# Same false-alarm class the cronjob tool fixed (#95947): the pid scan can transiently miss
|
|
# a live gateway (just after a restart) while the runtime lock — held for exactly the
|
|
# gateway's lifetime — proves the ticker's process is alive.
|
|
from gateway.status import get_running_pid, is_gateway_runtime_lock_active
|
|
gateway_alive_via_lock = is_gateway_runtime_lock_active()
|
|
lock_pid = get_running_pid() if gateway_alive_via_lock else None
|
|
pids = [lock_pid] if lock_pid else pids
|
|
# Multiplexer identity does not establish the active profile's ticker health.
|
|
if not gateway_alive_via_lock:
|
|
served_by_multiplexer = named_profile_served_by_running_multiplexer()
|
|
if host is not None:
|
|
print(f" Scheduler host: {host.describe()}")
|
|
# `hermes gateway restart` exits 78 for a served NAMED profile
|
|
# (_guard_named_profile_under_multiplexer): the one host process is the default's.
|
|
_print_ticker_health([host.pid], restart_command="hermes --profile default gateway restart")
|
|
elif pids or gateway_alive_via_lock or served_by_multiplexer:
|
|
if served_by_multiplexer:
|
|
print(" Scheduler host: the host gateway (multiplexing this profile)")
|
|
_print_ticker_health([], restart_command="hermes --profile default gateway restart")
|
|
else:
|
|
_print_ticker_health(pids)
|
|
else:
|
|
print(color("✗ No gateway is running on this host — cron jobs will NOT fire", Colors.RED))
|
|
# When scheduling last worked before the host went away: without this, a
|
|
# 7h-overdue job still reads as a normal upcoming "Next run" (#114309).
|
|
with contextlib.suppress(Exception):
|
|
from cron.jobs import TICKER_INTERVAL_SECONDS, get_ticker_heartbeat_age
|
|
hb_age = get_ticker_heartbeat_age()
|
|
if hb_age is not None and hb_age > TICKER_INTERVAL_SECONDS * 3 + 20:
|
|
print(color(" Scheduler last ticked "
|
|
f"{_format_lateness(hb_age)} ago — jobs that came due "
|
|
"since then have not fired.", Colors.YELLOW))
|
|
print("\n Start the ONE host gateway (it multiplexes every profile, this one included):\n"
|
|
" hermes --profile default gateway install # user service\n"
|
|
" sudo hermes --profile default gateway install --system # Linux servers: boot-time service\n"
|
|
" hermes --profile default gateway run # Or run in foreground")
|
|
if active not in ("default", "custom"):
|
|
print("\n It serves this profile automatically. If a per-profile service or gateway\n"
|
|
" from an older release is still installed, fold it in (preflight + dry run):\n"
|
|
" hermes --profile default gateway migrate --multiplex --dry-run\n"
|
|
" hermes --profile default gateway migrate --multiplex\n"
|
|
" Check: hermes cron status from this profile should show its ticker heartbeat.\n")
|
|
|
|
print()
|
|
_print_active_jobs_summary(list_jobs(include_disabled=False))
|
|
print()
|
|
|
|
|
|
def _print_active_jobs_summary(jobs) -> None:
|
|
"""Print the '<N> active job(s)' + next-run line shared by every status path."""
|
|
if not jobs:
|
|
print(" No active jobs")
|
|
return
|
|
from cron.jobs import _parse_aware
|
|
|
|
# Stored stamps carry mixed UTC offsets (an interval job keeps last_run_at's offset, a cron
|
|
# job its configured zone), so order by instant, never by ISO text; display the stored stamp.
|
|
# `_parse_aware` hands back one shared ZoneInfo, and Python compares same-tzinfo datetimes
|
|
# by wall clock (wrong across a DST fold) — normalise to UTC before ordering.
|
|
next_runs = [(parsed.astimezone(timezone.utc), j["next_run_at"]) for j in jobs
|
|
if (parsed := _parse_aware(j.get("next_run_at"))) is not None]
|
|
print(f" {len(jobs)} active job(s)")
|
|
if next_runs:
|
|
earliest = min(next_runs, key=lambda run: run[0])[1]
|
|
overdue_by = _next_run_overdue_seconds(earliest)
|
|
if overdue_by is not None and overdue_by > _OVERDUE_GRACE_SECONDS:
|
|
# #114309: a dead scheduler leaves next_run_at stranded in the past; presenting it
|
|
# as an upcoming "Next run" hides the outage. Same 15m grace as `cron doctor`
|
|
# (_OVERDUE_GRACE_SECONDS) so a job a few minutes behind the ticker's own
|
|
# cadence doesn't flash OVERDUE here while doctor still calls it healthy.
|
|
print(color(f" ⚠ Next run {earliest} is OVERDUE — passed "
|
|
f"{_format_lateness(overdue_by)} ago but the job has not fired "
|
|
"(is the scheduler running?)", Colors.YELLOW))
|
|
else:
|
|
print(f" Next run: {earliest}")
|
|
# Post-downtime late fires show at status level, not just per-job in `cron list`.
|
|
late = [j for j in jobs if isinstance(j.get("last_dispatch"), dict)
|
|
and j["last_dispatch"].get("kind") in ("late", "catch_up")]
|
|
if late:
|
|
print()
|
|
print(color(f" ⚠ {len(late)} job(s) last fired late:", Colors.YELLOW))
|
|
for j in late:
|
|
d = j["last_dispatch"]
|
|
late_by = _format_lateness(d.get("lateness_seconds", 0))
|
|
print(f" {j.get('id', '?')} {j.get('name', '(unnamed)')}: {_dispatch_kind_label(d.get('kind'))}, "
|
|
f"scheduled {d.get('scheduled_at', '?')}, ran {d.get('dispatched_at', '?')} "
|
|
+ color(f"({late_by} late)", Colors.YELLOW))
|
|
|
|
|
|
def _scripts_dir_for_cron() -> Path:
|
|
"""Scripts dir for cron jobs — via ``CRON_DIR`` so monkeypatched cron storage is honoured."""
|
|
from cron.jobs import CRON_DIR
|
|
return CRON_DIR.parent / "scripts"
|
|
|
|
|
|
def _script_health_issue(script: str) -> Optional[str]:
|
|
"""Human-readable script issue, or ``None`` when the path is OK."""
|
|
scripts_dir = _scripts_dir_for_cron().resolve()
|
|
raw = Path(script).expanduser()
|
|
path = raw.resolve() if raw.is_absolute() else (scripts_dir / raw).resolve()
|
|
try:
|
|
path.relative_to(scripts_dir)
|
|
except ValueError:
|
|
return f"script resolves outside {scripts_dir}: {script!r}"
|
|
if not path.exists():
|
|
return f"script not found: {path}"
|
|
if not path.is_file():
|
|
return f"script path is not a file: {path}"
|
|
return None
|
|
|
|
|
|
# A busy tick can push dispatch a few minutes late; only a next_run_at parked well in the past
|
|
# means the job is silently not firing (ticker dead, gateway down, wedged fire-claim).
|
|
# `cron status`'s OVERDUE line shares this grace so status, list, and doctor tell one
|
|
# consistent story about when a job counts as overdue.
|
|
_OVERDUE_GRACE_SECONDS = 15 * 60
|
|
|
|
|
|
def _next_run_overdue_issue(next_run: str) -> Optional[str]:
|
|
"""Issue string when ``next_run_at`` is parked in the past."""
|
|
overdue_s = _next_run_overdue_seconds(next_run)
|
|
if overdue_s is None:
|
|
return f"next_run_at is not a valid timestamp: {next_run!r}"
|
|
if overdue_s <= _OVERDUE_GRACE_SECONDS:
|
|
return None
|
|
amount = f"{overdue_s / 3600:.1f}h" if overdue_s >= 3600 else f"{overdue_s / 60:.0f}m"
|
|
return f"next_run_at is {amount} overdue — job is not firing (is the scheduler running?)"
|
|
|
|
|
|
def _cron_doctor_issues_for_job(job: Dict[str, Any]) -> List[str]:
|
|
issues: List[str] = []
|
|
last_status = str(job.get("last_status") or "").strip().lower()
|
|
# "delivery_failed" = the agent run succeeded; the delivery issue below reports it.
|
|
if last_status and last_status not in {"ok", "delivery_failed", "delivery_queued"}:
|
|
issues.append(f"last run failed: {str(job.get('last_error') or 'unknown error').strip()}")
|
|
if delivery_err := str(job.get("last_delivery_error") or "").strip():
|
|
issues.append(f"last run finished but the result was not delivered ({_short_reason(delivery_err)}). "
|
|
f"{_delivery_fix_hint(job)}")
|
|
if unverified := job.get("last_delivery_unverified"):
|
|
issues.append("last delivery unverified (adapter acked without evidence): "
|
|
+ _unverified_targets(unverified))
|
|
# Dispatch records measure lateness, not whether the scheduler process was running.
|
|
if isinstance(dispatch := job.get("last_dispatch"), dict):
|
|
if label := _dispatch_kind_label(dispatch.get("kind")):
|
|
issues.append(f"last fire was {label} (scheduled {dispatch.get('scheduled_at', '?')}, "
|
|
f"{_format_lateness(dispatch.get('lateness_seconds', 0))} late). "
|
|
"This warning clears at the next on-time fire.")
|
|
if isinstance(fire_err := job.get("last_fire_error"), dict) and fire_err.get("detail"):
|
|
# The handoff error survives next_run_at advancing beyond the failed dispatch.
|
|
issues.append(_missed_fire_issue(job, fire_err))
|
|
if job.get("enabled", True) and job.get("state") not in {"paused", "completed"}:
|
|
next_run = str(job.get("next_run_at") or "").strip()
|
|
issue = _next_run_overdue_issue(next_run) if next_run else "active job has no next_run_at"
|
|
if issue:
|
|
issues.append(issue)
|
|
script = str(job.get("script") or "").strip()
|
|
if job.get("no_agent") and not script:
|
|
issues.append("no-agent job has no script")
|
|
if script and (script_issue := _script_health_issue(script)):
|
|
issues.append(script_issue)
|
|
workdir = str(job.get("workdir") or "").strip()
|
|
if workdir and not Path(workdir).expanduser().exists():
|
|
issues.append(f"workdir not found: {workdir}")
|
|
return issues
|
|
|
|
|
|
def cron_doctor() -> int:
|
|
"""Run read-only cron health checks and return a shell-friendly status."""
|
|
from cron.jobs import list_jobs
|
|
jobs = list_jobs(include_disabled=False)
|
|
findings = [(job, issues) for job in jobs if (issues := _cron_doctor_issues_for_job(job))]
|
|
if not findings:
|
|
print(color("✓ Cron doctor found no issues", Colors.GREEN))
|
|
note = f" Checked {len(jobs)} active job(s)." if jobs else " No active jobs configured."
|
|
print(color(note, Colors.DIM))
|
|
return 0
|
|
issue_count = sum(len(issues) for _, issues in findings)
|
|
print(color(f"Cron doctor found {issue_count} issue(s) across {len(findings)} job(s):", Colors.YELLOW))
|
|
print()
|
|
for job, issues in findings:
|
|
print(f" {color(job.get('id', '?'), Colors.YELLOW)} {job.get('name', '(unnamed)')}")
|
|
for issue in issues:
|
|
print(f" - {issue}")
|
|
print()
|
|
print(color("Review the findings above, then run `hermes cron doctor` again.", Colors.DIM))
|
|
return 1
|
|
|
|
|
|
_JOB_ARG_FIELDS = (("name", "name"), ("deliver", "deliver"), ("failure_deliver", "failure_deliver"),
|
|
("repeat", "repeat"), ("script", "script"), ("workdir", "workdir"),
|
|
("model", "model"), ("provider", "model_provider"), ("pinned", "pinned"),
|
|
("monitor_script", "monitor_script"), ("monitor_url", "monitor_url"),
|
|
("continuity", "continuity"), ("reasoning_effort", "reasoning_effort"))
|
|
|
|
|
|
def _job_api_kwargs(args) -> Dict[str, Any]:
|
|
"""Collect the create/update kwargs shared by ``cron create`` and ``cron edit``."""
|
|
return {api_key: getattr(args, attr, None) for api_key, attr in _JOB_ARG_FIELDS}
|
|
|
|
|
|
_JOB_DETAIL_LINES = (
|
|
("script", " Script: {}"),
|
|
("monitor_script", " Monitor: {} (agent runs only on output change)"),
|
|
("monitor_url", " Monitor: {} (agent runs only on output change)"),
|
|
("no_agent", " Mode: no-agent (script stdout delivered directly)"),
|
|
("continuity", " Continuity: on (each run sees the previous run's output)"),
|
|
("workdir", " Workdir: {}"))
|
|
|
|
|
|
def _print_job_details(job_data: Dict[str, Any]) -> None:
|
|
"""Print the optional Script/Monitor/Mode/Continuity/Workdir lines of a job record."""
|
|
for key, template in _JOB_DETAIL_LINES:
|
|
if job_data.get(key):
|
|
print(template.format(job_data[key]))
|
|
|
|
|
|
def cron_create(args):
|
|
# The gateway-lifecycle guard lives in cron.jobs.create_job (every creation path); a block
|
|
# surfaces as result["error"].
|
|
result = _cron_api(
|
|
action="create", schedule=args.schedule, prompt=args.prompt,
|
|
skill=getattr(args, "skill", None),
|
|
skills=_normalize_skills(getattr(args, "skill", None), getattr(args, "skills", None)),
|
|
no_agent=getattr(args, "no_agent", False) or None,
|
|
**({"paused": args.paused, "paused_reason": getattr(args, "paused_reason", None)}
|
|
if getattr(args, "paused", False) or getattr(args, "paused_reason", None) is not None else {}),
|
|
**_job_api_kwargs(args))
|
|
if not result.get("success"):
|
|
print(color(f"Failed to create job: {result.get('error', 'unknown error')}", Colors.RED))
|
|
return 1
|
|
print(color(f"Created job: {result['job_id']}", Colors.GREEN))
|
|
print(f" Name: {result['name']}\n Schedule: {result['schedule']}")
|
|
if result.get("skills"):
|
|
print(f" Skills: {', '.join(result['skills'])}")
|
|
_print_job_details(result.get("job", {}))
|
|
if not result.get("job", {}).get("enabled", True):
|
|
print(" Created PAUSED — resume to schedule, or explicitly run now.")
|
|
else:
|
|
print(f" Next run: {result['next_run_at']}")
|
|
_warn_if_gateway_not_running()
|
|
return 0
|
|
|
|
|
|
def cron_edit(args):
|
|
from cron.jobs import AmbiguousJobReference, resolve_job_ref
|
|
try:
|
|
job = resolve_job_ref(args.job_id)
|
|
except AmbiguousJobReference as exc:
|
|
print(color(str(exc), Colors.RED))
|
|
for m in exc.matches:
|
|
print(f" {m['id']} (name: {m.get('name')!r})")
|
|
return 1
|
|
if not job:
|
|
print(color(f"Job not found: {args.job_id}", Colors.RED))
|
|
return 1
|
|
existing_skills = list(job.get("skills") or ([job["skill"]] if job.get("skill") else []))
|
|
replacement_skills = _normalize_skills(getattr(args, "skill", None), getattr(args, "skills", None))
|
|
add_skills = _normalize_skills(None, getattr(args, "add_skills", None)) or []
|
|
remove_skills = set(_normalize_skills(None, getattr(args, "remove_skills", None)) or [])
|
|
|
|
final_skills = None
|
|
if getattr(args, "clear_skills", False):
|
|
final_skills = []
|
|
elif replacement_skills is not None:
|
|
final_skills = replacement_skills
|
|
elif add_skills or remove_skills:
|
|
final_skills = [skill for skill in existing_skills if skill not in remove_skills]
|
|
final_skills += [skill for skill in add_skills if skill not in final_skills]
|
|
result = _cron_api(action="update", job_id=args.job_id,
|
|
schedule=getattr(args, "schedule", None),
|
|
prompt=getattr(args, "prompt", None), skills=final_skills,
|
|
no_agent=getattr(args, "no_agent", None), **_job_api_kwargs(args))
|
|
if not result.get("success"):
|
|
print(color(f"Failed to update job: {result.get('error', 'unknown error')}", Colors.RED))
|
|
return 1
|
|
updated = result["job"]
|
|
print(color(f"Updated job: {updated['job_id']}", Colors.GREEN))
|
|
print(f" Name: {updated['name']}\n Schedule: {updated['schedule']}")
|
|
print(f" Skills: {', '.join(updated['skills'])}" if updated.get("skills") else
|
|
" Skills: none")
|
|
_print_job_details(updated)
|
|
return 0
|
|
|
|
|
|
def _job_action(action: str, job_id: str, success_verb: str) -> int:
|
|
_stateless_token = None
|
|
if action == "run":
|
|
# One-shot CLI: a background-dispatched run (daemon thread, triggered when the CLI
|
|
# inherits HERMES_SESSION_KEY) would be orphaned mid-LLM-call, leaving the execution row
|
|
# stuck 'claimed'. Declaring the channel stateless forces a synchronous run; scoped to
|
|
# this call so in-process callers (tests, embedding apps) are not tainted.
|
|
with contextlib.suppress(Exception):
|
|
# The background path in ``_try_dispatch_background_run`` triggers when the CLI inherits a
|
|
# gateway/desktop session env (HERMES_SESSION_KEY); declare the channel stateless so
|
|
# ``async_delivery_supported()`` gates it off and the run executes synchronously to completion
|
|
# instead. See #86721.
|
|
from gateway.session_context import _SESSION_ASYNC_DELIVERY
|
|
_stateless_token = _SESSION_ASYNC_DELIVERY.set(False)
|
|
try:
|
|
result = _cron_api(action=action, job_id=job_id)
|
|
finally:
|
|
if _stateless_token is not None:
|
|
_SESSION_ASYNC_DELIVERY.reset(_stateless_token)
|
|
if not result.get("success"):
|
|
print(color(f"Failed to {action} job: {result.get('error', 'unknown error')}", Colors.RED))
|
|
return 1
|
|
job = result.get("job") or result.get("removed_job") or {}
|
|
print(color(f"{success_verb} job: {job.get('name', job_id)} ({job_id})", Colors.GREEN))
|
|
if action in {"resume", "run"} and result.get("job", {}).get("next_run_at"):
|
|
print(f" Next run: {result['job']['next_run_at']}")
|
|
if action == "run":
|
|
print(f" {_run_outcome(result.get('job', {}))}")
|
|
return 0
|
|
|
|
|
|
def _run_outcome(job: Dict[str, Any]) -> str:
|
|
"""One-line verdict for a manual run.
|
|
|
|
A background-dispatched run (execution_mode="background" / delegation_id) keeps running
|
|
after this CLI exits, so report the dispatch rather than a success/failure verdict.
|
|
"""
|
|
if job.get("delegation_id"):
|
|
return f"Running in background (delegation {job['delegation_id']})."
|
|
if job.get("execution_mode") == "background":
|
|
return "Running in background."
|
|
if job.get("executed"):
|
|
return f"Ran now: {'succeeded' if job.get('execution_success') else 'failed'}."
|
|
return job.get("execution_skipped") or "It will run on the next scheduler tick."
|
|
|
|
|
|
def cron_resume(args) -> int:
|
|
"""Resume a paused job or explicitly re-arm a completed one-shot."""
|
|
run_at = getattr(args, "run_at", None)
|
|
run_now = getattr(args, "run_now", False)
|
|
if run_at and run_now:
|
|
print(color("Use exactly one of --at or --run-now.", Colors.RED))
|
|
return 1
|
|
if not run_at and not run_now:
|
|
return _job_action("resume", args.job_id, "Resumed")
|
|
from cron.jobs import AmbiguousJobReference, _hermes_now, rearm_oneshot
|
|
try:
|
|
job = rearm_oneshot(args.job_id, _hermes_now().isoformat() if run_now else run_at)
|
|
except (AmbiguousJobReference, ValueError) as exc:
|
|
print(color(f"Failed to re-arm job: {exc}", Colors.RED))
|
|
return 1
|
|
if not job:
|
|
print(color(f"Job not found: {args.job_id}", Colors.RED))
|
|
return 1
|
|
print(color(f"Re-armed job: {job.get('name', args.job_id)} ({args.job_id})", Colors.GREEN)
|
|
+ f"\n Next run: {job.get('next_run_at')}")
|
|
return 0
|
|
|
|
|
|
def cron_notepad(args) -> int:
|
|
"""Handle ``hermes cron notepad <job_id> [get|set|delete|list]`` (per-job durable KV).
|
|
|
|
A running cron agent updates its own notepad via its terminal tool; the scheduler injects
|
|
non-empty notepads into the job prompt on each run.
|
|
"""
|
|
from cron import notepad
|
|
job_id = str(getattr(args, "job_id", "") or "")
|
|
action = getattr(args, "notepad_action", None) or "list"
|
|
key = getattr(args, "key", None)
|
|
value = getattr(args, "value", None)
|
|
if not job_id:
|
|
print(color("A job ID is required.", Colors.RED))
|
|
return 1
|
|
try:
|
|
if action not in ("set", "get", "delete"): # list (default)
|
|
notes = notepad.list_notes(job_id)
|
|
if not notes:
|
|
print(color(f"Notepad for job {job_id} is empty.", Colors.DIM))
|
|
for note in notes:
|
|
print(f" {color(note['key'], Colors.YELLOW)} = {note['value']}\n"
|
|
f" {color('updated: ' + str(note['updated_at']), Colors.DIM)}")
|
|
return 0
|
|
usage_args = "set <key> <value>" if action == "set" else f"{action} <key>"
|
|
if key is None or (action == "set" and value is None):
|
|
print(color(f"Usage: hermes cron notepad <job_id> {usage_args}", Colors.RED))
|
|
return 1
|
|
if action == "set":
|
|
notepad.set_note(job_id, key, value)
|
|
print(color(f"Set notepad key '{key}' for job {job_id}.", Colors.GREEN))
|
|
return 0
|
|
if action == "get":
|
|
stored = notepad.get_note(job_id, key)
|
|
if stored is not None:
|
|
print(stored)
|
|
return 0
|
|
elif notepad.delete_note(job_id, key):
|
|
print(color(f"Deleted notepad key '{key}' for job {job_id}.", Colors.GREEN))
|
|
return 0
|
|
print(color(f"No notepad key '{key}' for job {job_id}.", Colors.YELLOW))
|
|
return 1
|
|
except ValueError as exc:
|
|
print(color(f"Notepad error: {exc}", Colors.RED))
|
|
return 1
|
|
|
|
|
|
# Late-bound lambdas keep module-level monkeypatching working; list/status/runs return None -> 0.
|
|
_CRON_SUBCOMMANDS = {
|
|
"list": lambda a: cron_list(getattr(a, "all", False)) or 0,
|
|
"status": lambda a: cron_status() or 0,
|
|
"doctor": lambda a: cron_doctor(),
|
|
"tick": lambda a: cron_tick(),
|
|
"runs": lambda a: cron_runs(getattr(a, "job_id", None), getattr(a, "limit", 20)) or 0,
|
|
"incidents": lambda a: cron_incidents(a),
|
|
"notepad": lambda a: cron_notepad(a),
|
|
"create": lambda a: cron_create(a),
|
|
"edit": lambda a: cron_edit(a),
|
|
"pause": lambda a: _job_action("pause", a.job_id, "Paused"),
|
|
"resume": lambda a: cron_resume(a),
|
|
"run": lambda a: _job_action("run", a.job_id, "Triggered"),
|
|
"remove": lambda a: _job_action("remove", a.job_id, "Removed"),
|
|
}
|
|
_CRON_SUBCOMMANDS["history"] = _CRON_SUBCOMMANDS["runs"]
|
|
_CRON_SUBCOMMANDS["add"] = _CRON_SUBCOMMANDS["create"]
|
|
_CRON_SUBCOMMANDS["rm"] = _CRON_SUBCOMMANDS["delete"] = _CRON_SUBCOMMANDS["remove"]
|
|
|
|
|
|
def cron_command(args):
|
|
"""Handle cron subcommands."""
|
|
subcmd = getattr(args, 'cron_command', None)
|
|
handler = _CRON_SUBCOMMANDS.get("list" if subcmd is None else subcmd)
|
|
if handler is not None:
|
|
return handler(args)
|
|
print(f"Unknown cron command: {subcmd}\n"
|
|
"Usage: hermes cron [list|create|edit|pause|resume|run|remove|status|runs|doctor|tick]")
|
|
sys.exit(1)
|