KrustyPlanet/hermes-agent
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

feat(cron): Suggested Cron Jobs — one surface for proposed automations

Browse source 
Hermes can propose automations and let the user accept them with one tap
via /suggestions, instead of making them assemble cron jobs by hand. Every
proposal — wherever it originates — flows through one surface.

Sources (the 'where suggestions come from'):
- catalog: curated starter automations (daily briefing, important-mail
  monitor, weekly review, workday-start reminder) via /suggestions catalog
- recipe: installing a skill that carries a metadata.hermes.recipe block
  registers a suggestion instead of auto-scheduling
- usage / integration: reserved for the background-review detector and
  account-connect triggers (sources defined; emitters land next)

Pieces:
- cron/suggestions.py — the store. add/list/accept/dismiss, dedup+latch by
  key (dismissed proposals never re-offered), pending cap so it can't become
  a nag wall. Accepting calls the existing cron.jobs.create_job — there is
  NO second job engine. Mirrors jobs.py storage (atomic writes, lock, 0600).
- cron/suggestion_catalog.py — the curated set. The important-mail monitor
  entry is where the old proactive-monitor poll->classify->surface engine
  lives now (cron/scripts/classify_items.py + the 'monitor' aux task), as ONE
  catalog automation rather than a standalone feature.
- tools/recipes.py — recipe<->job bridge; register_recipe_suggestion() makes
  a recipe source 'recipe' of this surface. recipe_to_job_spec() is the single
  translation both the direct and suggestion paths share.
- hermes_cli/suggestions_cmd.py — shared /suggestions handler (CLI + gateway
  never drift); /suggestions [accept N|dismiss N|catalog|clear].
- Wired: CommandDef + CLI dispatch (cli.py) + gateway dispatch (gateway/run.py)
  + aux 'monitor' task (config.py) + recipe-install hook (skills_hub.py).

Consent-first throughout: nothing auto-schedules; acceptance is always
explicit; dismissals latch.

Supersedes #41122 (proactive-monitor) and #41127 (recipes): both fold in here
as a catalog entry and a suggestion source respectively.

Tests: store (dedup/cap/accept/dismiss/latch), catalog seeding+idempotency,
recipe->suggestion bridge, command handler, aux config. E2E: recipe SKILL.md
-> parsed -> suggested -> accepted -> real cron job persisted to jobs.json.
This commit is contained in:
teknium1 2026-06-07 06:47:24 -07:00 • committed by Teknium
parent 4d6a133a9f
commit 9a09ea69fb
12 changed files with 1599 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

226
cron/scripts/classify_items.py Normal file
View file

@ -0,0 +1,226 @@
#!/usr/bin/env python3
"""Classify candidate items by urgency/importance and emit only the urgent ones.
The proactive-monitor pattern: a fetch step (a watcher script, an inbox dump, a
feed) produces a list of candidate items; this script scores each with a cheap
LLM and prints ONLY the items at or above a threshold. Below-threshold runs
print nothing, so a cron job wrapping this stays silent unless something
actually matters -- mirroring Poke's email monitor (fetch -> classify urgency
-> surface only what's above the bar).
Design choices:
* Uses Hermes' auxiliary client with task="monitor", so the classifier model
is configured once in config.yaml (auxiliary.monitor.{provider,model}) and
can be a cheap fast model independent of the main chat model.
* Reads items as JSON (a list of objects) from stdin or --input-file.
* One LLM call scores the whole batch (cheap, single round-trip) and returns
structured scores; we filter locally.
* Empty result -> empty stdout -> the cron job's [SILENT]/empty-stdout path
suppresses delivery. No spam on quiet intervals.
Usage (standalone):
cat items.json | python classify_items.py --threshold 7 \
--criteria "Urgent if it needs a reply today or is from my manager/family"
Usage (wired to a watcher via cron, agent mode):
Ask the agent: "Every 10 minutes, run watch_http_json.py for my inbox feed,
pipe its JSON into classify_items.py with my urgency criteria, and deliver
whatever it prints. Stay silent if it prints nothing."
Item schema (flexible): each item is an object; the classifier sees the whole
object. A "title"/"subject"/"summary"/"text" field helps it judge. An "id"
field (any of id/guid/message_id/url) is echoed back so duplicates can be
deduped upstream.
"""
from __future__ import annotations
import argparse
import json
import sys
from typing import Any, Dict, List, Optional
def _eprint(*args: Any) -> None:
print(*args, file=sys.stderr)
def _load_items(input_file: Optional[str]) -> List[Dict[str, Any]]:
raw = ""
if input_file:
with open(input_file, encoding="utf-8") as f:
raw = f.read()
else:
raw = sys.stdin.read()
raw = raw.strip()
if not raw:
return []
try:
data = json.loads(raw)
except json.JSONDecodeError as e:
_eprint(f"classify_items: input is not valid JSON: {e}")
sys.exit(2)
if isinstance(data, dict):
# Allow {"items": [...]} or a single object.
if isinstance(data.get("items"), list):
return data["items"]
return [data]
if isinstance(data, list):
return [x for x in data if isinstance(x, dict)]
_eprint("classify_items: expected a JSON list or {items: [...]}")
sys.exit(2)
def _item_id(item: Dict[str, Any], index: int) -> str:
for key in ("id", "guid", "message_id", "url", "link"):
val = item.get(key)
if val:
return str(val)
return f"item-{index}"
_CLASSIFY_INSTRUCTIONS = (
"You are an urgency classifier for a proactive assistant. You will be given "
"a numbered list of items and the user's importance criteria. Score EACH "
"item from 0 (ignore entirely) to 10 (interrupt the user now). Return ONLY a "
"JSON array, one object per item, in the same order: "
'[{"index": <int>, "score": <int 0-10>, "reason": "<short>"}]. '
"No prose, no markdown fences. Be conservative: most items should score low. "
"Only score high when the item clearly meets the user's criteria."
)
def _build_prompt(items: List[Dict[str, Any]], criteria: str) -> str:
lines = [f"USER IMPORTANCE CRITERIA:\n{criteria}\n", "ITEMS:"]
for i, item in enumerate(items):
# Show a compact view; the model sees the salient fields.
view = {
k: item[k]
for k in ("title", "subject", "summary", "text", "body", "from", "sender", "url")
if k in item
}
if not view:
view = item # fall back to the whole object
lines.append(f"[{i}] {json.dumps(view, ensure_ascii=False)[:1200]}")
lines.append(
"\nReturn the JSON array of scores now (one object per item, same order)."
)
return "\n".join(lines)
def _parse_scores(content: str, n_items: int) -> Dict[int, Dict[str, Any]]:
text = (content or "").strip()
# Tolerate accidental markdown fences.
if text.startswith("```"):
text = text.strip("`")
if "\n" in text:
text = text.split("\n", 1)[1]
try:
arr = json.loads(text)
except json.JSONDecodeError:
# Last-ditch: find the first [...] block.
start = text.find("[")
end = text.rfind("]")
if start >= 0 and end > start:
try:
arr = json.loads(text[start : end + 1])
except json.JSONDecodeError:
_eprint("classify_items: could not parse classifier output")
return {}
else:
_eprint("classify_items: classifier returned no JSON array")
return {}
out: Dict[int, Dict[str, Any]] = {}
if isinstance(arr, list):
for obj in arr:
if not isinstance(obj, dict):
continue
idx = obj.get("index")
if isinstance(idx, int) and 0 <= idx < n_items:
out[idx] = obj
return out
def main() -> int:
parser = argparse.ArgumentParser(description="Classify items by urgency; emit only urgent ones.")
parser.add_argument("--criteria", required=True, help="Plain-language importance criteria.")
parser.add_argument("--threshold", type=int, default=7, help="Minimum score (0-10) to surface. Default 7.")
parser.add_argument("--input-file", default=None, help="Read items JSON from this file instead of stdin.")
parser.add_argument("--format", choices=["text", "json"], default="text", help="Output format for surfaced items.")
args = parser.parse_args()
items = _load_items(args.input_file)
if not items:
# Nothing to classify -> silent. This is the common quiet-interval case.
return 0
# Import here so --help works without the package importable.
try:
from agent.auxiliary_client import call_llm
except Exception as e: # pragma: no cover - import guard
_eprint(f"classify_items: cannot import auxiliary client: {e}")
return 3
prompt = _build_prompt(items, args.criteria)
try:
resp = call_llm(
task="monitor",
messages=[{"role": "user", "content": prompt}],
max_tokens=1024,
temperature=0,
)
content = resp.choices[0].message.content
if not isinstance(content, str):
content = str(content) if content else ""
except Exception as e:
# Classification failure is NOT silent -- surface it so a broken monitor
# doesn't quietly swallow important items. Non-zero exit -> cron alerts.
_eprint(f"classify_items: classifier call failed: {e}")
return 4
scores = _parse_scores(content, len(items))
surfaced = []
for i, item in enumerate(items):
s = scores.get(i)
score = s.get("score") if isinstance(s, dict) else None
if isinstance(score, int) and score >= args.threshold:
surfaced.append((i, item, s))
if not surfaced:
# Below threshold -> silent. Empty stdout; cron suppresses delivery.
return 0
if args.format == "json":
out = [
{
"id": _item_id(item, i),
"score": s.get("score"),
"reason": s.get("reason", ""),
"item": item,
}
for (i, item, s) in surfaced
]
print(json.dumps(out, ensure_ascii=False, indent=2))
else:
blocks = []
for (i, item, s) in surfaced:
title = (
item.get("title")
or item.get("subject")
or item.get("summary")
or _item_id(item, i)
)
url = item.get("url") or item.get("link") or ""
reason = s.get("reason", "")
block = f"## [{s.get('score')}/10] {title}"
if url:
block += f"\n{url}"
if reason:
block += f"\n_{reason}_"
blocks.append(block)
print("\n\n".join(blocks))
return 0
if __name__ == "__main__":
sys.exit(main())

