tegwick
b3280df742
CANP-WP-0004. The operator asked canned-prompts to build a database of versioned prompts recording where each came from and when — the first step toward a platform for collaborative prompting. The format had nowhere to put that. `provenance` records who wrote a prompt and where the idea came from; nothing recorded how a copy arrived in a particular store. Section 20.3 now specifies an `index.yaml` as store metadata rather than package data: how a copy arrived differs for every consumer, and recording an arrival must never rewrite the package that arrived. `add`, `install` and `publish` record registry, id, version, name, source, method, first-inclusion date, and the package's declared author, source and licence — the last three copied so a listing is readable without opening every package. A new `index` verb lists it. `included_at` is never overwritten; a re-run updates `last_seen_at`, because when a package first entered a collection is a fact about history rather than about the last command run. Tests 84 -> 90. Also records two findings from actually using the format: CANP-WP-0004-T03 — inclusion has no deduplication, so a diamond dependency renders shared content once per path. Found by composing a real collection. Not fixed here: deduplicating means choosing which occurrence survives and deciding what happens when two paths resolve different versions, which is resolver behaviour that section 10.4 deliberately avoids. Handed to CANP-WP-0005 with a leaning: document it, warn at validation time, do not deduplicate. The add_ons workaround in practice/pqrst-estimate is evidence about section 23's deferred "richer template syntax" — an optional appendix has to be an input with an empty default, because CPF has no conditionals. 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
1422 lines
55 KiB
Python
Executable file
1422 lines
55 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 datetime as dt
|
|
import shutil
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
import yaml
|
|
|
|
FORMAT = "canned-prompt/v0.2"
|
|
# v0.2 is additive, so a v0.1 package means exactly what it always meant (§ 3.2).
|
|
ACCEPTED_FORMATS = ("canned-prompt/v0.2", "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")
|
|
CAPABILITY_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
|
|
RESERVED_FILES = ("prompt.yaml", "README.md", "LICENSE")
|
|
RESERVED_DIRS = ("examples", "evals", "assets")
|
|
INDEX_FILE = "index.yaml"
|
|
INDEX_FORMAT = "canned-prompt-index/v0.1"
|
|
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*)"
|
|
r"(?:-([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?"
|
|
r"(?:\+[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?$"
|
|
)
|
|
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 validate_capabilities(names: Any, where: str) -> list[str]:
|
|
"""Validation rule 20 of § 18: shape is checked, membership is not (§ 10.2)."""
|
|
if names is None:
|
|
return []
|
|
if not isinstance(names, list):
|
|
raise CannedPromptError(f"{where} must be a list")
|
|
for name in names:
|
|
if not isinstance(name, str) or not CAPABILITY_RE.match(name):
|
|
raise CannedPromptError(f"{where}: {name!r} must be lowercase kebab-case")
|
|
return list(names)
|
|
|
|
|
|
def validate_context_dependencies(manifest: dict[str, Any]) -> list[dict[str, Any]]:
|
|
"""Validation rule 19 of § 18.
|
|
|
|
A context dependency names something CPF does not package and cannot
|
|
resolve, so `description` is the only thing a consumer has to go on.
|
|
"""
|
|
dependencies = manifest.get("dependencies") or {}
|
|
entries = dependencies.get("context") or []
|
|
if not isinstance(entries, list):
|
|
raise CannedPromptError("dependencies.context must be a list")
|
|
for entry in entries:
|
|
if not isinstance(entry, dict):
|
|
raise CannedPromptError("each context dependency must be a mapping")
|
|
name = entry.get("name")
|
|
if not isinstance(name, str) or not CAPABILITY_RE.match(name):
|
|
raise CannedPromptError(
|
|
f"context dependency name {name!r} must be lowercase kebab-case; "
|
|
"context entries use 'name', not 'id', because nothing can look them up"
|
|
)
|
|
if not isinstance(entry.get("description"), str) or not entry["description"].strip():
|
|
raise CannedPromptError(
|
|
f"context dependency {name!r} must declare a description; nothing "
|
|
"else can explain a dependency the format does not package"
|
|
)
|
|
requirement = entry.get("requirement", "required")
|
|
if requirement not in ("required", "optional"):
|
|
raise CannedPromptError(
|
|
f"context dependency {name!r}: requirement must be 'required' or 'optional'"
|
|
)
|
|
return list(entries)
|
|
|
|
|
|
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.3)"
|
|
)
|
|
|
|
|
|
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"] not in ACCEPTED_FORMATS:
|
|
raise CannedPromptError(
|
|
f"unsupported format: {manifest['format']!r}; expected one of "
|
|
+ ", ".join(ACCEPTED_FORMATS)
|
|
)
|
|
|
|
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)
|
|
|
|
validate_context_dependencies(manifest)
|
|
validate_capabilities(
|
|
(manifest.get("dependencies") or {}).get("capabilities"),
|
|
"dependencies.capabilities",
|
|
)
|
|
validate_capabilities(
|
|
(manifest.get("compatibility") or {}).get("capabilities"),
|
|
"compatibility.capabilities",
|
|
)
|
|
|
|
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.3)."""
|
|
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.3 selector.
|
|
|
|
Each dependency is selected independently against what is present. There is
|
|
no constraint solving across a dependency graph; that stays a non-goal.
|
|
|
|
Prereleases are excluded from `any`, `newest` and `>=` (§ 17.1); only an
|
|
exact pin selects one, so publishing a release candidate never changes what
|
|
existing consumers resolve to.
|
|
"""
|
|
if not available:
|
|
return None
|
|
if selector is not None and selector not in ("any", "newest") and not selector.startswith(">="):
|
|
return selector if selector in available else None
|
|
|
|
stable = [version for version in available if not is_prerelease(version)]
|
|
if not stable:
|
|
return None
|
|
if selector is None or selector in ("any", "newest"):
|
|
return stable[0]
|
|
floor = parse_semver(selector[2:])
|
|
for version in stable:
|
|
if parse_semver(version) >= floor:
|
|
return version
|
|
return None
|
|
|
|
|
|
def parse_semver(value: str) -> tuple[Any, ...]:
|
|
"""A precedence key following SemVer § 11 (spec § 17.1).
|
|
|
|
A prerelease ranks below its own release, so `1.0.0-rc1` < `1.0.0`. Build
|
|
metadata is ignored. Unparseable versions sort below everything.
|
|
"""
|
|
match = SEMVER_RE.match(value)
|
|
if not match:
|
|
return (-1, -1, -1, 0, (), value)
|
|
major, minor, patch = (int(match.group(i)) for i in (1, 2, 3))
|
|
prerelease = match.group(4)
|
|
if prerelease is None:
|
|
# 1 outranks the 0 given to a prerelease of the same numeric version.
|
|
return (major, minor, patch, 1, (), "")
|
|
identifiers: list[tuple[int, int, str]] = []
|
|
for part in prerelease.split("."):
|
|
# Numeric identifiers compare numerically and rank below alphanumeric ones.
|
|
identifiers.append((0, int(part), "") if part.isdigit() else (1, 0, part))
|
|
return (major, minor, patch, 0, tuple(identifiers), "")
|
|
|
|
|
|
def is_prerelease(value: str) -> bool:
|
|
match = SEMVER_RE.match(value)
|
|
return bool(match and match.group(4) is not None)
|
|
|
|
|
|
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:
|
|
available = versions_at(base)
|
|
if version:
|
|
path = base / version
|
|
if not (path / "prompt.yaml").is_file():
|
|
raise CannedPromptError(f"package not found: {label}@{version}")
|
|
return path
|
|
if not available:
|
|
raise CannedPromptError(f"package not found: {label}")
|
|
chosen = select_version(available, None)
|
|
if chosen is None:
|
|
raise CannedPromptError(
|
|
f"{label}: only prerelease versions are available "
|
|
f"({', '.join(available)}); name one exactly to use it"
|
|
)
|
|
return base / chosen
|
|
|
|
|
|
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 package_members(package_dir: Path, manifest: dict[str, Any]) -> set[str]:
|
|
"""Relative paths that belong to the package (§ 2).
|
|
|
|
Reserved paths plus whatever a manifest field references. Everything else in
|
|
the working directory — `.git`, a virtualenv, scratch files — is not part of
|
|
the package.
|
|
"""
|
|
members: set[str] = set()
|
|
for name in RESERVED_FILES:
|
|
if (package_dir / name).is_file():
|
|
members.add(name)
|
|
for name in RESERVED_DIRS:
|
|
directory = package_dir / name
|
|
if directory.is_dir():
|
|
for path in directory.rglob("*"):
|
|
if path.is_file():
|
|
members.add(path.relative_to(package_dir).as_posix())
|
|
|
|
referenced = [("template", manifest["template"])]
|
|
for key in ("examples", "evals"):
|
|
referenced.extend((key, relative) for relative in (manifest.get(key) or []))
|
|
for field, relative in referenced:
|
|
resolved = safe_relative_file(package_dir, relative, field)
|
|
members.add(resolved.relative_to(package_dir.resolve()).as_posix())
|
|
return members
|
|
|
|
|
|
def skipped_entries(package_dir: Path, members: set[str]) -> list[str]:
|
|
"""Top-level entries that will not be packaged.
|
|
|
|
Reported by name only; nothing walks into an ignored directory, so a stray
|
|
virtualenv costs nothing to skip.
|
|
"""
|
|
covered = {member.split("/", 1)[0] for member in members}
|
|
skipped = []
|
|
for entry in sorted(package_dir.iterdir()):
|
|
if entry.name in covered:
|
|
continue
|
|
skipped.append(entry.name + ("/" if entry.is_dir() else ""))
|
|
return skipped
|
|
|
|
|
|
def copy_package(src: Path, dst: Path, manifest: dict[str, Any], what: str) -> list[str]:
|
|
"""Copy only what belongs to the package (§ 2). Returns what was skipped."""
|
|
if dst.exists():
|
|
raise CannedPromptError(f"{what} already exists: {dst}")
|
|
members = package_members(src, manifest)
|
|
skipped = skipped_entries(src, members)
|
|
dst.mkdir(parents=True)
|
|
for relative in sorted(members):
|
|
target = dst / relative
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(src / relative, target)
|
|
return skipped
|
|
|
|
|
|
def read_index(store: Path) -> dict[str, Any]:
|
|
"""The catalog index: which packages entered this store, from where, when.
|
|
|
|
Registry-side metadata, deliberately outside the immutable package (§ 20).
|
|
A package records who wrote it; the index records how it got here.
|
|
"""
|
|
path = store / INDEX_FILE
|
|
if not path.is_file():
|
|
return {"format": INDEX_FORMAT, "entries": []}
|
|
try:
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
except yaml.YAMLError as exc:
|
|
raise CannedPromptError(f"invalid YAML in {path}: {exc}") from exc
|
|
if not isinstance(data, dict) or not isinstance(data.get("entries"), list):
|
|
raise CannedPromptError(f"{path}: index must be a mapping with an entries list")
|
|
if data.get("format") != INDEX_FORMAT:
|
|
raise CannedPromptError(f"unsupported index format: {data.get('format')!r}")
|
|
return data
|
|
|
|
|
|
def index_entry_key(entry: dict[str, Any]) -> tuple[str, str, str]:
|
|
return (entry.get("registry", ""), entry.get("id", ""), entry.get("version", ""))
|
|
|
|
|
|
def record_inclusion(
|
|
store: Path,
|
|
registry: str,
|
|
manifest: dict[str, Any],
|
|
source: str,
|
|
method: str,
|
|
) -> dict[str, Any]:
|
|
"""Record that a package version entered this store, and from where.
|
|
|
|
Re-recording the same registry/id/version keeps the original `included_at`:
|
|
the date a package first entered is a fact about history, not about the last
|
|
time someone re-ran a command.
|
|
"""
|
|
index = read_index(store)
|
|
entry = {
|
|
"registry": registry,
|
|
"id": manifest["id"],
|
|
"version": manifest["version"],
|
|
"name": manifest.get("name"),
|
|
"source": source,
|
|
"method": method,
|
|
"included_at": dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
}
|
|
provenance = manifest.get("provenance") or {}
|
|
if isinstance(provenance, dict):
|
|
for field in ("author", "source"):
|
|
if provenance.get(field):
|
|
entry[f"declared_{field}"] = str(provenance[field])
|
|
if manifest.get("license"):
|
|
entry["license"] = manifest["license"]
|
|
|
|
existing = {index_entry_key(e): e for e in index["entries"]}
|
|
key = index_entry_key(entry)
|
|
if key in existing:
|
|
entry["included_at"] = existing[key].get("included_at", entry["included_at"])
|
|
entry["last_seen_at"] = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
existing[key] = entry
|
|
|
|
index["entries"] = [existing[k] for k in sorted(existing)]
|
|
store.mkdir(parents=True, exist_ok=True)
|
|
(store / INDEX_FILE).write_text(
|
|
yaml.safe_dump(index, sort_keys=False, allow_unicode=True), encoding="utf-8"
|
|
)
|
|
return entry
|
|
|
|
|
|
def missing_dependencies(catalog: Path, manifest: dict[str, Any]) -> list[str]:
|
|
"""Declared prompt dependencies that this catalog cannot satisfy.
|
|
|
|
CPF leaves dependency *resolution* to the consumer (§ 10), so this tool does
|
|
not fetch anything. It does refuse to hand over a package that looks
|
|
installed but cannot render.
|
|
"""
|
|
missing: list[str] = []
|
|
for reference, entry in prompt_dependencies(manifest).items():
|
|
registry, package_id = parse_reference(reference)
|
|
names = [registry] if registry is not None else catalog_registries(catalog)
|
|
selector = entry.get("version")
|
|
if any(
|
|
select_version(versions_at(id_path(catalog / name, package_id)), selector)
|
|
for name in names
|
|
):
|
|
continue
|
|
missing.append(f"{reference}@{selector}" if selector else reference)
|
|
return missing
|
|
|
|
|
|
def report_missing_dependencies(catalog: Path, manifest: dict[str, Any]) -> None:
|
|
missing = missing_dependencies(catalog, manifest)
|
|
if missing:
|
|
print(
|
|
"declared dependencies not in this catalog: "
|
|
+ ", ".join(missing)
|
|
+ " — install them, or composition referencing them will not resolve",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
|
|
def report_skipped(skipped: list[str]) -> None:
|
|
if skipped:
|
|
print(
|
|
"not packaged (not a reserved path, not referenced by the manifest): "
|
|
+ ", ".join(skipped),
|
|
file=sys.stderr,
|
|
)
|
|
|
|
|
|
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"])
|
|
report_skipped(copy_package(src.resolve(), dst, manifest, "catalog package"))
|
|
record_inclusion(catalog, registry, manifest, str(src.resolve()), "add")
|
|
report_missing_dependencies(catalog, manifest)
|
|
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"])
|
|
report_skipped(copy_package(src.resolve(), dst, manifest, "published package"))
|
|
record_inclusion(registry, name or "", manifest, str(src.resolve()), "publish")
|
|
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_package(src, dst, manifest, "catalog package")
|
|
record_inclusion(catalog, name, manifest, str(registry.resolve()), "install")
|
|
report_missing_dependencies(catalog, manifest)
|
|
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.4). 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.4). 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.4), 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.4).
|
|
|
|
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}")
|
|
dependencies = manifest.get("dependencies") or {}
|
|
capabilities = dependencies.get("capabilities") or []
|
|
context = [
|
|
entry
|
|
for entry in (dependencies.get("context") or [])
|
|
if entry.get("requirement", "required") == "required"
|
|
]
|
|
if capabilities or context:
|
|
print()
|
|
print("requires (this tool cannot verify these):")
|
|
for name in capabilities:
|
|
print(f" capability {name}")
|
|
for entry in context:
|
|
print(f" context {entry['name']} — {entry['description'].strip()}")
|
|
|
|
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 cmd_index(args: argparse.Namespace) -> None:
|
|
store = Path(args.catalog).expanduser()
|
|
entries = read_index(store)["entries"]
|
|
if args.id:
|
|
_, package_id = parse_reference(args.id)
|
|
entries = [e for e in entries if e.get("id") == package_id]
|
|
if args.json:
|
|
print(json.dumps(entries, indent=2, ensure_ascii=False))
|
|
return
|
|
if not entries:
|
|
print("no packages recorded in this catalog")
|
|
return
|
|
width = max(len(f"{e.get('registry')}:{e.get('id')}@{e.get('version')}") for e in entries)
|
|
for entry in entries:
|
|
label = f"{entry.get('registry')}:{entry.get('id')}@{entry.get('version')}"
|
|
print(f"{label:<{width}} {entry.get('included_at')} {entry.get('method')}")
|
|
print(f"{'':<{width}} from {entry.get('source')}")
|
|
declared = entry.get("declared_source")
|
|
if declared:
|
|
print(f"{'':<{width}} declares source {declared}")
|
|
|
|
|
|
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)
|
|
|
|
index = sub.add_parser("index", help="show what entered this catalog, from where and when")
|
|
index.add_argument("id", nargs="?")
|
|
index.add_argument("--catalog", default=str(default_catalog()))
|
|
index.add_argument("--json", action="store_true")
|
|
index.set_defaults(func=cmd_index)
|
|
|
|
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())
|