#!/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._-]*$") 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 an input's default as 'none', 'static', or 'derived' (§ 6.1).""" if "default" not in item: return "none" default = item["default"] if isinstance(default, dict) and "derive" in default: return "derived" return "static" def prompt_dependency_ids(manifest: dict[str, Any]) -> set[str]: 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") ids: set[str] = set() for entry in prompts: if isinstance(entry, str): ids.add(entry) elif isinstance(entry, dict) and isinstance(entry.get("id"), str): ids.add(entry["id"]) else: raise CannedPromptError( "each prompt dependency must be an id string or a mapping with a string id" ) return ids def validate_input_default(item: dict[str, Any], dependency_ids: set[str]) -> None: """Validation rules 11-13 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 derive = item["default"]["derive"] if isinstance(derive, str): if not derive.strip(): raise CannedPromptError(f"input {name!r}: derive reference is empty") if derive not in dependency_ids: raise CannedPromptError( f"input {name!r} derives from {derive!r}, which is not declared " "in dependencies.prompts" ) 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) dependency_ids = prompt_dependency_ids(manifest) for item in manifest.get("inputs") or []: validate_input_default(item, dependency_ids) 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 `:` 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 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 `:` 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: str, spec: dict[str, Any]) -> Any: kind = spec.get("type", "string") 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]) -> Resolution: """Resolve inputs and parameters per § 5.1. This tool never calls a model, so a derived default is satisfied only by its static fallback `value`. An input with a derived default and no fallback is reported as underivable rather than silently rendered as empty. """ result = Resolution() known: set[str] = set() 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 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 == "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)" 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 "default" in spec: result.values[name] = spec["default"] result.origins[name] = "default" else: result.origins[name] = "unresolved (no default)" 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 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) 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) 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) 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:<32} {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 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) 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())