feat(compliance): wire multi-region data residency routing into ingestion pipeline

agentwatch/models/tenant.py

@@ -71,6 +71,8 @@ class TenantConfig:
     max_usd_per_month: float = 100.0
     ingestion_rate_limit: int = 100  # events per second
     retention_days: int = 30
+    residency_policy_name: str | None = None  # e.g. "eu_only", "us_only"
+    residency_region: str | None = None  # e.g. "eu-west-1", "us-east-1"
 
 
 PLAN_DEFAULTS: dict[TenantPlan, TenantConfig] = {
@@ -134,6 +136,8 @@ def to_dict(self) -> dict[str, Any]:
                 "max_usd_per_month": self.config.max_usd_per_month,
                 "ingestion_rate_limit": self.config.ingestion_rate_limit,
                 "retention_days": self.config.retention_days,
+                "residency_policy_name": self.config.residency_policy_name,
+                "residency_region": self.config.residency_region,
             },
             "created_at": self.created_at.isoformat(),
             "metadata": self.metadata,

agentwatch/telemetry/ingestion.py

@@ -16,6 +16,7 @@
 
 from agentwatch.api.tenant_auth import TenantStore
 from agentwatch.core.schema import AgentEvent
+from agentwatch.governance.residency import Region, ResidencyRouter, eu_only_policy, us_only_policy
 from agentwatch.models.tenant import TenantStatus
 
 logger = logging.getLogger(__name__)
@@ -75,6 +76,7 @@ def __init__(
         tenant_store: TenantStore,
         batch_size: int = 100,
         flush_interval: float = 5.0,
+        residency_router: ResidencyRouter | None = None,
     ):
         self._tenant_store = tenant_store
         self._batch_size = batch_size
@@ -84,6 +86,12 @@ def __init__(
         self._batches: dict[str, list[AgentEvent]] = defaultdict(list)
         self._handlers: list[Any] = []
         self._flush_task: asyncio.Task | None = None
+        self._residency_router = residency_router or ResidencyRouter()
+
+        # Register default compliance policies on the router
+        if residency_router is None:
+            self._residency_router.add_policy("eu_only", eu_only_policy())
+            self._residency_router.add_policy("us_only", us_only_policy())
 
     def add_handler(self, handler: Any) -> None:
         """Register an event handler to receive batched events."""
@@ -132,6 +140,11 @@ async def ingest(self, tenant_id: str, event: AgentEvent) -> bool:
             logger.warning("Event rejected: %s for tenant %s", reason, tenant_id)
             return False
 
+        # Resolve target region for data residency compliance
+        region = self.resolve_tenant_region(tenant_id)
+        if region is not None:
+            event.metadata["residency_region"] = region.value
+
         # Accept event
         self._batches[tenant_id].append(event)
         metrics.events_accepted += 1
@@ -209,6 +222,38 @@ async def _flush_tenant(self, tenant_id: str) -> int:
         logger.debug("Flushed %d events for tenant %s", len(batch), tenant_id)
         return len(batch)
 
+    def resolve_tenant_region(self, tenant_id: str) -> Region | None:
+        """Determine the target storage region for a tenant based on its config."""
+        tenant = self._tenant_store.get_tenant(tenant_id)
+        if not tenant:
+            return None
+
+        policy_name = tenant.config.residency_policy_name
+        explicit_region = tenant.config.residency_region
+
+        if policy_name:
+            decision = self._residency_router.route(
+                policy_name,
+                current_user_region=Region(explicit_region) if explicit_region else None,
+            )
+            return decision.region
+
+        if explicit_region:
+            try:
+                return Region(explicit_region)
+            except ValueError:
+                logger.warning("Unknown region '%s' for tenant %s", explicit_region, tenant_id)
+                return None
+
+        return None
+
+    def get_region_endpoint(self, tenant_id: str) -> str | None:
+        """Get the storage endpoint URL for a tenant's data residency region."""
+        region = self.resolve_tenant_region(tenant_id)
+        if region is None:
+            return None
+        return self._residency_router.endpoint_for(region)
+
     def get_metrics(self, tenant_id: str | None = None) -> dict[str, Any]:
         """Get ingestion metrics, optionally filtered by tenant."""
         if tenant_id:

tests/test_compliance.py

@@ -32,8 +32,13 @@
     Region,
     ResidencyRouter,
     eu_only_policy,
+    us_only_policy,
 )
