Chapter 11: Enforcing Multi-Tenancy: Isolating Workspaces, Chunks, and Execution Bounds

In enterprise AI architectures, multiple customers, business units, or project teams often share foundational models, inference infrastructure, and vector storage engines. Without strict perimeter enforcement and cryptographic partition checks, shared infrastructure risks critical cross-tenant data exfiltration: an agent acting on behalf of Tenant A might retrieve sensitive knowledge or invoke infrastructure tools belonging to Tenant B. Multi-Tenancy requires treating tenant boundaries as non-negotiable hard perimeters that are evaluated continuously across agent entry gates, ingestion pipelines, vector retrieval queries, and tool execution boundaries.

In this chapter, we use the incident response assistant (langchain-enterprise-incident-assistant) as an enterprise case study to demonstrate how to enforce multi-tenant isolation across your application stack using langchain-security-extension. We examine how to configure tenant perimeters on entry runnables, inject mandatory tenant filters into vector retrievers, and eliminate ad-hoc web filter checks in favor of declarative Policy Enforcement Points (PEPs).

The Strict Multi-Tenancy Invariant

The Strict Multi-Tenancy Invariant dictates that a tenant boundary can never be bypassed or weakened, regardless of administrative role elevation, user clearance levels, or permissive fallback policies:

Multi-Tenant Enforcement Across Lifecycle Tiers

Multi-tenancy is maintained as an ambient invariant across every tier of the agent and knowledge pipeline:

Pipeline Tier Enforcement Mechanism Policy Invariant Violation Handling
Agent Entry Gate SecureRunnable & EmbeddedPolicyEngine Caller tenant_id must match target_tenant configured boundary Raises TenantIsolationError; returns BLOCKED_TENANT (HTTP 403)
Vector Ingestion (ETL) DocumentSecurityTransformer & SecurityEnvelopeValidator Mandatory sec:tenant_id stamped; invariant preserved across permissive & quarantine modes Atomic chunk rejection or quarantine with tenant lock; batch rolled back
Vector Retrieval (RAG) SecureRetriever AST filter push-down & PDP in-memory pruning Query AST injects {SEC_TENANT_ID: identity.tenant_id}; unauthenticated calls receive deny-all sentinel; secondary in-memory pruning Foreign chunks filtered at query time; in-memory pruning drops mismatched chunks before prompt injection
Tool Execution (PEP) @secured_tool Policy Enforcement Point (PEP) & CEL constraints Dynamic entity attribute evaluation (e.g. ownerDepartment == subject.dept), clearance floors, and parameter constraints Mutative tools: hard abort (STATUS_BLOCKED_BY_POLICY); read-only tools: soft refusal

Tool PEP Multi-Tenancy & Dynamic ABAC: While hard tenant perimeters are anchored at agent entry gates and knowledge retrievers, the @secured_tool Policy Enforcement Point (PEP) enforces in-band tenant isolation on tools that consume explicit tenant identifiers (e.g., args.tenant_id == subject.tenantId) or dynamically resolve tenant ownership via asset registries. For a comprehensive guide on defining CEL rules, dynamic entity hydration, and dual-mode PEP enforcement contracts, see Chapter 7: Guarding Tools and Enterprise APIs.

Entry Gate Tenant Isolation in SecureRunnable

At the agent entry boundary, SecureRunnable inspects the caller's authenticated identity against the configured target_tenant boundary set. When engine.enforce_strict_tenant_isolation = True is enabled on the Policy Decision Point (PDP), any cross-tenant mismatch immediately logs a TENANT_BOUND_VIOLATION event with SEVERITY_CRITICAL_ALERT directly within pdp.evaluate(). The factory method AgentAccessDeniedException.for_agent() detects the TENANT_MISMATCH diagnostic reason and raises a typed TenantIsolationError.

langchain_security_extension/security_adapter_langchain/runnable_guard.py

