tegwick
c5f1640454
`evals/` was a reserved path holding unvalidated blobs: section 12 named the directory and gave an illustrative snippet, but nothing was specified, so no tool could act on an eval file. Every eval file must now declare a `schema`, and CPF defines exactly one — `canned-prompts/eval-rubric/v0.1`. Unrecognized schemas stay legal and are skipped rather than rejected, so the format gains something actionable without becoming an evaluation language, which remains a non-goal. The schema splits along the same seam as T01 and T03. Render checks (`contains`, `not_contains`, `resolves_all`) assert properties of the rendered prompt text, need no model, and are therefore run by the reference CLI. Output criteria describe a good result and are declared but not run, because judging them requires a model. That division is now the format's consistent answer to "deterministic locally, or not". An eval references a fixture already declared in the manifest's `examples` rather than carrying its own copy, so an example that is also an eval fixture stays honest — both break together. An eval declares assessment and must not record outcomes. Results are run evidence and live outside the immutable package, per INTENT.md and section 17. Spec: 12 rewritten with 12.1, 18 (rules 17-18), 21 (`eval` verb). Reference CLI: read_eval, validate_eval, load_example_values, run_render_checks, cmd_eval; a failed render check exits non-zero. Tests 42 -> 51. examples/pqrst-estimate/evals/quality.yaml is a real eval with four render checks and four output criteria, and it passes. Fixes a latent bug reaching a fixture exposed: coerce_value assumed every value was a command-line string, so a YAML fixture carrying a real type (include_rationale: true) crashed on .lower(). Typed values are now validated but not re-parsed. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bjefh8NUiEiahN4JLwoSKM Assistant: claude-code Assistant-Model: opus Assistant-Process: 388925@bnt-lap001 Assistant-Session: 3507023f-e0fd-4a1e-9d90-a0d4217d1502
1112 lines
42 KiB
Python
Executable file
1112 lines
42 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Tiny reference CLI for Canned Prompt Format v0.1.
|
|
|
|
This implementation intentionally favors readability over features. It uses a
|
|
filesystem-backed local catalog and registry and never calls a model.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
import yaml
|
|
|
|
FORMAT = "canned-prompt/v0.1"
|
|
REGISTRY_FORMAT = "canned-prompt-registry/v0.1"
|
|
LOCAL_REGISTRY = "local"
|
|
REGISTRY_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
|
|
EVAL_RUBRIC_SCHEMA = "canned-prompts/eval-rubric/v0.1"
|
|
RENDER_CHECKS = ("contains", "not_contains", "resolves_all")
|
|
PLACEHOLDER_RE = re.compile(r"{{\s*([A-Za-z_][A-Za-z0-9_.-]*)\s*}}")
|
|
SEMVER_RE = re.compile(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:[-+].*)?$")
|
|
REQUIRED_FIELDS = ("format", "id", "name", "version", "summary", "template")
|
|
|
|
|
|
class CannedPromptError(Exception):
|
|
pass
|
|
|
|
|
|
def home_dir() -> Path:
|
|
return Path(os.environ.get("CANNED_PROMPTS_HOME", Path.home() / ".canned-prompts"))
|
|
|
|
|
|
def default_catalog() -> Path:
|
|
return home_dir() / "catalog"
|
|
|
|
|
|
def default_registry() -> Path:
|
|
return home_dir() / "registry"
|
|
|
|
|
|
def read_manifest(package_dir: Path) -> dict[str, Any]:
|
|
manifest_path = package_dir / "prompt.yaml"
|
|
if not manifest_path.is_file():
|
|
raise CannedPromptError(f"missing manifest: {manifest_path}")
|
|
try:
|
|
data = yaml.safe_load(manifest_path.read_text(encoding="utf-8"))
|
|
except yaml.YAMLError as exc:
|
|
raise CannedPromptError(f"invalid YAML in {manifest_path}: {exc}") from exc
|
|
if not isinstance(data, dict):
|
|
raise CannedPromptError("prompt.yaml must contain a mapping")
|
|
return data
|
|
|
|
|
|
def safe_relative_file(package_dir: Path, relative: str, field: str) -> Path:
|
|
if not isinstance(relative, str) or not relative.strip():
|
|
raise CannedPromptError(f"{field} must be a non-empty relative path")
|
|
candidate = (package_dir / relative).resolve()
|
|
root = package_dir.resolve()
|
|
try:
|
|
candidate.relative_to(root)
|
|
except ValueError as exc:
|
|
raise CannedPromptError(f"{field} escapes the package: {relative}") from exc
|
|
if not candidate.is_file():
|
|
raise CannedPromptError(f"{field} does not reference a file: {relative}")
|
|
return candidate
|
|
|
|
|
|
def declared_names(manifest: dict[str, Any]) -> set[str]:
|
|
names: set[str] = set()
|
|
inputs = manifest.get("inputs") or []
|
|
if not isinstance(inputs, list):
|
|
raise CannedPromptError("inputs must be a list")
|
|
for item in inputs:
|
|
if not isinstance(item, dict) or not isinstance(item.get("name"), str):
|
|
raise CannedPromptError("each input must be a mapping with a string name")
|
|
name = item["name"]
|
|
if name in names:
|
|
raise CannedPromptError(f"duplicate input/parameter name: {name}")
|
|
names.add(name)
|
|
|
|
parameters = manifest.get("parameters") or {}
|
|
if not isinstance(parameters, dict):
|
|
raise CannedPromptError("parameters must be a mapping")
|
|
for name, spec in parameters.items():
|
|
if not isinstance(name, str) or not isinstance(spec, dict):
|
|
raise CannedPromptError("parameters must map names to mappings")
|
|
if name in names:
|
|
raise CannedPromptError(f"duplicate input/parameter name: {name}")
|
|
names.add(name)
|
|
return names
|
|
|
|
|
|
def input_default_kind(item: dict[str, Any]) -> str:
|
|
"""Classify a default as 'none', 'static', 'derived', or 'included' (§ 6.1)."""
|
|
if "default" not in item:
|
|
return "none"
|
|
default = item["default"]
|
|
if isinstance(default, dict):
|
|
if "derive" in default and "include" in default:
|
|
return "conflict"
|
|
if "derive" in default:
|
|
return "derived"
|
|
if "include" in default:
|
|
return "included"
|
|
return "static"
|
|
|
|
|
|
def prompt_dependencies(manifest: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
|
dependencies = manifest.get("dependencies") or {}
|
|
if not isinstance(dependencies, dict):
|
|
raise CannedPromptError("dependencies must be a mapping")
|
|
prompts = dependencies.get("prompts") or []
|
|
if not isinstance(prompts, list):
|
|
raise CannedPromptError("dependencies.prompts must be a list")
|
|
declared: dict[str, dict[str, Any]] = {}
|
|
for entry in prompts:
|
|
if isinstance(entry, str):
|
|
declared[entry] = {"id": entry}
|
|
elif isinstance(entry, dict) and isinstance(entry.get("id"), str):
|
|
if "version" in entry:
|
|
validate_version_selector(
|
|
entry["version"], f"dependency {entry['id']!r}"
|
|
)
|
|
declared[entry["id"]] = entry
|
|
else:
|
|
raise CannedPromptError(
|
|
"each prompt dependency must be an id string or a mapping with a string id"
|
|
)
|
|
return declared
|
|
|
|
|
|
def read_eval(package_dir: Path, relative: str) -> dict[str, Any]:
|
|
path = safe_relative_file(package_dir, relative, "evals")
|
|
try:
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
except yaml.YAMLError as exc:
|
|
raise CannedPromptError(f"invalid YAML in {relative}: {exc}") from exc
|
|
if not isinstance(data, dict):
|
|
raise CannedPromptError(f"{relative}: an eval file must contain a mapping")
|
|
if not isinstance(data.get("schema"), str) or not data["schema"].strip():
|
|
raise CannedPromptError(f"{relative}: an eval file must declare a schema")
|
|
return data
|
|
|
|
|
|
def validate_eval(relative: str, spec: dict[str, Any], declared_examples: list[str]) -> None:
|
|
"""Validation rules 17-18 of § 18. Unknown schemas are ignored, not rejected."""
|
|
if spec["schema"] != EVAL_RUBRIC_SCHEMA:
|
|
return
|
|
if not isinstance(spec.get("name"), str) or not spec["name"].strip():
|
|
raise CannedPromptError(f"{relative}: eval must declare a name")
|
|
|
|
render = spec.get("render") or []
|
|
output = spec.get("output") or {}
|
|
if not render and not output:
|
|
raise CannedPromptError(
|
|
f"{relative}: eval asserts nothing; declare render checks or output criteria"
|
|
)
|
|
|
|
if not isinstance(render, list):
|
|
raise CannedPromptError(f"{relative}: render must be a list of checks")
|
|
for check in render:
|
|
if not isinstance(check, dict) or len(check) != 1:
|
|
raise CannedPromptError(
|
|
f"{relative}: each render check must be a single-key mapping"
|
|
)
|
|
(kind,) = check
|
|
if kind not in RENDER_CHECKS:
|
|
raise CannedPromptError(
|
|
f"{relative}: unknown render check {kind!r}; expected one of "
|
|
+ ", ".join(RENDER_CHECKS)
|
|
)
|
|
|
|
if output:
|
|
if not isinstance(output, dict):
|
|
raise CannedPromptError(f"{relative}: output must be a mapping")
|
|
criteria = output.get("criteria") or []
|
|
if not isinstance(criteria, list) or not all(isinstance(c, str) for c in criteria):
|
|
raise CannedPromptError(f"{relative}: output.criteria must be a list of strings")
|
|
|
|
example = spec.get("example")
|
|
if example is not None:
|
|
if not isinstance(example, str):
|
|
raise CannedPromptError(f"{relative}: example must be a path string")
|
|
if example not in declared_examples:
|
|
raise CannedPromptError(
|
|
f"{relative}: example {example!r} is not declared in the manifest's examples"
|
|
)
|
|
|
|
|
|
def check_composition_reference(
|
|
name: str, reference: str, field: str, declared: dict[str, dict[str, Any]]
|
|
) -> None:
|
|
if not reference.strip():
|
|
raise CannedPromptError(f"input {name!r}: {field} reference is empty")
|
|
if reference not in declared:
|
|
raise CannedPromptError(
|
|
f"input {name!r} {field}s {reference!r}, which is not declared "
|
|
"in dependencies.prompts"
|
|
)
|
|
if "version" not in declared[reference]:
|
|
raise CannedPromptError(
|
|
f"input {name!r} {field}s {reference!r}, so that dependency must "
|
|
"declare a version (§ 10.1)"
|
|
)
|
|
|
|
|
|
def validate_input_default(
|
|
item: dict[str, Any], declared: dict[str, dict[str, Any]]
|
|
) -> None:
|
|
"""Validation rules 11-16 of § 18."""
|
|
kind = input_default_kind(item)
|
|
if kind == "none":
|
|
return
|
|
name = item["name"]
|
|
if item.get("required", False):
|
|
raise CannedPromptError(
|
|
f"input {name!r} declares a default and required: true; a required "
|
|
"input is always supplied, so the default could never apply"
|
|
)
|
|
if kind == "static":
|
|
return
|
|
if kind == "conflict":
|
|
raise CannedPromptError(
|
|
f"input {name!r}: a default declares at most one of 'include' or 'derive'"
|
|
)
|
|
|
|
if kind == "included":
|
|
include = item["default"]["include"]
|
|
if not isinstance(include, str):
|
|
raise CannedPromptError(
|
|
f"input {name!r}: include must name a declared prompt dependency"
|
|
)
|
|
check_composition_reference(name, include, "include", declared)
|
|
return
|
|
|
|
derive = item["default"]["derive"]
|
|
if isinstance(derive, str):
|
|
check_composition_reference(name, derive, "derive", declared)
|
|
return
|
|
if isinstance(derive, dict):
|
|
prompt = derive.get("prompt")
|
|
if not isinstance(prompt, str) or not prompt.strip():
|
|
raise CannedPromptError(
|
|
f"input {name!r}: inline derive requires a non-empty prompt"
|
|
)
|
|
extra = sorted(set(derive) - {"prompt"})
|
|
if extra:
|
|
raise CannedPromptError(
|
|
f"input {name!r}: an inline derive cannot also reference a "
|
|
"dependency; unexpected keys: " + ", ".join(extra)
|
|
)
|
|
return
|
|
raise CannedPromptError(
|
|
f"input {name!r}: derive must be a dependency id or a mapping with a prompt"
|
|
)
|
|
|
|
|
|
def validate_package(package_dir: Path) -> dict[str, Any]:
|
|
package_dir = package_dir.resolve()
|
|
if not package_dir.is_dir():
|
|
raise CannedPromptError(f"package directory not found: {package_dir}")
|
|
|
|
manifest = read_manifest(package_dir)
|
|
missing = [field for field in REQUIRED_FIELDS if field not in manifest]
|
|
if missing:
|
|
raise CannedPromptError("missing required fields: " + ", ".join(missing))
|
|
|
|
if manifest["format"] != FORMAT:
|
|
raise CannedPromptError(f"unsupported format: {manifest['format']!r}")
|
|
|
|
package_id = manifest["id"]
|
|
if not isinstance(package_id, str) or not package_id.strip():
|
|
raise CannedPromptError("id must be a non-empty string")
|
|
if any(part in ("", ".", "..") for part in package_id.split("/")):
|
|
raise CannedPromptError("id contains an invalid path segment")
|
|
if re.search(r"[^A-Za-z0-9._/-]", package_id):
|
|
raise CannedPromptError("id contains unsupported characters")
|
|
|
|
version = manifest["version"]
|
|
if not isinstance(version, str) or not SEMVER_RE.match(version):
|
|
raise CannedPromptError("version must be semantic-version-like, e.g. 1.2.0")
|
|
|
|
template_path = safe_relative_file(package_dir, manifest["template"], "template")
|
|
|
|
for field in ("examples", "evals"):
|
|
refs = manifest.get(field) or []
|
|
if not isinstance(refs, list):
|
|
raise CannedPromptError(f"{field} must be a list")
|
|
for relative in refs:
|
|
safe_relative_file(package_dir, relative, field)
|
|
|
|
names = declared_names(manifest)
|
|
|
|
declared = prompt_dependencies(manifest)
|
|
for item in manifest.get("inputs") or []:
|
|
validate_input_default(item, declared)
|
|
|
|
declared_examples = [str(path) for path in (manifest.get("examples") or [])]
|
|
for relative in manifest.get("evals") or []:
|
|
validate_eval(relative, read_eval(package_dir, relative), declared_examples)
|
|
|
|
template = template_path.read_text(encoding="utf-8")
|
|
placeholders = set(PLACEHOLDER_RE.findall(template))
|
|
undeclared = sorted(placeholders - names)
|
|
if undeclared:
|
|
raise CannedPromptError(
|
|
"template contains undeclared placeholders: " + ", ".join(undeclared)
|
|
)
|
|
|
|
return manifest
|
|
|
|
|
|
def check_registry_name(name: str) -> str:
|
|
if not isinstance(name, str) or not REGISTRY_NAME_RE.match(name):
|
|
raise CannedPromptError(
|
|
f"invalid registry name: {name!r}; expected letters, digits, '.', '-' or '_'"
|
|
)
|
|
return name
|
|
|
|
|
|
def parse_reference(reference: str) -> tuple[str | None, str]:
|
|
"""Split a qualified `<registry>:<id>` reference (§ 3.2)."""
|
|
if ":" not in reference:
|
|
return None, reference
|
|
registry, _, package_id = reference.partition(":")
|
|
if not registry or not package_id:
|
|
raise CannedPromptError(f"malformed reference: {reference!r}")
|
|
return check_registry_name(registry), package_id
|
|
|
|
|
|
def read_registry_manifest(registry: Path) -> dict[str, Any] | None:
|
|
"""Read an optional registry.yaml (§ 20.1). A bare directory is still valid."""
|
|
manifest_path = registry / "registry.yaml"
|
|
if not manifest_path.is_file():
|
|
return None
|
|
try:
|
|
data = yaml.safe_load(manifest_path.read_text(encoding="utf-8"))
|
|
except yaml.YAMLError as exc:
|
|
raise CannedPromptError(f"invalid YAML in {manifest_path}: {exc}") from exc
|
|
if not isinstance(data, dict):
|
|
raise CannedPromptError("registry.yaml must contain a mapping")
|
|
if data.get("format") != REGISTRY_FORMAT:
|
|
raise CannedPromptError(f"unsupported registry format: {data.get('format')!r}")
|
|
check_registry_name(data.get("name"))
|
|
|
|
namespaces = data.get("namespaces") or {}
|
|
if not isinstance(namespaces, dict):
|
|
raise CannedPromptError("registry namespaces must be a mapping")
|
|
for namespace, claim in namespaces.items():
|
|
if not isinstance(claim, dict):
|
|
raise CannedPromptError(f"namespace {namespace!r} must map to a mapping")
|
|
policy = claim.get("policy", "open")
|
|
if policy not in ("open", "closed"):
|
|
raise CannedPromptError(
|
|
f"namespace {namespace!r}: policy must be 'open' or 'closed'"
|
|
)
|
|
return data
|
|
|
|
|
|
def registry_name(registry: Path) -> str:
|
|
"""A registry's name: its manifest's, else its directory basename (§ 20.1)."""
|
|
manifest = read_registry_manifest(registry)
|
|
if manifest:
|
|
return manifest["name"]
|
|
return check_registry_name(registry.expanduser().resolve().name)
|
|
|
|
|
|
def namespace_policy(registry: Path, package_id: str) -> tuple[str, dict[str, Any] | None]:
|
|
manifest = read_registry_manifest(registry)
|
|
if not manifest:
|
|
return "open", None
|
|
namespace = package_id.split("/")[0] if "/" in package_id else package_id
|
|
claim = (manifest.get("namespaces") or {}).get(namespace)
|
|
if not claim:
|
|
return "open", None
|
|
return claim.get("policy", "open"), claim
|
|
|
|
|
|
def id_path(store: Path, package_id: str) -> Path:
|
|
parts = package_id.split("/")
|
|
if any(part in ("", ".", "..") for part in parts):
|
|
raise CannedPromptError("unsafe package id")
|
|
return store.joinpath(*parts)
|
|
|
|
|
|
def registry_package_path(registry: Path, package_id: str, version: str) -> Path:
|
|
"""A registry's own layout is flat: an id is unambiguous within a registry."""
|
|
return id_path(registry, package_id) / version
|
|
|
|
|
|
def catalog_package_path(catalog: Path, registry: str, package_id: str, version: str) -> Path:
|
|
"""The catalog is namespaced by registry (§ 20.2)."""
|
|
return id_path(catalog / check_registry_name(registry), package_id) / version
|
|
|
|
|
|
def validate_version_selector(value: Any, where: str) -> str:
|
|
"""A semver literal, `any`, `newest`, or a `>=` lower bound (§ 10.1)."""
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise CannedPromptError(f"{where}: version must be a non-empty string")
|
|
selector = value.strip()
|
|
if selector in ("any", "newest"):
|
|
return selector
|
|
compact = selector.replace(" ", "")
|
|
if compact.startswith(">="):
|
|
if not SEMVER_RE.match(compact[2:]):
|
|
raise CannedPromptError(f"{where}: {value!r} is not a valid lower bound")
|
|
return compact
|
|
if not SEMVER_RE.match(selector):
|
|
raise CannedPromptError(
|
|
f"{where}: unsupported version selector {value!r}; expected a semver "
|
|
"literal, 'any', 'newest', or '>= X.Y.Z'"
|
|
)
|
|
return selector
|
|
|
|
|
|
def select_version(available: list[str], selector: str | None) -> str | None:
|
|
"""Pick a version from `available` (newest first) per a § 10.1 selector.
|
|
|
|
Each dependency is selected independently against what is present. There is
|
|
no constraint solving across a dependency graph; that stays a non-goal.
|
|
"""
|
|
if not available:
|
|
return None
|
|
if selector is None or selector in ("any", "newest"):
|
|
return available[0]
|
|
if selector.startswith(">="):
|
|
floor = parse_semver(selector[2:])
|
|
for version in available:
|
|
if parse_semver(version) >= floor:
|
|
return version
|
|
return None
|
|
return selector if selector in available else None
|
|
|
|
|
|
def parse_semver(value: str) -> tuple[int, int, int, str]:
|
|
match = SEMVER_RE.match(value)
|
|
if not match:
|
|
return (-1, -1, -1, value)
|
|
return (int(match.group(1)), int(match.group(2)), int(match.group(3)), value)
|
|
|
|
|
|
def versions_at(base: Path) -> list[str]:
|
|
if not base.is_dir():
|
|
return []
|
|
versions = [p.name for p in base.iterdir() if p.is_dir() and (p / "prompt.yaml").is_file()]
|
|
return sorted(versions, key=parse_semver, reverse=True)
|
|
|
|
|
|
def pick_version(base: Path, label: str, version: str | None) -> Path:
|
|
if version:
|
|
path = base / version
|
|
if not (path / "prompt.yaml").is_file():
|
|
raise CannedPromptError(f"package not found: {label}@{version}")
|
|
return path
|
|
versions = versions_at(base)
|
|
if not versions:
|
|
raise CannedPromptError(f"package not found: {label}")
|
|
return base / versions[0]
|
|
|
|
|
|
def resolve_in_registry(registry: Path, package_id: str, version: str | None) -> Path:
|
|
return pick_version(id_path(registry, package_id), package_id, version)
|
|
|
|
|
|
def catalog_registries(catalog: Path) -> list[str]:
|
|
if not catalog.is_dir():
|
|
return []
|
|
return sorted(p.name for p in catalog.iterdir() if p.is_dir())
|
|
|
|
|
|
def legacy_catalog_entry(catalog: Path, package_id: str) -> bool:
|
|
"""True if the pre-registry-scoped catalog layout holds this id."""
|
|
try:
|
|
base = id_path(catalog, package_id)
|
|
except CannedPromptError:
|
|
return False
|
|
return bool(versions_at(base))
|
|
|
|
|
|
def resolve_installed(
|
|
catalog: Path, reference: str, version: str | None
|
|
) -> tuple[Path, str]:
|
|
"""Resolve a bare id or a qualified `<registry>:<id>` against the catalog.
|
|
|
|
Returns the package directory and the registry it was installed from. A
|
|
bare id present in more than one registry is ambiguous and is reported, not
|
|
guessed (§ 3.2).
|
|
"""
|
|
registry, package_id = parse_reference(reference)
|
|
if registry is not None:
|
|
matches = [registry] if versions_at(id_path(catalog / registry, package_id)) else []
|
|
else:
|
|
matches = [
|
|
name
|
|
for name in catalog_registries(catalog)
|
|
if versions_at(id_path(catalog / name, package_id))
|
|
]
|
|
|
|
if not matches:
|
|
if registry is None and legacy_catalog_entry(catalog, package_id):
|
|
raise CannedPromptError(
|
|
f"{package_id} is stored in the pre-registry-scoped catalog layout; "
|
|
"catalogs are now namespaced by registry (§ 20.2). Re-add or "
|
|
"re-install the package, or move it under a registry directory"
|
|
)
|
|
raise CannedPromptError(f"package not found: {reference}")
|
|
if len(matches) > 1:
|
|
raise CannedPromptError(
|
|
f"{package_id} is installed from more than one registry: "
|
|
+ ", ".join(f"{name}:{package_id}" for name in matches)
|
|
+ " — qualify the reference"
|
|
)
|
|
|
|
name = matches[0]
|
|
return pick_version(id_path(catalog / name, package_id), f"{name}:{package_id}", version), name
|
|
|
|
|
|
def copy_immutable(src: Path, dst: Path, what: str) -> None:
|
|
if dst.exists():
|
|
raise CannedPromptError(f"{what} already exists: {dst}")
|
|
dst.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copytree(src, dst)
|
|
|
|
|
|
def iter_catalog(catalog: Path) -> Iterable[tuple[str, Path, dict[str, Any]]]:
|
|
for name in catalog_registries(catalog):
|
|
for manifest_path in (catalog / name).rglob("prompt.yaml"):
|
|
package_dir = manifest_path.parent
|
|
try:
|
|
manifest = validate_package(package_dir)
|
|
except CannedPromptError:
|
|
continue
|
|
yield name, package_dir, manifest
|
|
|
|
|
|
def cmd_add(args: argparse.Namespace) -> None:
|
|
src = Path(args.path)
|
|
manifest = validate_package(src)
|
|
catalog = Path(args.catalog).expanduser()
|
|
registry = check_registry_name(args.as_registry)
|
|
dst = catalog_package_path(catalog, registry, manifest["id"], manifest["version"])
|
|
copy_immutable(src.resolve(), dst, "catalog package")
|
|
print(f"added {registry}:{manifest['id']}@{manifest['version']} -> {dst}")
|
|
|
|
|
|
def cmd_publish(args: argparse.Namespace) -> None:
|
|
src = Path(args.path)
|
|
manifest = validate_package(src)
|
|
registry = Path(args.registry).expanduser()
|
|
name = registry_name(registry) if registry.is_dir() else None
|
|
|
|
policy, claim = namespace_policy(registry, manifest["id"]) if registry.is_dir() else ("open", None)
|
|
if policy == "closed":
|
|
owner = (claim or {}).get("owner")
|
|
suffix = f" (owner: {owner})" if owner else ""
|
|
print(
|
|
f"warning: namespace {manifest['id'].split('/')[0]!r} is declared closed"
|
|
f"{suffix}; this tool cannot authenticate a publisher",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
dst = registry_package_path(registry, manifest["id"], manifest["version"])
|
|
copy_immutable(src.resolve(), dst, "published package")
|
|
label = f"{name}:{manifest['id']}" if name else manifest["id"]
|
|
print(f"published {label}@{manifest['version']} -> {dst}")
|
|
|
|
|
|
def cmd_install(args: argparse.Namespace) -> None:
|
|
registry = Path(args.registry).expanduser()
|
|
catalog = Path(args.catalog).expanduser()
|
|
name = registry_name(registry)
|
|
|
|
qualifier, package_id = parse_reference(args.id)
|
|
if qualifier is not None and qualifier != name:
|
|
raise CannedPromptError(
|
|
f"reference names registry {qualifier!r} but --registry is {name!r}"
|
|
)
|
|
|
|
src = resolve_in_registry(registry, package_id, args.version)
|
|
manifest = validate_package(src)
|
|
dst = catalog_package_path(catalog, name, manifest["id"], manifest["version"])
|
|
copy_immutable(src, dst, "catalog package")
|
|
print(f"installed {name}:{manifest['id']}@{manifest['version']} -> {dst}")
|
|
|
|
|
|
def cmd_search(args: argparse.Namespace) -> None:
|
|
catalog = Path(args.catalog).expanduser()
|
|
query = args.query.lower()
|
|
matches: list[dict[str, str]] = []
|
|
for registry, _, manifest in iter_catalog(catalog):
|
|
haystack = " ".join(
|
|
[
|
|
str(manifest.get("id", "")),
|
|
str(manifest.get("name", "")),
|
|
str(manifest.get("summary", "")),
|
|
" ".join(str(tag) for tag in (manifest.get("tags") or [])),
|
|
]
|
|
).lower()
|
|
if query in haystack:
|
|
matches.append(
|
|
{
|
|
"registry": registry,
|
|
"id": manifest["id"],
|
|
"version": manifest["version"],
|
|
"name": manifest["name"],
|
|
"summary": manifest["summary"],
|
|
}
|
|
)
|
|
matches.sort(key=lambda m: (m["registry"], m["id"], parse_semver(m["version"])))
|
|
if args.json:
|
|
print(json.dumps(matches, indent=2, ensure_ascii=False))
|
|
return
|
|
if not matches:
|
|
print("no matches")
|
|
return
|
|
for item in matches:
|
|
print(f"{item['registry']}:{item['id']}@{item['version']} {item['name']}")
|
|
print(f" {item['summary']}")
|
|
|
|
|
|
def cmd_show(args: argparse.Namespace) -> None:
|
|
catalog = Path(args.catalog).expanduser()
|
|
package_dir, _ = resolve_installed(catalog, args.id, args.version)
|
|
manifest = validate_package(package_dir)
|
|
if args.json:
|
|
print(json.dumps(manifest, indent=2, ensure_ascii=False))
|
|
else:
|
|
print(yaml.safe_dump(manifest, sort_keys=False, allow_unicode=True).rstrip())
|
|
|
|
|
|
def coerce_value(raw: Any, spec: dict[str, Any]) -> Any:
|
|
kind = spec.get("type", "string")
|
|
if not isinstance(raw, str):
|
|
# Already typed — from a YAML fixture rather than the command line.
|
|
# Validate membership, but do not try to parse it.
|
|
if kind == "enum" and raw not in (spec.get("values") or []):
|
|
raise CannedPromptError(
|
|
f"invalid enum value {raw!r}; expected one of {spec.get('values') or []}"
|
|
)
|
|
return raw
|
|
if kind == "boolean":
|
|
lowered = raw.lower()
|
|
if lowered in {"true", "1", "yes", "on"}:
|
|
return True
|
|
if lowered in {"false", "0", "no", "off"}:
|
|
return False
|
|
raise CannedPromptError(f"cannot parse boolean value: {raw}")
|
|
if kind == "integer":
|
|
try:
|
|
return int(raw)
|
|
except ValueError as exc:
|
|
raise CannedPromptError(f"cannot parse integer value: {raw}") from exc
|
|
if kind == "number":
|
|
try:
|
|
return float(raw)
|
|
except ValueError as exc:
|
|
raise CannedPromptError(f"cannot parse numeric value: {raw}") from exc
|
|
if kind == "enum":
|
|
values = spec.get("values") or []
|
|
if raw not in values:
|
|
raise CannedPromptError(f"invalid enum value {raw!r}; expected one of {values}")
|
|
return raw
|
|
|
|
|
|
def supplied_values(pairs: list[str]) -> dict[str, str]:
|
|
values: dict[str, str] = {}
|
|
for pair in pairs:
|
|
if "=" not in pair:
|
|
raise CannedPromptError(f"--set expects name=value, got: {pair}")
|
|
name, value = pair.split("=", 1)
|
|
if not name:
|
|
raise CannedPromptError("--set name cannot be empty")
|
|
values[name] = value
|
|
return values
|
|
|
|
|
|
@dataclass
|
|
class Resolution:
|
|
"""The outcome of § 5.1 resolution, separate from rendering."""
|
|
|
|
values: dict[str, Any] = field(default_factory=dict)
|
|
origins: dict[str, str] = field(default_factory=dict)
|
|
underivable: list[str] = field(default_factory=list)
|
|
|
|
|
|
def resolve_inputs(
|
|
manifest: dict[str, Any],
|
|
raw_values: dict[str, str],
|
|
composer: "Composer | None" = None,
|
|
inherited: dict[str, Any] | None = None,
|
|
) -> Resolution:
|
|
"""Resolve inputs and parameters per § 5.1.
|
|
|
|
`composer`, when supplied, satisfies `include` defaults by rendering the
|
|
included package (§ 10.2). Inclusion is deterministic, so a tool that can
|
|
render can satisfy it; derivation is not, and this tool never calls a model,
|
|
so a derived default is satisfied only by its static fallback `value`.
|
|
|
|
`inherited` carries the including package's already-resolved values into an
|
|
included package (§ 10.2). Those values are used as-is rather than coerced,
|
|
and do not count as caller-supplied.
|
|
"""
|
|
result = Resolution()
|
|
inherited = inherited or {}
|
|
known: set[str] = set()
|
|
declared = prompt_dependencies(manifest)
|
|
|
|
def compose(name: str, item: dict[str, Any]) -> None:
|
|
default = item["default"]
|
|
reference = default["include"]
|
|
if composer is not None:
|
|
entry = declared.get(reference) or {}
|
|
result.values[name] = composer(reference, entry.get("version"), result.values)
|
|
result.origins[name] = f"included from {reference}"
|
|
return
|
|
if "value" in default:
|
|
result.values[name] = default["value"]
|
|
result.origins[name] = "fallback (not included)"
|
|
else:
|
|
result.origins[name] = "unresolved (included, no fallback)"
|
|
result.underivable.append(name)
|
|
|
|
# Parameters resolve first: a composed input inherits the including
|
|
# package's resolved values (§ 10.2), so those must already be settled.
|
|
parameters = manifest.get("parameters") or {}
|
|
for name, spec in parameters.items():
|
|
known.add(name)
|
|
if name in raw_values:
|
|
result.values[name] = coerce_value(raw_values[name], spec)
|
|
result.origins[name] = "supplied"
|
|
elif name in inherited:
|
|
result.values[name] = inherited[name]
|
|
result.origins[name] = "inherited"
|
|
elif "default" in spec:
|
|
result.values[name] = spec["default"]
|
|
result.origins[name] = "default"
|
|
else:
|
|
result.origins[name] = "unresolved (no default)"
|
|
|
|
inputs = manifest.get("inputs") or []
|
|
for item in inputs:
|
|
name = item["name"]
|
|
known.add(name)
|
|
if name in raw_values:
|
|
result.values[name] = raw_values[name]
|
|
result.origins[name] = "supplied"
|
|
continue
|
|
if name in inherited:
|
|
result.values[name] = inherited[name]
|
|
result.origins[name] = "inherited"
|
|
continue
|
|
if item.get("required", False):
|
|
raise CannedPromptError(f"missing required input: {name}")
|
|
|
|
kind = input_default_kind(item)
|
|
if kind == "static":
|
|
result.values[name] = item["default"]
|
|
result.origins[name] = "default"
|
|
elif kind == "included":
|
|
compose(name, item)
|
|
elif kind == "derived":
|
|
default = item["default"]
|
|
if "value" in default:
|
|
result.values[name] = default["value"]
|
|
result.origins[name] = "fallback (not derived)"
|
|
else:
|
|
result.origins[name] = "unresolved (derived, no fallback)"
|
|
result.underivable.append(name)
|
|
else:
|
|
result.origins[name] = "unresolved (optional, no default)"
|
|
|
|
# Report inputs before parameters regardless of resolution order.
|
|
ordered = {item["name"]: result.origins[item["name"]] for item in inputs}
|
|
ordered.update({name: result.origins[name] for name in parameters})
|
|
result.origins = ordered
|
|
|
|
unknown = sorted(set(raw_values) - known)
|
|
if unknown:
|
|
raise CannedPromptError("unknown values: " + ", ".join(unknown))
|
|
|
|
return result
|
|
|
|
|
|
def resolve_values(manifest: dict[str, Any], raw_values: dict[str, str]) -> dict[str, Any]:
|
|
return resolve_inputs(manifest, raw_values).values
|
|
|
|
|
|
class CatalogComposer:
|
|
"""Satisfies `include` defaults from the catalog (§ 10.2).
|
|
|
|
Inclusion is text composition: the included package is rendered and its text
|
|
is inlined. No model is involved, so this is deterministic.
|
|
"""
|
|
|
|
def __init__(self, catalog: Path) -> None:
|
|
self.catalog = catalog
|
|
self._stack: list[str] = []
|
|
|
|
def __call__(
|
|
self, reference: str, selector: str | None, outer_values: dict[str, Any]
|
|
) -> str:
|
|
if reference in self._stack:
|
|
raise CannedPromptError(
|
|
"inclusion cycle: " + " -> ".join([*self._stack, reference])
|
|
)
|
|
|
|
registry, package_id = parse_reference(reference)
|
|
names = (
|
|
[registry]
|
|
if registry is not None
|
|
else [
|
|
name
|
|
for name in catalog_registries(self.catalog)
|
|
if versions_at(id_path(self.catalog / name, package_id))
|
|
]
|
|
)
|
|
if len(names) > 1:
|
|
raise CannedPromptError(
|
|
f"included package {package_id} is present in more than one registry: "
|
|
+ ", ".join(f"{name}:{package_id}" for name in names)
|
|
+ " — qualify the dependency id"
|
|
)
|
|
if not names:
|
|
raise CannedPromptError(f"included package not found: {reference}")
|
|
|
|
base = id_path(self.catalog / names[0], package_id)
|
|
version = select_version(versions_at(base), selector)
|
|
if version is None:
|
|
raise CannedPromptError(
|
|
f"included package {reference}: no version satisfies {selector!r}"
|
|
)
|
|
|
|
package_dir = base / version
|
|
manifest = validate_package(package_dir)
|
|
template_path = safe_relative_file(package_dir, manifest["template"], "template")
|
|
|
|
self._stack.append(reference)
|
|
try:
|
|
inner = resolve_inputs(manifest, {}, composer=self, inherited=outer_values)
|
|
except CannedPromptError as exc:
|
|
if str(exc).startswith("inclusion cycle:"):
|
|
raise
|
|
raise CannedPromptError(f"while including {reference}: {exc}") from exc
|
|
finally:
|
|
self._stack.pop()
|
|
|
|
text = template_path.read_text(encoding="utf-8")
|
|
used = set(PLACEHOLDER_RE.findall(text))
|
|
missing = sorted(used - set(inner.values))
|
|
if missing:
|
|
raise CannedPromptError(
|
|
f"included package {reference} has unresolved inputs the including "
|
|
"package does not supply: " + ", ".join(missing)
|
|
)
|
|
return render_template(text, inner.values)
|
|
|
|
|
|
Composer = CatalogComposer
|
|
|
|
|
|
def render_template(template: str, values: dict[str, Any]) -> str:
|
|
def replace(match: re.Match[str]) -> str:
|
|
name = match.group(1)
|
|
if name not in values:
|
|
raise CannedPromptError(f"unresolved placeholder: {name}")
|
|
value = values[name]
|
|
if isinstance(value, bool):
|
|
return "true" if value else "false"
|
|
if isinstance(value, (dict, list)):
|
|
return json.dumps(value, ensure_ascii=False)
|
|
return str(value)
|
|
|
|
return PLACEHOLDER_RE.sub(replace, template)
|
|
|
|
|
|
def cmd_render(args: argparse.Namespace) -> None:
|
|
catalog = Path(args.catalog).expanduser()
|
|
package_dir, _ = resolve_installed(catalog, args.id, args.version)
|
|
manifest = validate_package(package_dir)
|
|
template_path = safe_relative_file(package_dir, manifest["template"], "template")
|
|
raw = supplied_values(args.set_values)
|
|
resolution = resolve_inputs(manifest, raw, composer=CatalogComposer(catalog))
|
|
|
|
template = template_path.read_text(encoding="utf-8")
|
|
used = set(PLACEHOLDER_RE.findall(template))
|
|
blocked = sorted(used.intersection(resolution.underivable))
|
|
if blocked:
|
|
raise CannedPromptError(
|
|
"cannot render: this tool does not derive values, and these inputs "
|
|
"declare a derived default with no static fallback: "
|
|
+ ", ".join(blocked)
|
|
+ " — supply them with --set, or add a fallback `value` to the default"
|
|
)
|
|
|
|
rendered = render_template(template, resolution.values)
|
|
print(rendered, end="" if rendered.endswith("\n") else "\n")
|
|
|
|
|
|
def cmd_resolve(args: argparse.Namespace) -> None:
|
|
catalog = Path(args.catalog).expanduser()
|
|
package_dir, _ = resolve_installed(catalog, args.id, args.version)
|
|
manifest = validate_package(package_dir)
|
|
raw = supplied_values(args.set_values)
|
|
resolution = resolve_inputs(manifest, raw, composer=CatalogComposer(catalog))
|
|
|
|
if args.json:
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"values": resolution.values,
|
|
"origins": resolution.origins,
|
|
"underivable": resolution.underivable,
|
|
},
|
|
indent=2,
|
|
ensure_ascii=False,
|
|
)
|
|
)
|
|
return
|
|
|
|
width = max((len(name) for name in resolution.origins), default=0)
|
|
origin_width = max((len(o) for o in resolution.origins.values()), default=0)
|
|
for name, origin in resolution.origins.items():
|
|
if name in resolution.values:
|
|
preview = str(resolution.values[name]).replace("\n", " ")
|
|
if len(preview) > 48:
|
|
preview = preview[:45] + "..."
|
|
print(f"{name:<{width}} {origin:<{origin_width}} {preview}")
|
|
else:
|
|
print(f"{name:<{width}} {origin}")
|
|
if resolution.underivable:
|
|
print()
|
|
plural = len(resolution.underivable) > 1
|
|
print(
|
|
"note: "
|
|
+ ", ".join(resolution.underivable)
|
|
+ (" declare" if plural else " declares")
|
|
+ " a derived default this tool cannot satisfy; rendering will fail "
|
|
+ ("if the template uses them" if plural else "if the template uses it")
|
|
)
|
|
|
|
|
|
def load_example_values(package_dir: Path, relative: str) -> dict[str, Any]:
|
|
path = safe_relative_file(package_dir, relative, "examples")
|
|
try:
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
except yaml.YAMLError as exc:
|
|
raise CannedPromptError(f"invalid YAML in {relative}: {exc}") from exc
|
|
if not isinstance(data, dict):
|
|
raise CannedPromptError(f"{relative}: an example must contain a mapping")
|
|
values = data.get("values") or {}
|
|
if not isinstance(values, dict):
|
|
raise CannedPromptError(f"{relative}: example values must be a mapping")
|
|
return values
|
|
|
|
|
|
def run_render_checks(
|
|
rendered: str, resolution: Resolution, checks: list[dict[str, Any]]
|
|
) -> list[tuple[bool, str]]:
|
|
"""Evaluate § 12.1 render checks. Deterministic: no model is involved."""
|
|
results: list[tuple[bool, str]] = []
|
|
for check in checks:
|
|
(kind,) = check
|
|
expected = check[kind]
|
|
if kind == "contains":
|
|
results.append((str(expected) in rendered, f'contains "{expected}"'))
|
|
elif kind == "not_contains":
|
|
results.append((str(expected) not in rendered, f'not_contains "{expected}"'))
|
|
elif kind == "resolves_all":
|
|
unresolved = sorted(set(resolution.origins) - set(resolution.values))
|
|
ok = (not unresolved) if expected else bool(unresolved)
|
|
detail = f" ({', '.join(unresolved)})" if unresolved else ""
|
|
results.append((ok, f"resolves_all{detail}"))
|
|
return results
|
|
|
|
|
|
def cmd_eval(args: argparse.Namespace) -> None:
|
|
catalog = Path(args.catalog).expanduser()
|
|
package_dir, registry = resolve_installed(catalog, args.id, args.version)
|
|
manifest = validate_package(package_dir)
|
|
template_path = safe_relative_file(package_dir, manifest["template"], "template")
|
|
template = template_path.read_text(encoding="utf-8")
|
|
|
|
relatives = manifest.get("evals") or []
|
|
if not relatives:
|
|
print(f"{registry}:{manifest['id']}@{manifest['version']} no evals declared")
|
|
return
|
|
|
|
print(f"{registry}:{manifest['id']}@{manifest['version']}")
|
|
failures = 0
|
|
for relative in relatives:
|
|
spec = read_eval(package_dir, relative)
|
|
if spec["schema"] != EVAL_RUBRIC_SCHEMA:
|
|
print(f" {relative} schema {spec['schema']} not recognized — skipped")
|
|
continue
|
|
|
|
print(f" {relative} ({spec['name']})")
|
|
checks = spec.get("render") or []
|
|
if checks:
|
|
values = (
|
|
load_example_values(package_dir, spec["example"])
|
|
if spec.get("example")
|
|
else {}
|
|
)
|
|
resolution = resolve_inputs(
|
|
manifest, values, composer=CatalogComposer(catalog)
|
|
)
|
|
missing = sorted(set(PLACEHOLDER_RE.findall(template)) - set(resolution.values))
|
|
rendered = "" if missing else render_template(template, resolution.values)
|
|
if missing:
|
|
print(f" render FAIL cannot render: unresolved {', '.join(missing)}")
|
|
failures += 1
|
|
for ok, label in run_render_checks(rendered, resolution, checks):
|
|
print(f" render {'PASS' if ok else 'FAIL'} {label}")
|
|
failures += 0 if ok else 1
|
|
|
|
criteria = (spec.get("output") or {}).get("criteria") or []
|
|
if criteria:
|
|
print(
|
|
f" output -- {len(criteria)} criteria declared "
|
|
"(not run: judging output needs a model)"
|
|
)
|
|
|
|
if failures:
|
|
raise CannedPromptError(f"{failures} render check(s) failed")
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(prog="canned-prompts", description=__doc__)
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
add = sub.add_parser("add", help="add a package directory to the local catalog")
|
|
add.add_argument("path")
|
|
add.add_argument("--catalog", default=str(default_catalog()))
|
|
add.add_argument(
|
|
"--as",
|
|
dest="as_registry",
|
|
default=LOCAL_REGISTRY,
|
|
metavar="REGISTRY",
|
|
help=f"catalog registry name to file the package under (default: {LOCAL_REGISTRY})",
|
|
)
|
|
add.set_defaults(func=cmd_add)
|
|
|
|
search = sub.add_parser("search", help="search the local catalog")
|
|
search.add_argument("query")
|
|
search.add_argument("--catalog", default=str(default_catalog()))
|
|
search.add_argument("--json", action="store_true")
|
|
search.set_defaults(func=cmd_search)
|
|
|
|
show = sub.add_parser("show", help="show an installed package manifest")
|
|
show.add_argument("id")
|
|
show.add_argument("--version")
|
|
show.add_argument("--catalog", default=str(default_catalog()))
|
|
show.add_argument("--json", action="store_true")
|
|
show.set_defaults(func=cmd_show)
|
|
|
|
render = sub.add_parser("render", help="render an installed prompt template")
|
|
render.add_argument("id")
|
|
render.add_argument("--version")
|
|
render.add_argument("--catalog", default=str(default_catalog()))
|
|
render.add_argument("--set", dest="set_values", action="append", default=[], metavar="NAME=VALUE")
|
|
render.set_defaults(func=cmd_render)
|
|
|
|
evaluate = sub.add_parser("eval", help="run an installed package's render checks")
|
|
evaluate.add_argument("id")
|
|
evaluate.add_argument("--version")
|
|
evaluate.add_argument("--catalog", default=str(default_catalog()))
|
|
evaluate.set_defaults(func=cmd_eval)
|
|
|
|
resolve = sub.add_parser("resolve", help="report how each input and parameter resolves")
|
|
resolve.add_argument("id")
|
|
resolve.add_argument("--version")
|
|
resolve.add_argument("--catalog", default=str(default_catalog()))
|
|
resolve.add_argument("--set", dest="set_values", action="append", default=[], metavar="NAME=VALUE")
|
|
resolve.add_argument("--json", action="store_true")
|
|
resolve.set_defaults(func=cmd_resolve)
|
|
|
|
install = sub.add_parser("install", help="install a package from a filesystem registry")
|
|
install.add_argument("id")
|
|
install.add_argument("--version")
|
|
install.add_argument("--catalog", default=str(default_catalog()))
|
|
install.add_argument("--registry", default=str(default_registry()))
|
|
install.set_defaults(func=cmd_install)
|
|
|
|
publish = sub.add_parser("publish", help="publish a package to a filesystem registry")
|
|
publish.add_argument("path")
|
|
publish.add_argument("--registry", default=str(default_registry()))
|
|
publish.set_defaults(func=cmd_publish)
|
|
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
args.func(args)
|
|
return 0
|
|
except CannedPromptError as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|