feat: define bulk governance projection contract

src/govoplan_core/core/access.py

@@ -24,6 +24,7 @@ CAPABILITY_ACCESS_TENANT_PROVISIONER = "access.tenantProvisioner"
 CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER = "access.firstAdminProvisioner"
 CAPABILITY_ACCESS_ADMINISTRATION = "access.administration"
 CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER = "access.governanceMaterializer"
+CAPABILITY_ACCESS_GOVERNANCE_PROJECTION_V1 = "access.governanceProjection.v1"
 CAPABILITY_POLICY_ACCESS_EXPLANATION_SUBJECTS = (
     "policy.access_explanation_subjects"
 )
@@ -52,6 +53,7 @@ ACCESS_CAPABILITY_NAMES = frozenset(
         CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER,
         CAPABILITY_ACCESS_ADMINISTRATION,
         CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER,
+        CAPABILITY_ACCESS_GOVERNANCE_PROJECTION_V1,
         CAPABILITY_TENANCY_TENANT_RESOLVER,
         CAPABILITY_AUDIT_SINK,
         CAPABILITY_AUDIT_RECORDER,
@@ -390,6 +392,82 @@ class GovernanceTemplateMaterialization:
     required: bool = False
 
 
+GovernanceProjectionOperation = Literal["upsert", "remove"]
+GovernanceProjectionStatus = Literal[
+    "created",
+    "updated",
+    "unchanged",
+    "removed",
+    "absent",
+    "blocked",
+    "failed",
+]
+
+
+@dataclass(frozen=True, slots=True)
+class GovernanceProjectionCommand:
+    """Stable Access-owned input for one governance assignment projection."""
+
+    assignment_id: str
+    operation: GovernanceProjectionOperation
+    template: GovernanceTemplateMaterialization
+    provenance: Mapping[str, str] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        if not self.assignment_id or len(self.assignment_id) > 255:
+            raise ValueError("Governance projection assignment ids must contain at most 255 characters.")
+        if len(self.provenance) > 20:
+            raise ValueError("Governance projection provenance supports at most 20 entries.")
+        for key, value in self.provenance.items():
+            if not key or len(key) > 100 or len(value) > 500:
+                raise ValueError("Governance projection provenance entries exceed their bounds.")
+
+
+@dataclass(frozen=True, slots=True)
+class GovernanceProjectionBatch:
+    """Versioned, bounded reconciliation request independent of Admin internals."""
+
+    operation_id: str
+    commands: tuple[GovernanceProjectionCommand, ...]
+    version: Literal["1"] = "1"
+    dry_run: bool = False
+
+    def __post_init__(self) -> None:
+        if not self.operation_id or len(self.operation_id) > 255:
+            raise ValueError("Governance projection operation ids must contain at most 255 characters.")
+        if not self.commands or len(self.commands) > 500:
+            raise ValueError("Governance projection batches must contain between 1 and 500 commands.")
+        assignment_ids = [command.assignment_id for command in self.commands]
+        if len(assignment_ids) != len(set(assignment_ids)):
+            raise ValueError("Governance projection assignment ids must be unique within a batch.")
+
+
+@dataclass(frozen=True, slots=True)
+class GovernanceProjectionOutcome:
+    assignment_id: str
+    template_id: str
+    tenant_id: str
+    kind: Literal["group", "role"]
+    operation: GovernanceProjectionOperation
+    status: GovernanceProjectionStatus
+    resource_id: str | None = None
+    blocker_codes: tuple[str, ...] = ()
+    message: str | None = None
+    provenance: Mapping[str, str] = field(default_factory=dict)
+
+
+@dataclass(frozen=True, slots=True)
+class GovernanceProjectionResult:
+    operation_id: str
+    outcomes: tuple[GovernanceProjectionOutcome, ...]
+    version: Literal["1"] = "1"
+    dry_run: bool = False
+
+    @property
+    def blocked(self) -> tuple[GovernanceProjectionOutcome, ...]:
+        return tuple(item for item in self.outcomes if item.status in {"blocked", "failed"})
+
+
 @dataclass(frozen=True, slots=True)
 class AuditEvent:
     event_type: str
@@ -693,6 +771,18 @@ class AccessGovernanceMaterializer(Protocol):
         ...
 
 
+@runtime_checkable
+class AccessGovernanceProjectionV1(Protocol):
+    """Bulk reconciliation boundary for Admin-owned governance assignments."""
+
+    def reconcile(
+        self,
+        session: object,
+        batch: GovernanceProjectionBatch,
+    ) -> GovernanceProjectionResult:
+        ...
+
+
 @runtime_checkable
 class AuditSink(Protocol):
     def record(self, event: AuditEvent) -> None:

tests/test_api_smoke.py

@@ -5506,6 +5506,67 @@ class ApiSmokeTests(unittest.TestCase):
         self.assertEqual(deleted_delta.status_code, 200, deleted_delta.text)
         self.assertTrue(any(item["id"] == template_id and item["resource_type"] == "governance_template" for item in deleted_delta.json()["deleted"]))
 
+    def test_governance_template_bulk_synchronization_previews_and_applies(self) -> None:
+        headers, login = self._login()
+        tenant_id = login["tenant"]["id"]
+        approver_headers = self._create_system_approver(
+            headers,
+            tenant_id=tenant_id,
+            email="governance-sync-approver@example.local",
+        )
+        create_payload = {
+            "kind": "role",
+            "slug": "governance-sync",
+            "name": "Governance Sync",
+            "description": "Bulk reconciliation contract",
+            "permissions": ["admin:roles:read"],
+            "is_active": True,
+            "assignments": [{"tenant_id": tenant_id, "mode": "required"}],
+        }
+        change_request_id = self._approved_configuration_change(
+            headers,
+            approver_headers,
+            key="governance_templates",
+            value=create_payload,
+            target={"kind": "role", "slug": "governance-sync"},
+        )
+        created = self.client.post(
+            "/api/v1/admin/system/governance-templates",
+            headers=headers,
+            json={**create_payload, "change_request_id": change_request_id},
+        )
+        self.assertEqual(201, created.status_code, created.text)
+        template_id = created.json()["id"]
+
+        preview = self.client.post(
+            "/api/v1/admin/system/governance-templates/synchronize",
+            headers=headers,
+            json={"template_ids": [template_id], "dry_run": True},
+        )
+        self.assertEqual(200, preview.status_code, preview.text)
+        self.assertTrue(preview.json()["dry_run"])
+        self.assertEqual({"unchanged": 1}, preview.json()["counts"])
+        self.assertEqual("1", preview.json()["version"])
+        self.assertEqual("admin.bulk-synchronization", preview.json()["outcomes"][0]["provenance"]["source"])
+
+        applied = self.client.post(
+            "/api/v1/admin/system/governance-templates/synchronize",
+            headers=headers,
+            json={"template_ids": [template_id], "dry_run": False},
+        )
+        self.assertEqual(200, applied.status_code, applied.text)
+        self.assertFalse(applied.json()["dry_run"])
+        self.assertEqual({"unchanged": 1}, applied.json()["counts"])
+
+        audit = self.client.get(
+            "/api/v1/admin/audit",
+            headers=headers,
+            params={"all_tenants": True, "limit": 500},
+        )
+        actions = {item["action"] for item in audit.json()["items"]}
+        self.assertIn("governance_template.synchronization_previewed", actions)
+        self.assertIn("governance_template.synchronized", actions)
+
     def test_module_installer_history_supports_cursor_windows(self) -> None:
         from govoplan_core.core.module_installer import default_installer_runtime_dir
 

tests/test_governance_projection_contract.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import unittest
+
+from govoplan_core.core.access import (
+    AccessGovernanceProjectionV1,
+    GovernanceProjectionBatch,
+    GovernanceProjectionCommand,
+    GovernanceProjectionResult,
+    GovernanceTemplateMaterialization,
+)
+
+
+def _command(assignment_id: str) -> GovernanceProjectionCommand:
+    return GovernanceProjectionCommand(
+        assignment_id=assignment_id,
+        operation="upsert",
+        template=GovernanceTemplateMaterialization(
+            template_id="template-1",
+            kind="role",
+            tenant_id=f"tenant-{assignment_id}",
+            slug="reader",
+            name="Reader",
+        ),
+        provenance={"source": "contract-test"},
+    )
+
+
+class _Projection:
+    def reconcile(self, session: object, batch: GovernanceProjectionBatch) -> GovernanceProjectionResult:
+        del session
+        return GovernanceProjectionResult(
+            operation_id=batch.operation_id,
+            outcomes=(),
+            dry_run=batch.dry_run,
+        )
+
+
+class GovernanceProjectionContractTests(unittest.TestCase):
+    def test_protocol_is_runtime_checkable(self) -> None:
+        self.assertIsInstance(_Projection(), AccessGovernanceProjectionV1)
+
+    def test_batch_rejects_duplicate_assignment_ids(self) -> None:
+        with self.assertRaisesRegex(ValueError, "unique"):
+            GovernanceProjectionBatch(
+                operation_id="duplicate",
+                commands=(_command("same"), _command("same")),
+            )
+
+    def test_batch_enforces_bounds(self) -> None:
+        with self.assertRaisesRegex(ValueError, "between 1 and 500"):
+            GovernanceProjectionBatch(operation_id="empty", commands=())
+        with self.assertRaisesRegex(ValueError, "between 1 and 500"):
+            GovernanceProjectionBatch(
+                operation_id="large",
+                commands=tuple(_command(str(index)) for index in range(501)),
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()