SECURITY LIBRARY SNIPPET (INFORMATIVE) langchain_security_extension/security_adapter_langchain/runnable_guard.py
PYTHON 
class SecureRunnable:
    """Wraps LangChain runnables with entry authorization and tenant isolation boundaries."""

    def _evaluate_entry_access(self, config: Optional[Dict[str, Any]]) -> SecurityIdentity:
        identity = IdentityBridge.extract_from_config(config) or IdentityBridge.resolve_identity(None)
        self._prepare_config(config, identity)

        tenant = self._target_tenant if self._target_tenant is not None else identity.tenant_id
        request = PolicyEvaluationRequest.create(
            subject=identity,
            action=ACTION_AGENT_INVOKE,
            resource={
                "target_id": self._agent_name,
                "tenant_id": tenant,
                "required_roles": self._required_roles,
                "min_clearance": self._min_clearance,
            },
            context={"config": config} if config else {},
        )

        decision = self._pdp_engine.evaluate(request)
        if not decision.is_permitted:
            # PDP automatically logs TENANT_BOUND_VIOLATION with CRITICAL_ALERT under strict tenant isolation
            raise AgentAccessDeniedException.for_agent(
                agent_name=self._agent_name,
                message=f"Agent invocation denied for '{self._agent_name}': {decision.diagnostic_reason}",
                diagnostic_reason=decision.diagnostic_reason,
            )
        return identity

Configuring Agent Tenant Boundaries in Your Application

In an enterprise deployment, your application must declare authorized tenant boundaries and bind them directly to agent workflows. The reference incident assistant demonstrates this pattern by establishing tenant boundaries during PDP engine initialization and binding target_tenant=ALLOWED_TENANTS across its agent wrappers:

1. Policy Engine Configuration & Tenant Whitelisting

You can inspect the full implementation in app/security/security_integration.py of the assistant.

Your application's central security configuration initializes the PDP engine, declares permitted tenant IDs, and enables strict boundary enforcement. As shown in the reference app/security/security_integration.py:

DEMO APP (NEW FILE) app/security/security_integration.py
PYTHON 
# app/security/security_integration.py
ALLOWED_TENANTS: Set[str] = {
    DEFAULT_TENANT_ID,
    ENTERPRISE_TENANT_ID,
    "CORP_MAIN",
}

def get_pdp_engine() -> EmbeddedPolicyEngine:
    ...
    engine = EmbeddedPolicyEngine.with_components(...)
    engine.enforce_strict_tenant_isolation = True  # Escalates tenant mismatches to CRITICAL_ALERT
    return engine

2. Guarding Upstream Workflow Entry

You can inspect the full implementation in app/agents/triage_agent.py of the assistant.

Every user-facing or upstream agent entry point must declare its target tenant boundaries. As illustrated in the triage agent (app/agents/triage_agent.py), wrapping the core workflow with SecureRunnable.wrap() and binding target_tenant=ALLOWED_TENANTS guarantees that incoming requests from unauthorized tenants fail closed before consuming model tokens:

DEMO APP (DIFF) app/agents/triage_agent.py
DIFF 
--- a/app/agents/triage_agent.py
+++ b/app/agents/triage_agent.py
@@ -47,6 +59,13 @@ class TriageAgent:
             self.model_class,
             self.api_key
         )
+        self.secured_workflow = SecureRunnable.wrap(
+            runnable=self._execute_triage,
+            pdp_engine=get_pdp_engine(),
+            agent_name="triage_agent",
+            required_roles=TRIAGE_REQUIRED_ROLES,
+            min_clearance=TRIAGE_MIN_CLEARANCE,
+            target_tenant=ALLOWED_TENANTS,
         )

3. Guarding Downstream Workflow Handoffs

You can inspect the full implementation in app/agents/remediation_agent.py of the assistant.

Similarly, downstream or specialist execution agents must enforce identical tenant bounds. As shown in app/agents/remediation_agent.py, binding target_tenant=ALLOWED_TENANTS ensures that secondary agents cannot be invoked out-of-bounds, even when called internally during multi-agent handoffs:

DEMO APP (DIFF) app/agents/remediation_agent.py
DIFF 
--- a/app/agents/remediation_agent.py
+++ b/app/agents/remediation_agent.py
@@ -105,6 +89,14 @@ class RemediationAgent:
+        self.secured_workflow = SecureRunnable.wrap(
+            runnable=self._execute_remediation,
+            pdp_engine=get_pdp_engine(),
+            agent_name="remediation_agent",
+            required_roles=TRIAGE_REQUIRED_ROLES,
+            min_clearance=TRIAGE_MIN_CLEARANCE,
+            target_tenant=ALLOWED_TENANTS,
+        ).with_violation_handler(
+            violation_handler=self.violation_handler,
+            fallback_builder=build_violation_fallback_report,
+        )

Full Delegation: Cleaning HTTP Filters of Ad-Hoc Tenant Checks

