feat: reconcile governance assignments in bulk

README.md

@@ -63,7 +63,8 @@ This module will own:
 - compatibility FastAPI auth dependency API at `govoplan_access.auth`
 - provider-facing auth facade consumed by sibling modules at `govoplan_core.auth`
 - access administration, tenant provisioning, and governance materializer
-  capabilities
+  capabilities, including the bounded `access.governanceProjection.v1` bulk
+  reconciliation contract used by Admin for idempotent per-assignment outcomes
 - access-owned migrations
 
 The governance-template routes under `/admin/system/governance-templates` are

docs/ACCESS_MODULE_BOUNDARY.md

@@ -180,7 +180,12 @@ admin routers as base routers.
 
 Governance-template metadata CRUD is not access-owned. It is contributed by
 `govoplan-admin`; access only materializes those templates into access-owned
-groups and roles through the `access.governanceMaterializer` capability.
+groups and roles. The compatibility `access.governanceMaterializer` capability
+remains available for single-assignment callers. New Admin orchestration uses
+`access.governanceProjection.v1`: a bounded request of stable template and
+assignment DTOs that bulk-loads managed rows and assignment blockers, applies
+idempotent create/update/remove reconciliation, and returns one provenance-rich
+outcome per assignment. Admin never imports Access ORM models.
 
 The configuration-package Admin routes remain in Access as a compatibility
 surface. Their preflight context is assembled from the active Core registry,

src/govoplan_access/backend/governance_materializer.py

@@ -1,26 +1,186 @@
 from __future__ import annotations
 
+from collections import defaultdict
+
+from sqlalchemy import func
 from sqlalchemy.orm import Session
 
-from govoplan_access.backend.db.models import Group, GroupRoleAssignment, Role, UserGroupMembership, UserRoleAssignment
+from govoplan_access.backend.db.models import (
+    Group,
+    GroupRoleAssignment,
+    Role,
+    UserGroupMembership,
+    UserRoleAssignment,
+    new_uuid,
+)
 from govoplan_core.admin.common import AdminConflictError
-from govoplan_core.core.access import AccessGovernanceMaterializer, GovernanceTemplateMaterialization
+from govoplan_core.core.access import (
+    AccessGovernanceMaterializer,
+    AccessGovernanceProjectionV1,
+    GovernanceProjectionBatch,
+    GovernanceProjectionCommand,
+    GovernanceProjectionOutcome,
+    GovernanceProjectionResult,
+    GovernanceTemplateMaterialization,
+)
 from govoplan_core.core.runtime import get_registry
 
 
-class SqlAccessGovernanceMaterializer(AccessGovernanceMaterializer):
-    def sync_template(self, session: object, template: GovernanceTemplateMaterialization) -> None:
+class SqlAccessGovernanceMaterializer(
+    AccessGovernanceMaterializer,
+    AccessGovernanceProjectionV1,
+):
+    """Reconcile governance projections with a constant number of bulk reads."""
+
+    def reconcile(
+        self,
+        session: object,
+        batch: GovernanceProjectionBatch,
+    ) -> GovernanceProjectionResult:
         db = _session(session)
