CANP-WP-0002 T02: registry-scoped identity and namespace ownership

Section 17 already defined what a registry does when the same id@version is
republished; nothing defined what a *consumer* does. That is where namespace
conflict actually bites — in the catalog, after installing from two
registries.

Identity is now registry-scoped: an id names a package within a registry, the
way a path names a file within a repository. A local-first format with no
signing, no federation and no central authority cannot enforce global
uniqueness, and an unenforceable guarantee is worse than none — it invites
consumers to conflate two packages that merely share a name. A qualified
`<registry>:<id>` reference distinguishes them, and `:` is now barred from ids
so the separator stays available.

Installing the same id from two registries is therefore not a conflict. The
catalog is namespaced by registry and keeps both.

Ownership is registry policy, not package data. An optional `registry.yaml`
names a registry and records namespace claims. Those claims are explicitly
descriptive — a filesystem registry cannot authenticate a publisher, and
`publish` says so rather than implying it checked. Keeping the claim out of
packages leaves artifacts free of unverifiable assertions of authority, and
means package semantics do not change when a hosted registry appears later.

Spec: 3.2 (registry-scoped identity, qualified references), 17 (immutability
scoped to a registry), 20.1 and 20.2 (new), 18 (registry-manifest
validation), 21 (qualified references, reserved `local` name).

Reference CLI: parse_reference, check_registry_name, read_registry_manifest,
registry_name, namespace_policy; registry_package_path and
catalog_package_path split; resolve_installed reports ambiguity and returns
the source registry; iter_catalog; `add --as`; closed-namespace warning on
publish. Tests 11 -> 21.

The catalog layout changed. An existing catalog is detected and reported with
instructions rather than failing as "package not found". Signing, trust
scoring and federation remain non-goals and were not touched.

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

reference/canned_prompts.py

@@ -20,6 +20,9 @@ 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")
@@ -216,6 +219,72 @@ def validate_package(package_dir: Path) -> dict[str, Any]:
     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):
@@ -223,8 +292,14 @@ def id_path(store: Path, package_id: str) -> Path:
     return store.joinpath(*parts)
 
 
-def package_path(store: Path, package_id: str, version: str) -> Path:
-    return id_path(store, package_id) / version
+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]:
@@ -234,24 +309,80 @@ def parse_semver(value: str) -> tuple[int, int, int, str]:
     return (int(match.group(1)), int(match.group(2)), int(match.group(3)), value)
 
 
-def versions_for(store: Path, package_id: str) -> list[str]:
-    base = id_path(store, package_id)
+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 resolve_installed(store: Path, package_id: str, version: str | None) -> Path:
+def pick_version(base: Path, label: str, version: str | None) -> Path:
     if version:
-        path = package_path(store, package_id, version)
-        if not path.is_dir():
-            raise CannedPromptError(f"package not found: {package_id}@{version}")
+        path = base / version
+        if not (path / "prompt.yaml").is_file():
+            raise CannedPromptError(f"package not found: {label}@{version}")
         return path
-    versions = versions_for(store, package_id)
+    versions = versions_at(base)
     if not versions:
-        raise CannedPromptError(f"package not found: {package_id}")
-    return package_path(store, package_id, versions[0])
+        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:
@@ -261,51 +392,72 @@ def copy_immutable(src: Path, dst: Path, what: str) -> None:
     shutil.copytree(src, dst)
 
 
-def iter_packages(store: Path) -> Iterable[tuple[Path, dict[str, Any]]]:
-    if not store.exists():
-        return
-    for manifest_path in store.rglob("prompt.yaml"):
-        package_dir = manifest_path.parent
-        try:
-            manifest = validate_package(package_dir)
-        except CannedPromptError:
-            continue
-        yield package_dir, manifest
+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()
-    dst = package_path(catalog, manifest["id"], manifest["version"])
+    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 {manifest['id']}@{manifest['version']} -> {dst}")
+    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()
-    dst = package_path(registry, manifest["id"], manifest["version"])
+    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")
-    print(f"published {manifest['id']}@{manifest['version']} -> {dst}")
+    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()
-    src = resolve_installed(registry, args.id, args.version)
+    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 = package_path(catalog, manifest["id"], manifest["version"])
+    dst = catalog_package_path(catalog, name, manifest["id"], manifest["version"])
     copy_immutable(src, dst, "catalog package")
-    print(f"installed {manifest['id']}@{manifest['version']} -> {dst}")
+    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 _, manifest in iter_packages(catalog):
+    for registry, _, manifest in iter_catalog(catalog):
         haystack = " ".join(
             [
                 str(manifest.get("id", "")),
@@ -317,13 +469,14 @@ def cmd_search(args: argparse.Namespace) -> None:
         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["id"], parse_semver(m["version"])), reverse=False)
+    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
@@ -331,13 +484,13 @@ def cmd_search(args: argparse.Namespace) -> None:
         print("no matches")
         return
     for item in matches:
-        print(f"{item['id']}@{item['version']}  {item['name']}")
+        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)
+    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))
@@ -468,7 +621,7 @@ def render_template(template: str, values: dict[str, Any]) -> str:
 
 def cmd_render(args: argparse.Namespace) -> None:
     catalog = Path(args.catalog).expanduser()
-    package_dir = resolve_installed(catalog, args.id, args.version)
+    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)
@@ -491,7 +644,7 @@ def cmd_render(args: argparse.Namespace) -> None:
 
 def cmd_resolve(args: argparse.Namespace) -> None:
     catalog = Path(args.catalog).expanduser()
-    package_dir = resolve_installed(catalog, args.id, args.version)
+    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)
@@ -538,6 +691,13 @@ def build_parser() -> argparse.ArgumentParser:
     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")