Jurisdiction Routing Logic: Deterministic Signal Resolution at DSR Intake

Within the broader DSR Architecture & Intake Routing framework, jurisdiction routing is the control plane that decides which statute governs a request before any downstream worker touches personal data. It is the stage that assigns the primary regulatory framework — and therefore the statutory clock — to every data subject request (DSR). Get it wrong and everything after it is wrong: an EU resident routed under CCPA is given 45 days instead of the one month GDPR Art. 12(3) allows, a California opt-out is executed with GDPR erasure semantics, and the audit trail records a defensible-looking decision that is materially non-compliant. This page addresses the specific gap between raw, conflicting geographic signals and a single, attributable jurisdiction tag that the rest of the pipeline can trust.

Routing is not a lookup — it is a conflict-resolution problem. A single request can carry a billing-address country, an account-registration locale, an explicit residency declaration, and an IP geolocation, and those signals routinely disagree. The stage below resolves them deterministically through weighted precedence, fails closed to manual triage on genuine ambiguity, and stamps an immutable directive that hands off to the sibling GDPR vs CCPA Request Taxonomies stage for canonical-action mapping and to 30-Day vs 45-Day SLA Mapping for the deadline countdown. Upstream, it assumes a signed, deduplicated payload from Secure Intake Form Design; this page owns only the jurisdiction decision.

Phase 1: Payload Normalization and Schema Validation #

Routing fidelity depends on structurally sound input, so the router rejects malformed payloads at the edge before they consume the response window. Validation runs synchronously at the API gateway with Pydantic v2 in strict mode, returning an explicit 400 Bad Request with field-level error traces so clients can remediate immediately. The model captures each raw signal verbatim — no signal is discarded or coerced at parse time, because the precedence resolver in Phase 2 needs every source, and the audit trail must preserve exactly what arrived.

from datetime import datetime, timezone
from pydantic import BaseModel, ConfigDict, Field, field_validator


class IntakeSignals(BaseModel):
    # extra="forbid" makes schema drift a loud failure, not a silent leak.
    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)

    request_id: str = Field(..., pattern=r"^REQ-\d{8}$")
    declared_residency: str | None = Field(default=None, min_length=2, max_length=2)
    billing_country: str | None = Field(default=None, min_length=2, max_length=2)
    account_locale: str | None = Field(default=None, min_length=2, max_length=5)
    ip_country: str | None = Field(default=None, min_length=2, max_length=2)
    received_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc)
    )

    @field_validator("received_at")
    @classmethod
    def must_be_aware(cls, v: datetime) -> datetime:
        # datetime.utcnow() is deprecated in 3.12+; require tz-aware input so
        # the SLA clock in 30-Day vs 45-Day SLA Mapping never drifts.
        if v.tzinfo is None:
            raise ValueError("received_at must be timezone-aware (UTC)")
        return v

Country and locale fields use ISO 3166-1 alpha-2 / BCP 47 forms so the resolver compares like with like. A payload that arrives with a free-text country name (“California”, “Deutschland”) is rejected here rather than guessed at, because a misread country propagates into an irreversible fulfillment decision. Validated models serialize to Avro or Protobuf for compact broker transmission downstream.

Phase 2: Multi-Signal Jurisdiction Resolution #

Jurisdiction resolution is the heart of this stage: it collapses four potentially disagreeing signals into one jurisdiction with an attached confidence, using explicit weighted precedence rather than string heuristics scattered across services. Explicit user declarations outrank inferred signals, and IP geolocation is treated as a weak hint only — it is trivially spoofed by a VPN and, under GDPR Art. 3, physical location is not the same as the residency that establishes the applicable regime. When the highest-weight signal falls below a confidence floor, the resolver returns UNKNOWN and fails closed to triage rather than defaulting to a convenient regime.

from enum import Enum


class Jurisdiction(str, Enum):
    GDPR = "GDPR"          # EU/EEA data subjects (Art. 3 extraterritorial)
    CCPA = "CCPA"          # California consumers (CCPA as amended by CPRA)
    US_STATE = "US_STATE"  # other US state privacy laws (VCDPA, CPA, ...)
    UNKNOWN = "UNKNOWN"     # ambiguous -> fail closed to manual triage


# Signal source -> weight. Explicit declaration dominates; IP is a weak hint.
SIGNAL_WEIGHTS: dict[str, float] = {
    "declared_residency": 1.0,
    "billing_country": 0.6,
    "account_locale": 0.4,
    "ip_country": 0.2,
}
CONFIDENCE_FLOOR = 0.5

