"""Declarative skills and protocols: format, validation, rendering.
A skill or protocol is one document: a YAML front matter block followed by a
markdown body. The front matter carries what the catalog needs (name, purpose,
triggers, tags, area, and — for protocols — the steps and their escape hatches);
the body is the instruction text the agent reads when it loads the entry.
Two kinds live in one registry because they share a catalog, search, versioning
and ownership. They differ in what they must declare:
- **skill** — flexible knowledge about an approach: when it is useful and how to
go about it. Nothing is mandatory beyond a name, a purpose and a body.
- **protocol** — a strict order of actions. Steps, post-checks and the three
escape hatches are mandatory: what to do when the run drifts off plan, when it
cannot be completed, and when a step turns out to be already done. A protocol
the agent cannot act on when reality differs from the plan is worse than no
protocol at all, so validation refuses it.
Validation messages are English and agent-facing: the same text reaches the
model when it tries to save a malformed document.
"""
from __future__ import annotations
import hashlib
import re
from dataclasses import dataclass, field
from typing import Any
import yaml
KINDS = ("skill", "protocol")
AREAS = ("home", "infrastructure", "workflows", "general")
VISIBILITIES = ("personal", "shared")
# Mandatory front matter keys per kind. `steps` and the three handlers are the
# part that makes a protocol a protocol; a skill needs none of them.
_PROTOCOL_REQUIRED = ("steps", "post_checks", "on_deviation", "on_blocked", "on_already_done")
_FRONTMATTER = re.compile(r"\A---[ \t]*\r?\n(.*?)\r?\n---[ \t]*(?:\r?\n|\Z)", re.S)
_MAX_DOC_CHARS = 200_000
class SkillError(ValueError):
"""A document cannot be used; the message is meant for the caller."""
@dataclass
class SkillDoc:
kind: str
name: str
description: str = ""
triggers: list[str] = field(default_factory=list)
tags: list[str] = field(default_factory=list)
area: str = "general"
steps: list[str] = field(default_factory=list)
preconditions: list[str] = field(default_factory=list)
post_checks: list[str] = field(default_factory=list)
on_deviation: str = ""
on_blocked: str = ""
on_already_done: str = ""
body: str = ""
extra: dict[str, Any] = field(default_factory=dict)
def frontmatter(self) -> dict[str, Any]:
"""The searchable metadata as stored in the JSONB column."""
return {
"kind": self.kind,
"name": self.name,
"description": self.description,
"triggers": self.triggers,
"tags": self.tags,
"area": self.area,
"steps": self.steps,
"preconditions": self.preconditions,
"post_checks": self.post_checks,
"on_deviation": self.on_deviation,
"on_blocked": self.on_blocked,
"on_already_done": self.on_already_done,
**self.extra,
}
def content_hash(content: str) -> str:
return hashlib.sha256(content.encode("utf-8")).hexdigest()
def _as_list(value: Any) -> list[str]:
if value is None or value == "":
return []
if isinstance(value, str):
return [value.strip()] if value.strip() else []
if isinstance(value, (list, tuple)):
return [str(v).strip() for v in value if str(v).strip()]
raise SkillError(f"expected a list of strings, got {type(value).__name__}")
def _as_text(value: Any) -> str:
if value is None:
return ""
if isinstance(value, str):
return value.strip()
raise SkillError(f"expected a string, got {type(value).__name__}")
def parse_document(content: str, *, kind: str | None = None) -> SkillDoc:
"""Parse and validate one document.
``kind`` overrides the front matter (used when the caller already knows
whether it is creating a skill or a protocol). The front matter's own
``kind`` otherwise decides; neither present is an error.
"""
if not isinstance(content, str) or not content.strip():
raise SkillError("document is empty")
if len(content) > _MAX_DOC_CHARS:
raise SkillError(f"document is too large (limit {_MAX_DOC_CHARS} characters)")
match = _FRONTMATTER.match(content)
if not match:
raise SkillError(
"document must start with a YAML front matter block delimited by '---' lines"
)
try:
meta = yaml.safe_load(match.group(1)) or {}
except yaml.YAMLError as e:
raise SkillError(f"invalid YAML front matter: {e}") from e
if not isinstance(meta, dict):
raise SkillError("YAML front matter must be a mapping of keys to values")
body = content[match.end():].strip()
resolved_kind = (kind or meta.get("kind") or "").strip().lower()
if not resolved_kind:
raise SkillError("'kind' is required: 'skill' or 'protocol'")
if resolved_kind not in KINDS:
raise SkillError(f"'kind' must be one of {', '.join(KINDS)}; got {resolved_kind!r}")
if kind and meta.get("kind") and str(meta["kind"]).strip().lower() != resolved_kind:
raise SkillError(
f"front matter says kind={meta['kind']!r} but the request says {kind!r}"
)
name = _as_text(meta.get("name"))
if not name:
raise SkillError("'name' is required in the front matter")
description = _as_text(meta.get("description"))
area = _as_text(meta.get("area")) or "general"
if area not in AREAS:
raise SkillError(f"'area' must be one of {', '.join(AREAS)}; got {area!r}")
known = {
"kind", "name", "description", "triggers", "tags", "area",
"steps", "preconditions", "post_checks", "on_deviation",
"on_blocked", "on_already_done",
}
doc = SkillDoc(
kind=resolved_kind,
name=name,
description=description,
triggers=_as_list(meta.get("triggers")),
tags=_as_list(meta.get("tags")),
area=area,
steps=_as_list(meta.get("steps")),
preconditions=_as_list(meta.get("preconditions")),
post_checks=_as_list(meta.get("post_checks")),
on_deviation=_as_text(meta.get("on_deviation")),
on_blocked=_as_text(meta.get("on_blocked")),
on_already_done=_as_text(meta.get("on_already_done")),
body=body,
extra={k: v for k, v in meta.items() if k not in known},
)
if doc.kind == "skill":
if not doc.body:
raise SkillError("a skill needs a body (the instructions below the front matter)")
return doc
missing = [
key for key in _PROTOCOL_REQUIRED
if not getattr(doc, key, None)
]
if missing:
raise SkillError(
"a protocol must declare " + ", ".join(f"'{m}'" for m in missing) +
" in the front matter — steps, how each step is checked, and what to "
"do when the run drifts off plan, cannot be completed, or meets a step "
"that is already done"
)
return doc
def render_for_agent(
doc: SkillDoc,
*,
entry_id: str = "",
version: int = 0,
visibility: str = "personal",
) -> str:
"""The text the agent reads after loading an entry.
Compact on purpose: the model already has the conversation; this is the
methodology or the ordered steps, not a second copy of the catalog.
"""
lines: list[str] = []
title = "PROTOCOL" if doc.kind == "protocol" else "SKILL"
lines.append(f"{title}: {doc.name}" + (f" (v{version})" if version else ""))
if doc.description:
lines.append(doc.description)
meta_bits = []
if doc.area:
meta_bits.append(f"area: {doc.area}")
if doc.tags:
meta_bits.append("tags: " + ", ".join(doc.tags))
if doc.triggers:
meta_bits.append("use when: " + "; ".join(doc.triggers))
if meta_bits:
lines.append(" | ".join(meta_bits))
if doc.kind == "protocol":
if doc.preconditions:
lines.append("")
lines.append("Preconditions:")
lines.extend(f"- {p}" for p in doc.preconditions)
lines.append("")
lines.append("Steps (follow in order):")
lines.extend(f"{i}. {s}" for i, s in enumerate(doc.steps, 1))
lines.append("")
lines.append("After each step:" if len(doc.post_checks) == 1 else "Post-checks:")
lines.extend(f"- {c}" for c in doc.post_checks)
lines.append("")
lines.append("If the run goes off plan: " + doc.on_deviation)
lines.append("If the protocol cannot be completed: " + doc.on_blocked)
lines.append("If a step is already done: " + doc.on_already_done)
if doc.body:
lines.append("")
lines.append(doc.body)
else:
if doc.body:
lines.append("")
lines.append(doc.body)
if entry_id:
lines.append("")
lines.append(f"[{doc.kind} id: {entry_id}]")
return "\n".join(lines)
