Block data access at the infrastructure layer, not the prompt layer. The distinction matters. A prompt instruction saying “don’t access health data” can be overridden by a sufficiently complex task chain. A database permission saying “this connection cannot read health_category columns” cannot.

from dataclasses import dataclass
from enum import auto, Enum
from typing import Set

class DataCategory(Enum):
    CALENDAR = auto()
    EMAIL_METADATA = auto()
    HEALTH = auto()       # Article 9 special category
    BIOMETRIC = auto()
    FINANCIAL = auto()

@dataclass(frozen=True)
class TaskConstraints:
    goal: str
    allowed: Set[DataCategory]
    blocked: Set[DataCategory]

    def can_access(self, category: DataCategory) -> bool:
        if category in self.blocked:
            return False
        return category in self.allowed

# Per-task instantiation — not a global config
constraints = TaskConstraints(
    goal="schedule_meeting",
    allowed={DataCategory.CALENDAR, DataCategory.EMAIL_METADATA},
    blocked={DataCategory.HEALTH, DataCategory.BIOMETRIC, DataCategory.FINANCIAL}
)

# Called at every data access point in the infrastructure layer
if not constraints.can_access(detected_category):
    raise AccessDenied(
        f"Category {detected_category.name} blocked for goal "
        f"'{constraints.goal}' — not in approved DPIA scope"
    )

from abc import ABC, abstractmethod
from typing import Dict

class VectorStore(ABC):
    @abstractmethod
    def delete_by_user(self, user_id: str) -> Dict[str, object]:
        """Delete all vectors for user. Returns status and count deleted."""
        pass

    @abstractmethod
    def verify_deletion(self, user_id: str) -> bool:
        """Confirm zero vectors remain. Not optional — call this."""
        pass

def execute_erasure_request(user_id: str, store: VectorStore) -> dict:
    result = store.delete_by_user(user_id)
    confirmed = store.verify_deletion(user_id)

    if not confirmed:
        raise ErasureFailure(
            f"Deletion called for {user_id} but verification returned "
            f"non-zero vectors. Manual remediation required."
        )

    return {"user_id": user_id, "deleted": result, "verified": confirmed}

The agent calls only pre-approved processors. Not “we’ve blocked the bad ones.” The allowlist is the only thing that exists. If a processor isn’t in the registry, the call fails. This sounds extreme and it isn’t — it’s the only architecture that handles the “six months of incremental API additions” failure mode.

from dataclasses import dataclass
from typing import Optional, Set

@dataclass(frozen=True)
class Processor:
    name: str
    dpa_signed: bool
    location: str                         # "EU" or jurisdiction code
    transfer_mechanism: Optional[str]     # SCC_2021, BCR, ADEQUACY, None
    allowed_data: Set[str]                # Data categories in DPA scope
    sub_processors_disclosed: bool        # Article 28(2) requirement

class ProcessorRegistry:
    def __init__(self, processors: Set[Processor]):
        self._processors = {p.name: p for p in processors}

    def validate_call(self, name: str, data_category: str) -> Processor:
        # Allowlist: not in registry = blocked, not just unreviewed
        if name not in self._processors:
            raise ValueError(
                f"Processor '{name}' not in registry. "
                f"Add DPA and transfer mechanism before enabling."
            )

        proc = self._processors[name]

        if not proc.dpa_signed:
            raise ValueError(f"No DPA for '{name}' — Art. 28 violation")

        if data_category not in proc.allowed_data:
            raise ValueError(
                f"Category '{data_category}' not covered in DPA for '{name}'"
            )

        if proc.location != "EU" and not proc.transfer_mechanism:
            raise ValueError(
                f"Cross-border transfer to '{proc.location}' "
                f"without SCC/BCR — Art. 44 violation"
            )

        if not proc.sub_processors_disclosed:
            raise ValueError(
                f"'{name}' has not disclosed sub-processors — Art. 28(2)"
            )

        return proc

Log the reasoning structure, not just the I/O. For agentic systems with explicit planning steps, this is a matter of capturing what the agent decided to do and in what order — which is usually already represented in the agent’s internal state.

from dataclasses import dataclass
from datetime import datetime
from typing import Any, Dict, List

@dataclass
class DecisionTrace:
    trace_id: str
    timestamp: datetime
    goal: str
    plan_executed: List[str]          # Ordered steps the agent took
    tools_called: List[Dict[str, Any]]
    data_categories_accessed: List[str]   # Categories only, not raw values
    processors_accessed: List[str]
    automated_decision: bool

    def to_dsar_response(self) -> Dict:
        """Returns Art. 15-compliant logic description."""
        return {
            "processing_purpose": self.goal,
            "logic_applied": self.plan_executed,
            "data_categories_processed": self.data_categories_accessed,
            "recipients": self.processors_accessed,
            "automated_decision": self.automated_decision,
            "timestamp": self.timestamp.isoformat(),
            "trace_id": self.trace_id
        }