EU_EEA = frozenset({
    "AT", "BE", "BG", "HR", "CY", "CZ", "DK", "EE", "FI", "FR", "DE", "GR",
    "HU", "IE", "IT", "LV", "LT", "LU", "MT", "NL", "PL", "PT", "RO", "SK",
    "SI", "ES", "SE", "IS", "LI", "NO",
})


def _to_jurisdiction(country: str, region: str | None) -> Jurisdiction:
    """Map a resolved country (+ optional US region) to a regime."""
    if country in EU_EEA:
        return Jurisdiction.GDPR
    if country == "US":
        return Jurisdiction.CCPA if region == "CA" else Jurisdiction.US_STATE
    return Jurisdiction.UNKNOWN


def resolve_jurisdiction(signals: IntakeSignals) -> tuple[Jurisdiction, float]:
    """Resolve the governing regime and a confidence in [0, 1].

    Weighted precedence: each present signal votes for a jurisdiction with its
    weight; the winning jurisdiction's normalized score is the confidence.
    Below CONFIDENCE_FLOOR we return UNKNOWN so the caller fails closed.
    """
    votes: dict[Jurisdiction, float] = {}
    total = 0.0
    # declared_residency and billing_country map country->regime directly;
    # account_locale like "en-US" contributes a country hint.
    candidates = {
        "declared_residency": (signals.declared_residency, None),
        "billing_country": (signals.billing_country, None),
        "account_locale": (
            (signals.account_locale or "").split("-")[-1] or None, None
        ),
        "ip_country": (signals.ip_country, None),
    }
    region = "CA" if (signals.declared_residency == "US"
                      and (signals.account_locale or "").endswith("CA")) else None
    for source, (country, _) in candidates.items():
        if not country:
            continue
        weight = SIGNAL_WEIGHTS[source]
        total += weight
        j = _to_jurisdiction(country.upper(), region)
        votes[j] = votes.get(j, 0.0) + weight

    if not votes or total == 0.0:
        return Jurisdiction.UNKNOWN, 0.0
    winner = max(votes, key=votes.__getitem__)
    confidence = votes[winner] / total
    if winner is Jurisdiction.UNKNOWN or confidence < CONFIDENCE_FLOOR:
        return Jurisdiction.UNKNOWN, confidence
    return winner, confidence

Weighting the signals rather than short-circuiting on the first non-null one is deliberate: a spoofed IP that disagrees with a declared residency and a billing country will lose the vote, and a lone low-weight signal will fall under the confidence floor and route to triage. The resolved jurisdiction determines which statutory clock the 30-Day vs 45-Day SLA Mapping stage attaches, and it is recorded — with the confidence and the contributing signals — in the audit trail for GDPR Art. 5(2) accountability.

Phase 3: Immutable Directive Generation #

Directive generation stamps the routing decision onto an immutable envelope that no downstream worker may mutate. The directive carries the resolved jurisdiction, the compliance framework, the queue target, the statutory clock in hours, and a priority derived from how tight that clock is. Modeling it as a frozen Pydantic model means a worker that tries to “fix” a directive fails loudly rather than silently rerouting a request past its deadline.

class RoutingDirective(BaseModel):
    model_config = ConfigDict(frozen=True)

    request_id: str
    jurisdiction: Jurisdiction
    compliance_framework: str
    queue: str
    sla_hours: int
    priority: int  # lower = higher priority
    confidence: float


# Statutory baselines the router stamps; the SLA stage owns extensions.
# GDPR Art. 12(3): one month. CCPA 1798.130(a)(2): 45 days.
SLA_HOURS: dict[Jurisdiction, int] = {
    Jurisdiction.GDPR: 30 * 24,
    Jurisdiction.CCPA: 45 * 24,
    Jurisdiction.US_STATE: 45 * 24,
}
QUEUE: dict[Jurisdiction, str] = {
    Jurisdiction.GDPR: "gdpr_intake",
    Jurisdiction.CCPA: "ccpa_intake",
    Jurisdiction.US_STATE: "us_state_intake",
    Jurisdiction.UNKNOWN: "manual_triage",
}