+from agentwatch.api.tenant_auth import TenantStore
+from agentwatch.core.schema import AgentEvent, EventType
 from agentwatch.memory.causal_graph import CausalGraph, CausalNode, EdgeKind
+from agentwatch.models.tenant import Tenant, TenantConfig, TenantPlan, TenantStatus
+from agentwatch.telemetry.ingestion import TenantIngestionPipeline
 
 # ─────────────────────────────────────────────
 # CMP-001 — GDPR
@@ -214,6 +219,98 @@ def test_causal_attribution_signs_report():
 # ─────────────────────────────────────────────
 
 
+# ─────────────────────────────────────────────
+# CMP-007b — Residency-aware Ingestion
+# ─────────────────────────────────────────────
+
+
+def _make_tenant_store() -> TenantStore:
+    store = TenantStore()
+    store.add_tenant(Tenant(
+        tenant_id="tenant-eu",
+        name="EU Corp",
+        plan=TenantPlan.PROFESSIONAL,
+        config=TenantConfig(residency_policy_name="eu_only", residency_region="eu-west-1"),
+    ))
+    store.add_tenant(Tenant(
+        tenant_id="tenant-us",
+        name="US Corp",
+        plan=TenantPlan.PROFESSIONAL,
+        config=TenantConfig(residency_policy_name="us_only", residency_region="us-east-1"),
+    ))
+    store.add_tenant(Tenant(
+        tenant_id="tenant-default",
+        name="Default Corp",
+        plan=TenantPlan.FREE,
+        config=TenantConfig(),
+    ))
+    return store
+
+
+def test_residency_ingestion_resolves_eu_region():
+    store = _make_tenant_store()
+    pipeline = TenantIngestionPipeline(tenant_store=store)
+    region = pipeline.resolve_tenant_region("tenant-eu")
+    assert region is not None
+    assert region.value.startswith("eu-")
+
+
+def test_residency_ingestion_resolves_us_region():
+    store = _make_tenant_store()
+    pipeline = TenantIngestionPipeline(tenant_store=store)
+    region = pipeline.resolve_tenant_region("tenant-us")
+    assert region is not None
+    assert region.value.startswith("us-")
+
+
+def test_residency_ingestion_no_policy_returns_none():
+    store = _make_tenant_store()
+    pipeline = TenantIngestionPipeline(tenant_store=store)
+    region = pipeline.resolve_tenant_region("tenant-default")
+    assert region is None
+
+
+def test_residency_ingestion_unknown_tenant_returns_none():
+    store = _make_tenant_store()
+    pipeline = TenantIngestionPipeline(tenant_store=store)
+    region = pipeline.resolve_tenant_region("nonexistent")
+    assert region is None
+
+
+def test_residency_ingestion_attaches_region_metadata():
+    store = _make_tenant_store()
+    pipeline = TenantIngestionPipeline(tenant_store=store)
+
+    async def _run():
+        event = AgentEvent(
+            session_id="s1",
+            agent_id="agent-a",
+            event_type=EventType.TOOL_CALL,
+            timestamp="2026-01-01T00:00:00Z",
+        )
+        accepted = await pipeline.ingest("tenant-eu", event)
+        assert accepted
+        assert event.metadata.get("residency_region", "").startswith("eu-")
+
+    import asyncio
+    asyncio.run(_run())
+
+
+def test_residency_ingestion_endpoint_lookup():
+    store = _make_tenant_store()
+    pipeline = TenantIngestionPipeline(tenant_store=store)
+    pipeline._residency_router.register_endpoint(Region.EU_WEST, "https://eu-storage.example.com")
+    endpoint = pipeline.get_region_endpoint("tenant-eu")
+    assert endpoint == "https://eu-storage.example.com"
+
+
+def test_residency_ingestion_no_endpoint_for_default_tenant():
+    store = _make_tenant_store()
+    pipeline = TenantIngestionPipeline(tenant_store=store)
+    endpoint = pipeline.get_region_endpoint("tenant-default")
+    assert endpoint is None
+
+
 def test_iso42001_report_counts():
     ams = ISO42001AMS()
     ams.add_risk(RiskAssessment(risk_id="R1", description="x", likelihood=0.6, impact=0.9))