"""Strict function library manifest schema, validation, and deterministic hashing. Single contract for seed YAML, admin API, and future imports. Write paths use :func:`src.domain.edge_functions.service.EdgeFunctionsService.upsert_library_function_from_manifest`. """ from __future__ import annotations import hashlib import json import re from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Literal, Optional, Union from pydantic import BaseModel, Field, field_validator, model_validator from src.domain.connectors.catalog import supported_connectors # --------------------------------------------------------------------------- # Field classification # --------------------------------------------------------------------------- # Safe to PATCH on published functions without a new version snapshot. MUTABLE_LIBRARY_FIELDS: frozenset[str] = frozenset({ "name", "description", "category", "tags", "visibility", "minimum_plan", "codegen_hints", "observability_config", "deprecated_at", "replacement_slug", "seed_path", }) # Affect runtime — changing these on a published function requires a new version. VERSIONED_LIBRARY_FIELDS: frozenset[str] = frozenset({ "runtime", "code_template", "code_bundle_ref", "code_template_path", "trigger_type", "supported_trigger_types", "execution_mode", "input_schema", "output_schema", "required_integrations", "required_secrets", "required_capabilities", "required_bindings", "permission_scope", "external_access", "trigger_config_schema", "timeout_ms", "max_retries", "idempotency_required", "billing_config", }) MUTABLE_FIELDS = MUTABLE_LIBRARY_FIELDS VERSIONED_FIELDS = VERSIONED_LIBRARY_FIELDS # Excluded from manifest hash (lifecycle, ownership, derived counters). _HASH_EXCLUDE: frozenset[str] = frozenset({ "status", "deprecated_at", "replacement_slug", "admin_override", "managed_mode", "install_count", "last_synced_from_seed_at", "latest_seed_hash", "manifest_hash", "is_official", "is_active", }) _REDACT_PATH = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9_.\-]*$") SLUG_RE = re.compile(r"^[a-z][a-z0-9-]*$") TriggerLiteral = Literal["http", "event", "schedule", "webhook", "manual", "queue"] # --------------------------------------------------------------------------- # Manifest model # --------------------------------------------------------------------------- class FunctionLibraryManifest(BaseModel): """Canonical library function: metadata + code + schemas + policy + lifecycle.""" slug: str = Field(..., max_length=64) name: str = Field(..., max_length=120) description: Optional[str] = None category: Optional[str] = None tags: Optional[List[str]] = None status: Literal["draft", "published", "deprecated", "disabled"] = "draft" visibility: Literal["public", "private", "internal", "enterprise"] = "public" version: str = Field("1.0.0", max_length=20) semver_channel: Literal["alpha", "beta", "stable"] = "stable" deprecated_at: Optional[str] = None replacement_slug: Optional[str] = None runtime: Literal["js", "ts", "python"] = "js" trigger_type: TriggerLiteral = "http" supported_trigger_types: Optional[List[str]] = None execution_mode: Literal["sync", "async"] = "sync" timeout_ms: int = Field(10000, ge=1000, le=300000) max_retries: int = Field(3, ge=0, le=10) idempotency_required: bool = False minimum_plan: Literal["free", "starter", "pro", "enterprise"] = "free" required_integrations: Optional[List[str]] = None required_secrets: Optional[List[str]] = None required_capabilities: Optional[List[str]] = None required_bindings: Optional[List[str]] = None input_schema: Optional[Dict[str, Any]] = None output_schema: Optional[Dict[str, Any]] = None permission_scope: Optional[List[str]] = None external_access: Optional[Dict[str, Any]] = None trigger_config_schema: Optional[Dict[str, Any]] = None codegen_hints: Optional[Dict[str, Any]] = None observability_config: Optional[Dict[str, Any]] = None billing_config: Optional[Dict[str, Any]] = None code_template: Optional[str] = None code_bundle_ref: Optional[str] = None # Repo-relative path; resolved at seed/import time into code_template (not always stored) code_template_path: Optional[str] = None seed_path: Optional[str] = None managed_source: Literal["seed", "admin", "marketplace", "system"] = "admin" managed_mode: Literal["locked", "editable", "forked"] = "editable" @field_validator("slug") @classmethod def slug_shape(cls, v: str) -> str: if not SLUG_RE.match(v): raise ValueError("slug must match ^[a-z][a-z0-9-]*$") return v @model_validator(mode="after") def validate_trigger_requirements(self) -> FunctionLibraryManifest: if self.trigger_type == "webhook": tcs = self.trigger_config_schema or {} if "webhook" not in tcs: raise ValueError("trigger_config_schema must include a 'webhook' key when trigger_type is 'webhook'") if self.trigger_type == "schedule": tcs = self.trigger_config_schema or {} if "schedule" not in tcs: raise ValueError("trigger_config_schema must include a 'schedule' key when trigger_type is 'schedule'") return self @model_validator(mode="after") def validate_billing_units(self) -> FunctionLibraryManifest: known_units = { "function_invocation", "external_api_call", "ai_token", "storage_read", "storage_write", "record_read", "record_write", } if self.billing_config: for unit in (self.billing_config.get("units") or []): if unit not in known_units: raise ValueError( f"Unknown billing unit: {unit!r}. Valid: {sorted(known_units)}" ) return self class ManifestValidationError(ValueError): def __init__(self, errors: list[str]) -> None: self.errors = errors super().__init__("; ".join(errors)) class FunctionLibraryManifestValidator: """Semantic checks beyond Pydantic (registry lookups, JSON-shape hints).""" def __init__(self, *, require_integrations_in_registry: bool = True) -> None: self._require_int = require_integrations_in_registry def validate(self, manifest: FunctionLibraryManifest) -> list[str]: errors: list[str] = [] if manifest.status == "deprecated" and not manifest.replacement_slug: errors.append("deprecated status should set replacement_slug when possible") if manifest.minimum_plan in ("pro", "enterprise") and not manifest.required_capabilities: errors.append( f"minimum_plan={manifest.minimum_plan!r} should declare required_capabilities" ) if manifest.trigger_type == "webhook" and not manifest.required_integrations: errors.append("webhook trigger_type should declare required_integrations") if self._require_int and manifest.required_integrations: known = set(supported_connectors()) for name in manifest.required_integrations: if name not in known: errors.append( f"Unknown integration {name!r} — not in connector catalog. " f"Known: {sorted(known)}" ) if manifest.required_capabilities: for cap in manifest.required_capabilities: if not isinstance(cap, str) or not re.match(r"^[a-z][a-z0-9_.-]*$", cap): errors.append( f"Invalid capability id {cap!r} (expected lowercase [a-z0-9_.-])" ) if manifest.input_schema and manifest.input_schema.get("type") not in (None, "object"): errors.append("input_schema.type must be 'object' when present") if manifest.output_schema and manifest.output_schema.get("type") not in (None, "object"): errors.append("output_schema.type must be 'object' when present") if manifest.observability_config: for field in (manifest.observability_config.get("redact_fields") or []): if not isinstance(field, str): errors.append("observability_config.redact_fields entries must be strings") elif not _REDACT_PATH.match(field): errors.append( f"observability_config.redact_fields invalid path syntax: {field!r}" ) if manifest.code_template_path and manifest.code_template: errors.append("Specify only one of code_template_path or code_template") return errors def validate_or_raise(self, manifest: FunctionLibraryManifest) -> None: errors = self.validate(manifest) if errors: raise ManifestValidationError(errors) # --------------------------------------------------------------------------- # Hash helpers # --------------------------------------------------------------------------- def _stable_json(obj: Any) -> str: return json.dumps(obj, sort_keys=True, ensure_ascii=False, separators=(",", ":")) def compute_manifest_hash(manifest: FunctionLibraryManifest) -> str: data = manifest.model_dump(exclude=_HASH_EXCLUDE, exclude_none=False) keys = MUTABLE_LIBRARY_FIELDS | VERSIONED_LIBRARY_FIELDS # code_template_path is versioned but not stored on row — still part of logical manifest payload = {k: v for k, v in data.items() if k in keys} return hashlib.sha256(_stable_json(payload).encode()).hexdigest() def compute_code_hash(code_template: Optional[str]) -> str: return hashlib.sha256((code_template or "").encode()).hexdigest() def compute_config_hash(manifest: FunctionLibraryManifest) -> str: data = manifest.model_dump(exclude_none=False) payload: Dict[str, Any] = {} for k in VERSIONED_LIBRARY_FIELDS - {"code_template", "code_bundle_ref", "code_template_path"}: if k in data: payload[k] = data[k] return hashlib.sha256(_stable_json(payload).encode()).hexdigest() # --------------------------------------------------------------------------- # Load / apply # --------------------------------------------------------------------------- def parse_manifest( data: dict[str, Any], *, overrides: Optional[dict[str, Any]] = None, ) -> FunctionLibraryManifest: merged = {**data} if overrides: merged.update(overrides) return FunctionLibraryManifest.model_validate(merged) def resolve_code_template_path( manifest: FunctionLibraryManifest, repo_root: Path, ) -> FunctionLibraryManifest: """Load ``code_template`` from ``code_template_path`` relative to repo root.""" if not manifest.code_template_path: return manifest path = manifest.code_template_path full = path if Path(path).is_absolute() else (repo_root / path) if not full.is_file(): raise FileNotFoundError(f"code_template_path not found: {full}") body = full.read_text(encoding="utf-8") return manifest.model_copy(update={"code_template": body}) def _deprec_str_to_dt(value: Optional[Union[str, datetime]]) -> Optional[datetime]: if value is None: return None if isinstance(value, datetime): return value if isinstance(value, str): if value.endswith("Z"): value = value[:-1] + "+00:00" return datetime.fromisoformat(value) return None def function_library_row_to_manifest(row: Any) -> FunctionLibraryManifest: """Map ORM :class:`FunctionLibrary` to manifest for diffs (best-effort).""" dep = row.deprecated_at dep_s: Optional[str] = dep.isoformat() if dep is not None else None return FunctionLibraryManifest( slug=row.slug, name=row.name, description=row.description, category=row.category, tags=row.tags, status=row.status, # type: ignore[arg-type] visibility=row.visibility, # type: ignore[arg-type] version=row.version, semver_channel=row.semver_channel, # type: ignore[arg-type] deprecated_at=dep_s, replacement_slug=row.replacement_slug, runtime=row.runtime, # type: ignore[arg-type] trigger_type=row.trigger_type, # type: ignore[arg-type] supported_trigger_types=row.supported_trigger_types, execution_mode=row.execution_mode, # type: ignore[arg-type] timeout_ms=row.timeout_ms, max_retries=row.max_retries, idempotency_required=row.idempotency_required, minimum_plan=row.minimum_plan, # type: ignore[arg-type] required_integrations=row.required_integrations, required_secrets=row.required_secrets, required_capabilities=row.required_capabilities, required_bindings=row.required_bindings, input_schema=row.input_schema, output_schema=row.output_schema, permission_scope=row.permission_scope, external_access=row.external_access, trigger_config_schema=row.trigger_config_schema, codegen_hints=row.codegen_hints, observability_config=row.observability_config, billing_config=row.billing_config, code_template=row.code_template, code_bundle_ref=row.code_bundle_ref, code_template_path=None, seed_path=row.seed_path, managed_source=row.managed_source, # type: ignore[arg-type] managed_mode=row.managed_mode, # type: ignore[arg-type] ) def apply_manifest_to_row_dict(manifest: FunctionLibraryManifest) -> dict[str, Any]: """ORM kwargs for :class:`FunctionLibrary` (excludes id and timestamps).""" dep_at = _deprec_str_to_dt(manifest.deprecated_at) return { "name": manifest.name, "slug": manifest.slug, "description": manifest.description, "category": manifest.category, "tags": manifest.tags, "status": manifest.status, "visibility": manifest.visibility, "version": manifest.version, "semver_channel": manifest.semver_channel, "deprecated_at": dep_at, "replacement_slug": manifest.replacement_slug, "runtime": manifest.runtime, "trigger_type": manifest.trigger_type, "supported_trigger_types": manifest.supported_trigger_types, "execution_mode": manifest.execution_mode, "timeout_ms": manifest.timeout_ms, "max_retries": manifest.max_retries, "idempotency_required": manifest.idempotency_required, "minimum_plan": manifest.minimum_plan, "required_integrations": manifest.required_integrations, "required_secrets": manifest.required_secrets, "required_capabilities": manifest.required_capabilities, "required_bindings": manifest.required_bindings, "input_schema": manifest.input_schema, "output_schema": manifest.output_schema, "permission_scope": manifest.permission_scope, "external_access": manifest.external_access, "trigger_config_schema": manifest.trigger_config_schema, "codegen_hints": manifest.codegen_hints, "observability_config": manifest.observability_config, "billing_config": manifest.billing_config, "code_template": manifest.code_template, "code_bundle_ref": manifest.code_bundle_ref, "managed_source": manifest.managed_source, "managed_mode": manifest.managed_mode, "seed_path": manifest.seed_path, } def diff_fields( current_row: Any, incoming: FunctionLibraryManifest, ) -> tuple[list[str], list[str]]: """Return ``(mutable_changes, versioned_changes)`` vs DB row.""" incoming_data = incoming.model_dump() mutable_changes: list[str] = [] versioned_changes: list[str] = [] for field in MUTABLE_LIBRARY_FIELDS: current_val = getattr(current_row, field, None) incoming_val = incoming_data.get(field) if field == "deprecated_at": cur_s = None cv = getattr(current_row, "deprecated_at", None) if cv is not None: cur_s = cv.isoformat() if isinstance(cv, datetime) else str(cv) in_s = incoming_val if in_s and isinstance(in_s, str) and in_s.endswith("Z"): in_s = in_s[:-1] + "+00:00" if cur_s != in_s: mutable_changes.append(field) continue if current_val != incoming_val: mutable_changes.append(field) for field in VERSIONED_LIBRARY_FIELDS: if field == "code_template_path": continue current_val = getattr(current_row, field, None) incoming_val = incoming_data.get(field) if current_val != incoming_val: versioned_changes.append(field) return mutable_changes, versioned_changes