152
cron/suggestion_catalog.py Normal file
View file

@ -0,0 +1,152 @@
"""Curated catalog of starter cron-job suggestions.
These are the built-in automations Hermes can offer a new user out of the box —
the ``catalog`` source of the unified suggestion surface. Each entry is a
ready-to-run ``cron.jobs.create_job`` spec wrapped as a suggestion; the user
accepts via ``/suggestions``. Nothing here auto-schedules.
The "important-mail monitor" entry is where the old proactive-monitor engine
lives now: its ``classify_items.py`` (poll a source -> LLM-score urgency ->
surface only above-threshold) is ONE catalog automation, not a standalone
feature.
Adding a catalog entry: append a CatalogEntry. Keep prompts self-contained
(cron jobs run with no chat context) and schedules sensible. The ``job_spec``
is passed verbatim to ``create_job`` on accept.
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
__all__ = ["CatalogEntry", "CATALOG", "seed_catalog_suggestions", "classify_items_script_path"]
def classify_items_script_path() -> str:
"""Absolute path to the urgency classifier script shipped with cron/."""
return str((Path(__file__).resolve().parent / "scripts" / "classify_items.py"))
@dataclass(frozen=True)
class CatalogEntry:
"""A curated starter automation offered as a suggestion."""
key: str # stable dedup key (never re-offered once dismissed)
title: str
description: str
job_spec: Dict[str, Any] # kwargs for cron.jobs.create_job
# The curated set. Schedules use the cron/interval syntax create_job accepts.
CATALOG: List[CatalogEntry] = [
CatalogEntry(
key="catalog:daily-briefing",
title="Daily briefing",
description="Every morning at 8am, a short briefing: today's calendar, "
"weather, and anything urgent waiting on you.",
job_spec={
"prompt": (
"Produce a concise morning briefing for the user: today's "
"calendar events, the local weather, and any urgent items "
"(unread important email, due tasks). Keep it short and "
"scannable. If you have no connected data sources, give a brief "
"general good-morning with the date and offer to connect "
"calendar/email."
),
"schedule": "0 8 * * *",
"name": "Daily briefing",
"deliver": "origin",
},
),
CatalogEntry(
key="catalog:important-mail-monitor",
title="Important-mail monitor",
description="Check your inbox periodically and ping you ONLY about mail "
"that actually needs attention — never the newsletters.",
job_spec={
"prompt": (
"Check the user's inbox for new messages since the last run. "
"For each candidate, judge urgency against this rule: surface "
"only mail that needs a reply today, is from a manager/family "
"member, or mentions a deadline. Pipe candidates through the "
f"urgency classifier at {classify_items_script_path()} "
"(--threshold 7) and deliver ONLY what it returns. If nothing "
"clears the bar, respond with [SILENT] so the user is not "
"pinged. Requires a connected mail source; if none is "
"configured, explain how to connect one and then stop."
),
"schedule": "every 30m",
"name": "Important-mail monitor",
"deliver": "origin",
},
),
CatalogEntry(
key="catalog:weekly-review",
title="Weekly review",
description="Every Sunday evening, a recap of the week: what got done, "
"what's still open, and what's coming up next week.",
job_spec={
"prompt": (
"Produce a weekly review for the user: summarize what was "
"accomplished this week, list still-open items, and preview "
"next week's calendar. Pull from whatever sources are connected "
"(calendar, task tools, recent conversations). Keep it tight."
),
"schedule": "0 18 * * 0",
"name": "Weekly review",
"deliver": "origin",
},
),
CatalogEntry(
key="catalog:standup-reminder",
title="Workday start reminder",
description="A weekday nudge at 9am with your day's agenda and top "
"priorities, so you start focused.",
job_spec={
"prompt": (
"Give the user a brief weekday start-of-day nudge: their "
"calendar for today and the 1-3 highest-priority things to "
"focus on, inferred from recent context and any task tools. "
"Encouraging, short, one message."
),
"schedule": "0 9 * * 1-5",
"name": "Workday start reminder",
"deliver": "origin",
},
),
]
def seed_catalog_suggestions(
*,
add_fn: Optional[Callable[..., Optional[Dict[str, Any]]]] = None,
keys: Optional[List[str]] = None,
) -> List[Dict[str, Any]]:
"""Register catalog entries as pending suggestions.
``add_fn`` defaults to ``cron.suggestions.add_suggestion`` (injectable for
tests). ``keys`` restricts to specific catalog entries; omit to seed all.
Entries already dismissed/accepted (by dedup key) or beyond the pending cap
are skipped by the store, so re-seeding is safe and idempotent. Returns the
list of suggestion records actually created.
"""
if add_fn is None:
from cron.suggestions import add_suggestion as add_fn # type: ignore[assignment]
wanted = set(keys) if keys else None
created: List[Dict[str, Any]] = []
for entry in CATALOG:
if wanted is not None and entry.key not in wanted:
continue
rec = add_fn(
title=entry.title,
description=entry.description,
source="catalog",
job_spec=dict(entry.job_spec),
dedup_key=entry.key,
)
if rec is not None:
created.append(rec)
return created

257
cron/suggestions.py Normal file
View file

@ -0,0 +1,257 @@
"""Suggested cron jobs — proposed automations the user accepts with one tap.
A *suggestion* is a ready-to-run cron job spec that Hermes surfaces to the
user, who accepts it (creates the real cron job) or dismisses it (latched so
it is never re-offered). This is the single surface every automation proposal
flows through, regardless of where it came from:
* ``catalog`` — a curated starter automation (daily briefing, important-mail
monitor, weekly digest, ...).
* ``recipe`` — the user installed a skill that carries a ``recipe:`` block
(see ``tools/recipes.py``); installing it registers a
suggestion instead of auto-scheduling.
* ``usage`` — the background self-improvement review noticed a recurring
ask that a scheduled job would serve.
* ``integration`` — the user connected an account (Gmail, GitHub, ...) and
the obvious automations for that surface are offered.
Accepting a suggestion just calls the existing ``cron.jobs.create_job`` with
the stored ``job_spec`` — there is NO second job engine. Suggestions never
auto-create jobs; acceptance is always explicit (consent-first). Dismissed
suggestions latch by a stable ``dedup_key`` so the same proposal is not
re-offered after the user says no.
Storage mirrors ``cron/jobs.py``: ``~/.hermes/cron/suggestions.json``, atomic
writes, an in-process lock, and 0600 perms.
"""
from __future__ import annotations
import json
import logging
import os
import tempfile
import threading
import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional
from hermes_constants import get_hermes_home
from hermes_time import now as _hermes_now
from utils import atomic_replace
logger = logging.getLogger(__name__)
CRON_DIR = get_hermes_home().resolve() / "cron"
SUGGESTIONS_FILE = CRON_DIR / "suggestions.json"
# In-process lock protecting load->modify->save cycles (the background review
# fork and the main agent can both write).
_suggestions_lock = threading.Lock()
# Cap pending suggestions so the list never becomes a nag wall. When full,
# new suggestions are dropped (the user should clear the backlog first).
MAX_PENDING = 5
VALID_SOURCES = frozenset({"catalog", "recipe", "usage", "integration"})
_STATUS_PENDING = "pending"
_STATUS_ACCEPTED = "accepted"
_STATUS_DISMISSED = "dismissed"
def _secure_file(path: Path) -> None:
try:
os.chmod(path, 0o600)
except OSError:
pass
def _ensure_dir() -> None:
CRON_DIR.mkdir(parents=True, exist_ok=True)
def _load_raw() -> Dict[str, Any]:
if not SUGGESTIONS_FILE.exists():
return {"suggestions": []}
try:
with open(SUGGESTIONS_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
except (json.JSONDecodeError, OSError) as e:
logger.warning("suggestions.json unreadable (%s); starting empty", e)
return {"suggestions": []}
if isinstance(data, dict) and isinstance(data.get("suggestions"), list):
return data
if isinstance(data, list):
return {"suggestions": data}
logger.warning("suggestions.json malformed; starting empty")
return {"suggestions": []}
def _save_raw(suggestions: List[Dict[str, Any]]) -> None:
_ensure_dir()
fd, tmp_path = tempfile.mkstemp(dir=str(SUGGESTIONS_FILE.parent), suffix=".tmp", prefix=".sugg_")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(
{"suggestions": suggestions, "updated_at": _hermes_now().isoformat()},
f,
indent=2,
)
f.flush()
os.fsync(f.fileno())
atomic_replace(tmp_path, SUGGESTIONS_FILE)
_secure_file(SUGGESTIONS_FILE)
except BaseException:
try:
os.unlink(tmp_path)
except OSError:
pass
raise
def load_suggestions() -> List[Dict[str, Any]]:
"""Return all suggestion records (any status)."""
return _load_raw().get("suggestions", [])
def list_pending() -> List[Dict[str, Any]]:
"""Return pending suggestions in creation order (oldest first)."""
return [s for s in load_suggestions() if s.get("status") == _STATUS_PENDING]
def add_suggestion(
*,
title: str,
description: str,
source: str,
job_spec: Dict[str, Any],
dedup_key: str,
) -> Optional[Dict[str, Any]]:
"""Register a pending suggestion. Returns the record, or None if skipped.
Skipped when: the source is unknown, the same ``dedup_key`` was already
dismissed or accepted (never re-offer), an identical pending suggestion
exists, or the pending list is full (``MAX_PENDING``).
``job_spec`` is a dict of kwargs for ``cron.jobs.create_job`` — accepting
the suggestion passes it straight through, so there is no second schema to
keep in sync.
"""
if source not in VALID_SOURCES:
raise ValueError(f"unknown suggestion source: {source!r}")
if not title.strip() or not dedup_key.strip():
raise ValueError("title and dedup_key are required")
with _suggestions_lock:
suggestions = _load_raw().get("suggestions", [])
# Never re-offer something the user already saw and decided on, and
# never duplicate a still-pending proposal.
for existing in suggestions:
if existing.get("dedup_key") == dedup_key:
if existing.get("status") in (_STATUS_DISMISSED, _STATUS_ACCEPTED):
return None
if existing.get("status") == _STATUS_PENDING:
return None
pending_count = sum(1 for s in suggestions if s.get("status") == _STATUS_PENDING)
if pending_count >= MAX_PENDING:
logger.info("Suggestion backlog full (%d); dropping %r", MAX_PENDING, title)
return None
record = {
"id": uuid.uuid4().hex[:12],
"title": title.strip(),
"description": description.strip(),
"source": source,
"job_spec": job_spec,
"dedup_key": dedup_key.strip(),
"status": _STATUS_PENDING,
"created_at": _hermes_now().isoformat(),
}
suggestions.append(record)
_save_raw(suggestions)
return record
def get_suggestion(ref: str) -> Optional[Dict[str, Any]]:
"""Resolve a suggestion by id, 1-based pending index, or title (exact)."""
suggestions = load_suggestions()
# By id.
for s in suggestions:
if s.get("id") == ref:
return s
# By 1-based pending index.
if ref.isdigit():
pending = [s for s in suggestions if s.get("status") == _STATUS_PENDING]
idx = int(ref) - 1
if 0 <= idx < len(pending):
return pending[idx]
# By exact title (case-insensitive).
for s in suggestions:
if s.get("title", "").lower() == ref.lower():
return s
return None
def _set_status(suggestion_id: str, status: str) -> bool:
with _suggestions_lock:
suggestions = _load_raw().get("suggestions", [])
changed = False
for s in suggestions:
if s.get("id") == suggestion_id:
s["status"] = status
s["resolved_at"] = _hermes_now().isoformat()
changed = True
break
if changed:
_save_raw(suggestions)
return changed
def dismiss_suggestion(ref: str) -> bool:
"""Dismiss a suggestion (latched — never re-offered for its dedup_key)."""
s = get_suggestion(ref)
if not s:
return False
return _set_status(s["id"], _STATUS_DISMISSED)
def accept_suggestion(ref: str, *, origin: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]:
"""Accept a suggestion: create the real cron job from its ``job_spec``.
Returns the created cron job dict, or None if the suggestion isn't found /
not pending. The job_spec is passed straight to ``cron.jobs.create_job``;
an ``origin`` (platform/chat) is merged so "origin" delivery routes back to
the chat where the user accepted.
"""
s = get_suggestion(ref)
if not s or s.get("status") != _STATUS_PENDING:
return None
from cron.jobs import create_job
spec = dict(s.get("job_spec") or {})
if origin is not None and "origin" not in spec:
spec["origin"] = origin
job = create_job(**spec)
_set_status(s["id"], _STATUS_ACCEPTED)
return job
def clear_resolved() -> int:
"""Drop accepted/dismissed records from disk. Returns the count removed.
Pending suggestions and the dedup memory of dismissed ones are the only
things that matter long-term, but dismissed records must be RETAINED for
their dedup_key (so they aren't re-offered). This only prunes ACCEPTED
records, which have served their purpose once the job exists.
"""
with _suggestions_lock:
suggestions = _load_raw().get("suggestions", [])
kept = [s for s in suggestions if s.get("status") != _STATUS_ACCEPTED]
removed = len(suggestions) - len(kept)
if removed:
_save_raw(kept)
return removed

33
gateway/run.py
View file

@ -7171,6 +7171,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if canonical == "kanban":
return await self._handle_kanban_command(event)
if canonical == "suggestions":
return await self._handle_suggestions_command(event)
if canonical == "retry":
return await self._handle_retry_command(event)
@ -9237,6 +9240,36 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
async def _handle_suggestions_command(self, event: MessageEvent) -> str:
"""Handle /suggestions in the gateway.
Delegates to the shared handler so CLI and gateway never drift. The
origin is built from the event source so an accepted suggestion's job
delivers back to this chat/thread.
"""
args = (event.get_command_args() or "").strip()
source = event.source
origin = None
try:
platform = getattr(source.platform, "value", None) or str(getattr(source, "platform", "") or "")
chat_id = getattr(source, "chat_id", None)
if platform and chat_id:
origin = {
"platform": platform,
"chat_id": str(chat_id),
"chat_name": getattr(source, "chat_name", None),
"thread_id": getattr(source, "thread_id", None),
}
except Exception:
origin = None
try:
from hermes_cli.suggestions_cmd import handle_suggestions_command
return handle_suggestions_command(args, origin=origin)
except Exception as e:
logger.debug("suggestions command failed: %s", e)
return f"Suggestions command failed: {e}"
# ────────────────────────────────────────────────────────────────
# /goal — persistent cross-turn goals (Ralph-style loop)
# ────────────────────────────────────────────────────────────────

3
hermes_cli/commands.py
View file

@ -179,6 +179,9 @@ COMMAND_REGISTRY: list[CommandDef] = [
CommandDef("cron", "Manage scheduled tasks", "Tools & Skills",
cli_only=True, args_hint="[subcommand]",
subcommands=("list", "add", "create", "edit", "pause", "resume", "run", "remove")),
CommandDef("suggestions", "Review suggested automations (accept/dismiss)",
"Tools & Skills", aliases=("suggest",), args_hint="[accept|dismiss N | catalog]",
subcommands=("accept", "dismiss", "catalog", "clear")),
CommandDef("curator", "Background skill maintenance (status, run, pin, archive, list-archived)",
"Tools & Skills", args_hint="[subcommand]",
subcommands=("status", "run", "pause", "resume", "pin", "unpin", "restore", "list-archived")),

14
hermes_cli/config.py
View file

@ -1369,6 +1369,20 @@ DEFAULT_CONFIG = {
"timeout": 600,
"extra_body": {},
},
# Monitor — urgency/importance classifier used by the important-mail
# monitor catalog automation (cron/scripts/classify_items.py). Scores
# candidate items 0-10 against the user's criteria so only above-
# threshold items get delivered. "auto" = main chat model; override to
# a cheap fast model (e.g. openrouter google/gemini-3-flash-preview,
# haiku) since per-item scoring is high-volume and a small model is fine.
"monitor": {
"provider": "auto",
"model": "",
"base_url": "",
"api_key": "",
"timeout": 60,
"extra_body": {},
},
},
"display": {

27
hermes_cli/skills_hub.py
View file

@ -691,6 +691,33 @@ def do_install(identifier: str, category: str = "", force: bool = False,
c.print(f"[bold green]Installed:[/] {install_dir.relative_to(SKILLS_DIR)}")
c.print(f"[dim]Files: {', '.join(bundle.files.keys())}[/]\n")
# Recipe detection: if the installed skill declares a
# metadata.hermes.recipe block, it is a runnable automation. Register it as
# a Suggested Cron Job rather than auto-scheduling — installing never
# silently creates a recurring job; the user accepts it via /suggestions.
# This is the single surface every automation proposal flows through.
try:
from tools.recipes import RecipeError, recipe_spec_for_installed, register_recipe_suggestion
try:
spec = recipe_spec_for_installed(bundle.name)
except RecipeError as _rec_err:
c.print(f"[yellow]Recipe block present but invalid:[/] {_rec_err}\n")
spec = None
if spec is not None:
registered = register_recipe_suggestion(spec)
if registered is not None:
c.print(
f"[bold cyan]Recipe:[/] '{bundle.name}' is an automation "
f"(schedule [bold]{spec.schedule}[/])."
)
c.print(
"[dim]Added to your suggestions — run[/] [bold]/suggestions[/] "
"[dim]to schedule or dismiss it.[/]\n"
)
except Exception: # pragma: no cover - recipe detection is best-effort
pass
if invalidate_cache:
# Invalidate the skills prompt cache so the new skill appears immediately
try:

145
hermes_cli/suggestions_cmd.py Normal file
View file

@ -0,0 +1,145 @@
"""Shared ``/suggestions`` command logic for CLI and gateway.
Both surfaces call ``handle_suggestions_command(args, origin=...)`` and present
the returned text however they present command output. Keeping the logic here
(not in cli.py / gateway/run.py) means the two surfaces can never drift.
Subcommands:
/suggestions list pending suggestions (numbered)
/suggestions accept <N|id> create the cron job for that suggestion
/suggestions dismiss <N|id> dismiss it (latched, never re-offered)
/suggestions catalog seed the curated starter automations as pending
/suggestions clear drop accepted records (housekeeping)
"""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
logger = logging.getLogger(__name__)
def _fmt_pending(pending: list) -> str:
if not pending:
return (
"No suggested automations right now.\n"
"Try `/suggestions catalog` to see the curated starter set, or "
"install a recipe skill to get one."
)
lines = ["Suggested automations — `/suggestions accept N` or `dismiss N`:\n"]
for i, s in enumerate(pending, 1):
spec = s.get("job_spec", {}) or {}
sched = spec.get("schedule", "?")
src = s.get("source", "?")
lines.append(f" {i}. {s.get('title', '(untitled)')} [{sched}] ({src})")
desc = s.get("description", "").strip()
if desc:
lines.append(f" {desc}")
return "\n".join(lines)
def _resolve_origin() -> Optional[Dict[str, Any]]:
"""Best-effort current-chat origin from session env (CLI and gateway both set it).
Mirrors cron's ``_origin_from_env`` so an accepted suggestion's job delivers
back to the chat where it was accepted. Returns None if unavailable, in
which case create_job falls back to a configured home channel.
"""
try:
from gateway.session_context import get_session_env
platform = get_session_env("HERMES_SESSION_PLATFORM")
chat_id = get_session_env("HERMES_SESSION_CHAT_ID")
if platform and chat_id:
return {
"platform": platform,
"chat_id": chat_id,
"chat_name": get_session_env("HERMES_SESSION_CHAT_NAME") or None,
"thread_id": get_session_env("HERMES_SESSION_THREAD_ID") or None,
}
except Exception:
pass
return None
def handle_suggestions_command(
args: str,
*,
origin: Optional[Dict[str, Any]] = None,
) -> str:
"""Dispatch a ``/suggestions`` invocation. Returns text to show the user.
``args`` is everything after ``/suggestions`` (already stripped of the
command word). ``origin`` is the platform/chat dict so an accepted job's
"origin" delivery routes back to where the user accepted; when omitted it
is resolved from the session environment.
"""
if origin is None:
origin = _resolve_origin()
try:
from cron import suggestions as store
except Exception as e: # pragma: no cover - import guard
logger.debug("suggestions store import failed: %s", e)
return "Suggestions are unavailable in this build."
parts = (args or "").strip().split()
sub = parts[0].lower() if parts else ""
rest = " ".join(parts[1:]).strip()
# Bare /suggestions -> list pending.
if not sub:
return _fmt_pending(store.list_pending())
if sub in ("accept", "add", "schedule"):
if not rest:
return "Usage: /suggestions accept <number|id>"
job = store.accept_suggestion(rest, origin=origin)
if job is None:
return f"No pending suggestion matches '{rest}'. Run /suggestions to list them."
sched = job.get("schedule_display") or (job.get("job_spec", {}) or {}).get("schedule", "")
name = job.get("name", "automation")
return (
f"Scheduled '{name}'"
+ (f" ({sched})" if sched else "")
+ ". Manage it with /cron."
)
if sub in ("dismiss", "no", "reject"):
if not rest:
return "Usage: /suggestions dismiss <number|id>"
ok = store.dismiss_suggestion(rest)
return (
f"Dismissed. Won't suggest that again."
if ok
else f"No pending suggestion matches '{rest}'."
)
if sub == "catalog":
try:
from cron.suggestion_catalog import seed_catalog_suggestions
created = seed_catalog_suggestions()
except Exception as e:
logger.debug("catalog seed failed: %s", e)
return "Couldn't load the catalog."
if not created:
return (
"No new catalog automations to add (already offered, dismissed, "
"or your suggestion list is full). Run /suggestions to see pending."
)
added = ", ".join(c.get("title", "?") for c in created)
return f"Added {len(created)} suggestion(s): {added}.\nRun /suggestions to review."
if sub == "clear":
removed = store.clear_resolved()
return f"Cleared {removed} resolved suggestion record(s)."
return (
"Usage:\n"
" /suggestions list pending\n"
" /suggestions accept N schedule suggestion N\n"
" /suggestions dismiss N dismiss suggestion N\n"
" /suggestions catalog add curated starter automations\n"
" /suggestions clear housekeeping"
)

193
tests/cron/test_suggestions.py Normal file
View file

@ -0,0 +1,193 @@
"""Tests for the Suggested Cron Jobs feature.
Covers the store (add/dedup/cap/accept/dismiss/latch), catalog seeding, the
recipe->suggestion bridge, and the shared command handler. Uses an isolated
HERMES_HOME so the real suggestions.json is never touched.
"""
import importlib
import json
from pathlib import Path
from unittest.mock import patch
import pytest
@pytest.fixture
def store(tmp_path, monkeypatch):
"""A cron.suggestions module bound to an isolated HERMES_HOME."""
home = tmp_path / ".hermes"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
# Reload so module-level CRON_DIR/SUGGESTIONS_FILE pick up the temp home.
import hermes_constants
importlib.reload(hermes_constants)
import cron.suggestions as s
importlib.reload(s)
return s
def _add(store, key="k1", title="Test", source="catalog", schedule="0 9 * * *"):
return store.add_suggestion(
title=title,
description="desc",
source=source,
job_spec={"prompt": "do it", "schedule": schedule, "name": title, "deliver": "origin"},
dedup_key=key,
)
class TestStore:
def test_add_and_list_pending(self, store):
rec = _add(store)
assert rec is not None
pending = store.list_pending()
assert len(pending) == 1
assert pending[0]["title"] == "Test"
assert pending[0]["status"] == "pending"
def test_dedup_blocks_duplicate_pending(self, store):
assert _add(store, key="dup") is not None
assert _add(store, key="dup") is None # same key already pending
assert len(store.list_pending()) == 1
def test_dismiss_latches_against_redisplay(self, store):
_add(store, key="latch")
assert store.dismiss_suggestion("1") is True
assert store.list_pending() == []
# Re-adding the same key is refused (never re-offer a dismissed one).
assert _add(store, key="latch") is None
def test_unknown_source_rejected(self, store):
with pytest.raises(ValueError):
store.add_suggestion(title="x", description="d", source="bogus", job_spec={}, dedup_key="k")
def test_pending_cap(self, store):
for i in range(store.MAX_PENDING):
assert _add(store, key=f"k{i}") is not None
# One past the cap is dropped.
assert _add(store, key="over") is None
assert len(store.list_pending()) == store.MAX_PENDING
def test_accept_creates_job_and_marks_accepted(self, store):
_add(store, key="acc", title="My Job")
created = {}
def fake_create_job(**kwargs):
created.update(kwargs)
return {"id": "job123", "name": kwargs.get("name"), **kwargs}
with patch("cron.jobs.create_job", fake_create_job):
job = store.accept_suggestion("1", origin={"platform": "telegram", "chat_id": "5"})
assert job is not None
assert created["schedule"] == "0 9 * * *"
assert created["origin"] == {"platform": "telegram", "chat_id": "5"}
# No longer pending.
assert store.list_pending() == []
# And accepting again is a no-op (not pending anymore).
assert store.accept_suggestion("acc") is None
def test_get_by_id_and_index_and_title(self, store):
rec = _add(store, key="byref", title="Findable")
assert store.get_suggestion(rec["id"])["id"] == rec["id"]
assert store.get_suggestion("1")["id"] == rec["id"]
assert store.get_suggestion("findable")["id"] == rec["id"]
assert store.get_suggestion("nope") is None
def test_clear_resolved_drops_accepted_only(self, store):
_add(store, key="a")
_add(store, key="b")
store.dismiss_suggestion("2") # b dismissed (retained for latch)
with patch("cron.jobs.create_job", lambda **k: {"id": "j"}):
store.accept_suggestion("1") # a accepted
removed = store.clear_resolved()
assert removed == 1 # only the accepted record pruned
# Dismissed record retained so its dedup_key still latches.
assert _add(store, key="b") is None
class TestCatalog:
def test_seed_registers_all_entries(self, store):
from cron.suggestion_catalog import CATALOG, seed_catalog_suggestions
created = seed_catalog_suggestions(add_fn=store.add_suggestion)
assert len(created) == len(CATALOG)
assert len(store.list_pending()) == min(len(CATALOG), store.MAX_PENDING)
def test_seed_is_idempotent(self, store):
from cron.suggestion_catalog import seed_catalog_suggestions
first = seed_catalog_suggestions(add_fn=store.add_suggestion)
second = seed_catalog_suggestions(add_fn=store.add_suggestion)
assert len(first) >= 1
assert second == [] # already present -> nothing new
def test_monitor_entry_references_classifier_script(self):
from cron.suggestion_catalog import CATALOG, classify_items_script_path
monitor = next(e for e in CATALOG if e.key == "catalog:important-mail-monitor")
assert classify_items_script_path() in monitor.job_spec["prompt"]
assert Path(classify_items_script_path()).name == "classify_items.py"
class TestRecipeBridge:
def test_recipe_registers_suggestion(self, store):
from tools.recipes import RecipeSpec, register_recipe_suggestion
spec = RecipeSpec(skill_name="morning-brief", schedule="0 8 * * *", deliver="telegram")
with patch("cron.suggestions.add_suggestion", store.add_suggestion):
rec = register_recipe_suggestion(spec)
assert rec is not None
assert rec["source"] == "recipe"
assert rec["job_spec"]["skills"] == ["morning-brief"]
assert rec["job_spec"]["schedule"] == "0 8 * * *"
def test_recipe_to_job_spec_matches_create_recipe_job(self):
from tools.recipes import RecipeSpec, recipe_to_job_spec
spec = RecipeSpec(skill_name="x", schedule="every 2h", deliver="origin", prompt="p")
js = recipe_to_job_spec(spec)
assert js["skills"] == ["x"]
assert js["schedule"] == "every 2h"
assert js["prompt"] == "p"
class TestCommandHandler:
def test_bare_lists_pending(self, store):
_add(store, key="c1", title="Daily thing")
with patch("cron.suggestions.list_pending", store.list_pending):
from hermes_cli.suggestions_cmd import handle_suggestions_command
# Patch the module the handler imports.
with patch.dict("sys.modules"):
out = handle_suggestions_command("")
assert "Daily thing" in out
def test_accept_via_handler(self, store):
_add(store, key="ha", title="Acceptable")
from hermes_cli.suggestions_cmd import handle_suggestions_command
with patch("cron.jobs.create_job", lambda **k: {"id": "j", "name": k.get("name"), "job_spec": k}):
out = handle_suggestions_command("accept 1", origin={"platform": "cli", "chat_id": "1"})
assert "Scheduled" in out
assert store.list_pending() == []
def test_dismiss_via_handler(self, store):
_add(store, key="hd", title="Dismissable")
from hermes_cli.suggestions_cmd import handle_suggestions_command
out = handle_suggestions_command("dismiss 1")
assert "Dismissed" in out
assert store.list_pending() == []
def test_empty_list_message(self, store):
from hermes_cli.suggestions_cmd import handle_suggestions_command
out = handle_suggestions_command("")
assert "No suggested automations" in out
def test_aux_monitor_config_default(self):
from hermes_cli.config import DEFAULT_CONFIG
assert "monitor" in DEFAULT_CONFIG["auxiliary"]
assert DEFAULT_CONFIG["auxiliary"]["monitor"]["provider"] == "auto"

169
tests/tools/test_recipes.py Normal file
View file

@ -0,0 +1,169 @@
"""Tests for the recipes layer (skill frontmatter <-> cron automation bridge).
A recipe is a skill with a metadata.hermes.recipe block. These verify parsing,
the create-job bridge, and the export round-trip without touching the real
cron store.
"""
import sys
from pathlib import Path
from unittest.mock import patch
import pytest
from tools.recipes import (
RecipeError,
RecipeSpec,
create_recipe_job,
export_recipe,
parse_recipe,
recipe_spec_for_installed,
)
RECIPE_SKILL = """---
name: morning-brief
description: Summarize unread email and calendar every morning.
version: 1.0.0
metadata:
hermes:
tags: [recipe, email]
recipe:
schedule: "0 8 * * *"
deliver: telegram
prompt: "Summarize my unread email and today's calendar."
---
# Morning Brief
Every morning, gather unread email and the day's calendar and send a digest.
"""
PLAIN_SKILL = """---
name: not-a-recipe
description: Just a regular skill.
metadata:
hermes:
tags: [misc]
---
# Not a recipe
"""
MALFORMED_RECIPE = """---
name: broken
description: Recipe with no schedule.
metadata:
hermes:
recipe:
deliver: origin
---
# Broken
"""
class TestParseRecipe:
def test_parses_full_recipe(self):
spec = parse_recipe(RECIPE_SKILL)
assert spec is not None
assert spec.skill_name == "morning-brief"
assert spec.schedule == "0 8 * * *"
assert spec.deliver == "telegram"
assert spec.prompt is not None and spec.prompt.startswith("Summarize")
def test_plain_skill_is_not_a_recipe(self):
assert parse_recipe(PLAIN_SKILL) is None
def test_no_frontmatter_is_not_a_recipe(self):
assert parse_recipe("just some text, no frontmatter") is None
def test_missing_schedule_raises(self):
with pytest.raises(RecipeError):
parse_recipe(MALFORMED_RECIPE)
def test_recipe_not_mapping_raises(self):
bad = "---\nname: x\nmetadata:\n hermes:\n recipe: not-a-dict\n---\n\nbody"
with pytest.raises(RecipeError):
parse_recipe(bad)
def test_deliver_defaults_to_origin(self):
skill = (
"---\nname: r\ndescription: d\nmetadata:\n hermes:\n"
' recipe:\n schedule: "every 1h"\n---\n\nbody'
)
spec = parse_recipe(skill)
assert spec is not None
assert spec.deliver == "origin"
class TestRecipeSpecForInstalled:
def test_finds_and_parses_installed_recipe(self, tmp_path):
skills_dir = tmp_path / "skills"
rec_dir = skills_dir / "productivity" / "morning-brief"
rec_dir.mkdir(parents=True)
(rec_dir / "SKILL.md").write_text(RECIPE_SKILL, encoding="utf-8")
with patch("tools.skills_hub.SKILLS_DIR", skills_dir):
spec = recipe_spec_for_installed("morning-brief")
assert spec is not None
assert spec.schedule == "0 8 * * *"
def test_missing_skill_returns_none(self, tmp_path):
skills_dir = tmp_path / "skills"
skills_dir.mkdir()
with patch("tools.skills_hub.SKILLS_DIR", skills_dir):
assert recipe_spec_for_installed("nope") is None
def test_plain_skill_returns_none(self, tmp_path):
skills_dir = tmp_path / "skills"
d = skills_dir / "misc" / "not-a-recipe"
d.mkdir(parents=True)
(d / "SKILL.md").write_text(PLAIN_SKILL, encoding="utf-8")
with patch("tools.skills_hub.SKILLS_DIR", skills_dir):
assert recipe_spec_for_installed("not-a-recipe") is None
class TestCreateRecipeJob:
def test_bridges_to_create_job(self):
spec = parse_recipe(RECIPE_SKILL)
assert spec is not None
captured = {}
def fake_create_job(**kwargs):
captured.update(kwargs)
return {"id": "abc123", **kwargs}
with patch("cron.jobs.create_job", fake_create_job):
job = create_recipe_job(spec, origin={"platform": "telegram"})
assert captured["schedule"] == "0 8 * * *"
assert captured["skills"] == ["morning-brief"]
assert captured["deliver"] == "telegram"
assert captured["prompt"].startswith("Summarize")
assert job["id"] == "abc123"
class TestExportRecipe:
def test_round_trips_job_to_skill_md(self):
job = {
"name": "My Morning Brief",
"schedule_display": "0 8 * * *",
"skills": ["morning-brief"],
"deliver": "telegram",
"prompt": "Summarize my unread email.",
}
md = export_recipe(job, "# Morning Brief\n\nDoes the morning digest.")
# The exported SKILL.md must itself parse back as a recipe.
spec = parse_recipe(md)
assert spec is not None
assert spec.schedule == "0 8 * * *"
assert spec.deliver == "telegram"
# Name is sanitized to a valid skill identifier.
assert spec.skill_name == "my-morning-brief"
def test_export_has_recipe_tag(self):
job = {"name": "x", "schedule_display": "every 2h", "skills": ["x"]}
md = export_recipe(job, "body")
assert "recipe" in md
assert "automation" in md

317
tools/recipes.py Normal file
View file

@ -0,0 +1,317 @@
"""Recipes: shareable plain-language automations layered on skills + cron.
A "recipe" is NOT a new object type. It is an ordinary skill (a SKILL.md the
agent loads) that additionally declares an automation schedule in its
frontmatter:
metadata:
hermes:
recipe:
schedule: "0 9 * * *" # presence of `recipe:` marks it runnable
deliver: origin # optional (default "origin")
prompt: "..." # optional task instruction for the run
no_agent: false # optional
Because a recipe is just a skill, it flows through the ENTIRE existing
skills-hub pipeline for free — search, inspect, quarantine, security scan,
install, lock-file provenance, audit log, taps, the centralized index, and
`hermes skills publish` for sharing. No new source type, no new store, no new
transport. This module is the thin bridge between that skill metadata and the
existing cron `create_job()` API:
* ``parse_recipe(skill_md_text)`` -> RecipeSpec | None
* ``recipe_spec_for_installed(name)`` -> RecipeSpec | None
* ``create_recipe_job(spec, ...)`` -> the created cron job dict
* ``export_recipe(job, body)`` -> a shareable SKILL.md string
The dev guide's "Extend, Don't Duplicate" rule is the whole design: the recipe
is a skill, the schedule is a cron job, sharing is the existing publish/tap/
index path.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
__all__ = [
"RecipeSpec",
"parse_recipe",
"recipe_spec_for_installed",
"recipe_to_job_spec",
"create_recipe_job",
"register_recipe_suggestion",
"export_recipe",
"RecipeError",
]
class RecipeError(ValueError):
"""Raised when a recipe block is present but malformed."""
@dataclass
class RecipeSpec:
"""Parsed ``metadata.hermes.recipe`` automation spec for a skill."""
skill_name: str
schedule: str
deliver: str = "origin"
prompt: Optional[str] = None
no_agent: bool = False
model: Optional[str] = None
provider: Optional[str] = None
enabled_toolsets: Optional[List[str]] = None
raw: Dict[str, Any] = field(default_factory=dict)
def _split_frontmatter(text: str) -> Optional[Dict[str, Any]]:
"""Return the parsed YAML frontmatter mapping, or None if absent/invalid."""
if not isinstance(text, str):
return None
stripped = text.lstrip()
if not stripped.startswith("---"):
return None
# Find the closing fence after the opening one.
after_open = stripped[3:]
end = after_open.find("\n---")
if end == -1:
return None
fm_text = after_open[:end]
try:
import yaml
data = yaml.safe_load(fm_text)
except Exception as e: # pragma: no cover - malformed YAML
logger.debug("recipe: frontmatter YAML parse failed: %s", e)
return None
return data if isinstance(data, dict) else None
def parse_recipe(skill_md_text: str) -> Optional[RecipeSpec]:
"""Extract a RecipeSpec from a SKILL.md string, or None if not a recipe.
A skill is a recipe iff ``metadata.hermes.recipe`` is a mapping containing
a non-empty ``schedule``. Raises RecipeError if the block exists but is
structurally invalid (so a typo surfaces instead of silently no-op'ing).
"""
fm = _split_frontmatter(skill_md_text)
if not fm:
return None
name = str(fm.get("name", "")).strip()
meta = fm.get("metadata")
hermes = meta.get("hermes") if isinstance(meta, dict) else None
recipe = hermes.get("recipe") if isinstance(hermes, dict) else None
if recipe is None:
return None
if not isinstance(recipe, dict):
raise RecipeError("metadata.hermes.recipe must be a mapping")
schedule = str(recipe.get("schedule", "")).strip()
if not schedule:
raise RecipeError("recipe.schedule is required and must be non-empty")
deliver = str(recipe.get("deliver", "origin")).strip() or "origin"
prompt = recipe.get("prompt")
if prompt is not None:
prompt = str(prompt)
no_agent = bool(recipe.get("no_agent", False))
model = recipe.get("model")
provider = recipe.get("provider")
toolsets = recipe.get("enabled_toolsets")
if toolsets is not None and not isinstance(toolsets, list):
raise RecipeError("recipe.enabled_toolsets must be a list when present")
return RecipeSpec(
skill_name=name,
schedule=schedule,
deliver=deliver,
prompt=prompt,
no_agent=no_agent,
model=str(model).strip() if model else None,
provider=str(provider).strip() if provider else None,
enabled_toolsets=[str(t) for t in toolsets] if toolsets else None,
raw=recipe,
)
def recipe_spec_for_installed(skill_name: str) -> Optional[RecipeSpec]:
"""Locate an installed skill's SKILL.md and parse its recipe block.
Searches the standard skills tree for ``<skill_name>/SKILL.md``. Returns
None if the skill isn't found or isn't a recipe.
"""
try:
from tools.skills_hub import SKILLS_DIR
except Exception: # pragma: no cover - import guard
return None
base = Path(SKILLS_DIR)
# Skills live at skills/<category>/<name>/SKILL.md or skills/<name>/SKILL.md.
candidates = list(base.glob(f"**/{skill_name}/SKILL.md"))
for path in candidates:
try:
text = path.read_text(encoding="utf-8")
except OSError:
continue
spec = parse_recipe(text)
if spec is not None:
# Prefer the frontmatter name, fall back to the directory name.
if not spec.skill_name:
spec.skill_name = skill_name
return spec
return None
def recipe_to_job_spec(
spec: RecipeSpec,
*,
name: Optional[str] = None,
) -> Dict[str, Any]:
"""Build the ``cron.jobs.create_job`` kwargs dict for a RecipeSpec.
This is the single source of truth for translating a recipe into a job.
Both the direct ``create_recipe_job`` path and the suggestion path
(``register_recipe_suggestion``) build on it, so a recipe scheduled now and
a recipe accepted from a suggestion produce an identical job.
"""
return {
"prompt": spec.prompt,
"schedule": spec.schedule,
"name": name or f"recipe:{spec.skill_name}",
"deliver": spec.deliver,
"skills": [spec.skill_name] if spec.skill_name else None,
"model": spec.model,
"provider": spec.provider,
"enabled_toolsets": spec.enabled_toolsets,
"no_agent": spec.no_agent,
}
def create_recipe_job(
spec: RecipeSpec,
*,
origin: Optional[Dict[str, Any]] = None,
name: Optional[str] = None,
) -> Dict[str, Any]:
"""Create the cron job described by a RecipeSpec via the existing cron API.
The recipe's skill is loaded before the run (cron ``skills=[name]``); the
optional ``prompt`` becomes the task instruction. Delivery, model, and
toolsets carry through. Returns the created job dict.
"""
from cron.jobs import create_job
job_spec = recipe_to_job_spec(spec, name=name)
if origin is not None:
job_spec["origin"] = origin
return create_job(**job_spec)
def register_recipe_suggestion(spec: RecipeSpec) -> Optional[Dict[str, Any]]:
"""Turn an installed recipe into a pending Suggested Cron Job.
Recipes are source ``recipe`` of the unified suggestion surface: installing
a skill that carries a ``recipe:`` block does NOT auto-schedule it — it
registers a suggestion the user accepts (or dismisses) like any other.
Returns the suggestion record, or None if it was skipped (already
seen/dismissed, backlog full, etc.).
"""
if not spec.skill_name:
return None
try:
from cron.suggestions import add_suggestion
except Exception: # pragma: no cover - import guard
return None
return add_suggestion(
title=f"Schedule '{spec.skill_name}'",
description=(
f"The '{spec.skill_name}' recipe runs on schedule {spec.schedule}"
+ (f", delivering to {spec.deliver}" if spec.deliver and spec.deliver != "origin" else "")
+ "."
),
source="recipe",
job_spec=recipe_to_job_spec(spec),
dedup_key=f"recipe:{spec.skill_name}:{spec.schedule}",
)
def export_recipe(job: Dict[str, Any], body: str, *, recipe_name: Optional[str] = None) -> str:
"""Render a shareable recipe SKILL.md from an existing cron job dict.
The inverse of ``create_recipe_job``: take a cron job a user already built
and emit a SKILL.md (with a ``metadata.hermes.recipe`` block) they can hand
to ``hermes skills publish`` to share. ``body`` is the plain-language
description / instructions that become the SKILL.md body.
"""
import yaml
name = recipe_name or job.get("name") or "shared-recipe"
# Sanitize to a valid skill identifier.
name = "".join(c if (c.isalnum() or c in "-_") else "-" for c in str(name).lower())
name = name.strip("-_") or "shared-recipe"
schedule = job.get("schedule_display") or _schedule_to_string(job.get("schedule"))
skills = job.get("skills") or ([job["skill"]] if job.get("skill") else [])
recipe_block: Dict[str, Any] = {"schedule": schedule}
deliver = job.get("deliver")
if deliver and deliver != "origin":
recipe_block["deliver"] = deliver
if job.get("prompt"):
recipe_block["prompt"] = job["prompt"]
if job.get("no_agent"):
recipe_block["no_agent"] = True
if job.get("model"):
recipe_block["model"] = job["model"]
if job.get("provider"):
recipe_block["provider"] = job["provider"]
if job.get("enabled_toolsets"):
recipe_block["enabled_toolsets"] = job["enabled_toolsets"]
description = (
(body.strip().splitlines() or ["Shared automation recipe."])[0][:200]
if body.strip()
else "Shared automation recipe."
)
frontmatter = {
"name": name,
"description": description,
"version": "1.0.0",
"license": "MIT",
"metadata": {
"hermes": {
"tags": ["recipe", "automation"],
"recipe": recipe_block,
}
},
}
fm_yaml = yaml.safe_dump(frontmatter, sort_keys=False, allow_unicode=True).strip()
body_text = body.strip() or f"# {name}\n\nShared automation recipe."
return f"---\n{fm_yaml}\n---\n\n{body_text}\n"
def _schedule_to_string(schedule: Any) -> str:
"""Best-effort render of a parsed schedule dict back to a string."""
if isinstance(schedule, str):
return schedule
if isinstance(schedule, dict):
kind = schedule.get("kind")
if kind == "cron" and schedule.get("expr"):
return str(schedule["expr"])
if kind == "interval" and schedule.get("seconds"):
secs = int(schedule["seconds"])
if secs % 3600 == 0:
return f"every {secs // 3600}h"
if secs % 60 == 0:
return f"every {secs // 60}m"
return f"every {secs}s"
return "0 9 * * *" # safe daily fallback

63
website/docs/developer-guide/creating-skills.md
View file

@ -66,6 +66,11 @@ metadata:
description: "What this setting controls"
default: "sensible-default"
prompt: "Display prompt for setup"
recipe: # Optional — marks this skill a runnable automation
schedule: "0 9 * * *" # cron expr / "every 2h" / ISO timestamp
deliver: origin # optional (default origin)
prompt: "Task instruction for each run" # optional
no_agent: false # optional
required_environment_variables: # Optional — env vars the skill needs
- name: MY_API_KEY
prompt: "Enter your API key"
@ -334,6 +339,64 @@ If your skill is official and useful but not universally needed (e.g., a paid se
If your skill is specialized, community-contributed, or niche, it's better suited for a **Skills Hub** — upload it to a registry and share it via `hermes skills install`.
## Recipes: skills that are also automations
A **recipe** is an ordinary skill that additionally declares a schedule in its frontmatter. Add a `metadata.hermes.recipe` block and the skill becomes a shareable, runnable automation:
```yaml
metadata:
hermes:
tags: [recipe, email]
recipe:
schedule: "0 8 * * *" # presence of `recipe:` marks it runnable
deliver: telegram # optional (default: origin)
prompt: "Summarize my unread email and today's calendar." # optional
no_agent: false # optional
```
Because a recipe **is** a skill, it flows through the entire skills pipeline unchanged — search, inspect, install, security scan, provenance, taps, the centralized index, and `hermes skills publish` for sharing. Nothing new to learn.
**Installing a recipe.** When you install a skill that carries a `recipe:` block, Hermes registers it as a **suggested cron job** rather than scheduling it. Scheduling is **opt-in** — installing never silently creates a recurring job. You review and accept it via `/suggestions`:
```bash
hermes skills install owner/morning-brief
# → Recipe: 'morning-brief' is an automation (schedule 0 8 * * *).
# Added to your suggestions — run /suggestions to schedule or dismiss it.
# then, in a session:
/suggestions # lists pending suggestions, numbered
/suggestions accept 1 # creates the cron job
/suggestions dismiss 1 # never offer it again
```
Recipes are one **source** of the unified Suggested Cron Jobs surface — the same place curated starter automations and (later) usage-pattern and integration suggestions appear. See [Suggested Cron Jobs](#suggested-cron-jobs) below.
**Sharing an automation you built.** A recipe loaded by a cron job (`hermes cron create --skill <name> ...`) can be exported back to a SKILL.md and published like any other skill, so an automation you tuned for yourself becomes a one-command install for someone else.
The recipe layer adds no new object type, store, or transport — the recipe is a skill, the schedule is a cron job, and sharing is the existing publish/tap/index path.
## Suggested Cron Jobs
Hermes can *propose* automations and let you accept them with one tap, instead of making you assemble cron jobs by hand. Every proposal flows through one surface — the `/suggestions` command — regardless of where it came from:
| Source | Trigger |
|--------|---------|
| `catalog` | Curated starter automations (`/suggestions catalog`) — daily briefing, important-mail monitor, weekly review, workday-start reminder |
| `recipe` | You installed a skill carrying a `recipe:` block |
| `usage` | The background review noticed a recurring ask a schedule would serve |
| `integration` | You connected an account (Gmail, GitHub, ...) and the obvious automations are offered |
```bash
/suggestions # list pending
/suggestions accept N # schedule suggestion N (creates the cron job)
/suggestions dismiss N # dismiss it — latched, never re-offered
/suggestions catalog # add the curated starter automations
```
Accepting a suggestion calls the same `cron.jobs.create_job` the `cronjob` tool uses — there is no second job engine. Suggestions **never** auto-create jobs; acceptance is always explicit. Dismissed suggestions latch by a stable key so the same proposal is never re-offered. The pending list is capped so it never becomes a nag wall.
The **important-mail monitor** catalog entry is the poll→classify→surface pattern: it scores inbox items with a cheap classifier model (`auxiliary.monitor` in `config.yaml`) and delivers only the ones above an urgency threshold, staying silent otherwise.
## Publishing Skills
### To the Skills Hub