-        if template.kind == "group":
-            group = (
-                db.query(Group)
-                .filter(Group.tenant_id == template.tenant_id, Group.system_template_id == template.template_id)
-                .first()
+        commands = tuple(batch.commands)
+        group_commands = tuple(item for item in commands if item.template.kind == "group")
+        role_commands = tuple(item for item in commands if item.template.kind == "role")
+
+        groups, duplicate_group_keys = _managed_groups(db, group_commands)
+        roles, duplicate_role_keys = _managed_roles(db, role_commands)
+        used_group_slugs = _used_slugs(db, Group, group_commands)
+        used_role_slugs = _used_slugs(db, Role, role_commands)
+
+        group_ids = {item.id for item in groups.values()}
+        role_ids = {item.id for item in roles.values()}
+        group_memberships = _assignment_counts(db, UserGroupMembership, UserGroupMembership.group_id, group_ids)
+        group_role_links = _assignment_counts(db, GroupRoleAssignment, GroupRoleAssignment.group_id, group_ids)
+        role_user_links = _assignment_counts(db, UserRoleAssignment, UserRoleAssignment.role_id, role_ids)
+        role_group_links = _assignment_counts(db, GroupRoleAssignment, GroupRoleAssignment.role_id, role_ids)
+
+        outcomes: list[GovernanceProjectionOutcome] = []
+        for command in commands:
+            key = (command.template.tenant_id, command.template.template_id)
+            if command.template.kind == "group":
+                outcome = self._reconcile_group(
+                    db,
+                    command,
+                    groups,
+                    duplicate_group_keys,
+                    used_group_slugs,
+                    group_memberships,
+                    group_role_links,
+                    dry_run=batch.dry_run,
+                )
+            else:
+                outcome = self._reconcile_role(
+                    db,
+                    command,
+                    roles,
+                    duplicate_role_keys,
+                    used_role_slugs,
+                    role_user_links,
+                    role_group_links,
+                    dry_run=batch.dry_run,
+                )
+            outcomes.append(outcome)
+            if outcome.status in {"removed", "absent"}:
+                groups.pop(key, None)
+                roles.pop(key, None)
+
+        if not batch.dry_run:
+            db.flush()
+        return GovernanceProjectionResult(
+            operation_id=batch.operation_id,
+            outcomes=tuple(outcomes),
+            dry_run=batch.dry_run,
+        )
+
+    def sync_template(self, session: object, template: GovernanceTemplateMaterialization) -> None:
+        self._legacy_reconcile(session, template, operation="upsert")
+
+    def remove_template(self, session: object, template: GovernanceTemplateMaterialization) -> None:
+        self._legacy_reconcile(session, template, operation="remove")
+
+    def _legacy_reconcile(
+        self,
+        session: object,
+        template: GovernanceTemplateMaterialization,
+        *,
+        operation: str,
+    ) -> None:
+        command = GovernanceProjectionCommand(
+            assignment_id=f"legacy:{template.kind}:{template.template_id}:{template.tenant_id}",
+            operation=operation,  # type: ignore[arg-type]
+            template=template,
+            provenance={"contract": "access.governanceMaterializer"},
+        )
+        result = self.reconcile(
+            session,
+            GovernanceProjectionBatch(
+                operation_id=command.assignment_id,
+                commands=(command,),
+            ),
+        )
+        if result.blocked:
+            raise AdminConflictError(result.blocked[0].message or "Governance projection was blocked.")
+
+    def _reconcile_group(
+        self,
+        db: Session,
+        command: GovernanceProjectionCommand,
+        existing: dict[tuple[str, str], Group],
+        duplicate_keys: set[tuple[str, str]],
+        used_slugs: dict[str, set[str]],
+        membership_counts: dict[str, int],
+        role_counts: dict[str, int],
+        *,
+        dry_run: bool,
+    ) -> GovernanceProjectionOutcome:
+        template = command.template
+        key = (template.tenant_id, template.template_id)
+        group = existing.get(key)
+        if key in duplicate_keys:
+            return _outcome(
+                command,
+                status="failed",
+                blocker_codes=("duplicate_managed_projection",),
+                message="Multiple managed groups exist for this template and tenant.",
             )
+        if command.operation == "remove":
             if group is None:
+                return _outcome(command, status="absent")
+            blockers: list[str] = []
+            if membership_counts.get(group.id, 0):
+                blockers.append("group_has_members")
+            if role_counts.get(group.id, 0):
+                blockers.append("group_has_roles")
+            if blockers:
+                return _outcome(
+                    command,
+                    status="blocked",
+                    resource_id=group.id,
+                    blocker_codes=tuple(blockers),
+                    message=f"Cannot remove {template.name!r} while its managed group has members or roles.",
+                )
+            if not dry_run:
+                try:
+                    _run_delete_vetoes(db, "group", template.tenant_id, group.id)
+                except AdminConflictError as exc:
+                    return _outcome(
+                        command,
+                        status="blocked",
+                        resource_id=group.id,
+                        blocker_codes=("module_delete_veto",),
+                        message=str(exc),
+                    )
+                db.delete(group)
+            return _outcome(command, status="removed", resource_id=group.id)
+
+        if group is None:
+            resource_id = new_uuid()
+            slug = _available_slug(used_slugs[template.tenant_id], template.slug)
+            if not dry_run:
                 group = Group(
+                    id=resource_id,
                     tenant_id=template.tenant_id,
-                    slug=_available_slug(db, Group, template.tenant_id, template.slug),
+                    slug=slug,
                     name=template.name,
                     description=template.description,
                     is_active=template.is_active,
@@ -28,78 +188,199 @@ class SqlAccessGovernanceMaterializer(AccessGovernanceMaterializer):
                     system_required=template.required,
                 )
                 db.add(group)
-            else:
-                group.name = template.name
-                group.description = template.description
-                group.system_required = template.required
-                if template.required:
-                    group.is_active = template.is_active
-            db.flush()
-            return
+                existing[key] = group
+            return _outcome(command, status="created", resource_id=resource_id)
 
-        role = (
-            db.query(Role)
-            .filter(Role.tenant_id == template.tenant_id, Role.system_template_id == template.template_id)
-            .first()
-        )
-        if role is None:
-            role = Role(
-                tenant_id=template.tenant_id,
-                slug=_available_slug(db, Role, template.tenant_id, template.slug),
-                name=template.name,
-                description=template.description,
-                permissions=list(template.permissions),
-                is_builtin=False,
-                is_assignable=template.is_active,
-                system_template_id=template.template_id,
-                system_required=template.required,
-            )
-            db.add(role)
-        else:
-            role.name = template.name
-            role.description = template.description
-            role.permissions = list(template.permissions)
-            role.system_required = template.required
-            if template.required:
-                role.is_assignable = template.is_active
-        db.flush()
+        changes = {
+            "name": template.name,
+            "description": template.description,
+            "system_required": template.required,
+        }
+        if template.required:
+            changes["is_active"] = template.is_active
+        changed = any(getattr(group, field) != value for field, value in changes.items())
+        if changed and not dry_run:
+            for field, value in changes.items():
+                setattr(group, field, value)
+        return _outcome(command, status="updated" if changed else "unchanged", resource_id=group.id)
 
-    def remove_template(self, session: object, template: GovernanceTemplateMaterialization) -> None:
-        db = _session(session)
-        if template.kind == "group":
-            group = (
-                db.query(Group)
-                .filter(Group.tenant_id == template.tenant_id, Group.system_template_id == template.template_id)
-                .first()
+    def _reconcile_role(
+        self,
+        db: Session,
+        command: GovernanceProjectionCommand,
+        existing: dict[tuple[str, str], Role],
+        duplicate_keys: set[tuple[str, str]],
+        used_slugs: dict[str, set[str]],
+        user_counts: dict[str, int],
+        group_counts: dict[str, int],
+        *,
+        dry_run: bool,
+    ) -> GovernanceProjectionOutcome:
+        template = command.template
+        key = (template.tenant_id, template.template_id)
+        role = existing.get(key)
+        if key in duplicate_keys:
+            return _outcome(
+                command,
+                status="failed",
+                blocker_codes=("duplicate_managed_projection",),
+                message="Multiple managed roles exist for this template and tenant.",
             )
-            if group is None:
-                return
-            membership_count = db.query(UserGroupMembership).filter(UserGroupMembership.group_id == group.id).count()
-            role_count = db.query(GroupRoleAssignment).filter(GroupRoleAssignment.group_id == group.id).count()
-            if membership_count or role_count:
-                raise AdminConflictError(
-                    f"Cannot remove {template.name!r} from the tenant while its managed group has members or roles."
+        if command.operation == "remove":
+            if role is None:
+                return _outcome(command, status="absent")
+            blockers: list[str] = []
+            if user_counts.get(role.id, 0):
+                blockers.append("role_has_users")
+            if group_counts.get(role.id, 0):
+                blockers.append("role_has_groups")
+            if blockers:
+                return _outcome(
+                    command,
+                    status="blocked",
+                    resource_id=role.id,
+                    blocker_codes=tuple(blockers),
+                    message=f"Cannot remove {template.name!r} while its managed role is assigned to users or groups.",
                 )
-            _run_delete_vetoes(db, "group", template.tenant_id, group.id)
-            db.delete(group)
-            db.flush()
-            return
+            if not dry_run:
+                db.delete(role)
+            return _outcome(command, status="removed", resource_id=role.id)
 
-        role = (
-            db.query(Role)
-            .filter(Role.tenant_id == template.tenant_id, Role.system_template_id == template.template_id)
-            .first()
-        )
         if role is None:
-            return
-        user_count = db.query(UserRoleAssignment).filter(UserRoleAssignment.role_id == role.id).count()
-        group_count = db.query(GroupRoleAssignment).filter(GroupRoleAssignment.role_id == role.id).count()
-        if user_count or group_count:
-            raise AdminConflictError(
-                f"Cannot remove {template.name!r} from the tenant while its managed role is assigned to users or groups."
-            )
-        db.delete(role)
-        db.flush()
+            resource_id = new_uuid()
+            slug = _available_slug(used_slugs[template.tenant_id], template.slug)
+            if not dry_run:
+                role = Role(
+                    id=resource_id,
+                    tenant_id=template.tenant_id,
+                    slug=slug,
+                    name=template.name,
+                    description=template.description,
+                    permissions=list(template.permissions),
+                    is_builtin=False,
+                    is_assignable=template.is_active,
+                    system_template_id=template.template_id,
+                    system_required=template.required,
+                )
+                db.add(role)
+                existing[key] = role
+            return _outcome(command, status="created", resource_id=resource_id)
+
+        changes: dict[str, object] = {
+            "name": template.name,
+            "description": template.description,
+            "permissions": list(template.permissions),
+            "system_required": template.required,
+        }
+        if template.required:
+            changes["is_assignable"] = template.is_active
+        changed = any(getattr(role, field) != value for field, value in changes.items())
+        if changed and not dry_run:
+            for field, value in changes.items():
+                setattr(role, field, value)
+        return _outcome(command, status="updated" if changed else "unchanged", resource_id=role.id)
+
+
+def _managed_groups(
+    session: Session,
+    commands: tuple[GovernanceProjectionCommand, ...],
+) -> tuple[dict[tuple[str, str], Group], set[tuple[str, str]]]:
+    if not commands:
+        return {}, set()
+    tenants = {item.template.tenant_id for item in commands}
+    templates = {item.template.template_id for item in commands}
+    rows = session.query(Group).filter(
+        Group.tenant_id.in_(tenants),
+        Group.system_template_id.in_(templates),
+    ).all()
+    return _indexed_managed(rows)
+
+
+def _managed_roles(
+    session: Session,
+    commands: tuple[GovernanceProjectionCommand, ...],
+) -> tuple[dict[tuple[str, str], Role], set[tuple[str, str]]]:
+    if not commands:
+        return {}, set()
+    tenants = {item.template.tenant_id for item in commands}
+    templates = {item.template.template_id for item in commands}
+    rows = session.query(Role).filter(
+        Role.tenant_id.in_(tenants),
+        Role.system_template_id.in_(templates),
+    ).all()
+    return _indexed_managed(rows)
+
+
+def _indexed_managed(rows):
+    indexed = {}
+    duplicates = set()
+    for row in rows:
+        key = (row.tenant_id, row.system_template_id)
+        if key in indexed:
+            duplicates.add(key)
+        else:
+            indexed[key] = row
+    return indexed, duplicates
+
+
+def _used_slugs(
+    session: Session,
+    model: type[Group] | type[Role],
+    commands: tuple[GovernanceProjectionCommand, ...],
+) -> dict[str, set[str]]:
+    used: dict[str, set[str]] = defaultdict(set)
+    tenants = {item.template.tenant_id for item in commands}
+    if tenants:
+        for tenant_id, slug in session.query(model.tenant_id, model.slug).filter(model.tenant_id.in_(tenants)).all():
+            used[str(tenant_id)].add(slug)
+    for tenant_id in tenants:
+        used[tenant_id]
+    return used
+
+
+def _assignment_counts(session: Session, model, column, resource_ids: set[str]) -> dict[str, int]:
+    if not resource_ids:
+        return {}
+    return {
+        resource_id: count
+        for resource_id, count in session.query(column, func.count(model.id))
+        .filter(column.in_(resource_ids))
+        .group_by(column)
+        .all()
+    }
+
+
+def _available_slug(used: set[str], base: str) -> str:
+    candidate = base
+    suffix = 2
+    while candidate in used:
+        candidate = f"{base}-{suffix}"
+        suffix += 1
+    used.add(candidate)
+    return candidate
+
+
+def _outcome(
+    command: GovernanceProjectionCommand,
+    *,
+    status: str,
+    resource_id: str | None = None,
+    blocker_codes: tuple[str, ...] = (),
+    message: str | None = None,
+) -> GovernanceProjectionOutcome:
+    template = command.template
+    return GovernanceProjectionOutcome(
+        assignment_id=command.assignment_id,
+        template_id=template.template_id,
+        tenant_id=template.tenant_id,
+        kind=template.kind,
+        operation=command.operation,
+        status=status,  # type: ignore[arg-type]
+        resource_id=resource_id,
+        blocker_codes=blocker_codes,
+        message=message,
+        provenance=dict(command.provenance),
+    )
 
 
 def _session(session: object) -> Session:
@@ -119,12 +400,3 @@ def _run_delete_vetoes(session: Session, resource_type: str, tenant_id: str, res
             raise
         except Exception as exc:
             raise AdminConflictError(str(exc)) from exc
-
-
-def _available_slug(session: Session, model: type[Group] | type[Role], tenant_id: str, base: str) -> str:
-    candidate = base
-    suffix = 2
-    while session.query(model).filter(model.tenant_id == tenant_id, model.slug == candidate).first():
-        candidate = f"{base}-{suffix}"
-        suffix += 1
-    return candidate

src/govoplan_access/backend/manifest.py

@@ -12,6 +12,7 @@ from govoplan_core.core.access import (
     CAPABILITY_ACCESS_EXPLANATION,
     CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER,
     CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER,
+    CAPABILITY_ACCESS_GOVERNANCE_PROJECTION_V1,
     CAPABILITY_ACCESS_PERMISSION_EVALUATOR,
     CAPABILITY_ACCESS_PRINCIPAL_RESOLVER,
     CAPABILITY_ACCESS_TENANT_PROVISIONER,
@@ -474,6 +475,44 @@ ACCESS_DOCUMENTATION: tuple[DocumentationTopic, ...] = (
             ],
         },
     ),
+    DocumentationTopic(
+        id="access.operator.governance-projection",
+        title="Reconcile managed groups and roles in bulk",
+        summary="Access projects centrally assigned governance templates into tenant groups and roles through one bounded, versioned reconciliation contract.",
+        body=(
+            "Admin supplies stable template and tenant-assignment inputs to the Access governance projection v1 capability. Access bulk-loads managed groups, roles, memberships, and role mappings, then returns one created, updated, unchanged, removed, absent, blocked, or failed outcome per assignment. Repeating the same request is idempotent. A removal remains blocked while the managed resource has members or role assignments, and protected or unrelated tenant resources are never adopted. Dry runs calculate the same outcomes without changing Access data; applied and preview runs retain source and assignment-mode provenance in the Admin audit event."
+        ),
+        layer="configured",
+        documentation_types=("admin",),
+        audience=("system_admin", "operator"),
+        order=31,
+        conditions=(
+            DocumentationCondition(
+                required_modules=("access", "admin"),
+                any_scopes=("access:governance:read", "access:governance:write"),
+            ),
+        ),
+        links=(
+            DocumentationLink(label="Governance templates", href="/admin?section=system-role-templates", kind="runtime"),
+            DocumentationLink(label="Bulk synchronization API", href="/api/v1/admin/system/governance-templates/synchronize", kind="api"),
+            DocumentationLink(label="Access module boundary", href="docs/ACCESS_MODULE_BOUNDARY.md", kind="repository"),
+        ),
+        metadata={
+            "kind": "operator_workflow",
+            "help_contexts": ["admin.governance-templates"],
+            "capability": "access.governanceProjection.v1",
+            "batch_limit": 500,
+            "outcome_statuses": [
+                "created",
+                "updated",
+                "unchanged",
+                "removed",
+                "absent",
+                "blocked",
+                "failed",
+            ],
+        },
+    ),
     DocumentationTopic(
         id="access.reference.admin-access-fields",
         title="Access administration fields",
@@ -1180,6 +1219,7 @@ manifest = ModuleManifest(
         CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER: _first_admin_provisioner,
         CAPABILITY_ACCESS_ADMINISTRATION: _access_administration,
         CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER: _governance_materializer,
+        CAPABILITY_ACCESS_GOVERNANCE_PROJECTION_V1: _governance_materializer,
         CAPABILITY_ACCESS_PEOPLE_SEARCH: _people_search,
         CAPABILITY_ACCESS_REFERENCE_OPTIONS: _access_reference_options,
         ACCESS_CONFIGURATION_CAPABILITY: _configuration_provider,

tests/test_governance_projection.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_access.backend.db.models import Account, Group, Role, User, UserRoleAssignment
+from govoplan_access.backend.governance_materializer import SqlAccessGovernanceMaterializer
+from govoplan_core.core.access import (
+    GovernanceProjectionBatch,
+    GovernanceProjectionCommand,
+    GovernanceTemplateMaterialization,
+)
+from govoplan_core.db.base import Base
+
+
+def _command(index: int, *, kind: str = "role", operation: str = "upsert") -> GovernanceProjectionCommand:
+    return GovernanceProjectionCommand(
+        assignment_id=f"assignment-{kind}-{index}",
+        operation=operation,  # type: ignore[arg-type]
+        template=GovernanceTemplateMaterialization(
+            template_id=f"template-{kind}",
+            kind=kind,  # type: ignore[arg-type]
+            tenant_id=f"tenant-{index}",
+            slug=f"managed-{kind}",
+            name=f"Managed {kind}",
+            permissions=("access:role:read",) if kind == "role" else (),
+            required=True,
+        ),
+        provenance={"source": "test", "assignment_mode": "required"},
+    )
+
+
+class GovernanceProjectionTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(bind=self.engine)
+        self.Session = sessionmaker(bind=self.engine)
+        self.session = self.Session()
+        self.materializer = SqlAccessGovernanceMaterializer()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(bind=self.engine)
+        self.engine.dispose()
+
+    def test_bulk_projection_is_idempotent_and_returns_per_assignment_outcomes(self) -> None:
+        commands = tuple(_command(index) for index in range(5))
+        first = self.materializer.reconcile(
+            self.session,
+            GovernanceProjectionBatch(operation_id="first", commands=commands),
+        )
+        second = self.materializer.reconcile(
+            self.session,
+            GovernanceProjectionBatch(operation_id="second", commands=commands),
+        )
+
+        self.assertEqual(["created"] * 5, [item.status for item in first.outcomes])
+        self.assertEqual(["unchanged"] * 5, [item.status for item in second.outcomes])
+        self.assertEqual(5, self.session.query(Role).count())
+        self.assertEqual(
+            {item.assignment_id for item in commands},
+            {item.assignment_id for item in second.outcomes},
+        )
+        self.assertTrue(all(item.provenance["source"] == "test" for item in second.outcomes))
+
+    def test_removal_isolated_blocker_preserves_other_batch_outcomes(self) -> None:
+        first, second = _command(1), _command(2)
+        created = self.materializer.reconcile(
+            self.session,
+            GovernanceProjectionBatch(operation_id="create", commands=(first, second)),
+        )
+        roles = {item.tenant_id: item.resource_id for item in created.outcomes}
+        account = Account(
+            id="account-1",
+            email="assigned@example.test",
+            normalized_email="assigned@example.test",
+        )
+        user = User(
+            id="user-1",
+            tenant_id="tenant-1",
+            account_id=account.id,
+            email=account.email,
+        )
+        self.session.add_all([account, user])
+        self.session.flush()
+        self.session.add(
+            UserRoleAssignment(
+                tenant_id="tenant-1",
+                user_id=user.id,
+                role_id=roles["tenant-1"],
+            )
+        )
+        self.session.flush()
+
+        removals = tuple(
+            GovernanceProjectionCommand(
+                assignment_id=item.assignment_id,
+                operation="remove",
+                template=item.template,
+                provenance=item.provenance,
+            )
+            for item in (first, second)
+        )
+        result = self.materializer.reconcile(
+            self.session,
+            GovernanceProjectionBatch(operation_id="remove", commands=removals),
+        )
+
+        self.assertEqual(["blocked", "removed"], [item.status for item in result.outcomes])
+        self.assertEqual(("role_has_users",), result.outcomes[0].blocker_codes)
+        self.assertIsNotNone(self.session.get(Role, roles["tenant-1"]))
+        self.assertIsNone(self.session.get(Role, roles["tenant-2"]))
+
+    def test_dry_run_does_not_mutate(self) -> None:
+        result = self.materializer.reconcile(
+            self.session,
+            GovernanceProjectionBatch(
+                operation_id="preview",
+                commands=(_command(1, kind="group"),),
+                dry_run=True,
+            ),
+        )
+        self.assertEqual("created", result.outcomes[0].status)
+        self.assertEqual(0, self.session.query(Group).count())
+
+    def test_bulk_read_query_count_does_not_grow_per_assignment(self) -> None:
+        def select_count(size: int) -> int:
+            count = 0
+
+            def record_select(_conn, _cursor, statement, _parameters, _context, _executemany):
+                nonlocal count
+                if statement.lstrip().upper().startswith("SELECT"):
+                    count += 1
+
+            event.listen(self.engine, "before_cursor_execute", record_select)
+            try:
+                self.materializer.reconcile(
+                    self.session,
+                    GovernanceProjectionBatch(
+                        operation_id=f"preview-{size}",
+                        commands=tuple(
+                            _command(index, kind="group" if index % 2 else "role")
+                            for index in range(size)
+                        ),
+                        dry_run=True,
+                    ),
+                )
+            finally:
+                event.remove(self.engine, "before_cursor_execute", record_select)
+            return count
+
+        small = select_count(2)
+        large = select_count(200)
+        self.assertEqual(small, large)
+        self.assertLessEqual(large, 4)
+
+
+if __name__ == "__main__":
+    unittest.main()