diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 525682d..80383c6 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -4,7 +4,7 @@ on:
   workflow_dispatch:
     inputs:
       release_tag:
-        description: Numeric semantic release tag, for example v0.20.15
+        description: Numeric semantic release tag, for example v0.20.16
         required: true
         type: string
 
diff --git a/.github/workflows/security-release-gate.yml b/.github/workflows/security-release-gate.yml
index 01ab585..a372024 100644
--- a/.github/workflows/security-release-gate.yml
+++ b/.github/workflows/security-release-gate.yml
@@ -20,7 +20,7 @@ jobs:
     env:
       DAPPER_IMAGE: pasturestack/catalog-service-dapper:${{ github.sha }}
       TRIVY_IMAGE: aquasec/trivy:0.74.0@sha256:62b1e65e8869bc4b4c6aa4fa2b21595256c7c2f6018a9d9ad61caf87187c1969
-      VERSION_OVERRIDE: v0.20.15
+      VERSION_OVERRIDE: v0.20.16
 
     steps:
       - name: Check out candidate
@@ -74,7 +74,7 @@ jobs:
           }
 
           run_ci
-          artifact="dist/artifacts/catalog-service-0.20.15.tar.xz"
+          artifact="dist/artifacts/catalog-service-0.20.16.tar.xz"
           test -s "$artifact"
           cp "$artifact" /tmp/catalog-service-first.tar.xz
 
@@ -88,8 +88,8 @@ jobs:
           test "$(find evidence/product -maxdepth 1 -type f | wc -l)" -eq 2
           test -x evidence/product/catalog-service
           test -x evidence/product/catalog-service-sqlite
-          test "$(evidence/product/catalog-service --version)" = 'v0.20.15'
-          test "$(evidence/product/catalog-service-sqlite --version)" = 'v0.20.15'
+          test "$(evidence/product/catalog-service --version)" = 'v0.20.16'
+          test "$(evidence/product/catalog-service-sqlite --version)" = 'v0.20.16'
           sha256sum "$artifact" > evidence/catalog-service.tar.xz.sha256
           docker run --rm --entrypoint sh \
             --volume "$PWD:/work:ro" \
diff --git a/COMPATIBILITY.md b/COMPATIBILITY.md
index e3ff524..c62282b 100644
--- a/COMPATIBILITY.md
+++ b/COMPATIBILITY.md
@@ -2,6 +2,17 @@
 
 The migration preserves `/v1-catalog` routes, generated catalog schemas, database table names, template and version resource fields, legacy Compose metadata, version suffix ordering, and upgrade-link behavior.
 
+The v0.20.16 candidate adds only `GET /v1-catalog/ai/templates`, a private
+same-container metadata contract for an authenticated Engine gateway. It uses
+the effective project/global visibility of the original Catalog API, with
+template/version projection, bounded paging and stable logical identifiers.
+It does not replace public Catalog resource IDs or migrate/delete catalog
+data. Caller authorization and live Key platform-read evaluation remain in
+the Engine; forwarding a client role or IP header cannot authorize this route.
+Version bounds are metadata, not a compatibility or launch verdict. Existing
+files, questions, readme and launch APIs remain unchanged and outside this
+typed interface. Component publication and runtime acceptance remain separate.
+
 Legacy metadata parsing remains compatible without the archived general-purpose Compose parser. Top-level `catalog` and `.catalog` blocks continue to work, Compose v2 continues to read `services[".catalog"]`, and an explicit top-level block keeps precedence when both forms are present. Unrelated service definitions are not interpreted by Catalog Service.
 
 `platformVersion` is the preferred public query parameter. Historical version-query names, `rancher-compose.yml`, `minimumRancherVersion` and `maximumRancherVersion` schema fields, generated `RancherCompose` and `RancherClient` types, legacy repository fixtures, and vendored `github.com/rancher/*` paths remain only as catalog-data, HTTP, or inherited dependency contracts.
diff --git a/Dockerfile.dapper b/Dockerfile.dapper
index d701176..f9e9b37 100644
--- a/Dockerfile.dapper
+++ b/Dockerfile.dapper
@@ -6,6 +6,7 @@ ADD --checksum=sha256:5b6326f1932cfd7be57130561dc78192f09ed9f58bfbe6606ed0c50510
 ADD --checksum=sha256:2c418a61d7165d0cde01d9cb05ae025cc41fc75ef6d3f095996bd045ba0945dc https://security.ubuntu.com/ubuntu/pool/main/c/curl/curl_8.18.0-1ubuntu2.8_amd64.deb /tmp/curl.deb
 ADD --checksum=sha256:8db913e1fa5a694ca6ba7b828bd18c5204e6853ac07fe7ed9c6e1f94e3444d8e https://security.ubuntu.com/ubuntu/pool/main/c/curl/libcurl3t64-gnutls_8.18.0-1ubuntu2.8_amd64.deb /tmp/libcurl3t64-gnutls.deb
 ADD --checksum=sha256:7c0b37592f83a629cc22f0c98effa176da9edbd6d77b3f38184a9d0ddc5e0266 https://security.ubuntu.com/ubuntu/pool/main/c/curl/libcurl4t64_8.18.0-1ubuntu2.8_amd64.deb /tmp/libcurl4t64.deb
+ADD --checksum=sha256:bedeef1cf3268f90338c411cdb0a45b224d0660fd2240f310f380316c8244606 https://security.ubuntu.com/ubuntu/pool/main/b/brotli/libbrotli1_1.2.0-3ubuntu0.1_amd64.deb /tmp/libbrotli1.deb
 
 ARG DAPPER_HOST_ARCH=amd64
 ARG GO_VERSION=1.27.3
@@ -64,6 +65,12 @@ RUN set -eux; \
     for package in curl libcurl3t64-gnutls libcurl4t64; do \
         test "$(dpkg-query -W -f='${Version}' "${package}")" = "${UBUNTU_APT_CURL_VERSION}"; \
     done; \
+    test "$(dpkg-query -W -f='${db:Status-Status}' libbrotli1)" = installed; \
+    test "$(dpkg-deb -f /tmp/libbrotli1.deb Package)" = libbrotli1; \
+    test "$(dpkg-deb -f /tmp/libbrotli1.deb Version)" = "${UBUNTU_APT_BROTLI_VERSION}"; \
+    test "$(dpkg-deb -f /tmp/libbrotli1.deb Architecture)" = amd64; \
+    dpkg -i /tmp/libbrotli1.deb; \
+    test "$(dpkg-query -W -f='${Version}' libbrotli1)" = "${UBUNTU_APT_BROTLI_VERSION}"; \
     dpkg -i /opt/pasturestack-ca-certificates.deb && \
     update-ca-certificates --fresh && \
     test "$(dpkg-query -W -f='${Version}' ca-certificates)" = '20260816ubuntu0.26.04.1' && \
diff --git a/README.md b/README.md
index ebdf843..e04251c 100644
--- a/README.md
+++ b/README.md
@@ -17,6 +17,21 @@ publication and reproducible packaging checks passed; matching Server
 integration and acceptance remain separate and are not established by this
 component release.
 
