"""Workspace-Scope-Registry aggregation (Phase N1, Roadmap Phase N). Plugins declare the scope dimensions their modules support via the contract hook ``workspace_scopes()`` — same aggregation pattern as ``document_placeholders()`` (#359 philosophy): the module owns its domain, the generic consumer stays module-agnostic. The N2 workspace editor renders filter UI from these definitions; scope VALUES are stored per workspace in ``workspace_modules.config`` (JSONB). N3 list filtering applies them as pure AND-restrictions. Security-Invariante (Phase N): a workspace can never GRANT visibility — effective visibility is always Workspace-Scope ∧ RLS ∧ ABAC ∧ Permissions. Without an active workspace there is no filter (backward compatible). """ from __future__ import annotations import logging from typing import Any from pydantic import ValidationError from app.schemas.workspace import WorkspaceModuleScopes logger = logging.getLogger(__name__) def _parse_contribution(plugin_name: str, contribution: Any) -> WorkspaceModuleScopes | None: """Validate a single raw contribution; ``None`` when invalid (fail-closed). Every declaration must pass the Pydantic schema — invalid dimensions (e.g. a multiselect without options AND value_source) are dropped with a warning instead of breaking the editor endpoint. """ if not isinstance(contribution, dict): return None try: return WorkspaceModuleScopes.model_validate(contribution) except ValidationError as exc: logger.warning( "Invalid workspace_scopes contribution from plugin '%s': %s", plugin_name, exc, ) return None def get_scope_definitions() -> dict[str, list[dict[str, Any]]]: """Aggregate ``workspace_scopes()`` contributions from all discovered plugins. Returns ``module_key -> [dimension, ...]`` for the N2 editor. Contracts of explicitly deactivated plugins stay gone (ARCH-014 unregister marker); never-loaded contracts lazy-load on first access (document_placeholders precedent). A plugin crash while declaring scopes never fails the endpoint — its contribution is simply skipped. """ from app.plugins.builtins.contracts import get_contract_registry from app.plugins.registry import get_registry modules: dict[str, list[dict[str, Any]]] = {} for plugin_name in get_registry().list_discovered(): contract = get_contract_registry().get_contract(plugin_name) if contract is None: continue fn = getattr(contract, "workspace_scopes", None) if fn is None: continue try: contributions = fn() or [] except Exception: # noqa: BLE001 logger.exception( "workspace_scopes() raised for plugin '%s' — skipping", plugin_name ) continue for contribution in contributions: parsed = _parse_contribution(plugin_name, contribution) if parsed is None: continue modules.setdefault(parsed.module_key, []).extend( dimension.model_dump() for dimension in parsed.dimensions ) return modules