In conventional web applications, developers frequently rely on custom HTTP middleware or API filters to inspect incoming tenant headers (e.g., if tenant not in allowed: raise HTTPException(403)). While simple web filters work well for standard, single-shot REST requests, they are inadequate for enterprise GenAI architectures. Once a request passes through a web filter into an autonomous agent pipeline, sub-agents, asynchronous tool runners, and vector retrievers lose visibility into the HTTP context unless complex thread-locals or global state are maintained.

By delegating tenant perimeter checks entirely to the security library's SecureRunnable and EmbeddedPolicyEngine, your enterprise codebase eliminates scattered, hardcoded tenant checks from web middleware. Instead, tenant validation is anchored directly into the LangChain execution lifecycle, providing in-band propagation, centralized audit attribution (TENANT_BOUND_VIOLATION), and typed exception flows (TenantIsolationError). In our reference application, this pattern cleans up app/security/auth.py by removing legacy ad-hoc filter checks:

You can inspect the full implementation in app/security/auth.py of the assistant.

app/security/auth.py

DEMO APP (DIFF) app/security/auth.py
DIFF 
--- a/app/security/auth.py
+++ b/app/security/auth.py
@@ -152,9 +137,0 @@ async def get_current_security_context(...)
-    # Cross-tenant isolation check: reject external/unauthorized tenants
-    if sec_context.tenantId not in (DEFAULT_TENANT_ID, ENTERPRISE_TENANT_ID):
-        logger.warning(f"[RequestSecurityFilter] Cross-tenant isolation violation: tenantId={sec_context.tenantId}")
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail={"error": "Forbidden", "message": f"Tenant '{sec_context.tenantId}' is unauthorized for target resources"}
-        )

Code Hardening: Intercepting Cross-Tenant Breaches in API Routers

Enterprise API routers that serve agent workflows must intercept security policy exceptions cleanly at the transport boundary. Rather than letting authorization failures bubble up as unhandled 500 errors or leaking stack traces, your API endpoints should catch typed TenantIsolationError exceptions. This allows the API to terminate foreign tenant access immediately, record structured telemetry, and return an HTTP 403 response before model tokens or compute resources are consumed. In the reference implementation, this defense is organized across two distinct operational phases in app/routers/tickets.py:

Phase 1: Inbound Identity Resolution & Triage Entry PEP

At your API's front door, the endpoint extracts the caller's authenticated identity and translates it into a native LangChain RunnableConfig via IdentityBridge.create_runnable_config(). Invoking the primary agent within a try/except TenantIsolationError block ensures that any tenant boundary violation is intercepted immediately with zero model token expenditure, as demonstrated in the triage phase of app/routers/tickets.py:

You can inspect the full implementation in app/routers/tickets.py of the assistant.

app/routers/tickets.py (Triage Phase)