def build_directive(signals: IntakeSignals) -> RoutingDirective:
    """Resolve jurisdiction and stamp an immutable routing directive."""
    jurisdiction, confidence = resolve_jurisdiction(signals)
    if jurisdiction is Jurisdiction.UNKNOWN:
        return RoutingDirective(
            request_id=signals.request_id,
            jurisdiction=jurisdiction,
            compliance_framework="MANUAL_REVIEW",
            queue=QUEUE[Jurisdiction.UNKNOWN],
            sla_hours=30 * 24,  # assume the tightest clock until a human decides
            priority=0,
            confidence=confidence,
        )
    sla = SLA_HOURS[jurisdiction]
    # Tighter statutory windows outrank looser ones on the shared bus.
    priority = 1 if sla <= 30 * 24 else 2
    return RoutingDirective(
        request_id=signals.request_id,
        jurisdiction=jurisdiction,
        compliance_framework=jurisdiction.value,
        queue=QUEUE[jurisdiction],
        sla_hours=sla,
        priority=priority,
        confidence=confidence,
    )

An UNKNOWN result stamps the tightest clock (the GDPR one month) as a safety default while routing to manual_triage — over-restricting the timeline is recoverable, under-restricting it is a breach. For teams extending this with dynamic SLA overrides, framework versioning, and residency-attestation providers, Building a jurisdiction-aware intake router in Python walks through the full production configuration.

Phase 4: Broker Dispatch and Fulfillment Handoff #

Dispatch publishes the immutable directive to the message broker keyed by jurisdiction, then hands off to the taxonomy stage without ever reopening the routing decision. Keying Kafka partitions on jurisdiction keeps GDPR and CCPA traffic on separate consumer groups so a backlog in one regime cannot starve the other’s statutory clock, and it enforces strict ordering within each regulatory boundary. Consumer groups are isolated per queue so high-priority erasures scale independently of routine access requests.

import json


def dispatch(directive: RoutingDirective, producer) -> None:
    """Publish the directive to a jurisdiction-partitioned Kafka topic.

    `producer` is a configured kafka-python KafkaProducer. Partitioning by
    jurisdiction preserves per-regime ordering and isolates consumer groups.
    """
    producer.send(
        topic="dsr-routing",
        key=directive.jurisdiction.value.encode("utf-8"),
        value=json.dumps(directive.model_dump()).encode("utf-8"),
    )
    producer.flush()

Once dispatched, the request crosses a hard boundary: the routing layer never mutates the payload again, and the fulfillment layer never re-resolves jurisdiction. The consuming stage maps the stamped framework onto a canonical action using GDPR vs CCPA Request Taxonomies, keeping access, erasure, and opt-out semantics physically isolated so a GDPR extraction can never leak into a CCPA deletion path. Consumer-lag monitoring on each partition surfaces requests approaching their deadline before they breach.

Edge Cases and Conflict Resolution #

Jurisdiction resolution is where conflicting and ambiguous signals surface. Resolve each deterministically rather than guessing:

• Conflicting high-weight signals. A declared EU residency with a US billing country lands below the confidence floor and routes to triage. Never let a single signal silently override an explicit conflict — the weighted vote is what makes the ambiguity visible.
• UNKNOWN jurisdiction. When no signal clears the floor, tag Jurisdiction.UNKNOWN, stamp the tightest clock, and fail closed to manual_triage. A human resolves residency before the clock is trusted; the pipeline never assumes a regime to keep automation flowing.
• VPN / spoofed IP. IP geolocation carries the lowest weight precisely because it is spoofable. When it is the only signal present, its 0.2 weight cannot clear the floor alone, so a VPN user with no other signal is triaged rather than misrouted.
• Multi-jurisdiction subject. A subject who is both an EU resident and a California consumer holds rights under both regimes for the same identifier. The router emits one directive per resolvable regime and defers reconciliation to the audit layer — it never collapses two obligations into one “best” jurisdiction.
• US state without a resolver entry. A US country with a non-California region maps to US_STATE and a generic state-law queue until a state-specific resolver (VCDPA, CPA) is added; it is never silently treated as CCPA.

Performance and Scale Considerations #

Routing sits on the hot path of every request, so resolution must be cheap and dispatch must isolate failure domains:

• Cache the maps, not the decision. SIGNAL_WEIGHTS, EU_EEA, SLA_HOURS, and QUEUE are tiny immutable structures — load them once at process start. Cache resolved directives per request_id in Redis only to make retries idempotent, never to skip re-validation.
• Redis-back the residency lookups. Where residency attestation calls an external identity provider, cache the attestation result (not the raw PII) in Redis with a short TTL so high-volume intake bursts do not hammer the provider or add latency to the hot path.
• Partition the broker by jurisdiction. Keying Kafka partitions on jurisdiction keeps regimes on separate consumer groups so a backlog in one cannot delay another’s statutory clock.
• Isolate consumer groups per queue. GDPR, CCPA, US-state, and triage workers scale independently; an opt-out spike after a marketing campaign must not delay time-critical GDPR erasures.
• Throughput target. Signal resolution and directive stamping should complete in single-digit milliseconds; any request approaching its window should already have paged an operator via consumer-lag alerting.

