ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

refactor(tools): compact credential_files, blueprints, arg_coercion

Browse Source 
- credential_files: _walk_skill_tree unifies the os.walk/EXCLUDED_SKILL_DIRS/symlink-skip
  loop shared by _safe_skills_path and iter_skills_files (drops _iter_syncable_files);
  _remap_cache_path unifies map_cache_path_to_container/from_agent_visible_cache_path;
  _iter_regular_files inlined into its only caller iter_cache_files.
- arg_coercion: _SCALAR_COERCERS dispatch table replaces the if/elif type ladder in
  _coerce_value; identity-preserving normalize helpers compacted. Logger name and
  every message text unchanged.
- blueprints: frontmatter split, job-spec and export compacted; __all__ unchanged.
This commit is contained in:
Teknium
2026-09-02 22:36:15 -07:00
parent b428887b98
commit ffbd5ef633
3 changed files with 194 additions and 445 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/blueprints.py
+36 -102
View File
@@ -1,21 +1,10 @@
"""Blueprints: shareable plain-language automations layered on skills + cron.
A "blueprint" 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:
blueprint:
schedule: "0 9 * * *" # presence of `blueprint:` marks it runnable
deliver: origin # optional (default "origin")
prompt: "..." # optional task instruction for the run
no_agent: false # optional
Because a blueprint is just a skill it rides the whole skills-hub pipeline
(search, scan, install, provenance, publish) for free; this module is only the
bridge from that frontmatter to the cron ``create_job()`` API, plus the inverse
(``export_blueprint``) that renders a cron job back into a shareable SKILL.md.
A blueprint is NOT a new object type: it is an ordinary skill whose frontmatter declares
``metadata.hermes.blueprint`` (``schedule`` required; optional ``deliver`` [default "origin"],
``prompt``, ``no_agent``, ``model``, ``provider``, ``enabled_toolsets``), so it rides the whole
skills-hub pipeline for free. This module only bridges that block to cron ``create_job()``,
plus the inverse (``export_blueprint``) back to a SKILL.md.
"""
from __future__ import annotations
@@ -27,16 +16,8 @@ from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
__all__ = [
"BlueprintSpec",
"parse_blueprint",
"blueprint_spec_for_installed",
"blueprint_to_job_spec",
"create_blueprint_job",
"register_blueprint_suggestion",
"export_blueprint",
"BlueprintError",
]
__all__ = ["BlueprintSpec", "parse_blueprint", "blueprint_spec_for_installed", "blueprint_to_job_spec",
"create_blueprint_job", "register_blueprint_suggestion", "export_blueprint", "BlueprintError"]
class BlueprintError(ValueError):
@@ -63,18 +44,12 @@ def _split_frontmatter(text: str) -> Optional[Dict[str, Any]]:
if not isinstance(text, str):
return None
stripped = text.lstrip("\ufeff").lstrip() # BOM is not whitespace; strip explicitly
if not stripped.startswith("---"):
if not stripped.startswith("---") or (end := stripped.find("\n---", 3)) == -1:
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)
data = yaml.safe_load(stripped[3:end])
except Exception as e: # pragma: no cover - malformed YAML
logger.debug("blueprint: frontmatter YAML parse failed: %s", e)
return None
@@ -84,16 +59,14 @@ def _split_frontmatter(text: str) -> Optional[Dict[str, Any]]:
def parse_blueprint(skill_md_text: str) -> Optional[BlueprintSpec]:
"""Extract a BlueprintSpec from a SKILL.md string, or None if not a blueprint.
A skill is a blueprint iff ``metadata.hermes.blueprint`` is a mapping containing
a non-empty ``schedule``. Raises BlueprintError if the block exists but is
A skill is a blueprint iff ``metadata.hermes.blueprint`` is a mapping with a
non-empty ``schedule``. Raises BlueprintError 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
blueprint = hermes.get("blueprint") if isinstance(hermes, dict) else None
@@ -106,16 +79,13 @@ def parse_blueprint(skill_md_text: str) -> Optional[BlueprintSpec]:
if not schedule:
raise BlueprintError("blueprint.schedule is required and must be non-empty")
prompt = blueprint.get("prompt")
model = blueprint.get("model")
provider = blueprint.get("provider")
prompt, model, provider = blueprint.get("prompt"), blueprint.get("model"), blueprint.get("provider")
toolsets = blueprint.get("enabled_toolsets")
if toolsets is not None and not isinstance(toolsets, list):
raise BlueprintError("blueprint.enabled_toolsets must be a list when present")
return BlueprintSpec(
skill_name=name,
schedule=schedule,
skill_name=str(fm.get("name", "")).strip(), schedule=schedule,
deliver=str(blueprint.get("deliver", "origin")).strip() or "origin",
prompt=str(prompt) if prompt is not None else None,
no_agent=bool(blueprint.get("no_agent", False)),
@@ -133,48 +103,31 @@ def blueprint_spec_for_installed(skill_name: str) -> Optional[BlueprintSpec]:
from tools.skills_hub import SKILLS_DIR
except Exception: # pragma: no cover - import guard
return None
# Skills live at skills/<category>/<name>/SKILL.md or skills/<name>/SKILL.md.
for path in Path(SKILLS_DIR).glob(f"**/{skill_name}/SKILL.md"):
try:
text = path.read_text(encoding="utf-8")
spec = parse_blueprint(path.read_text(encoding="utf-8"))
except OSError:
continue
spec = parse_blueprint(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
spec.skill_name = spec.skill_name or skill_name # frontmatter name wins over dir name
return spec
return None
def blueprint_to_job_spec(
spec: BlueprintSpec,
*,
name: Optional[str] = None,
) -> Dict[str, Any]:
def blueprint_to_job_spec(spec: BlueprintSpec, *, name: Optional[str] = None) -> Dict[str, Any]:
"""``cron.jobs.create_job`` kwargs for a spec — the single translation used by
both ``create_blueprint_job`` and the suggestion path so they never drift."""
return {
"prompt": spec.prompt,
"schedule": spec.schedule,
"name": name or f"blueprint:{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,
"prompt": spec.prompt, "schedule": spec.schedule, "name": name or f"blueprint:{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_blueprint_job(
spec: BlueprintSpec,
*,
origin: Optional[Dict[str, Any]] = None,
name: Optional[str] = None,
) -> Dict[str, Any]:
def create_blueprint_job(spec: BlueprintSpec, *, origin: Optional[Dict[str, Any]] = None,
name: Optional[str] = None) -> Dict[str, Any]:
"""Create the cron job for a spec (skill preloaded via ``skills=[name]``); returns the job dict."""
from cron.scheduler import create_job_with_scheduler_registration
@@ -194,13 +147,10 @@ def register_blueprint_suggestion(spec: BlueprintSpec) -> Optional[Dict[str, Any
except Exception: # pragma: no cover - import guard
return None
deliver = f", delivering to {spec.deliver}" if spec.deliver and spec.deliver != "origin" else ""
return add_suggestion(
title=f"Schedule '{spec.skill_name}'",
description=(
f"The '{spec.skill_name}' blueprint runs on schedule {spec.schedule}"
+ (f", delivering to {spec.deliver}" if spec.deliver and spec.deliver != "origin" else "")
+ "."
),
description=f"The '{spec.skill_name}' blueprint runs on schedule {spec.schedule}{deliver}.",
source="blueprint",
job_spec=blueprint_to_job_spec(spec),
dedup_key=f"blueprint:{spec.skill_name}:{spec.schedule}",
@@ -213,38 +163,24 @@ def export_blueprint(job: Dict[str, Any], body: str, *, blueprint_name: Optional
``body`` becomes the SKILL.md body; its first line is the description."""
import yaml
name = blueprint_name or job.get("name") or "shared-blueprint"
# 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-blueprint"
name = str(blueprint_name or job.get("name") or "shared-blueprint").lower()
name = "".join(c if (c.isalnum() or c in "-_") else "-" for c in name).strip("-_") or "shared-blueprint"
blueprint_block: Dict[str, Any] = {
"schedule": job.get("schedule_display") or _schedule_to_string(job.get("schedule")),
}
block: Dict[str, Any] = {"schedule": job.get("schedule_display") or _schedule_to_string(job.get("schedule"))}
if job.get("deliver") and job["deliver"] != "origin":
blueprint_block["deliver"] = job["deliver"]
block["deliver"] = job["deliver"]
if job.get("prompt"):
blueprint_block["prompt"] = job["prompt"]
block["prompt"] = job["prompt"]
if job.get("no_agent"):
blueprint_block["no_agent"] = True
for key in ("model", "provider", "enabled_toolsets"):
if job.get(key):
blueprint_block[key] = job[key]
block["no_agent"] = True
block.update({k: job[k] for k in ("model", "provider", "enabled_toolsets") if job.get(k)})
body = body.strip()
description = body.splitlines()[0][:200] if body else "Shared automation blueprint."
frontmatter = {
"name": name,
"description": description,
"version": "1.0.0",
"license": "MIT",
"metadata": {
"hermes": {
"tags": ["blueprint", "automation"],
"blueprint": blueprint_block,
}
},
"name": name, "description": body.splitlines()[0][:200] if body else "Shared automation blueprint.",
"version": "1.0.0", "license": "MIT",
"metadata": {"hermes": {"tags": ["blueprint", "automation"], "blueprint": block}},
}
fm_yaml = yaml.safe_dump(frontmatter, sort_keys=False, allow_unicode=True).strip()
body_text = body or f"# {name}\n\nShared automation blueprint."
@@ -260,14 +196,12 @@ def _schedule_to_string(schedule: Any) -> str:
if kind == "cron" and schedule.get("expr"):
return str(schedule["expr"])
if kind == "interval":
# parse_schedule stores interval periods as "minutes"; tolerate a
# legacy/foreign "seconds" form too.
# parse_schedule stores interval periods as "minutes"; tolerate a legacy/foreign "seconds" form too.
if schedule.get("minutes"):
mins = int(schedule["minutes"])
return f"every {mins // 60}h" if mins % 60 == 0 else f"every {mins}m"
if schedule.get("seconds"):
secs = int(schedule["seconds"])
if secs % 3600 == 0:
return f"every {secs // 3600}h"
return f"every {secs // 60}m" if secs % 60 == 0 else f"every {secs}s"
return (f"every {secs // 3600}h" if secs % 3600 == 0
else f"every {secs // 60}m" if secs % 60 == 0 else f"every {secs}s")
return "0 9 * * *" # safe daily fallback