DEMO APP (DIFF) app/routers/tickets.py
DIFF 
--- a/app/routers/tickets.py
+++ b/app/routers/tickets.py
@@ -51,5 +54,14 @@ def submit_ticket(
-    # Triage Phase
-    incident_context = triage_agent_instance.triage(request)
+    sec_identity = security_identity_from_context(sec_context)
+    run_config = IdentityBridge.create_runnable_config(sec_identity)
+
+    # Triage Phase (Guarded by SecureRunnable Entry PEP)
+    try:
+        incident_context = triage_agent_instance.triage(request, config=run_config)
+    except TenantIsolationError as exc:
+        logger.warning(f"Triage entry tenant isolation violation: {exc}")
+        return build_entry_access_denied_report(request, exc, response, "BLOCKED_TENANT")
+    except AgentAccessDeniedException as exc:
+        logger.warning(f"Triage entry access denied: {exc}")
+        return build_entry_access_denied_report(request, exc, response, "BLOCKED")

Phase 2: Remediation Handoff & In-Band Tenant Context Preservation

When an enterprise workflow transitions between multiple agents (such as routing from an intent classifier or triage agent to a domain specialist), passing run_config ensures that SecureRunnable.handoff() immutably carries the caller's verified SecurityIdentity in-band across agent boundaries without relying on fragile global variables or thread-locals. The API router maintains error boundaries around each handoff stage to catch any tenant isolation breaches before mutative tools or expensive operations run. In the reference application's remediation handoff:

app/routers/tickets.py (Remediation Handoff Phase)

DEMO APP (DIFF) app/routers/tickets.py
DIFF 
--- a/app/routers/tickets.py
+++ b/app/routers/tickets.py
@@ -63,4 +75,12 @@ def submit_ticket(
-    # Remediation Phase
-    report = remediation_agent_instance.process_incident(incident_context)
+    # Remediation Phase (In-band SecureRunnable.handoff preserving context)
+    try:
+        report = remediation_agent_instance.process_incident(incident_context, config=run_config)
+    except TenantIsolationError as exc:
+        logger.warning(f"Remediation handoff tenant isolation violation: {exc}")
+        return build_entry_access_denied_report(incident_context, exc, response, "BLOCKED_TENANT", stage="remediation")
+    except AgentAccessDeniedException as exc:
+        logger.warning(f"Remediation handoff entry access denied: {exc}")
+        return build_entry_access_denied_report(incident_context, exc, response, "BLOCKED", stage="remediation")

Retrieval AST Tenant Partition Push-Down

In enterprise RAG architectures, multi-tenancy cannot rely on system prompts instructing the model to "only read runbooks for your organization"—a pattern vulnerable to prompt injection and retrieval leakage. Retrieval must enforce tenant isolation at query compilation time using Deterministic Filter Push-Down: SecureRetriever automatically injects {SEC_TENANT_ID: identity.tenant_id} into the vector store AST filter, ensuring foreign-tenant chunks are excluded at the database index layer before similarity vectors are compared.

langchain_security_extension/security_adapter_langchain/retriever_wrapper.py

SECURITY LIBRARY SNIPPET (INFORMATIVE) langchain_security_extension/security_adapter_langchain/retriever_wrapper.py
PYTHON 
class SecureRetriever:
    """Wraps vector store retrievers with AST pre-filtering and post-retrieval clearance pruning."""

    def build_full_ast(self, identity: Optional[SecurityIdentity]) -> PrefilterAST:
        """Constructs unconstrained full AST from SecurityIdentity pushing down tenant boundary."""
        if identity is None or identity.is_anonymous or identity.subject_id is None:
            return PrefilterAST({SEC_TENANT_ID: "__DENY_ALL_UNAUTHENTICATED__"})

        and_clauses: List[Dict[str, Any]] = [
            {SEC_TENANT_ID: identity.tenant_id},
            {SEC_CLEARANCE_FLOOR: {"$lte": identity.clearance_level}},
        ]
        if identity.roles:
            and_clauses.append({SEC_ALLOWED_ROLES: {"$in": list(identity.roles)}})
        return PrefilterAST({"$and": and_clauses})

To harden your enterprise knowledge base, wrap your standard vector retriever with SecureRetriever.wrap(). The reference implementation demonstrates this in app/agents/knowledge_base_store.py, upgrading a basic ChromaDB retriever into a secure, tenant-isolated retrieval pipeline:

You can inspect the full implementation in app/agents/knowledge_base_store.py of the assistant.

app/agents/knowledge_base_store.py

DEMO APP (DIFF) app/agents/knowledge_base_store.py
DIFF 
--- a/app/agents/knowledge_base_store.py
+++ b/app/agents/knowledge_base_store.py
@@ -193,3 +246,10 @@
-        self._retriever = self._vector_store.as_retriever(search_kwargs={"k": TOP_K_DEFAULT})
+        pdp_engine = get_pdp_engine()
+        base_retriever = self._vector_store.as_retriever(search_kwargs={"k": TOP_K_DEFAULT})
+        self._retriever = SecureRetriever.wrap(
+            base_retriever,
+            pdp_engine=pdp_engine,
+            oversampling_factor=DEFAULT_OVERSAMPLING_FACTOR,
+            secret_key=config.security_hmac_secret,
+            store_profile=STORE_PROFILE_FULL_AST,
+            secondary_fetch_fn=self._secondary_fetch_chroma_page,
+        )

Second Line of Defense (Post-Retrieval In-Memory Pruning): If a vector store profile only supports partial AST filtering (or fails to filter correctly), SecureRetriever evaluates each candidate chunk's doc_tenant against caller identity via PDP evaluate(), ensuring foreign-tenant chunks are pruned in memory before reaching prompt context.

Verification Tests (Guardrail Tests)

Automated regression testing is essential for verifying that multi-tenant boundaries cannot be bypassed by role elevation, parameter manipulation, or permissive configuration changes. When building CI/CD test pipelines for your enterprise agent systems, structure verification tests to assert boundary enforcement across each architectural tier. The reference project provides standardized guardrail test scripts illustrating these assertions: