From c2f7048d11697b809acd558edf72d07c0a48daa5 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Rodrigo=20M=C3=A9ndez?= <rodrigo.mendez@wgu.edu>
Date: Wed, 23 Sep 2026 10:03:27 -0600
Subject: [PATCH 01/10] feat: add authz schema compilation

---
 .../engine/schema/compilation.py              | 339 ++++++++++++++++
 .../tests/schema/test_compilation.py          | 365 ++++++++++++++++++
 2 files changed, 704 insertions(+)
 create mode 100644 src/openedx_authz/engine/schema/compilation.py
 create mode 100644 src/openedx_authz/tests/schema/test_compilation.py

diff --git a/src/openedx_authz/engine/schema/compilation.py b/src/openedx_authz/engine/schema/compilation.py
new file mode 100644
index 00000000..5c7dc974
--- /dev/null
+++ b/src/openedx_authz/engine/schema/compilation.py
@@ -0,0 +1,339 @@
+"""Resolve documents into one set of static definitions (the ``compile`` step).
+
+Compilation (ADR 0018 §1) merges base definitions across all documents and
+applies ``role_extensions`` per ADR 0023:
+
+    * Extensions resolve only after every role and permission is loaded.
+    * An extension changes only the fields it includes; absent fields keep
+      their current value; it cannot change a role ID.
+    * Different fields from different contributions combine.
+    * ``priority`` resolves conflicts on the same metadata field or the same
+      permission (higher wins). Equal priority with disagreeing values raises
+      :class:`SchemaCompileError` so deployment stops before the database
+      changes.
+    * Adding a permission the role already has, or removing one it lacks, is a
+      no-op logged as a warning.
+
+Every resulting :class:`CompiledDefinition` retains all contributing
+:class:`SourceRecord` values, and each role-permission grant is attributed at
+the (role, permission) grain with its origin (base vs extension) for ADR 0025
+source tracking. Output is deterministic regardless of discovery order. No
+Casbin/Django imports.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field, replace
+
+from openedx_authz.constants import SchemaOriginKind
+from openedx_authz.engine.schema.exceptions import SchemaCompileError
+from openedx_authz.engine.schema.types import (
+    CompiledDefinition,
+    CompiledSchema,
+    RelationshipSource,
+    RoleDefinition,
+    SchemaDocument,
+    SourceRecord,
+)
+
+logger = logging.getLogger(__name__)
+
+# Metadata fields an extension may replace on a role.
+_METADATA_FIELDS = ("display_name", "description", "icon", "hidden")
+
+# Singular labels for operator-facing messages, keyed by document attribute.
+_KIND_LABELS = {"categories": "category", "permissions": "permission", "roles": "role"}
+
+
+@dataclass
+class _Tracked:
+    """A base definition plus the sources and priority that produced it."""
+
+    definition: object
+    sources: list[SourceRecord] = field(default_factory=list)
+    priority: int = 0
+
+
+class SchemaCompiler:
+    """Merges validated documents into a :class:`CompiledSchema`."""
+
+    def compile(self, documents: list[SchemaDocument]) -> CompiledSchema:
+        """Resolve categories, permissions, roles, and extensions.
+
+        Assumes ``documents`` already passed validation.
+
+        Raises:
+            SchemaCompileError: On an unresolvable equal-priority conflict.
+        """
+        categories = self._collect(documents, "categories", key=lambda c: c.id)
+        permissions = self._collect(documents, "permissions", key=lambda p: p.identifier)
+        roles = self._collect(documents, "roles", key=lambda r: r.id)
+
+        role_permission_sources = self._resolve_roles_and_provenance(roles, documents)
+
+        return CompiledSchema(
+            categories=self._finalize(categories),
+            permissions=self._finalize(permissions),
+            roles=self._finalize(roles),
+            role_permission_sources=role_permission_sources,
+        )
+
+    # ---- base collection --------------------------------------------------
+
+    def _collect(self, documents: list[SchemaDocument], attr: str, key) -> dict[str, _Tracked]:
+        """Gather base definitions keyed by identifier, resolving by priority.
+
+        Higher priority wins on conflict; equal priority with differing content
+        raises; identical duplicates merge their sources.
+        """
+        tracked: dict[str, _Tracked] = {}
+        for document in documents:
+            for definition in getattr(document, attr):
+                identifier = key(definition)
+                existing = tracked.get(identifier)
+                if existing is None:
+                    tracked[identifier] = _Tracked(
+                        definition=definition,
+                        sources=[document.source],
+                        priority=document.priority,
+                    )
+                    continue
+
+                kind = _KIND_LABELS.get(attr, attr)
+                if existing.definition == definition:
+                    existing.sources.append(document.source)
+                elif document.priority > existing.priority:
+                    # The loser is discarded; say so, otherwise the contributing
+                    # file looks like it took effect (ADR 0017 §4).
+                    self._warn_discarded(
+                        kind,
+                        identifier,
+                        loser=existing.sources[0],
+                        loser_priority=existing.priority,
+                        winner=document.source,
+                        winner_priority=document.priority,
+                    )
+                    tracked[identifier] = _Tracked(
+                        definition=definition,
+                        sources=[document.source],
+                        priority=document.priority,
+                    )
+                elif document.priority == existing.priority:
+                    raise SchemaCompileError(
+                        f"Conflicting {kind} definition for {identifier!r} at equal priority "
+                        f"{document.priority} ({existing.sources[0].source_id} vs {document.source.source_id})."
+                    )
+                else:
+                    self._warn_discarded(
+                        kind,
+                        identifier,
+                        loser=document.source,
+                        loser_priority=document.priority,
+                        winner=existing.sources[0],
+                        winner_priority=existing.priority,
+                    )
+        return tracked
+
+    @staticmethod
+    def _warn_discarded(
+        kind: str,
+        identifier: str,
+        *,
+        loser: SourceRecord,
+        loser_priority: int,
+        winner: SourceRecord,
+        winner_priority: int,
+    ) -> None:
+        """Report a contribution that lost to a higher-priority one.
+
+        Priority silently picking a winner is the behavior operators find hardest
+        to debug: the losing file is valid, was loaded, and simply has no effect.
+        ADR 0017 §4 requires warning about exactly this.
+        """
+        logger.warning(
+            "authz schema: %s %r from %s (priority %s) has no effect; %s (priority %s) takes precedence.",
+            kind,
+            identifier,
+            loser.source_id,
+            loser_priority,
+            winner.source_id,
+            winner_priority,
+        )
+
+    # ---- roles + provenance ----------------------------------------------
+
+    def _resolve_roles_and_provenance(
+        self, roles: dict[str, _Tracked], documents: list[SchemaDocument]
+    ) -> dict[tuple[str, str], list[RelationshipSource]]:
+        """Apply extensions and build per-(role, permission) provenance.
+
+        Seeds base provenance from each role's own definition, then folds in
+        ``role_extensions`` (metadata replacement + permission add/remove),
+        honoring priority. Returns the relationship provenance map.
+        """
+        metadata_changes, perm_changes = self._gather_extension_changes(roles, documents)
+        rp_sources: dict[tuple[str, str], list[RelationshipSource]] = {}
+
+        for role_id, tracked in roles.items():
+            role: RoleDefinition = tracked.definition
+            base_sources = list(tracked.sources)
+            base_priority = tracked.priority
+
+            # Seed base provenance for every permission the role declares.
+            provenance: dict[str, list[RelationshipSource]] = {
+                perm: [RelationshipSource(src, SchemaOriginKind.BASE, base_priority) for src in base_sources]
+                for perm in role.permissions
+            }
+
+            md = metadata_changes.get(role_id, {})
+            if md:
+                new_values, contributing_sources = self._resolve_metadata(role_id, md)
+                tracked.definition = replace(role, **new_values)
+                role = tracked.definition
+                for src in contributing_sources:
+                    if src not in tracked.sources:
+                        tracked.sources.append(src)
+
+            pc = perm_changes.get(role_id)
+            if pc and (pc["add"] or pc["remove"]):
+                final_perms, provenance = self._resolve_permissions(
+                    role_id, role.permissions, base_sources, base_priority, pc
+                )
+                tracked.definition = replace(tracked.definition, permissions=final_perms)
+
+            for perm, sources in provenance.items():
+                rp_sources[(role_id, perm)] = sources
+
+        return rp_sources
+
+    def _gather_extension_changes(self, roles: dict[str, _Tracked], documents: list[SchemaDocument]):
+        """Collect per-role metadata and permission changes from all extensions.
+
+        Entries carry the full :class:`SourceRecord` and priority so provenance
+        and conflict resolution have everything they need.
+        """
+        metadata_changes: dict[str, dict[str, list[tuple[object, int, SourceRecord]]]] = {}
+        perm_changes: dict[str, dict[str, list[tuple[str, int, SourceRecord]]]] = {}
+
+        for document in documents:
+            for extension in document.role_extensions:
+                role_id = extension.role
+                if role_id not in roles:
+                    # Validation already errors on this; skip defensively.
+                    continue
+                md = metadata_changes.setdefault(role_id, {})
+                for field_name in _METADATA_FIELDS:
+                    value = getattr(extension, field_name)
+                    if value is not None:
+                        md.setdefault(field_name, []).append((value, document.priority, document.source))
+                pc = perm_changes.setdefault(role_id, {"add": [], "remove": []})
+                for perm in extension.add_permissions:
+                    pc["add"].append((perm, document.priority, document.source))
+                for perm in extension.remove_permissions:
+                    pc["remove"].append((perm, document.priority, document.source))
+        return metadata_changes, perm_changes
+
+    def _resolve_metadata(self, role_id: str, md: dict[str, list[tuple[object, int, SourceRecord]]]):
+        """Pick winning metadata values by priority; error on equal-priority ties."""
+        new_values: dict[str, object] = {}
+        contributing: set[SourceRecord] = set()
+        for field_name, entries in md.items():
+            max_priority = max(priority for _, priority, _ in entries)
+            top_values = {value for value, priority, _ in entries if priority == max_priority}
+            if len(top_values) > 1:
+                raise SchemaCompileError(
+                    f"Conflicting {field_name!r} for role {role_id!r} at equal priority "
+                    f"{max_priority}: {sorted(map(str, top_values))}."
+                )
+            new_values[field_name] = next(iter(top_values))
+            contributing.update(src for _, priority, src in entries if priority == max_priority)
+            winner = next(src for _, priority, src in entries if priority == max_priority)
+            for _, priority, src in entries:
+                if priority < max_priority:
+                    self._warn_discarded(
+                        f"role_extension {field_name}",
+                        role_id,
+                        loser=src,
+                        loser_priority=priority,
+                        winner=winner,
+                        winner_priority=max_priority,
+                    )
+        return new_values, contributing
+
+    def _resolve_permissions(
+        self,
+        role_id: str,
+        base: tuple[str, ...],
+        base_sources: list[SourceRecord],
+        base_priority: int,
+        pc: dict[str, list[tuple[str, int, SourceRecord]]],
+    ):
+        """Apply add/remove per permission, returning (final_perms, provenance).
+
+        Add-vs-remove conflicts resolve by priority; equal priority raises.
+        Provenance keeps base attribution and appends extension attribution for
+        added permissions.
+        """
+        current = set(base)
+        provenance: dict[str, list[RelationshipSource]] = {
+            perm: [RelationshipSource(src, SchemaOriginKind.BASE, base_priority) for src in base_sources]
+            for perm in base
+        }
+
+        actions: dict[str, list[tuple[str, int, SourceRecord]]] = {}
+        for perm, priority, src in pc["add"]:
+            actions.setdefault(perm, []).append(("add", priority, src))
+        for perm, priority, src in pc["remove"]:
+            actions.setdefault(perm, []).append(("remove", priority, src))
+
+        for perm, entries in actions.items():
+            max_priority = max(priority for _, priority, _ in entries)
+            top = {action for action, priority, _ in entries if priority == max_priority}
+            if len(top) > 1:
+                raise SchemaCompileError(
+                    f"Conflicting add/remove for permission {perm!r} on role {role_id!r} "
+                    f"at equal priority {max_priority}."
+                )
+            action = next(iter(top))
+            winning_sources = [src for act, priority, src in entries if priority == max_priority and act == action]
+
+            for act, priority, src in entries:
+                if priority < max_priority:
+                    self._warn_discarded(
+                        f"role_extension {act} of {perm!r} on role",
+                        role_id,
+                        loser=src,
+                        loser_priority=priority,
+                        winner=winning_sources[0],
+                        winner_priority=max_priority,
+                    )
+
+            if action == "add":
+                if perm in current:
+                    logger.warning("role_extension adds %r already on role %r; no-op.", perm, role_id)
+                current.add(perm)
+                provenance.setdefault(perm, [])
+                provenance[perm].extend(
+                    RelationshipSource(src, SchemaOriginKind.EXTENSION, max_priority) for src in winning_sources
+                )
+            else:  # remove
+                if perm not in current:
+                    logger.warning("role_extension removes %r not on role %r; no-op.", perm, role_id)
+                current.discard(perm)
+                provenance.pop(perm, None)
+
+        return tuple(sorted(current)), provenance
+
+    # ---- finalize ---------------------------------------------------------
+
+    def _finalize(self, tracked: dict[str, _Tracked]) -> dict[str, CompiledDefinition]:
+        """Turn tracked definitions into CompiledDefinition entries."""
+        return {
+            identifier: CompiledDefinition(
+                key=identifier,
+                definition=entry.definition,
+                sources=tuple(entry.sources),
+            )
+            for identifier, entry in tracked.items()
+        }
diff --git a/src/openedx_authz/tests/schema/test_compilation.py b/src/openedx_authz/tests/schema/test_compilation.py
new file mode 100644
index 00000000..6285c945
--- /dev/null
+++ b/src/openedx_authz/tests/schema/test_compilation.py
@@ -0,0 +1,365 @@
+"""Unit tests for the schema compilation step (merge + extensions + priority).
+
+Grouped by concern: base compilation, extensions, priority resolution,
+provenance, discarded-contribution warnings, no-op extension warnings, and the
+defensive branches guarding states validation is expected to have rejected.
+"""
+
+import logging
+
+import pytest
+
+from openedx_authz.constants import SchemaOriginKind
+from openedx_authz.engine.schema.compilation import SchemaCompiler
+from openedx_authz.engine.schema.exceptions import SchemaCompileError
+
+from .factories import category, extension, make_document, make_source, permission, role
+
+PERMS = [
+    permission(name="view_course", cat="cat"),
+    permission(name="export_course", cat="cat"),
+    permission(name="manage_tags", cat="cat"),
+]
+
+
+def _base(**role_kwargs):
+    return make_document(
+        "base",
+        priority=100,
+        categories=[category("cat")],
+        permissions=PERMS,
+        roles=[role(rid="course_editor", permissions=("courses.view_course", "courses.manage_tags"), **role_kwargs)],
+    )
+
+
+class TestBaseCompilation:
+    """Compiling base definitions with no extensions applied."""
+
+    def test_base_definitions_compile(self):
+        """A single base document yields its roles and permissions verbatim."""
+        schema = SchemaCompiler().compile([_base()])
+        assert set(schema.roles) == {"course_editor"}
+        assert len(schema.permissions) == 3
+        # Base definitions keep their declared order; rendering sorts later.
+        assert schema.roles["course_editor"].definition.permissions == (
+            "courses.view_course",
+            "courses.manage_tags",
+        )
+
+
+class TestExtensions:
+    """Applying ``role_extensions`` on top of a base role definition."""
+
+    def test_extension_adds_and_removes_permissions_and_metadata(self):
+        """One extension can add, remove, rename, and hide in a single pass."""
+        ext = make_document(
+            "ext",
+            priority=200,
+            role_extensions=[
+                extension(
+                    "course_editor",
+                    add_permissions=("courses.export_course",),
+                    remove_permissions=("courses.manage_tags",),
+                    display_name="Author",
+                    hidden=True,
+                )
+            ],
+        )
+        definition = SchemaCompiler().compile([_base(), ext]).roles["course_editor"].definition
+        assert "courses.export_course" in definition.permissions
+        assert "courses.manage_tags" not in definition.permissions
+        assert definition.display_name == "Author"
+        assert definition.hidden is True
+
+    def test_extension_sources_are_retained(self):
+        """A role touched by an extension keeps both the base and extension sources."""
+        ext = make_document("ext", priority=200, role_extensions=[extension("course_editor", display_name="X")])
+        compiled = SchemaCompiler().compile([_base(), ext])
+        assert len(compiled.roles["course_editor"].sources) == 2
+
+
+class TestPriorityResolution:
+    """How priority resolves conflicting contributions (higher wins; ties fail)."""
+
+    def test_equal_priority_metadata_conflict_raises(self):
+        """Two extensions setting the same field at equal priority is an error."""
+        a = make_document("a", priority=200, role_extensions=[extension("course_editor", display_name="A")])
+        b = make_document("b", priority=200, role_extensions=[extension("course_editor", display_name="B")])
+        with pytest.raises(SchemaCompileError):
+            SchemaCompiler().compile([_base(), a, b])
+
+    def test_higher_priority_metadata_wins(self):
+        """The higher-priority extension's metadata value takes effect."""
+        lo = make_document("lo", priority=150, role_extensions=[extension("course_editor", display_name="Lo")])
+        hi = make_document("hi", priority=300, role_extensions=[extension("course_editor", display_name="Hi")])
+        definition = SchemaCompiler().compile([_base(), lo, hi]).roles["course_editor"].definition
+        assert definition.display_name == "Hi"
+
+    def test_equal_priority_add_remove_conflict_raises(self):
+        """An add and a remove of the same permission at equal priority is an error."""
+        add = make_document(
+            "add",
+            priority=200,
+            role_extensions=[extension("course_editor", add_permissions=("courses.export_course",))],
+        )
+        rem = make_document(
+            "rem",
+            priority=200,
+            role_extensions=[extension("course_editor", remove_permissions=("courses.export_course",))],
+        )
+        with pytest.raises(SchemaCompileError):
+            SchemaCompiler().compile([_base(), add, rem])
+
+    def test_conflicting_base_definition_equal_priority_raises(self):
+        """Two base definitions of the same role at equal priority is an error."""
+        a = make_document("a", priority=100, roles=[role(rid="dup", display_name="A", permissions=())])
+        b = make_document("b", priority=100, roles=[role(rid="dup", display_name="B", permissions=())])
+        with pytest.raises(SchemaCompileError):
+            SchemaCompiler().compile([a, b])
+
+    def test_higher_priority_base_definition_wins(self):
+        """The higher-priority base definition replaces the lower one."""
+        lo = make_document("lo", priority=100, roles=[role(rid="dup", display_name="Lo", permissions=())])
+        hi = make_document("hi", priority=200, roles=[role(rid="dup", display_name="Hi", permissions=())])
+        compiled = SchemaCompiler().compile([lo, hi])
+        assert compiled.roles["dup"].definition.display_name == "Hi"
+
+
+class TestProvenance:
+    """Each role-permission grant records where it came from (ADR 0025)."""
+
+    def test_base_permissions_get_base_provenance(self):
+        """Permissions from the role's own definition are tagged ``BASE``."""
+        schema = SchemaCompiler().compile([_base()])
+        for perm in ("courses.view_course", "courses.manage_tags"):
+            prov = schema.role_permission_sources[("course_editor", perm)]
+            assert [(rs.source.distribution, rs.origin_kind) for rs in prov] == [("test-dist", SchemaOriginKind.BASE)]
+
+    def test_extension_grant_is_attributed_to_the_module_not_core(self):
+        """An extension-added grant is tagged ``EXTENSION``, base grants stay ``BASE``."""
+        ext = make_document(
+            "modx",
+            priority=200,
+            role_extensions=[extension("course_editor", add_permissions=("courses.export_course",))],
+        )
+        schema = SchemaCompiler().compile([_base(), ext])
+
+        core = schema.role_permission_sources[("course_editor", "courses.view_course")]
+        added = schema.role_permission_sources[("course_editor", "courses.export_course")]
+
+        # Both permissions coexist on the role, but their origins remain distinct.
+        assert [rs.origin_kind for rs in core] == [SchemaOriginKind.BASE]
+        assert [rs.origin_kind for rs in added] == [SchemaOriginKind.EXTENSION]
+
+    def test_removed_permission_has_no_provenance(self):
+        """A permission removed by an extension leaves no provenance entry."""
+        ext = make_document(
+            "modx",
+            priority=200,
+            role_extensions=[extension("course_editor", remove_permissions=("courses.manage_tags",))],
+        )
+        schema = SchemaCompiler().compile([_base(), ext])
+        assert ("course_editor", "courses.manage_tags") not in schema.role_permission_sources
+
+
+class TestDiscardedContributionWarnings:
+    """Priority silently picks a winner; the loser must be reported.
+
+    ADR 0017 §4 requires warning about contributions that do not take effect
+    because another file has a higher priority. A losing file is valid and was
+    loaded, so without a warning it looks like it applied.
+    """
+
+    @staticmethod
+    def _compile(*documents):
+        return SchemaCompiler().compile(list(documents))
+
+    def test_lower_priority_base_definition_warns(self, caplog):
+        """A base definition that loses on priority is reported as having no effect."""
+        low = make_document("low", priority=100, roles=[role(rid="course_editor", display_name="Editor")])
+        high = make_document("high", priority=200, roles=[role(rid="course_editor", display_name="Author")])
+
+        with caplog.at_level(logging.WARNING):
+            compiled = self._compile(high, low)
+
+        assert compiled.roles["course_editor"].definition.display_name == "Author"
+        assert "has no effect" in caplog.text
+        assert make_source("low").source_id in caplog.text
+        assert make_source("high").source_id in caplog.text
+
+    def test_warning_names_both_priorities(self, caplog):
+        """The discard warning names both the losing and winning priorities."""
+        low = make_document("low", priority=100, roles=[role(rid="course_editor", display_name="Editor")])
+        high = make_document("high", priority=200, roles=[role(rid="course_editor", display_name="Author")])
+
+        with caplog.at_level(logging.WARNING):
+            self._compile(low, high)
+
+        assert "priority 100" in caplog.text
+        assert "priority 200" in caplog.text
+
+    def test_warns_regardless_of_document_order(self, caplog):
+        """Discovery order must not decide whether the operator is told."""
+        low = make_document("low", priority=100, roles=[role(rid="course_editor", display_name="Editor")])
+        high = make_document("high", priority=200, roles=[role(rid="course_editor", display_name="Author")])
+
+        with caplog.at_level(logging.WARNING):
+            self._compile(low, high)
+        ascending = caplog.text
+        caplog.clear()
+        with caplog.at_level(logging.WARNING):
+            self._compile(high, low)
+
+        assert "has no effect" in ascending
+        assert "has no effect" in caplog.text
+
+    def test_identical_duplicate_does_not_warn(self):
+        """An identical definition merges sources; nothing is discarded."""
+        first = make_document("first", priority=100, roles=[role(rid="course_editor")])
+        second = make_document("second", priority=200, roles=[role(rid="course_editor")])
+
+        compiled = self._compile(first, second)
+
+        assert len(compiled.roles["course_editor"].sources) == 2
+
+    def test_uses_singular_kind_label(self, caplog):
+        """Messages say 'category', not the truncated attribute name."""
+        low = make_document("low", priority=100, categories=[category("cat", display_name="Low")])
+        high = make_document("high", priority=200, categories=[category("cat", display_name="High")])
+
+        with caplog.at_level(logging.WARNING):
+            self._compile(low, high)
+
+        assert "category 'cat'" in caplog.text
+        assert "categorie" not in caplog.text
+
+    def test_conflict_error_uses_singular_kind_label(self):
+        """A conflict error uses the singular kind label ('category')."""
+        left = make_document("left", priority=100, categories=[category("cat", display_name="Left")])
+        right = make_document("right", priority=100, categories=[category("cat", display_name="Right")])
+
+        with pytest.raises(SchemaCompileError, match="Conflicting category definition"):
+            self._compile(left, right)
+
+    def test_losing_metadata_extension_warns(self, caplog):
+        """A metadata extension that loses on priority is reported, naming its source."""
+        base = make_document("base", priority=100, roles=[role(rid="course_editor")])
+        low = make_document("low", priority=100, role_extensions=[extension("course_editor", display_name="Low")])
+        high = make_document("high", priority=200, role_extensions=[extension("course_editor", display_name="High")])
+
+        with caplog.at_level(logging.WARNING):
+            compiled = self._compile(base, low, high)
+
+        assert compiled.roles["course_editor"].definition.display_name == "High"
+        assert "role_extension display_name" in caplog.text
+        assert make_source("low").source_id in caplog.text
+
+    def test_losing_permission_extension_warns(self, caplog):
+        """A permission-changing extension that loses on priority is reported."""
+        base = make_document(
+            "base",
+            priority=100,
+            permissions=[permission(cat="cat")],
+            roles=[role(rid="course_editor", permissions=("courses.view_course",))],
+        )
+        low = make_document(
+            "low",
+            priority=100,
+            role_extensions=[extension("course_editor", remove_permissions=("courses.view_course",))],
+        )
+        high = make_document(
+            "high",
+            priority=200,
+            role_extensions=[extension("course_editor", add_permissions=("courses.view_course",))],
+        )
+
+        with caplog.at_level(logging.WARNING):
+            compiled = self._compile(base, low, high)
+
+        # The higher-priority add wins, so the permission stays.
+        assert "courses.view_course" in compiled.roles["course_editor"].definition.permissions
+        assert "role_extension remove of 'courses.view_course'" in caplog.text
+
+
+class TestNoOpExtensionWarnings:
+    """ADR 0023 §3: a no-op add/remove warns and leaves the result unchanged."""
+
+    @staticmethod
+    def _compile(*documents):
+        return SchemaCompiler().compile(list(documents))
+
+    def test_adding_an_existing_permission_warns(self, caplog):
+        """Adding a permission the role already has warns and is a no-op."""
+        base = make_document(
+            "base",
+            priority=100,
+            permissions=[permission(cat="cat")],
+            roles=[role(rid="course_editor", permissions=("courses.view_course",))],
+        )
+        ext = make_document(
+            "ext", priority=200, role_extensions=[extension("course_editor", add_permissions=("courses.view_course",))]
+        )
+
+        with caplog.at_level(logging.WARNING):
+            compiled = self._compile(base, ext)
+
+        assert compiled.roles["course_editor"].definition.permissions == ("courses.view_course",)
+        assert "already on role" in caplog.text
+
+    def test_removing_an_absent_permission_warns(self, caplog):
+        """Removing a permission the role does not have warns and is a no-op."""
+        base = make_document("base", priority=100, roles=[role(rid="course_editor", permissions=())])
+        ext = make_document(
+            "ext",
+            priority=200,
+            role_extensions=[extension("course_editor", remove_permissions=("courses.manage_tags",))],
+        )
+
+        with caplog.at_level(logging.WARNING):
+            compiled = self._compile(base, ext)
+
+        assert compiled.roles["course_editor"].definition.permissions == ()
+        assert "not on role" in caplog.text
+
+
+class TestDefensiveBranches:
+    """Paths guarded against states validation is expected to have rejected."""
+
+    def test_extension_for_an_unknown_role_is_skipped(self):
+        """Validation errors on this; compilation must not raise on it."""
+        ext = make_document(
+            "ext", priority=200, role_extensions=[extension("ghost", add_permissions=("courses.view_course",))]
+        )
+
+        compiled = SchemaCompiler().compile([ext])
+
+        assert not compiled.roles
+        assert not compiled.role_permission_sources
+
+    def test_identical_duplicate_categories_merge_sources(self):
+        """Two identical category definitions merge into one, keeping both sources."""
+        first = make_document("first", categories=[category("cat")])
+        second = make_document("second", categories=[category("cat")])
+
+        compiled = SchemaCompiler().compile([first, second])
+
+        assert len(compiled.categories["cat"].sources) == 2
+
+    def test_identical_duplicate_permissions_merge_sources(self):
+        """Two identical permission definitions merge into one, keeping both sources."""
+        first = make_document("first", permissions=[permission(cat="cat")])
+        second = make_document("second", permissions=[permission(cat="cat")])
+
+        compiled = SchemaCompiler().compile([first, second])
+
+        assert len(compiled.permissions["courses.view_course"].sources) == 2
+
+    def test_lower_priority_base_definition_is_kept_out(self):
+        """The 'keep existing' branch: a later, lower-priority file loses."""
+        high = make_document("high", priority=200, roles=[role(rid="course_editor", display_name="Author")])
+        low = make_document("low", priority=100, roles=[role(rid="course_editor", display_name="Editor")])
+
+        compiled = SchemaCompiler().compile([high, low])
+
+        assert compiled.roles["course_editor"].definition.display_name == "Author"
+        assert [s.source_id for s in compiled.roles["course_editor"].sources] == [make_source("high").source_id]

