Ben Stull
e336e31812
Collections declare a `fields:` schema in `.collection.yaml`; entries carry typed metadata (enum/tags/text). New central `metadata_schema` module parses the schema leniently (INV-3) and validates entry values — advisory at read (corpus mirror flags violations as `metadata_malformed` without hard-failing), the enforcement point for the write boundary (edit endpoints land SLICE-4/5). - app/metadata_schema.py: parse_fields (lenient/normalizing) + validate - registry.parse_collection_manifest reads `fields:` into collection config - collections.get_collection unpacks `fields`; served by the collection API - cache._refresh_collection_corpus validates each entry advisory-only Non-breaking, opt-in: no `fields:` → unchanged (INV-5); default `document` collection declares none (N=1 unchanged). No DB migration — schema rides in collections.config_json. SLICE-2 of docs/design/2026-06-06-configurable-collection-metadata.md §7.2. Backend suite green (601 passed). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
359 lines
16 KiB
Python
359 lines
16 KiB
Python
"""§22.2 project registry mirror — the config-side analogue of
|
|
cache.refresh_meta_repo.
|
|
|
|
A deployment declares its projects in a `projects.yaml` at the root of
|
|
REGISTRY_REPO. This module mirrors that file into the `projects` cache table
|
|
and the `deployment` singleton. Per §22.2, `projects` rows flow from the
|
|
registry only — never from user actions. The mirror runs on the registry-repo
|
|
webhook and on every reconciler sweep (Option A wiring, Task 5).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import logging
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
|
|
import yaml
|
|
|
|
from . import db
|
|
from . import metadata_schema
|
|
from .config import Config
|
|
from .gitea import Gitea
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
VALID_TYPES = {"document", "specification", "bdd"}
|
|
VALID_VISIBILITY = {"gated", "public", "unlisted"}
|
|
VALID_INITIAL_STATE = {"super-draft", "active"}
|
|
# §22.4b: per-type default landing state.
|
|
_TYPE_DEFAULT_INITIAL_STATE = {
|
|
"document": "super-draft",
|
|
"specification": "super-draft",
|
|
"bdd": "active",
|
|
}
|
|
_SLUG_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$")
|
|
|
|
|
|
class RegistryError(Exception):
|
|
"""A registry document that fails validation, or a missing registry file.
|
|
|
|
Raised by parse_registry/refresh_registry. The caller decides severity:
|
|
fatal at startup (no last-good to serve), tolerated on a running deployment
|
|
(keep the last-good projects rows). See Task 5 wiring.
|
|
"""
|
|
|
|
|
|
@dataclass
|
|
class ProjectEntry:
|
|
id: str
|
|
name: str
|
|
type: str
|
|
content_repo: str
|
|
visibility: str
|
|
initial_state: str
|
|
config: dict = field(default_factory=dict) # theme, enabled_models
|
|
|
|
|
|
@dataclass
|
|
class CollectionEntry:
|
|
"""A named collection declared by a `.collection.yaml` manifest inside a
|
|
project's content repo (S2). `visibility=None` means "inherit the project's
|
|
visibility"."""
|
|
type: str
|
|
visibility: str | None
|
|
initial_state: str
|
|
name: str | None
|
|
config: dict = field(default_factory=dict) # §22.12 enabled_models
|
|
|
|
|
|
@dataclass
|
|
class RegistryDoc:
|
|
deployment_name: str
|
|
deployment_tagline: str
|
|
projects: list[ProjectEntry]
|
|
|
|
|
|
def parse_registry(text: str) -> RegistryDoc:
|
|
"""Parse + validate projects.yaml. Pure (no I/O). Raises RegistryError."""
|
|
raw = yaml.safe_load(text) or {}
|
|
dep = raw.get("deployment") or {}
|
|
projects_raw = raw.get("projects") or []
|
|
if not isinstance(projects_raw, list) or not projects_raw:
|
|
raise RegistryError("registry must declare at least one project")
|
|
seen: set[str] = set()
|
|
entries: list[ProjectEntry] = []
|
|
for p in projects_raw:
|
|
if not isinstance(p, dict):
|
|
raise RegistryError(f"each project entry must be a mapping, got {type(p).__name__}")
|
|
pid = str(p.get("id") or "").strip()
|
|
if not _SLUG_RE.match(pid):
|
|
raise RegistryError(f"project id {pid!r} is not a valid slug")
|
|
if pid in seen:
|
|
raise RegistryError(f"duplicate project id {pid!r}")
|
|
seen.add(pid)
|
|
name = str(p.get("name") or "").strip()
|
|
if not name:
|
|
raise RegistryError(f"project {pid!r} missing name")
|
|
ptype = str(p.get("type") or "").strip()
|
|
if ptype not in VALID_TYPES:
|
|
raise RegistryError(f"project {pid!r} has invalid type {ptype!r}")
|
|
content_repo = str(p.get("content_repo") or "").strip()
|
|
if not content_repo:
|
|
raise RegistryError(f"project {pid!r} missing content_repo")
|
|
vis = str(p.get("visibility") or "gated").strip()
|
|
if vis not in VALID_VISIBILITY:
|
|
raise RegistryError(f"project {pid!r} has invalid visibility {vis!r}")
|
|
initial_state = str(
|
|
p.get("initial_state") or _TYPE_DEFAULT_INITIAL_STATE[ptype]
|
|
).strip()
|
|
if initial_state not in VALID_INITIAL_STATE:
|
|
raise RegistryError(
|
|
f"project {pid!r} has invalid initial_state {initial_state!r}"
|
|
)
|
|
cfg: dict = {}
|
|
if p.get("theme") is not None:
|
|
cfg["theme"] = p["theme"]
|
|
if p.get("enabled_models") is not None:
|
|
cfg["enabled_models"] = [str(m) for m in p["enabled_models"]]
|
|
entries.append(
|
|
ProjectEntry(pid, name, ptype, content_repo, vis, initial_state, cfg)
|
|
)
|
|
return RegistryDoc(
|
|
deployment_name=str(dep.get("name") or "").strip(),
|
|
deployment_tagline=str(dep.get("tagline") or "").strip(),
|
|
projects=entries,
|
|
)
|
|
|
|
|
|
def parse_collection_manifest(text: str) -> CollectionEntry:
|
|
"""Parse + validate a `.collection.yaml`. Pure (no I/O). Raises RegistryError.
|
|
|
|
`type` is required and immutable (§22.4a, enforced at upsert). `visibility`
|
|
is optional — omitted means inherit the project's. `initial_state` defaults
|
|
per type (§22.4b)."""
|
|
raw = yaml.safe_load(text) or {}
|
|
if not isinstance(raw, dict):
|
|
raise RegistryError("collection manifest must be a mapping")
|
|
ctype = str(raw.get("type") or "").strip()
|
|
if ctype not in VALID_TYPES:
|
|
raise RegistryError(f"collection has invalid type {ctype!r}")
|
|
vis = raw.get("visibility")
|
|
if vis is not None:
|
|
vis = str(vis).strip()
|
|
if vis not in VALID_VISIBILITY:
|
|
raise RegistryError(f"collection has invalid visibility {vis!r}")
|
|
initial_state = str(
|
|
raw.get("initial_state") or _TYPE_DEFAULT_INITIAL_STATE[ctype]
|
|
).strip()
|
|
if initial_state not in VALID_INITIAL_STATE:
|
|
raise RegistryError(f"collection has invalid initial_state {initial_state!r}")
|
|
name = raw.get("name")
|
|
name = str(name).strip() if name else None
|
|
# §22.12: an optional per-collection enabled_models list that narrows the
|
|
# project's universe. Absent → no narrowing (inherit). Present (incl. empty)
|
|
# → narrowing applies; [] opts the collection out of AI.
|
|
cfg: dict = {}
|
|
if raw.get("enabled_models") is not None:
|
|
em = raw["enabled_models"]
|
|
if not isinstance(em, list):
|
|
raise RegistryError("collection enabled_models must be a list")
|
|
cfg["enabled_models"] = [str(m) for m in em]
|
|
# §22.4a SLICE-2: the collection's metadata field schema. Parsed leniently —
|
|
# a bad field def is skipped with a warning, never fatal (INV-3), so a typo
|
|
# in one field can't drop the whole collection from the mirror. Stored only
|
|
# when at least one valid field survives.
|
|
fields = metadata_schema.parse_fields(raw.get("fields"))
|
|
if fields:
|
|
cfg["fields"] = fields
|
|
return CollectionEntry(ctype, vis, initial_state, name, cfg)
|
|
|
|
|
|
def _default_collection_id(project_id: str, default_id: str) -> str:
|
|
"""The id of a project's default collection. The deployment's primary
|
|
project (== `default_id`, the §22.13 resolved default) gets the stable
|
|
literal `'default'` — matching migration 029's seed so the upsert *merges*
|
|
onto the migration-seeded row rather than duplicating it (critical on a
|
|
fresh deploy where the bootstrap `default` project is later restamped to the
|
|
configured id). Any additional project keys its default collection by its own
|
|
id, keeping the collection PK globally unique (pre-S5 multi-project)."""
|
|
return "default" if project_id == default_id else project_id
|
|
|
|
|
|
def apply_registry(doc: RegistryDoc, registry_sha: str, default_id: str) -> None:
|
|
"""Upsert the parsed registry into projects + their default collections +
|
|
the deployment singleton. Idempotent.
|
|
|
|
§22 three-tier (S1): a project carries the grouping-tier fields (name,
|
|
content_repo, visibility, config); the per-corpus fields (`type`,
|
|
`initial_state`) live on the project's default collection. §22.4a: `type` is
|
|
immutable — a change against an existing collection is rejected (skip the
|
|
type change + log), never applied. Projects absent from the registry are
|
|
left in place (archival is out of scope for M3; they stop refreshing).
|
|
"""
|
|
with db.tx() as conn:
|
|
for e in doc.projects:
|
|
cid = _default_collection_id(e.id, default_id)
|
|
existing = conn.execute(
|
|
"SELECT type FROM collections WHERE id = ?", (cid,)
|
|
).fetchone()
|
|
type_locked = existing is not None and existing["type"] != e.type
|
|
if type_locked:
|
|
log.error(
|
|
"registry: refusing immutable type change on collection %s (%s -> %s)",
|
|
cid, existing["type"], e.type,
|
|
)
|
|
# The project (grouping tier) always refreshes.
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO projects
|
|
(id, name, content_repo, visibility, config_json, registry_sha, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, datetime('now'))
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
name = excluded.name,
|
|
content_repo = excluded.content_repo,
|
|
visibility = excluded.visibility,
|
|
config_json = excluded.config_json,
|
|
registry_sha = excluded.registry_sha,
|
|
updated_at = datetime('now')
|
|
""",
|
|
(e.id, e.name, e.content_repo, e.visibility, json.dumps(e.config), registry_sha),
|
|
)
|
|
# The default collection (corpus tier). On an immutable-type
|
|
# conflict, keep the stored type but still refresh the rest.
|
|
effective_type = existing["type"] if type_locked else e.type
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO collections
|
|
(id, project_id, type, subfolder, initial_state, visibility, name, registry_sha, updated_at)
|
|
VALUES (?, ?, ?, '', ?, ?, ?, ?, datetime('now'))
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
project_id = excluded.project_id,
|
|
type = excluded.type,
|
|
initial_state = excluded.initial_state,
|
|
visibility = excluded.visibility,
|
|
name = excluded.name,
|
|
registry_sha = excluded.registry_sha,
|
|
updated_at = datetime('now')
|
|
""",
|
|
(cid, e.id, effective_type, e.initial_state, e.visibility, e.name, registry_sha),
|
|
)
|
|
conn.execute(
|
|
"""
|
|
UPDATE deployment
|
|
SET name = ?, tagline = ?, registry_sha = ?, updated_at = datetime('now')
|
|
WHERE id = 1
|
|
""",
|
|
(doc.deployment_name, doc.deployment_tagline, registry_sha),
|
|
)
|
|
|
|
|
|
def _strictest_visibility(a: str, b: str) -> str:
|
|
"""The stricter of two §22.5 visibilities on the public-exposure axis
|
|
(`public` < `unlisted` < `gated`). Used to enforce that a collection is set
|
|
only as strict or stricter than its project (S3 operator decision)."""
|
|
rank = {"public": 0, "unlisted": 1, "gated": 2}
|
|
return a if rank.get(a, 2) >= rank.get(b, 2) else b
|
|
|
|
|
|
def _upsert_named_collection(
|
|
proj: ProjectEntry, subdir: str, ce: CollectionEntry, sha: str
|
|
) -> None:
|
|
"""Upsert one named collection (S2). Type is immutable (§22.4a): a type
|
|
change against an existing row is refused (logged, not applied). A None
|
|
manifest visibility inherits the project's visibility; a manifest that tries
|
|
to be *looser* than its project is clamped to the project's (S3 strictness:
|
|
a collection may narrow but never widen its project's visibility)."""
|
|
requested = ce.visibility or proj.visibility
|
|
visibility = _strictest_visibility(requested, proj.visibility)
|
|
if visibility != requested:
|
|
log.warning(
|
|
"registry: collection %s visibility %r looser than project %s %r — "
|
|
"clamped to %r (S3 strictness)",
|
|
subdir, requested, proj.id, proj.visibility, visibility,
|
|
)
|
|
with db.tx() as conn:
|
|
existing = conn.execute(
|
|
"SELECT type FROM collections WHERE id = ?", (subdir,)
|
|
).fetchone()
|
|
if existing is not None and existing["type"] != ce.type:
|
|
log.error(
|
|
"registry: refusing immutable type change on collection %s (%s -> %s)",
|
|
subdir, existing["type"], ce.type,
|
|
)
|
|
return
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO collections
|
|
(id, project_id, type, subfolder, initial_state, visibility, name, config_json, registry_sha, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
project_id = excluded.project_id,
|
|
initial_state = excluded.initial_state,
|
|
visibility = excluded.visibility,
|
|
name = excluded.name,
|
|
config_json = excluded.config_json,
|
|
registry_sha = excluded.registry_sha,
|
|
updated_at = datetime('now')
|
|
""",
|
|
(subdir, proj.id, ce.type, subdir, ce.initial_state, visibility, ce.name,
|
|
json.dumps(ce.config), sha),
|
|
)
|
|
|
|
|
|
async def _mirror_named_collections(config: Config, gitea: Gitea, doc: RegistryDoc, sha: str) -> None:
|
|
"""§22 S2: named collections are declared by `.collection.yaml` manifests
|
|
inside each project's content repo (the default collection comes from
|
|
projects.yaml). Walk each content repo root; a subdir carrying a manifest
|
|
becomes a collection keyed by the subdir name. Tolerant: a transport or
|
|
parse failure on one project/collection logs and is skipped, never aborts
|
|
the wider mirror (keep last-good)."""
|
|
for proj in doc.projects:
|
|
try:
|
|
items = await gitea.list_dir(config.gitea_org, proj.content_repo, "", ref="main")
|
|
except Exception as e: # noqa: BLE001 — GiteaError/transport: tolerate
|
|
log.warning("registry: cannot list %s root: %s", proj.content_repo, e)
|
|
continue
|
|
for it in items:
|
|
if it.get("type") != "dir":
|
|
continue
|
|
subdir = it["name"]
|
|
manifest = await gitea.get_contents(
|
|
config.gitea_org, proj.content_repo, f"{subdir}/.collection.yaml", ref="main"
|
|
)
|
|
if not manifest or manifest.get("type") != "file":
|
|
continue
|
|
mtext = base64.b64decode(manifest["content"]).decode("utf-8")
|
|
try:
|
|
ce = parse_collection_manifest(mtext)
|
|
except RegistryError as e:
|
|
log.error("registry: bad manifest %s/%s: %s", proj.content_repo, subdir, e)
|
|
continue
|
|
_upsert_named_collection(proj, subdir, ce, sha)
|
|
|
|
|
|
async def refresh_registry(config: Config, gitea: Gitea) -> None:
|
|
"""Mirror REGISTRY_REPO/projects.yaml into projects + deployment.
|
|
|
|
Idempotent. Raises RegistryError on a missing/invalid file and GiteaError
|
|
on transport failure; the caller chooses fatal-vs-tolerated.
|
|
"""
|
|
item = await gitea.get_contents(
|
|
config.gitea_org, config.registry_repo, "projects.yaml", ref="main"
|
|
)
|
|
if not item or item.get("type") != "file":
|
|
raise RegistryError(
|
|
f"{config.gitea_org}/{config.registry_repo}/projects.yaml not found"
|
|
)
|
|
text = base64.b64decode(item["content"]).decode("utf-8")
|
|
# Prefer the file's last commit sha for provenance (production Gitea
|
|
# includes it on the contents response); fall back to the blob sha.
|
|
sha = item.get("last_commit_sha") or item.get("sha") or ""
|
|
doc = parse_registry(text)
|
|
from . import projects as projects_mod
|
|
apply_registry(doc, sha, projects_mod.resolved_default_id(config))
|
|
# §22 S2: discover + upsert named collections from each content repo.
|
|
await _mirror_named_collections(config, gitea, doc, sha)
|
|
log.info("registry: mirrored %d project(s) at %s", len(doc.projects), sha)
|