From 019a29f9653bb784dbe21405e05d9bcb1633eaa1 Mon Sep 17 00:00:00 2001 From: Debasmita Date: Thu, 2 Jul 2026 21:27:51 +0530 Subject: [PATCH] feat(compliance): wire multi-region data residency routing into ingestion pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CMP-007 — implements data residency routing for compliance telemetry. - Add residency_policy_name and residency_region fields to TenantConfig - Inject ResidencyRouter into TenantIngestionPipeline with default EU/US policies - resolve_tenant_region() selects storage region from tenant config - Ingestion attaches 'residency_region' metadata to each event - get_region_endpoint() exposes the storage endpoint for a tenant - 7 new tests covering EU/US routing, unknown tenant, metadata attachment Closes #366 --- agentwatch/models/tenant.py | 4 ++ agentwatch/telemetry/ingestion.py | 45 ++++++++++++++ tests/test_compliance.py | 97 +++++++++++++++++++++++++++++++ 3 files changed, 146 insertions(+) diff --git a/agentwatch/models/tenant.py b/agentwatch/models/tenant.py index b0f32580..4e4d01af 100644 --- a/agentwatch/models/tenant.py +++ b/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, diff --git a/agentwatch/telemetry/ingestion.py b/agentwatch/telemetry/ingestion.py index 13c3310c..26935f0d 100644 --- a/agentwatch/telemetry/ingestion.py +++ b/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: diff --git a/tests/test_compliance.py b/tests/test_compliance.py index 5c142821..a07038c3 100644 --- a/tests/test_compliance.py +++ b/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))