From 2b42335c6efe90addec31c7abf49819bd57a425a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Rodrigo=20M=C3=A9ndez?= <rodrigo.mendez@wgu.edu>
Date: Fri, 2 Oct 2026 12:13:44 -0600
Subject: [PATCH 02/10] squash!: Fix rebase issues

---
 src/openedx_authz/engine/schema/compilation.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/openedx_authz/engine/schema/compilation.py b/src/openedx_authz/engine/schema/compilation.py
index 5c7dc974..8c5b19b3 100644
--- a/src/openedx_authz/engine/schema/compilation.py
+++ b/src/openedx_authz/engine/schema/compilation.py
@@ -218,7 +218,7 @@ def _gather_extension_changes(self, roles: dict[str, _Tracked], documents: list[
 
         for document in documents:
             for extension in document.role_extensions:
-                role_id = extension.role
+                role_id = extension.role_id
                 if role_id not in roles:
                     # Validation already errors on this; skip defensively.
                     continue

From 1f6d1c511d1411014fe1ec8b72224aeaff911db3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Rodrigo=20M=C3=A9ndez?= <rodrigo.mendez@wgu.edu>
Date: Fri, 2 Oct 2026 12:55:36 -0600
Subject: [PATCH 03/10] squash!: refactor conflict resolution methods

---
 .../engine/schema/compilation.py              | 190 +++++++++++-------
 1 file changed, 117 insertions(+), 73 deletions(-)

diff --git a/src/openedx_authz/engine/schema/compilation.py b/src/openedx_authz/engine/schema/compilation.py
index 8c5b19b3..8bef2edb 100644
--- a/src/openedx_authz/engine/schema/compilation.py
+++ b/src/openedx_authz/engine/schema/compilation.py
@@ -79,8 +79,6 @@ def compile(self, documents: list[SchemaDocument]) -> CompiledSchema:
             role_permission_sources=role_permission_sources,
         )
 
-    # ---- base collection --------------------------------------------------
-
     def _collect(self, documents: list[SchemaDocument], attr: str, key) -> dict[str, _Tracked]:
         """Gather base definitions keyed by identifier, resolving by priority.
 
@@ -101,40 +99,65 @@ def _collect(self, documents: list[SchemaDocument], attr: str, key) -> dict[str,
                     continue
 
                 kind = _KIND_LABELS.get(attr, attr)
-                if existing.definition == definition:
-                    existing.sources.append(document.source)
-                elif document.priority > existing.priority:
-                    # The loser is discarded; say so, otherwise the contributing
-                    # file looks like it took effect (ADR 0017 §4).
-                    self._warn_discarded(
-                        kind,
-                        identifier,
-                        loser=existing.sources[0],
-                        loser_priority=existing.priority,
-                        winner=document.source,
-                        winner_priority=document.priority,
-                    )
-                    tracked[identifier] = _Tracked(
+                tracked[identifier] = self._resolve_priority(
+                    kind,
+                    identifier,
+                    existing=existing,
+                    incoming=_Tracked(
                         definition=definition,
                         sources=[document.source],
                         priority=document.priority,
-                    )
-                elif document.priority == existing.priority:
-                    raise SchemaCompileError(
-                        f"Conflicting {kind} definition for {identifier!r} at equal priority "
-                        f"{document.priority} ({existing.sources[0].source_id} vs {document.source.source_id})."
-                    )
-                else:
-                    self._warn_discarded(
-                        kind,
-                        identifier,
-                        loser=document.source,
-                        loser_priority=document.priority,
-                        winner=existing.sources[0],
-                        winner_priority=existing.priority,
-                    )
+                    ),
+                )
         return tracked
 
+    def _resolve_priority(
+        self,
+        kind: str,
+        identifier: str,
+        *,
+        existing: _Tracked,
+        incoming: _Tracked,
+    ) -> _Tracked:
+        """Resolve two competing definitions for one identifier by priority.
+
+        Single responsibility: decide which of two :class:`_Tracked` entries for
+        the same identifier survives.
+
+        * Identical definitions merge their sources (both files contributed the
+          same thing).
+        * Otherwise higher priority wins; the loser is warned about (ADR 0017
+          §4) so a valid-but-ineffective file does not look like it took effect.
+        * Equal priority with disagreeing definitions is unresolvable and raises
+          :class:`SchemaCompileError` so deployment stops before the database
+          changes.
+
+        Each ``_Tracked`` carries a single source; ``existing`` and ``incoming``
+        are symmetric, so this also serves conflicts between extension
+        contributions where only priority (not load order) decides the winner.
+        """
+        if existing.definition == incoming.definition:
+            existing.sources.extend(incoming.sources)
+            return existing
+        if incoming.priority == existing.priority:
+            raise SchemaCompileError(
+                f"Conflicting {kind} definition for {identifier!r} at equal priority "
+                f"{incoming.priority} "
+                f"({existing.sources[0].source_id} vs {incoming.sources[0].source_id})."
+            )
+        winner, loser = (
+            (incoming, existing) if incoming.priority > existing.priority else (existing, incoming)
+        )
+        self._warn_discarded(
+            kind,
+            identifier,
+            loser=loser.sources[0],
+            loser_priority=loser.priority,
+            winner=winner.sources[0],
+            winner_priority=winner.priority,
+        )
+        return winner
+
     @staticmethod
     def _warn_discarded(
         kind: str,
@@ -161,8 +184,6 @@ def _warn_discarded(
             winner_priority,
         )
 
-    # ---- roles + provenance ----------------------------------------------
-
     def _resolve_roles_and_provenance(
         self, roles: dict[str, _Tracked], documents: list[SchemaDocument]
     ) -> dict[tuple[str, str], list[RelationshipSource]]:
@@ -234,31 +255,67 @@ def _gather_extension_changes(self, roles: dict[str, _Tracked], documents: list[
                     pc["remove"].append((perm, document.priority, document.source))
         return metadata_changes, perm_changes
 
+    def _resolve_contributions(
+        self,
+        identifier: str,
+        entries: list[tuple[object, int, SourceRecord]],
+        *,
+        loser_kind,
+        on_tie,
+    ) -> tuple[object, int, list[SourceRecord]]:
+        """Pick the winning value among competing extension contributions.
+
+        The batch counterpart to :meth:`_resolve_priority`: where that method
+        decides between two base definitions pairwise, this decides among any
+        number of ``(value, priority, source)`` contributions to the same
+        extension field or permission.
+
+        The highest priority wins. Every contribution in the top-priority group
+        must agree; if they do not, ``on_tie(top_values)`` builds the message
+        for a :class:`SchemaCompileError` (the caller knows how to describe its
+        own conflict). Lower-priority contributions are warned about so a
+        valid-but-ineffective file is not mistaken for one that took effect
+        (ADR 0017 §4); ``loser_kind(value)`` labels each loser so the warning
+        can name what that contribution tried to do.
+
+        Returns the winning value, the winning priority, and every source in the
+        winning group (for provenance).
+        """
+        max_priority = max(priority for _, priority, _ in entries)
+        top_values = {value for value, priority, _ in entries if priority == max_priority}
+        if len(top_values) > 1:
+            raise SchemaCompileError(on_tie(top_values))
+
+        winner_value = next(iter(top_values))
+        winning_sources = [src for value, priority, src in entries if priority == max_priority]
+        for value, priority, src in entries:
+            if priority < max_priority:
+                self._warn_discarded(
+                    loser_kind(value),
+                    identifier,
+                    loser=src,
+                    loser_priority=priority,
+                    winner=winning_sources[0],
+                    winner_priority=max_priority,
+                )
+        return winner_value, max_priority, winning_sources
+
     def _resolve_metadata(self, role_id: str, md: dict[str, list[tuple[object, int, SourceRecord]]]):
         """Pick winning metadata values by priority; error on equal-priority ties."""
         new_values: dict[str, object] = {}
         contributing: set[SourceRecord] = set()
         for field_name, entries in md.items():
-            max_priority = max(priority for _, priority, _ in entries)
-            top_values = {value for value, priority, _ in entries if priority == max_priority}
-            if len(top_values) > 1:
-                raise SchemaCompileError(
+            value, _, winning_sources = self._resolve_contributions(
+                role_id,
+                entries,
+                loser_kind=lambda _value: f"role_extension {field_name}",
+                on_tie=lambda top: (
                     f"Conflicting {field_name!r} for role {role_id!r} at equal priority "
-                    f"{max_priority}: {sorted(map(str, top_values))}."
-                )
-            new_values[field_name] = next(iter(top_values))
-            contributing.update(src for _, priority, src in entries if priority == max_priority)
-            winner = next(src for _, priority, src in entries if priority == max_priority)
-            for _, priority, src in entries:
-                if priority < max_priority:
-                    self._warn_discarded(
-                        f"role_extension {field_name}",
-                        role_id,
-                        loser=src,
-                        loser_priority=priority,
-                        winner=winner,
-                        winner_priority=max_priority,
-                    )
+                    f"{max(p for _, p, _ in entries)}: {sorted(map(str, top))}."
+                ),
+            )
+            new_values[field_name] = value
+            contributing.update(winning_sources)
         return new_values, contributing
 
     def _resolve_permissions(
@@ -288,26 +345,15 @@ def _resolve_permissions(
             actions.setdefault(perm, []).append(("remove", priority, src))
 
         for perm, entries in actions.items():
-            max_priority = max(priority for _, priority, _ in entries)
-            top = {action for action, priority, _ in entries if priority == max_priority}
-            if len(top) > 1:
-                raise SchemaCompileError(
+            action, max_priority, winning_sources = self._resolve_contributions(
+                role_id,
+                entries,
+                loser_kind=lambda act, _perm=perm: f"role_extension {act} of {_perm!r} on role",
+                on_tie=lambda _top: (
                     f"Conflicting add/remove for permission {perm!r} on role {role_id!r} "
-                    f"at equal priority {max_priority}."
-                )
-            action = next(iter(top))
-            winning_sources = [src for act, priority, src in entries if priority == max_priority and act == action]
-
-            for act, priority, src in entries:
-                if priority < max_priority:
-                    self._warn_discarded(
-                        f"role_extension {act} of {perm!r} on role",
-                        role_id,
-                        loser=src,
-                        loser_priority=priority,
-                        winner=winning_sources[0],
-                        winner_priority=max_priority,
-                    )
+                    f"at equal priority {max(p for _, p, _ in entries)}."
+                ),
+            )
 
             if action == "add":
                 if perm in current:
@@ -325,8 +371,6 @@ def _resolve_permissions(
 
         return tuple(sorted(current)), provenance
 
-    # ---- finalize ---------------------------------------------------------
-
     def _finalize(self, tracked: dict[str, _Tracked]) -> dict[str, CompiledDefinition]:
         """Turn tracked definitions into CompiledDefinition entries."""
         return {

From dec9b9084e5b800a7dfeb0da7fcc3efd9a2f8356 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Rodrigo=20M=C3=A9ndez?= <rodrigo.mendez@wgu.edu>
Date: Fri, 2 Oct 2026 12:58:37 -0600
Subject: [PATCH 04/10] squash!: Refactor var names

---
 .../engine/schema/compilation.py              | 50 +++++++++++--------
 1 file changed, 28 insertions(+), 22 deletions(-)

diff --git a/src/openedx_authz/engine/schema/compilation.py b/src/openedx_authz/engine/schema/compilation.py
index 8bef2edb..7997a5ad 100644
--- a/src/openedx_authz/engine/schema/compilation.py
+++ b/src/openedx_authz/engine/schema/compilation.py
@@ -193,8 +193,8 @@ def _resolve_roles_and_provenance(
         ``role_extensions`` (metadata replacement + permission add/remove),
         honoring priority. Returns the relationship provenance map.
         """
-        metadata_changes, perm_changes = self._gather_extension_changes(roles, documents)
-        rp_sources: dict[tuple[str, str], list[RelationshipSource]] = {}
+        metadata_changes, permission_changes = self._gather_extension_changes(roles, documents)
+        role_permission_sources: dict[tuple[str, str], list[RelationshipSource]] = {}
 
         for role_id, tracked in roles.items():
             role: RoleDefinition = tracked.definition
@@ -207,26 +207,28 @@ def _resolve_roles_and_provenance(
                 for perm in role.permissions
             }
 
-            md = metadata_changes.get(role_id, {})
-            if md:
-                new_values, contributing_sources = self._resolve_metadata(role_id, md)
+            metadata_changes_for_role = metadata_changes.get(role_id, {})
+            if metadata_changes_for_role:
+                new_values, contributing_sources = self._resolve_metadata(role_id, metadata_changes_for_role)
                 tracked.definition = replace(role, **new_values)
                 role = tracked.definition
                 for src in contributing_sources:
                     if src not in tracked.sources:
                         tracked.sources.append(src)
 
-            pc = perm_changes.get(role_id)
-            if pc and (pc["add"] or pc["remove"]):
+            permission_changes_for_role = permission_changes.get(role_id)
+            if permission_changes_for_role and (
+                permission_changes_for_role["add"] or permission_changes_for_role["remove"]
+            ):
                 final_perms, provenance = self._resolve_permissions(
-                    role_id, role.permissions, base_sources, base_priority, pc
+                    role_id, role.permissions, base_sources, base_priority, permission_changes_for_role
                 )
                 tracked.definition = replace(tracked.definition, permissions=final_perms)
 
             for perm, sources in provenance.items():
-                rp_sources[(role_id, perm)] = sources
+                role_permission_sources[(role_id, perm)] = sources
 
-        return rp_sources
+        return role_permission_sources
 
     def _gather_extension_changes(self, roles: dict[str, _Tracked], documents: list[SchemaDocument]):
         """Collect per-role metadata and permission changes from all extensions.
@@ -235,7 +237,7 @@ def _gather_extension_changes(self, roles: dict[str, _Tracked], documents: list[
         and conflict resolution have everything they need.
         """
         metadata_changes: dict[str, dict[str, list[tuple[object, int, SourceRecord]]]] = {}
-        perm_changes: dict[str, dict[str, list[tuple[str, int, SourceRecord]]]] = {}
+        permission_changes: dict[str, dict[str, list[tuple[str, int, SourceRecord]]]] = {}
 
         for document in documents:
             for extension in document.role_extensions:
@@ -243,17 +245,19 @@ def _gather_extension_changes(self, roles: dict[str, _Tracked], documents: list[
                 if role_id not in roles:
                     # Validation already errors on this; skip defensively.
                     continue
-                md = metadata_changes.setdefault(role_id, {})
+                metadata_changes_for_role = metadata_changes.setdefault(role_id, {})
                 for field_name in _METADATA_FIELDS:
                     value = getattr(extension, field_name)
                     if value is not None:
-                        md.setdefault(field_name, []).append((value, document.priority, document.source))
-                pc = perm_changes.setdefault(role_id, {"add": [], "remove": []})
+                        metadata_changes_for_role.setdefault(field_name, []).append(
+                            (value, document.priority, document.source)
+                        )
+                permission_changes_for_role = permission_changes.setdefault(role_id, {"add": [], "remove": []})
                 for perm in extension.add_permissions:
-                    pc["add"].append((perm, document.priority, document.source))
+                    permission_changes_for_role["add"].append((perm, document.priority, document.source))
                 for perm in extension.remove_permissions:
-                    pc["remove"].append((perm, document.priority, document.source))
-        return metadata_changes, perm_changes
+                    permission_changes_for_role["remove"].append((perm, document.priority, document.source))
+        return metadata_changes, permission_changes
 
     def _resolve_contributions(
         self,
@@ -300,11 +304,13 @@ def _resolve_contributions(
                 )
         return winner_value, max_priority, winning_sources
 
-    def _resolve_metadata(self, role_id: str, md: dict[str, list[tuple[object, int, SourceRecord]]]):
+    def _resolve_metadata(
+        self, role_id: str, metadata_changes: dict[str, list[tuple[object, int, SourceRecord]]]
+    ):
         """Pick winning metadata values by priority; error on equal-priority ties."""
         new_values: dict[str, object] = {}
         contributing: set[SourceRecord] = set()
-        for field_name, entries in md.items():
+        for field_name, entries in metadata_changes.items():
             value, _, winning_sources = self._resolve_contributions(
                 role_id,
                 entries,
@@ -324,7 +330,7 @@ def _resolve_permissions(
         base: tuple[str, ...],
         base_sources: list[SourceRecord],
         base_priority: int,
-        pc: dict[str, list[tuple[str, int, SourceRecord]]],
+        permission_changes: dict[str, list[tuple[str, int, SourceRecord]]],
     ):
         """Apply add/remove per permission, returning (final_perms, provenance).
 
@@ -339,9 +345,9 @@ def _resolve_permissions(
         }
 
         actions: dict[str, list[tuple[str, int, SourceRecord]]] = {}
-        for perm, priority, src in pc["add"]:
+        for perm, priority, src in permission_changes["add"]:
             actions.setdefault(perm, []).append(("add", priority, src))
-        for perm, priority, src in pc["remove"]:
+        for perm, priority, src in permission_changes["remove"]:
             actions.setdefault(perm, []).append(("remove", priority, src))
 
         for perm, entries in actions.items():

From d13d42a2f3a3a8bd24a58618601f5e1754f071e4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Rodrigo=20M=C3=A9ndez?= <rodrigo.mendez@wgu.edu>
Date: Fri, 2 Oct 2026 13:18:24 -0600
Subject: [PATCH 05/10] squash!: Refactor _Tracked to be immutable

---
 .../engine/schema/compilation.py              | 129 ++++++++++++------
 1 file changed, 90 insertions(+), 39 deletions(-)

diff --git a/src/openedx_authz/engine/schema/compilation.py b/src/openedx_authz/engine/schema/compilation.py
index 7997a5ad..c39a3d53 100644
--- a/src/openedx_authz/engine/schema/compilation.py
+++ b/src/openedx_authz/engine/schema/compilation.py
@@ -24,7 +24,7 @@
 from __future__ import annotations
 
 import logging
-from dataclasses import dataclass, field, replace
+from dataclasses import dataclass, replace
 
 from openedx_authz.constants import SchemaOriginKind
 from openedx_authz.engine.schema.exceptions import SchemaCompileError
@@ -46,12 +46,16 @@
 _KIND_LABELS = {"categories": "category", "permissions": "permission", "roles": "role"}
 
 
-@dataclass
+@dataclass(frozen=True)
 class _Tracked:
-    """A base definition plus the sources and priority that produced it."""
+    """A base definition plus the sources and priority that produced it.
+
+    Immutable: resolution steps return a new ``_Tracked`` via
+    :func:`dataclasses.replace` rather than mutating one in place.
+    """
 
     definition: object
-    sources: list[SourceRecord] = field(default_factory=list)
+    sources: tuple[SourceRecord, ...] = ()
     priority: int = 0
 
 
@@ -70,12 +74,12 @@ def compile(self, documents: list[SchemaDocument]) -> CompiledSchema:
         permissions = self._collect(documents, "permissions", key=lambda p: p.identifier)
         roles = self._collect(documents, "roles", key=lambda r: r.id)
 
-        role_permission_sources = self._resolve_roles_and_provenance(roles, documents)
+        resolved_roles, role_permission_sources = self._resolve_roles_and_provenance(roles, documents)
 
         return CompiledSchema(
             categories=self._finalize(categories),
             permissions=self._finalize(permissions),
-            roles=self._finalize(roles),
+            roles=self._finalize(resolved_roles),
             role_permission_sources=role_permission_sources,
         )
 
@@ -93,7 +97,7 @@ def _collect(self, documents: list[SchemaDocument], attr: str, key) -> dict[str,
                 if existing is None:
                     tracked[identifier] = _Tracked(
                         definition=definition,
-                        sources=[document.source],
+                        sources=(document.source,),
                         priority=document.priority,
                     )
                     continue
@@ -105,7 +109,7 @@ def _collect(self, documents: list[SchemaDocument], attr: str, key) -> dict[str,
                     existing=existing,
                     incoming=_Tracked(
                         definition=definition,
-                        sources=[document.source],
+                        sources=(document.source,),
                         priority=document.priority,
                     ),
                 )
@@ -137,8 +141,7 @@ def _resolve_priority(
         contributions where only priority (not load order) decides the winner.
         """
         if existing.definition == incoming.definition:
-            existing.sources.extend(incoming.sources)
-            return existing
+            return replace(existing, sources=existing.sources + incoming.sources)
         if incoming.priority == existing.priority:
             raise SchemaCompileError(
                 f"Conflicting {kind} definition for {identifier!r} at equal priority "
@@ -186,49 +189,97 @@ def _warn_discarded(
 
     def _resolve_roles_and_provenance(
         self, roles: dict[str, _Tracked], documents: list[SchemaDocument]
-    ) -> dict[tuple[str, str], list[RelationshipSource]]:
+    ) -> tuple[dict[str, _Tracked], dict[tuple[str, str], list[RelationshipSource]]]:
         """Apply extensions and build per-(role, permission) provenance.
 
         Seeds base provenance from each role's own definition, then folds in
         ``role_extensions`` (metadata replacement + permission add/remove),
-        honoring priority. Returns the relationship provenance map.
+        honoring priority. Returns the resolved roles (a new mapping; inputs are
+        left untouched) alongside the relationship provenance map.
         """
         metadata_changes, permission_changes = self._gather_extension_changes(roles, documents)
+        resolved_roles: dict[str, _Tracked] = {}
         role_permission_sources: dict[tuple[str, str], list[RelationshipSource]] = {}
 
         for role_id, tracked in roles.items():
-            role: RoleDefinition = tracked.definition
-            base_sources = list(tracked.sources)
+            base_sources = tracked.sources
             base_priority = tracked.priority
 
-            # Seed base provenance for every permission the role declares.
-            provenance: dict[str, list[RelationshipSource]] = {
-                perm: [RelationshipSource(src, SchemaOriginKind.BASE, base_priority) for src in base_sources]
-                for perm in role.permissions
-            }
-
-            metadata_changes_for_role = metadata_changes.get(role_id, {})
-            if metadata_changes_for_role:
-                new_values, contributing_sources = self._resolve_metadata(role_id, metadata_changes_for_role)
-                tracked.definition = replace(role, **new_values)
-                role = tracked.definition
-                for src in contributing_sources:
-                    if src not in tracked.sources:
-                        tracked.sources.append(src)
-
-            permission_changes_for_role = permission_changes.get(role_id)
-            if permission_changes_for_role and (
-                permission_changes_for_role["add"] or permission_changes_for_role["remove"]
-            ):
-                final_perms, provenance = self._resolve_permissions(
-                    role_id, role.permissions, base_sources, base_priority, permission_changes_for_role
-                )
-                tracked.definition = replace(tracked.definition, permissions=final_perms)
+            tracked = self._apply_metadata_changes(role_id, tracked, metadata_changes.get(role_id, {}))
+
+            tracked, provenance = self._apply_permission_changes(
+                role_id,
+                tracked,
+                base_sources,
+                base_priority,
+                permission_changes.get(role_id),
+            )
 
+            resolved_roles[role_id] = tracked
             for perm, sources in provenance.items():
                 role_permission_sources[(role_id, perm)] = sources
 
-        return role_permission_sources
+        return resolved_roles, role_permission_sources
+
+    def _apply_metadata_changes(
+        self,
+        role_id: str,
+        tracked: _Tracked,
+        metadata_changes_for_role: dict[str, list[tuple[object, int, SourceRecord]]],
+    ) -> _Tracked:
+        """Return the role with winning metadata applied and its sources recorded.
+
+        Single responsibility: resolve the winning metadata values for this role
+        and produce a new :class:`_Tracked` carrying the updated definition and
+        the sources that contributed them. Returns ``tracked`` unchanged when the
+        role has no metadata extensions.
+        """
+        if not metadata_changes_for_role:
+            return tracked
+        new_values, contributing_sources = self._resolve_metadata(role_id, metadata_changes_for_role)
+        merged_sources = tracked.sources + tuple(
+            src for src in contributing_sources if src not in tracked.sources
+        )
+        return replace(
+            tracked,
+            definition=replace(tracked.definition, **new_values),
+            sources=merged_sources,
+        )
+
+    def _apply_permission_changes(
+        self,
+        role_id: str,
+        tracked: _Tracked,
+        base_sources: tuple[SourceRecord, ...],
+        base_priority: int,
+        permission_changes_for_role: dict[str, list[tuple[str, int, SourceRecord]]] | None,
+    ) -> tuple[_Tracked, dict[str, list[RelationshipSource]]]:
+        """Return the role with permission changes applied and its provenance.
+
+        Single responsibility: own the per-permission provenance for this role.
+        Seeds base provenance from the role's declared permissions, then, if the
+        role has permission extensions, resolves the final permission set,
+        produces a new :class:`_Tracked`, and reflects the add/remove in
+        provenance. Returns ``(tracked, base_provenance)`` unchanged when the
+        role has no permission extensions.
+        """
+        base_provenance: dict[str, list[RelationshipSource]] = {
+            perm: [RelationshipSource(src, SchemaOriginKind.BASE, base_priority) for src in base_sources]
+            for perm in tracked.definition.permissions
+        }
+        if not permission_changes_for_role or not (
+            permission_changes_for_role["add"] or permission_changes_for_role["remove"]
+        ):
+            return tracked, base_provenance
+        final_perms, provenance = self._resolve_permissions(
+            role_id,
+            tracked.definition.permissions,
+            base_sources,
+            base_priority,
+            permission_changes_for_role,
+        )
+        tracked = replace(tracked, definition=replace(tracked.definition, permissions=final_perms))
+        return tracked, provenance
 
     def _gather_extension_changes(self, roles: dict[str, _Tracked], documents: list[SchemaDocument]):
         """Collect per-role metadata and permission changes from all extensions.
@@ -383,7 +434,7 @@ def _finalize(self, tracked: dict[str, _Tracked]) -> dict[str, CompiledDefinitio
             identifier: CompiledDefinition(
                 key=identifier,
                 definition=entry.definition,
-                sources=tuple(entry.sources),
+                sources=entry.sources,
             )
             for identifier, entry in tracked.items()
         }

From 9012cce24d568b6d4253f854939add45345c5453 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Rodrigo=20M=C3=A9ndez?= <rodrigo.mendez@wgu.edu>
Date: Fri, 2 Oct 2026 13:25:41 -0600
Subject: [PATCH 06/10] squash!: RoleMetadataField

---
 src/openedx_authz/engine/schema/compilation.py | 14 +++++++++++---
 1 file changed, 11 insertions(+), 3 deletions(-)

diff --git a/src/openedx_authz/engine/schema/compilation.py b/src/openedx_authz/engine/schema/compilation.py
index c39a3d53..1a3b2008 100644
--- a/src/openedx_authz/engine/schema/compilation.py
+++ b/src/openedx_authz/engine/schema/compilation.py
@@ -25,6 +25,7 @@
 
 import logging
 from dataclasses import dataclass, replace
+from enum import Enum
 
 from openedx_authz.constants import SchemaOriginKind
 from openedx_authz.engine.schema.exceptions import SchemaCompileError
@@ -39,8 +40,14 @@
 
 logger = logging.getLogger(__name__)
 
-# Metadata fields an extension may replace on a role.
-_METADATA_FIELDS = ("display_name", "description", "icon", "hidden")
+class RoleMetadataField(str, Enum):
+    """Metadata fields a ``RoleExtension`` may replace on a role (ADR 0023)."""
+
+    DISPLAY_NAME = "display_name"
+    DESCRIPTION = "description"
+    ICON = "icon"
+    HIDDEN = "hidden"
+
 
 # Singular labels for operator-facing messages, keyed by document attribute.
 _KIND_LABELS = {"categories": "category", "permissions": "permission", "roles": "role"}
@@ -297,7 +304,8 @@ def _gather_extension_changes(self, roles: dict[str, _Tracked], documents: list[
                     # Validation already errors on this; skip defensively.
                     continue
                 metadata_changes_for_role = metadata_changes.setdefault(role_id, {})
-                for field_name in _METADATA_FIELDS:
+                for field in RoleMetadataField:
+                    field_name = field.value
                     value = getattr(extension, field_name)
                     if value is not None:
                         metadata_changes_for_role.setdefault(field_name, []).append(

From db0d20bccdc810b141d839a7e1ea3107e9b4898f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Rodrigo=20M=C3=A9ndez?= <rodrigo.mendez@wgu.edu>
Date: Fri, 2 Oct 2026 16:35:30 -0600
Subject: [PATCH 07/10] squash!: Make sure RoleMetadataField is a subset of
 RoleExtension fields

---
 .../tests/schema/test_compilation.py          | 25 ++++++++++++++++++-
 1 file changed, 24 insertions(+), 1 deletion(-)

diff --git a/src/openedx_authz/tests/schema/test_compilation.py b/src/openedx_authz/tests/schema/test_compilation.py
index 6285c945..d23ee835 100644
--- a/src/openedx_authz/tests/schema/test_compilation.py
+++ b/src/openedx_authz/tests/schema/test_compilation.py
@@ -6,12 +6,14 @@
 """
 
 import logging
+from dataclasses import fields
 
 import pytest
 
 from openedx_authz.constants import SchemaOriginKind
-from openedx_authz.engine.schema.compilation import SchemaCompiler
+from openedx_authz.engine.schema.compilation import RoleMetadataField, SchemaCompiler
 from openedx_authz.engine.schema.exceptions import SchemaCompileError
+from openedx_authz.engine.schema.types import RoleExtension
 
 from .factories import category, extension, make_document, make_source, permission, role
 
@@ -32,6 +34,27 @@ def _base(**role_kwargs):
     )
 
 
+class TestRoleMetadataFieldInvariant:
+    """``RoleMetadataField`` must stay in lockstep with ``RoleExtension``.
+
+    ``_gather_extension_changes`` reads each member via ``getattr`` with no
+    default, so a member naming a field that ``RoleExtension`` does not declare
+    would raise at compile time. This guards that coupling at the enum level,
+    turning a rename drift into a fast, obvious test failure.
+    """
+
+    def test_every_member_is_a_role_extension_field(self):
+        """Each enum value names a real ``RoleExtension`` field."""
+        extension_fields = {f.name for f in fields(RoleExtension)}
+        enum_values = {member.value for member in RoleMetadataField}
+        assert enum_values <= extension_fields
+
+    def test_members_exclude_non_metadata_fields(self):
+        """Permission and identity fields are not metadata an extension replaces."""
+        enum_values = {member.value for member in RoleMetadataField}
+        assert enum_values.isdisjoint({"role_id", "add_permissions", "remove_permissions"})
+
+
 class TestBaseCompilation:
     """Compiling base definitions with no extensions applied."""
 

From 768d36aab7787cafbf0f9b4a62454aae76c02adf Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Rodrigo=20M=C3=A9ndez?= <rodrigo.mendez@wgu.edu>
Date: Fri, 2 Oct 2026 16:50:34 -0600
Subject: [PATCH 08/10] squash!: refactor _seed_base_provenance

---
 .../engine/schema/compilation.py              | 31 +++++++++++++------
 1 file changed, 22 insertions(+), 9 deletions(-)

diff --git a/src/openedx_authz/engine/schema/compilation.py b/src/openedx_authz/engine/schema/compilation.py
index 1a3b2008..afd72018 100644
--- a/src/openedx_authz/engine/schema/compilation.py
+++ b/src/openedx_authz/engine/schema/compilation.py
@@ -253,6 +253,23 @@ def _apply_metadata_changes(
             sources=merged_sources,
         )
 
+    @staticmethod
+    def _seed_base_provenance(
+        permissions: tuple[str, ...],
+        base_sources: tuple[SourceRecord, ...],
+        base_priority: int,
+    ) -> dict[str, list[RelationshipSource]]:
+        """Attribute each of a role's declared permissions to its base sources.
+
+        Every permission the role declares in its own definition is a ``BASE``
+        grant from each contributing source. Extension-driven add/remove layers
+        on top of this seed.
+        """
+        return {
+            perm: [RelationshipSource(src, SchemaOriginKind.BASE, base_priority) for src in base_sources]
+            for perm in permissions
+        }
+
     def _apply_permission_changes(
         self,
         role_id: str,
@@ -270,10 +287,9 @@ def _apply_permission_changes(
         provenance. Returns ``(tracked, base_provenance)`` unchanged when the
         role has no permission extensions.
         """
-        base_provenance: dict[str, list[RelationshipSource]] = {
-            perm: [RelationshipSource(src, SchemaOriginKind.BASE, base_priority) for src in base_sources]
-            for perm in tracked.definition.permissions
-        }
+        base_provenance = self._seed_base_provenance(
+            tracked.definition.permissions, base_sources, base_priority
+        )
         if not permission_changes_for_role or not (
             permission_changes_for_role["add"] or permission_changes_for_role["remove"]
         ):
@@ -387,7 +403,7 @@ def _resolve_permissions(
         self,
         role_id: str,
         base: tuple[str, ...],
-        base_sources: list[SourceRecord],
+        base_sources: tuple[SourceRecord, ...],
         base_priority: int,
         permission_changes: dict[str, list[tuple[str, int, SourceRecord]]],
     ):
@@ -398,10 +414,7 @@ def _resolve_permissions(
         added permissions.
         """
         current = set(base)
-        provenance: dict[str, list[RelationshipSource]] = {
-            perm: [RelationshipSource(src, SchemaOriginKind.BASE, base_priority) for src in base_sources]
-            for perm in base
-        }
+        provenance = self._seed_base_provenance(base, base_sources, base_priority)
 
         actions: dict[str, list[tuple[str, int, SourceRecord]]] = {}
         for perm, priority, src in permission_changes["add"]:

From 63a05323a15772eb87969f15179f9bfbc9944a04 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Rodrigo=20M=C3=A9ndez?= <rodrigo.mendez@wgu.edu>
Date: Fri, 2 Oct 2026 16:55:32 -0600
Subject: [PATCH 09/10] squash!: Fix lint issues

---
 src/openedx_authz/engine/schema/compilation.py | 15 +++++++--------
 1 file changed, 7 insertions(+), 8 deletions(-)

diff --git a/src/openedx_authz/engine/schema/compilation.py b/src/openedx_authz/engine/schema/compilation.py
index afd72018..3e25adaa 100644
--- a/src/openedx_authz/engine/schema/compilation.py
+++ b/src/openedx_authz/engine/schema/compilation.py
@@ -33,7 +33,6 @@
     CompiledDefinition,
     CompiledSchema,
     RelationshipSource,
-    RoleDefinition,
     SchemaDocument,
     SourceRecord,
 )
@@ -389,10 +388,10 @@ def _resolve_metadata(
             value, _, winning_sources = self._resolve_contributions(
                 role_id,
                 entries,
-                loser_kind=lambda _value: f"role_extension {field_name}",
-                on_tie=lambda top: (
-                    f"Conflicting {field_name!r} for role {role_id!r} at equal priority "
-                    f"{max(p for _, p, _ in entries)}: {sorted(map(str, top))}."
+                loser_kind=lambda _value, _field=field_name: f"role_extension {_field}",
+                on_tie=lambda top, _field=field_name, _entries=entries: (
+                    f"Conflicting {_field!r} for role {role_id!r} at equal priority "
+                    f"{max(p for _, p, _ in _entries)}: {sorted(map(str, top))}."
                 ),
             )
             new_values[field_name] = value
@@ -427,9 +426,9 @@ def _resolve_permissions(
                 role_id,
                 entries,
                 loser_kind=lambda act, _perm=perm: f"role_extension {act} of {_perm!r} on role",
-                on_tie=lambda _top: (
-                    f"Conflicting add/remove for permission {perm!r} on role {role_id!r} "
-                    f"at equal priority {max(p for _, p, _ in entries)}."
+                on_tie=lambda _top, _perm=perm, _entries=entries: (
+                    f"Conflicting add/remove for permission {_perm!r} on role {role_id!r} "
+                    f"at equal priority {max(p for _, p, _ in _entries)}."
                 ),
             )
 

From 31fbae7d62624c946dd40c965192d541ab286cc8 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Rodrigo=20M=C3=A9ndez?= <rodrigo.mendez@wgu.edu>
Date: Fri, 2 Oct 2026 17:02:31 -0600
Subject: [PATCH 10/10] squash!: Refactor _gather_extension_changes

---
 .../engine/schema/compilation.py              | 74 +++++++++++++------
 1 file changed, 53 insertions(+), 21 deletions(-)

diff --git a/src/openedx_authz/engine/schema/compilation.py b/src/openedx_authz/engine/schema/compilation.py
index 3e25adaa..9e012b3b 100644
--- a/src/openedx_authz/engine/schema/compilation.py
+++ b/src/openedx_authz/engine/schema/compilation.py
@@ -306,32 +306,64 @@ def _apply_permission_changes(
     def _gather_extension_changes(self, roles: dict[str, _Tracked], documents: list[SchemaDocument]):
         """Collect per-role metadata and permission changes from all extensions.
 
+        Thin orchestrator over the two independent concerns; see
+        :meth:`_gather_metadata_changes` and :meth:`_gather_permission_changes`.
+        """
+        return (
+            self._gather_metadata_changes(roles, documents),
+            self._gather_permission_changes(roles, documents),
+        )
+
+    @staticmethod
+    def _extensions_for_known_roles(roles: dict[str, _Tracked], documents: list[SchemaDocument]):
+        """Yield ``(document, extension)`` for extensions targeting a known role.
+
+        Extensions naming an unknown role are skipped defensively; validation is
+        expected to have already errored on them.
+        """
+        for document in documents:
+            for extension in document.role_extensions:
+                if extension.role_id in roles:
+                    yield document, extension
+
+    def _gather_metadata_changes(
+        self, roles: dict[str, _Tracked], documents: list[SchemaDocument]
+    ) -> dict[str, dict[str, list[tuple[object, int, SourceRecord]]]]:
+        """Collect per-role metadata field contributions from all extensions.
+
         Entries carry the full :class:`SourceRecord` and priority so provenance
         and conflict resolution have everything they need.
         """
         metadata_changes: dict[str, dict[str, list[tuple[object, int, SourceRecord]]]] = {}
-        permission_changes: dict[str, dict[str, list[tuple[str, int, SourceRecord]]]] = {}
+        for document, extension in self._extensions_for_known_roles(roles, documents):
+            metadata_changes_for_role = metadata_changes.setdefault(extension.role_id, {})
+            for field in RoleMetadataField:
+                field_name = field.value
+                value = getattr(extension, field_name)
+                if value is not None:
+                    metadata_changes_for_role.setdefault(field_name, []).append(
+                        (value, document.priority, document.source)
+                    )
+        return metadata_changes
 
-        for document in documents:
-            for extension in document.role_extensions:
-                role_id = extension.role_id
-                if role_id not in roles:
-                    # Validation already errors on this; skip defensively.
-                    continue
-                metadata_changes_for_role = metadata_changes.setdefault(role_id, {})
-                for field in RoleMetadataField:
-                    field_name = field.value
-                    value = getattr(extension, field_name)
-                    if value is not None:
-                        metadata_changes_for_role.setdefault(field_name, []).append(
-                            (value, document.priority, document.source)
-                        )
-                permission_changes_for_role = permission_changes.setdefault(role_id, {"add": [], "remove": []})
-                for perm in extension.add_permissions:
-                    permission_changes_for_role["add"].append((perm, document.priority, document.source))
-                for perm in extension.remove_permissions:
-                    permission_changes_for_role["remove"].append((perm, document.priority, document.source))
-        return metadata_changes, permission_changes
+    def _gather_permission_changes(
+        self, roles: dict[str, _Tracked], documents: list[SchemaDocument]
+    ) -> dict[str, dict[str, list[tuple[str, int, SourceRecord]]]]:
+        """Collect per-role permission add/remove contributions from all extensions.
+
+        Entries carry the full :class:`SourceRecord` and priority so provenance
+        and conflict resolution have everything they need.
+        """
+        permission_changes: dict[str, dict[str, list[tuple[str, int, SourceRecord]]]] = {}
+        for document, extension in self._extensions_for_known_roles(roles, documents):
+            permission_changes_for_role = permission_changes.setdefault(
+                extension.role_id, {"add": [], "remove": []}
+            )
+            for perm in extension.add_permissions:
+                permission_changes_for_role["add"].append((perm, document.priority, document.source))
+            for perm in extension.remove_permissions:
+                permission_changes_for_role["remove"].append((perm, document.priority, document.source))
+        return permission_changes
 
     def _resolve_contributions(
         self,