+The next numeric candidate, **v0.20.16**, adds a private loopback-only Catalog
+metadata interface for typed Engine reads. Template and version IDs derive
+from logical catalog identity rather than disposable cache rows. Bounded
+responses exclude repository credentials, launch files, questions, defaults
+and readme; existing Catalog APIs keep their contracts. See
+[the metadata contract](docs/ai-metadata.md). This candidate is not published;
+its official Go 1.27.3/security gates and matching Engine/Web runtime
+integration are pending.
+
+The candidate builder applies Ubuntu's checksum-pinned
+`libbrotli1=1.2.0-3ubuntu0.1` same-series security update for
+[USN-8956-1](https://ubuntu.com/security/notices/USN-8956-1). Installed package
+identity/version/architecture are checked before the existing raw-to-VEX
+equality gate; no Brotli VEX exception is added.
+
 Earlier isolated upgrade checks confirmed removal of empty Git-metadata and
 README-only index entries while preserving valid catalog content, plus the
 remaining upgrade steps and cleanup of an existing catalog. That evidence
diff --git a/docs/ai-metadata.md b/docs/ai-metadata.md
new file mode 100644
index 0000000..b36ce77
--- /dev/null
+++ b/docs/ai-metadata.md
@@ -0,0 +1,71 @@
+# Bounded Catalog metadata
+
+The new `GET /v1-catalog/ai/templates` route is a private same-container
+interface for the Engine's typed Catalog metadata resource. It does not change
+the existing Catalog endpoints or their public response contracts.
+
+The handler accepts only a real loopback `RemoteAddr` and one effective
+`X-API-Project-Id` header. It does not trust forwarded IP or role headers. The
+Engine must authenticate the original Cookie or Basic caller, verify current
+project access, and evaluate the original Key's **platform Catalog metadata
+read** permission before calling this route. A project-only grant is not a
+platform grant. The gateway must use its fixed loopback destination without
+redirects, forwarded credentials, arbitrary URLs or new user JWTs. This route
+must not be published directly or exposed by a general AI proxy.
+
+Inputs are optional `query` (name substring, at most 256 UTF-16 units), `marker`
+(`cmc-<opaque cursor>`), or exclusive logical `templateId` for detail. Unknown or
+duplicate selectors are rejected. Project/global visibility and exact catalog
+parent ownership are applied in SQL before the fixed 51-row lookahead and
+50-row response. Public identities are `ctm-<SHA256 logical identity>` of the
+Catalog's global or verified wire-project namespace, Catalog name, canonical
+repository URL without userinfo/query/fragment, configured branch/pin, and
+template folder/base. They exclude cache row IDs, description, versions and
+local filesystem paths. Same-name resources in different scopes/repositories
+remain distinct. Credential rotation and cache rebuild do not change identity;
+source, branch/pin or template location changes do. No row-ID fallback exists.
+
+The short cursor contains a private cache locator, its logical identity and
+query/project binding. It is not a grant or persistent resource ID. The source
+revalidates the anchor under the same project/global SQL predicate before the
+next page; a changed/reused anchor returns `CatalogMetadataCursorStale`.
+Detail recomputes identities from at most 10000 authorized rows and matches the
+logical hash, without accepting a caller's DB ID. A larger scan fails closed.
+
+The response is `{resourceType: "catalogTemplate", data: [...], nextMarker}`.
+Only template name, description, catalog name, base, scope, configured default
+version, maintainer, license and `templateKey` are returned. The key is the
+indexer's logical single-component `FolderName`, distinct from display name;
+local cache `Path`, absolute paths, URLs, separators and control characters are
+rejected. This makes same-name/same-description templates distinguishable
+without exposing a local path. Descriptions preserve 1024 UTF-16
+units. Truncation is explicit and never splits a Unicode scalar. Catalog URLs,
+credentials, compose files, labels, questions/default answers, readme, icons
+and version links are excluded. Configured default version is metadata; this
+does not verify launch compatibility or implement the Catalog create/launch
+form.
+
+Version selection uses the same route with `templateId` and `versions=true`;
+an optional marker pages the list and an exclusive `versionId` requests detail.
+Name queries are not accepted in this mode. The response resource type is
+`catalogTemplateVersion`. Each `cvm-<SHA256>` identity binds the logical template,
+version string and nullable revision, never a database row. The parent is
+re-authorized on every request, and the version SQL joins and rechecks the
+current source identity to reject a source replacement between the two reads.
+The cursor is bound to the effective project and logical parent template.
+
+Only version, nullable numeric revision, minimum/maximum platform versions and
+upgrade-from metadata are projected. Revision values must fit the consumer's
+safe integer range. This is selection metadata, not a compatibility verdict,
+launch preview or permission to load files, questions, defaults or readme.
+
+Focused checks:
+
+```sh
+go test -count=1 ./internal/aimetadata
+go test -count=1 ./service -run '^TestAIMetadataRoute'
+```
+
+Engine/Web wiring is developed in separate B3 worktrees. Actual role/Key
+integration, release gates with the existing Go 1.27.3 toolchain and
+browser/weak-model acceptance remain separate work.
diff --git a/docs/releases/catalog-service-0.20.16.md b/docs/releases/catalog-service-0.20.16.md
new file mode 100644
index 0000000..d9c235d
--- /dev/null
+++ b/docs/releases/catalog-service-0.20.16.md
@@ -0,0 +1,36 @@
+# Catalog Service v0.20.16 candidate
+
+Status: unpublished candidate. The official Go 1.27.3 build, full tests/race,
+reproducible archives, source/binary/image security and required CodeQL checks
+must pass before publication. No release assets, producer or runtime success
+are claimed here.
+
+This release adds a fixed, private loopback metadata handler for the Engine's
+typed Catalog reader. It returns project-visible and global template metadata
+and parent-bound version selection metadata without source credentials, files,
+questions, default answers, readme or arbitrary links.
+
+Safe `templateKey` values preserve logical folder identity beside the display
+name, so two folders with identical names/descriptions remain selectable. They
+cannot contain an absolute/local path, URL or separator. The builder also
+applies Ubuntu's exact checksum-pinned libbrotli1 1.2.0-3ubuntu0.1 update for
+USN-8956-1, without adding a vulnerability exception or changing the frozen
+package baseline.
+
+Logical template IDs bind the catalog scope, name, sanitized repository origin,
+branch/pin and template path. Version IDs also bind version and nullable
+revision. Cache rebuilds preserve identity; a changed source or reused cache
+row cannot rebind a prior ID. Cursor anchors are re-read under the same
+authorized parent/project predicates before paging.
+
+Existing read, refresh and launch APIs, database tables, catalog content,
+operator origin policy and build gates retain their behavior. The new handler
+trusts only the actual loopback peer and an effective project supplied by the
+Engine, with no forwarded credentials or user JWT. Engine owner/Key decisions
+and Web opaque references are separate integration responsibilities. A
+project-only Key grant does not become a platform Catalog grant.
+
+Catalog create/edit/launch, compatibility resolution, browser/role/weak-model
+acceptance and Server deployment are outside this metadata component result.
+The previous immutable release remains v0.20.15; rollback uses that numeric
+component release and the existing operator configuration/data.
diff --git a/internal/aimetadata/handler.go b/internal/aimetadata/handler.go
new file mode 100644
index 0000000..4e168fa
--- /dev/null
+++ b/internal/aimetadata/handler.go
@@ -0,0 +1,237 @@
+// Package aimetadata owns the bounded Catalog metadata projection. The Engine
+// authorizes the real caller and Key; this private same-container route accepts
+// only its loopback transport and effective project, never forwarded client IPs.
+package aimetadata
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net"
+	"net/http"
+	"net/url"
+	"regexp"
+	"unicode/utf16"
+)
+
+const Path = "/v1-catalog/ai/templates"
+const PageSize = 50
+const ProjectHeader = "X-API-Project-Id"
+
+var projectID = regexp.MustCompile(`^1a[1-9][0-9]{0,18}$`)
+var metadataID = regexp.MustCompile(`^ctm-[0-9a-f]{64}$`)
+
+// Template is the complete output allowlist. Compose files, labels, questions,
+// default answers, URLs, credentials, icons and readme are never loaded/copied.
+type Template struct {
+	ID              string   `json:"id"`
+	Type            string   `json:"type"`
+	Name            string   `json:"name"`
+	Description     string   `json:"description"`
+	CatalogID       string   `json:"catalogId"`
+	TemplateBase    string   `json:"templateBase"`
+	TemplateKey     string   `json:"templateKey"`
+	CatalogScope    string   `json:"catalogScope"`
+	DefaultVersion  string   `json:"defaultVersion"`
+	Maintainer      string   `json:"maintainer"`
+	License         string   `json:"license"`
+	TruncatedFields []string `json:"truncatedFields,omitempty"`
+	rowID           int64
+}
+
+type Query struct {
+	Text          string
+	AfterID       int64
+	AfterIdentity string
+	ExactID       string
+	Versions      bool
+	VersionID     string
+}
+
+// Source applies the effective project/global visibility and keyset condition
+// before its fixed SQL limit. It must return strictly increasing persisted IDs.
+type Source interface {
+	Read(context.Context, string, Query, int) ([]Template, error)
+}
+
+type Handler struct{ source Source }
+
+func NewHandler(source Source) http.Handler { return &Handler{source: source} }
+
+type response struct {
+	ResourceType string     `json:"resourceType"`
+	Data         []Template `json:"data"`
+	NextMarker   *string    `json:"nextMarker"`
+}
+
+func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Cache-Control", "private, no-store")
+	w.Header().Set("X-Content-Type-Options", "nosniff")
+	if r.Method != http.MethodGet {
+		w.Header().Set("Allow", http.MethodGet)
+		fail(w, http.StatusMethodNotAllowed, "CatalogMetadataMethodDenied")
+		return
+	}
+	host, _, err := net.SplitHostPort(r.RemoteAddr)
+	if err != nil || net.ParseIP(host) == nil || !net.ParseIP(host).IsLoopback() {
+		fail(w, http.StatusForbidden, "CatalogMetadataCallerDenied")
+		return
+	}
+	project := r.Header.Get(ProjectHeader)
+	if !projectID.MatchString(project) || len(r.Header.Values(ProjectHeader)) != 1 {
+		fail(w, http.StatusForbidden, "CatalogMetadataProjectDenied")
+		return
+	}
+	query, ok := parseQuery(r, project)
+	if !ok {
+		fail(w, http.StatusUnprocessableEntity, "CatalogMetadataInputInvalid")
+		return
+	}
+	if h.source == nil {
+		fail(w, http.StatusServiceUnavailable, "CatalogMetadataUnavailable")
+		return
+	}
+	if query.Versions {
+		h.serveVersions(w, r, project, query)
+		return
+	}
+	rows, err := h.source.Read(r.Context(), project, query, PageSize+1)
+	if errors.Is(err, errCursorStale) {
+		fail(w, http.StatusConflict, "CatalogMetadataCursorStale")
+		return
+	}
+	if err != nil || len(rows) > PageSize+1 {
+		fail(w, http.StatusServiceUnavailable, "CatalogMetadataUnavailable")
+		return
+	}
+	prior := query.AfterID
+	seen := map[string]bool{}
+	for i := range rows {
+		if !metadataID.MatchString(rows[i].ID) || rows[i].rowID <= prior || seen[rows[i].ID] || query.ExactID != "" && rows[i].ID != query.ExactID ||
+			(rows[i].CatalogScope != "global" && rows[i].CatalogScope != "project") || !safeTemplateKey(rows[i].TemplateKey) {
+			fail(w, http.StatusServiceUnavailable, "CatalogMetadataUnavailable")
+			return
+		}
+		prior = rows[i].rowID
+		seen[rows[i].ID] = true
+		rows[i].Type = "catalogTemplate"
+		rows[i].TruncatedFields = nil
+		clipField(&rows[i].Name, "name", 512, &rows[i].TruncatedFields)
+		clipField(&rows[i].Description, "description", 1024, &rows[i].TruncatedFields)
+		clipField(&rows[i].CatalogID, "catalogId", 512, &rows[i].TruncatedFields)
+		clipField(&rows[i].TemplateBase, "templateBase", 512, &rows[i].TruncatedFields)
+		clipField(&rows[i].DefaultVersion, "defaultVersion", 512, &rows[i].TruncatedFields)
+		clipField(&rows[i].Maintainer, "maintainer", 512, &rows[i].TruncatedFields)
+		clipField(&rows[i].License, "license", 512, &rows[i].TruncatedFields)
+	}
+	if query.ExactID != "" && len(rows) != 1 {
+		fail(w, http.StatusNotFound, "CatalogMetadataNotFound")
+		return
+	}
+	result := response{ResourceType: "catalogTemplate", Data: []Template{}}
+	if len(rows) > PageSize {
+		rows = rows[:PageSize]
+		last := rows[len(rows)-1]
+		marker := cursor(project, query.Text, last.rowID, last.ID)
+		result.NextMarker = &marker
+	}
+	result.Data = append(result.Data, rows...)
+	writeJSON(w, http.StatusOK, result)
+}
+
+// FolderName is the logical single-component key set by the Git/Helm indexer,
+// unlike Template.Path, which is a local cache path and is never projected.
+func safeTemplateKey(key string) bool {
+	if key == "" || key == "." || key == ".." || len(utf16.Encode([]rune(key))) > 512 {
+		return false
+	}
+	for _, value := range key {
+		if value < 0x20 || value == 0x7f || value == '/' || value == '\\' || value == ':' {
+			return false
+		}
+	}
+	return true
+}
+
+func parseQuery(r *http.Request, project string) (Query, bool) {
+	if len(r.URL.RawQuery) > 4096 {
+		return Query{}, false
+	}
+	// URL.Query silently drops malformed escapes; ParseQuery must fail closed.
+	values, err := url.ParseQuery(r.URL.RawQuery)
+	if err != nil {
+		return Query{}, false
+	}
+	for key, values := range values {
+		if len(values) != 1 || (key != "query" && key != "marker" && key != "templateId" && key != "versions" && key != "versionId") {
+			return Query{}, false
+		}
+	}
+	query := Query{Text: values.Get("query")}
+	if len(utf16.Encode([]rune(query.Text))) > 256 {
+		return Query{}, false
+	}
+	if values.Has("templateId") {
+		if values.Has("query") || values.Has("marker") && values.Get("versions") != "true" {
+			return Query{}, false
+		}
+		id := values.Get("templateId")
+		if !metadataID.MatchString(id) {
+			return Query{}, false
+		}
+		query.ExactID = id
+	}
+	if values.Has("versions") || values.Has("versionId") {
+		if values.Get("versions") != "true" || query.ExactID == "" || values.Has("query") {
+			return Query{}, false
+		}
+		query.Versions = true
+		if values.Has("versionId") {
+			if !versionID.MatchString(values.Get("versionId")) || values.Has("marker") {
+				return Query{}, false
+			}
+			query.VersionID = values.Get("versionId")
+		}
+	}
+	if values.Has("marker") {
+		binding := query.Text
+		if query.Versions {
+			binding = query.ExactID
+		}
+		id, identity, valid := parseCursor(values.Get("marker"), project, binding)
+		if !valid {
+			return Query{}, false
+		}
+		query.AfterID = id
+		query.AfterIdentity = identity
+		if query.Versions {
+			query.AfterIdentity = "cvm-" + identity[4:]
+		}
+	}
+	return query, true
+}
+
+func clipField(value *string, field string, limit int, truncated *[]string) {
+	units := 0
+	for offset, char := range *value {
+		size := 1
+		if char > 0xffff {
+			size = 2
+		}
+		if units+size > limit {
+			*value = (*value)[:offset]
+			*truncated = append(*truncated, field)
+			return
+		}
+		units += size
+	}
+}
+
+func fail(w http.ResponseWriter, status int, code string) {
+	writeJSON(w, status, map[string]string{"code": code})
+}
+func writeJSON(w http.ResponseWriter, status int, value interface{}) {
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(value)
+}
diff --git a/internal/aimetadata/handler_test.go b/internal/aimetadata/handler_test.go
new file mode 100644
index 0000000..66ce75a
--- /dev/null
+++ b/internal/aimetadata/handler_test.go
@@ -0,0 +1,231 @@
+package aimetadata
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql/driver"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"unicode/utf16"
+)
+
+type sourceFunc func(context.Context, string, Query, int) ([]Template, error)
+
+func (f sourceFunc) Read(c context.Context, p string, q Query, n int) ([]Template, error) {
+	return f(c, p, q, n)
+}
+func request(query string) *http.Request {
+	r := httptest.NewRequest(http.MethodGet, Path+query, nil)
+	r.RemoteAddr = "127.0.0.1:12345"
+	r.Header.Set(ProjectHeader, "1a5")
+	return r
+}
+func serve(h http.Handler, r *http.Request) *httptest.ResponseRecorder {
+	w := httptest.NewRecorder()
+	h.ServeHTTP(w, r)
+	return w
+}
+func template(id int) Template {
+	digest := sha256.Sum256([]byte(fmt.Sprintf("fixture-%d", id)))
+	return Template{ID: "ctm-" + hex.EncodeToString(digest[:]), rowID: int64(id), Name: "same-name", TemplateKey: fmt.Sprintf("folder-%d", id), CatalogScope: "global"}
+}
+
+func TestMetadataTemplateKeyUsesOnlyFolderNameAndDistinguishesDuplicateDisplayNames(t *testing.T) {
+	s := sourceFixture(t)
+	fixtureRows = [][]driver.Value{sourceRow(51, "https://example.test/repo.git"), sourceRow(52, "https://example.test/repo.git")}
+	fixtureRows[0][12] = "wordpress-small"
+	fixtureRows[1][12] = "wordpress-large"
+	rows, err := s.Read(context.Background(), "1a5", Query{}, 51)
+	if err != nil || len(rows) != 2 || rows[0].Name != rows[1].Name || rows[0].Description != rows[1].Description || rows[0].ID == rows[1].ID {
+		t.Fatal("fixture must preserve duplicate display metadata and different identity")
+	}
+	h := NewHandler(sourceFunc(func(context.Context, string, Query, int) ([]Template, error) { return rows, nil }))
+	w := serve(h, request(""))
+	if w.Code != 200 || !strings.Contains(w.Body.String(), `"templateKey":"wordpress-small"`) || !strings.Contains(w.Body.String(), `"templateKey":"wordpress-large"`) || strings.Contains(w.Body.String(), "https://") {
+		t.Fatal("safe logical template keys unavailable")
+	}
+	for _, bad := range []string{"", ".", "..", "/tmp/cache/template", `C:\cache\template`, "https://example.test/repo", "folder/child", "key\nother"} {
+		rows[0].TemplateKey = bad
+		if serve(h, request("")).Code != 503 {
+			t.Fatalf("local path or non-key accepted: %q", bad)
+		}
+	}
+}
+
+func TestMetadataCallerUsesRealRemoteAddressNotForwardedOrRoleHeaders(t *testing.T) {
+	calls := 0
+	h := NewHandler(sourceFunc(func(_ context.Context, _ string, _ Query, _ int) ([]Template, error) { calls++; return nil, nil }))
+	for _, remote := range []string{"192.0.2.1:9000", "[2001:db8::1]:9000", "127.0.0.1", "malformed"} {
+		r := request("")
+		r.RemoteAddr = remote
+		r.Header.Set("X-Forwarded-For", "127.0.0.1")
+		r.Header.Set("X-Real-IP", "127.0.0.1")
+		r.Header.Set("X-API-Roles", "admin")
+		if w := serve(h, r); w.Code != 403 {
+			t.Fatalf("remote %q status=%d", remote, w.Code)
+		}
+	}
+	if calls != 0 {
+		t.Fatal("untrusted caller reached source")
+	}
+	for _, remote := range []string{"127.0.0.1:1", "[::1]:1"} {
+		r := request("")
+		r.RemoteAddr = remote
+		if w := serve(h, r); w.Code != 200 {
+			t.Fatalf("loopback status=%d", w.Code)
+		}
+	}
+	if calls != 2 {
+		t.Fatal("loopback reads did not reach source")
+	}
+}
+
+func TestMetadataRequiresEffectiveProjectHeaderAndRejectsClientProjectSelectors(t *testing.T) {
+	h := NewHandler(sourceFunc(func(_ context.Context, p string, _ Query, _ int) ([]Template, error) {
+		if p != "1a5" {
+			t.Fatal("project changed")
+		}
+		return nil, nil
+	}))
+	for _, project := range []string{"", "global", "1a0", "1a5/other", "1a5,1a6"} {
+		r := request("")
+		r.Header.Set(ProjectHeader, project)
+		if w := serve(h, r); w.Code != 403 {
+			t.Fatalf("project %q status=%d", project, w.Code)
+		}
+	}
+	r := request("")
+	r.Header.Add(ProjectHeader, "1a6")
+	if serve(h, r).Code != 403 {
+		t.Fatal("duplicate header accepted")
+	}
+	if serve(h, request("?projectId=1a6")).Code != 422 {
+		t.Fatal("query project accepted")
+	}
+}
+
+func TestMetadataInputIsClosedAndReadOnlyBeforeAnySourceRead(t *testing.T) {
+	calls := 0
+	h := NewHandler(sourceFunc(func(context.Context, string, Query, int) ([]Template, error) { calls++; return nil, nil }))
+	for _, query := range []string{"?url=https://other", "?action=refresh", "?query=a&query=b", "?marker=ctm-0", "?marker=../x", "?marker=ctm-9223372036854775808", "?templateId=ctm-1&query=", "?query=%ZZ", "?query=" + strings.Repeat("a", 257)} {
+		if w := serve(h, request(query)); w.Code != 422 {
+			t.Fatalf("query %q status=%d", query, w.Code)
+		}
+	}
+	r := request("")
+	r.Method = http.MethodPost
+	if w := serve(h, r); w.Code != 405 || w.Header().Get("Allow") != "GET" {
+		t.Fatal("write method accepted")
+	}
+	if calls != 0 {
+		t.Fatal("invalid input reached source")
+	}
+}
+
+func TestMetadataKeysetPagingPassesMarkerBeforeFixedLimitAndKeepsDuplicateNames(t *testing.T) {
+	calls := 0
+	h := NewHandler(sourceFunc(func(_ context.Context, p string, q Query, n int) ([]Template, error) {
+		calls++
+		if p != "1a5" || n != 51 {
+			t.Fatal("unbounded or foreign read")
+		}
+		if calls == 2 && q.AfterID != 50 {
+			t.Fatal("cursor reloaded first page")
+		}
+		rows := []Template{}
+		for i := int(q.AfterID) + 1; i <= 70 && len(rows) < n; i++ {
+			rows = append(rows, template(i))
+		}
+		return rows, nil
+	}))
+	first := serve(h, request(""))
+	var page response
+	if err := json.Unmarshal(first.Body.Bytes(), &page); err != nil {
+		t.Fatal(err)
+	}
+	if first.Code != 200 || len(page.Data) != 50 || page.NextMarker == nil {
+		t.Fatal("first page not bounded")
+	}
+	if after, id, valid := parseCursor(*page.NextMarker, "1a5", ""); !valid || after != 50 || id != template(50).ID {
+		t.Fatal("cursor did not bind private locator to logical identity")
+	}
+	second := serve(h, request("?marker="+*page.NextMarker))
+	if err := json.Unmarshal(second.Body.Bytes(), &page); err != nil {
+		t.Fatal(err)
+	}
+	if second.Code != 200 || len(page.Data) != 20 || page.Data[0].ID != template(51).ID || page.NextMarker != nil {
+		t.Fatal("later page skipped/reloaded")
+	}
+	if page.Data[0].Name != page.Data[1].Name || page.Data[0].ID == page.Data[1].ID {
+		t.Fatal("same names were collapsed")
+	}
+}
+
+func TestMetadataDetailUsesExactIdentityOrNotFoundWithoutLeakingSourceErrors(t *testing.T) {
+	h := NewHandler(sourceFunc(func(_ context.Context, _ string, q Query, _ int) ([]Template, error) {
+		if q.ExactID == template(12).ID {
+			return []Template{template(12)}, nil
+		}
+		if q.ExactID == template(13).ID {
+			return nil, nil
+		}
+		return nil, errors.New("PRIVATE_DATABASE_CREDENTIAL")
+	}))
+	if w := serve(h, request("?templateId="+template(12).ID)); w.Code != 200 || !strings.Contains(w.Body.String(), `"id":"`+template(12).ID+`"`) {
+		t.Fatal("wrong detail")
+	}
+	if serve(h, request("?templateId="+template(13).ID)).Code != 404 {
+		t.Fatal("missing detail was not404")
+	}
+	w := serve(h, request("?templateId="+template(14).ID))
+	if w.Code != 503 || strings.Contains(w.Body.String(), "PRIVATE_DATABASE_CREDENTIAL") {
+		t.Fatal("source error leaked")
+	}
+}
+
+func TestMetadataProjectionPreservesUTF16BoundaryAndMarksTruncation(t *testing.T) {
+	h := NewHandler(sourceFunc(func(context.Context, string, Query, int) ([]Template, error) {
+		r := template(1)
+		r.Name = strings.Repeat("n", 513)
+		r.Description = strings.Repeat("🌿", 513)
+		return []Template{r}, nil
+	}))
+	w := serve(h, request(""))
+	var page response
+	if err := json.Unmarshal(w.Body.Bytes(), &page); err != nil {
+		t.Fatal(err)
+	}
+	if w.Code != 200 || len(utf16.Encode([]rune(page.Data[0].Description))) != 1024 || len(page.Data[0].Name) != 512 {
+		t.Fatal("metadata lost description or exceeded bounds")
+	}
+	if strings.Join(page.Data[0].TruncatedFields, ",") != "name,description" {
+		t.Fatal("truncation was not explicit")
+	}
+	for _, forbidden := range []string{`"questions"`, `"files"`, `"readme"`, `"links"`, `"environment"`, `"path"`} {
+		if strings.Contains(w.Body.String(), forbidden) {
+			t.Fatalf("unsafe field %s", forbidden)
+		}
+	}
+	if w.Header().Get("Cache-Control") != "private, no-store" {
+		t.Fatal("private response is cacheable")
+	}
+}
+
+func TestMetadataRejectsNonprogressingOrOversizedSourceBeforeResponse(t *testing.T) {
+	for _, rows := range [][]Template{{template(1), template(1)}, {template(2), template(1)}, make([]Template, 52)} {
+		h := NewHandler(sourceFunc(func(context.Context, string, Query, int) ([]Template, error) { return rows, nil }))
+		if w := serve(h, request("")); w.Code != 503 {
+			t.Fatal("invalid source page accepted")
+		}
+	}
+	h := NewHandler(sourceFunc(func(context.Context, string, Query, int) ([]Template, error) { return []Template{template(1)}, nil }))
+	if serve(h, request("?marker="+cursor("1a5", "", 1, template(1).ID))).Code != 503 {
+		t.Fatal("old source row repeated")
+	}
+}
diff --git a/internal/aimetadata/identity.go b/internal/aimetadata/identity.go
new file mode 100644
index 0000000..6103040
--- /dev/null
+++ b/internal/aimetadata/identity.go
@@ -0,0 +1,75 @@
+package aimetadata
+
+import (
+	"crypto/sha256"
+	"crypto/subtle"
+	"encoding/base64"
+	"encoding/binary"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"net/url"
+	"strings"
+)
+
+// The cache primary key is deliberately absent from this logical identity.
+// Project scope uses the existing Engine-verified account namespace; global
+// Catalogs use the literal global namespace. Mutable metadata/versions are not
+// identity, while a source/branch/pin/template-location change is.
+type identity struct {
+	Scope, Catalog, Repo, Branch, Pin, Folder, Base string
+}
+
+func logicalID(value identity) (string, error) {
+	u, err := url.Parse(value.Repo)
+	if err != nil || u.Scheme == "" || u.Hostname() == "" || u.Opaque != "" {
+		return "", errors.New("invalid catalog source identity")
+	}
+	u.User = nil
+	u.RawQuery, u.Fragment, u.RawFragment = "", "", ""
+	u.ForceQuery = false
+	u.Scheme, u.Host = strings.ToLower(u.Scheme), strings.ToLower(u.Host)
+	if value.Scope != "global" && !projectID.MatchString(value.Scope) {
+		return "", errors.New("invalid catalog scope")
+	}
+	encoded, err := json.Marshal([]string{"catalog-template-v1", value.Scope, value.Catalog,
+		u.String(), value.Branch, strings.ToLower(value.Pin), value.Folder, value.Base})
+	if err != nil {
+		return "", err
+	}
+	digest := sha256.Sum256(encoded)
+	return "ctm-" + hex.EncodeToString(digest[:]), nil
+}
+
+// A cursor is a private locator hint, never an authorization target or grant.
+// Its context binding prevents accidental reuse across project/query. The
+// source must re-read its anchor and compare the logical identity before use.
+func cursor(project, query string, row int64, id string) string {
+	data := make([]byte, 57)
+	data[0] = 1
+	binary.BigEndian.PutUint64(data[1:9], uint64(row))
+	digest, _ := hex.DecodeString(strings.TrimPrefix(id, "ctm-"))
+	copy(data[9:41], digest)
+	bound := sha256.Sum256([]byte(project + "\x00" + query))
+	copy(data[41:], bound[:16])
+	return "cmc-" + base64.RawURLEncoding.EncodeToString(data)
+}
+
+func parseCursor(value, project, query string) (int64, string, bool) {
+	if !strings.HasPrefix(value, "cmc-") {
+		return 0, "", false
+	}
+	data, err := base64.RawURLEncoding.DecodeString(strings.TrimPrefix(value, "cmc-"))
+	if err != nil || len(data) != 57 || data[0] != 1 {
+		return 0, "", false
+	}
+	row := binary.BigEndian.Uint64(data[1:9])
+	if row == 0 || row > uint64(^uint64(0)>>1) {
+		return 0, "", false
+	}
+	bound := sha256.Sum256([]byte(project + "\x00" + query))
+	if subtle.ConstantTimeCompare(data[41:], bound[:16]) != 1 {
+		return 0, "", false
+	}
+	return int64(row), "ctm-" + hex.EncodeToString(data[9:41]), true
+}
diff --git a/internal/aimetadata/identity_test.go b/internal/aimetadata/identity_test.go
new file mode 100644
index 0000000..7211f5b
--- /dev/null
+++ b/internal/aimetadata/identity_test.go
@@ -0,0 +1,52 @@
+package aimetadata
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestMetadataLogicalIdentityBindsScopeSourceBranchPinAndTemplateLocation(t *testing.T) {
+	base := identity{Scope: "1a5", Catalog: "same-name", Repo: "https://example.test/team/repo.git", Branch: "main", Folder: "web", Base: "cattle"}
+	id, err := logicalID(base)
+	if err != nil || !metadataID.MatchString(id) {
+		t.Fatal("logical ID invalid")
+	}
+	for _, change := range []func(*identity){func(v *identity) { v.Scope = "1a6" }, func(v *identity) { v.Scope = "global" }, func(v *identity) { v.Repo = "https://example.test/other/repo.git" }, func(v *identity) { v.Branch = "other" }, func(v *identity) { v.Pin = strings.Repeat("a", 40) }, func(v *identity) { v.Folder = "other" }, func(v *identity) { v.Base = "k8s" }, func(v *identity) { v.Catalog = "other" }} {
+		changed := base
+		change(&changed)
+		next, err := logicalID(changed)
+		if err != nil || next == id {
+			t.Fatal("logical identity aliased another source/scope/path")
+		}
+	}
+}
+
+func TestMetadataCredentialRotationAndQuerySecretsDoNotEnterIdentity(t *testing.T) {
+	base := identity{Scope: "1a5", Catalog: "repo", Repo: "https://example.test/team/repo.git", Branch: "main", Folder: "web"}
+	id, _ := logicalID(base)
+	for _, repo := range []string{"https://PRIVATE_USER:PRIVATE_PASSWORD@example.test/team/repo.git?token=PRIVATE_TOKEN#PRIVATE_FRAGMENT", "https://ROTATED_USER:ROTATED_PASSWORD@example.test/team/repo.git?token=ROTATED_TOKEN"} {
+		changed := base
+		changed.Repo = repo
+		got, err := logicalID(changed)
+		if err != nil || got != id {
+			t.Fatal("credential rotation changed identity or secrets were hashed")
+		}
+	}
+}
+
+func TestMetadataCursorIsBoundToQueryAndProjectWithoutGrantingRowIdentity(t *testing.T) {
+	id := template(1).ID
+	value := cursor("1a5", "web", 123, id)
+	row, got, valid := parseCursor(value, "1a5", "web")
+	if !valid || row != 123 || got != id || len(value) > 128 {
+		t.Fatal("private cursor roundtrip invalid")
+	}
+	for _, pair := range [][2]string{{"1a6", "web"}, {"1a5", "other"}} {
+		if _, _, valid := parseCursor(value, pair[0], pair[1]); valid {
+			t.Fatal("foreign cursor context accepted")
+		}
+	}
+	if _, _, valid := parseCursor("ctm-123", "1a5", "web"); valid {
+		t.Fatal("public row-id cursor fallback accepted")
+	}
+}
diff --git a/internal/aimetadata/source.go b/internal/aimetadata/source.go
new file mode 100644
index 0000000..c2e6a57
--- /dev/null
+++ b/internal/aimetadata/source.go
@@ -0,0 +1,99 @@
+package aimetadata
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+)
+
+type SQLSource struct{ DB *sql.DB }
+
+var errCursorStale = errors.New("catalog metadata cursor changed")
+var errUnavailable = errors.New("catalog metadata source unavailable")
+
+const maxDetailRows = 10000
+
+// Source columns remain private. Exact detail recomputes logical identity from
+// authorized rows; a cache row ID is never accepted as an identity fallback.
+func (s SQLSource) Read(ctx context.Context, project string, query Query, limit int) ([]Template, error) {
+	if s.DB == nil || limit != PageSize+1 {
+		return nil, errUnavailable
+	}
+	if query.AfterID > 0 {
+		anchors, err := s.query(ctx, project, Query{}, query.AfterID, 1)
+		if err != nil {
+			return nil, err
+		}
+		if len(anchors) != 1 || anchors[0].ID != query.AfterIdentity {
+			return nil, errCursorStale
+		}
+	}
+	if query.ExactID != "" {
+		if !metadataID.MatchString(query.ExactID) {
+			return nil, errUnavailable
+		}
+		rows, err := s.query(ctx, project, Query{}, 0, maxDetailRows+1)
+		if err != nil {
+			return nil, err
+		}
+		if len(rows) > maxDetailRows {
+			return nil, errUnavailable
+		}
+		for _, row := range rows {
+			if row.ID == query.ExactID {
+				return []Template{row}, nil
+			}
+		}
+		return nil, nil
+	}
+	return s.query(ctx, project, query, 0, limit)
+}
+
+func (s SQLSource) query(ctx context.Context, project string, query Query, anchor int64, limit int) ([]Template, error) {
+	statement := `SELECT t.id, t.name, t.description, c.name, t.base, t.default_version, t.maintainer, t.license, t.environment_id, c.url, c.branch, c.pinned_commit, t.folder_name
+FROM catalog_template t JOIN catalog c ON c.id = t.catalog_id
+WHERE (t.environment_id = ? OR t.environment_id = 'global')
+AND c.environment_id = t.environment_id`
+	args := []interface{}{project}
+	if anchor != 0 {
+		statement += " AND t.id = ?"
+		args = append(args, anchor)
+	} else {
+		statement += " AND t.id > ?"
+		args = append(args, query.AfterID)
+		if query.Text != "" {
+			statement += " AND INSTR(LOWER(t.name), LOWER(?)) > 0"
+			args = append(args, query.Text)
+		}
+	}
+	statement += " ORDER BY t.id ASC LIMIT ?"
+	args = append(args, limit)
+	rows, err := s.DB.QueryContext(ctx, statement, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	result := []Template{}
+	for rows.Next() {
+		var id int64
+		var name, description, catalog, base, version, maintainer, license, environment, repo, branch, pin, folder sql.NullString
+		if err := rows.Scan(&id, &name, &description, &catalog, &base, &version, &maintainer, &license, &environment, &repo, &branch, &pin, &folder); err != nil {
+			return nil, err
+		}
+		if id <= 0 {
+			return nil, errUnavailable
+		}
+		scope := "project"
+		if environment.String == "global" {
+			scope = "global"
+		}
+		logical, err := logicalID(identity{Scope: environment.String, Catalog: catalog.String, Repo: repo.String,
+			Branch: branch.String, Pin: pin.String, Folder: folder.String, Base: base.String})
+		if err != nil {
+			return nil, errUnavailable
+		}
+		result = append(result, Template{ID: logical, rowID: id, Name: name.String, Description: description.String,
+			CatalogID: catalog.String, TemplateBase: base.String, TemplateKey: folder.String, CatalogScope: scope, DefaultVersion: version.String, Maintainer: maintainer.String, License: license.String})
+	}
+	return result, rows.Err()
+}
diff --git a/internal/aimetadata/source_test.go b/internal/aimetadata/source_test.go
new file mode 100644
index 0000000..7f73715
--- /dev/null
+++ b/internal/aimetadata/source_test.go
@@ -0,0 +1,177 @@
+package aimetadata
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"encoding/json"
+	"errors"
+	"io"
+	"os"
+	"strings"
+	"testing"
+)
+
+type captureDriver struct{}
+type captureConnection struct{}
+type captureRows struct {
+	rows   [][]driver.Value
+	offset int
+}
+
+var recordedQuery string
+var recordedArgs []driver.NamedValue
+var fixtureRows [][]driver.Value
+var fixtureVersionRows [][]driver.Value
+var queryCount int
+
+func (captureDriver) Open(string) (driver.Conn, error)        { return captureConnection{}, nil }
+func (captureConnection) Prepare(string) (driver.Stmt, error) { panic("not used") }
+func (captureConnection) Close() error                        { return nil }
+func (captureConnection) Begin() (driver.Tx, error)           { panic("not used") }
+func (captureConnection) QueryContext(_ context.Context, q string, args []driver.NamedValue) (driver.Rows, error) {
+	recordedQuery = q
+	recordedArgs = append([]driver.NamedValue(nil), args...)
+	queryCount++
+	rows := [][]driver.Value{}
+	if strings.Contains(q, "FROM catalog_version v") {
+		point := args[2].Value.(int64)
+		limit := args[len(args)-1].Value.(int64)
+		for _, row := range fixtureVersionRows {
+			id := row[0].(int64)
+			if strings.Contains(q, "v.id = ?") && id != point || strings.Contains(q, "v.id > ?") && id <= point {
+				continue
+			}
+			rows = append(rows, row)
+			if int64(len(rows)) == limit {
+				break
+			}
+		}
+		return &captureRows{rows: rows}, nil
+	}
+	point := args[1].Value.(int64)
+	limit := args[len(args)-1].Value.(int64)
+	for _, row := range fixtureRows {
+		id := row[0].(int64)
+		if strings.Contains(q, "t.id = ?") && id != point || strings.Contains(q, "t.id > ?") && id <= point {
+			continue
+		}
+		rows = append(rows, row)
+		if int64(len(rows)) == limit {
+			break
+		}
+	}
+	return &captureRows{rows: rows}, nil
+}
+func (*captureRows) Columns() []string {
+	return []string{"id", "name", "description", "catalog", "base", "version", "maintainer", "license", "environment", "url", "branch", "pin", "folder"}
+}
+func (*captureRows) Close() error { return nil }
+func (r *captureRows) Next(values []driver.Value) error {
+	if r.offset >= len(r.rows) {
+		return io.EOF
+	}
+	copy(values, r.rows[r.offset])
+	r.offset++
+	return nil
+}
+func init() { sql.Register("ai-metadata-capture", captureDriver{}) }
+func sourceFixture(t *testing.T) SQLSource {
+	t.Helper()
+	db, err := sql.Open("ai-metadata-capture", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { db.Close() })
+	queryCount = 0
+	fixtureVersionRows = nil
+	fixtureRows = [][]driver.Value{sourceRow(51, "https://example.test/repo.git")}
+	return SQLSource{DB: db}
+}
+func sourceRow(id int64, repo string) []driver.Value {
+	return []driver.Value{id, "name", "desc", "catalog", "", "1.0", "maintainer", "license", "1a5", repo, "main", "", "folder"}
+}
+
+func TestMetadataSQLFiltersProjectParentAndMarkerBeforeBoundedProjection(t *testing.T) {
+	s := sourceFixture(t)
+	query := "name%' OR 1=1 --"
+	rows, err := s.query(context.Background(), "1a5", Query{Text: query, AfterID: 50}, 0, 51)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, clause := range []string{"t.environment_id = ? OR t.environment_id = 'global'", "c.environment_id = t.environment_id", "t.id > ?", "INSTR(LOWER(t.name), LOWER(?)) > 0", "ORDER BY t.id ASC LIMIT ?"} {
+		if !strings.Contains(recordedQuery, clause) {
+			t.Fatalf("missing %s", clause)
+		}
+	}
+	if strings.Contains(recordedQuery, query) || strings.Contains(recordedQuery, "SELECT *") || strings.Contains(recordedQuery, "t.*") {
+		t.Fatal("unbounded fields or query interpolation")
+	}
+	if len(recordedArgs) != 4 || recordedArgs[0].Value != "1a5" || recordedArgs[1].Value != int64(50) || recordedArgs[2].Value != query || recordedArgs[3].Value != int64(51) {
+		t.Fatal("query arguments changed")
+	}
+	if len(rows) != 1 || !metadataID.MatchString(rows[0].ID) || rows[0].ID == "ctm-51" {
+		t.Fatal("logical identity missing")
+	}
+	collectionSQL := recordedQuery
+	collectionArgs := append([]driver.NamedValue(nil), recordedArgs...)
+	_, err = s.query(context.Background(), "1a5", Query{}, 51, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(recordedQuery, "t.id = ?") {
+		t.Fatal("private anchor did not use exact locator")
+	}
+	if file := os.Getenv("CATALOG_METADATA_SQL_PROOF_FILE"); file != "" {
+		proof := map[string]interface{}{"collectionSQL": collectionSQL, "collectionArgs": collectionArgs, "anchorSQL": recordedQuery, "anchorArgs": recordedArgs}
+		encoded, err := json.MarshalIndent(proof, "", "  ")
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := os.WriteFile(file, encoded, 0600); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestMetadataDetailRecomputesLogicalIdentityAcrossCacheRebuildAndSourceChange(t *testing.T) {
+	s := sourceFixture(t)
+	first, err := s.Read(context.Background(), "1a5", Query{}, 51)
+	if err != nil {
+		t.Fatal(err)
+	}
+	id := first[0].ID
+	fixtureRows = [][]driver.Value{sourceRow(901, "https://example.test/repo.git")}
+	fixtureRows[0][2] = "changed description"
+	fixtureRows[0][5] = "2.0"
+	rebuilt, err := s.Read(context.Background(), "1a5", Query{ExactID: id}, 51)
+	if err != nil || len(rebuilt) != 1 || rebuilt[0].ID != id || rebuilt[0].rowID != 901 {
+		t.Fatal("cache rebuild changed logical resource")
+	}
+	fixtureRows = [][]driver.Value{sourceRow(51, "https://example.test/other.git")}
+	rebound, err := s.Read(context.Background(), "1a5", Query{ExactID: id}, 51)
+	if err != nil || len(rebound) != 0 {
+		t.Fatal("old logical identity rebound to new repository")
+	}
+	if _, err := s.Read(context.Background(), "1a5", Query{ExactID: "ctm-51"}, 51); err == nil {
+		t.Fatal("rowid fallback accepted")
+	}
+	fixtureRows = [][]driver.Value{sourceRow(901, "https://PRIVATE_USER:PRIVATE_PASSWORD@example.test/repo.git?token=PRIVATE_TOKEN#PRIVATE_FRAGMENT")}
+	rotated, err := s.Read(context.Background(), "1a5", Query{ExactID: id}, 51)
+	encoded, _ := json.Marshal(rotated)
+	if err != nil || len(rotated) != 1 || rotated[0].ID != id || strings.Contains(string(encoded), "PRIVATE_") || strings.Contains(string(encoded), "rowID") {
+		t.Fatal("credential rotation changed public identity or private source/locator leaked")
+	}
+}
+
+func TestMetadataCursorChecksAnchorLogicalIdentityBeforeNextPage(t *testing.T) {
+	s := sourceFixture(t)
+	first, _ := s.Read(context.Background(), "1a5", Query{}, 51)
+	anchor := first[0].ID
+	fixtureRows = [][]driver.Value{sourceRow(51, "https://example.test/replaced.git")}
+	queryCount = 0
+	_, err := s.Read(context.Background(), "1a5", Query{AfterID: 51, AfterIdentity: anchor}, 51)
+	if !errors.Is(err, errCursorStale) || queryCount != 1 {
+		t.Fatal("stale cache anchor advanced to next page")
+	}
+}
diff --git a/internal/aimetadata/versions.go b/internal/aimetadata/versions.go
new file mode 100644
index 0000000..5f1daa2
--- /dev/null
+++ b/internal/aimetadata/versions.go
@@ -0,0 +1,188 @@
+package aimetadata
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"regexp"
+)
+
+var versionID = regexp.MustCompile(`^cvm-[0-9a-f]{64}$`)
+
+// Only version selection metadata. Files, questions, defaults and readme are
+// deliberately not queried, and version bounds are not a compatibility verdict.
+type Version struct {
+	ID                    string   `json:"id"`
+	Type                  string   `json:"type"`
+	TemplateID            string   `json:"templateId"`
+	Version               string   `json:"version"`
+	Revision              *int64   `json:"revision"`
+	MinimumRancherVersion string   `json:"minimumRancherVersion"`
+	MaximumRancherVersion string   `json:"maximumRancherVersion"`
+	UpgradeFrom           string   `json:"upgradeFrom"`
+	TruncatedFields       []string `json:"truncatedFields,omitempty"`
+	rowID                 int64
+}
+
+type versionSource interface {
+	ReadVersions(context.Context, string, Query, int) ([]Version, error)
+}
+
+func versionLogicalID(template, version string, revision *int64) string {
+	encoded, _ := json.Marshal([]interface{}{"catalog-version-v1", template, version, revision})
+	digest := sha256.Sum256(encoded)
+	return "cvm-" + hex.EncodeToString(digest[:])
+}
+
+func (h *Handler) serveVersions(w http.ResponseWriter, r *http.Request, project string, query Query) {
+	source, ok := h.source.(versionSource)
+	if !ok {
+		fail(w, 503, "CatalogMetadataUnavailable")
+		return
+	}
+	rows, err := source.ReadVersions(r.Context(), project, query, PageSize+1)
+	if errors.Is(err, errCursorStale) {
+		fail(w, 409, "CatalogMetadataCursorStale")
+		return
+	}
+	if errors.Is(err, sql.ErrNoRows) {
+		fail(w, 404, "CatalogMetadataNotFound")
+		return
+	}
+	if err != nil || len(rows) > PageSize+1 {
+		fail(w, 503, "CatalogMetadataUnavailable")
+		return
+	}
+	prior := query.AfterID
+	seen := map[string]bool{}
+	for i := range rows {
+		row := &rows[i]
+		if !versionID.MatchString(row.ID) || row.rowID <= prior || seen[row.ID] || row.TemplateID != query.ExactID ||
+			query.VersionID != "" && row.ID != query.VersionID {
+			fail(w, 503, "CatalogMetadataUnavailable")
+			return
+		}
+		prior = row.rowID
+		seen[row.ID] = true
+		row.Type = "catalogTemplateVersion"
+		row.TruncatedFields = nil
+		clipField(&row.Version, "version", 512, &row.TruncatedFields)
+		clipField(&row.MinimumRancherVersion, "minimumRancherVersion", 512, &row.TruncatedFields)
+		clipField(&row.MaximumRancherVersion, "maximumRancherVersion", 512, &row.TruncatedFields)
+		clipField(&row.UpgradeFrom, "upgradeFrom", 512, &row.TruncatedFields)
+	}
+	if query.VersionID != "" && len(rows) != 1 {
+		fail(w, 404, "CatalogMetadataNotFound")
+		return
+	}
+	var marker *string
+	if len(rows) > PageSize {
+		rows = rows[:PageSize]
+		last := rows[len(rows)-1]
+		value := cursor(project, query.ExactID, last.rowID, "ctm-"+last.ID[4:])
+		marker = &value
+	}
+	if rows == nil {
+		rows = []Version{}
+	}
+	writeJSON(w, 200, struct {
+		ResourceType string    `json:"resourceType"`
+		Data         []Version `json:"data"`
+		NextMarker   *string   `json:"nextMarker"`
+	}{"catalogTemplateVersion", rows, marker})
+}
+
+func (s SQLSource) ReadVersions(ctx context.Context, project string, query Query, limit int) ([]Version, error) {
+	if !query.Versions || !metadataID.MatchString(query.ExactID) || limit != PageSize+1 {
+		return nil, errUnavailable
+	}
+	// Resolve the parent from authorized rows each time. Private cache IDs are
+	// never caller input or a durable authorization target.
+	parents, err := s.Read(ctx, project, Query{ExactID: query.ExactID}, PageSize+1)
+	if err != nil {
+		return nil, err
+	}
+	if len(parents) != 1 {
+		return nil, sql.ErrNoRows
+	}
+	parent := parents[0]
+	if query.AfterID > 0 {
+		anchor, err := s.versions(ctx, project, parent, query.AfterID, 0, 1)
+		if err != nil {
+			return nil, err
+		}
+		if len(anchor) != 1 || anchor[0].ID != query.AfterIdentity {
+			return nil, errCursorStale
+		}
+	}
+	if query.VersionID != "" {
+		if !versionID.MatchString(query.VersionID) {
+			return nil, errUnavailable
+		}
+		rows, err := s.versions(ctx, project, parent, 0, 0, maxDetailRows+1)
+		if err != nil {
+			return nil, err
+		}
+		if len(rows) > maxDetailRows {
+			return nil, errUnavailable
+		}
+		for _, row := range rows {
+			if row.ID == query.VersionID {
+				return []Version{row}, nil
+			}
+		}
+		return nil, nil
+	}
+	return s.versions(ctx, project, parent, 0, query.AfterID, limit)
+}
+
+func (s SQLSource) versions(ctx context.Context, project string, parent Template, anchor, after int64, limit int) ([]Version, error) {
+	statement := `SELECT v.id, v.version, v.revision, v.minimum_rancher_version, v.maximum_rancher_version, v.upgrade_from,
+t.environment_id, c.name, c.url, c.branch, c.pinned_commit, t.folder_name, t.base
+FROM catalog_version v JOIN catalog_template t ON v.template_id = t.id JOIN catalog c ON c.id = t.catalog_id
+WHERE (t.environment_id = ? OR t.environment_id = 'global') AND c.environment_id = t.environment_id AND t.id = ?`
+	args := []interface{}{project, parent.rowID}
+	if anchor > 0 {
+		statement += " AND v.id = ?"
+		args = append(args, anchor)
+	} else {
+		statement += " AND v.id > ?"
+		args = append(args, after)
+	}
+	statement += " ORDER BY v.id ASC LIMIT ?"
+	args = append(args, limit)
+	rows, err := s.DB.QueryContext(ctx, statement, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	result := []Version{}
+	for rows.Next() {
+		var id int64
+		var version, minimum, maximum, upgrade, environment, catalog, repo, branch, pin, folder, base sql.NullString
+		var revision sql.NullInt64
+		if err := rows.Scan(&id, &version, &revision, &minimum, &maximum, &upgrade, &environment, &catalog, &repo, &branch, &pin, &folder, &base); err != nil {
+			return nil, err
+		}
+		if id <= 0 || revision.Valid && (revision.Int64 < 0 || revision.Int64 > 9007199254740991) {
+			return nil, errUnavailable
+		}
+		current, err := logicalID(identity{Scope: environment.String, Catalog: catalog.String, Repo: repo.String,
+			Branch: branch.String, Pin: pin.String, Folder: folder.String, Base: base.String})
+		if err != nil || current != parent.ID {
+			return nil, errCursorStale
+		}
+		var rev *int64
+		if revision.Valid {
+			value := revision.Int64
+			rev = &value
+		}
+		result = append(result, Version{ID: versionLogicalID(parent.ID, version.String, rev), TemplateID: parent.ID, Version: version.String,
+			Revision: rev, MinimumRancherVersion: minimum.String, MaximumRancherVersion: maximum.String, UpgradeFrom: upgrade.String, rowID: id})
+	}
+	return result, rows.Err()
+}
diff --git a/internal/aimetadata/versions_test.go b/internal/aimetadata/versions_test.go
new file mode 100644
index 0000000..dc75b1c
--- /dev/null
+++ b/internal/aimetadata/versions_test.go
@@ -0,0 +1,136 @@
+package aimetadata
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"os"
+	"strings"
+	"testing"
+)
+
+func versionRow(id int64, value string, revision int64) []driver.Value {
+	return []driver.Value{id, value, revision, "1.6.0", "1.9.0", "1.x", "1a5", "catalog", "https://example.test/repo.git", "main", "", "folder", ""}
+}
+
+func TestVersionMetadataRevalidatesParentAndUsesOnlyBoundedSafeSQL(t *testing.T) {
+	s := sourceFixture(t)
+	parents, _ := s.Read(context.Background(), "1a5", Query{}, 51)
+	parent := parents[0].ID
+	fixtureVersionRows = [][]driver.Value{versionRow(10, "1.0", 1)}
+	query := Query{ExactID: parent, Versions: true}
+	rows, err := s.ReadVersions(context.Background(), "1a5", query, 51)
+	if err != nil || len(rows) != 1 || !versionID.MatchString(rows[0].ID) || rows[0].TemplateID != parent {
+		t.Fatalf("version metadata unavailable: %v", err)
+	}
+	for _, clause := range []string{"v.template_id = t.id", "c.id = t.catalog_id", "t.environment_id = ? OR t.environment_id = 'global'", "c.environment_id = t.environment_id", "t.id = ?", "v.id > ?", "ORDER BY v.id ASC LIMIT ?"} {
+		if !strings.Contains(recordedQuery, clause) {
+			t.Fatalf("missing %s", clause)
+		}
+	}
+	for _, field := range []string{"v.*", "files", "questions", "readme", "default_answers"} {
+		if strings.Contains(recordedQuery, field) {
+			t.Fatalf("unsafe query: %s", field)
+		}
+	}
+	if recordedArgs[0].Value != "1a5" || recordedArgs[1].Value != int64(51) || recordedArgs[2].Value != int64(0) || recordedArgs[3].Value != int64(51) {
+		t.Fatal("unbound selector")
+	}
+	if file := os.Getenv("CATALOG_VERSION_SQL_PROOF_FILE"); file != "" {
+		encoded, err := json.MarshalIndent(map[string]interface{}{"sql": recordedQuery, "args": recordedArgs}, "", "  ")
+		if err != nil || os.WriteFile(file, encoded, 0600) != nil {
+			t.Fatal("cannot save SQL proof")
+		}
+	}
+	id := rows[0].ID
+	fixtureRows = [][]driver.Value{sourceRow(901, "https://example.test/repo.git")}
+	fixtureVersionRows[0][0] = int64(999)
+	query.VersionID = id
+	rebuilt, err := s.ReadVersions(context.Background(), "1a5", query, 51)
+	if err != nil || len(rebuilt) != 1 || rebuilt[0].ID != id {
+		t.Fatal("cache rebuild changed version identity")
+	}
+	fixtureVersionRows[0][8] = "https://example.test/replaced.git"
+	if _, err = s.ReadVersions(context.Background(), "1a5", query, 51); !errors.Is(err, errCursorStale) {
+		t.Fatal("source changed between parent and version query")
+	}
+	fixtureVersionRows[0] = versionRow(999, "1.0", 9007199254740992)
+	if _, err = s.ReadVersions(context.Background(), "1a5", query, 51); !errors.Is(err, errUnavailable) {
+		t.Fatal("unsafe numeric revision projected")
+	}
+	fixtureRows = nil
+	if _, err = s.ReadVersions(context.Background(), "1a5", query, 51); !errors.Is(err, sql.ErrNoRows) {
+		t.Fatal("foreign or removed parent accepted")
+	}
+}
+
+type versionsFixture struct {
+	rows    []Version
+	queries []Query
+}
+
+func (*versionsFixture) Read(context.Context, string, Query, int) ([]Template, error) {
+	return nil, nil
+}
+func (s *versionsFixture) ReadVersions(_ context.Context, _ string, q Query, _ int) ([]Version, error) {
+	s.queries = append(s.queries, q)
+	return s.rows, nil
+}
+
+func TestVersionsRouteClosedInputsProjectionPagingAndTemplateBinding(t *testing.T) {
+	parent := template(1).ID
+	source := &versionsFixture{}
+	for i := 1; i <= 51; i++ {
+		revision := int64(i)
+		source.rows = append(source.rows, Version{ID: versionLogicalID(parent, fmt.Sprint(i), &revision), TemplateID: parent, Version: fmt.Sprint(i), Revision: &revision, rowID: int64(i)})
+	}
+	h := NewHandler(source)
+	for _, suffix := range []string{"?versions=true", "?templateId=" + parent + "&versions=false", "?templateId=" + parent + "&versionId=cvm-1", "?templateId=" + parent + "&versions=true&query=x", "?templateId=" + parent + "&versions=true&versionId=cvm-" + strings.Repeat("a", 64) + "&marker=x"} {
+		if w := serve(h, request(suffix)); w.Code != 422 {
+			t.Fatalf("input accepted %s", suffix)
+		}
+	}
+	w := serve(h, request("?templateId="+parent+"&versions=true"))
+	if w.Code != http.StatusOK {
+		t.Fatal(w.Body.String())
+	}
+	var body struct {
+		ResourceType string    `json:"resourceType"`
+		Data         []Version `json:"data"`
+		NextMarker   string    `json:"nextMarker"`
+	}
+	if json.Unmarshal(w.Body.Bytes(), &body) != nil || body.ResourceType != "catalogTemplateVersion" || len(body.Data) != 50 || body.NextMarker == "" {
+		t.Fatal("version pagination invalid")
+	}
+	if strings.Contains(w.Body.String(), "rowID") || strings.Contains(w.Body.String(), "files") || strings.Contains(w.Body.String(), "questions") {
+		t.Fatal("private projection leaked")
+	}
+	other := template(2).ID
+	if serve(h, request("?templateId="+other+"&versions=true&marker="+body.NextMarker)).Code != 422 {
+		t.Fatal("cross-template cursor accepted")
+	}
+	source.rows = []Version{{ID: versionLogicalID(other, "1", nil), TemplateID: other, Version: "1", rowID: 1}}
+	if serve(h, request("?templateId="+parent+"&versions=true")).Code != 503 {
+		t.Fatal("foreign version projection accepted")
+	}
+}
+
+func TestVersionCursorRechecksExactVersionAndNeverRebindsReusedCacheRow(t *testing.T) {
+	s := sourceFixture(t)
+	parents, _ := s.Read(context.Background(), "1a5", Query{}, 51)
+	parent := parents[0].ID
+	fixtureVersionRows = [][]driver.Value{versionRow(10, "1.0", 1)}
+	rows, _ := s.ReadVersions(context.Background(), "1a5", Query{ExactID: parent, Versions: true}, 51)
+	query := Query{ExactID: parent, Versions: true, AfterID: 10, AfterIdentity: rows[0].ID}
+	fixtureVersionRows[0][1] = "2.0"
+	if _, err := s.ReadVersions(context.Background(), "1a5", query, 51); !errors.Is(err, errCursorStale) {
+		t.Fatal("reused version row advanced")
+	}
+	if versionLogicalID(parent, "1.0", nil) == versionLogicalID(parent, "1.0", new(int64)) {
+		t.Fatal("nil and zero revision conflated")
+	}
+}
diff --git a/service/ai_metadata_route_test.go b/service/ai_metadata_route_test.go
new file mode 100644
index 0000000..c34c9e5
--- /dev/null
+++ b/service/ai_metadata_route_test.go
@@ -0,0 +1,36 @@
+package service
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/PastureStack/catalog-service/internal/aimetadata"
+	"github.com/gorilla/mux"
+)
+
+func TestAIMetadataRouteIsPrivateAndLeavesLegacyReadRoutesIntact(t *testing.T) {
+	previousManager, previousDB := m, db
+	defer func() { m, db = previousManager, previousDB }()
+	router := NewRouter(nil, nil)
+	for path, expected := range map[string]string{
+		aimetadata.Path:                        "GetAIMetadata",
+		"/v1-catalog/templates":                "GetTemplates",
+		"/v1-catalog/templates/catalog:folder": "GetTemplate",
+		"/v1-catalog/catalogs":                 "GetCatalogs",
+	} {
+		request := httptest.NewRequest(http.MethodGet, path, nil)
+		var match mux.RouteMatch
+		if !router.Match(request, &match) || match.Route.GetName() != expected {
+			t.Fatalf("route changed: %s", path)
+		}
+	}
+	request := httptest.NewRequest(http.MethodGet, aimetadata.Path, nil)
+	request.Header.Set(aimetadata.ProjectHeader, "1a5")
+	request.Header.Set("X-Forwarded-For", "127.0.0.1")
+	response := httptest.NewRecorder()
+	router.ServeHTTP(response, request)
+	if response.Code != http.StatusForbidden {
+		t.Fatalf("external metadata caller status=%d", response.Code)
+	}
+}
diff --git a/service/routes.go b/service/routes.go
index 1dbb60a..2c451bc 100644
--- a/service/routes.go
+++ b/service/routes.go
@@ -1,6 +1,7 @@
 package service
 
 import (
+	"github.com/PastureStack/catalog-service/internal/aimetadata"
 	"net/http"
 
 	api "github.com/PastureStack/catalog-service/internal/catalogapi"
@@ -82,6 +83,13 @@ func NewRouter(manager *manager.Manager, gormDb *gorm.DB) *mux.Router {
 
 	// API framework routes
 	router := mux.NewRouter().StrictSlash(true)
+	var metadataSource aimetadata.Source
+	if gormDb != nil {
+		if sqlDB, err := gormDb.DB(); err == nil {
+			metadataSource = aimetadata.SQLSource{DB: sqlDB}
+		}
+	}
+	router.Methods("GET").Path(aimetadata.Path).Name("GetAIMetadata").Handler(aimetadata.NewHandler(metadataSource))
 
 	router.Methods("GET").Path("/").Handler(api.VersionsHandler(schemas, "v1-catalog"))
 	router.Methods("GET").Path("/v1-catalog/schemas").Handler(api.SchemasHandler(schemas))
diff --git a/ubuntu-apt.lock b/ubuntu-apt.lock
index 24dfddf..50e940b 100644
--- a/ubuntu-apt.lock
+++ b/ubuntu-apt.lock
@@ -10,6 +10,7 @@ UBUNTU_APT_CA_CERTIFICATES_BOOTSTRAP_VERSION='20260601~26.04.1'
 UBUNTU_APT_CA_CERTIFICATES_VERSION='20260816ubuntu0.26.04.1'
 UBUNTU_APT_CURL_BOOTSTRAP_VERSION='8.18.0-1ubuntu2.7'
 UBUNTU_APT_CURL_VERSION='8.18.0-1ubuntu2.8'
+UBUNTU_APT_BROTLI_VERSION='1.2.0-3ubuntu0.1'
 UBUNTU_APT_FILE_VERSION='1:5.46-5build2'
 UBUNTU_APT_GCC_VERSION='4:15.2.0-5ubuntu1'
 UBUNTU_APT_GIT_VERSION='1:2.53.0-1ubuntu1'