Testing and Compliance Verification #

Treat the resolver as a compliance control and prove it before it touches production requests:

• Signal matrix. Feed resolve_jurisdiction a matrix crossing every combination of present/absent signals and agreeing/conflicting values; assert each unambiguous case returns the expected regime and each genuine conflict returns UNKNOWN.
• Confidence-floor regression. Assert that an IP-only signal never clears the floor and that a declared residency alone always does, guarding against a future weight edit that lets a weak signal decide.
• Held-out regions. Include fixtures for jurisdictions live traffic does not yet cover (e.g. a Brazilian LGPD country code) so UNKNOWN handling is exercised before it is needed in production.
• Clock-attribution completeness. Assert every non-UNKNOWN directive carries an sla_hours value traceable to a specific statute (GDPR Art. 12(3), CCPA §1798.130(a)(2)) so the audit artifact is always attributable.

import pytest


def test_spoofed_ip_alone_fails_closed():
    signals = IntakeSignals(request_id="REQ-00000001", ip_country="US")
    jurisdiction, confidence = resolve_jurisdiction(signals)
    assert jurisdiction is Jurisdiction.UNKNOWN
    assert confidence < CONFIDENCE_FLOOR


def test_declared_eu_residency_routes_to_gdpr():
    signals = IntakeSignals(request_id="REQ-00000002", declared_residency="DE")
    directive = build_directive(signals)
    assert directive.jurisdiction is Jurisdiction.GDPR
    assert directive.sla_hours == 30 * 24
    assert directive.queue == "gdpr_intake"


def test_conflicting_signals_route_to_triage():
    signals = IntakeSignals(
        request_id="REQ-00000003",
        declared_residency="FR",
        billing_country="US",
        ip_country="US",
    )
    directive = build_directive(signals)
    assert directive.jurisdiction is Jurisdiction.UNKNOWN
    assert directive.queue == "manual_triage"

Frequently Asked Questions #

Why weight the signals instead of trusting the IP address? #

Because IP geolocation is trivially spoofed by a VPN and, under GDPR Art. 3, physical location is not the residency that establishes the applicable regime. Weighting explicit declarations above billing and account signals — and treating IP as a weak hint — means a spoofed or lone IP loses the vote or falls under the confidence floor and routes to triage rather than misrouting a request.

What happens when the signals conflict? #

The weighted vote makes the conflict visible: no single regime clears the confidence floor, so the resolver returns UNKNOWN and the request fails closed to manual triage with the tightest statutory clock stamped as a safety default. A human resolves residency before the clock is trusted, which is safer than letting one signal silently override the others.

How does routing relate to the SLA clock? #

The router only decides which statute applies and records the baseline window (GDPR one month per Art. 12(3), CCPA 45 days per §1798.130(a)(2)). The countdown state machine, extension logic, and escalation live in the 30-Day vs 45-Day SLA Mapping stage, which consumes the directive this stage stamps.

How should a subject covered by both GDPR and CCPA be routed? #

Emit one immutable directive per resolvable regime — each with its own jurisdiction, framework, and clock — and reconcile them only at the audit layer. Collapsing them into a single “best” jurisdiction risks applying the wrong deadline or the wrong exceptions to one of the two obligations.

Where does the routing decision get recorded for audit? #

Every directive — including UNKNOWN results — is written to the immutable audit trail with the resolved jurisdiction, the confidence score, and the contributing signals, satisfying GDPR Art. 5(2) accountability. Downstream workers read the stamped decision; they never re-resolve jurisdiction, so the audit record is the single source of truth for why a request was routed as it was.

Related #

• GDPR vs CCPA Request Taxonomies — mapping the framework this stage stamps onto a canonical action for each downstream worker.
• 30-Day vs 45-Day SLA Mapping — the countdown state machine and extension logic that consume the stamped clock.
• Secure Intake Form Design — signed, deduplicated payloads and anti-replay protection upstream of this routing gate.
• Building a jurisdiction-aware intake router in Python — the full production router with dynamic SLA overrides and residency-attestation providers.
• Up to DSR Architecture & Intake Routing — the intake and routing control plane this stage anchors.